256 lines
11 KiB
Python
256 lines
11 KiB
Python
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()
|