xiaobaifupan/tools/build_architecture_inventory.py

from __future__ import annotations

import argparse
import ast
import json
import re
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "docs" / "governance" / "architecture-inventory.json"


def relative(path: Path) -> str:
    return path.relative_to(ROOT).as_posix()


def source(path: str) -> str:
    return (ROOT / path).read_text(encoding="utf-8")


def page_inventory(html: str) -> list[dict[str, str]]:
    items: list[dict[str, str]] = []
    seen: set[str] = set()
    pattern = re.compile(
        r'<button[^>]+data-view="(?P<id>[^"]+)"[^>]+title="(?P<title>[^"]+)"',
        re.IGNORECASE,
    )
    for match in pattern.finditer(html):
        page_id = match.group("id")
        if page_id in seen:
            continue
        seen.add(page_id)
        items.append({"id": page_id, "title": match.group("title")})
    return items


def api_inventory(server: str) -> dict[str, list[str]]:
    exact = sorted(set(re.findall(r'parsed\.path\s*==\s*"(/api/[^"]+)"', server)))
    prefixes = sorted(
        set(re.findall(r'parsed\.path\.startswith\(\s*"(/api/[^"]+)"', server))
    )
    patterns = sorted(
        set(
            item
            for item in re.findall(r'r?["\']([^"\']*?/api/[^"\']+)["\']', server)
            if "\\d" in item or ".+" in item or "(?P" in item
        )
    )
    return {"exact": exact, "prefixes": prefixes, "patterns": patterns}


def database_inventory(database: str) -> list[str]:
    return re.findall(r"CREATE TABLE IF NOT EXISTS\s+([a-zA-Z0-9_]+)", database)


def python_functions(path: str, prefixes: tuple[str, ...]) -> list[str]:
    tree = ast.parse(source(path), filename=path)
    result: list[str] = []
    for node in ast.walk(tree):
        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith(prefixes):
            result.append(node.name)
    return sorted(set(result))


def css_layers(html: str) -> list[str]:
    return re.findall(r'<link[^>]+rel="stylesheet"[^>]+href="([^"]+)"', html)


def code_hotspots() -> list[dict[str, Any]]:
    candidates = [
        "server.py",
        "database.py",
        "screener.py",
        "market_insights.py",
        "tushare_client.py",
        "static/index.html",
        "static/app.js",
        "static/styles.css",
        "static/redesign-v2.css",
        "static/renovation.css",
        "static/theme.css",
        "static/wentian-v2.css",
    ]
    rows = []
    for name in candidates:
        path = ROOT / name
        rows.append(
            {
                "path": name,
                "bytes": path.stat().st_size,
                "lines": len(path.read_text(encoding="utf-8").splitlines()),
            }
        )
    return sorted(rows, key=lambda item: item["bytes"], reverse=True)


def build() -> dict[str, Any]:
    html = source("static/index.html")
    server = source("server.py")
    database = source("database.py")
    pages = page_inventory(html)
    api = api_inventory(server)
    tables = database_inventory(database)
    return {
        "schema_version": 1,
        "captured_from": "governed source tree",
        "runtime": {
            "http_server": "http.server.ThreadingHTTPServer",
            "application_processes": 1,
            "database": "SQLite WAL",
            "frontend": "build-free HTML/CSS/JavaScript",
            "container_port": 8765,
        },
        "counts": {
            "primary_pages": len(pages),
            "api_exact_paths": len(api["exact"]),
            "api_prefixes": len(api["prefixes"]),
            "api_patterns": len(api["patterns"]),
            "database_tables": len(tables),
        },
        "pages": pages,
        "api": api,
        "database_tables": tables,
        "background_job_methods": python_functions(
            "server.py", ("_background", "_run_background", "_schedule_", "run_automatic")
        ),
        "external_data_adapters": [
            {"provider": "tushare", "path": "tushare_client.py", "runtime_role": "primary deterministic market data"},
            {"provider": "ifind", "path": "ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
            {"provider": "eastmoney", "path": "chart_data_provider.py", "runtime_role": "display chart fallback"},
            {"provider": "eastmoney", "path": "realtime_aggregator.py", "runtime_role": "isolated realtime observation"},
            {"provider": "tencent", "path": "realtime_aggregator.py", "runtime_role": "index observation fallback"},
        ],
        "llm_entrypoints": [
            {"function": "stream_with_mentor", "path": "mentor_agent.py"},
            {"function": "interpret_heaven", "path": "heaven_agent.py"},
            {"function": "stream_review_assistant", "path": "assistant_agent.py"},
            {"function": "compile_strategy_with_llm", "path": "llm_strategy.py"},
            {"function": "test_llm_connection", "path": "llm_strategy.py"},
        ],
        "css_layers": css_layers(html),
        "code_hotspots": code_hotspots(),
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Build the current architecture inventory")
    parser.add_argument("--check", action="store_true", help="fail when the committed inventory is stale")
    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("architecture inventory is stale; run tools/build_architecture_inventory.py")
        print("Architecture inventory is current.")
        return 0
    OUTPUT.parent.mkdir(parents=True, exist_ok=True)
    OUTPUT.write_text(rendered, encoding="utf-8")
    print(relative(OUTPUT))
    return 0


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