scripts/check_contract.py

"""
check_contract.py — static conformance check against the frontend contract.

Compares every response model in app/schemas/ with the corresponding interface
in outreach-frontend/types/, and every route in app/api/ with the paths
outreach-frontend/lib/api/ actually calls. Pure stdlib: it parses both trees, so
it runs without installing anything and without a database.

    uv run python scripts/check_contract.py     (or: python3 scripts/check_contract.py)

Exit code is the number of mismatches.
"""

from __future__ import annotations

import ast
import re
import sys
from pathlib import Path

BACKEND = Path(__file__).resolve().parent.parent
FRONTEND = BACKEND.parent / "outreach-frontend"

PAIRS = [
    ("types/lead.ts", "Lead", "schemas/lead.py", "LeadOut"),
    ("types/problem.ts", "Problem", "schemas/problem.py", "ProblemOut"),
    ("types/problem.ts", "ProblemComment", "schemas/problem.py", "ProblemCommentOut"),
    ("types/metrics.ts", "CrmMetrics", "schemas/metrics.py", "CrmMetrics"),
    ("types/metrics.ts", "WeekStats", "schemas/metrics.py", "WeekStats"),
    ("types/metrics.ts", "TeamMetrics", "schemas/metrics.py", "TeamMetrics"),
    ("types/metrics.ts", "SystemStatus", "schemas/metrics.py", "SystemStatus"),
    ("types/scan.ts", "ScanResult", "schemas/scan.py", "ScanResult"),
    ("types/scan.ts", "DecisionMaker", "schemas/scan.py", "DecisionMaker"),
    ("types/scan.ts", "LinkedInResearch", "schemas/scan.py", "LinkedInResearch"),
    ("types/api.ts", "Job", "schemas/job.py", "JobOut"),
]


def ts_interface_fields(rel: str, name: str) -> list[str]:
    src = (FRONTEND / rel).read_text(encoding="utf-8")
    m = re.search(rf"export interface {name}(?:<[^>]*>)?\s*\{{(.*?)\n\}}", src, re.S)
    if not m:
        raise SystemExit(f"interface {name} not found in {rel}")
    body = re.sub(r"//.*", "", re.sub(r"/\*.*?\*/", "", m.group(1), flags=re.S))
    return re.findall(r"^\s*([a-z_][A-Za-z0-9_]*)\??\s*:", body, re.M)


def py_model_fields(rel: str, cls: str) -> list[str]:
    tree = ast.parse((BACKEND / "app" / rel).read_text(encoding="utf-8"))
    for node in ast.walk(tree):
        if isinstance(node, ast.ClassDef) and node.name == cls:
            return [s.target.id for s in node.body if isinstance(s, ast.AnnAssign)]
    raise SystemExit(f"class {cls} not found in {rel}")


def backend_routes() -> set[tuple[str, str]]:
    routes: set[tuple[str, str]] = set()
    for path in (BACKEND / "app" / "api").glob("*.py"):
        tree = ast.parse(path.read_text(encoding="utf-8"))
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for dec in node.decorator_list:
                if (
                    isinstance(dec, ast.Call)
                    and isinstance(dec.func, ast.Attribute)
                    and dec.func.attr in {"get", "post", "patch", "delete"}
                    and dec.args
                    and isinstance(dec.args[0], ast.Constant)
                ):
                    routes.add((dec.func.attr.upper(), _normalise(dec.args[0].value)))
    return routes


def frontend_routes() -> set[tuple[str, str]]:
    """Pulls api.get/post/patch/delete("...") calls out of lib/api/*.ts."""
    routes: set[tuple[str, str]] = set()
    for path in (FRONTEND / "lib" / "api").glob("*.ts"):
        src = path.read_text(encoding="utf-8")
        for verb, raw in re.findall(
            r"api\.(get|post|patch|delete)<[^>]*>\(\s*[`\"']([^`\"']+)", src
        ):
            routes.add((verb.upper(), _normalise(raw)))
    return routes


def _normalise(p: str) -> str:
    """/leads/${id} and /leads/{lead_id} both become /leads/:id."""
    p = re.sub(r"\$\{[^}]+\}", ":id", p)
    p = re.sub(r"\{[^}]+\}", ":id", p)
    return "/" + p.strip("/")


def main() -> int:
    failures = 0

    print("── response shapes ──")
    for ts_rel, iface, py_rel, cls in PAIRS:
        want, got = ts_interface_fields(ts_rel, iface), py_model_fields(py_rel, cls)
        missing = [f for f in want if f not in got]
        extra = [f for f in got if f not in want]
        if missing or extra:
            failures += 1
            print(f"DIFF {iface:18s} missing={missing} extra={extra}")
        else:
            print(f"OK   {iface:18s} {len(want)} fields")

    print("\n── routes ──")
    be, fe = backend_routes(), frontend_routes()
    # /jobs/{discover|scan|import} is one client call with a variable kind slug.
    fe.discard(("POST", "/jobs/:id"))
    for kind in ("discover", "scan", "import"):
        fe.add(("POST", f"/jobs/{kind}"))

    for verb, path in sorted(fe):
        if (verb, path) in be:
            print(f"OK   {verb:6s} {path}")
        else:
            failures += 1
            print(f"MISS {verb:6s} {path}  — frontend calls it, backend has no route")

    unused = sorted(be - fe)
    if unused:
        print("\nbackend routes the frontend does not call (fine, just noting):")
        for verb, path in unused:
            print(f"     {verb:6s} {path}")

    print(f"\nmismatches: {failures}")
    return failures


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