tools/python_analyze.py

"""
python_analyze.py — Analisi statica Python in sandbox sicura (AST-only, zero exec).

Funzionalità:
  • Syntax check via ast.parse() — cattura SyntaxError/IndentationError
  • Metriche: LOC, SLOC, blank, comment_lines, functions, classes, imports
  • Cyclomatic complexity per funzione (McCabe approximato via ast.NodeVisitor)
  • Max nesting depth (annidamento if/for/while/with/try)
  • Long lines (> max_line_len chars)
  • Duplicate lines detection (righe di codice identiche ripetute ≥ 2 volte)
  • Suggerimenti automatici basati sulle metriche

Sicurezza:
  • Nessun exec(), eval(), compile() a runtime
  • ast.parse() è safe — nessun effetto collaterale
  • Timeout interno 5s via asyncio.wait_for (per codice enorme)
  • Risultato max 8 KB per non saturare il contesto LLM
"""

from __future__ import annotations

import ast
import asyncio
import re
import textwrap
from typing import Any


# ─── Cyclomatic Complexity Visitor ───────────────────────────────────────────

class _ComplexityVisitor(ast.NodeVisitor):
    """McCabe cyclomatic complexity approssimata.

    Conta i branch decisionali (if/elif/for/while/except/with/assert/and/or/?).
    Base = 1 per ogni funzione/metodo.
    """

    def __init__(self) -> None:
        self.complexity: int = 1  # base

    def visit_If(self, node: ast.If) -> None:  # noqa: N802
        self.complexity += 1
        for orelse in node.orelse:
            if isinstance(orelse, ast.If):
                self.complexity += 1  # elif conta come branch aggiuntivo
        self.generic_visit(node)

    def visit_For(self, node: ast.For) -> None:  # noqa: N802
        self.complexity += 1
        self.generic_visit(node)

    def visit_While(self, node: ast.While) -> None:  # noqa: N802
        self.complexity += 1
        self.generic_visit(node)

    def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:  # noqa: N802
        self.complexity += 1
        self.generic_visit(node)

    def visit_With(self, node: ast.With) -> None:  # noqa: N802
        self.complexity += 1
        self.generic_visit(node)

    def visit_Assert(self, node: ast.Assert) -> None:  # noqa: N802
        self.complexity += 1
        self.generic_visit(node)

    def visit_BoolOp(self, node: ast.BoolOp) -> None:  # noqa: N802
        # and/or aggiungono un branch per ogni operando extra
        self.complexity += len(node.values) - 1
        self.generic_visit(node)

    def visit_IfExp(self, node: ast.IfExp) -> None:  # noqa: N802
        # x if cond else y
        self.complexity += 1
        self.generic_visit(node)


# ─── Nesting Depth Visitor ────────────────────────────────────────────────────

_NESTING_NODES = (ast.If, ast.For, ast.While, ast.With, ast.Try,
                  ast.AsyncFor, ast.AsyncWith)


def _max_depth(node: ast.AST, current: int = 0) -> int:
    """Calcola il massimo annidamento di blocchi decisionali/iterativi."""
    max_d = current
    for child in ast.iter_child_nodes(node):
        if isinstance(child, _NESTING_NODES):
            d = _max_depth(child, current + 1)
        else:
            d = _max_depth(child, current)
        if d > max_d:
            max_d = d
    return max_d


# ─── Core analysis ───────────────────────────────────────────────────────────

