tools/registry_web.py

"""registry_web.py — Tool web: search, meteo, calcolo, Python, immagini, browser.

Estratto da registry.py per ridurre il file principale.

Funzioni esportate:
  _web_search, _read_page, _get_weather, _calculate, _run_python,
  _generate_image, _browser_navigate, _browser_session_open,
  _browser_session_act, _browser_session_close, _get_news
"""
from __future__ import annotations
import httpx
import asyncio
import subprocess
import tempfile
import os
import sys
import logging
_logger = logging.getLogger("tools.registry")

async def _web_search(query: str, max_results: int = 5) -> dict:
    """
    S357: Parallelismo completo — tutti e 4 i provider lanciati con asyncio.gather.
    Worst case: 10s (timeout singolo provider) invece di 40s+ (4 × 10s sequenziali).
    Merge con deduplicazione per URL, priorità Brave > Tavily > Wikipedia > DDG.
    """
    import re as _re, html as _html, 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 [
                        # S602: snippet 300→500 — Brave description[:300] parity con Tavily
                        {"title": it["title"], "snippet": (it.get("description") or "")[:500],
                         "url": it["url"], "source": "Brave"}
                        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("[registry] 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 [
                        # S600: snippet 300→500 — Tavily restituisce snippet più ricchi
                    {"title": it["title"], "snippet": (it.get("content") or "")[:500],
                         "url": it["url"], "source": "Tavily"}
                        for it in r.json().get("results", [])[:max_results]
                        if it.get("title") and it.get("url")
                    ]
        except Exception as _exc:
            _logger.debug("[registry] 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",
            })
            wiki_req = _urlreq.Request(
                f"https://en.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":   item.get("title", ""),
                    "url_trunc": ("https://en.wikipedia.org/wiki/" + item.get("title", "")).replace(" ", "_"),  # S600: snippet 300→500 — Wikipedia snippet può essere più lungo
                    # S600: snippet 300→500 — Wikipedia snippet può essere più lungo
                    "snippet": _html.unescape(_re.sub(r"<[^>]+>", "", item.get("snippet", "")))[:500],
                    "source":  "Wikipedia",
                }
                for item in wiki_data.get("query", {}).get("search", [])[:max_results]
            ]
        except Exception:
            return []

    async def _ddg() -> list:
        try:
            ddg_qs  = _urlparse.urlencode({"q": query, "kl": "it-it"})
            ddg_req = _urlreq.Request(
                f"https://html.duckduckgo.com/html/?{ddg_qs}",
                headers={
                    "User-Agent": "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Safari/604.1",
                    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                    "Accept-Language": "it-IT,it;q=0.9,en;q=0.8",
                },
            )
            raw = await asyncio.to_thread(
                lambda: _urlreq.urlopen(ddg_req, timeout=10).read().decode("utf-8", errors="replace")
            )
            links    = _re.findall(r'class="result__a"[^>]+href="([^"]+)"[^>]*>(.*?)</a>', raw, _re.S)
            snippets = _re.findall(r'class="result__snippet"[^>]*>(.*?)</a>', raw, _re.S)
            out = []
            for i, (href, title_html) in enumerate(links[:max_results]):
                title   = _html.unescape(_re.sub(r"<[^>]+>", "", title_html)).strip()
                snippet = _html.unescape(_re.sub(r"<[^>]+>", "", snippets[i] if i < len(snippets) else "")).strip()
                if title and href.startswith("http"):
                    # S600: snippet 300→500 — DDG snippet spesso viene troncato a 300
                    out.append({"title": title, "url": href, "snippet": snippet[:500], "source": "DDG"})  # S600
            return out
        except Exception:
            return []

    async def _jina() -> list:
        """S378: Jina Search — gratuito, no API key, fallback affidabile."""
        try:
            import urllib.parse as _up
            jina_url = f"https://s.jina.ai/?q={_up.quote(query)}"
            async with httpx.AsyncClient(timeout=10) as c:
                r = await c.get(
                    jina_url,
                    headers={**_headers, "Accept": "application/json", "X-No-Cache": "true"},
                )
                if r.status_code == 200:
                    data = r.json().get("data", [])
                    return [
                        {
                            "title":   item.get("title", ""),
                            "url":     item.get("url", ""),
                            # S602: snippet 300→500 — Jina description/content parity con altri provider
                            "snippet": (item.get("description") or item.get("content") or "")[:500],
                            "source":  "Jina",
                        }
                        for item in data[:max_results]
                        if item.get("title") and item.get("url")
                    ]
        except Exception as _exc:
            _logger.debug("[registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    async def _hackernews() -> list:
        """S763: HackerNews via Algolia — gratuito, no API key, qualità alta su query tech."""
        try:
            async with httpx.AsyncClient(timeout=8) as c:
                r = await c.get(
                    "https://hn.algolia.com/api/v1/search",
                    params={"query": query, "hitsPerPage": min(max_results, 4), "tags": "story"},
                )
                if r.status_code == 200:
                    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"}
                        for h in r.json().get("hits", [])
                        if h.get("title")
                    ]
        except Exception as _exc:
            _logger.debug("[registry] silenced %s", type(_exc).__name__)  # noqa: BLE001
        return []

    # S357: lancio parallelo — worst case = max timeout singolo (10s), non 4×10s=40s
    # S378: Jina aggiunto come 5° provider (gratuito, no API key)
    # GAP3-fix: HackerNews aggiunto come 6° provider (tech quality, da web_search.py ora integrato)
    _gather_results = await asyncio.gather(
        _brave(), _tavily(), _wikipedia(), _ddg(), _jina(), _hackernews(),
        return_exceptions=True,
    )
    all_lists = [r for r in _gather_results if not isinstance(r, BaseException)]

    # Merge con deduplicazione per URL (priorità: Brave > Tavily > Wikipedia > DDG > Jina)
    seen_urls: set = set()
    results: list = []
    for res_list in all_lists:
        for item in (res_list if isinstance(res_list, list) else []):
            url = item.get("url", "")
            if url and url not in seen_urls:
                seen_urls.add(url)
                results.append(item)
                if len(results) >= max_results:
                    break
        if len(results) >= max_results:
            break

    return {"query": query, "results": results[:max_results]}

