#!/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 import os 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" ) DEFAULT_HIGH_RISK_HELP_BASELINE = ( META_ROOT / "tools" / "inventory" / "high-risk-help-baseline.json" ) REQUIRED_LOCALES = ("de", "en") REFERENCE_LOCALE = "de" def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--workspace-root", type=Path, help="Authoritative directory containing registered checkouts; never falls back to another workspace.", ) 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.", ) parser.add_argument( "--high-risk-help-baseline", type=Path, default=DEFAULT_HIGH_RISK_HELP_BASELINE, help="Versioned upper bound for high-risk controls without exact F1 help.", ) args = parser.parse_args() catalog = json.loads((META_ROOT / "repositories.json").read_text(encoding="utf-8")) workspace_root = _resolve_workspace_root(catalog, args.workspace_root) _validate_repository_roots(catalog, workspace_root) webui = _extract_webui(workspace_root) backend_endpoints = _extract_backend_endpoints(catalog, workspace_root) manifests = _extract_manifests(catalog, workspace_root) endpoint_declarations = _load_endpoint_declarations( args.endpoint_declarations.resolve() ) high_risk_help_baseline = _load_high_risk_help_baseline( args.high_risk_help_baseline.resolve() ) inventory = _assemble_inventory( webui=webui, backend_endpoints=backend_endpoints, manifests=manifests, endpoint_declarations=endpoint_declarations, high_risk_help_baseline=high_risk_help_baseline, runtime_snapshot=( _load_runtime_snapshot(args.runtime_snapshot.resolve()) if args.runtime_snapshot is not None else None ), ) inventory["workspace_root"] = str(workspace_root) inventory["workspace_selection"] = ( "explicit" if args.workspace_root is not None else "legacy-discovery" ) 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" ) help_health = inventory.get("help_health", {}) if check_declarations and help_health.get("invalid_risk_annotations"): failures.append( f"{len(help_health['invalid_risk_annotations'])} controls use an " "unsupported contextual-help risk class" ) if check_declarations and help_health.get("baseline_regression"): failures.append( f"{len(help_health['missing_exact_high_risk_help'])} high-risk " "controls lack exact F1 help; baseline permits at most " f"{help_health['baseline_maximum_missing']}" ) if check_declarations and help_health.get("unresolved_exact_high_risk_help"): failures.append( f"{len(help_health['unresolved_exact_high_risk_help'])} high-risk " "controls reference no manifest DocumentationTopic help context" ) if check_declarations and help_health.get("high_risk_help_without_german"): failures.append( f"{len(help_health['high_risk_help_without_german'])} high-risk " "controls resolve to documentation without complete German content" ) 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], explicit_root: Path | None = None ) -> Path: if explicit_root is not None: root = explicit_root.expanduser().resolve() if not root.is_dir(): raise ValueError("The explicit inventory workspace root must be an existing directory") return root # Compatibility for direct legacy callers only. Managed callers always # supply their selected root; a partial checkout must not borrow sources. 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 _validate_repository_roots(catalog: dict[str, Any], workspace_root: Path) -> None: repositories = catalog.get("repositories") if not isinstance(repositories, list): raise ValueError("repository catalog has no repositories array") for repository in repositories: if not isinstance(repository, dict) or not isinstance(repository.get("path"), str): raise ValueError("Invalid inventory repository path") relative = Path(repository["path"]) if not repository["path"] or relative.is_absolute() or ".." in relative.parts: raise ValueError("Inventory repository paths must remain inside the selected workspace") root = workspace_root / relative # Missing optional checkouts are allowed; links to another checkout are # not evidence for the selected workspace. if not root.resolve().is_relative_to(workspace_root): raise ValueError("Inventory repository path escapes the selected workspace") for source in (root / "src", root / "webui/src"): if not source.resolve().is_relative_to(workspace_root): raise ValueError("Inventory source root escapes the selected workspace") def _extract_webui(workspace_root: Path | None = None) -> dict[str, Any]: helper = META_ROOT / "tools" / "inventory" / "extract-webui-structure.mjs" argv = [os.environ.get("NODE", "node"), str(helper), str(META_ROOT)] if workspace_root is not None: argv.extend(["--workspace-root", str(workspace_root)]) completed = subprocess.run( argv, check=True, capture_output=True, text=True, ) result = json.loads(completed.stdout) if workspace_root is not None and ( not isinstance(result, dict) or result.get("workspaceRoot") != str(workspace_root.resolve()) ): raise ValueError("WebUI collector did not confirm the selected inventory workspace") return result 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")): if not source_path.resolve().is_relative_to(workspace_root): raise ValueError("Backend source path escapes the selected workspace") 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]]: _validate_repository_roots(catalog, workspace_root) source_roots = [ workspace_root / repository["path"] / "src" for repository in catalog["repositories"] if (workspace_root / repository["path"] / "src").is_dir() ] core_root = next( (workspace_root / repository["path"] / "src" for repository in catalog["repositories"] if repository.get("name") == "govoplan-core"), workspace_root / "govoplan-core/src", ) if not (core_root / "govoplan_core/core/platform_interfaces.py").is_file(): raise ValueError("Inventory requires Core interface sources in the selected workspace") _assert_workspace_imports(workspace_root) sys.path[:0] = [str(path) for path in source_roots] from govoplan_core.core.platform_interfaces import ( # noqa: PLC0415 manifest_interface_catalog, ) _assert_workspace_imports(workspace_root) 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) source = getattr(loaded, "__file__", None) if not isinstance(source, str) or Path(source).resolve() != manifest_path.resolve(): raise ValueError("Manifest import did not resolve to its selected workspace source") _assert_workspace_imports(workspace_root) manifest = loaded.get_manifest() _assert_workspace_imports(workspace_root) 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 ], "documentation": [ { "id": topic.id, "help_contexts": sorted( { str(context) for context in topic.metadata.get( "help_contexts", () ) if isinstance(context, str) and context.strip() } ), "german_complete": ( isinstance(topic.translations.get("de"), dict) and all( isinstance(topic.translations["de"].get(field), str) and topic.translations["de"][field].strip() for field in ("title", "summary", "body") ) ), } for topic in manifest.documentation ], "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 _assert_workspace_imports(workspace_root: Path) -> None: # An editable installation or cached import must not stand in for a missing # optional checkout. Direct callers with another workspace use a fresh # process instead of replacing already-loaded application packages. for name, module in list(sys.modules.items()): # The Meta tools may audit a separate checkout; they are not module # contributions and must not be confused with application packages. package = name.partition(".")[0] if not package.startswith("govoplan_") or package in { "govoplan_devkit", "govoplan_release" }: continue filename = getattr(module, "__file__", None) locations = list(getattr(module, "__path__", ())) if isinstance(filename, str): locations.append(filename) if any(not Path(location).resolve().is_relative_to(workspace_root) for location in locations): raise ValueError("A GovOPlaN import originates outside the selected inventory workspace") 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]], high_risk_help_baseline: dict[str, Any] | None = None, 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"] actions = webui.get("actions", []) help_candidates = [field for field in fields if field["helpCandidate"]] dynamic_help = [field for field in fields if field.get("helpDynamic")] controls = [*fields, *actions] high_risk_controls = [item for item in controls if item.get("helpRisk")] missing_exact_high_risk_help = [ item for item in controls if item.get("highRiskHelpMissing") ] invalid_risk_annotations = [ item for item in controls if item.get("helpRiskSource") == "invalid_explicit" ] documentation_contexts = { context: { "module_id": manifest["id"], "topic_id": topic["id"], "german_complete": topic["german_complete"], } for manifest in manifests for topic in manifest.get("documentation", []) for context in topic.get("help_contexts", []) } unresolved_exact_high_risk_help = [ item for item in high_risk_controls if item.get("helpExact") and not item.get("helpContextDynamic") and item.get("helpContextId") not in documentation_contexts ] high_risk_help_without_german = [ item for item in high_risk_controls if item.get("helpContextId") in documentation_contexts and not documentation_contexts[item["helpContextId"]]["german_complete"] ] baseline_maximum_missing = ( high_risk_help_baseline["maximum_missing_exact_help"] if high_risk_help_baseline is not None else None ) 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, "help_health": { "supported_risk_classes": sorted( { str(item["helpRisk"]) for item in high_risk_controls if item.get("helpRisk") } ), "high_risk_controls": high_risk_controls, "missing_exact_high_risk_help": missing_exact_high_risk_help, "invalid_risk_annotations": invalid_risk_annotations, "unresolved_exact_high_risk_help": unresolved_exact_high_risk_help, "high_risk_help_without_german": high_risk_help_without_german, "dynamic_owner_context_controls": [ item for item in high_risk_controls if item.get("helpContextDynamic") ], "baseline_maximum_missing": baseline_maximum_missing, "baseline_regression": ( baseline_maximum_missing is not None and len(missing_exact_high_risk_help) > baseline_maximum_missing ), }, "runtime_comparison": runtime_comparison, "ui": { "fields": fields, "actions": 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(actions), "high_risk_controls": len(high_risk_controls), "high_risk_controls_with_exact_help": ( len(high_risk_controls) - len(missing_exact_high_risk_help) ), "high_risk_controls_missing_exact_help": len( missing_exact_high_risk_help ), "invalid_help_risk_annotations": len(invalid_risk_annotations), "unresolved_exact_high_risk_help": len( unresolved_exact_high_risk_help ), "high_risk_help_without_german": len(high_risk_help_without_german), "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"] high_risk_by_repository = Counter( item["repository"] for item in inventory["help_health"]["missing_exact_high_risk_help"] ) 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"- High-risk controls: {summary['high_risk_controls']}", f"- High-risk controls with exact F1 help: {summary['high_risk_controls_with_exact_help']}", f"- High-risk controls missing exact F1 help: {summary['high_risk_controls_missing_exact_help']}", f"- Invalid help-risk annotations: {summary['invalid_help_risk_annotations']}", f"- High-risk exact contexts missing a manifest topic: {summary['unresolved_exact_high_risk_help']}", f"- High-risk contexts without complete German topic content: {summary['high_risk_help_without_german']}", 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( [ "", "## High-risk Contextual-help Debt", "", "Inferred or explicitly classified high-risk controls require an exact", "F1 context. `data-help-risk-reviewed=\"standard\"` records a reviewed", "false positive. The versioned baseline makes this queue non-regressing.", "", "| Repository | Missing exact contexts |", "| --- | ---: |", ] ) lines.extend( f"| `{repository}` | {count} |" for repository, count in sorted(high_risk_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_high_risk_help_baseline(path: Path) -> dict[str, Any]: try: payload = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise ValueError( f"High-risk contextual-help baseline does not exist: {path}" ) from exc except json.JSONDecodeError as exc: raise ValueError( f"High-risk contextual-help baseline is invalid JSON: {exc}" ) from exc if not isinstance(payload, dict) or payload.get("schema_version") != 1: raise ValueError( "High-risk contextual-help baseline must use schema_version 1." ) maximum = payload.get("maximum_missing_exact_help") if not isinstance(maximum, int) or isinstance(maximum, bool) or maximum < 0: raise ValueError( "High-risk contextual-help baseline maximum must be a non-negative integer." ) return payload 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())