mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 04:21:57 +00:00
ThearchyHelios
277982ead3
- 蓝图：bp_* 操作和逐卡白名单、蓝图检查、留痕、按蓝图判权限；五个场景模板加自由形（骨架、关节、应用骨架、定时任务），实例化与骨架比对；BlueprintApplier / TemplateApplier（只收模板声明过的关节）与自查；look 新类别口径卡；蓝图对话和填关节两个流式接口，不同组的关节并行填、出错的组串行重试；模板列表 - 试运行与发布单：四种起点（对话、模板、问数据的一轮、已发布版本）；试运行是探索运行，草稿应用按这次运行解析、印章写试运行，流程类派给自己、不触发升级；发布单九步、预检只读、每步留痕、停下后接着做、核对卡片指纹和应用管理权、启动时整理中断的单；白话层覆盖全部问题编号 - 记录表与文件：平台托管只追加的记录表、按高水位固定数据、连接上的临时视图做行级可见、写入走动作（records_append）、表版本在发布时切换、草稿不影响线上、记录列表遮敏感列；文件按哈希存、两步上传、提交时绑定运行、下载按读权限、清理、file_text、按配方导入一期（table_import_period） - 个人应用：存成个人应用（问数据的约束）、刷新（遮罩、只读、每日上限）、只有拥有者看得见、提交审核与共享副本、开关下发到认人接口；页面编辑的候选；隐藏组件不下发；三条新检查
316 lines
17 KiB
Python
316 lines
17 KiB
Python
"""表单 agentlab.form/1：校验和输入节点字段的转换（§3.3）。
|
||
|
||
前端只校验同一份子集（lib/schemaForm.ts）；这里用 jsonschema 复核，以这里为准。错误格式两边一样：{path, code, message}。
|
||
path 是字段名（表单只有一层；数组的某一项出错也挂在这个字段上）。
|
||
|
||
子集之外的写法一律不收，最要紧的是 pattern：JS 和 Python 的正则语义不同，会出现前端放行、后端拒收（P5）。format 也不交给
|
||
jsonschema 的 FormatChecker：date-time、uri 要装额外的包才查，装没装决定查不查，两边就对不上。这里自己按同一套规则查。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from datetime import date, datetime
|
||
from typing import Any, TypedDict
|
||
from urllib.parse import urlsplit
|
||
|
||
from app.apps.spec import FILE_FIELD_TYPE, FILE_REF_KEYS, FORM_FORMAT, FORM_FORMATS, FORM_TYPES, FORM_WIDGETS
|
||
|
||
|
||
class FormError(TypedDict):
|
||
path: str
|
||
code: str
|
||
message: str
|
||
|
||
|
||
_TOP_KEYS = frozenset({"format", "type", "title", "description", "required", "properties", "x-order"})
|
||
_FIELD_KEYS = frozenset({
|
||
"type", "title", "description", "default", "readOnly", "enum", "x-enum-labels", "minimum", "maximum",
|
||
"minLength", "maxLength", "maxItems", "format", "items", "x-widget",
|
||
# 文件控件的限制：accept（类型或扩展名的列表）、max_mb（docs/design/normal-user-apps.md §3.8）
|
||
"x-file",
|
||
})
|
||
#: 文件控件只认这几项写法
|
||
_FILE_FIELD_KEYS = frozenset({"type", "title", "description", "readOnly", "x-widget", "x-file"})
|
||
_NUMBER_KEYS = ("minimum", "maximum")
|
||
_COUNT_KEYS = ("minLength", "maxLength", "maxItems")
|
||
_LIMIT_LABEL = {"minimum": "最小值", "maximum": "最大值", "minLength": "最少字数", "maxLength": "最多字数",
|
||
"maxItems": "最多选几项"}
|
||
_TYPE_LABEL = {"string": "文字", "number": "数", "integer": "整数", "boolean": "是或否", "array": "多选"}
|
||
_FORMAT_LABEL = {"date": "日期（如 2026-10-14）", "date-time": "日期时间（如 2026-10-14T09:30:00）",
|
||
"email": "邮箱地址", "uri": "网址（如 https://example.com）"}
|
||
_EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
|
||
_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
|
||
|
||
|
||
def _number(value: Any) -> bool:
|
||
return isinstance(value, (int, float)) and not isinstance(value, bool)
|
||
|
||
|
||
def validate_form_schema(form: Any) -> list[str]:
|
||
"""表单本身合不合目录（只用允许的 type / format / x-widget，没有 pattern……）。返回问题说明，空表示合规。"""
|
||
if not isinstance(form, dict):
|
||
return ["表单需要是一个对象"]
|
||
problems: list[str] = []
|
||
if form.get("format") not in (None, FORM_FORMAT):
|
||
problems.append(f"表单格式只能是 {FORM_FORMAT}")
|
||
if form.get("type") != "object":
|
||
problems.append("表单最外层的类型只能是对象")
|
||
if extra := sorted(k for k in form if k not in _TOP_KEYS):
|
||
problems.append(f"表单不支持这些写法：{'、'.join(extra)}")
|
||
props = form.get("properties")
|
||
if not isinstance(props, dict):
|
||
return [*problems, "表单需要列出字段"]
|
||
required = form.get("required", [])
|
||
if not isinstance(required, list) or not all(isinstance(r, str) for r in required):
|
||
problems.append("必填项需要是字段名的列表")
|
||
else:
|
||
problems += [f"必填项「{r}」不是表单里的字段" for r in required if r not in props]
|
||
order = form.get("x-order", [])
|
||
if not isinstance(order, list) or not all(isinstance(r, str) for r in order):
|
||
problems.append("字段顺序需要是字段名的列表")
|
||
else:
|
||
problems += [f"字段顺序里的「{r}」不是表单里的字段" for r in order if r not in props]
|
||
for name, field in props.items():
|
||
problems += [f"字段「{name}」{p}" for p in _field_problems(field)]
|
||
return problems
|
||
|
||
|
||
def _field_problems(field: Any) -> list[str]:
|
||
if not isinstance(field, dict):
|
||
return ["需要是一个对象"]
|
||
out: list[str] = []
|
||
if "pattern" in field:
|
||
out.append("不支持正则校验：前端和服务端的正则规则不同，会出现一边放行、一边拒收。请改用枚举、长度或格式")
|
||
if extra := sorted(k for k in field if k not in _FIELD_KEYS and k != "pattern"):
|
||
out.append(f"不支持这些写法：{'、'.join(extra)}")
|
||
kind = field.get("type")
|
||
if field.get("x-widget") == "file" or kind == FILE_FIELD_TYPE:
|
||
return out + _file_field_problems(field)
|
||
if "x-file" in field:
|
||
out.append("不是文件控件，不能写文件的限制")
|
||
if kind not in FORM_TYPES:
|
||
out.append(f"的类型只能是 {' / '.join(FORM_TYPES)}")
|
||
if "format" in field and field["format"] not in FORM_FORMATS:
|
||
out.append(f"的格式只能是 {' / '.join(FORM_FORMATS)}")
|
||
if "format" in field and kind != "string":
|
||
out.append("只有文字字段能设格式")
|
||
if "x-widget" in field and field["x-widget"] not in FORM_WIDGETS:
|
||
out.append(f"的控件只能是 {' / '.join(FORM_WIDGETS)}")
|
||
if "enum" in field:
|
||
enum = field["enum"]
|
||
if not isinstance(enum, list) or not enum or not all(isinstance(v, (str, int, float)) for v in enum):
|
||
out.append("的可选值需要是非空的列表")
|
||
labels = field.get("x-enum-labels")
|
||
if labels is not None and (not isinstance(labels, list) or len(labels) != len(enum or [])):
|
||
out.append("的选项文字需要和可选值一一对应")
|
||
out += [f"的{_LIMIT_LABEL[k]}需要是数" for k in _NUMBER_KEYS if k in field and not _number(field[k])]
|
||
out += [f"的{_LIMIT_LABEL[k]}需要是不小于 0 的整数" for k in _COUNT_KEYS
|
||
if k in field and not (isinstance(field[k], int) and not isinstance(field[k], bool) and field[k] >= 0)]
|
||
if kind == "array":
|
||
items = field.get("items")
|
||
if not (isinstance(items, dict) and items.get("type") == "string" and isinstance(items.get("enum"), list)
|
||
and items["enum"] and set(items) <= {"type", "enum"}):
|
||
out.append("是多选：只能从一组文字选项里选，需要写明可选项")
|
||
elif "items" in field:
|
||
out.append("不是多选，不能写可选项列表")
|
||
return out
|
||
|
||
|
||
def _file_field_problems(field: dict[str, Any]) -> list[str]:
|
||
"""文件控件：类型只能是对象、控件只能是 file（只有这个控件能用对象），限制写在 x-file 里。"""
|
||
out: list[str] = []
|
||
if field.get("x-widget") != "file":
|
||
out.append(f"的类型只能是 {' / '.join(FORM_TYPES)}（对象只有文件控件能用）")
|
||
if field.get("type") != FILE_FIELD_TYPE:
|
||
out.append("是文件控件，类型只能是对象")
|
||
if extra := sorted(k for k in field if k not in _FILE_FIELD_KEYS and k != "pattern"):
|
||
out.append(f"是文件控件，不支持这些写法：{'、'.join(extra)}")
|
||
setting = field.get("x-file")
|
||
if setting is not None:
|
||
if not isinstance(setting, dict) or not set(setting) <= {"accept", "max_mb"}:
|
||
out.append("的文件限制只能写可选的类型（accept）和大小上限（max_mb）")
|
||
else:
|
||
accept = setting.get("accept")
|
||
if accept is not None and not (isinstance(accept, list) and all(isinstance(a, str) and a for a in accept)):
|
||
out.append("的可选类型需要是类型或扩展名的列表，例如 [\".pdf\", \"image/png\"]")
|
||
max_mb = setting.get("max_mb")
|
||
if max_mb is not None and not (_number(max_mb) and max_mb > 0):
|
||
out.append("的大小上限需要是大于 0 的数（MB）")
|
||
return out
|
||
|
||
|
||
def _file_fields(form: dict[str, Any]) -> set[str]:
|
||
return {name for name, field in (form.get("properties") or {}).items()
|
||
if isinstance(field, dict) and (field.get("x-widget") == "file" or field.get("type") == FILE_FIELD_TYPE)}
|
||
|
||
|
||
def _clean_schema(form: dict[str, Any]) -> dict[str, Any]:
|
||
"""交给 jsonschema 的那一份：去掉展示用的键和 format（format 自己查，见文件开头）。文件控件自己查（_file_errors）。"""
|
||
props: dict[str, Any] = {}
|
||
files = _file_fields(form)
|
||
for name, field in (form.get("properties") or {}).items():
|
||
if not isinstance(field, dict) or name in files:
|
||
continue
|
||
props[name] = {k: v for k, v in field.items()
|
||
if k in _FIELD_KEYS and k not in ("format", "x-widget", "x-enum-labels", "title",
|
||
"description", "readOnly", "default")}
|
||
return {"type": "object", "properties": props, "required": [r for r in form.get("required") or [] if r in props]}
|
||
|
||
|
||
_CODES = {"required": "required", "type": "type", "enum": "enum", "minimum": "minimum", "maximum": "maximum",
|
||
"minLength": "min_length", "maxLength": "max_length", "maxItems": "max_items"}
|
||
|
||
|
||
def validate_form(form: dict[str, Any], values: Any) -> list[FormError]:
|
||
"""按表单校验一份填写的值。表单本身不合目录的不在这里报（发布检查报），按能认的部分校验。
|
||
|
||
没填（值为 None 或空串）的非必填字段不校验：前端把清空的输入框交成空串，不该报「格式不对」。"""
|
||
from jsonschema import Draft202012Validator
|
||
|
||
if not isinstance(values, dict):
|
||
return [{"path": "", "code": "type", "message": "提交的内容需要是一组字段和值"}]
|
||
props = form.get("properties") or {}
|
||
required = [r for r in form.get("required") or [] if r in props]
|
||
present = {k: v for k, v in values.items() if v is not None and v != ""}
|
||
# 必填自己按顺序查（jsonschema 的 required 错误不带是哪个字段的路径）
|
||
errors: list[FormError] = [_error(name, "required", props.get(name) or {}) for name in required
|
||
if name not in present]
|
||
validator = Draft202012Validator(_clean_schema(form))
|
||
seen: set[tuple[str, str]] = {(e["path"], e["code"]) for e in errors}
|
||
for err in sorted(validator.iter_errors(present), key=lambda e: list(e.absolute_path)):
|
||
code = _CODES.get(str(err.validator))
|
||
if code is None or code == "required":
|
||
continue
|
||
path = str(err.absolute_path[0]) if err.absolute_path else ""
|
||
if (path, code) in seen:
|
||
continue
|
||
seen.add((path, code))
|
||
errors.append(_error(path, code, props.get(path) or {}))
|
||
for name, value in present.items():
|
||
field = props.get(name)
|
||
if isinstance(field, dict) and field.get("format") and isinstance(value, str) and (name, "type") not in seen \
|
||
and not _format_ok(field["format"], value):
|
||
errors.append(_error(name, "format", field))
|
||
errors += _file_errors(form, present)
|
||
return errors
|
||
|
||
|
||
def _file_errors(form: dict[str, Any], present: dict[str, Any]) -> list[FormError]:
|
||
"""文件控件的值：上传后拿回的引用 {file_id, name, size, sha256, type}，按 x-file 的类型、大小核对。引用是不是这个人
|
||
传的、有没有用过，提交时绑定那一步（files/store.py）按登记核对。"""
|
||
from app.files.store import FILE_ERROR_CODE, accept_problem
|
||
|
||
out: list[FormError] = []
|
||
props = form.get("properties") or {}
|
||
for name in sorted(_file_fields(form)):
|
||
if name not in present:
|
||
continue
|
||
field = props.get(name) or {}
|
||
label = field.get("title") or name
|
||
value = present[name]
|
||
if not (isinstance(value, dict) and set(value) == set(FILE_REF_KEYS) and isinstance(value.get("file_id"), str)
|
||
and value["file_id"] and isinstance(value.get("size"), int) and isinstance(value.get("name"), str)
|
||
and isinstance(value.get("type"), str) and isinstance(value.get("sha256"), str)):
|
||
out.append({"path": name, "code": FILE_ERROR_CODE, "message": f"「{label}」需要先上传文件"})
|
||
continue
|
||
setting = field.get("x-file") if isinstance(field.get("x-file"), dict) else None
|
||
if problem := accept_problem(setting, name=value["name"], mime=value["type"], size=value["size"]):
|
||
out.append({"path": name, "code": FILE_ERROR_CODE, "message": f"「{label}」{problem}"})
|
||
return out
|
||
|
||
|
||
def _format_ok(fmt: str, value: str) -> bool:
|
||
if fmt == "date":
|
||
try:
|
||
return bool(_DATE.match(value)) and isinstance(date.fromisoformat(value), date)
|
||
except ValueError:
|
||
return False
|
||
if fmt == "date-time":
|
||
try:
|
||
datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||
return "T" in value or " " in value
|
||
except ValueError:
|
||
return False
|
||
if fmt == "email":
|
||
return bool(_EMAIL.match(value))
|
||
if fmt == "uri":
|
||
parts = urlsplit(value)
|
||
return bool(parts.scheme in ("http", "https") and parts.netloc)
|
||
return True
|
||
|
||
|
||
def _error(path: str, code: str, field: dict[str, Any]) -> FormError:
|
||
name = field.get("title") or path or "提交的内容"
|
||
if code == "required":
|
||
message = f"请填写「{name}」"
|
||
elif code == "type":
|
||
message = f"「{name}」需要填{_TYPE_LABEL.get(str(field.get('type')), '正确类型的值')}"
|
||
elif code == "enum":
|
||
message = f"「{name}」只能从给出的选项里选"
|
||
elif code == "minimum":
|
||
message = f"「{name}」不能小于 {field.get('minimum')}"
|
||
elif code == "maximum":
|
||
message = f"「{name}」不能大于 {field.get('maximum')}"
|
||
elif code == "min_length":
|
||
message = f"「{name}」至少要 {field.get('minLength')} 个字"
|
||
elif code == "max_length":
|
||
message = f"「{name}」最多 {field.get('maxLength')} 个字"
|
||
elif code == "max_items":
|
||
message = f"「{name}」最多选 {field.get('maxItems')} 项"
|
||
else:
|
||
message = f"「{name}」需要是{_FORMAT_LABEL.get(str(field.get('format')), '正确的格式')}"
|
||
return {"path": path, "code": code, "message": message}
|
||
|
||
|
||
def input_schema(fields: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""输入节点的 fields（name / required / default / description / example，以及新增的 type / enum / format，数字类型的
|
||
minimum / maximum）→ agentlab.form/1。没写 type 的按 string。上下限只带数字类型的、写成数的那几个：表单子集只有闭区间的
|
||
minimum / maximum（没有 exclusiveMinimum），「大于 0」的金额写成 minimum 0.01。type 为 file 的是文件字段（可带 accept、
|
||
max_mb，进 x-file）。
|
||
|
||
有默认值的必填字段不进 required：没填时入口节点补默认值（engine/nodes/io.py 的 run_input），表单不该逼人填。
|
||
不认识的 type、format 退回文字：老图里随手写的值不该让整张表单作废。"""
|
||
props: dict[str, Any] = {}
|
||
required: list[str] = []
|
||
for f in fields or []:
|
||
if not isinstance(f, dict) or not isinstance(f.get("name"), str) or not f["name"].strip():
|
||
continue
|
||
name = f["name"].strip()
|
||
if f.get("type") == "file" or f.get("widget") == "file":
|
||
# 文件字段（docs/design/normal-user-apps.md N9）：值是上传后拿回的引用；限制（可选类型、大小上限）照写
|
||
upload: dict[str, Any] = {"type": FILE_FIELD_TYPE, "x-widget": "file", "title": str(f.get("label") or name)}
|
||
if isinstance(f.get("description"), str) and f["description"].strip():
|
||
upload["description"] = f["description"].strip()
|
||
limits = {k: f[k] for k in ("accept", "max_mb") if f.get(k) not in (None, "", [])}
|
||
if limits:
|
||
upload["x-file"] = limits
|
||
props[name] = upload
|
||
if f.get("required"):
|
||
required.append(name)
|
||
continue
|
||
kind = f.get("type") if f.get("type") in FORM_TYPES else "string"
|
||
prop: dict[str, Any] = {"type": kind, "title": str(f.get("label") or name)}
|
||
if isinstance(f.get("description"), str) and f["description"].strip():
|
||
prop["description"] = f["description"].strip()
|
||
enum = f.get("enum")
|
||
if isinstance(enum, list) and enum and all(isinstance(v, (str, int, float)) for v in enum):
|
||
if kind == "array":
|
||
prop["items"] = {"type": "string", "enum": [str(v) for v in enum]}
|
||
else:
|
||
prop["enum"] = list(enum)
|
||
elif kind == "array":
|
||
# 多选没写可选项的，不能当多选填：退回文字，照旧按 JSON 交（同以前的发起表单）
|
||
prop["type"] = kind = "string"
|
||
if kind == "string" and f.get("format") in FORM_FORMATS:
|
||
prop["format"] = f["format"]
|
||
if kind in ("number", "integer"):
|
||
prop.update({k: f[k] for k in _NUMBER_KEYS if _number(f.get(k))})
|
||
default = f.get("default")
|
||
if default not in (None, ""):
|
||
prop["default"] = default
|
||
props[name] = prop
|
||
if f.get("required") and default in (None, ""):
|
||
required.append(name)
|
||
return {"format": FORM_FORMAT, "type": "object", "properties": props, "required": required,
|
||
"x-order": list(props)}
|