from __future__ import annotations import hashlib from collections.abc import Mapping from dataclasses import dataclass from urllib.parse import quote from sqlalchemy.orm import Session from govoplan_core.core.semantic_documentation import ( SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION, SemanticDocumentationBreadcrumb, SemanticDocumentationSubjectAnchor, SemanticDocumentationSubjectDescriptor, SemanticDocumentationSubjectPage, SemanticDocumentationSubjectQuery, SemanticDocumentationSubjectReference, SemanticDocumentationSubjectResolution, semantic_documentation_fingerprint, ) from govoplan_forms.backend.db.models import FormDefinitionRevision from govoplan_forms.backend.service import definition_from_mapping SUBJECT_KIND = "form_definition" READ_SCOPE = "forms:definition:read" _MAX_SUBJECTS = 20_000 @dataclass(frozen=True, slots=True) class _IdentityState: field_ids: Mapping[str, str] section_ids: Mapping[tuple[str, str], str] historical_field_ids: frozenset[str] historical_section_ids: frozenset[str] class FormsSemanticDocumentationSubjectProvider: provider_id = "forms.semantic_subjects" module_id = "forms" contract_version = SEMANTIC_DOCUMENTATION_SUBJECT_CONTRACT_VERSION def list_subjects( self, session: object, principal: object, *, request: SemanticDocumentationSubjectQuery, ) -> SemanticDocumentationSubjectPage: if not _authorized(principal, request.tenant_id): return SemanticDocumentationSubjectPage() db = _session(session) if request.subject_kinds and SUBJECT_KIND not in request.subject_kinds: return SemanticDocumentationSubjectPage() rows = ( db.query(FormDefinitionRevision) .filter( FormDefinitionRevision.tenant_id == request.tenant_id, FormDefinitionRevision.superseded_at.is_(None), ) .order_by(FormDefinitionRevision.form_key, FormDefinitionRevision.form_id) .all() ) subjects: list[SemanticDocumentationSubjectDescriptor] = [] for row in rows: definition = definition_from_mapping(row.payload) identities = _identity_state(db, row) subjects.extend(_descriptors(definition, identities)) if len(subjects) > _MAX_SUBJECTS: raise ValueError( "Forms semantic subject limit exceeded; narrow the query." ) query = request.query.casefold().strip() if query: subjects = [ item for item in subjects if query in _descriptor_search_text(item).casefold() ] offset = _cursor_offset(request.cursor) selected = tuple(subjects[offset : offset + request.limit]) next_offset = offset + len(selected) has_more = next_offset < len(subjects) return SemanticDocumentationSubjectPage( subjects=selected, next_cursor=str(next_offset) if has_more else None, has_more=has_more, ) def resolve_subject( self, session: object, principal: object, *, reference: SemanticDocumentationSubjectReference, ) -> SemanticDocumentationSubjectResolution | None: if ( reference.module_id != self.module_id or reference.subject_kind != SUBJECT_KIND or not _authorized(principal, reference.tenant_id) ): return None db = _session(session) row = ( db.query(FormDefinitionRevision) .filter( FormDefinitionRevision.tenant_id == reference.tenant_id, FormDefinitionRevision.form_id == reference.subject_id, FormDefinitionRevision.superseded_at.is_(None), ) .one_or_none() ) if row is None: return SemanticDocumentationSubjectResolution( requested_reference=reference, availability="missing", reason_code="form_missing", ) definition = definition_from_mapping(row.payload) identities = _identity_state(db, row) descriptor = next( ( item for item in _descriptors(definition, identities) if item.reference.stable_key == reference.stable_key ), None, ) if descriptor is None: anchor = reference.anchor reason = "subject_missing" if anchor is not None and anchor.kind == "field": reason = ( "field_deleted" if anchor.id in identities.historical_field_ids else "field_missing" ) elif anchor is not None and anchor.kind == "section": reason = ( "section_deleted" if anchor.id in identities.historical_section_ids else "section_missing" ) return SemanticDocumentationSubjectResolution( requested_reference=reference, availability="missing", reason_code=reason, ) changed = any( expected is not None and expected != actual for expected, actual in ( (reference.observed_revision, descriptor.reference.observed_revision), ( reference.observed_fingerprint, descriptor.reference.observed_fingerprint, ), ) ) return SemanticDocumentationSubjectResolution( requested_reference=reference, availability="changed" if changed else "available", subject=descriptor, ) def _descriptors(definition, identities: _IdentityState): form_reference = _reference( definition, revision=definition.temporal.revision, fingerprint=_form_fingerprint(definition), ) route = f"/forms?formId={quote(definition.reference.object_id, safe='')}" form_labels = _form_labels(definition) form_descriptions = _form_descriptions(definition) result = [ SemanticDocumentationSubjectDescriptor( reference=form_reference, labels=form_labels, descriptions=form_descriptions, route=route, required_scopes=(READ_SCOPE,), ) ] field_locations = _field_locations(definition) for field in definition.fields: identity = identities.field_ids[field.key] page, section = field_locations.get(field.key, (None, None)) labels = _field_labels(definition, field.key, field.label) descriptions = _field_descriptions(definition, field.key, field.help_text) fingerprint = _field_fingerprint(definition, field, page, section, labels) breadcrumbs = [ SemanticDocumentationBreadcrumb( label=_label(form_labels), subject_kind=SUBJECT_KIND, subject_id=definition.reference.object_id, ) ] if page is not None: breadcrumbs.append( SemanticDocumentationBreadcrumb( label=page.title, subject_kind=SUBJECT_KIND, subject_id=definition.reference.object_id, ) ) if section is not None: breadcrumbs.append( SemanticDocumentationBreadcrumb( label=section.title, subject_kind=SUBJECT_KIND, subject_id=definition.reference.object_id, anchor=SemanticDocumentationSubjectAnchor( kind="section", id=identities.section_ids[(page.key, section.key)], ), ) ) result.append( SemanticDocumentationSubjectDescriptor( reference=_reference( definition, anchor=SemanticDocumentationSubjectAnchor( kind="field", id=identity ), revision=fingerprint, fingerprint=fingerprint, ), labels=labels, descriptions=descriptions, breadcrumbs=tuple(breadcrumbs), route=route, route_anchor=f"field-{field.key}", required_scopes=(READ_SCOPE,), ) ) for page in definition.pages: for section in page.sections: identity = identities.section_ids[(page.key, section.key)] labels = _section_labels(definition, section.key, section.title) fingerprint = _section_fingerprint(definition, page, section, labels) result.append( SemanticDocumentationSubjectDescriptor( reference=_reference( definition, anchor=SemanticDocumentationSubjectAnchor( kind="section", id=identity ), revision=fingerprint, fingerprint=fingerprint, ), labels=labels, descriptions=( {_fallback_locale(definition): section.description} if section.description else {} ), breadcrumbs=( SemanticDocumentationBreadcrumb( label=_label(form_labels), subject_kind=SUBJECT_KIND, subject_id=definition.reference.object_id, ), SemanticDocumentationBreadcrumb( label=page.title, subject_kind=SUBJECT_KIND, subject_id=definition.reference.object_id, ), ), route=route, route_anchor=f"section-{page.key}-{section.key}", required_scopes=(READ_SCOPE,), ) ) return tuple(result) def _reference( definition, *, revision: str, fingerprint: str, anchor: SemanticDocumentationSubjectAnchor | None = None, ) -> SemanticDocumentationSubjectReference: return SemanticDocumentationSubjectReference( module_id="forms", tenant_id=definition.reference.tenant_id, subject_kind=SUBJECT_KIND, subject_id=definition.reference.object_id, anchor=anchor, observed_revision=revision, observed_fingerprint=fingerprint, ) def _identity_state( session: Session, current: FormDefinitionRevision, ) -> _IdentityState: rows = ( session.query(FormDefinitionRevision) .filter( FormDefinitionRevision.tenant_id == current.tenant_id, FormDefinitionRevision.form_id == current.form_id, ) .order_by(FormDefinitionRevision.recorded_at, FormDefinitionRevision.id) .all() ) active_fields: dict[str, str] = {} active_sections: dict[tuple[str, str], str] = {} historical_fields: set[str] = set() historical_sections: set[str] = set() for row in rows: definition = definition_from_mapping(row.payload) field_keys = {field.key for field in definition.fields} section_keys = { (page.key, section.key) for page in definition.pages for section in page.sections } active_fields = { key: value for key, value in active_fields.items() if key in field_keys } active_sections = { key: value for key, value in active_sections.items() if key in section_keys } for key in sorted(field_keys): active_fields.setdefault(key, _lineage_id("field", row.id, key)) historical_fields.add(active_fields[key]) for page_key, section_key in sorted(section_keys): key = (page_key, section_key) active_sections.setdefault( key, _lineage_id("section", row.id, page_key, section_key), ) historical_sections.add(active_sections[key]) if row.id == current.id: break return _IdentityState( field_ids=active_fields, section_ids=active_sections, historical_field_ids=frozenset(historical_fields), historical_section_ids=frozenset(historical_sections), ) def _lineage_id(kind: str, *parts: str) -> str: value = "\x1f".join((kind, *parts)).encode() return f"{kind}-{hashlib.sha256(value).hexdigest()[:40]}" def _form_fingerprint(definition) -> str: return semantic_documentation_fingerprint( { "revision": definition.temporal.revision, "publication_state": definition.publication_state, } ) def _field_fingerprint(definition, field, page, section, labels) -> str: return semantic_documentation_fingerprint( { "canonical_label": field.label, "canonical_help": field.help_text, "labels": labels, "help": _field_descriptions(definition, field.key, field.help_text), "value_type": field.value_type, "required": field.required, "options": list(field.options), "constraints": dict(field.constraints), "visibility": ( field.visibility_condition.to_dict() if field.visibility_condition is not None else None ), "page": page.key if page else None, "section": section.key if section else None, } ) def _section_fingerprint(definition, page, section, labels) -> str: return semantic_documentation_fingerprint( { "labels": labels, "description": section.description, "page": page.key, "field_keys": list(section.field_keys), "visibility": ( section.visibility_condition.to_dict() if section.visibility_condition is not None else None ), } ) def _form_labels(definition) -> dict[str, str]: labels = {_fallback_locale(definition): definition.title} for localization in definition.localizations: if localization.title: labels[localization.locale] = localization.title return labels def _form_descriptions(definition) -> dict[str, str]: descriptions = ( {_fallback_locale(definition): definition.description} if definition.description else {} ) for localization in definition.localizations: if localization.description: descriptions[localization.locale] = localization.description return descriptions def _field_labels(definition, key: str, canonical: str) -> dict[str, str]: labels = {_fallback_locale(definition): canonical} for localization in definition.localizations: label = localization.field_labels.get(key) if label: labels[localization.locale] = label return labels def _field_descriptions( definition, key: str, canonical: str | None ) -> dict[str, str]: descriptions = ( {_fallback_locale(definition): canonical} if canonical else {} ) for localization in definition.localizations: value = localization.field_help_texts.get(key) if value: descriptions[localization.locale] = value return descriptions def _section_labels(definition, key: str, canonical: str) -> dict[str, str]: labels = {_fallback_locale(definition): canonical} for localization in definition.localizations: label = localization.section_titles.get(key) if label: labels[localization.locale] = label return labels def _fallback_locale(definition) -> str: return definition.fallback_locale or "en" def _field_locations(definition) -> dict[str, tuple[object, object]]: return { field_key: (page, section) for page in definition.pages for section in page.sections for field_key in section.field_keys } def _label(labels: Mapping[str, str]) -> str: return labels.get("de") or labels.get("en") or next(iter(labels.values())) def _descriptor_search_text(item: SemanticDocumentationSubjectDescriptor) -> str: return " ".join( ( item.reference.subject_id, *(item.labels.values()), *(item.descriptions.values()), *(breadcrumb.label for breadcrumb in item.breadcrumbs), ) ) def _authorized(principal: object, tenant_id: str) -> bool: if str(getattr(principal, "tenant_id", "") or "") != tenant_id: return False checker = getattr(principal, "has", None) if callable(checker): return bool(checker(READ_SCOPE)) return READ_SCOPE in getattr(principal, "scopes", ()) def _cursor_offset(value: str | None) -> int: if value is None: return 0 if not value.isdigit() or int(value) < 0: raise ValueError("Forms semantic subject cursor is invalid.") return int(value) def _session(value: object) -> Session: if not isinstance(value, Session): raise TypeError("Forms semantic subjects require a SQLAlchemy session.") return value __all__ = [ "FormsSemanticDocumentationSubjectProvider", "SUBJECT_KIND", ]