Spaces:
Runtime error
Runtime error
File size: 5,762 Bytes
feef734 c64f4ef | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 | """Бесплатный интернет-агент с минимум 3 способами поиска"""
import re
import logging
import hashlib
from datetime import datetime, timedelta
from typing import Dict, List, Tuple, Any, Optional
from collections import Counter
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class FreeInternetAgent:
"""Интернет-агент с бесплатными поисковыми системами"""
def __init__(self, cache_ttl: int = 3600):
self.session = requests.Session()
retry = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503, 504])
self.session.mount("https://", HTTPAdapter(max_retries=retry))
self.session.headers.update({
'User-Agent': 'PinkSky/7.0 (Linux; Research)'
})
self.cache: Dict[str, Tuple[Any, datetime]] = {}
self.cache_ttl = cache_ttl
self.logger = logging.getLogger(__name__)
self._has_bs4 = False
try:
import bs4
self._has_bs4 = True
except ImportError:
pass
# SearXNG инстансы
self.searxng_instances = [
"https://searx.be",
"https://search.bus-hit.me",
"https://searx.nixnet.xyz",
"https://searx.tuxcloud.net",
"https://searx.moe",
]
# Асинхронная проверка доступности при старте
self.healthy_searxng = self._check_searxng_health()
def _check_searxng_health(self) -> List[str]:
healthy = []
for instance in self.searxng_instances:
try:
response = self.session.get(f"{instance}/health", timeout=5)
if response.status_code == 200:
healthy.append(instance)
except Exception as e:
self.logger.warning(f"SearXNG health check failed for {instance}: {e}")
self.logger.info(f"Healthy SearXNG instances: {len(healthy)}/{len(self.searxng_instances)}")
return healthy
def _get_cache_key(self, *args, **kwargs) -> str:
key = f"{args}_{sorted(kwargs.items())}"
return hashlib.md5(key.encode()).hexdigest()
def _get_from_cache(self, key: str) -> Optional[Any]:
if key in self.cache:
data, timestamp = self.cache[key]
if datetime.now() - timestamp < timedelta(seconds=self.cache_ttl):
return data
else:
del self.cache[key]
return None
def _save_to_cache(self, key: str, data: Any) -> None:
self.cache[key] = (data, datetime.now())
def search_web(self, query: str, num_results: int = 5) -> List[Dict[str, str]]:
cache_key = self._get_cache_key('search', query, num_results)
cached = self._get_from_cache(cache_key)
if cached is not None:
return cached
results = []
# Способ 1: SearXNG (мета-поиск)
results = self._search_searxng(query, num_results)
# Способ 2: DuckDuckGo API
if not results:
results = self._search_duckduckgo(query, num_results)
# Способ 3: Google
if not results:
results = self._search_google(query, num_results)
# Способ 4: Яндекс
if not results and any(ord(c) > 1024 for c in query):
results = self._search_yandex(query, num_results)
self._save_to_cache(cache_key, results)
return results
def _search_searxng(self, query: str, num_results: int) -> List[Dict[str, str]]:
if not self.healthy_searxng:
self.healthy_searxng = self._check_searxng_health()
if not self.healthy_searxng:
self.logger.warning("No healthy SearXNG instances available")
return []
results = []
for instance in self.healthy_searxng:
try:
url = f"{instance}/search"
params = {
"q": query,
"format": "json",
"categories": "general",
"engines": "google,bing,duckduckgo,startpage",
"language": "en",
"pageno": 1
}
response = self.session.get(url, params=params, timeout=20)
response.raise_for_status()
data = response.json()
if 'results' in data:
for item in data['results'][:num_results]:
results.append({
'title': item.get('title', '')[:100],
'url': item.get('url', ''),
'snippet': item.get('content', '')[:200],
'source': 'searxng',
'engine': item.get('engine', '')
})
self.logger.info(f"SearXNG ({instance}): {len(results)} results")
break
except Exception as e:
self.logger.warning(f"SearXNG error for {instance}: {e}")
self.healthy_searxng.remove(instance)
return results
# Удалить эту строку:
# INTERNET_AGENT = FreeInternetAgent(cache_ttl=3600)
# Вместо неё добавляем функцию-фабрику:
def get_internet_agent() -> FreeInternetAgent:
"""Lazy-инициализация интернет-агента с глобальным состоянием."""
if not hasattr(get_internet_agent, 'instance'):
get_internet_agent.instance = FreeInternetAgent(cache_ttl=3600)
return get_internet_agent.instance
|