#!/usr/bin/env python3
"""Build a deterministic legacy FastAPI -> Go OpenAPI route contract matrix.

The script uses Python AST only; importing the legacy application is deliberately
avoided so the audit does not require its runtime dependencies or secrets.
"""
from __future__ import annotations

import argparse
import ast
import re
from dataclasses import dataclass
from pathlib import Path

import yaml

METHODS = {"get", "post", "put", "patch", "delete"}
PARAM = re.compile(r"\{[^{}]+\}")


@dataclass(frozen=True)
class Route:
    method: str
    path: str
    source: str
    line: int


def literal(node: ast.AST | None, default: str = "") -> str:
    return node.value if isinstance(node, ast.Constant) and isinstance(node.value, str) else default


def legacy_routes(root: Path) -> list[Route]:
    result: list[Route] = []
    for source in sorted((root / "app").glob("*.py")):
        tree = ast.parse(source.read_text(encoding="utf-8"), filename=str(source))
        prefixes: dict[str, str] = {"app": ""}
        for node in tree.body:
            if not isinstance(node, (ast.Assign, ast.AnnAssign)):
                continue
            value = node.value
            if not isinstance(value, ast.Call) or not isinstance(value.func, ast.Name) or value.func.id != "APIRouter":
                continue
            targets = node.targets if isinstance(node, ast.Assign) else [node.target]
            prefix = next((literal(item.value) for item in value.keywords if item.arg == "prefix"), "")
            for target in targets:
                if isinstance(target, ast.Name):
                    prefixes[target.id] = prefix
        for node in ast.walk(tree):
            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for decorator in node.decorator_list:
                if not isinstance(decorator, ast.Call) or not isinstance(decorator.func, ast.Attribute):
                    continue
                owner = decorator.func.value
                if not isinstance(owner, ast.Name) or decorator.func.attr.lower() not in METHODS or not decorator.args:
                    continue
                route_path = literal(decorator.args[0], "")
                if route_path == "" and not (isinstance(decorator.args[0], ast.Constant) and decorator.args[0].value == ""):
                    continue
                full = (prefixes.get(owner.id, "") + route_path) or "/"
                result.append(Route(decorator.func.attr.upper(), full, source.name, decorator.lineno))
    return sorted(set(result), key=lambda item: (item.path, item.method, item.source, item.line))


def current_routes(spec: Path) -> set[tuple[str, str]]:
    document = yaml.safe_load(spec.read_text(encoding="utf-8"))
    return {(method.upper(), path) for path, item in document["paths"].items() for method in item if method.lower() in METHODS}


def canonical(path: str) -> str:
    if path.startswith("/admin/api"):
        path = "/api/v1/admin" + path[len("/admin/api"):]
    elif path.startswith("/portal/api"):
        path = "/api/v1/portal" + path[len("/portal/api"):]
    return PARAM.sub("{}", path.rstrip("/") or "/")


ALIASES = {
    "/api/v1/admin/keys": "/api/v1/admin/api-keys",
    "/api/v1/admin/kb": "/api/v1/admin/knowledge-bases",
    "/api/v1/admin/model_pricing": "/api/v1/admin/model-prices",
    "/api/v1/admin/logs": "/api/v1/admin/audit-events",
    "/api/v1/admin/users": "/api/v1/admin/identities/portal-users",
    "/api/v1/admin/admins": "/api/v1/admin/identities/admins",
    "/api/v1/admin/system_info": "/api/v1/admin/system-info",
    "/api/v1/admin/totp/backup_codes": "/api/v1/admin/totp/backup-codes",
    "/api/v1/portal/totp/backup_codes": "/api/v1/portal/totp/backup-codes",
    "/v1/apps": "/v1/applications",
}