def _analyze_sync(
    code: str,
    filename: str,
    check_style: bool,
    max_complexity: int,
    max_line_len: int,
) -> dict[str, Any]:
    """Analisi sincrona — wrappata in asyncio.wait_for per timeout."""

    result: dict[str, Any] = {
        "syntax_ok": False,
        "errors": [],
        "warnings": [],
        "metrics": {},
        "functions": [],
        "suggestions": [],
    }

    lines = code.splitlines()
    total_loc = len(lines)

    # ── Metriche base ────────────────────────────────────────────────────────
    blank_lines   = sum(1 for l in lines if not l.strip())
    comment_lines = sum(1 for l in lines if l.strip().startswith("#"))
    sloc          = total_loc - blank_lines - comment_lines

    result["metrics"] = {
        "loc":           total_loc,
        "sloc":          sloc,
        "blank":         blank_lines,
        "comment_lines": comment_lines,
    }

    # ── Syntax check ─────────────────────────────────────────────────────────
    try:
        tree = ast.parse(code, filename=filename)
    except SyntaxError as exc:
        result["errors"].append({
            "type":    "SyntaxError",
            "message": str(exc.msg),
            "line":    exc.lineno,
            "col":     exc.offset,
            "text":    (exc.text or "").rstrip(),
        })
        return result  # inutile continuare senza AST
    except IndentationError as exc:
        result["errors"].append({
            "type":    "IndentationError",
            "message": str(exc.msg),
            "line":    exc.lineno,
            "col":     exc.offset,
            "text":    (exc.text or "").rstrip(),
        })
        return result

    result["syntax_ok"] = True

    # ── Imports ───────────────────────────────────────────────────────────────
    imports: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, ast.Import):
            for alias in node.names:
                imports.append(alias.asname or alias.name)
        elif isinstance(node, ast.ImportFrom):
            mod = node.module or ""
            for alias in node.names:
                imports.append(f"{mod}.{alias.asname or alias.name}")

    result["metrics"]["imports"] = len(imports)
    result["metrics"]["import_names"] = imports[:20]  # cap a 20

    # ── Funzioni e classi ─────────────────────────────────────────────────────
    func_nodes  = [n for n in ast.walk(tree) if isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef))]
    class_nodes = [n for n in ast.walk(tree) if isinstance(n, ast.ClassDef)]

    result["metrics"]["functions"] = len(func_nodes)
    result["metrics"]["classes"]   = len(class_nodes)
    result["metrics"]["class_names"] = [c.name for c in class_nodes[:10]]

    # ── Cyclomatic complexity per funzione ────────────────────────────────────
    func_details: list[dict] = []
    max_cc = 0
    for fn in func_nodes:
        visitor = _ComplexityVisitor()
        visitor.visit(fn)
        cc = visitor.complexity
        if cc > max_cc:
            max_cc = cc
        decorators = [ast.unparse(d) for d in fn.decorator_list] if hasattr(ast, "unparse") else []
        args_count = len(fn.args.args) + len(fn.args.posonlyargs) + len(fn.args.kwonlyargs)
        func_details.append({
            "name":        fn.name,
            "line":        fn.lineno,
            "complexity":  cc,
            "args":        args_count,
            "is_async":    isinstance(fn, ast.AsyncFunctionDef),
            "decorators":  decorators[:3],
            "high_cc":     cc > max_complexity,
        })

    result["metrics"]["max_complexity"]  = max_cc
    result["metrics"]["avg_complexity"]  = (
        round(sum(f["complexity"] for f in func_details) / len(func_details), 1)
        if func_details else 0
    )
    result["functions"] = sorted(func_details, key=lambda f: -f["complexity"])[:15]

    # ── Nesting depth ─────────────────────────────────────────────────────────
    nesting = _max_depth(tree)
    result["metrics"]["max_nesting"] = nesting

    # ── Style checks ─────────────────────────────────────────────────────────
    if check_style:
        long_lines = [
            {"line": i + 1, "len": len(l), "text": l[:80] + "…" if len(l) > 80 else l}
            for i, l in enumerate(lines)
            if len(l) > max_line_len
        ]
        result["metrics"]["long_lines"] = len(long_lines)
        if long_lines:
            result["warnings"].extend(
                [{"type": "long_line", "line": ll["line"], "len": ll["len"]} for ll in long_lines[:5]]
            )

        # Bare except
        bare_except_pat = re.compile(r"^\s*except\s*:\s*$")
        for i, l in enumerate(lines):
            if bare_except_pat.match(l):
                result["warnings"].append({
                    "type": "bare_except",
                    "line": i + 1,
                    "text": l.strip(),
                    "fix":  "Usa 'except Exception as e:' o cattura l'eccezione specifica",
                })

        # print() in produzione
        print_pat = re.compile(r"^\s*print\(")
        for i, l in enumerate(lines):
            if print_pat.match(l) and not l.strip().startswith("#"):
                result["warnings"].append({
                    "type":    "debug_print",
                    "line":    i + 1,
                    "text":    l.strip()[:60],
                    "fix":     "Usa logging.getLogger(__name__) al posto di print()",
                })

        # type(x) == — confronta tipo con type() invece di isinstance()
        type_eq_pat = re.compile(r"type\(\w+\)\s*[!=]=")
        for i, l in enumerate(lines):
            if type_eq_pat.search(l):
                result["warnings"].append({
                    "type": "type_compare",
                    "line": i + 1,
                    "text": l.strip()[:60],
                    "fix":  "Usa isinstance() al posto di type() == per supportare sottoclassi",
                })

        # Duplicate lines (non-trivial: >= 20 chars, non solo punteggiatura)
        code_lines = [l.strip() for l in lines if len(l.strip()) >= 20 and not l.strip().startswith("#")]
        seen: dict[str, list[int]] = {}
        for i, l in enumerate(code_lines):
            seen.setdefault(l, []).append(i)
        dupes = {l: idxs for l, idxs in seen.items() if len(idxs) >= 2}
        result["metrics"]["duplicate_lines"] = len(dupes)
        if dupes:
            result["warnings"].append({
                "type":  "duplicate_code",
                "count": len(dupes),
                "examples": [{"line": l[:60]} for l in list(dupes)[:3]],
                "fix":   "Considera di estrarre le righe duplicate in funzioni riutilizzabili",
            })

    # ── Suggerimenti automatici ───────────────────────────────────────────────
    suggestions: list[str] = []
    m = result["metrics"]

    if m.get("max_complexity", 0) > max_complexity:
        high = [f["name"] for f in result["functions"] if f["high_cc"]]
        suggestions.append(
            f"Alta complessità ciclomatica (max {m['max_complexity']}) "
            f"nelle funzioni: {', '.join(high[:5])}. "
            f"Considera di suddividerle in funzioni più piccole (soglia: {max_complexity})."
        )

    if m.get("max_nesting", 0) >= 5:
        suggestions.append(
            f"Annidamento profondo ({m['max_nesting']} livelli). "
            "Usa early-return / guard clauses per appiattire la struttura."
        )

    if m.get("functions", 0) == 0 and m.get("sloc", 0) > 50:
        suggestions.append(
            "Nessuna funzione definita su >50 SLOC. "
            "Considera di modularizzare il codice in funzioni con responsabilità singola."
        )

    if m.get("imports", 0) > 20:
        suggestions.append(
            f"Molte dipendenze ({m['imports']} import). "
            "Verifica che tutti siano necessari e valuta di suddividere il modulo."
        )

    if m.get("duplicate_lines", 0) >= 3:
        suggestions.append(
            f"{m['duplicate_lines']} righe di codice duplicate. "
            "Estrai il codice ripetuto in funzioni helper o costanti."
        )

    if not result["errors"] and not result["warnings"] and not suggestions:
        suggestions.append("✅ Codice OK — nessun problema rilevato dall'analisi statica.")

    result["suggestions"] = suggestions

    # ── Cap output per non saturare il contesto LLM ─────────────────────────
    result_str = str(result)
    if len(result_str) > 8000:
        result["functions"] = result["functions"][:5]
        result["_truncated"] = True

    return result


