# -*- coding: utf-8 -*- """pbl_blueprint 表定义/CRUD定义生成器（M1a）。 单一事实源：本文件的 TABLES 声明 -> 生成 pbl_blueprint/models/{tbl}.json 四段式表定义（summary/fields/indexes/codes） pbl_blueprint/json/{tbl}.json CRUD 定义（tblname/params/editable/browserfields） 用法（机构工作空间根目录）： python3 modules/pbl_blueprint/scripts/gen_defs.py # 生成 + 清理陈旧定义 python3 modules/pbl_blueprint/scripts/gen_defs.py --clean # 仅清理非 11 表的陈旧文件 python3 modules/pbl_blueprint/scripts/gen_defs.py --dump-ddl # 打印建表 DDL 约定（QC 逐条核验点）： * primary 恒为 ["id"]；id 为 str(32) notnull * tenant_id 为首业务字段（紧跟 id），str(32) notnull * 金额/分值字段一律 double(18,2) * 所有索引 tenant_id 打头 * CRUD 定义根键仅 tblname/params/editable/browserfields，editable 为 3 个 api/*.dspy """ import io import json import os import sys HERE = os.path.dirname(os.path.abspath(__file__)) MODULE_ROOT = os.path.dirname(HERE) PKG = os.path.join(MODULE_ROOT, "pbl_blueprint") MODEL_DIR = os.path.join(PKG, "models") JSON_DIR = os.path.join(PKG, "json") TABLE_ORDER = [ "pbl_blueprint", "pbl_blueprint_node", "pbl_blueprint_edge", "pbl_blueprint_version", "pbl_blueprint_version_delta", "pbl_blueprint_template", "pbl_blueprint_publish", "pbl_blueprint_offline", "pbl_blueprint_fork", "pbl_blueprint_lock", "pbl_blueprint_audit", ] AUDIT_FIELDS = ("create_user", "create_time", "update_user", "update_time") IMMUTABLE_FIELDS = AUDIT_FIELDS + ("deleted",) # 服务端写入的大字段：CRUD 定义里 editable=false（前端只读） SERVER_BLOB_FIELDS = ("snapshot_json", "pkg_json", "before_json", "after_json", "game_def_json", "checksum", "file_size", "use_count", "node_count", "edge_count", "current_version", "rev", "publish_time", "expire_time", "cost_ms", "err_code") QUALITY_CODE = { "summary": "质量状态（5 级，M2 校验引擎回写）", "items": { "q0": "未校验", "q1": "不合格", "q2": "基本合格", "q3": "合格", "q4": "优秀", "q5": "标杆", }, } def F(name, typ, size=None, nn=False, default=None, summary=""): """字段声明。""" spec = {"type": typ} if size is not None: spec["size"] = size if nn: spec["notnull"] = True if default is not None: spec["default"] = default spec["summary"] = summary return (name, spec) def ID(): return F("id", "str", 32, True, None, "主键ID（str32，UUID去横线）") def TENANT(extra=""): return F("tenant_id", "str", 32, True, None, "租户ID（首业务字段，所有读写强制打头，缺失即 fail-closed 拒绝）" + extra) def DELETED(): return F("deleted", "int", None, True, 0, "逻辑删除标记 0正常 1删除") def IDX(name, fields, unique, summary): return (name, {"fields": list(fields), "unique": bool(unique), "summary": summary}) # ------------------------------------------------------------------ 11 表定义 TABLES = {} TABLES["pbl_blueprint"] = { "summary": "PBL 蓝图聚合根表：一个租户下的一份项目式学习蓝图（学科/年级/学段/状态/质量状态/当前版本/预算）。所有读写 tenant_id 强制打头。", "fields": [ ID(), TENANT(), F("code", "str", 64, True, None, "蓝图编码（租户内唯一，PBL-BP-序号）"), F("name", "str", 128, True, None, "蓝图名称"), F("subject", "str", 64, False, "", "学科"), F("grade", "str", 32, False, "", "年级"), F("phase", "str", 32, False, "", "学段"), F("status", "str", 16, True, "draft", "蓝图状态"), F("quality_status", "str", 16, True, "q0", "质量状态（5 级）"), F("source", "str", 16, True, "manual", "来源方式"), F("source_id", "str", 32, False, "", "来源对象ID（模板ID或父蓝图ID）"), F("owner_id", "str", 32, False, "", "负责人（教师）ID"), F("class_id", "str", 32, False, "", "关联班级ID"), F("current_version", "int", None, True, 0, "当前版本号（save_version 回写）"), F("duration_hours", "int", None, True, 0, "预计课时数"), F("budget_amount", "double", [18, 2], True, 0, "预算金额（double(18,2)）"), F("summary", "str", 512, False, "", "蓝图简介"), F("ext_json", "text", None, False, "", "扩展属性JSON"), F("publish_time", "datetime", None, False, None, "最近发布时间"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), F("update_user", "str", 32, False, "", "更新人"), F("update_time", "datetime", None, False, None, "更新时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_blueprint_code", ["tenant_id", "code"], True, "租户内蓝图编码唯一"), IDX("idx_pbl_blueprint_tenant", ["tenant_id", "deleted", "status"], False, "租户蓝图列表主索引（tenant_id 打头）"), IDX("idx_pbl_blueprint_owner", ["tenant_id", "owner_id"], False, "按负责人查蓝图"), IDX("idx_pbl_blueprint_source", ["tenant_id", "source", "source_id"], False, "按来源（模板/fork）追溯"), IDX("idx_pbl_blueprint_class", ["tenant_id", "class_id"], False, "按班级查蓝图"), ], "codes": { "status": { "summary": "蓝图状态", "items": { "draft": "草稿", "validating": "校验中", "ready": "就绪", "published": "已发布", "archived": "已归档", "disabled": "已停用", }, }, "quality_status": QUALITY_CODE, "source": { "summary": "来源方式", "items": { "manual": "手工创建", "template": "模板实例化", "fork": "蓝图fork派生", "agent": "Designer Agent 生成", "import": "离线包导入", }, }, }, "browserfields": ["code", "name", "subject", "status", "quality_status", "owner_id", "update_time"], "queryfields": ["code", "name", "subject", "grade", "phase", "status", "quality_status", "source_id", "owner_id", "class_id"], } TABLES["pbl_blueprint_node"] = { "summary": "蓝图子对象节点表（7 类子对象泛化契约）：role/task/rule/artifact/scene/assessment/resource 统一存一张表，差异载荷放 spec_json，按 node_type 分派校验器（M2）。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, None, "所属蓝图ID"), F("node_type", "str", 16, True, None, "子对象类型（7 类泛化）"), F("parent_id", "str", 32, True, "", "父节点ID（构成蓝图树，空串=根节点）"), F("code", "str", 64, True, "", "节点编码（蓝图内同类型唯一）"), F("name", "str", 128, True, None, "节点名称"), F("sort_no", "int", None, True, 0, "同级排序号"), F("spec_json", "text", None, False, "", "泛化载荷JSON（结构随 node_type 不同）"), F("ref_module", "str", 32, True, "", "引用模块名（scense/world/assessment 等）"), F("ref_id", "str", 32, True, "", "引用对象ID"), F("status", "str", 16, True, "active", "节点状态"), F("required", "int", None, True, 0, "是否必需节点 0否 1是"), F("weight", "double", [18, 2], True, 0, "权重/分值（double(18,2)）"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), F("update_user", "str", 32, False, "", "更新人"), F("update_time", "datetime", None, False, None, "更新时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_node_code", ["tenant_id", "blueprint_id", "node_type", "code"], True, "蓝图内同类型节点编码唯一"), IDX("idx_pbl_node_bp", ["tenant_id", "blueprint_id", "deleted", "node_type"], False, "按蓝图+类型列节点（tenant_id 打头）"), IDX("idx_pbl_node_parent", ["tenant_id", "blueprint_id", "parent_id", "sort_no"], False, "构树查询"), IDX("idx_pbl_node_ref", ["tenant_id", "ref_module", "ref_id"], False, "按引用对象反查节点"), ], "codes": { "node_type": { "summary": "子对象类型（7 类泛化契约）", "items": { "role": "角色", "task": "任务", "rule": "规则", "artifact": "产出物", "scene": "场景引用", "assessment": "评估项", "resource": "资源", }, }, "status": { "summary": "节点状态", "items": { "active": "生效", "draft": "草稿", "invalid": "校验不通过", "disabled": "停用", }, }, }, "browserfields": ["code", "name", "node_type", "parent_id", "sort_no", "status"], "queryfields": ["blueprint_id", "node_type", "parent_id", "code", "name", "ref_module", "ref_id", "status"], } TABLES["pbl_blueprint_edge"] = { "summary": "蓝图节点关系边表：描述子对象之间的依赖/解锁/产出/消耗/评估/包含关系，供 Compiler（M3）编译为 Game Definition 图结构。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, None, "所属蓝图ID"), F("from_node_id", "str", 32, True, None, "起点节点ID"), F("to_node_id", "str", 32, True, None, "终点节点ID"), F("edge_type", "str", 16, True, "depends", "边类型"), F("weight", "double", [18, 2], True, 0, "边权重（double(18,2)）"), F("condition_json", "text", None, False, "", "触发条件JSON"), F("sort_no", "int", None, True, 0, "排序号"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), F("update_time", "datetime", None, False, None, "更新时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_edge", ["tenant_id", "blueprint_id", "from_node_id", "to_node_id", "edge_type"], True, "同类型边唯一，防重复连边"), IDX("idx_pbl_edge_bp", ["tenant_id", "blueprint_id", "deleted"], False, "按蓝图取全部边"), IDX("idx_pbl_edge_from", ["tenant_id", "from_node_id"], False, "出边查询"), IDX("idx_pbl_edge_to", ["tenant_id", "to_node_id"], False, "入边查询（删节点前校验）"), ], "codes": { "edge_type": { "summary": "边类型", "items": { "depends": "依赖", "unlocks": "解锁", "produces": "产出", "consumes": "消耗", "assesses": "评估", "contains": "包含", }, }, }, "browserfields": ["from_node_id", "to_node_id", "edge_type", "weight", "sort_no"], "queryfields": ["blueprint_id", "from_node_id", "to_node_id", "edge_type"], } TABLES["pbl_blueprint_version"] = { "summary": "蓝图版本表：每次 save_version 生成一个版本快照（snapshot_json 存整棵树），支持版本回溯、change_delta 对比与回滚。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, None, "所属蓝图ID"), F("version_no", "int", None, True, None, "版本号（蓝图内递增）"), F("version_type", "str", 16, True, "draft", "版本类型"), F("change_summary", "str", 512, True, "", "变更说明"), F("snapshot_json", "text", None, False, "", "蓝图整树快照JSON（主表字段+节点+边）"), F("node_count", "int", None, True, 0, "快照节点数"), F("edge_count", "int", None, True, 0, "快照边数"), F("quality_status", "str", 16, True, "q0", "该版本质量状态"), F("score_amount", "double", [18, 2], True, 0, "该版本评估得分（double(18,2)，M6 回写）"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_version", ["tenant_id", "blueprint_id", "version_no"], True, "蓝图内版本号唯一"), IDX("idx_pbl_version_bp", ["tenant_id", "blueprint_id", "deleted", "version_no"], False, "版本列表（倒序取最新）"), ], "codes": { "version_type": { "summary": "版本类型", "items": { "draft": "草稿版本", "minor": "小版本", "major": "大版本", "publish": "发布版本", "rollback": "回滚版本", }, }, "quality_status": QUALITY_CODE, }, "browserfields": ["version_no", "version_type", "change_summary", "node_count", "edge_count", "quality_status", "create_time"], "queryfields": ["blueprint_id", "version_no", "version_type", "quality_status"], } TABLES["pbl_blueprint_version_delta"] = { "summary": "蓝图版本差异表：记录两个版本之间的 change_delta（新增/修改/删除的节点、边与主表字段），供版本对比与回滚审计。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, None, "所属蓝图ID"), F("from_version", "int", None, True, 0, "基准版本号"), F("to_version", "int", None, True, 0, "目标版本号"), F("delta_type", "str", 16, True, None, "差异对象类型"), F("op_type", "str", 16, True, None, "操作类型"), F("target_id", "str", 32, True, "", "差异对象ID（节点ID/边ID/蓝图ID）"), F("target_name", "str", 128, True, "", "差异对象名称（冗余便于展示）"), F("before_json", "text", None, False, "", "变更前JSON"), F("after_json", "text", None, False, "", "变更后JSON"), F("sort_no", "int", None, True, 0, "排序号"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_delta", ["tenant_id", "blueprint_id", "from_version", "to_version", "delta_type", "op_type", "target_id"], True, "同一版本对同一对象同一操作唯一，保证幂等写入"), IDX("idx_pbl_delta_bp", ["tenant_id", "blueprint_id", "deleted", "to_version"], False, "按蓝图+版本查差异"), ], "codes": { "delta_type": { "summary": "差异对象类型", "items": {"blueprint": "蓝图主表字段", "node": "节点", "edge": "边"}, }, "op_type": { "summary": "操作类型", "items": {"add": "新增", "update": "修改", "delete": "删除"}, }, }, "browserfields": ["from_version", "to_version", "delta_type", "op_type", "target_name"], "queryfields": ["blueprint_id", "from_version", "to_version", "delta_type", "op_type", "target_id"], } TABLES["pbl_blueprint_template"] = { "summary": "蓝图模板表：可复用的蓝图骨架（payload_json 存模板树），支持实例化为新蓝图；离线兜底时作为本地模板源。", "fields": [ ID(), TENANT("；空串=平台公共模板"), F("code", "str", 64, True, None, "模板编码（租户内唯一）"), F("name", "str", 128, True, None, "模板名称"), F("category", "str", 32, True, "general", "模板分类"), F("subject", "str", 64, True, "", "适用学科"), F("grade", "str", 32, True, "", "适用年级"), F("scope", "str", 16, True, "tenant", "可见范围"), F("payload_json", "text", None, False, "", "模板树载荷JSON（blueprint+nodes+edges）"), F("node_count", "int", None, True, 0, "模板节点数"), F("edge_count", "int", None, True, 0, "模板边数"), F("price_amount", "double", [18, 2], True, 0, "模板定价金额（double(18,2)，0=免费）"), F("use_count", "int", None, True, 0, "被实例化次数"), F("status", "str", 16, True, "enabled", "模板状态"), F("summary", "str", 512, True, "", "模板说明"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), F("update_user", "str", 32, False, "", "更新人"), F("update_time", "datetime", None, False, None, "更新时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_template_code", ["tenant_id", "code"], True, "租户内模板编码唯一"), IDX("idx_pbl_template_list", ["tenant_id", "deleted", "status", "category"], False, "模板列表查询（tenant_id 打头）"), IDX("idx_pbl_template_subject", ["tenant_id", "subject", "grade"], False, "按学科年级筛模板"), ], "codes": { "scope": { "summary": "可见范围", "items": {"tenant": "本租户", "platform": "平台公共", "private": "仅创建人"}, }, "status": { "summary": "模板状态", "items": {"enabled": "启用", "disabled": "停用", "draft": "草稿"}, }, "category": { "summary": "模板分类", "items": { "stem": "STEM", "science": "自然科学", "humanity": "人文社科", "art": "艺术", "labor": "劳动实践", "general": "通用", }, }, }, "browserfields": ["code", "name", "category", "subject", "scope", "status", "use_count"], "queryfields": ["code", "name", "category", "subject", "grade", "scope", "status"], } TABLES["pbl_blueprint_publish"] = { "summary": "蓝图发布/投放记录表：蓝图某版本投放到班级/团队/学生的一次发布，记录运行时句柄与编译产物，供 M11 runtime_ext 关联。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, None, "蓝图ID"), F("version_no", "int", None, True, 0, "发布的版本号"), F("class_id", "str", 32, True, "", "投放班级ID"), F("target_type", "str", 16, True, "class", "投放对象类型"), F("target_id", "str", 32, True, "", "投放对象ID"), F("runtime_ref", "str", 64, True, "", "运行时引用（scense_runtime 会话/世界ID）"), F("game_def_json", "text", None, False, "", "编译产物 Game Definition JSON（M3 回写）"), F("start_time", "datetime", None, False, None, "开始时间"), F("end_time", "datetime", None, False, None, "结束时间"), F("budget_amount", "double", [18, 2], True, 0, "本次投放预算金额（double(18,2)）"), F("status", "str", 16, True, "published", "投放状态"), F("remark", "str", 512, True, "", "备注"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), F("update_time", "datetime", None, False, None, "更新时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_publish", ["tenant_id", "blueprint_id", "version_no", "target_type", "target_id"], True, "同版本同对象只投放一次，保证幂等"), IDX("idx_pbl_publish_bp", ["tenant_id", "blueprint_id", "deleted"], False, "按蓝图查投放记录"), IDX("idx_pbl_publish_class", ["tenant_id", "class_id", "status"], False, "按班级查在跑的投放"), ], "codes": { "target_type": { "summary": "投放对象类型", "items": {"class": "班级", "team": "团队", "student": "学生", "preview": "预览"}, }, "status": { "summary": "投放状态", "items": { "published": "已发布", "running": "进行中", "finished": "已结束", "revoked": "已撤回", }, }, }, "browserfields": ["blueprint_id", "version_no", "target_type", "target_id", "status", "start_time", "end_time"], "queryfields": ["blueprint_id", "version_no", "class_id", "target_type", "target_id", "runtime_ref", "status"], } TABLES["pbl_blueprint_offline"] = { "summary": "蓝图离线兜底包表：把蓝图/模板打包为自描述 JSON 包（含 schema_version 与 sha256 校验和），网络或服务不可用时本地兜底实例化。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, "", "来源蓝图ID（与 template_id 二选一）"), F("template_id", "str", 32, True, "", "来源模板ID"), F("pkg_code", "str", 64, True, None, "离线包编码"), F("pkg_name", "str", 128, True, None, "离线包名称"), F("pkg_version", "int", None, True, 1, "离线包版本"), F("pkg_json", "text", None, False, "", "离线包内容JSON（自描述：schema_version+蓝图+节点+边）"), F("file_size", "int", None, True, 0, "包体字节数"), F("checksum", "str", 64, True, "", "内容校验和（sha256）"), F("status", "str", 16, True, "ready", "离线包状态"), F("expire_time", "datetime", None, False, None, "过期时间"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), F("update_time", "datetime", None, False, None, "更新时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_offline_code", ["tenant_id", "pkg_code", "pkg_version"], True, "离线包编码+版本唯一"), IDX("idx_pbl_offline_bp", ["tenant_id", "blueprint_id", "deleted"], False, "按蓝图查离线包"), IDX("idx_pbl_offline_status", ["tenant_id", "status", "expire_time"], False, "过期包清理扫描"), ], "codes": { "status": { "summary": "离线包状态", "items": { "ready": "可用", "building": "构建中", "expired": "已过期", "invalid": "校验失败", }, }, }, "browserfields": ["pkg_code", "pkg_name", "pkg_version", "status", "file_size", "create_time"], "queryfields": ["blueprint_id", "template_id", "pkg_code", "pkg_name", "pkg_version", "status"], } TABLES["pbl_blueprint_fork"] = { "summary": "蓝图 fork 血缘表：记录蓝图之间的派生关系（源蓝图→新蓝图）与模板实例化来源，支持 fork 溯源与派生树查询。", "fields": [ ID(), TENANT(), F("source_blueprint_id", "str", 32, True, "", "源蓝图ID（模板实例化时为空串）"), F("source_version", "int", None, True, 0, "fork 时的源版本号"), F("target_blueprint_id", "str", 32, True, None, "派生出的新蓝图ID"), F("fork_type", "str", 16, True, "copy", "fork 方式"), F("template_id", "str", 32, True, "", "若由模板实例化，记录模板ID"), F("node_count", "int", None, True, 0, "复制的节点数"), F("edge_count", "int", None, True, 0, "复制的边数"), F("cost_amount", "double", [18, 2], True, 0, "本次 fork 产生的费用金额（double(18,2)，付费模板）"), F("remark", "str", 512, True, "", "备注"), F("create_user", "str", 32, False, "", "创建人"), F("create_time", "datetime", None, False, None, "创建时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_fork", ["tenant_id", "source_blueprint_id", "target_blueprint_id"], True, "同一对蓝图血缘唯一，保证幂等"), IDX("idx_pbl_fork_source", ["tenant_id", "source_blueprint_id", "deleted"], False, "查某蓝图的派生列表"), IDX("idx_pbl_fork_target", ["tenant_id", "target_blueprint_id"], False, "反查某蓝图的来源"), IDX("idx_pbl_fork_template", ["tenant_id", "template_id"], False, "查模板被实例化记录"), ], "codes": { "fork_type": { "summary": "fork 方式", "items": { "copy": "完整复制", "structure": "仅结构（不带载荷）", "template": "模板实例化", "branch": "分支派生", }, }, }, "browserfields": ["source_blueprint_id", "source_version", "target_blueprint_id", "fork_type", "node_count", "create_time"], "queryfields": ["source_blueprint_id", "target_blueprint_id", "fork_type", "template_id"], } TABLES["pbl_blueprint_lock"] = { "summary": "蓝图编辑锁表：多人协作时对蓝图/节点加编辑锁（悲观锁+rev 乐观修订号），锁带过期时间，过期可被抢占。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, None, "被锁蓝图ID"), F("node_id", "str", 32, True, "", "被锁节点ID（空串=整蓝图锁）"), F("lock_type", "str", 16, True, "edit", "锁类型"), F("holder_id", "str", 32, True, None, "持锁人ID"), F("holder_name", "str", 64, True, "", "持锁人名称"), F("rev", "int", None, True, 0, "乐观锁修订号（每次抢占/续锁 +1）"), F("expire_time", "datetime", None, False, None, "锁过期时间"), F("status", "str", 16, True, "holding", "锁状态"), F("create_time", "datetime", None, False, None, "加锁时间"), F("update_time", "datetime", None, False, None, "续锁时间"), DELETED(), ], "indexes": [ IDX("uk_pbl_lock", ["tenant_id", "blueprint_id", "node_id", "lock_type"], True, "同对象同类型仅一把锁记录"), IDX("idx_pbl_lock_holder", ["tenant_id", "holder_id", "status"], False, "查某人持有的锁"), IDX("idx_pbl_lock_expire", ["tenant_id", "status", "expire_time"], False, "过期锁清理扫描"), ], "codes": { "lock_type": { "summary": "锁类型", "items": {"edit": "编辑锁", "publish": "发布锁", "compile": "编译锁"}, }, "status": { "summary": "锁状态", "items": {"holding": "持有中", "released": "已释放", "expired": "已过期"}, }, }, "browserfields": ["blueprint_id", "node_id", "lock_type", "holder_name", "status", "expire_time"], "queryfields": ["blueprint_id", "node_id", "lock_type", "holder_id", "status"], } TABLES["pbl_blueprint_audit"] = { "summary": "蓝图操作审计表（append-only）：记录蓝图/节点/边/版本/模板/发布/锁的关键写操作与结果，只增不改不删，供追溯与合规审计。", "fields": [ ID(), TENANT(), F("blueprint_id", "str", 32, True, "", "蓝图ID"), F("target_type", "str", 16, True, None, "操作对象类型"), F("target_id", "str", 32, True, "", "操作对象ID"), F("action", "str", 32, True, None, "操作动作"), F("version_no", "int", None, True, 0, "关联版本号"), F("before_json", "text", None, False, "", "操作前数据JSON"), F("after_json", "text", None, False, "", "操作后数据JSON"), F("result", "str", 16, True, "success", "操作结果"), F("err_code", "str", 32, True, "", "失败错误码"), F("cost_ms", "int", None, True, 0, "耗时毫秒"), F("op_user", "str", 32, True, "", "操作人"), F("op_source", "str", 32, True, "api", "操作来源（web/api/agent/offline）"), F("client_ip", "str", 64, True, "", "客户端IP"), F("create_time", "datetime", None, False, None, "操作时间"), ], "indexes": [ IDX("idx_pbl_audit_bp", ["tenant_id", "blueprint_id", "create_time"], False, "按蓝图时间线查审计"), IDX("idx_pbl_audit_target", ["tenant_id", "target_type", "target_id"], False, "按对象查审计"), IDX("idx_pbl_audit_user", ["tenant_id", "op_user", "create_time"], False, "按操作人查审计"), IDX("idx_pbl_audit_action", ["tenant_id", "action", "result"], False, "按动作与结果统计"), ], "codes": { "target_type": { "summary": "操作对象类型", "items": { "blueprint": "蓝图", "node": "节点", "edge": "边", "version": "版本", "delta": "版本差异", "template": "模板", "publish": "发布", "offline": "离线包", "fork": "血缘", "lock": "编辑锁", }, }, "action": { "summary": "操作动作", "items": { "create": "创建", "update": "更新", "delete": "删除", "fork": "fork派生", "save_version": "存版本", "rollback": "回滚", "publish": "发布", "revoke": "撤回发布", "instantiate": "模板实例化", "validate": "校验", "compile": "编译", "lock": "加锁", "unlock": "解锁", "export_offline": "导出离线包", "import_offline": "导入离线包", }, }, "result": { "summary": "操作结果", "items": {"success": "成功", "fail": "失败", "denied": "被拒绝"}, }, }, "browserfields": ["blueprint_id", "target_type", "action", "result", "op_user", "op_source", "create_time"], "queryfields": ["blueprint_id", "target_type", "target_id", "action", "version_no", "result", "op_user", "op_source", "create_time"], } # ------------------------------------------------------------------ 生成 def build_model(tbl): d = TABLES[tbl] fields = {} for name, spec in d["fields"]: fields[name] = spec return { "summary": d["summary"], "primary": ["id"], "fields": fields, "indexes": dict(d["indexes"]), "codes": d["codes"], } def build_crud(tbl): d = TABLES[tbl] codes = d.get("codes") or {} params = {} for name, spec in d["fields"]: p = {"type": spec["type"]} if "size" in spec: p["size"] = spec["size"] if spec.get("notnull"): p["notnull"] = True if "default" in spec: p["default"] = spec["default"] if name in IMMUTABLE_FIELDS or name in SERVER_BLOB_FIELDS: p["editable"] = False else: p["editable"] = True p["query"] = name in (d.get("queryfields") or ()) or name == "tenant_id" if name in codes: p["code"] = name p["summary"] = spec["summary"] params[name] = p return { "tblname": tbl, "params": params, "editable": [ "api/%s_list.dspy" % tbl, "api/%s_edit.dspy" % tbl, "api/%s_view.dspy" % tbl, ], "browserfields": list(d.get("browserfields") or []), } def col_type(spec): t = (spec.get("type") or "str").lower() size = spec.get("size") if t in ("str", "string", "varchar", "char"): return "varchar(%d)" % int(size or 64) if t == "int": return "int" if t == "bigint": return "bigint" if t == "double": if isinstance(size, (list, tuple)) and len(size) == 2: return "decimal(%d,%d)" % (int(size[0]), int(size[1])) return "double" if t in ("text", "longtext", "json"): return "text" if t in ("datetime", "timestamp"): return "datetime" if t == "date": return "date" return "varchar(64)" def ddl(): out = ["-- pbl_blueprint M1a DDL（%d 表，由 scripts/gen_defs.py 生成）" % len(TABLE_ORDER), ""] for tbl in TABLE_ORDER: m = build_model(tbl) cols = [] for name, spec in m["fields"].items(): nn = " not null" if spec.get("notnull") else "" df = "" if "default" in spec: dv = spec["default"] df = (" default '%s'" % dv) if isinstance(dv, str) else (" default %s" % dv) cm = " comment '%s'" % str(spec.get("summary") or "").replace("'", "") cols.append(" `%s` %s%s%s%s" % (name, col_type(spec), nn, df, cm)) cols.append(" primary key (`%s`)" % "` , `".join(m["primary"])) for iname, ispec in m["indexes"].items(): uq = "unique " if ispec.get("unique") else "" cols.append(" %skey `%s` (%s)" % (uq, iname, ", ".join(["`%s`" % c for c in ispec["fields"]]))) out.append("-- %s" % m["summary"]) out.append("create table if not exists `%s` (" % tbl) out.append(",\n".join(cols)) out.append(") engine=InnoDB default charset=utf8mb4 comment='%s';" % tbl.replace("'", "")) out.append("") return "\n".join(out) def clean_stale(): """删除 models/ 与 json/ 下不属于 11 表清单的陈旧定义文件。""" removed = [] for d in (MODEL_DIR, JSON_DIR): if not os.path.isdir(d): continue for fn in sorted(os.listdir(d)): if not fn.endswith(".json"): continue if fn[:-5] not in TABLE_ORDER: p = os.path.join(d, fn) os.remove(p) removed.append(os.path.relpath(p, MODULE_ROOT)) return removed def write_json(path, obj): d = os.path.dirname(path) if not os.path.isdir(d): os.makedirs(d) with io.open(path, "w", encoding="utf-8") as f: f.write(json.dumps(obj, ensure_ascii=False, indent=2) + "\n") def generate(): removed = clean_stale() written = [] for tbl in TABLE_ORDER: p1 = os.path.join(MODEL_DIR, "%s.json" % tbl) write_json(p1, build_model(tbl)) written.append(os.path.relpath(p1, MODULE_ROOT)) p2 = os.path.join(JSON_DIR, "%s.json" % tbl) write_json(p2, build_crud(tbl)) written.append(os.path.relpath(p2, MODULE_ROOT)) return removed, written def main(argv=None): argv = list(sys.argv[1:] if argv is None else argv) if "--dump-ddl" in argv: print(ddl()) return 0 if "--clean" in argv: rm = clean_stale() print("已清理陈旧定义 %d 个: %s" % (len(rm), rm)) return 0 rm, wr = generate() print("清理陈旧定义 %d 个: %s" % (len(rm), rm)) print("生成定义文件 %d 个（models 11 + json 11）:" % len(wr)) for w in wr: print(" %s" % w) return 0 if __name__ == "__main__": sys.exit(main())