"""api/coding.py — Code analysis endpoint (S437→P37-PA) S437: CodeAgent Ollama rimosso. Sostituito con python_analyze reale. Analisi statica Python via ast (zero deps) + complessità ciclomatica + metriche. Opzionale: esecuzione in sandbox isolata via exec_sandbox. Endpoints: POST /code/analyze — analisi statica Python (+ opzionale esecuzione sandbox) POST /code/session — stub 501 (rimosso Ollama S437 — usa /api/agent/run) Sprint P37-PA — 2026-06-21 """ from __future__ import annotations import ast import logging import textwrap import time from typing import Optional, List from fastapi import APIRouter from fastapi.responses import JSONResponse from pydantic import BaseModel, field_validator _logger = logging.getLogger("api.coding") router = APIRouter(prefix="/code", tags=["coding"]) # ── Constants ───────────────────────────────────────────────────────────────── _MAX_CODE_LEN = 50_000 # byte max per /code/analyze _MAX_RUN_LINES = 200 # righe max per esecuzione sandbox # ── Request / Response schemas ──────────────────────────────────────────────── class AnalyzeReq(BaseModel): filepath: str = "code.py" content: str goal: str = "" run_code: bool = False # se True → esegue in sandbox dopo analisi @field_validator("content") @classmethod def _check_len(cls, v: str) -> str: if len(v) > _MAX_CODE_LEN: raise ValueError(f"content troppo grande ({len(v)} byte, max {_MAX_CODE_LEN})") return v class SessionReq(BaseModel): goal: str files: List[dict] model: Optional[str] = None # ── AST helpers ─────────────────────────────────────────────────────────────── def _lint_python(code: str) -> dict: """Analisi sintattica via ast.parse — zero dipendenze aggiuntive.""" try: ast.parse(code, mode="exec") return {"ok": True, "errors": []} except SyntaxError as e: return {"ok": False, "errors": [f"SyntaxError riga {e.lineno}: {e.msg} ('{e.text or ''}')".strip()]} except Exception as e: return {"ok": False, "errors": [str(e)[:300]]} def _complexity(tree: ast.AST) -> dict: """ Complessità ciclomatica approssimata: CC = 1 + somma rami decisionali (if/elif, while, for, except, with, assert, and/or booleani). Per funzione e per modulo. """ def _cc_node(node: ast.AST) -> int: """Conta branch points in un sotto-albero.""" count = 0 for n in ast.walk(node): if isinstance(n, (ast.If, ast.While, ast.For, ast.ExceptHandler, ast.With, ast.AsyncFor, ast.AsyncWith)): count += 1 elif isinstance(n, ast.BoolOp) and isinstance(n.op, (ast.And, ast.Or)): count += len(n.values) - 1 elif isinstance(n, ast.Assert): count += 1 return count funcs: list[dict] = [] for node in ast.walk(tree): if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): cc = 1 + _cc_node(node) funcs.append({ "name": node.name, "line": node.lineno, "cc": cc, "risk": "HIGH" if cc > 10 else "MEDIUM" if cc > 5 else "LOW", }) total_cc = 1 + _cc_node(tree) return { "module_cc": total_cc, "functions": sorted(funcs, key=lambda f: -f["cc"])[:20], # top 20 per cc "avg_func_cc": round(sum(f["cc"] for f in funcs) / len(funcs), 1) if funcs else 0, } def _imports(tree: ast.AST) -> list[str]: """Lista moduli importati.""" mods: list[str] = [] for node in ast.walk(tree): if isinstance(node, ast.Import): mods.extend(a.name for a in node.names) elif isinstance(node, ast.ImportFrom) and node.module: mods.append(node.module) return sorted(set(mods)) def _suggestions(lint: dict, cc: dict, code_lines: int) -> list[str]: """Genera suggerimenti leggibili in base all'analisi.""" hints: list[str] = [] if not lint["ok"]: hints.append("🔴 Correggere errori di sintassi prima di procedere.") for f in cc["functions"]: if f["risk"] == "HIGH": hints.append(f"⚠️ Funzione '{f['name']}' (riga {f['line']}): CC={f['cc']} — considera di spezzarla.") elif f["risk"] == "MEDIUM": hints.append(f"📌 Funzione '{f['name']}' (riga {f['line']}): CC={f['cc']} — complessità moderata.") if code_lines > 300: hints.append(f"📏 File lungo ({code_lines} righe) — considera di suddividere in moduli.") if cc["module_cc"] > 20: hints.append(f"🔺 Complessità ciclomatica modulo alta (CC={cc['module_cc']}) — molti rami.") return hints[:8] # max 8 suggerimenti # ── Routes ──────────────────────────────────────────────────────────────────── @router.post("/analyze") async def analyze(req: AnalyzeReq) -> JSONResponse: """ Analisi statica Python: sintassi, complessità ciclomatica, imports, suggerimenti. Se run_code=True: esegue in sandbox isolata via exec_sandbox (max _MAX_RUN_LINES righe). Response JSON: syntax_ok, syntax_errors, complexity, imports, suggestions, metrics, execution (solo se run_code=True e sintassi ok) """ t0 = time.monotonic() code = req.content.strip() code_lines = code.count("\n") + 1 # 1. Analisi sintattica lint = _lint_python(code) # 2. AST analysis (solo se sintassi ok) cc_result: dict = {"module_cc": 0, "functions": [], "avg_func_cc": 0} imp_list: list = [] if lint["ok"]: try: tree = ast.parse(code, mode="exec") cc_result = _complexity(tree) imp_list = _imports(tree) except Exception as e: _logger.warning("AST analysis error: %s", e) suggestions = _suggestions(lint, cc_result, code_lines) result: dict = { "syntax_ok": lint["ok"], "syntax_errors": lint["errors"], "complexity": cc_result, "imports": imp_list, "suggestions": suggestions, "metrics": { "lines": code_lines, "chars": len(code), "functions": len(cc_result["functions"]), }, "elapsed_ms": round((time.monotonic() - t0) * 1000, 1), } # 3. Esecuzione sandbox opzionale if req.run_code and lint["ok"]: if code_lines > _MAX_RUN_LINES: result["execution"] = { "ok": False, "error": f"Codice troppo lungo per esecuzione sandbox ({code_lines} righe, max {_MAX_RUN_LINES}).", } else: try: from api.exec_sandbox import run_in_sandbox_async exec_result = await run_in_sandbox_async( code=code, language="python", timeout_s=10, ) result["execution"] = exec_result except ImportError: result["execution"] = {"ok": False, "error": "exec_sandbox non disponibile in questo ambiente."} except Exception as e: result["execution"] = {"ok": False, "error": str(e)[:300]} _logger.info("analyze %s: syntax=%s cc=%s t=%.0fms", req.filepath, lint["ok"], cc_result["module_cc"], result["elapsed_ms"]) return JSONResponse(content=result) _STUB_MSG = ( "Endpoint sessione rimosso (S437) — usa /api/agent/run con AIClient multi-provider." ) class _SR(BaseModel): goal: str files: list model: str | None = None @router.post("/session") async def session(_req: _SR) -> JSONResponse: return JSONResponse(status_code=501, content={"error": _STUB_MSG})