async def _read_page(url: str, query: str = "") -> dict:
    """S763: upgrade — usa extract_with_trafilatura (Readability-quality).
    Fallback a regex strip se trafilatura non disponibile.
    query opzionale per filtrare paragrafi rilevanti.
    """
    try:
        async with httpx.AsyncClient(timeout=15, follow_redirects=True) as c:
            r = await c.get(
                url,
                headers={"User-Agent": "Mozilla/5.0 (compatible; AgentBot/1.0)"},
            )
            try:
                from tools.content_cleaner import extract_with_trafilatura
                result = extract_with_trafilatura(
                    html=r.text, url=url, query=query or None, max_chars=5000,
                )
                return {
                    "url":       url,
                    "content":   result["content"],
                    "status":    r.status_code,
                    "extractor": result.get("extractor", "trafilatura"),
                    "chars":     result.get("chars", 0),
                }
            except ImportError:
                import re as _re
                _c = _re.sub(r"<[^>]+>", " ", r.text)
                _c = _re.sub(r"\s+", " ", _c).strip()
                return {"url": url, "content": _c[:5000], "status": r.status_code, "extractor": "regex"}
    except Exception as e:
        return {"url": url, "content": "", "error": str(e)[:300]}

async def _get_weather(city: str) -> dict:
    try:
        async with httpx.AsyncClient(timeout=8) as c:
            # S390-B-H: usa params= invece di f-string per URL encoding corretto.
            # f-string non codifica spazi/accenti → API restituisce 0 risultati per "New York", "Reggio Emilia", ecc.
            geo = await c.get(
                "https://geocoding-api.open-meteo.com/v1/search",
                params={"name": city, "count": 1, "language": "it", "format": "json"},
            )
            results = geo.json().get("results", [])
            if not results:
                return {"error": f"Città '{city}' non trovata"}
            loc = results[0]
            weather = await c.get(
                f"https://api.open-meteo.com/v1/forecast?latitude={loc['latitude']}&longitude={loc['longitude']}"
                f"&current=temperature_2m,weather_code,wind_speed_10m&timezone=auto"
            )
            w = weather.json().get("current", {})
            return {"city": loc["name"], "country": loc.get("country", ""), "temp_c": w.get("temperature_2m"), "wind_kmh": w.get("wind_speed_10m"), "code": w.get("weather_code")}
    except Exception as e:
        return {"error": str(e)}

