"""Search provider interface — neutral ranking, no trust implication.""" from __future__ import annotations import abc import os import time from datetime import datetime, timezone from typing import Any from pydantic import BaseModel, Field, field_validator # --------------------------------------------------------------------------- # NormalizedResult # --------------------------------------------------------------------------- class NormalizedResult(BaseModel): """Normalisiertes Suchergebnis — provider-unabhängig. WICHTIG: rank ist KEIN truth_score. Suchergebnis auf Position 1 ist nicht automatisch glaubwürdiger als Position 8. """ title: str = Field(..., min_length=1, description="Titel des Suchergebnisses.") url: str = Field(..., description="Original-URL der Quelle.") snippet: str = Field(default="", description="Auszug/Snippet aus dem Suchergebnis.") provider: str = Field(..., description="Name des Providers, der dieses Ergebnis geliefert hat.") rank: int = Field( ..., ge=1, description="Original-Ranking des Providers. KEIN Vertrauens- oder Wahrheitsindikator.", ) retrieved_at: datetime = Field( default_factory=lambda: datetime.now(timezone.utc), description="Zeitpunkt der Abrufung (UTC).", ) extra: dict[str, Any] = Field( default_factory=dict, description="Zusätzliche provider-spezifische Metadaten.", ) @field_validator("url") @classmethod def _validate_url(cls, v: str) -> str: """Stelle sicher, dass URL einen validen Scheme hat.""" if not v.startswith(("http://", "https://")): raise ValueError("URL muss mit http:// oder https:// beginnen") return v @field_validator("title") @classmethod def _validate_title(cls, v: str) -> str: if not v.strip(): raise ValueError("title darf nicht leer sein") return v @classmethod def from_raw( cls, provider_name: str, title: str, url: str, snippet: str = "", rank: int = 1, extra: dict[str, Any] | None = None, retrieved_at: datetime | None = None, ) -> "NormalizedResult": """Factory: Erzeuge ein NormalizedResult aus rohen Provider-Daten.""" if retrieved_at is None: retrieved_at = datetime.now(timezone.utc) return cls( title=title.strip() if title else "", url=url.strip(), snippet=snippet.strip() if snippet else "", provider=provider_name, rank=rank, retrieved_at=retrieved_at, extra=extra or {}, ) # --------------------------------------------------------------------------- # ProviderConfig # --------------------------------------------------------------------------- class ProviderConfig(BaseModel): """Konfiguration für einen einzelnen SearchProvider.""" name: str = Field(..., description="Eindeutiger Provider-Name.") enabled: bool = Field(default=True, description="Ob der Provider aktiviert ist.") timeout_seconds: float = Field( default=10.0, description="Timeout pro Provider-Anfrage in Sekunden.", gt=0, ) max_results: int = Field( default=10, ge=1, description="Max Ergebnisse pro Provider.", ) # --------------------------------------------------------------------------- # SearchProvider Interface # --------------------------------------------------------------------------- class SearchProvider(abc.ABC): """Schnittstelle für Suchmaschinen. Kein Provider darf als Evidenz-Ranking verwendet werden. Das Ergebnis-Ranking einer Suchmaschine ist KEIN Vertrauensindikator. """ _provider_name: str = "" _provider_config: ProviderConfig | None = None @abc.abstractmethod async def search( self, query: str, language: str = "de", max_results: int = 10, ) -> list[NormalizedResult]: """Suche durchführen. Parameters ---------- query: Suchanfrage-Text. language: Sprachcode (z.B. 'de', 'en'). max_results: Maximale Anzahl zurückzugebender Ergebnisse. Returns ------- list[NormalizedResult] — ohne Ranking-Implikation. """ ... @abc.abstractmethod async def get_metadata(self) -> dict[str, Any]: """Provider-Metadata (Name, Version, capabilities).""" ... async def health_check(self) -> bool: """Prüfe, ob der Provider erreichbar ist.""" try: meta = await self.get_metadata() return bool(meta.get("name")) except Exception: return False # --------------------------------------------------------------------------- # Multi-Provider-Orchestrator # --------------------------------------------------------------------------- class MultiProviderSearch: """Orchestrator für mehrere SearchProvider parallel. - Sammelt Ergebnisse aller aktiven Provider. - Dedupliziert nach URL. - Behält provider-Information für jedes Ergebnis. - Fällt stillschweigend auf aus, wenn ein Provider ausfällt. """ def __init__(self, providers: list[SearchProvider] | None = None) -> None: """Initialisiere mit einer Liste von SearchProvider-Instanzen.""" self._providers: list[SearchProvider] = providers or [] self._name_index: dict[str, SearchProvider] = {} for p in self._providers: name = getattr(p, "_provider_name", p.__class__.__name__) self._name_index[name] = p def add_provider(self, provider: SearchProvider, config: ProviderConfig | None = None) -> None: """Füge einen Provider hinzu.""" self._providers.append(provider) name = getattr(provider, "_provider_name", provider.__class__.__name__) self._name_index[name] = provider def get_provider(self, name: str) -> SearchProvider | None: """Gib einen Provider nach Namen zurück.""" return self._name_index.get(name) def enabled_providers(self) -> list[tuple[str, SearchProvider, ProviderConfig]]: """Gib alle aktivierten (Name, Provider, Config)-Tupel zurück.""" results: list[tuple[str, SearchProvider, ProviderConfig]] = [] for p in self._providers: name = getattr(p, "_provider_name", p.__class__.__name__) cfg = getattr(p, "_provider_config", None) if cfg is None: cfg = ProviderConfig(name=name, enabled=True) if not isinstance(cfg, ProviderConfig): cfg = ProviderConfig(name=name, enabled=True) if cfg.enabled: results.append((name, p, cfg)) return results async def search( self, query: str, language: str = "de", max_results: int = 10, provider_names: list[str] | None = None, timeout_seconds: float | None = None, ) -> list[NormalizedResult]: """Starte parallele Suche über alle (oder ausgewählte) Provider. Returns ------- Deduplizierte, normalisierte Ergebnisse. """ import asyncio candidates: list[tuple[str, SearchProvider, ProviderConfig]] = [] if provider_names: for pname in provider_names: p = self.get_provider(pname) if p is None: continue cfg = getattr(p, "_provider_config", None) if cfg is None: cfg = ProviderConfig(name=pname, enabled=True) elif not isinstance(cfg, ProviderConfig): cfg = ProviderConfig(name=pname, enabled=True) candidates.append((pname, p, cfg)) else: candidates = self.enabled_providers() if not candidates: return [] async def _run(p: SearchProvider, cfg: ProviderConfig) -> list[NormalizedResult]: to = timeout_seconds if timeout_seconds else cfg.timeout_seconds try: return await asyncio.wait_for( p.search(query, language=language, max_results=cfg.max_results), timeout=to, ) except asyncio.TimeoutError: return [] except Exception: return [] tasks = [_run(p, cfg) for _, p, cfg in candidates] raw_results: list[list[NormalizedResult]] = await asyncio.gather(*tasks) # Zusammenführen und deduplizieren nach URL seen_urls: set[str] = set() deduped: list[NormalizedResult] = [] for batch in raw_results: for r in batch: url_key = r.url.lower().rstrip("/") if url_key not in seen_urls: seen_urls.add(url_key) deduped.append(r) if len(deduped) >= max_results: break if len(deduped) >= max_results: break return deduped[:max_results] async def search_with_debug( self, query: str, language: str = "de", max_results: int = 10, provider_names: list[str] | None = None, timeout_seconds: float | None = None, ) -> dict[str, Any]: """Parallele Suche mit Debug-Informationen (Timing, Provider-Status).""" import asyncio start = time.monotonic() candidates: list[tuple[str, SearchProvider, ProviderConfig]] = [] if provider_names: for pname in provider_names: p = self.get_provider(pname) if p is None: continue cfg = getattr(p, "_provider_config", None) if cfg is None: cfg = ProviderConfig(name=pname, enabled=True) elif not isinstance(cfg, ProviderConfig): cfg = ProviderConfig(name=pname, enabled=True) candidates.append((pname, p, cfg)) else: candidates = self.enabled_providers() timings: dict[str, float] = {} results_by_provider: dict[str, list[NormalizedResult]] = {} errors: dict[str, str] = {} async def _run_debug(p: SearchProvider, cfg: ProviderConfig) -> None: to = timeout_seconds if timeout_seconds else cfg.timeout_seconds t0 = time.monotonic() try: res = await asyncio.wait_for( p.search(query, language=language, max_results=cfg.max_results), timeout=to, ) timings[p.__class__.__name__] = time.monotonic() - t0 results_by_provider[p.__class__.__name__] = res except asyncio.TimeoutError: timings[p.__class__.__name__] = time.monotonic() - t0 errors[p.__class__.__name__] = "timeout" except Exception as exc: timings[p.__class__.__name__] = time.monotonic() - t0 errors[p.__class__.__name__] = str(exc) tasks = [_run_debug(p, cfg) for _, p, cfg in candidates] await asyncio.gather(*tasks) # Dedupliziere seen: set[str] = set() deduped: list[NormalizedResult] = [] for batch in results_by_provider.values(): for r in batch: u = r.url.lower().rstrip("/") if u not in seen: seen.add(u) deduped.append(r) deduped = deduped[:max_results] elapsed = time.monotonic() - start return { "results": deduped, "debug": { "total_time_seconds": round(elapsed, 3), "providers": { name: { "timing_seconds": round(timings.get(name, 0), 3), "result_count": len(results_by_provider.get(name, [])), "error": errors.get(name), } for name, _, _ in candidates }, "total_results": len(deduped), }, }