creator3/backend/suche.py

"""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)