tools/web_search.py

"""
web_search.py — Web search pipeline (allineato con registry.py - S434)
Usa gli stessi provider di registry._web_search: Brave → Tavily → Wikipedia → HN
con chiamate parallele. Mantiene la firma pubblica originale per compatibilità con api/web.py.
"""
import httpx
import asyncio
import os

import logging
_logger = logging.getLogger("tools.web_search")


async def web_search(query: str, focus: str = "general", max_results: int = 6) -> dict:
    """
    Pipeline parallela: Brave → Tavily → Wikipedia → HackerNews.
    Merge con deduplicazione per URL, priorità per source.
    Allineata con registry._web_search — zero DDG scraping.
    """
    import re as _re, urllib.parse as _urlparse, urllib.request as _urlreq

    _headers = {"User-Agent": "Mozilla/5.0 (compatible; AgentBot/1.0)"}
    brave_key  = os.environ.get("BRAVE_SEARCH_API_KEY", "")
    tavily_key = os.environ.get("TAVILY_API_KEY", "")

    async def _brave() -> list:
        if not brave_key:
            return []
        try:
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.get(
                    "https://api.search.brave.com/res/v1/web/search",
                    params={"q": query, "count": max_results, "text_decorations": "0"},
                    headers={**_headers, "Accept": "application/json",
                             "X-Subscription-Token": brave_key},
                )
                if r.status_code == 200:
                    return [
                        {"title": it["title"],
                         # S602: snippet 300→500 — Brave web_search parity con Tavily/Wikipedia
                         "snippet": (it.get("description") or "")[:500],
                         "url": it["url"], "source": "Brave", "score": 1.0}
                        for it in r.json().get("web", {}).get("results", [])[:max_results]
                        if it.get("title") and it.get("url")
                    ]
        except Exception as _exc:
            _logger.debug("[web_search] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    async def _tavily() -> list:
        if not tavily_key:
            return []
        try:
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.post(
                    "https://api.tavily.com/search",
                    json={"api_key": tavily_key, "query": query,
                          "max_results": max_results, "include_answer": False},
                    headers={**_headers, "Content-Type": "application/json"},
                )
                if r.status_code == 200:
                    return [
                        {"title": it["title"],
                         # S601: snippet 300→500 — Tavily web_search parity con registry.py
                         "snippet": (it.get("content") or "")[:500],
                         "url": it["url"], "source": "Tavily", "score": 0.9}
                        for it in r.json().get("results", [])[:max_results]
                        if it.get("title") and it.get("url")
                    ]
        except Exception as _exc:
            _logger.debug("[web_search] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    async def _wikipedia() -> list:
        try:
            wiki_qs = _urlparse.urlencode({
                "action": "query", "list": "search", "srsearch": query,
                "format": "json", "utf8": "1",
                "srlimit": min(max_results, 4), "srnamespace": "0",
            })
            # B-GAP-D: IT Wikipedia per query italiane — migliore pertinenza su task IT
            _it_signals = {'come', 'cosa', 'perché', 'quando', 'dove', 'quale', 'quali',
                           'creare', 'fare', 'usare', 'per', 'con', 'una', 'del', 'della'}
            _wiki_lang = "it" if len({w.lower() for w in _re.split(r'\W+', query) if w} & _it_signals) >= 2 else "en"
            wiki_req = _urlreq.Request(
                f"https://{_wiki_lang}.wikipedia.org/w/api.php?{wiki_qs}",
                headers={"User-Agent": "agente-ai/3.2"},
            )
            wiki_data = await asyncio.to_thread(
                lambda: __import__("json").loads(_urlreq.urlopen(wiki_req, timeout=7).read())
            )
            return [
                {"title": it["title"],
                 "snippet": _re.sub(r"<[^>]+>", "", it.get("snippet", ""))[:500],
                 "url": f"https://{_wiki_lang}.wikipedia.org/wiki/{_urlparse.quote(it['title'].replace(' ', '_'))}",
                 "source": "Wikipedia", "score": 0.75}
                for it in wiki_data.get("query", {}).get("search", [])[:3]
                if it.get("title")
            ]
        except Exception:
            return []

    async def _hackernews() -> list:
        if focus not in ("news", "general", "technical"):
            return []
        try:
            async with httpx.AsyncClient(timeout=8) as c:
                r = await c.get(
                    "https://hn.algolia.com/api/v1/search",
                    params={"query": query, "hitsPerPage": 4, "tags": "story"},
                )
                return [
                    {"title": h.get("title", "")[:300],  # S607: 200→300
                     "snippet": (h.get("story_text") or "")[:300],  # S583: 250→300
                     "url": h.get("url") or f"https://news.ycombinator.com/item?id={h.get('objectID','')}",
                     "source": "HackerNews",
                     "score": min(1.0, (h.get("points", 0) or 0) / 500)}
                    for h in r.json().get("hits", [])
                    if h.get("title")
                ]
        except Exception:
            return []

    tasks = [_brave(), _tavily(), _wikipedia()]
    if focus in ("technical", "code"):
        # Stack Overflow via HN for technical focus
        tasks.append(_hackernews())
    elif focus in ("news", "general"):
        tasks.append(_hackernews())

    all_results: list = []
    for batch in await asyncio.gather(*tasks, return_exceptions=True):
        if isinstance(batch, list):
            all_results.extend(batch)

    # Deduplica per URL, mantieni priorità source
    seen: set = set()
    ranked: list = []
    for r in sorted(all_results, key=lambda x: x.get("score", 0), reverse=True):
        url = r.get("url", "")
        if url and url not in seen:
            seen.add(url)
            ranked.append(r)
        elif not url:
            ranked.append(r)

    ranked = ranked[:max_results]
    return {
        "query": query,
        "focus": focus,
        "total": len(ranked),
        "results": ranked,
        "sources": list({r["source"] for r in ranked}),
    }