async def _calculate(expression: str) -> dict:
    try:
        import ast, operator
        # S390-B-M: aggiunto Mod (%) e FloorDiv (//) agli operator consentiti
        allowed = {ast.Add: operator.add, ast.Sub: operator.sub, ast.Mult: operator.mul, ast.Div: operator.truediv, ast.Pow: operator.pow, ast.USub: operator.neg, ast.Mod: operator.mod, ast.FloorDiv: operator.floordiv}
        def eval_node(node):
            if isinstance(node, ast.Constant): return node.value
            if isinstance(node, ast.BinOp): return allowed[type(node.op)](eval_node(node.left), eval_node(node.right))
            if isinstance(node, ast.UnaryOp): return allowed[type(node.op)](eval_node(node.operand))
            raise ValueError("Operazione non supportata")
        tree = ast.parse(expression, mode='eval')
        result = eval_node(tree.body)
        return {"expression": expression, "result": result}
    except Exception as e:
        return {"expression": expression, "error": str(e)}

async def _run_python(code: str) -> dict:
    """S574/S749: run_python con backend-exec microservice (sandbox persistente per sessione).

    Priorità:
    1. backend-exec (EXEC_ENGINE_URL configurato) — sandbox persistente, pip install per sessione,
       resource limits 512 MB, TTL cleaner. Usa AGENT_SESSION_ID env come session_id.
    2. Fallback: exec_sandbox locale (S574) — tmpdir effimera, comportamento pre-S749.

    Vantaggio chiave: se lo step precedente ha installato pandas, questo step lo trova.
    """
    # S749-D: session_id da ContextVar (impostata da unified_loop per task isolation)
    _session_id = _agent_session_id_var.get()
    _remote = await _call_exec_engine({
        "code": code,
        "language": "python",
        "timeout": 15,
        "session_id": _session_id,
    })
    if _remote is not None:
        # Normalizza formato: backend-exec → {exit_code, stdout, stderr}
        #                      exec_sandbox → {returncode, stdout, stderr}
        return {
            "returncode": _remote.get("exit_code", -1),
            "stdout":     _remote.get("stdout", ""),
            "stderr":     _remote.get("stderr", ""),
        }
    # Fallback locale (S574)
    from api.exec_sandbox import run_in_sandbox_async
    return await run_in_sandbox_async(code, lang="python",
                                      task_id="tool_run_python", timeout=15.0)  # S574: task isolamento per run_python



