tools/registry_macro.py

"""registry_macro.py — Macro tool compositi: bulk-write, fetch, webhook, notion, python_analyze.

Estratto da registry.py per ridurre il file principale.

Funzioni esportate:
  _write_and_check, _bulk_write_files, _fetch_and_extract,
  _read_multiple_files, _trigger_webhook, _notion_rw, _jina_fetch, _python_analyze
"""
from __future__ import annotations
import httpx
import asyncio
import subprocess
import tempfile
import os
import sys
import logging
_logger = logging.getLogger("tools.registry")

# ─── P24-F1: Macro tools — workflow compositi ─────────────────────────────────

async def _write_and_check(
    path: str,
    content: str,
    language: str = "auto",
    run_lint: bool = True,
) -> dict:
    """Macro: write_file → lint_code — scrive e verifica in un unico step."""
    write_res = await _write_file(path=path, content=content)
    lint_res: "dict | None" = None
    if run_lint:
        ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
        if ext in ("py", "js", "ts", "tsx", "jsx", "json", "css", "html"):
            lint_res = await _lint_code_tool(content=content, language=language, path=path)
    has_errors = bool(lint_res and (lint_res.get("errors") or lint_res.get("error")))
    return {
        "ok": True,
        "path": path,
        "written": True,
        "bytes_written": len(content.encode("utf-8")),
        "lint": lint_res,
        "has_errors": has_errors,
        "message": (
            f"Scritto {path} ({len(content)} chars)"
            + (" — lint OK" if lint_res and not has_errors else " — errori lint" if has_errors else "")
        ),
    }


async def _bulk_write_files(
    files: dict,
    stop_on_error: bool = False,
    run_lint: bool = False,
) -> dict:
    """Macro: write_file x N in parallelo — scrive più file in un solo step."""
    if not isinstance(files, dict) or not files:
        return {"ok": False, "error": "files deve essere un dict {path: content} non vuoto"}

    async def _single(p: str, c: str) -> "tuple[str, bool, str]": # type: ignore[type-arg]
        try:
            await _write_file(path=p, content=str(c))
            if run_lint:
                ext = p.rsplit(".", 1)[-1].lower() if "." in p else ""
                if ext in ("py", "js", "ts", "tsx", "jsx", "json"):
                    lr = await _lint_code_tool(content=str(c), path=p)
                    if lr.get("errors"):
                        return p, False, f"lint errors: {str(lr['errors'])[:200]}"
            return p, True, ""
        except Exception as exc:  # noqa: BLE001
            return p, False, str(exc)[:200]

    results = await asyncio.gather(*[_single(p, c) for p, c in files.items()])
    written = [p for p, ok, _ in results if ok]
    failed  = [{"path": p, "error": e} for p, ok, e in results if not ok]
    return {
        "ok": len(failed) == 0,
        "written": written,
        "failed": failed,
        "total": len(files),
        "written_count": len(written),
        "message": f"Scritti {len(written)}/{len(files)} file" + (f" — {len(failed)} errori" if failed else ""),
    }


async def _fetch_and_extract(url: str, extract: str = "all") -> dict:
    """Macro: read_page → estrazione strutturata (title, testo, word_count, links)."""
    import re as _re
    page = await _read_page(url=url, query="")
    raw_text = page.get("text") or page.get("content") or ""
    title    = page.get("title", "")
    paras    = [p.strip() for p in raw_text.split("\n") if len(p.strip()) > 60]
    result: dict = {
        "ok": page.get("ok", True),
        "url": url,
        "title": title,
        "word_count": len(raw_text.split()),
        "char_count": len(raw_text),
    }
    if page.get("error"):
        result["error"] = page["error"]
    if extract in ("all", "text", "summary"):
        result["paragraphs"] = paras[:8]
        result["excerpt"]    = raw_text[:2500]
    if extract in ("all", "links"):
        result["links"] = list(dict.fromkeys(_re.findall(r'https?://[^\s"<>]{10,}', raw_text)))[:30]
    return result


