"""Such-Schicht (R2): Wikipedia de+en immer, dazu Kette Tavily→Brave→ddgs. Circuit Breaker je Backend; Teilausfall = degraded-Befund, nie Lauf-Crash.""" import asyncio import os from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit import httpx from . import config, db, engine _TRACKING = {"utm_source", "utm_medium", "utm_campaign", "utm_term", "utm_content", "fbclid", "gclid", "ref"} _ausfaelle: dict[str, int] = {} _wiki_sem = asyncio.Semaphore(config.WIKI_MAX_PARALLEL) def url_normieren(url: str) -> str: s = urlsplit(url) query = urlencode([(k, v) for k, v in parse_qsl(s.query) if k.lower() not in _TRACKING]) return urlunsplit((s.scheme.lower(), s.netloc.lower(), s.path.rstrip("/") or "/", query, "")) def _tot(backend: str) -> bool: return _ausfaelle.get(backend, 0) >= config.CIRCUIT_BREAKER_N def _ausfall(backend: str) -> None: _ausfaelle[backend] = _ausfaelle.get(backend, 0) + 1 async def _http_json(url: str, **kw) -> dict | list | None: async with httpx.AsyncClient( timeout=config.SUCHE_TIMEOUT_S, headers={"User-Agent": config.user_agent()}) as client: r = await client.get(url, **kw) r.raise_for_status() return r.json() async def _wikipedia(query: str) -> list[dict]: treffer = [] for sprache in config.SPRACHEN_SUCHE: backend = f"wikipedia_{sprache}" if _tot(backend): continue try: async with _wiki_sem: daten = await _http_json( f"https://{sprache}.wikipedia.org/w/rest.php/v1/search/page", params={"q": query, "limit": 4}) for s in (daten or {}).get("pages", []): treffer.append({ "titel": s.get("title", ""), "backend": backend, "url": f"https://{sprache}.wikipedia.org/wiki/" f"{s.get('key', s.get('title', ''))}"}) _ausfaelle[backend] = 0 except Exception: _ausfall(backend) return treffer async def _tavily(query: str) -> list[dict]: key = os.getenv("TAVILY_API_KEY") if not key: return [] async with httpx.AsyncClient(timeout=config.SUCHE_TIMEOUT_S) as client: r = await client.post("https://api.tavily.com/search", json={"api_key": key, "query": query, "max_results": config.SUCHE_MAX_TREFFER}) r.raise_for_status() return [{"titel": t.get("title", ""), "url": t["url"], "backend": "tavily"} for t in r.json().get("results", [])] async def _brave(query: str) -> list[dict]: key = os.getenv("BRAVE_API_KEY") if not key: return [] locale = config.SUCHE_LOCALE.get(config.SPRACHE, {}) daten = await _http_json("https://api.search.brave.com/res/v1/web/search", params={"q": query, "count": config.SUCHE_MAX_TREFFER, "country": locale.get("country", "US"), "search_lang": locale.get("search_lang", "en")}, headers={"X-Subscription-Token": key}) return [{"titel": t.get("title", ""), "url": t["url"], "backend": "brave"} for t in (daten or {}).get("web", {}).get("results", [])] async def _ddgs(query: str) -> list[dict]: def _sync(): from ddgs import DDGS region = config.SUCHE_LOCALE.get(config.SPRACHE, {}).get("region", "us-en") return DDGS().text(query, region=region, max_results=config.SUCHE_MAX_TREFFER) await asyncio.sleep(config.DDGS_PAUSE_S) # Drosselung (ToS-Grauzone) rows = await asyncio.to_thread(_sync) return [{"titel": r.get("title", ""), "url": r.get("href") or r.get("link", ""), "backend": "ddgs"} for r in rows or [] if r.get("href") or r.get("link")] _KETTE = (("tavily", _tavily), ("brave", _brave), ("ddgs", _ddgs)) async def web_suchen(query: str) -> tuple[list[dict], list[str]]: """→ (treffer, fehler). Kette bis zum ersten Backend mit Ergebnissen (Fallback statt Fan-out, R2). Wikipedia ist KEINE privilegierte Extra-Quelle mehr — sie kommt über die Startquellen (Seed) und über normale Web-Treffer herein (Lektion 106).""" treffer: list[dict] = [] fehler = [] for name, fn in _KETTE: if _tot(name): continue for versuch in range(config.SUCHE_RETRIES + 1): try: web = await asyncio.wait_for(fn(query), config.SUCHE_TIMEOUT_S * 2) if web: _ausfaelle[name] = 0 return treffer + web, fehler break # Backend ohne Key/Treffer → nächstes except Exception as e: if versuch == config.SUCHE_RETRIES: _ausfall(name) fehler.append(f"{name}: {type(e).__name__}") return treffer, fehler @engine.worker("suche.suchen") async def suchen(task: dict) -> engine.Ergebnis: payload = db.uj(task["payload"], {}) query = payload.get("query", "") zweck = payload.get("zweck", "korpus") seed = bool(payload.get("seed")) if seed: # Startquellen-Seed: Pflicht-Kandidaten (umgehen den Deckel), aber # OHNE weitere Privilegien — das Inhalts-Urteil läuft wie überall. # Titel erst über die Wiki-Suche auflösen („AlpineJS" heißt dort # „Alpine.js"); der direkte URL-Bau ist nur Fallback. from urllib.parse import quote zweck = "seed" titel = db.one("SELECT COALESCE(NULLIF(titel,''), name) t FROM topics " "WHERE name=?", task["topic"])["t"] gefunden = {} if config.FAKE else { t["backend"]: t for t in reversed(await _wikipedia(titel))} treffer = [gefunden.get(f"wikipedia_{s}", {"titel": f"Wikipedia ({s}): {titel}", "url": f"https://{s}.wikipedia.org/wiki/" f"{quote(titel.replace(' ', '_'))}", "backend": f"wikipedia_{s}"}) for s in config.SPRACHEN_SUCHE] fehler = [] elif config.FAKE: from . import fakes treffer, fehler = fakes.suche(query), [] else: treffer, fehler = await web_suchen(query) # Keine Vorsortierung, kein Titel-Filter — beurteilt wird nach dem Laden # auf dem Inhalt (Lektion 102/106). # Quellen-Deckel je Topic (0 = unbegrenzt, dann stoppt nur die Sättigung) einstellung = db.one("SELECT quellen_max FROM topics WHERE name=?", task["topic"]) or {} deckel = einstellung.get("quellen_max") if deckel is None: deckel = config.QUELLEN_MAX neue = [] for t in treffer: if zweck == "korpus" and deckel > 0: vorhanden = db.one("SELECT COUNT(*) c FROM quellen WHERE topic=? AND " "zweck='korpus' AND status NOT IN " "('fehler','leer','braucht_js','aussortiert')", task["topic"])["c"] if vorhanden >= deckel: break url_norm = url_normieren(t["url"]) qid = db.insert("quellen", ignore=True, topic=task["topic"], titel=t["titel"][:200], url=t["url"], url_norm=url_norm, backend=t["backend"], runde=task["runde"], zweck=zweck, soll_id=payload.get("soll_id")) if qid: neue.append({"knoten": "laden", "item": f"quelle:{qid}", "payload": {"quelle_id": qid}}) if fehler and not treffer: db.insert("befunde", run_id=task["run_id"], stufe="korpus", knoten="suche", art="degraded", item=query[:80], detail="; ".join(fehler)[:300]) return engine.Ergebnis(daten={"treffer": len(treffer), "neu": len(neue)}, neue_tasks=neue)