async def _generate_image(prompt: str, width: int = 512, height: int = 512) -> dict:
    """
    Genera immagine AI: FLUX.1-schnell via backend /api/vision/generate (HF Space, gratuito),
    fallback Pollinations AI se FLUX non disponibile.
    """
    import urllib.parse

    # V001: Prova prima il backend interno FLUX (stesso processo, localhost)
    _base_url = os.environ.get("BACKEND_BASE_URL", "http://localhost:7860")
    _token    = os.environ.get("INTERNAL_TOKEN", "")
    try:
        async with httpx.AsyncClient(timeout=30.0) as c:
            _r = await c.post(
                f"{_base_url}/api/vision/generate",
                json={"prompt": prompt.strip()[:600], "width": width, "height": height},
                headers={**({"X-Internal-Token": _token} if _token else {})},
            )
            if _r.status_code == 200:
                _data = _r.json()
                if _data.get("ok") and _data.get("image_url"):
                    return {
                        "url":    _data["image_url"],
                        "prompt": prompt,
                        "width":  width,
                        "height": height,
                        "ready":  True,
                        "source": "flux",
                        "note":   "Immagine generata con FLUX.1-schnell via backend.",
                    }
    except Exception:
        pass  # FLUX non raggiungibile → Pollinations fallback

    # Fallback: Pollinations AI (gratuito, nessuna API key)
    encoded = urllib.parse.quote(prompt.strip(), safe="")
    seed    = sum(ord(c) for c in prompt) % 9999 + 1
    url = (
        f"https://image.pollinations.ai/prompt/{encoded}"
        f"?width={width}&height={height}&seed={seed}&nologo=true&enhance=true"
    )
    return {
        "url":    url,
        "prompt": prompt,
        "width":  width,
        "height": height,
        "ready":  True,
        "source": "pollinations",
        "note":   "Copia l\'URL nel browser o incollalo in un tag <img> per vedere l\'immagine.",
    }