async def _read_multiple_files(paths: list, max_chars_per_file: int = 4000) -> dict:
    """Macro: read_file x N in parallelo — legge più file e aggrega i contenuti."""
    if not paths:
        return {"ok": False, "error": "paths non può essere vuoto"}

    async def _single(path: str) -> "tuple[str, str | None, str | None]": # type: ignore[type-arg]
        try:
            res = await _read_file(path=path)
            content = (res.get("content") or res.get("text") or "")[:max_chars_per_file]
            return path, content, None
        except Exception as exc:  # noqa: BLE001
            return path, None, str(exc)[:200]

    results = await asyncio.gather(*[_single(p) for p in paths])
    files_out  = {p: c for p, c, e in results if c is not None}
    errors_out = {p: e for p, c, e in results if e is not None}
    return {
        "ok": len(errors_out) == 0,
        "files": files_out,
        "errors": errors_out,
        "read_count": len(files_out),
        "total": len(paths),
    }


async def _trigger_webhook(
    url: str,
    payload: "dict | str | None" = None,
    method: str = "POST",
    headers: "dict | None" = None,
    timeout: float = 10.0,
) -> dict:
    """Wrapper — delega all'implementazione in tools/trigger_webhook.py."""
    try:
        from tools.trigger_webhook import trigger_webhook as _tw
        return await _tw(url=url, payload=payload, method=method, headers=headers, timeout=timeout)
    except ImportError:
        import httpx as _hx
        body = None
        _hdrs: dict = {"User-Agent": "agente-ai/1.0"}
        if headers:
            _hdrs.update(headers)
        if payload is not None:
            import json as _j
            body = _j.dumps(payload).encode() if isinstance(payload, dict) else str(payload).encode()
            _hdrs.setdefault("Content-Type", "application/json")
        async with _hx.AsyncClient(timeout=min(float(timeout), 10.0), follow_redirects=True) as _c:
            _m = method.upper()
            if _m == "GET":
                r = await _c.get(url, headers=_hdrs)
            elif _m == "PUT":
                r = await _c.put(url, content=body, headers=_hdrs)
            else:
                r = await _c.post(url, content=body, headers=_hdrs)
        return {"ok": r.is_success, "status": r.status_code, "body": r.text[:2000], "error": None}


# ─── P24-F2: notion_rw wrapper ────────────────────────────────────────────────
async def _notion_rw(
    action: str,
    query: str = "",
    page_id: str = "",
    parent_id: str = "",
    title: str = "",
    content: str = "",
    max_results: int = 5,
) -> dict:
    """Wrapper — delega all'implementazione in tools/notion_tool.py."""
    from tools.notion_tool import notion_rw as _nrw  # noqa: PLC0415
    return await _nrw(
        action=action, query=query, page_id=page_id,
        parent_id=parent_id, title=title, content=content,
        max_results=max_results,
    )


# ─── P24-F3: jina_fetch wrapper ────────────────────────────────────────────────
async def _jina_fetch(
    url: str,
    query: str           = "",
    target_selector: str = "",
    remove_selector: str = "",
    max_length: int      = 12000,
) -> dict:
    """Wrapper — delega a tools/jina_reader.py (Jina Reader per SPA e siti JS)."""
    from tools.jina_reader import jina_fetch as _jr  # noqa: PLC0415
    return await _jr(
        url=url, query=query,
        target_selector=target_selector,
        remove_selector=remove_selector,
        max_length=max_length,
    )