# Capability replacements whose new API intentionally does not mirror the old
# path shape. They remain migration work (client payload/response validation),
# but are not incorrectly reported as an unimplemented domain.
CAPABILITY_REPLACEMENTS = [
    (re.compile(r"^/api/v1/admin/(rules|rule_dimensions|content_audit_scopes)(/|$)"), "/api/v1/admin/content-policies"),
    (re.compile(r"^/api/v1/admin/(rate_limit_rules|model_qps)(/|$)"), "/api/v1/admin/api-keys and /api/v1/admin/model-routes"),
    (re.compile(r"^/api/v1/admin/(model_aliases|routing_groups)(/|$)"), "/api/v1/admin/model-routes"),
    (re.compile(r"^/api/v1/admin/(stats|dept_stats|cost_report|pricing_config|dept_budgets)(/|$)"), "/api/v1/admin/usage/daily and /api/v1/admin/model-prices"),
    (re.compile(r"^/api/v1/admin/(notify_config)(/|$)"), "/api/v1/admin/notification-channels"),
    (re.compile(r"^/api/v1/admin/(logs|log_search|log_views)(/|$)"), "/api/v1/admin/audit-events"),
    (re.compile(r"^/api/v1/admin/malicious_domains(/|$)"), "/api/v1/admin/content-policies"),
    (re.compile(r"^/api/v1/admin/prompts/categories(/|$)"), "/api/v1/admin/prompt-categories"),
    (re.compile(r"^/api/v1/admin/prompts/\{\}/preview$"), "/api/v1/admin/prompts/{prompt_id}/render"),
    (re.compile(r"^/api/v1/admin/knowledge-bases/\{\}/documents/(paste|upload)$"), "/api/v1/admin/knowledge-bases/{knowledge_base_id}/documents"),
    (re.compile(r"^/api/v1/admin/knowledge-bases/\{\}/test_search$"), "/api/v1/admin/knowledge-bases/{knowledge_base_id}/search"),
    (re.compile(r"^/api/v1/admin/tools/\{\}/test_invoke$"), "/api/v1/admin/tools/{tool_id}/test"),
    (re.compile(r"^/api/v1/admin/applications/\{\}/composition$"), "/api/v1/admin/applications/{application_id}"),
    (re.compile(r"^/api/v1/admin/audit_settings(/|$)"), "/api/v1/admin/content-policies and /api/v1/admin/notification-channels"),
    (re.compile(r"^/api/v1/admin/feishu_auth(/|$)"), "/api/v1/admin/identity-providers (generic OIDC/SAML)"),
    (re.compile(r"^/api/v1/portal/feishu(/|$)"), "/api/v1/portal/sso (generic OIDC/SAML)"),
    (re.compile(r"^/api/v1/admin/providers/\{\}/backends(/|$)"), "/api/v1/admin/providers/{provider_id}/models and /api/v1/admin/model-routes"),
    (re.compile(r"^/api/v1/admin/providers/\{\}/fetch_models$"), "/api/v1/admin/providers/{provider_id}/models/sync"),
    (re.compile(r"^/api/v1/admin/audit-events(/search|/\{\}/view_history)?$"), "/api/v1/admin/audit-events (bounded immutable audit query)"),
    (re.compile(r"^/api/v1/admin/applications/\{\}/test$"), "/v1/applications/{application_code}/chat/completions"),
    (re.compile(r"^/api/v1/admin/applications/\{\}/keys$"), "/api/v1/admin/api-keys (scoped credentials)"),
    (re.compile(r"^/api/v1/admin/identities/portal-users/\{\}/approve$"), "/api/v1/admin/identities/portal-users/{identity_id} active=true"),
    (re.compile(r"^/api/v1/admin/identities/portal-users/\{\}/reset_password$"), "/api/v1/admin/identities/portal-users/{identity_id} password field"),
    (re.compile(r"^/api/v1/admin/identities/portal-users/\{\}/feishu$"), "/api/v1/admin/identity-providers (generic identity link lifecycle)"),
    (re.compile(r"^/api/v1/admin/identities/(admins|portal-users)/\{\}$"), "/api/v1/admin/identities/*/{identity_id} active=false"),
    (re.compile(r"^/api/v1/admin/departments/\{\}$"), "/api/v1/admin/departments/{department_id} active=false"),
    (re.compile(r"^/api/v1/admin/identities/admins/change_password$"), "/api/v1/admin/password"),
    (re.compile(r"^/api/v1/admin/api-keys/\{\}$"), "/api/v1/admin/api-keys/{api_key_id}/limits"),
    (re.compile(r"^/api/v1/admin/providers/\{\}$"), "/api/v1/admin/providers/{provider_id} enabled=false"),
    (re.compile(r"^/v1/apps/\{\}/chat/completions$"), "/v1/applications/{application_code}/chat/completions"),
]

