from __future__ import annotations import hashlib from dataclasses import dataclass from typing import Annotated, Any, Literal, Mapping from pydantic import BaseModel, ConfigDict, Field from govoplan_core.core.modules import ( DocumentationCondition, DocumentationSourceDefinition, DocumentationType, ModuleManifest, ) from govoplan_core.security.redaction import redact_secret_values class _SourceModel(BaseModel): model_config = ConfigDict(extra="forbid") class DocumentationSourceProvenance(_SourceModel): source: str version: str | None = None revision: str | None = None published_at: str | None = None checksum: str | None = None class DocumentationSourceVisibility(_SourceModel): documentation_types: list[DocumentationType] required_modules: list[str] = Field(default_factory=list) any_modules: list[str] = Field(default_factory=list) missing_modules: list[str] = Field(default_factory=list) required_capabilities: list[str] = Field(default_factory=list) required_scopes: list[str] = Field(default_factory=list) any_scopes: list[str] = Field(default_factory=list) configuration_keys: list[str] = Field(default_factory=list) class ManifestInspection(_SourceModel): kind: Literal["manifest"] = "manifest" module_id: str name: str version: str dependencies: list[str] optional_dependencies: list[str] class RouteInspection(_SourceModel): kind: Literal["route"] = "route" path: str component: str | None = None order: int class CapabilityInspection(_SourceModel): kind: Literal["capability"] = "capability" capability: str label: str | None = None summary: str | None = None contract_version: str | None = None stability: Literal["experimental", "stable", "deprecated"] | None = None audience: list[str] = Field(default_factory=list) class PolicyInspection(_SourceModel): kind: Literal["policy"] = "policy" capability: str label: str | None = None summary: str | None = None contract_version: str | None = None stability: Literal["experimental", "stable", "deprecated"] | None = None audience: list[str] = Field(default_factory=list) class ReleaseCatalogEntryInspection(_SourceModel): id: str name: str version: str | None = None description: str | None = None action: str | None = None tags: list[str] = Field(default_factory=list) class ReleaseCatalogInspection(_SourceModel): kind: Literal["release_catalog"] = "release_catalog" catalog_type: Literal["modules", "configuration_packages"] channel: str | None = None sequence: int | None = None generated_at: str | None = None entry_count: int signed: bool trusted: bool cache_used: bool warnings: list[str] = Field(default_factory=list) entries: list[ReleaseCatalogEntryInspection] = Field(default_factory=list) class ConfigurationPackageInspection(_SourceModel): kind: Literal["configuration_package"] = "configuration_package" package_id: str name: str | None = None version: str | None = None schema_version: str | None = None description: str | None = None publisher: str | None = None category: str | None = None tags: list[str] = Field(default_factory=list) required_modules: list[str] = Field(default_factory=list) required_capabilities: list[str] = Field(default_factory=list) class WikiInspection(_SourceModel): kind: Literal["wiki"] = "wiki" href: str published_at: str | None = None revision: str | None = None class RepositoryInspection(_SourceModel): kind: Literal["repository"] = "repository" href: str revision: str | None = None DocumentationSourceInspection = Annotated[ ManifestInspection | RouteInspection | CapabilityInspection | PolicyInspection | ReleaseCatalogInspection | ConfigurationPackageInspection | WikiInspection | RepositoryInspection, Field(discriminator="kind"), ] class DocumentationSourceItem(_SourceModel): id: str kind: str owner_module_id: str label: str state: Literal["configured", "disabled", "unavailable"] state_reason: str | None = None inspection_url: str provenance: DocumentationSourceProvenance visibility: DocumentationSourceVisibility inspection: DocumentationSourceInspection def summary(self) -> dict[str, Any]: return self.model_dump(mode="json", exclude={"inspection"}) @dataclass(frozen=True, slots=True) class RegisteredDocumentationSource: item: DocumentationSourceItem condition: DocumentationCondition documentation_types: tuple[DocumentationType, ...] configuration_key: str | None = None class DocumentationSourceRegistry: def __init__(self) -> None: self._sources: dict[str, RegisteredDocumentationSource] = {} def register(self, source: RegisteredDocumentationSource) -> None: if source.item.id in self._sources: raise ValueError(f"Duplicate documentation source id: {source.item.id}") self._sources[source.item.id] = source def sources(self) -> tuple[RegisteredDocumentationSource, ...]: return tuple( self._sources[source_id] for source_id in sorted(self._sources) ) def get(self, source_id: str) -> RegisteredDocumentationSource | None: return self._sources.get(source_id) def build_documentation_source_registry( manifests: tuple[ModuleManifest, ...], *, include_runtime_catalogs: bool = True, ) -> DocumentationSourceRegistry: registry = DocumentationSourceRegistry() for manifest in manifests: registry.register(_manifest_source(manifest)) for source in _route_sources(manifest): registry.register(source) for source in _capability_sources(manifest): registry.register(source) for definition in manifest.documentation_sources: registry.register(_defined_source(manifest, definition)) for source in _linked_sources(manifest): if registry.get(source.item.id) is None: registry.register(source) if include_runtime_catalogs: for source in _catalog_sources(): registry.register(source) return registry def _manifest_source(manifest: ModuleManifest) -> RegisteredDocumentationSource: return _registered_source( source_id=f"{manifest.id}.manifest", kind="manifest", owner_module_id=manifest.id, label=f"{manifest.name} module manifest", provenance={"source": "module_manifest", "version": manifest.version}, inspection=ManifestInspection( module_id=manifest.id, name=manifest.name, version=manifest.version, dependencies=list(manifest.dependencies), optional_dependencies=list(manifest.optional_dependencies), ), ) def _route_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]: if manifest.frontend is None: return () return tuple( _registered_source( source_id=_derived_source_id(manifest.id, "route", route.path), kind="route", owner_module_id=manifest.id, label=f"{manifest.name} route {route.path}", provenance={"source": "frontend_manifest", "version": manifest.version}, inspection=RouteInspection( path=route.path, component=route.component, order=route.order, ), condition=DocumentationCondition( required_modules=(manifest.id,), required_scopes=route.required_all, any_scopes=route.required_any, ), ) for route in manifest.frontend.routes ) def _capability_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]: sources: list[RegisteredDocumentationSource] = [] for capability in sorted(manifest.capability_factories): metadata = manifest.capability_documentation.get(capability) is_policy = capability.startswith("policy.") kind = "policy" if is_policy else "capability" inspection: DocumentationSourceInspection = ( PolicyInspection( capability=capability, label=metadata.label if metadata else None, summary=metadata.summary if metadata else None, contract_version=metadata.contract_version if metadata else None, stability=metadata.stability if metadata else None, audience=list(metadata.audience) if metadata else [], ) if is_policy else CapabilityInspection( capability=capability, label=metadata.label if metadata else None, summary=metadata.summary if metadata else None, contract_version=metadata.contract_version if metadata else None, stability=metadata.stability if metadata else None, audience=list(metadata.audience) if metadata else [], ) ) sources.append(_registered_source( source_id=_derived_source_id(manifest.id, kind, capability), kind=kind, owner_module_id=manifest.id, label=metadata.label if metadata else f"{manifest.name} {kind} {capability}", provenance={"source": "module_manifest", "version": manifest.version}, inspection=inspection, condition=DocumentationCondition(required_modules=(manifest.id,)), documentation_types=metadata.documentation_types if metadata else ("admin",), )) return tuple(sources) def _defined_source( manifest: ModuleManifest, definition: DocumentationSourceDefinition, ) -> RegisteredDocumentationSource: inspection = _defined_inspection(definition) return _registered_source( source_id=definition.id, kind=definition.kind, owner_module_id=manifest.id, label=definition.label, provenance={ "source": str(definition.provenance.get("source") or "module_manifest"), **_safe_source_fields(definition.provenance), }, inspection=inspection, condition=definition.condition, documentation_types=definition.documentation_types, state=definition.state, state_reason=definition.state_reason, configuration_key=definition.configuration_key, ) def _defined_inspection(definition: DocumentationSourceDefinition) -> DocumentationSourceInspection: safe = _safe_source_fields(definition.inspection) if definition.kind == "configuration_package": return ConfigurationPackageInspection( package_id=str(safe.get("package_id") or definition.id), name=_optional_text(safe.get("name")), version=_optional_text(safe.get("version")), schema_version=_optional_text(safe.get("schema_version")), description=_optional_text(safe.get("description")), publisher=_optional_text(safe.get("publisher")), category=_optional_text(safe.get("category")), tags=_string_list(safe.get("tags")), required_modules=_string_list(safe.get("required_modules")), required_capabilities=_string_list(safe.get("required_capabilities")), ) href = definition.link.href if definition.link else str(safe.get("href") or "") if definition.kind == "wiki": return WikiInspection( href=href, published_at=_optional_text(safe.get("published_at")), revision=_optional_text(safe.get("revision")), ) if definition.kind == "repository": return RepositoryInspection( href=href, revision=_optional_text(safe.get("revision")), ) if definition.kind == "route": return RouteInspection( path=str(safe.get("path") or ""), component=_optional_text(safe.get("component")), order=int(safe.get("order") or 0), ) if definition.kind == "policy": return PolicyInspection( capability=str(safe.get("capability") or definition.id), label=_optional_text(safe.get("label")), summary=_optional_text(safe.get("summary")), contract_version=_optional_text(safe.get("contract_version")), stability=_capability_stability(safe.get("stability")), audience=_string_list(safe.get("audience")), ) if definition.kind == "capability": return CapabilityInspection( capability=str(safe.get("capability") or definition.id), label=_optional_text(safe.get("label")), summary=_optional_text(safe.get("summary")), contract_version=_optional_text(safe.get("contract_version")), stability=_capability_stability(safe.get("stability")), audience=_string_list(safe.get("audience")), ) return ManifestInspection( module_id=str(safe.get("module_id") or definition.id.split(".", 1)[0]), name=str(safe.get("name") or definition.label), version=str(safe.get("version") or ""), dependencies=_string_list(safe.get("dependencies")), optional_dependencies=_string_list(safe.get("optional_dependencies")), ) def _linked_sources(manifest: ModuleManifest) -> tuple[RegisteredDocumentationSource, ...]: sources: list[RegisteredDocumentationSource] = [] for topic in manifest.documentation: for link in topic.links: if link.kind not in {"wiki", "repository"}: continue source_id = _derived_source_id(manifest.id, link.kind, link.href) inspection: DocumentationSourceInspection = ( WikiInspection(href=link.href) if link.kind == "wiki" else RepositoryInspection(href=link.href) ) sources.append(_registered_source( source_id=source_id, kind=link.kind, owner_module_id=manifest.id, label=link.label, provenance={"source": "documentation_link", "version": manifest.version}, inspection=inspection, condition=topic.conditions[0] if len(topic.conditions) == 1 else DocumentationCondition(), documentation_types=topic.documentation_types, )) return tuple(sources) def _registered_source( *, source_id: str, kind: str, owner_module_id: str, label: str, provenance: Mapping[str, Any], inspection: DocumentationSourceInspection, condition: DocumentationCondition | None = None, documentation_types: tuple[DocumentationType, ...] = ("admin",), state: Literal["configured", "disabled", "unavailable"] = "configured", state_reason: str | None = None, configuration_key: str | None = None, ) -> RegisteredDocumentationSource: clean_provenance = _safe_source_fields(provenance) return RegisteredDocumentationSource( item=DocumentationSourceItem( id=source_id, kind=kind, owner_module_id=owner_module_id, label=label, state=state, state_reason=state_reason, inspection_url=f"/api/v1/docs/sources/{source_id}", provenance=DocumentationSourceProvenance( source=str(clean_provenance.get("source") or "unknown"), version=_optional_text(clean_provenance.get("version")), revision=_optional_text(clean_provenance.get("revision")), published_at=_optional_text(clean_provenance.get("published_at")), checksum=_optional_text(clean_provenance.get("checksum")), ), visibility=_visibility_payload(condition or DocumentationCondition(), documentation_types), inspection=inspection, ), condition=condition or DocumentationCondition(), documentation_types=documentation_types, configuration_key=configuration_key, ) def _catalog_sources() -> tuple[RegisteredDocumentationSource, ...]: from govoplan_core.core.configuration_packages import validate_configuration_package_catalog from govoplan_core.core.module_package_catalog import validate_module_package_catalog return ( *_release_catalog_sources( "modules", "Module release catalog", validate_module_package_catalog(), ), *_release_catalog_sources( "configuration_packages", "Configuration package catalog", validate_configuration_package_catalog(), ), ) def _release_catalog_sources( catalog_type: Literal["modules", "configuration_packages"], label: str, validation: Mapping[str, object], ) -> tuple[RegisteredDocumentationSource, ...]: entry_key = "modules" if catalog_type == "modules" else "packages" raw_entries = validation.get(entry_key) entries = [item for item in raw_entries if isinstance(item, Mapping)] if isinstance(raw_entries, list) else [] configured = bool(validation.get("configured")) valid = bool(validation.get("valid")) state: Literal["configured", "disabled", "unavailable"] = ( "configured" if configured and valid else "unavailable" if configured else "disabled" ) reason = _optional_text(validation.get("error")) if state == "disabled": reason = "No catalog source is configured." catalog_id = f"docs.release-catalog.{catalog_type.replace('_', '-')}" catalog_source = _registered_source( source_id=catalog_id, kind="release_catalog", owner_module_id="docs", label=label, provenance={ "source": "core_catalog_contract", "published_at": validation.get("generated_at"), }, inspection=ReleaseCatalogInspection( catalog_type=catalog_type, channel=_optional_text(validation.get("channel")), sequence=_optional_int(validation.get("sequence")), generated_at=_optional_text(validation.get("generated_at")), entry_count=len(entries), signed=bool(validation.get("signed")), trusted=bool(validation.get("trusted")), cache_used=bool(validation.get("cache_used")), warnings=_string_list(validation.get("warnings")), entries=[_release_catalog_entry(item, catalog_type) for item in entries], ), state=state, state_reason=reason, ) if catalog_type == "modules" or not valid: return (catalog_source,) return ( catalog_source, *( _configuration_package_source(item, validation) for item in entries ), ) def _release_catalog_entry( item: Mapping[str, object], catalog_type: Literal["modules", "configuration_packages"], ) -> ReleaseCatalogEntryInspection: item_id = ( _optional_text(item.get("module_id")) if catalog_type == "modules" else _optional_text(item.get("package_id")) ) return ReleaseCatalogEntryInspection( id=item_id or "unknown", name=_optional_text(item.get("name")) or item_id or "Unknown", version=_optional_text(item.get("version")), description=_optional_text(item.get("description")), action=_optional_text(item.get("action")), tags=_string_list(item.get("tags")), ) def _configuration_package_source( item: Mapping[str, object], validation: Mapping[str, object], ) -> RegisteredDocumentationSource: package_id = _optional_text(item.get("package_id")) or "unknown" required_modules = [ str(requirement.get("module_id")) for requirement in item.get("required_modules", ()) if isinstance(requirement, Mapping) and requirement.get("module_id") ] if isinstance(item.get("required_modules"), (list, tuple)) else [] return _registered_source( source_id=_derived_source_id("docs", "configuration-package", package_id), kind="configuration_package", owner_module_id="docs", label=_optional_text(item.get("name")) or package_id, provenance={ "source": "configuration_package_catalog", "version": item.get("version"), "published_at": validation.get("generated_at"), }, inspection=ConfigurationPackageInspection( package_id=package_id, name=_optional_text(item.get("name")), version=_optional_text(item.get("version")), schema_version=_optional_text(item.get("schema_version")), description=_optional_text(item.get("description")), publisher=_optional_text(item.get("publisher")), category=_optional_text(item.get("category")), tags=_string_list(item.get("tags")), required_modules=required_modules, required_capabilities=_string_list(item.get("required_capabilities")), ), ) def _visibility_payload( condition: DocumentationCondition, documentation_types: tuple[DocumentationType, ...], ) -> DocumentationSourceVisibility: return DocumentationSourceVisibility( documentation_types=list(documentation_types), required_modules=list(condition.required_modules), any_modules=list(condition.any_modules), missing_modules=list(condition.missing_modules), required_capabilities=list(condition.required_capabilities), required_scopes=list(condition.required_scopes), any_scopes=list(condition.any_scopes), configuration_keys=list(condition.configuration_keys), ) def _derived_source_id(module_id: str, kind: str, value: str) -> str: digest = hashlib.sha256(value.encode("utf-8")).hexdigest()[:16] return f"{module_id}.{kind}.{digest}" def _safe_source_fields(value: Mapping[str, Any]) -> dict[str, Any]: redacted = redact_secret_values(dict(value)) if not isinstance(redacted, Mapping): return {} return { str(key): item for key, item in redacted.items() if isinstance(item, (str, int, float, bool, list, tuple)) or item is None } def _optional_text(value: object) -> str | None: if value is None: return None text = str(value).strip() return text[:2_000] if text else None def _optional_int(value: object) -> int | None: if value is None: return None try: return int(value) except (TypeError, ValueError): return None def _capability_stability( value: object, ) -> Literal["experimental", "stable", "deprecated"] | None: text = _optional_text(value) return text if text in {"experimental", "stable", "deprecated"} else None def _string_list(value: object) -> list[str]: if not isinstance(value, (list, tuple)): return [] return [str(item)[:500] for item in value[:100]]