async def _python_analyze(code: str = "", content: str = "", filename: str = "<string>") -> dict:
    """P30-B1: Analisi statica Python in-process — zero exec_engine, zero deps esterne.

    Usa ast.parse() + visitor per:
    - Syntax check con posizione esatta (riga, colonna, testo)
    - Metriche strutturali: funzioni/classi/imports/nesting/righe
    - Suggerimenti actionable (funzioni troppo lunghe, nesting alto, ecc.)

    Zero I/O, zero network. Tipicamente <5ms.

    GAP-1: accetta sia 'code' che 'content' — alias per compatibilità frontend.
    Il frontend (toolDefsCode.ts) invia 'content', il backend usava solo 'code'.
    """
    # GAP-1: alias — frontend può inviare 'content' invece di 'code'
    code = code or content
    import ast as _ast

    result: dict = {"syntax_ok": False, "errors": [], "complexity": {}, "suggestions": [], "summary": ""}

    if not code or not code.strip():
        result["errors"] = [{"type": "EmptyCode", "message": "Nessun codice fornito (parametro 'code' o 'content' richiesto)", "line": 0}]
        result["summary"] = "Codice vuoto"
        return result

    # 1. Syntax check
    try:
        tree = _ast.parse(code, filename=filename)
        result["syntax_ok"] = True
    except SyntaxError as _e:
        result["errors"] = [{
            "type": "SyntaxError", "message": str(_e.msg or _e),
            "line": _e.lineno or 0, "col": _e.offset or 0,
            "text": (_e.text or "").rstrip(),
        }]
        result["summary"] = f"SyntaxError alla riga {_e.lineno}: {_e.msg}"
        return result
    except IndentationError as _e:
        result["errors"] = [{
            "type": "IndentationError", "message": str(_e.msg or _e), "line": _e.lineno or 0,
        }]
        result["summary"] = f"IndentationError alla riga {_e.lineno}"
        return result

    # 2. AST visitor per metriche
    class _V(_ast.NodeVisitor):
        def __init__(self):
            self.fns: list[dict] = []
            self.classes: list[dict] = []
            self.imports: list[str] = []
            self.max_nesting = 0
            self._depth = 0
        def _ent(self): self._depth += 1; self.max_nesting = max(self.max_nesting, self._depth)
        def _ex(self):  self._depth -= 1
        def visit_FunctionDef(self, n):
            _ln = (n.end_lineno or n.lineno) - n.lineno + 1
            self.fns.append({"name": n.name, "line": n.lineno, "lines": _ln})
            self._ent(); self.generic_visit(n); self._ex()
        visit_AsyncFunctionDef = visit_FunctionDef
        def visit_ClassDef(self, n):
            self.classes.append({"name": n.name, "line": n.lineno})
            self._ent(); self.generic_visit(n); self._ex()
        def visit_For(self, n):   self._ent(); self.generic_visit(n); self._ex()
        def visit_While(self, n): self._ent(); self.generic_visit(n); self._ex()
        def visit_If(self, n):    self._ent(); self.generic_visit(n); self._ex()
        def visit_With(self, n):  self._ent(); self.generic_visit(n); self._ex()
        def visit_Try(self, n):   self._ent(); self.generic_visit(n); self._ex()
        def visit_Import(self, n):
            for _a in n.names: self.imports.append(_a.name)
        def visit_ImportFrom(self, n):
            if n.module: self.imports.append(n.module)

    _v = _V(); _v.visit(tree)
    _tot = len(code.splitlines())
    result["complexity"] = {
        "total_lines": _tot,
        "functions": len(_v.fns),
        "classes": len(_v.classes),
        "imports": _v.imports[:20],
        "max_nesting": _v.max_nesting,
    }

    # 3. Suggerimenti actionable
    _sug: list[str] = []
    for _f in _v.fns:
        if _f["lines"] > 50:
            _sug.append(f"Funzione '{_f['name']}' (riga {_f['line']}) ha {_f['lines']} righe — valuta di dividerla")
    if _v.max_nesting >= 5:
        _sug.append(f"Nesting max {_v.max_nesting} livelli — rischio complessità ciclomatica alta")
    if not _v.fns and _tot > 30:
        _sug.append("Nessuna funzione definita su codice lungo — struttura in funzioni")
    if "import *" in code:
        _sug.append("Evita 'import *' — importa esplicitamente solo ciò che serve")
    _dup_imports = {_i for _i in _v.imports if _v.imports.count(_i) > 1}
    if _dup_imports:
        _sug.append(f"Import duplicati rilevati: {', '.join(sorted(_dup_imports))}")
    result["suggestions"] = _sug[:5]

    # 4. Summary
    _parts = [f"OK — {_tot} righe"]
    if _v.fns:    _parts.append(f"{len(_v.fns)} funzioni")
    if _v.classes:_parts.append(f"{len(_v.classes)} classi")
    if _v.imports:_parts.append(f"{len(_v.imports)} import")
    if _v.max_nesting: _parts.append(f"nesting max {_v.max_nesting}")
    result["summary"] = "Sintassi " + ", ".join(_parts)
    return result