from __future__ import annotations from collections import Counter from datetime import datetime, timezone import hashlib import json from pathlib import Path import re from typing import Any from uuid import UUID from app.persistence import atomic_write_bytes SCHEMA_VERSION = 1 FONT_SIZE_MIN = 8 FONT_SIZE_MAX = 64 TEXT_LENGTH_MAX = 2000 INSERTION_LIMIT = 500 DRAFT_FIELDS = frozenset({ "schema_version", "base_catalog_digest", "revision", "updated_at", "overrides", "insertions", }) OVERRIDE_FIELDS = frozenset({"text", "font_size_px", "hidden"}) INSERTION_FIELDS = frozenset({ "id", "workspace_id", "anchor_id", "placement", "text", "font_size_px", }) PLACEHOLDER_PATTERN = re.compile(r"\{([a-zA-Z][a-zA-Z0-9_]*)\}") class UiCopyDraftError(RuntimeError): pass class UiCopyDraftConflictError(UiCopyDraftError): pass class UiCopyDraftValidationError(UiCopyDraftError): pass def _canonical_bytes(document: dict[str, Any]) -> bytes: return (json.dumps(document, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8") def _now() -> str: return datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z") class UiCopyDraftStore: def __init__(self, data_dir: Path, catalog_path: Path) -> None: self.directory = Path(data_dir) / "development" / "ui-copy-editor" self.path = self.directory / "draft.json" self.catalog_path = Path(catalog_path) self.catalog_bytes = self.catalog_path.read_bytes() self.catalog_digest = hashlib.sha256(self.catalog_bytes).hexdigest() try: self.catalog = json.loads(self.catalog_bytes) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise UiCopyDraftValidationError("UI copy catalog is not valid UTF-8 JSON") from exc self._validate_catalog() self.document = self._load() @property def items(self) -> dict[str, dict[str, Any]]: return self.catalog["items"] @property def anchors(self) -> dict[str, dict[str, Any]]: return self.catalog["anchors"] def _validate_catalog(self) -> None: if not isinstance(self.catalog, dict) or self.catalog.get("schema_version") != 2: raise UiCopyDraftValidationError("UI copy catalog schema is unsupported") if set(self.catalog) != {"schema_version", "items", "anchors", "insertions"}: raise UiCopyDraftValidationError("UI copy catalog contains unknown fields") if not isinstance(self.catalog["items"], dict) or not isinstance(self.catalog["anchors"], dict): raise UiCopyDraftValidationError("UI copy catalog collections are invalid") for copy_id, item in self.catalog["items"].items(): if not isinstance(copy_id, str) or not isinstance(item, dict): raise UiCopyDraftValidationError("UI copy catalog item is invalid") render = item.get("render") if not isinstance(render, dict) or render.get("type") not in { "static_text", "attribute", "dynamic_template", }: raise UiCopyDraftValidationError(f"UI copy binding is invalid: {copy_id}") if type(render.get("default_hidden", False)) is not bool: raise UiCopyDraftValidationError(f"UI copy default hidden state is invalid: {copy_id}") def empty_document(self) -> dict[str, Any]: return { "schema_version": SCHEMA_VERSION, "base_catalog_digest": self.catalog_digest, "revision": 0, "updated_at": None, "overrides": {}, "insertions": [], } def _load(self) -> dict[str, Any]: if not self.path.exists(): return self.empty_document() try: raw = json.loads(self.path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: raise UiCopyDraftValidationError("UI copy draft is not valid UTF-8 JSON") from exc return self.validate_document(raw, allow_catalog_mismatch=True) def validate_document( self, raw: Any, *, allow_catalog_mismatch: bool = False, ) -> dict[str, Any]: if not isinstance(raw, dict): raise UiCopyDraftValidationError("UI copy draft must be a JSON object") unknown = set(raw) - DRAFT_FIELDS missing = DRAFT_FIELDS - set(raw) if unknown or missing: raise UiCopyDraftValidationError("UI copy draft fields are invalid") version = raw.get("schema_version") if type(version) is not int or version != SCHEMA_VERSION: raise UiCopyDraftValidationError("UI copy draft schema_version is unsupported") digest = raw.get("base_catalog_digest") if not isinstance(digest, str) or not re.fullmatch(r"[0-9a-f]{64}", digest): raise UiCopyDraftValidationError("UI copy draft catalog digest is invalid") if not allow_catalog_mismatch and digest != self.catalog_digest: raise UiCopyDraftConflictError("UI copy catalog changed; export the existing draft before editing") revision = raw.get("revision") if type(revision) is not int or revision < 0: raise UiCopyDraftValidationError("UI copy draft revision is invalid") updated_at = raw.get("updated_at") if updated_at is not None and not isinstance(updated_at, str): raise UiCopyDraftValidationError("UI copy draft updated_at is invalid") overrides = self._validate_overrides(raw.get("overrides")) insertions = self._validate_insertions(raw.get("insertions")) return { "schema_version": SCHEMA_VERSION, "base_catalog_digest": digest, "revision": revision, "updated_at": updated_at, "overrides": overrides, "insertions": insertions, } def _validate_overrides(self, raw: Any) -> dict[str, dict[str, Any]]: if not isinstance(raw, dict): raise UiCopyDraftValidationError("UI copy overrides must be an object") checked: dict[str, dict[str, Any]] = {} for copy_id, value in raw.items(): if copy_id not in self.items: raise UiCopyDraftValidationError(f"unknown UI copy id: {copy_id}") if not isinstance(value, dict) or not value or set(value) - OVERRIDE_FIELDS: raise UiCopyDraftValidationError(f"invalid UI copy override: {copy_id}") item = self.items[copy_id] normalized: dict[str, Any] = {} if "text" in value: text = value["text"] if not isinstance(text, str) or not text or len(text) > TEXT_LENGTH_MAX: raise UiCopyDraftValidationError(f"invalid UI copy text: {copy_id}") expected = Counter(item.get("placeholders", [])) if Counter(PLACEHOLDER_PATTERN.findall(text)) != expected: raise UiCopyDraftValidationError(f"UI copy placeholders changed: {copy_id}") normalized["text"] = text if "font_size_px" in value: size = value["font_size_px"] if item.get("kind") == "html_attribute": raise UiCopyDraftValidationError(f"attribute copy cannot have a font size: {copy_id}") if type(size) is not int or not FONT_SIZE_MIN <= size <= FONT_SIZE_MAX: raise UiCopyDraftValidationError(f"invalid UI copy font size: {copy_id}") normalized["font_size_px"] = size if "hidden" in value: hidden = value["hidden"] if item.get("kind") == "html_attribute": raise UiCopyDraftValidationError(f"attribute copy cannot be hidden: {copy_id}") if type(hidden) is not bool: raise UiCopyDraftValidationError(f"invalid UI copy hidden state: {copy_id}") normalized["hidden"] = hidden checked[copy_id] = normalized return checked def _validate_insertions(self, raw: Any) -> list[dict[str, Any]]: if not isinstance(raw, list) or len(raw) > INSERTION_LIMIT: raise UiCopyDraftValidationError("UI copy insertions are invalid") checked: list[dict[str, Any]] = [] seen: set[str] = set() for item in raw: if not isinstance(item, dict) or set(item) != INSERTION_FIELDS: raise UiCopyDraftValidationError("UI copy insertion fields are invalid") identifier = item["id"] try: UUID(str(identifier)) except (ValueError, TypeError, AttributeError) as exc: raise UiCopyDraftValidationError("UI copy insertion id is invalid") from exc if identifier in seen: raise UiCopyDraftValidationError("UI copy insertion id is duplicated") seen.add(identifier) anchor_id = item["anchor_id"] anchor = self.anchors.get(anchor_id) if anchor is None: raise UiCopyDraftValidationError("UI copy insertion anchor is unknown") workspace_id = item["workspace_id"] if workspace_id != anchor.get("scope"): raise UiCopyDraftValidationError("UI copy insertion workspace does not match its anchor") placement = item["placement"] if placement not in {"before", "after"}: raise UiCopyDraftValidationError("UI copy insertion placement is invalid") text = item["text"] if not isinstance(text, str) or not text or len(text) > TEXT_LENGTH_MAX: raise UiCopyDraftValidationError("UI copy insertion text is invalid") size = item["font_size_px"] if type(size) is not int or not FONT_SIZE_MIN <= size <= FONT_SIZE_MAX: raise UiCopyDraftValidationError("UI copy insertion font size is invalid") checked.append({ "id": identifier, "workspace_id": workspace_id, "anchor_id": anchor_id, "placement": placement, "text": text, "font_size_px": size, }) return checked def status_document(self) -> dict[str, Any]: return { **self.document, "catalog_mismatch": self.document["base_catalog_digest"] != self.catalog_digest, "catalog_digest": self.catalog_digest, } def replace(self, raw: dict[str, Any], *, expected_revision: int) -> dict[str, Any]: if self.document["base_catalog_digest"] != self.catalog_digest: raise UiCopyDraftConflictError("UI copy catalog changed; export the existing draft before editing") if expected_revision != self.document["revision"]: raise UiCopyDraftConflictError("UI copy draft was modified by another page") candidate = {**raw, "revision": expected_revision, "updated_at": self.document["updated_at"]} checked = self.validate_document(candidate) checked["revision"] = expected_revision + 1 checked["updated_at"] = _now() self.directory.mkdir(parents=True, exist_ok=True) atomic_write_bytes(self.path, _canonical_bytes(checked)) self.document = checked return self.status_document()