1118 lines
42 KiB
Python
1118 lines
42 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"
|
|
)
|
|
REQUIRED_LOCALES = ("de", "en")
|
|
REFERENCE_LOCALE = "de"
|
|
|
|
|
|
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(
|
|
"--strict-endpoints",
|
|
action="store_true",
|
|
help="Fail only on incomplete or stale endpoint-surface declarations.",
|
|
)
|
|
parser.add_argument(
|
|
"--strict-declarations",
|
|
action="store_true",
|
|
help=(
|
|
"Fail on duplicate stable IDs, WebUI surfaces absent from runtime "
|
|
"metadata, or stale runtime route declarations."
|
|
),
|
|
)
|
|
parser.add_argument(
|
|
"--runtime-snapshot",
|
|
type=Path,
|
|
help=(
|
|
"Compare a saved /api/v1/platform/interface-catalog response with "
|
|
"the static manifest inventory. Any installed module combination "
|
|
"is accepted; every module present in the snapshot must match."
|
|
),
|
|
)
|
|
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 = _resolve_workspace_root(catalog)
|
|
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,
|
|
runtime_snapshot=(
|
|
_load_runtime_snapshot(args.runtime_snapshot.resolve())
|
|
if args.runtime_snapshot is not None
|
|
else None
|
|
),
|
|
)
|
|
|
|
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 or args.strict_endpoints or args.strict_declarations:
|
|
failures = _strict_failures(
|
|
inventory,
|
|
check_translations=args.strict,
|
|
check_endpoints=args.strict or args.strict_endpoints,
|
|
check_declarations=args.strict or args.strict_declarations,
|
|
)
|
|
if failures:
|
|
print(
|
|
"Strict platform inventory failed: " + "; ".join(failures) + ".",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
return 0
|
|
|
|
|
|
def _strict_failures(
|
|
inventory: dict[str, Any],
|
|
*,
|
|
check_translations: bool,
|
|
check_endpoints: bool,
|
|
check_declarations: bool = False,
|
|
) -> list[str]:
|
|
failures: list[str] = []
|
|
if (
|
|
check_translations
|
|
and inventory["translation_health"]["missing_catalog_entries"]
|
|
):
|
|
failures.append("used translation keys are missing from generated catalogs")
|
|
if check_endpoints and inventory["api"]["unclassified_endpoints"]:
|
|
failures.append(
|
|
f"{len(inventory['api']['unclassified_endpoints'])} backend "
|
|
"endpoints have no WebUI evidence or surface declaration"
|
|
)
|
|
if check_endpoints and inventory["api"]["stale_endpoint_declarations"]:
|
|
failures.append(
|
|
f"{len(inventory['api']['stale_endpoint_declarations'])} "
|
|
"endpoint declarations do not match a backend endpoint"
|
|
)
|
|
declaration_health = inventory.get("declaration_health", {})
|
|
if check_declarations and declaration_health.get("duplicate_ids"):
|
|
failures.append(
|
|
f"{len(declaration_health['duplicate_ids'])} platform interface "
|
|
"IDs are declared more than once"
|
|
)
|
|
if check_declarations and declaration_health.get("undeclared_source_surfaces"):
|
|
failures.append(
|
|
f"{len(declaration_health['undeclared_source_surfaces'])} public "
|
|
"WebUI surfaces have no runtime manifest declaration"
|
|
)
|
|
if check_declarations and declaration_health.get("stale_runtime_routes"):
|
|
failures.append(
|
|
f"{len(declaration_health['stale_runtime_routes'])} runtime route "
|
|
"declarations have no WebUI implementation"
|
|
)
|
|
runtime_comparison = inventory.get("runtime_comparison")
|
|
if (
|
|
check_declarations
|
|
and runtime_comparison is not None
|
|
and runtime_comparison.get("mismatches")
|
|
):
|
|
failures.append(
|
|
f"{len(runtime_comparison['mismatches'])} runtime catalog entries "
|
|
"do not match the release inventory"
|
|
)
|
|
return failures
|
|
|
|
|
|
def _resolve_workspace_root(catalog: dict[str, Any]) -> Path:
|
|
sibling_root = META_ROOT.parent.resolve()
|
|
configured_root = Path(str(catalog["default_parent"])).expanduser().resolve()
|
|
repositories = catalog.get("repositories")
|
|
if not isinstance(repositories, list):
|
|
raise ValueError("repository catalog has no repositories array")
|
|
|
|
def source_count(root: Path) -> int:
|
|
return sum(
|
|
1
|
|
for item in repositories
|
|
if isinstance(item, dict)
|
|
and isinstance(item.get("path"), str)
|
|
and (root / item["path"] / "src").is_dir()
|
|
)
|
|
|
|
sibling_count = source_count(sibling_root)
|
|
configured_count = source_count(configured_root)
|
|
return sibling_root if sibling_count >= configured_count else configured_root
|
|
|
|
|
|
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]
|
|
from govoplan_core.core.platform_interfaces import ( # noqa: PLC0415
|
|
manifest_interface_catalog,
|
|
)
|
|
|
|
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
|
|
],
|
|
"architecture": (
|
|
manifest.architecture.to_dict()
|
|
if manifest.architecture is not None
|
|
else None
|
|
),
|
|
"information_governance": (
|
|
manifest.information_governance.to_dict()
|
|
),
|
|
"interface_catalog": manifest_interface_catalog(manifest),
|
|
"frontend": (
|
|
{
|
|
"package": frontend.package_name,
|
|
"routes": [
|
|
_plain_value(route) for route in frontend.routes
|
|
],
|
|
"public_routes": [
|
|
_plain_value(route)
|
|
for route in frontend.public_routes
|
|
],
|
|
"nav_items": [
|
|
_plain_value(item) for item in frontend.nav_items
|
|
],
|
|
"settings_routes": [
|
|
_plain_value(route)
|
|
for route in frontend.settings_routes
|
|
],
|
|
"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]],
|
|
runtime_snapshot: dict[str, Any] | None = None,
|
|
) -> 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(set(catalog_keys) | set(REQUIRED_LOCALES))
|
|
missing_catalog_entries = [
|
|
{
|
|
"key": key,
|
|
"missing_locales": [
|
|
locale
|
|
for locale in expected_locales
|
|
if key not in catalog_keys.get(locale, set())
|
|
],
|
|
}
|
|
for key in sorted(usages)
|
|
if any(key not in catalog_keys.get(locale, set()) for locale in expected_locales)
|
|
]
|
|
fields = webui["fields"]
|
|
help_candidates = [field for field in fields if field["helpCandidate"]]
|
|
dynamic_help = [field for field in fields if field.get("helpDynamic")]
|
|
governance_adoption = Counter(
|
|
dimension["adoption"]
|
|
for manifest in manifests
|
|
for dimension in manifest["information_governance"]["dimensions"].values()
|
|
)
|
|
source_declarations = _source_interface_declarations(webui, manifests)
|
|
declaration_health = _declaration_health(source_declarations, manifests)
|
|
runtime_comparison = (
|
|
_compare_runtime_snapshot(runtime_snapshot, manifests)
|
|
if runtime_snapshot is not None
|
|
else None
|
|
)
|
|
return {
|
|
"schema_version": 2,
|
|
"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.",
|
|
"Low-level field and action IDs use line-independent source anchors unless an explicit interfaceId, DOM id, or name is declared.",
|
|
],
|
|
},
|
|
"modules": manifests,
|
|
"interface_declarations": source_declarations,
|
|
"declaration_health": declaration_health,
|
|
"runtime_comparison": runtime_comparison,
|
|
"ui": {
|
|
"fields": fields,
|
|
"actions": webui.get("actions", []),
|
|
"labels": webui["labels"],
|
|
"visible_text": webui["visibleText"],
|
|
"routes": webui["routes"],
|
|
"navigation": webui["navigation"],
|
|
"ui_capabilities": webui["uiCapabilities"],
|
|
"help_candidates": help_candidates,
|
|
"dynamic_help": dynamic_help,
|
|
"contributions": webui.get("contributions", []),
|
|
},
|
|
"translations": {
|
|
"catalog": catalogs,
|
|
"usages": webui["translationUsages"],
|
|
"dynamic_usages": webui["dynamicTranslationUsages"],
|
|
},
|
|
"translation_health": {
|
|
"locales": expected_locales,
|
|
"reference_locale": REFERENCE_LOCALE,
|
|
"reference_locale_entries": len(catalog_keys.get(REFERENCE_LOCALE, set())),
|
|
"reference_locale_complete": not any(
|
|
REFERENCE_LOCALE in item["missing_locales"]
|
|
for item in missing_catalog_entries
|
|
),
|
|
"used_keys": len(usages),
|
|
"missing_catalog_entries": missing_catalog_entries,
|
|
},
|
|
"information_governance_health": {
|
|
"dimensions": len(manifests) * 4,
|
|
"adoption_counts": dict(sorted(governance_adoption.items())),
|
|
"modules": [
|
|
{
|
|
"module_id": manifest["id"],
|
|
"dimensions": manifest["information_governance"]["dimensions"],
|
|
}
|
|
for manifest in manifests
|
|
],
|
|
},
|
|
"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),
|
|
"ui_fields_with_resolvable_f1_context": len(fields),
|
|
"help_review_candidates": len(help_candidates),
|
|
"dynamic_help_references": len(dynamic_help),
|
|
"ui_actions": len(webui.get("actions", [])),
|
|
"interface_declarations": len(source_declarations),
|
|
"duplicate_interface_ids": len(declaration_health["duplicate_ids"]),
|
|
"undeclared_source_surfaces": len(
|
|
declaration_health["undeclared_source_surfaces"]
|
|
),
|
|
"stale_runtime_routes": len(
|
|
declaration_health["stale_runtime_routes"]
|
|
),
|
|
"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),
|
|
"information_governance_dimensions": len(manifests) * 4,
|
|
"information_governance_enforced": governance_adoption["enforced"],
|
|
"information_governance_partial": governance_adoption["partial"],
|
|
"information_governance_contract_only": governance_adoption[
|
|
"contract_only"
|
|
],
|
|
},
|
|
}
|
|
|
|
|
|
def _source_interface_declarations(
|
|
webui: dict[str, Any],
|
|
manifests: list[dict[str, Any]],
|
|
) -> list[dict[str, Any]]:
|
|
module_by_repository = {
|
|
str(manifest["repository"]): str(manifest["id"])
|
|
for manifest in manifests
|
|
}
|
|
declarations: list[dict[str, Any]] = []
|
|
|
|
def module_id(repository: str) -> str:
|
|
if repository in module_by_repository:
|
|
return module_by_repository[repository]
|
|
if repository == "govoplan-core":
|
|
return "core"
|
|
return repository.removeprefix("govoplan-").replace("-", "_")
|
|
|
|
def source_evidence(item: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
key: item[key]
|
|
for key in ("repository", "file", "line", "column")
|
|
if key in item
|
|
}
|
|
|
|
for kind, items in (
|
|
("field", webui["fields"]),
|
|
("action", webui.get("actions", [])),
|
|
):
|
|
for item in items:
|
|
repository = str(item["repository"])
|
|
owner = module_id(repository)
|
|
raw_id = str(item["id"])
|
|
stable_id = (
|
|
f"{owner}.{raw_id[len(repository) + 1:]}"
|
|
if raw_id.startswith(f"{repository}.")
|
|
else _namespaced_interface_id(owner, kind, raw_id)
|
|
)
|
|
declarations.append(
|
|
{
|
|
"key": f"{kind}:{stable_id}",
|
|
"id": stable_id,
|
|
"module_id": owner,
|
|
"kind": kind,
|
|
"origin": "webui_source",
|
|
"id_source": item.get("idSource", "source_anchor"),
|
|
"explicit_id": item.get("explicitId"),
|
|
"context": item.get("context"),
|
|
**source_evidence(item),
|
|
}
|
|
)
|
|
if kind == "field" and item.get("helpId"):
|
|
help_raw_id = str(item["helpId"])
|
|
help_id = (
|
|
f"{owner}.{help_raw_id[len(repository) + 1:]}"
|
|
if help_raw_id.startswith(f"{repository}.")
|
|
else _namespaced_interface_id(owner, "help", help_raw_id)
|
|
)
|
|
declarations.append(
|
|
{
|
|
"key": f"help:{help_id}",
|
|
"id": help_id,
|
|
"module_id": owner,
|
|
"kind": "help",
|
|
"origin": "webui_source",
|
|
"field_id": stable_id,
|
|
"dynamic": bool(item.get("helpDynamic")),
|
|
**source_evidence(item),
|
|
}
|
|
)
|
|
|
|
for item in webui.get("contributions", []):
|
|
repository = str(item["repository"])
|
|
owner = module_id(repository)
|
|
kind = str(item["kind"])
|
|
raw_id = str(item["id"])
|
|
path = item.get("path")
|
|
if kind == "frontend_route" and isinstance(path, str):
|
|
stable_id = f"{owner}.route.{_surface_slug(path)}"
|
|
elif kind == "public_route" and isinstance(path, str):
|
|
stable_id = f"{owner}.public.{_surface_slug(path)}"
|
|
elif kind == "navigation" and isinstance(path, str):
|
|
stable_id = f"{owner}.nav.{_surface_slug(path)}"
|
|
else:
|
|
stable_id = _namespaced_interface_id(owner, kind, raw_id)
|
|
declarations.append(
|
|
{
|
|
"key": f"{kind}:{stable_id}",
|
|
"id": stable_id,
|
|
"module_id": owner,
|
|
"kind": kind,
|
|
"origin": "webui_contribution",
|
|
"declared_value": raw_id,
|
|
**({"path": path} if isinstance(path, str) else {}),
|
|
**source_evidence(item),
|
|
}
|
|
)
|
|
|
|
translation_entries: dict[tuple[str, str], dict[str, Any]] = {}
|
|
for locale, entries in webui["translationCatalog"].items():
|
|
for key, item in entries.items():
|
|
repository = str(item["repository"])
|
|
owner = module_id(repository)
|
|
declaration_key = (owner, str(key))
|
|
declaration = translation_entries.setdefault(
|
|
declaration_key,
|
|
{
|
|
"key": f"translation:{key}",
|
|
"id": key,
|
|
"module_id": owner,
|
|
"kind": "translation",
|
|
"origin": "translation_catalog",
|
|
"locales": [],
|
|
**source_evidence(item),
|
|
},
|
|
)
|
|
declaration["locales"].append(locale)
|
|
declarations.extend(translation_entries.values())
|
|
|
|
return sorted(
|
|
declarations,
|
|
key=lambda item: (
|
|
item["module_id"],
|
|
item["kind"],
|
|
item["id"],
|
|
item.get("file", ""),
|
|
item.get("line", 0),
|
|
),
|
|
)
|
|
|
|
|
|
def _declaration_health(
|
|
source_declarations: list[dict[str, Any]],
|
|
manifests: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = {}
|
|
for declaration in source_declarations:
|
|
key = (
|
|
str(declaration["module_id"]),
|
|
str(declaration["kind"]),
|
|
str(declaration["id"]),
|
|
)
|
|
grouped.setdefault(key, []).append(declaration)
|
|
duplicate_ids = [
|
|
{
|
|
"module_id": key[0],
|
|
"kind": key[1],
|
|
"id": key[2],
|
|
"evidence": values,
|
|
}
|
|
for key, values in sorted(grouped.items())
|
|
if len(values) > 1
|
|
]
|
|
|
|
comparable_kinds = {
|
|
"frontend_route",
|
|
"navigation",
|
|
"public_route",
|
|
"view_surface",
|
|
}
|
|
source_surfaces = {
|
|
(str(item["module_id"]), str(item["kind"]), str(item["id"])): item
|
|
for item in source_declarations
|
|
if item["origin"] == "webui_contribution"
|
|
and item["kind"] in comparable_kinds
|
|
}
|
|
runtime_surfaces: dict[tuple[str, str, str], dict[str, Any]] = {}
|
|
for manifest in manifests:
|
|
catalog = manifest["interface_catalog"]
|
|
for declaration in catalog["declarations"]:
|
|
if declaration["kind"] not in comparable_kinds:
|
|
continue
|
|
key = (
|
|
str(manifest["id"]),
|
|
str(declaration["kind"]),
|
|
str(declaration["id"]),
|
|
)
|
|
runtime_surfaces[key] = {
|
|
"repository": manifest["repository"],
|
|
**declaration,
|
|
}
|
|
surface_id = declaration.get("metadata", {}).get("surface_id")
|
|
if (
|
|
declaration["kind"]
|
|
in {"frontend_route", "navigation", "settings_route"}
|
|
and isinstance(surface_id, str)
|
|
and surface_id
|
|
):
|
|
runtime_surfaces[
|
|
(str(manifest["id"]), "view_surface", surface_id)
|
|
] = {
|
|
"repository": manifest["repository"],
|
|
"key": f"view_surface:{surface_id}",
|
|
"id": surface_id,
|
|
"module_id": manifest["id"],
|
|
"kind": "view_surface",
|
|
"metadata": {
|
|
"derived_from": declaration["kind"],
|
|
"path": declaration.get("path"),
|
|
},
|
|
}
|
|
|
|
undeclared_source_surfaces = [
|
|
source_surfaces[key]
|
|
for key in sorted(source_surfaces.keys() - runtime_surfaces.keys())
|
|
]
|
|
route_kinds = {"frontend_route", "public_route"}
|
|
stale_runtime_routes = [
|
|
runtime_surfaces[key]
|
|
for key in sorted(runtime_surfaces.keys() - source_surfaces.keys())
|
|
if key[1] in route_kinds
|
|
]
|
|
return {
|
|
"duplicate_ids": duplicate_ids,
|
|
"undeclared_source_surfaces": undeclared_source_surfaces,
|
|
"stale_runtime_routes": stale_runtime_routes,
|
|
"source_declaration_count": len(source_declarations),
|
|
"runtime_declaration_count": sum(
|
|
len(manifest["interface_catalog"]["declarations"])
|
|
for manifest in manifests
|
|
),
|
|
}
|
|
|
|
|
|
def _compare_runtime_snapshot(
|
|
snapshot: dict[str, Any],
|
|
manifests: list[dict[str, Any]],
|
|
) -> dict[str, Any]:
|
|
static_by_module = {
|
|
str(manifest["id"]): manifest["interface_catalog"]
|
|
for manifest in manifests
|
|
}
|
|
modules = snapshot.get("modules")
|
|
if not isinstance(modules, list):
|
|
raise ValueError("Runtime interface snapshot must contain a modules list.")
|
|
mismatches: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
matched: list[str] = []
|
|
for item in modules:
|
|
if not isinstance(item, dict) or not isinstance(item.get("module_id"), str):
|
|
raise ValueError("Runtime interface snapshot has an invalid module entry.")
|
|
module_id = item["module_id"]
|
|
if module_id in seen:
|
|
mismatches.append({"module_id": module_id, "reason": "duplicate_module"})
|
|
continue
|
|
seen.add(module_id)
|
|
expected = static_by_module.get(module_id)
|
|
if expected is None:
|
|
mismatches.append({"module_id": module_id, "reason": "unknown_module"})
|
|
continue
|
|
for field in ("contract_version", "module_version", "digest"):
|
|
if item.get(field) != expected.get(field):
|
|
mismatches.append(
|
|
{
|
|
"module_id": module_id,
|
|
"reason": f"{field}_mismatch",
|
|
"expected": expected.get(field),
|
|
"actual": item.get(field),
|
|
}
|
|
)
|
|
if not any(
|
|
mismatch["module_id"] == module_id for mismatch in mismatches
|
|
):
|
|
matched.append(module_id)
|
|
return {
|
|
"contract_version": snapshot.get("contract_version"),
|
|
"matched_modules": sorted(matched),
|
|
"mismatches": mismatches,
|
|
}
|
|
|
|
|
|
def _load_runtime_snapshot(path: Path) -> dict[str, Any]:
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8"))
|
|
except FileNotFoundError as exc:
|
|
raise ValueError(f"Runtime interface snapshot does not exist: {path}") from exc
|
|
except json.JSONDecodeError as exc:
|
|
raise ValueError(f"Runtime interface snapshot is invalid JSON: {exc}") from exc
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("Runtime interface snapshot must be a JSON object.")
|
|
return payload
|
|
|
|
|
|
def _namespaced_interface_id(module_id: str, kind: str, value: str) -> str:
|
|
if value.startswith(f"{module_id}."):
|
|
return value
|
|
return f"{module_id}.{kind}.{_surface_slug(value)}"
|
|
|
|
|
|
def _surface_slug(value: str) -> str:
|
|
normalized = re.sub(r"[^a-z0-9]+", ".", value.strip().lower()).strip(".")
|
|
return normalized or "root"
|
|
|
|
|
|
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"- UI actions: {summary['ui_actions']}",
|
|
f"- Fields with statically associated help: {summary['ui_fields_with_static_help']}",
|
|
f"- Fields with a resolvable F1 context: {summary['ui_fields_with_resolvable_f1_context']}",
|
|
f"- Fields with dynamic help references: {summary['dynamic_help_references']}",
|
|
f"- Help review candidates: {summary['help_review_candidates']}",
|
|
f"- Stable interface declarations: {summary['interface_declarations']}",
|
|
f"- Duplicate interface IDs: {summary['duplicate_interface_ids']}",
|
|
f"- WebUI surfaces missing runtime declarations: {summary['undeclared_source_surfaces']}",
|
|
f"- Runtime routes missing WebUI implementations: {summary['stale_runtime_routes']}",
|
|
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"- Reference locale: `{inventory['translation_health']['reference_locale']}`",
|
|
f"- Reference locale complete: `{str(inventory['translation_health']['reference_locale_complete']).lower()}`",
|
|
f"- Used translation keys missing from a required locale catalog: {len(missing)}",
|
|
f"- Information-governance dimensions enforced: {summary['information_governance_enforced']}",
|
|
f"- Information-governance dimensions partial: {summary['information_governance_partial']}",
|
|
f"- Information-governance dimensions contract-only: {summary['information_governance_contract_only']}",
|
|
"",
|
|
"## 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(
|
|
[
|
|
"",
|
|
"## Declaration Reconciliation",
|
|
"",
|
|
"Routes, navigation, View surfaces, fields, actions, help references,",
|
|
"translations, settings, widgets, search objects, and backend",
|
|
"capabilities use normalized stable IDs. CI rejects duplicate IDs,",
|
|
"WebUI public surfaces absent from runtime metadata, and runtime routes",
|
|
"without a WebUI implementation.",
|
|
"",
|
|
"",
|
|
"## 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 or typed PlatformWebModule metadata to become",
|
|
"canonically visible. The generated files are release evidence, not an",
|
|
"editable source of platform behavior.",
|
|
"",
|
|
]
|
|
)
|
|
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())
|