from __future__ import annotations import hashlib import json import re from collections.abc import Mapping, Sequence from html import escape from html.parser import HTMLParser from sqlalchemy import or_, select from sqlalchemy.orm import Session from govoplan_core.auth import ApiPrincipal from govoplan_core.core.concurrency import claim_revision from govoplan_core.core.templates import ( TemplateCompatibility, TemplateCompatibilityError, TemplateFieldRequirement, TemplateNotFoundError, TemplateOutputProfile, TemplateRef, TemplateRevisionRef, ) from govoplan_core.security.time import utc_now from govoplan_templates.backend.db.models import TemplateDefinition, TemplateRevision from govoplan_templates.backend.schemas import TemplateCreateRequest, TemplateRevisionPayload, TemplateUpdateRequest READ_SCOPE = "templates:template:read" WRITE_SCOPE = "templates:template:write" PUBLISH_SCOPE = "templates:template:publish" RENDER_SCOPE = "templates:template:render" ADMIN_SCOPE = "templates:template:admin" _TOKEN_PATTERN = re.compile(r"{{\s*([A-Za-z_][A-Za-z0-9_.-]*)\s*}}") def visible_templates_statement(principal: ApiPrincipal, *, include_deleted: bool = False): statement = select(TemplateDefinition).where( TemplateDefinition.tenant_id == principal.tenant_id ) if not principal.has(ADMIN_SCOPE): statement = statement.where( or_( TemplateDefinition.scope_type == "tenant", ( (TemplateDefinition.scope_type == "user") & (TemplateDefinition.scope_id == principal.account_id) ), ( (TemplateDefinition.scope_type == "group") & TemplateDefinition.scope_id.in_(tuple(principal.group_ids) or ("",)) ), ) ) if not include_deleted: statement = statement.where(TemplateDefinition.deleted_at.is_(None)) return statement def list_templates( session: Session, principal: ApiPrincipal, *, query: str = "", usage: str | None = None, template_type: str | None = None, locale: str | None = None, include_deleted: bool = False, limit: int = 200, ) -> list[TemplateDefinition]: statement = visible_templates_statement(principal, include_deleted=include_deleted) normalized_query = query.strip() if normalized_query: pattern = f"%{normalized_query}%" statement = statement.where( or_( TemplateDefinition.name.ilike(pattern), TemplateDefinition.slug.ilike(pattern), TemplateDefinition.description.ilike(pattern), ) ) if template_type: statement = statement.where(TemplateDefinition.template_type == template_type) rows = list( session.scalars( statement.order_by(TemplateDefinition.name, TemplateDefinition.id).limit( max(1, min(limit, 500)) ) ) ) if not usage and not locale: return rows filtered: list[TemplateDefinition] = [] for item in rows: revision = get_template_revision(session, item, published_preferred=True) if usage and usage.strip().lower() not in revision.usages: continue if locale and revision.locale.lower() != locale.lower(): continue filtered.append(item) return filtered def get_template( session: Session, principal: ApiPrincipal, template_id: str, *, include_deleted: bool = False, ) -> TemplateDefinition: item = session.scalar( visible_templates_statement(principal, include_deleted=include_deleted).where( TemplateDefinition.id == template_id ) ) if item is None: raise TemplateNotFoundError("Template not found.") return item def get_template_revision( session: Session, definition: TemplateDefinition, *, revision: int | None = None, published_preferred: bool = False, ) -> TemplateRevision: if revision is not None: revision_number = revision elif published_preferred and definition.published_revision_id: published = session.get(TemplateRevision, definition.published_revision_id) if published is not None and published.template_id == definition.id: return published revision_number = definition.current_revision else: revision_number = definition.current_revision item = session.scalar( select(TemplateRevision).where( TemplateRevision.template_id == definition.id, TemplateRevision.tenant_id == definition.tenant_id, TemplateRevision.revision == revision_number, ) ) if item is None: raise TemplateNotFoundError("Template revision not found.") return item def list_template_revisions( session: Session, definition: TemplateDefinition, ) -> list[TemplateRevision]: return list( session.scalars( select(TemplateRevision) .where( TemplateRevision.tenant_id == definition.tenant_id, TemplateRevision.template_id == definition.id, ) .order_by(TemplateRevision.revision.desc()) .limit(500) ) ) def create_template( session: Session, principal: ApiPrincipal, payload: TemplateCreateRequest, ) -> tuple[TemplateDefinition, TemplateRevision]: _ensure_requested_scope(principal, payload.scope_type, payload.scope_id) slug = _slug(payload.slug or payload.name) _ensure_unique_slug( session, principal, slug=slug, scope_type=payload.scope_type, scope_id=payload.scope_id, ) definition = TemplateDefinition( tenant_id=principal.tenant_id, scope_type=payload.scope_type, scope_id=payload.scope_id, name=payload.name.strip(), slug=slug, description=_text(payload.description), template_type=payload.template_type, status="draft", current_revision_id="pending", current_revision=1, resource_revision=1, created_by_account_id=principal.account_id, updated_by_account_id=principal.account_id, metadata_={}, ) session.add(definition) session.flush() revision = _create_revision( session, principal, definition=definition, revision_number=1, payload=payload, ) definition.current_revision_id = revision.id session.flush() return definition, revision def update_template( session: Session, principal: ApiPrincipal, definition: TemplateDefinition, payload: TemplateUpdateRequest, ) -> tuple[TemplateDefinition, TemplateRevision]: _ensure_mutable_scope(principal, definition) _ensure_requested_scope(principal, payload.scope_type, payload.scope_id) slug = _slug(payload.slug or payload.name) _ensure_unique_slug( session, principal, slug=slug, scope_type=payload.scope_type, scope_id=payload.scope_id, exclude_id=definition.id, ) next_resource_revision = claim_revision( session, model=TemplateDefinition, filters=( TemplateDefinition.id == definition.id, TemplateDefinition.tenant_id == definition.tenant_id, TemplateDefinition.deleted_at.is_(None), ), revision_attribute="resource_revision", expected_revision=payload.base_revision, resource_type="template_definition", resource_id=definition.id, refresh_path=f"/api/v1/templates/{definition.id}", ) definition.resource_revision = next_resource_revision definition.scope_type = payload.scope_type definition.scope_id = payload.scope_id definition.name = payload.name.strip() definition.slug = slug definition.description = _text(payload.description) definition.template_type = payload.template_type definition.current_revision += 1 definition.current_revision_id = "pending" definition.status = "draft" definition.updated_by_account_id = principal.account_id revision = _create_revision( session, principal, definition=definition, revision_number=definition.current_revision, payload=payload, ) definition.current_revision_id = revision.id session.flush() return definition, revision def publish_template( session: Session, principal: ApiPrincipal, definition: TemplateDefinition, *, revision: int | None, base_revision: int, ) -> tuple[TemplateDefinition, TemplateRevision]: _ensure_mutable_scope(principal, definition) next_resource_revision = claim_revision( session, model=TemplateDefinition, filters=( TemplateDefinition.id == definition.id, TemplateDefinition.tenant_id == definition.tenant_id, TemplateDefinition.deleted_at.is_(None), ), revision_attribute="resource_revision", expected_revision=base_revision, resource_type="template_definition", resource_id=definition.id, refresh_path=f"/api/v1/templates/{definition.id}", ) item_revision = get_template_revision(session, definition, revision=revision) now = utc_now() item_revision.published_at = now item_revision.published_by_account_id = principal.account_id definition.published_revision_id = item_revision.id definition.status = "active" definition.resource_revision = next_resource_revision definition.updated_by_account_id = principal.account_id session.add(item_revision) session.add(definition) session.flush() return definition, item_revision def delete_template( session: Session, principal: ApiPrincipal, definition: TemplateDefinition, *, base_revision: int, ) -> TemplateDefinition: _ensure_mutable_scope(principal, definition) next_resource_revision = claim_revision( session, model=TemplateDefinition, filters=( TemplateDefinition.id == definition.id, TemplateDefinition.tenant_id == definition.tenant_id, TemplateDefinition.deleted_at.is_(None), ), revision_attribute="resource_revision", expected_revision=base_revision, resource_type="template_definition", resource_id=definition.id, refresh_path="/api/v1/templates", ) definition.resource_revision = next_resource_revision definition.status = "deleted" definition.deleted_at = utc_now() definition.updated_by_account_id = principal.account_id session.flush() return definition def compatibility( revision: TemplateRevision, *, usage: str | None, output_format: str | None, available_fields: Mapping[str, str] | Sequence[str], ) -> TemplateCompatibility: field_types = ( {str(key): str(value) for key, value in available_fields.items()} if isinstance(available_fields, Mapping) else {str(item): "unknown" for item in available_fields} ) normalized_usage = usage.strip().lower() if usage else None diagnostics: list[dict[str, object]] = [] missing: list[str] = [] incompatible: list[str] = [] if normalized_usage and normalized_usage not in revision.usages: diagnostics.append( { "code": "template.usage_incompatible", "severity": "error", "message": f"This template is not published for {normalized_usage}.", } ) formats = {str(item.get("output_format")) for item in revision.output_profiles} if output_format and output_format not in formats: diagnostics.append( { "code": "template.output_format_incompatible", "severity": "error", "message": f"This template does not provide {output_format} output.", } ) for requirement in revision.required_fields: path = str(requirement.get("path") or "") if not path or not bool(requirement.get("required", True)): continue if path not in field_types: missing.append(path) continue actual = field_types[path] expected = str(requirement.get("value_type") or "string") if actual not in {"unknown", expected} and not ( expected == "number" and actual in {"integer", "number"} ): incompatible.append(path) if missing: diagnostics.append( { "code": "template.required_fields_missing", "severity": "error", "message": f"Missing required fields: {', '.join(sorted(missing))}.", } ) if incompatible: diagnostics.append( { "code": "template.field_types_incompatible", "severity": "error", "message": f"Fields have incompatible types: {', '.join(sorted(incompatible))}.", } ) return TemplateCompatibility( compatible=not diagnostics, template_id=revision.template_id, revision_id=revision.id, usage=normalized_usage, output_format=output_format, missing_fields=tuple(sorted(missing)), incompatible_fields=tuple(sorted(incompatible)), diagnostics=tuple(diagnostics), ) def template_ref( definition: TemplateDefinition, revision: TemplateRevision, *, read_only: bool = False, ) -> TemplateRef: return TemplateRef( id=definition.id, tenant_id=definition.tenant_id, name=definition.name, slug=definition.slug, template_type=definition.template_type, # type: ignore[arg-type] status=definition.status, current_revision=definition.current_revision, current_revision_id=definition.current_revision_id, published_revision_id=definition.published_revision_id, description=definition.description, scope_type=definition.scope_type, scope_id=definition.scope_id, read_only=read_only, updated_at=definition.updated_at, revision=revision_ref(revision), metadata=dict(definition.metadata_ or {}), ) def revision_ref(revision: TemplateRevision) -> TemplateRevisionRef: return TemplateRevisionRef( id=revision.id, template_id=revision.template_id, revision=revision.revision, definition_hash=revision.definition_hash, template_type=revision.template_type, # type: ignore[arg-type] usages=tuple(revision.usages or []), locale=revision.locale, required_fields=tuple( TemplateFieldRequirement(**item) for item in revision.required_fields ), output_profiles=tuple( TemplateOutputProfile(**item) for item in revision.output_profiles ), content_text=revision.content_text, content_html=revision.content_html, layout=dict(revision.layout or {}), metadata=dict(revision.metadata_ or {}), published_at=revision.published_at, provenance={ "module": "templates", "created_by_account_id": revision.created_by_account_id, }, ) def referenced_fields(revision: TemplateRevision) -> tuple[str, ...]: content = f"{revision.content_text or ''}\n{revision.content_html or ''}" return tuple(sorted(set(_TOKEN_PATTERN.findall(content)))) def _create_revision( session: Session, principal: ApiPrincipal, *, definition: TemplateDefinition, revision_number: int, payload: TemplateRevisionPayload, ) -> TemplateRevision: output_profiles = [item.model_dump(mode="json") for item in payload.output_profiles] if not output_profiles: output_profiles = _default_output_profiles(payload.template_type) sanitized_html = sanitize_template_html(payload.content_html) definition_payload = { "template_type": payload.template_type, "usages": payload.usages, "locale": payload.locale, "required_fields": [item.model_dump(mode="json") for item in payload.required_fields], "output_profiles": output_profiles, "content_text": payload.content_text, "content_html": sanitized_html, "layout": payload.layout, "metadata": payload.metadata, } revision = TemplateRevision( tenant_id=principal.tenant_id, template_id=definition.id, revision=revision_number, definition_hash=_canonical_hash(definition_payload), template_type=payload.template_type, usages=list(payload.usages), locale=payload.locale, required_fields=definition_payload["required_fields"], output_profiles=output_profiles, content_text=payload.content_text, content_html=sanitized_html, layout=dict(payload.layout), metadata_=dict(payload.metadata), created_by_account_id=principal.account_id, ) session.add(revision) session.flush() return revision def _default_output_profiles(template_type: str) -> list[dict[str, object]]: if template_type == "content_fragment": return [] media = "A4" if template_type == "envelope": media = "DL" return [ { "id": "print-html", "label": "Printable HTML", "output_format": "html", "media_type": "text/html", "channel": "print", "capabilities": ["browser_print", template_type], "page": {"size": media}, }, { "id": "plain-text", "label": "Plain text", "output_format": "text", "media_type": "text/plain", "channel": "download", "capabilities": ["deterministic_text"], "page": {}, }, ] def _ensure_unique_slug( session: Session, principal: ApiPrincipal, *, slug: str, scope_type: str, scope_id: str | None, exclude_id: str | None = None, ) -> None: statement = select(TemplateDefinition.id).where( TemplateDefinition.tenant_id == principal.tenant_id, TemplateDefinition.scope_type == scope_type, TemplateDefinition.slug == slug, TemplateDefinition.deleted_at.is_(None), ) if scope_id is None: statement = statement.where(TemplateDefinition.scope_id.is_(None)) else: statement = statement.where(TemplateDefinition.scope_id == scope_id) if exclude_id: statement = statement.where(TemplateDefinition.id != exclude_id) if session.scalar(statement) is not None: raise TemplateCompatibilityError("A template with this slug already exists in the selected scope.") def _ensure_requested_scope( principal: ApiPrincipal, scope_type: str, scope_id: str | None, ) -> None: if principal.has(ADMIN_SCOPE): return if scope_type == "tenant": return if scope_type == "user" and scope_id == principal.account_id: return if scope_type == "group" and scope_id in principal.group_ids: return raise TemplateCompatibilityError("The requested template scope is not writable by this principal.") def _ensure_mutable_scope(principal: ApiPrincipal, definition: TemplateDefinition) -> None: _ensure_requested_scope(principal, definition.scope_type, definition.scope_id) def _slug(value: str) -> str: normalized = re.sub(r"[^a-z0-9_.-]+", "-", value.strip().lower()).strip("-.") if not normalized: raise TemplateCompatibilityError("Template slug cannot be empty.") return normalized[:160] def _text(value: str | None) -> str | None: normalized = str(value or "").strip() return normalized or None def _canonical_hash(value: object) -> str: payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False) return hashlib.sha256(payload.encode("utf-8")).hexdigest() class _TemplateHtmlSanitizer(HTMLParser): allowed_tags = { "a", "blockquote", "br", "code", "div", "em", "h1", "h2", "h3", "h4", "h5", "h6", "hr", "li", "ol", "p", "pre", "span", "strong", "table", "tbody", "td", "th", "thead", "tr", "u", "ul", } void_tags = {"br", "hr"} blocked_tags = {"script", "style", "iframe", "object", "embed", "svg", "math"} allowed_attrs = {"a": {"href", "title"}, "td": {"colspan", "rowspan"}, "th": {"colspan", "rowspan"}} def __init__(self) -> None: super().__init__(convert_charrefs=True) self.parts: list[str] = [] self.blocked_depth = 0 def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: tag = tag.lower() if tag in self.blocked_tags: self.blocked_depth += 1 return if self.blocked_depth or tag not in self.allowed_tags: return clean_attrs: list[str] = [] for name, raw_value in attrs: name = name.lower() value = str(raw_value or "") if name not in self.allowed_attrs.get(tag, set()): continue if name == "href" and not _safe_href(value): continue clean_attrs.append(f' {name}="{escape(value, quote=True)}"') self.parts.append(f"<{tag}{''.join(clean_attrs)}>") def handle_startendtag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: self.handle_starttag(tag, attrs) def handle_endtag(self, tag: str) -> None: tag = tag.lower() if tag in self.blocked_tags: self.blocked_depth = max(0, self.blocked_depth - 1) return if not self.blocked_depth and tag in self.allowed_tags and tag not in self.void_tags: self.parts.append(f"") def handle_data(self, data: str) -> None: if not self.blocked_depth: self.parts.append(escape(data, quote=False)) def sanitize_template_html(value: str | None) -> str | None: if not value or not value.strip(): return None parser = _TemplateHtmlSanitizer() parser.feed(value) parser.close() return "".join(parser.parts).strip() or None def _safe_href(value: str) -> bool: normalized = value.strip().lower() return normalized.startswith(("https://", "http://", "mailto:", "#", "/")) __all__ = [ "ADMIN_SCOPE", "PUBLISH_SCOPE", "READ_SCOPE", "RENDER_SCOPE", "WRITE_SCOPE", "compatibility", "create_template", "delete_template", "get_template", "get_template_revision", "list_template_revisions", "list_templates", "publish_template", "referenced_fields", "revision_ref", "sanitize_template_html", "template_ref", "update_template", ]