api/coding.py

"""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})