xiaobai-review/tools/build_api_registry.py

from __future__ import annotations

import argparse
import ast
import json
import re
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "config" / "api.config.json"
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))


def _owner(path: str) -> str:
    ordered = (
        ("/api/health", "health"),
        ("/api/auth", "auth"),
        ("/api/admin", "admin"),
        ("/api/account", "account"),
        ("/api/sentiment", "sentiment"),
        ("/api/dashboard", "market"),
        ("/api/realtime-aggregate", "market"),
        ("/api/rotation", "rotation"),
        ("/api/auction", "auction"),
        ("/api/themes", "themes"),
        ("/api/popularity", "popularity"),
        ("/api/dragon-tiger", "dragon_tiger"),
        ("/api/search", "search"),
        ("/api/chart", "charts"),
        ("/api/stock", "market"),
        ("/api/screener", "screener"),
        ("/api/mentors", "mentor"),
        ("/api/heaven", "heaven"),
        ("/api/alerts", "alerts"),
        ("/api/trades", "review"),
        ("/api/assistant", "review"),
        ("/api/watchlist", "review"),
        ("/api/notes", "review"),
        ("/api/reasons", "admin"),
        ("/api/seat-aliases", "admin"),
        ("/api/backfill", "admin"),
    )
    for prefix, owner in ordered:
        if path.startswith(prefix):
            return owner
    raise ValueError(f"API owner is not registered: {path}")


def _role(method: str, path: str) -> str:
    if path == "/api/health" or path in {
        "/api/auth/register",
        "/api/auth/login",
        "/api/auth/accounts",
        "/api/auth/switch",
        "/api/auth/forget",
    }:
        return "public"
    from api_access import required_role

    sample = re.sub(r"\\d\{6\}", "000001", path)
    sample = re.sub(r"\\d\+", "1", sample)
    sample = sample.replace("(.+)", "sample").replace("(", "").replace(")", "")
    return required_role(method, sample)


def _mapped_paths(text: str) -> dict[str, set[str]]:
    paths = {method: set() for method in ("GET", "POST", "DELETE")}
    tree = ast.parse(text)
    for node in tree.body:
        if not isinstance(node, ast.Assign) or len(node.targets) != 1:
            continue
        target = node.targets[0]
        if not isinstance(target, ast.Name):
            continue
        match = re.fullmatch(
            r"(?:PUBLIC_|AUTHENTICATED_)?(GET|POST|DELETE)_HANDLERS", target.id
        )
        if not match:
            continue
        mapping = ast.literal_eval(node.value)
        if not isinstance(mapping, dict) or not all(
            isinstance(path, str) and path.startswith("/api/") for path in mapping
        ):
            raise ValueError(f"Invalid route handler map: {target.id}")
        paths[match.group(1)].update(mapping)
    return paths


def _route_sources() -> list[tuple[Path, str]]:
    paths = [ROOT / "backend" / "http" / "dispatch.py"]
    paths.extend(sorted((ROOT / "backend" / "features").glob("*/routes.py")))
    return [(path, path.read_text(encoding="utf-8")) for path in paths]


def _method_blocks(path: Path, text: str) -> list[tuple[str, str]]:
    tree = ast.parse(text, filename=str(path))
    lines = text.splitlines()
    blocks: list[tuple[str, str]] = []
    for node in ast.walk(tree):
        if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
            continue
        method = ""
        if node.name.startswith("do_") and node.name[3:] in {"GET", "POST", "DELETE"}:
            method = node.name[3:]
        else:
            suffix = node.name.rsplit("_", 1)[-1].upper()
            if node.name.startswith("_handle_") and suffix in {"GET", "POST", "DELETE"}:
                method = suffix
        if method:
            blocks.append((method, "\n".join(lines[node.lineno - 1 : node.end_lineno])))
    return blocks


def build() -> dict:
    sources = _route_sources()
    mapped_paths = {method: set() for method in ("GET", "POST", "DELETE")}
    for _, text in sources:
        discovered = _mapped_paths(text)
        for method in mapped_paths:
            mapped_paths[method].update(discovered[method])

    discovered_routes = {method: {"exact": set(), "patterns": set()} for method in mapped_paths}
    for path, text in sources:
        for method, block in _method_blocks(path, text):
            discovered_routes[method]["exact"].update(
                re.findall(r'parsed\.path\s*==\s*"(/api/[^"]+)"', block)
            )
            discovered_routes[method]["patterns"].update(
                re.findall(
                    r're\.(?:fullmatch|match)\(\s*r?["\']([^"\']*?/api/[^"\']+)["\']\s*,\s*parsed\.path',
                    block,
                )
            )

    routes = []
    for method in ("GET", "POST", "DELETE"):
        exact_paths = discovered_routes[method]["exact"] | mapped_paths[method]
        patterns = discovered_routes[method]["patterns"]
        for path in sorted(exact_paths):
            routes.append(
                {"method": method, "path": path, "match": "exact", "feature": _owner(path), "access": _role(method, path)}
            )
        for path in sorted(patterns):
            normalized = path.replace("^", "").replace("$", "")
            routes.append(
                {"method": method, "path": normalized, "match": "regex", "feature": _owner(normalized), "access": _role(method, normalized)}
            )
    routes.sort(key=lambda item: (item["path"], item["method"], item["match"]))
    return {
        "schema_version": 1,
        "generated_from": "server.py",
        "routes": routes,
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Build the transitional API ownership registry")
    parser.add_argument("--check", action="store_true")
    args = parser.parse_args()
    rendered = json.dumps(build(), ensure_ascii=False, indent=2) + "\n"
    if args.check:
        if not OUTPUT.exists() or OUTPUT.read_text(encoding="utf-8") != rendered:
            raise SystemExit("API registry is stale; run tools/build_api_registry.py")
        print("API registry is current.")
        return 0
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(rendered, encoding="utf-8")
    print(OUTPUT.relative_to(ROOT).as_posix())
    return 0


if __name__ == "__main__":
    raise SystemExit(main())