RETIRED = [
    (re.compile(r"^/(admin|portal)/ui$"), "Art Design Pro SPA deployment"),
    (re.compile(r"^/api/v1/admin/logs/\{\}/view_raw$"), "retired: unbounded raw-body access is excluded by storage/compliance ADR"),
    (re.compile(r"^/api/v1/admin/logs$"), "retired: immutable audit data is removed only by the configured retention worker"),
    (re.compile(r"^/api/v1/admin/tools/\{\}/headers$"), "retired: decrypted secret headers are intentionally write-only"),
]


def alias(path: str) -> str:
    for old, new in sorted(ALIASES.items(), key=lambda item: len(item[0]), reverse=True):
        if path == old or path.startswith(old + "/"):
            return new + path[len(old):]
    return path


def compare(legacy: list[Route], current: set[tuple[str, str]]) -> list[tuple[Route, str, str]]:
    normalized_current = {(method, canonical(path)): path for method, path in current}
    rows = []
    for route in legacy:
        expected = canonical(route.path)
        direct = normalized_current.get((route.method, expected))
        if direct:
            rows.append((route, "covered", direct))
            continue
        replacement = alias(expected)
        target = normalized_current.get((route.method, replacement))
        if target:
            rows.append((route, "replaced", target))
            continue
        if route.method == "PATCH":
            target = normalized_current.get(("PUT", replacement))
            if target:
                rows.append((route, "replaced", target + " (PUT with full revision payload)"))
                continue
        decision = next((target for pattern, target in CAPABILITY_REPLACEMENTS if pattern.search(replacement)), None)
        if decision:
            rows.append((route, "replaced", decision))
            continue
        retirement = next((target for pattern, target in RETIRED if pattern.search(expected)), None)
        rows.append((route, "retired" if retirement else "missing", retirement or replacement))
    return rows


def write_markdown(rows: list[tuple[Route, str, str]], destination: Path) -> None:
    counts = {status: sum(1 for _, value, _ in rows if value == status) for status in ("covered", "replaced", "retired", "missing")}
    lines = [
        "# 旧 Python → Go 路由契约矩阵", "",
        "> 由 `scripts/compare_route_contracts.py` 通过 AST 与 OpenAPI 自动生成；`replaced` 表示能力有新契约承接，仍需兼容层或客户端迁移，不能视为原路由原样可用。", "",
        f"- 旧源码路由装饰器：{len(rows)}", f"- 同方法同规范路径：{counts['covered']}",
        f"- 新契约替代：{counts['replaced']}", f"- 明确退役：{counts['retired']}", f"- 尚无契约：{counts['missing']}", "",
        "| 状态 | 方法 | 旧路径 | Go 契约/预期目标 | 来源 |", "|---|---|---|---|---|",
    ]
    labels = {"covered": "已覆盖", "replaced": "新契约替代", "retired": "明确退役", "missing": "缺口"}
    for route, status, target in rows:
        lines.append(f"| {labels[status]} | `{route.method}` | `{route.path or '/'}` | `{target}` | `{route.source}:{route.line}` |")
    destination.write_text("\n".join(lines) + "\n", encoding="utf-8")


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--legacy-root", type=Path, default=Path("/home/ben/ai-gateway-work"))
    parser.add_argument("--openapi", type=Path, default=Path("api/openapi/gateway.yaml"))
    parser.add_argument("--output", type=Path, default=Path("docs/route-contract-matrix.md"))
    parser.add_argument("--fail-on-missing", action="store_true", help="return a non-zero status when an undecided route remains")
    args = parser.parse_args()
    rows = compare(legacy_routes(args.legacy_root), current_routes(args.openapi))
    args.output.parent.mkdir(parents=True, exist_ok=True)
    write_markdown(rows, args.output)
    counts = {status: sum(1 for _, value, _ in rows if value == status) for status in ("covered", "replaced", "retired", "missing")}
    print(f"legacy={len(rows)} covered={counts['covered']} replaced={counts['replaced']} retired={counts['retired']} missing={counts['missing']}")
    if args.fail_on_missing and counts["missing"]:
        raise SystemExit(1)


if __name__ == "__main__":
    main()