async def _browser_navigate(url: str, wait_ms: int = 2000, mobile: bool = False) -> dict:
    """
    S403 — Limite 1 (criticità 10/10): Browser execution layer.
    Naviga a una URL con Playwright headless, restituisce titolo + testo + DOM links/inputs.
    Stateless (nessuna sessione persistente) — usa per lettura/scraping rapido.
    Playwright è già installato nel backend (browser.py v5 S65).
    """
    try:
        from api.browser import (
            _safe_url, _LAUNCH_ARGS, _make_context, _DOM_SCRIPT,
            GOTO_TIMEOUT, MAX_LINKS, MAX_INPUTS, MAX_TEXT
        )
        if not _safe_url(url):
            return {"error": "URL non consentita (localhost/intranet bloccato per sicurezza)"}
        from playwright.async_api import async_playwright
        async with async_playwright() as pw:
            browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS)
            ctx     = await _make_context(browser, 1280, 800, mobile)
            page    = await ctx.new_page()
            try:
                await page.goto(url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT)
                await page.wait_for_timeout(wait_ms)
                title   = await page.title()
                dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
                text    = (dom_raw.get("text") or "")[:2000] if isinstance(dom_raw, dict) else ""
                links   = (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else []
                inputs  = (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else []
                return {
                    "url":    page.url,
                    "title":  title,
                    "text":   text,
                    "links":  links,
                    "inputs": inputs,
                }
            except Exception as e:
                # S600: 300→500 — browser inner exception può contenere stack/path
                return {"url": url, "error": str(e)[:500]}
            finally:
                await ctx.close()
                await browser.close()
    except ImportError:
        return {"error": "Playwright non disponibile — usa read_page come alternativa"}
    except Exception as e:
        # S600: 300→500 — outer exception
        return {"error": str(e)[:500]}


async def _browser_session_open(url: str, wait_ms: int = 1500, mobile: bool = False) -> dict:
    """
    S403 — Apre sessione Playwright persistente (stateful).
    Restituisce session_id da usare con browser_session_act e browser_session_close.
    Max 2 sessioni simultanee (OOM guard HF free tier).
    Usa questa modalità per task multi-step: login → naviga → compila → invia.
    """
    try:
        from api.browser import (
            _safe_url, _LAUNCH_ARGS, _make_context, _DOM_SCRIPT,
            _sessions, SESSION_LIMIT, GOTO_TIMEOUT, MAX_LINKS, MAX_INPUTS, MAX_TEXT,
            _close_session
        )
        import uuid, time
        if not _safe_url(url):
            return {"error": "URL non consentita"}
        if len(_sessions) >= SESSION_LIMIT:
            oldest = min(_sessions, key=lambda sid: _sessions[sid]["last_used"])
            await _close_session(oldest, "OOM guard from tool")
        from playwright.async_api import async_playwright
        pw      = await async_playwright().start()
        browser = await pw.chromium.launch(headless=True, args=_LAUNCH_ARGS)
        ctx     = await _make_context(browser, 1280, 800, mobile)
        page    = await ctx.new_page()
        await page.goto(url, wait_until="domcontentloaded", timeout=GOTO_TIMEOUT)
        await page.wait_for_timeout(wait_ms)
        title   = await page.title()
        dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
        sid     = uuid.uuid4().hex[:16]
        _sessions[sid] = {
            "pw": pw, "browser": browser, "context": ctx, "page": page,
            "created_at": time.time(), "last_used": time.time(), "url": page.url,
        }
        return {
            "session_id": sid,
            "url":        page.url,
            "title":      title,
            "text":       (dom_raw.get("text") or "")[:1500] if isinstance(dom_raw, dict) else "",
            "links":      (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else [],
            "inputs":     (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else [],
        }
    except ImportError:
        return {"error": "Playwright non disponibile"}
    except Exception as e:
        # S600: 300→500 — registry browser session exception
        return {"error": str(e)[:500]}


async def _browser_session_act(session_id: str, actions: list, wait_ms: int = 1000) -> dict:
    """
    S403 — Esegue azioni su sessione Playwright esistente.
    actions: lista di {type, selector?, value?, key?, ms?}
    Tipi supportati: click, fill, select, press, hover, wait_for, wait, scroll.
    Restituisce DOM aggiornato (titolo + testo + links + inputs).
    """
    try:
        from api.browser import (
            _sessions, _execute_actions, _DOM_SCRIPT,
            MAX_LINKS, MAX_INPUTS, MAX_TEXT
        )
        import time
        sess = _sessions.get(session_id)
        if not sess:
            return {"error": f"Sessione {session_id} non trovata o scaduta (TTL 8 min)"}
        sess["last_used"] = time.time()
        page = sess["page"]

        # Converti lista dict → oggetti con attributo .type, .selector, ecc.
        class _A:
            def __init__(self, d: dict):
                self.type     = d.get("type", "")
                self.selector = d.get("selector")
                self.value    = d.get("value")
                self.key      = d.get("key")
                self.ms       = d.get("ms")

        await _execute_actions(page, [_A(a) for a in (actions or [])])
        await page.wait_for_timeout(wait_ms)
        sess["url"] = page.url
        title   = await page.title()
        dom_raw = await page.evaluate(_DOM_SCRIPT % (MAX_LINKS, MAX_INPUTS, MAX_TEXT))
        return {
            "session_id": session_id,
            "url":        page.url,
            "title":      title,
            "text":       (dom_raw.get("text") or "")[:1500] if isinstance(dom_raw, dict) else "",
            "links":      (dom_raw.get("links") or [])[:15] if isinstance(dom_raw, dict) else [],
            "inputs":     (dom_raw.get("inputs") or [])[:10] if isinstance(dom_raw, dict) else [],
        }
    except ImportError:
        return {"error": "Playwright non disponibile"}
    except Exception as e:
        # S601: 300→500 — parity con altri session handler
        return {"error": str(e)[:500]}


async def _browser_session_close(session_id: str) -> dict:
    """S403 — Chiude sessione Playwright e libera memoria (~300 MB/sessione)."""
    try:
        from api.browser import _sessions, _close_session
        if session_id not in _sessions:
            return {"ok": True, "note": "Sessione già chiusa o non trovata"}
        await _close_session(session_id, "tool call")
        return {"ok": True, "session_id": session_id}
    except Exception as e:
        return {"ok": False, "error": str(e)[:300]}  # S589: 200→300


async def _get_news(query: str, max_results: int = 5) -> dict:
    """S378: get_news — alias di web_search con query ottimizzata per notizie recenti.
    smolagents chiama questo tool per richieste di tipo 'notizie su X'.
    """
    news_query = f"notizie recenti {query}" if not any(
        kw in query.lower() for kw in ("notizie", "news", "ultime", "latest", "breaking")
    ) else query
    return await _web_search(news_query, max_results=max_results)