siftlode/scripts/audit_metrics.py

"""The codebase's health numbers in one deterministic run: size, complexity, tests, governance
weight, comment noise and the footprint of features marked for removal. It REPORTS, it never fails
— what used to be a gate (the size/complexity ratchet) is read here instead, and an epic compares
its before and after.

    python scripts/audit_metrics.py              the static numbers (seconds), as JSON on stdout
    python scripts/audit_metrics.py --speed      plus the speed baseline: frontend/e2e/speed.measure.ts
                                                 against the localdev stack (E2E_PASSWORD must be set;
                                                 the ops wrapper exports it)

The readings are counts of files and lines, so a number moves only when the code does.
"""

import ast
import json
import os
import re
import subprocess
import sys
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
BE_APP = ROOT / "backend" / "app"
BE_TESTS = ROOT / "backend" / "tests"
FE_SRC = ROOT / "frontend" / "src"
BIG_FILE = 800
BE_COMPLEXITY = 10  # ruff's C901 default, the backend ratchet's limit
FE_COMPLEXITY = 20  # the frontend ratchet's limit

EPIC_CODE = re.compile(r"\b(E\d{2}|R\d{1,2})(\s?S\d+)?\b")
NARRATION = re.compile(r"\b(used to|no longer)\b", re.IGNORECASE)


def _is_fe_test(p: Path) -> bool:
    return bool(re.search(r"\.(test|testkit)\.tsx?$", p.name))


def _files(root: Path, pattern: str) -> list[Path]:
    return sorted(p for p in root.rglob(pattern) if "__pycache__" not in p.parts)


def _lines(paths: list[Path]) -> list[tuple[Path, list[str]]]:
    return [(p, p.read_text(encoding="utf-8").splitlines()) for p in paths]


def _comment_lines(lines: list[str], python: bool) -> int:
    """Comment lines; for Python the docstrings count too — this codebase writes its "why" there."""
    if not python:
        return sum(1 for line in lines if line.lstrip().startswith(("//", "/*", "*")))
    hashes = sum(1 for line in lines if line.lstrip().startswith("#"))
    docs = 0
    for node in ast.walk(ast.parse("\n".join(lines))):
        body = getattr(node, "body", None)
        if not isinstance(body, list) or not body:
            continue
        first = body[0]
        if (
            isinstance(first, ast.Expr)
            and isinstance(first.value, ast.Constant)
            and isinstance(first.value.value, str)
        ):
            docs += (first.end_lineno or first.lineno) - first.lineno + 1
    return hashes + docs


def _size(be: list[tuple[Path, list[str]]], fe: list[tuple[Path, list[str]]]) -> dict[str, object]:
    be_tests = _lines(_files(BE_TESTS, "*.py"))
    fe_tests = _lines([p for p in _files(FE_SRC, "*.ts*") if _is_fe_test(p)])
    be_n = sum(len(ls) for _, ls in be)
    fe_n = sum(len(ls) for _, ls in fe)
    return {
        "backend_lines": be_n,
        "frontend_lines": fe_n,
        "backend_test_ratio": round(sum(len(ls) for _, ls in be_tests) / be_n, 2),
        "frontend_test_ratio": round(sum(len(ls) for _, ls in fe_tests) / fe_n, 2),
        "files_over_800": sum(1 for _, ls in be + fe if len(ls) > BIG_FILE),
        "largest_file": max(((len(ls), p) for p, ls in be + fe), key=lambda t: t[0])[1]
        .relative_to(ROOT)
        .as_posix(),
        "tables": len(re.findall(r'__tablename__ = "', (BE_APP / "models.py").read_text("utf-8"))),
    }


def _complexity() -> dict[str, object]:
    ruff = subprocess.run(
        [sys.executable, "-m", "ruff", "check", "--config", "backend/ruff.toml", "--select", "C901"]
        + ["--output-format", "json", "--exit-zero", "backend/app"],
        cwd=ROOT,
        capture_output=True,
        text=True,
        check=True,
    )
    found = [
        int(m) for d in json.loads(ruff.stdout) for m in re.findall(r"\((\d+) > ", d["message"])
    ]
    fe = subprocess.run(
        ["node", "scripts/fe_complexity.mjs", str(FE_COMPLEXITY)],
        cwd=ROOT,
        capture_output=True,
        text=True,
        check=True,
    )
    fe_json = json.loads(fe.stdout)
    return {
        f"backend_functions_over_{BE_COMPLEXITY}": len(found),
        "backend_max_complexity": max(found, default=0),
        f"frontend_functions_over_{FE_COMPLEXITY}": fe_json["over"],
        "frontend_max_complexity": fe_json["max"],
    }


