220 lines
8.2 KiB
Python
220 lines
8.2 KiB
Python
"""大屏公共结构；组件属性、事件与数据源配置保持开放"""
|
||
|
||
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
|