147 lines
6.3 KiB
Python
147 lines
6.3 KiB
Python
"""Validated, local presentation blocks. Component props come from the frontend catalog."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Annotated, Any, Literal
|
|
|
|
from jsonschema import Draft7Validator
|
|
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_serializer, model_validator
|
|
|
|
CHART_SAMPLE_TIME_PATTERN = (
|
|
r"^(?:[1-9][0-9]{3}|0[1-9][0-9]{2}|00[1-9][0-9]|000[1-9])-[0-9]{2}-[0-9]{2}T(?:[01][0-9]|2[0-3]):[0-5][0-9]:[0-5][0-9]"
|
|
r"(?:\.[0-9]+)?(?:Z|[+-](?:[01][0-9]|2[0-3]):[0-5][0-9])$"
|
|
)
|
|
|
|
|
|
def validate_chart_sample_time(value: str) -> None:
|
|
"""Match the bounded frontend RFC3339 spelling and an actual calendar date."""
|
|
if len(value) > 64 or re.fullmatch(CHART_SAMPLE_TIME_PATTERN, value) is None:
|
|
raise ValueError("Chart sample times require RFC3339 seconds and a timezone")
|
|
if datetime.fromisoformat(value).tzinfo is None:
|
|
raise ValueError("Chart sample times require a timezone")
|
|
|
|
|
|
_catalog = json.loads(
|
|
(Path(__file__).resolve().parents[3] / "contracts/answer-ui-catalog.json").read_text()
|
|
)
|
|
_validators = {name: Draft7Validator(schema) for name, schema in _catalog["props"].items()}
|
|
Key = Annotated[str, Field(pattern=r"^[a-zA-Z0-9_-]{1,80}$")]
|
|
|
|
|
|
class UiModel(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
|
|
class AnswerUiElement(UiModel):
|
|
type: str
|
|
props: dict[str, Any]
|
|
children: list[Key] = Field(max_length=48)
|
|
|
|
|
|
class AnswerUiSpec(UiModel):
|
|
root: Key
|
|
elements: dict[Key, AnswerUiElement]
|
|
|
|
|
|
class AnswerUiBlock(UiModel):
|
|
version: Literal["answer-ui@1"]
|
|
block_id: Key
|
|
phase: Literal["process", "final"] | None = None
|
|
fallback_text: Annotated[str, Field(min_length=1, max_length=4000)]
|
|
spec: AnswerUiSpec
|
|
|
|
@field_validator("phase", mode="before")
|
|
@classmethod
|
|
def validate_explicit_phase(cls, value):
|
|
if value is None:
|
|
raise ValueError("UI phase must be process or final when supplied")
|
|
return value
|
|
|
|
@model_serializer(mode="wrap")
|
|
def serialize_optional_phase(self, handler):
|
|
data = handler(self)
|
|
if self.phase is None:
|
|
data.pop("phase", None)
|
|
return data
|
|
|
|
@model_validator(mode="after")
|
|
def validate_components(self):
|
|
spec = self.spec
|
|
if not 1 <= len(spec.elements) <= 48:
|
|
raise ValueError("UI requires 1 to 48 elements")
|
|
# Match the existing JavaScript serialized character limit, including UTF-16.
|
|
serialized = json.dumps(self.model_dump(), ensure_ascii=False, separators=(",", ":"))
|
|
if len(serialized.encode("utf-16-le")) // 2 > 100000:
|
|
raise ValueError("UI block is too large")
|
|
for element in spec.elements.values():
|
|
validator = _validators.get(element.type)
|
|
if validator is None or not validator.is_valid(element.props):
|
|
raise ValueError("Unknown UI component or invalid properties")
|
|
if element.children and element.type not in {"Stack", "Section"}:
|
|
raise ValueError("Leaf components cannot contain children")
|
|
props = element.props
|
|
if element.type == "Chart":
|
|
if any(len(series["data"]) != len(props["x_data"]) for series in props["series"]):
|
|
raise ValueError("Chart series must match the axis")
|
|
if any(value is not None and not math.isfinite(value) for series in props["series"] for value in series["data"]):
|
|
raise ValueError("Chart values must be finite")
|
|
for series in props["series"]:
|
|
times = series.get("sample_times")
|
|
if times is not None:
|
|
if len(times) != len(series["data"]):
|
|
raise ValueError("Chart sample times must match the axis")
|
|
for value, at in zip(series["data"], times, strict=True):
|
|
if at is not None:
|
|
if value is None:
|
|
raise ValueError("Chart sample times require an observed value")
|
|
validate_chart_sample_time(at)
|
|
if element.type == "Table" and any(len(row) != len(props["columns"]) for row in props["rows"]):
|
|
raise ValueError("Table row width must match its columns")
|
|
if element.type == "FollowUpForm":
|
|
names = [field["name"] for field in props["fields"]]
|
|
if len(names) != len(set(names)):
|
|
raise ValueError("Form field names must be unique")
|
|
visited: set[str] = set()
|
|
|
|
def visit(key: str, depth: int):
|
|
if depth > 12 or key in visited or key not in spec.elements:
|
|
raise ValueError("UI must be a bounded tree without duplicate or missing references")
|
|
visited.add(key)
|
|
for child in spec.elements[key].children:
|
|
visit(child, depth + 1)
|
|
|
|
visit(spec.root, 0)
|
|
if len(visited) != len(spec.elements):
|
|
raise ValueError("UI contains unreachable elements")
|
|
return self
|
|
|
|
|
|
class ReportUiBlock(UiModel):
|
|
block: AnswerUiBlock
|
|
evidence_refs: list[Annotated[str, Field(min_length=1, max_length=128)]] = Field(min_length=1, max_length=100)
|
|
|
|
@model_validator(mode="after")
|
|
def validate_report_phase(self):
|
|
if self.block.phase == "process":
|
|
raise ValueError("Report UI blocks must contain final results")
|
|
return self
|
|
|
|
|
|
def validate_report_ui_blocks(blocks: list[ReportUiBlock], evidence_refs: list[str]) -> None:
|
|
ids = [item.block.block_id for item in blocks]
|
|
if len(ids) != len(set(ids)):
|
|
raise ValueError("Report UI block ids must be unique")
|
|
serialized = json.dumps([item.model_dump() for item in blocks], ensure_ascii=False, separators=(",", ":"))
|
|
if len(serialized.encode("utf-8")) > 400000:
|
|
raise ValueError("Report UI exceeds 400 KB")
|
|
for item in blocks:
|
|
if not set(item.evidence_refs) <= set(evidence_refs):
|
|
raise ValueError("UI evidence must be included in report evidence")
|
|
for element in item.block.spec.elements.values():
|
|
if element.type == "Evidence" and element.props.get("evidence_id") not in item.evidence_refs:
|
|
raise ValueError("UI evidence component must reference its block evidence")
|