"""大屏公共结构;组件属性、事件与数据源配置保持开放""" import json import re from typing import Annotated, Any, Literal, Self from pydantic import ( BaseModel, ConfigDict, Field, field_validator, model_serializer, model_validator, ) from .base import ExtensibleModel, JsonObject, NonBlankString Number = int | float PositiveNumber = Annotated[Number, Field(gt=0)] # 变量的值:JSON 标量;相等按 JSON 严格比较,"1" 与 1 不等 VarValue = str | int | float | bool | None # 模板占位符 {{变量ID}}:括号内可有空格,ID 由字母、数字、_、- 组成;和前端 fillTemplate 同一条规则 # 用显式字符集而不是 \w:Python 的 \w 匹配中文,JS 的不匹配 TEMPLATE = re.compile(r"\{\{\s*([A-Za-z0-9_-]+)\s*\}\}") def template_refs(value: Any) -> set[str]: """值里(递归到字符串)引用到的变量 ID""" if isinstance(value, str): return set(TEMPLATE.findall(value)) if isinstance(value, dict): return {ref for item in value.values() for ref in template_refs(item)} if isinstance(value, list): return {ref for item in value for ref in template_refs(item)} return set() def _as_text(value: Any) -> str: # 按 JS 的 String(value):null 为空串,布尔小写,整数值的小数不带 .0 if value is None: return "" if isinstance(value, bool): return "true" if value else "false" if isinstance(value, float) and value.is_integer(): return str(int(value)) return str(value) def fill_template(value: Any, variables: dict[str, Any]) -> Any: """整个字符串就是一个占位符时换成变量原值(保留类型,null 也原样);嵌在文字中时转成文字拼接, null 和未定义为空串;对象和数组递归,其他原样返回""" if isinstance(value, str): if whole := TEMPLATE.fullmatch(value): return variables.get(whole.group(1)) return TEMPLATE.sub(lambda m: _as_text(variables.get(m.group(1))), value) if isinstance(value, dict): return {key: fill_template(item, variables) for key, item in value.items()} if isinstance(value, list): return [fill_template(item, variables) for item in value] return value def json_key(value: Any) -> tuple[str, Any]: """相等判断的键:键相同即相等。数字之间按值(1 与 1.0 相同),其余类型不同就不等("1" 与 1、true 与 1 都不等) 可以放进集合或字典里分组;对象和数组按内容逐层用同样的规则比较(前端的 === 不会认为它们相等,变量值只用标量) """ if isinstance(value, bool): return "boolean", value if isinstance(value, (int, float)): # int 与 float 值相等时哈希也相等,1 和 1.0 落在同一个键 return "number", value if isinstance(value, str): return "string", value if value is None: return "null", None if isinstance(value, dict): return "object", tuple(sorted((key, json_key(item)) for key, item in value.items())) if isinstance(value, list): return "array", tuple(json_key(item) for item in value) return "json", json.dumps(value, sort_keys=True, ensure_ascii=False) def json_equal(left: Any, right: Any) -> bool: """按 JSON 严格比较,规则见 json_key""" return json_key(left) == json_key(right) def is_scalar(value: Any) -> bool: """变量的值只能是 JSON 标量""" return value is None or isinstance(value, (str, int, float, bool)) class ProtocolModel(BaseModel): """前后端约定的交互结构:字段固定,写错的键直接报错,不当作扩展字段保留""" model_config = ConfigDict(extra="forbid", strict=True, allow_inf_nan=False) class Variable(ProtocolModel): """大屏变量:点击写变量,组件读变量(显隐、筛选数据、填文字、取数参数)""" id: NonBlankString = Field( description="变量 ID,组件按它引用;文本内容和数据源参数里写 {{ID}} 取它的值,所以只用字母、数字、_、-" ) name: str | None = Field( default=None, description="给用户看的变量名,例如 选中模型" ) default: VarValue = Field( default=None, description="打开大屏时的值:字符串、数字、布尔或 null;null 时按它筛选的组件不过滤。" "被 filters 引用时要取数据里存在的值", ) class VisibleWhen(ProtocolModel): """变量等于 equals 时组件才显示""" var: NonBlankString = Field(description="变量 ID") equals: VarValue = Field(description="变量等于这个值时显示,按 JSON 严格比较") class RowFilter(ProtocolModel): """只保留 row[field] 等于变量值的数据行;变量为 null 时不过滤""" field: NonBlankString = Field(description="组件数据行里的列名") var: NonBlankString = Field(description="变量 ID") class EventAction(ProtocolModel): """事件触发时写变量:value 写常量,fromField 写被点击数据行的某列(只用于 itemClick)""" setVar: NonBlankString = Field(description="要写的变量 ID") value: VarValue = Field(default=None, description="写入的常量;与 fromField 二选一") fromField: NonBlankString | None = Field( default=None, description="写入被点击数据行这一列的值,只用于 itemClick 事件;与 value 二选一", ) @model_validator(mode="after") def one_source(self) -> Self: # value 可以是 null,按是否写出判断,不按值判断 if ("value" in self.model_fields_set) == (self.fromField is not None): raise ValueError("value 与 fromField 必须二选一") return self @model_serializer(mode="wrap") def written_source_only(self, handler): # 二选一按"写了哪个键"区分,没写的那个不能序列化出来 data = handler(self) if "value" not in self.model_fields_set: data.pop("value", None) if self.fromField is None: data.pop("fromField", None) return data class Canvas(ExtensibleModel): width: PositiveNumber height: PositiveNumber theme: Literal["light", "dark"] | None = Field( default=None, description="画布主题:组件里没单独设置的文字、坐标轴、分隔线、卡片底等颜色按主题取;旧大屏没有时前端按背景色推断", ) backgroundColor: str | None = Field( default=None, description="画布背景色;为 null 时用主题的背景色" ) variables: list[Variable] = Field( default_factory=list, description="大屏变量,整组替换;组件用 visibleWhen、filters、事件和 {{ID}} 模板读写它们", ) class Layout(ExtensibleModel): """组件在画布上的绝对坐标和尺寸,单位 px""" x: Number = Field(description="左上角横坐标") y: Number = Field(description="左上角纵坐标") width: PositiveNumber height: PositiveNumber class ScreenNode(ExtensibleModel): id: NonBlankString type: NonBlankString layout: Layout name: str | None = None props: JsonObject = Field(default_factory=dict) style: JsonObject | None = None # 事件保持开放(前端还有 title、name、code 等字段);带 action 时按 EventAction 校验结构 events: list[JsonObject] | None = None visibleWhen: VisibleWhen | None = None filters: list[RowFilter] | None = None @field_validator("events") @classmethod def event_actions(cls, events: list[JsonObject] | None): for event in events or []: if "action" in event: EventAction.model_validate(event["action"]) return events class Screen(ExtensibleModel): canvas: Canvas nodes: list[ScreenNode] # 保持前端字段名;数据源的具体类型与绑定校验后续再引入 dataSource: list[JsonObject] = Field(default_factory=list) @model_validator(mode="after") def unique_node_ids(self) -> Self: seen: set[str] = set() for node in self.nodes: if node.id in seen: raise ValueError(f"重复的节点 ID: {node.id}") seen.add(node.id) return self