Files
govoplan/tools/inventory/platform-interface-inventory.py
T

629 lines
23 KiB
Python

#!/usr/bin/env python3
"""Extract a source-derived GovOPlaN UI, translation, module, and API inventory."""
from __future__ import annotations
import argparse
import ast
from collections import Counter
from dataclasses import asdict, is_dataclass
import importlib
import json
from pathlib import Path
import re
import subprocess
import sys
from typing import Any
META_ROOT = Path(__file__).resolve().parents[2]
HTTP_METHODS = {"delete", "get", "head", "options", "patch", "post", "put"}
PATH_PARAMETER = re.compile(r"\$\{[^{}]*\}|\{[^{}]+\}")
ENDPOINT_SURFACE_CATEGORIES = {
"ui_reachable",
"intentionally_headless",
"public_integration",
"worker_internal",
"compatibility",
"missing_ui",
"removable",
}
DEFAULT_ENDPOINT_DECLARATIONS = (
META_ROOT / "tools" / "inventory" / "endpoint-surface-declarations.json"
)
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--output-dir",
type=Path,
default=META_ROOT / "audit-reports" / "platform-inventory",
)
parser.add_argument(
"--strict",
action="store_true",
help="Fail on missing translations or incomplete endpoint-surface declarations.",
)
parser.add_argument(
"--endpoint-declarations",
type=Path,
default=DEFAULT_ENDPOINT_DECLARATIONS,
help="Versioned endpoint-surface declaration registry.",
)
args = parser.parse_args()
catalog = json.loads((META_ROOT / "repositories.json").read_text(encoding="utf-8"))
workspace_root = Path(catalog["default_parent"]).resolve()
webui = _extract_webui()
backend_endpoints = _extract_backend_endpoints(catalog, workspace_root)
manifests = _extract_manifests(catalog, workspace_root)
endpoint_declarations = _load_endpoint_declarations(
args.endpoint_declarations.resolve()
)
inventory = _assemble_inventory(
webui=webui,
backend_endpoints=backend_endpoints,
manifests=manifests,
endpoint_declarations=endpoint_declarations,
)
output_dir = args.output_dir.resolve()
output_dir.mkdir(parents=True, exist_ok=True)
json_path = output_dir / "platform-interface-inventory.json"
markdown_path = output_dir / "platform-interface-inventory.md"
json_path.write_text(
json.dumps(inventory, indent=2, sort_keys=True) + "\n",
encoding="utf-8",
)
markdown_path.write_text(_render_markdown(inventory), encoding="utf-8")
print(f"Platform inventory JSON: {json_path}")
print(f"Platform inventory summary: {markdown_path}")
if args.strict:
failures: list[str] = []
if inventory["translation_health"]["missing_catalog_entries"]:
failures.append("used translation keys are missing from generated catalogs")
if inventory["api"]["unclassified_endpoints"]:
failures.append(
f"{len(inventory['api']['unclassified_endpoints'])} backend "
"endpoints have no WebUI evidence or surface declaration"
)
if inventory["api"]["stale_endpoint_declarations"]:
failures.append(
f"{len(inventory['api']['stale_endpoint_declarations'])} "
"endpoint declarations do not match a backend endpoint"
)
if failures:
print(
"Strict platform inventory failed: " + "; ".join(failures) + ".",
file=sys.stderr,
)
return 1
return 0
def _extract_webui() -> dict[str, Any]:
helper = META_ROOT / "tools" / "inventory" / "extract-webui-structure.mjs"
completed = subprocess.run(
["node", str(helper), str(META_ROOT)],
check=True,
capture_output=True,
text=True,
)
return json.loads(completed.stdout)
def _extract_backend_endpoints(
catalog: dict[str, Any],
workspace_root: Path,
) -> list[dict[str, Any]]:
endpoints: list[dict[str, Any]] = []
for repository in catalog["repositories"]:
repository_root = workspace_root / repository["path"]
source_root = repository_root / "src"
if not source_root.is_dir():
continue
for source_path in sorted(source_root.rglob("*.py")):
try:
tree = ast.parse(
source_path.read_text(encoding="utf-8"),
filename=str(source_path),
)
except (OSError, SyntaxError, UnicodeDecodeError):
continue
prefixes = _router_prefixes(tree)
for node in ast.walk(tree):
if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
continue
for decorator in node.decorator_list:
endpoint = _endpoint_from_decorator(
decorator,
prefixes=prefixes,
)
if endpoint is None:
continue
endpoints.append(
{
"repository": repository["name"],
"file": str(source_path.relative_to(repository_root)),
"line": node.lineno,
"handler": node.name,
**endpoint,
}
)
return sorted(
endpoints,
key=lambda item: (
item["repository"],
item["path"],
item["method"],
item["file"],
item["line"],
),
)
def _router_prefixes(tree: ast.AST) -> dict[str, str]:
prefixes: dict[str, str] = {}
for node in ast.walk(tree):
if not isinstance(node, (ast.Assign, ast.AnnAssign)):
continue
value = node.value
if not isinstance(value, ast.Call):
continue
function_name = _call_name(value.func)
if function_name not in {"APIRouter", "fastapi.APIRouter"}:
continue
prefix = ""
for keyword in value.keywords:
if keyword.arg == "prefix":
prefix = _static_string(keyword.value) or ""
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
for target in targets:
if isinstance(target, ast.Name):
prefixes[target.id] = prefix
return prefixes
def _endpoint_from_decorator(
decorator: ast.expr,
*,
prefixes: dict[str, str],
) -> dict[str, str] | None:
if not isinstance(decorator, ast.Call) or not isinstance(
decorator.func, ast.Attribute
):
return None
method = decorator.func.attr.lower()
if method not in HTTP_METHODS or not decorator.args:
return None
route = _static_string(decorator.args[0])
if route is None:
return None
owner = (
decorator.func.value.id if isinstance(decorator.func.value, ast.Name) else ""
)
prefix = prefixes.get(owner, "")
return {
"method": method.upper(),
"path": _join_route(prefix, route),
"router": owner,
}
def _extract_manifests(
catalog: dict[str, Any],
workspace_root: Path,
) -> list[dict[str, Any]]:
source_roots = [
workspace_root / repository["path"] / "src"
for repository in catalog["repositories"]
if (workspace_root / repository["path"] / "src").is_dir()
]
sys.path[:0] = [str(path) for path in source_roots]
manifests: list[dict[str, Any]] = []
for repository in catalog["repositories"]:
source_root = workspace_root / repository["path"] / "src"
if not source_root.is_dir():
continue
for manifest_path in sorted(source_root.glob("*/backend/manifest.py")):
module_name = ".".join(
manifest_path.relative_to(source_root).with_suffix("").parts
)
loaded = importlib.import_module(module_name)
manifest = loaded.get_manifest()
frontend = manifest.frontend
manifests.append(
{
"repository": repository["name"],
"id": manifest.id,
"name": manifest.name,
"version": manifest.version,
"dependencies": list(manifest.dependencies),
"optional_dependencies": list(manifest.optional_dependencies),
"required_capabilities": list(manifest.required_capabilities),
"provided_interfaces": [
{"name": item.name, "version": item.version}
for item in manifest.provides_interfaces
],
"runtime_capabilities": sorted(manifest.capability_factories),
"permissions": [
{
"scope": permission.scope,
"label": permission.label,
"level": permission.level,
}
for permission in manifest.permissions
],
"frontend": (
{
"package": frontend.package_name,
"routes": [
_plain_value(route) for route in frontend.routes
],
"nav_items": [
_plain_value(item) for item in frontend.nav_items
],
"view_surfaces": [
_plain_value(surface)
for surface in frontend.view_surfaces
],
}
if frontend is not None
else None
),
}
)
return sorted(manifests, key=lambda item: item["id"])
def _assemble_inventory(
*,
webui: dict[str, Any],
backend_endpoints: list[dict[str, Any]],
manifests: list[dict[str, Any]],
endpoint_declarations: dict[tuple[str, str, str], dict[str, Any]],
) -> dict[str, Any]:
frontend_refs = webui["frontendApiReferences"]
frontend_paths = {
canonical_api_path(reference["path"])
for reference in frontend_refs
if canonical_api_path(reference["path"])
}
endpoint_keys = {endpoint_key(endpoint) for endpoint in backend_endpoints}
classified_endpoints: list[dict[str, Any]] = []
for endpoint in backend_endpoints:
key = endpoint_key(endpoint)
canonical_path = key[2]
static_webui_reference = canonical_path in frontend_paths
declaration = endpoint_declarations.get(key)
surface = (
{
"category": "ui_reachable",
"rationale": "A canonical API path reference exists in WebUI source.",
"source": "static_webui_scan",
}
if static_webui_reference
else (
{**declaration, "source": "declaration"}
if declaration is not None
else None
)
)
classified_endpoints.append(
{
**endpoint,
"canonical_path": canonical_path,
"static_webui_reference": static_webui_reference,
"surface": surface,
}
)
unreferenced = [
endpoint
for endpoint in classified_endpoints
if not endpoint["static_webui_reference"]
]
unclassified = [
endpoint for endpoint in unreferenced if endpoint["surface"] is None
]
stale_declarations = [
declaration
for key, declaration in endpoint_declarations.items()
if key not in endpoint_keys
]
classification_counts = Counter(
endpoint["surface"]["category"]
for endpoint in classified_endpoints
if endpoint["surface"] is not None
)
usages = {item["key"] for item in webui["translationUsages"]}
catalogs = webui["translationCatalog"]
catalog_keys = {locale: set(entries) for locale, entries in catalogs.items()}
expected_locales = sorted(catalog_keys)
missing_catalog_entries = [
{
"key": key,
"missing_locales": [
locale for locale in expected_locales if key not in catalog_keys[locale]
],
}
for key in sorted(usages)
if any(key not in catalog_keys[locale] for locale in expected_locales)
]
fields = webui["fields"]
help_candidates = [field for field in fields if field["helpCandidate"]]
return {
"schema_version": 1,
"scope": {
"source": "local GovOPlaN repository catalog",
"limitations": [
"Static extraction cannot resolve runtime-computed labels, routes, or API paths.",
"A backend endpoint without a static WebUI reference may intentionally serve public clients, workers, connectors, or external integrations.",
"A field marked as a help candidate may receive contextual help from a surrounding dynamic component.",
],
},
"modules": manifests,
"ui": {
"fields": fields,
"labels": webui["labels"],
"visible_text": webui["visibleText"],
"routes": webui["routes"],
"navigation": webui["navigation"],
"ui_capabilities": webui["uiCapabilities"],
"help_candidates": help_candidates,
},
"translations": {
"catalog": catalogs,
"usages": webui["translationUsages"],
"dynamic_usages": webui["dynamicTranslationUsages"],
},
"translation_health": {
"locales": expected_locales,
"used_keys": len(usages),
"missing_catalog_entries": missing_catalog_entries,
},
"api": {
"backend_endpoints": classified_endpoints,
"frontend_references": frontend_refs,
"unreferenced_by_static_webui_scan": unreferenced,
"unclassified_endpoints": unclassified,
"stale_endpoint_declarations": stale_declarations,
"classification_counts": dict(sorted(classification_counts.items())),
},
"summary": {
"modules": len(manifests),
"ui_fields": len(fields),
"ui_fields_with_static_help": len(fields) - len(help_candidates),
"help_review_candidates": len(help_candidates),
"label_attributes": len(webui["labels"]),
"visible_text_nodes": len(webui["visibleText"]),
"frontend_routes": len(webui["routes"]),
"backend_endpoints": len(backend_endpoints),
"frontend_api_references": len(frontend_refs),
"backend_endpoints_without_static_webui_reference": len(unreferenced),
"unclassified_backend_endpoints": len(unclassified),
"stale_endpoint_declarations": len(stale_declarations),
},
}
def _render_markdown(inventory: dict[str, Any]) -> str:
summary = inventory["summary"]
missing = inventory["translation_health"]["missing_catalog_entries"]
help_by_repository = Counter(
item["repository"] for item in inventory["ui"]["help_candidates"]
)
endpoint_by_repository = Counter(
item["repository"]
for item in inventory["api"]["unreferenced_by_static_webui_scan"]
)
classification_counts = inventory["api"]["classification_counts"]
lines = [
"# GovOPlaN Platform Interface Inventory",
"",
"Generated from runtime module manifests plus TypeScript and Python ASTs.",
"Counts are source evidence, not a claim that every surface is enabled or reachable.",
"",
"## Summary",
"",
f"- Modules: {summary['modules']}",
f"- UI fields: {summary['ui_fields']}",
f"- Fields with statically associated help: {summary['ui_fields_with_static_help']}",
f"- Help review candidates: {summary['help_review_candidates']}",
f"- Label attributes: {summary['label_attributes']}",
f"- Frontend routes: {summary['frontend_routes']}",
f"- Backend endpoints: {summary['backend_endpoints']}",
f"- Frontend API references: {summary['frontend_api_references']}",
(
"- Backend endpoints without a static WebUI reference: "
f"{summary['backend_endpoints_without_static_webui_reference']}"
),
f"- Unclassified backend endpoints: {summary['unclassified_backend_endpoints']}",
f"- Stale endpoint declarations: {summary['stale_endpoint_declarations']}",
f"- Used translation keys missing from a locale catalog: {len(missing)}",
"",
"## Help Review Candidates",
"",
"| Repository | Fields |",
"| --- | ---: |",
]
lines.extend(
f"| `{repository}` | {count} |"
for repository, count in sorted(help_by_repository.items())
)
lines.extend(
[
"",
"## Endpoints Without Static WebUI References",
"",
"These are review candidates. Public APIs, worker callbacks, connector",
"endpoints, health checks, and dynamically assembled paths are legitimate",
"reasons for appearing here.",
"",
"| Repository | Endpoints |",
"| --- | ---: |",
]
)
lines.extend(
f"| `{repository}` | {count} |"
for repository, count in sorted(endpoint_by_repository.items())
)
lines.extend(
[
"",
"## Endpoint Surface Classifications",
"",
"| Classification | Endpoints |",
"| --- | ---: |",
]
)
lines.extend(
f"| `{category}` | {count} |"
for category, count in sorted(classification_counts.items())
)
lines.extend(
[
"",
"## Interpretation",
"",
"Use the JSON artifact for exact file and line evidence. Missing help is",
"a triage list, not an automatic defect. Endpoint coverage requires an",
"owner classification before enforcement. Runtime-computed structures",
"need explicit manifest metadata to become canonically visible.",
"",
]
)
return "\n".join(lines)
def canonical_api_path(value: str) -> str:
path = value.split("?", 1)[0].strip()
if "/api/" in path:
path = path[path.index("/api/") :]
path = re.sub(r"^/api/v\d+", "", path)
path = re.sub(r"(?<=[^/])\$\{[^{}]*\}$", "", path)
path = PATH_PARAMETER.sub("{}", path)
path = re.sub(r"/+", "/", path)
return path.rstrip("/") or "/"
def endpoint_key(endpoint: dict[str, Any]) -> tuple[str, str, str]:
return (
str(endpoint["repository"]),
str(endpoint["method"]).upper(),
canonical_api_path(str(endpoint["path"])),
)
def _load_endpoint_declarations(
path: Path,
) -> dict[tuple[str, str, str], dict[str, Any]]:
try:
payload = json.loads(path.read_text(encoding="utf-8"))
except FileNotFoundError as exc:
raise ValueError(
f"Endpoint declaration registry does not exist: {path}"
) from exc
except json.JSONDecodeError as exc:
raise ValueError(
f"Endpoint declaration registry is invalid JSON: {exc}"
) from exc
if not isinstance(payload, dict) or payload.get("schema_version") != 1:
raise ValueError("Endpoint declaration registry must use schema_version 1.")
entries = payload.get("endpoints")
if not isinstance(entries, list):
raise ValueError(
"Endpoint declaration registry must contain an endpoints list."
)
declarations: dict[tuple[str, str, str], dict[str, Any]] = {}
for index, entry in enumerate(entries):
if not isinstance(entry, dict):
raise ValueError(f"Endpoint declaration {index} must be an object.")
repository = entry.get("repository")
method = entry.get("method")
raw_path = entry.get("path")
category = entry.get("category")
rationale = entry.get("rationale")
if not isinstance(repository, str) or not repository.strip():
raise ValueError(f"Endpoint declaration {index} has no repository.")
if not isinstance(method, str) or method.lower() not in HTTP_METHODS:
raise ValueError(f"Endpoint declaration {index} has an invalid method.")
if not isinstance(raw_path, str) or not raw_path.startswith("/"):
raise ValueError(f"Endpoint declaration {index} has an invalid path.")
canonical_path = canonical_api_path(raw_path)
if raw_path != canonical_path:
raise ValueError(
f"Endpoint declaration {index} path must be canonical: {canonical_path}"
)
if category not in ENDPOINT_SURFACE_CATEGORIES:
raise ValueError(f"Endpoint declaration {index} has an invalid category.")
if not isinstance(rationale, str) or not rationale.strip():
raise ValueError(f"Endpoint declaration {index} has no rationale.")
tracking_issue = entry.get("tracking_issue")
if category == "missing_ui" and (
not isinstance(tracking_issue, str) or not tracking_issue.strip()
):
raise ValueError(
f"Endpoint declaration {index} requires a tracking_issue for missing_ui."
)
key = (repository, method.upper(), canonical_path)
if key in declarations:
raise ValueError(f"Duplicate endpoint declaration: {key!r}.")
declarations[key] = {
"repository": repository,
"method": method.upper(),
"path": canonical_path,
"category": category,
"rationale": rationale.strip(),
**(
{"tracking_issue": tracking_issue.strip()}
if isinstance(tracking_issue, str) and tracking_issue.strip()
else {}
),
}
return declarations
def _join_route(prefix: str, route: str) -> str:
return f"/{prefix.strip('/')}/{route.strip('/')}".replace("//", "/")
def _static_string(node: ast.AST) -> str | None:
if isinstance(node, ast.Constant) and isinstance(node.value, str):
return node.value
if isinstance(node, ast.JoinedStr):
pieces: list[str] = []
for value in node.values:
if isinstance(value, ast.Constant) and isinstance(value.value, str):
pieces.append(value.value)
elif isinstance(value, ast.FormattedValue):
pieces.append("${}")
else:
return None
return "".join(pieces)
return None
def _call_name(node: ast.AST) -> str:
if isinstance(node, ast.Name):
return node.id
if isinstance(node, ast.Attribute):
parent = _call_name(node.value)
return f"{parent}.{node.attr}" if parent else node.attr
return ""
def _plain_value(value: Any) -> Any:
if is_dataclass(value):
return {key: _plain_value(item) for key, item in asdict(value).items()}
if isinstance(value, tuple):
return [_plain_value(item) for item in value]
if isinstance(value, dict):
return {str(key): _plain_value(item) for key, item in value.items()}
return value
if __name__ == "__main__":
raise SystemExit(main())