# ─── Public async entry point ─────────────────────────────────────────────────

async def analyze_python(
    code: str,
    filename: str       = "script.py",
    check_style: bool   = True,
    max_complexity: int = 10,
    max_line_len: int   = 100,
) -> dict[str, Any]:
    """Analizza codice Python in modo sicuro (AST-only, nessun exec).

    Args:
        code:            Codice Python da analizzare.
        filename:        Nome file per i messaggi di errore (default: 'script.py').
        check_style:     Se True, include check di stile (long lines, bare except, print, ecc.).
        max_complexity:  Soglia di complessità ciclomatica per i warning (default: 10).
        max_line_len:    Lunghezza massima riga per i warning (default: 100).

    Returns:
        dict con: syntax_ok, errors[], warnings[], metrics{}, functions[], suggestions[].
    """
    if not isinstance(code, str) or not code.strip():
        return {
            "syntax_ok": False,
            "errors": [{"type": "InputError", "message": "code deve essere una stringa Python non vuota"}],
            "warnings": [], "metrics": {}, "functions": [], "suggestions": [],
        }

    # Normalizza: rimuovi BOM e normalizza newline
    code = code.lstrip("\ufeff").replace("\r\n", "\n").replace("\r", "\n")

    # Timeout 5s — ast.parse() è O(n) ma codice molto grande potrebbe essere lento
    try:
        result = await asyncio.wait_for(
            asyncio.get_event_loop().run_in_executor(
                None,
                lambda: _analyze_sync(code, filename, check_style, max_complexity, max_line_len),
            ),
            timeout=5.0,
        )
    except asyncio.TimeoutError:
        return {
            "syntax_ok": False,
            "errors": [{"type": "TimeoutError", "message": "Analisi timeout (>5s) — codice troppo grande"}],
            "warnings": [], "metrics": {"loc": len(code.splitlines())}, "functions": [], "suggestions": [],
        }

    return result