"""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="([^"]+)"[^>]*>(.*?)', raw, _re.S) snippets = _re.findall(r'class="result__snippet"[^>]*>(.*?)', 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"¤t=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 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)