def _noise(be: list[tuple[Path, list[str]]], fe: list[tuple[Path, list[str]]]) -> dict[str, object]:
    every = be + fe
    epic = [(p, ln) for p, ls in every for ln in ls if EPIC_CODE.search(ln)]
    be_c = sum(_comment_lines(ls, True) for _, ls in be)
    fe_c = sum(_comment_lines(ls, False) for _, ls in fe)
    return {
        "epic_code_lines": len(epic),
        "epic_code_files": len({p for p, _ in epic}),
        "history_narration_lines": sum(1 for _, ls in every for ln in ls if NARRATION.search(ln)),
        "backend_comment_share": round(be_c / sum(len(ls) for _, ls in be), 3),
        "frontend_comment_share": round(fe_c / sum(len(ls) for _, ls in fe), 3),
    }


def _governance() -> dict[str, object]:
    eslint = (ROOT / "frontend" / "eslint.config.js").read_text("utf-8")
    check = (ROOT / "scripts" / "check.sh").read_text("utf-8")
    contract = [p for p in _files(FE_SRC, "*.contract.test.ts*")]
    return {
        "gate_lanes": len(re.findall(r'^\s*lane "', check, re.MULTILINE)),
        "local_eslint_rules": len(set(re.findall(r'"local/([a-z0-9-]+)"', eslint))),
        "frontend_contract_tests": len(contract),
        "gates_snapshot_lines": len(
            (FE_SRC / "lib" / "gates.snapshot.json").read_text("utf-8").splitlines()
        ),
        "claude_md_lines": len((ROOT / "CLAUDE.md").read_text("utf-8").splitlines()),
        "gates_md_lines": len((ROOT / "docs" / "gates.md").read_text("utf-8").splitlines()),
    }


def _footprint(
    be: list[tuple[Path, list[str]]], fe: list[tuple[Path, list[str]]]
) -> dict[str, int]:
    """Files that still mention a feature marked for removal: 0 once it is gone."""
    patterns = {
        "demo_files": re.compile(r"\bdemo|isDemo|is_demo", re.IGNORECASE),
        "chat_files": re.compile(r"\bchat\b|\bmessages?\.py|\bChat[A-Z]|message_keys"),
        "access_request_files": re.compile(r"access[ _-]?request|AccessRequest", re.IGNORECASE),
    }
    return {
        k: sum(1 for _, ls in be + fe if any(rx.search(ln) for ln in ls))
        for k, rx in patterns.items()
    }


def _speed() -> dict[str, object]:
    env = {**os.environ, "PW_SPEED": "1"}
    out = ROOT / "frontend" / "e2e" / ".speed" / "latest.json"
    out.unlink(missing_ok=True)  # never report an earlier run's readings
    run = subprocess.run(
        ["npx", "playwright", "test", "--project=speed", "--reporter=line"],
        cwd=ROOT / "frontend",
        env=env,
        capture_output=True,
        text=True,
        encoding="utf-8",
        shell=os.name == "nt",  # npx is a .cmd on Windows
        check=False,  # a failed run is reported in the result, not raised
    )
    if run.returncode != 0 or not out.exists():
        lines = [ln.strip() for ln in (run.stdout + run.stderr).splitlines() if ln.strip()]
        first_error = next(
            (i for i, ln in enumerate(lines) if "Error" in ln), max(0, len(lines) - 15)
        )
        return {"error": "the speed run failed", "output": lines[first_error : first_error + 15]}
    return json.loads(out.read_text("utf-8"))


def main() -> int:
    be = _lines(_files(BE_APP, "*.py"))
    fe = _lines(
        [p for p in _files(FE_SRC, "*.ts*") if not _is_fe_test(p) and not p.name.endswith(".d.ts")]
    )
    report: dict[str, object] = {
        "size": _size(be, fe),
        "complexity": _complexity(),
        "noise": _noise(be, fe),
        "governance": _governance(),
        "footprint": _footprint(be, fe),
    }
    if "--speed" in sys.argv[1:]:
        report["speed"] = _speed()
    sys.stdout.write(json.dumps(report, indent=2) + "\n")
    return 0


if __name__ == "__main__":
    sys.exit(main())