# -*- coding: utf-8 -*- """M1a 表定义对齐设计定稿生成器（权威：projects/pbls/docs/01-design/modules/pbl_blueprint.md §2）。 产出 ---- 1. pbl_blueprint/models/*.json —— 设计定稿 B1~B11 共 11 张表，四段式 （summary/fields/indexes/codes），primary=["id"]，id str32，tenant_id 首业务字段， 金额字段 double(18,2) 2. pbl_blueprint/json/*.json —— 11 个 CRUD 契约，根键 tblname+params， editable 3 个 .dspy URL，browserfields 非空 3. pbl_blueprint/sql/*.sql —— mariadb 方言 DDL（无 FK / 无 ENUM / 无 TIMESTAMP） core.sql = B1~B3 + 模板；subobjects.sql = B4~B11；support.sql = 运维支撑表 4. 在途旧表定义（node/edge/lock/fork/audit/offline/publish/version_delta/template） 归档到 models/support/ 与 json/support/，不计入 11 表主清单，功能不丢失 幂等：可重复执行。 """ from __future__ import annotations import json import os import shutil SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) REPO = os.path.dirname(SCRIPT_DIR) PKG = os.path.join(REPO, "pbl_blueprint") MODELS = os.path.join(PKG, "models") JSOND = os.path.join(PKG, "json") SQLD = os.path.join(PKG, "sql") # 设计定稿 B1~B11 主清单（权威表名，顺序即编号） MAIN_TABLES = [ "pbl_blueprint", "pbl_blueprint_version", "pbl_blueprint_approval", "pbl_learner", "pbl_learning_goal", "pbl_problem", "pbl_driving_question", "pbl_project", "pbl_role", "pbl_mission", "pbl_artifact_def", ] # 在途旧表 -> 归档为支撑表（不计入 11 表主清单） SUPPORT_TABLES = [ "pbl_blueprint_audit", "pbl_blueprint_edge", "pbl_blueprint_fork", "pbl_blueprint_lock", "pbl_blueprint_node", "pbl_blueprint_offline", "pbl_blueprint_publish", "pbl_blueprint_template", "pbl_blueprint_version_delta", ] # ------------------------------------------------------------------ 字段构造助手 def f_str(size, summary, notnull=True, default="", query=False, editable=True, code=None): d = {"type": "str", "size": size, "notnull": notnull, "default": default, "editable": editable, "query": query, "summary": summary} if code: d["code"] = code return d def f_id(): return {"type": "str", "size": 32, "notnull": True, "default": "", "editable": True, "query": False, "summary": "主键ID（str32，UUID去横线）"} def f_tenant(): return {"type": "str", "size": 32, "notnull": True, "default": "", "editable": True, "query": True, "summary": "租户ID（首业务字段，所有读写强制打头，缺失即 fail-closed 拒绝）"} def f_int(summary, default=0, query=False, editable=True, notnull=True): return {"type": "int", "notnull": notnull, "default": default, "editable": editable, "query": query, "summary": summary} def f_money(summary, query=False): """金额字段：double(18,2)（规范硬约束）""" return {"type": "double", "size": [18, 2], "notnull": True, "default": 0, "editable": True, "query": query, "summary": summary} def f_dt(summary, editable=False, query=False): return {"type": "datetime", "notnull": False, "default": None, "editable": editable, "query": query, "summary": summary} def f_text(summary, editable=True, query=False, ttype="text"): return {"type": ttype, "notnull": False, "editable": editable, "query": query, "summary": summary} def audit_tail(): """通用审计尾字段（所有表统一）""" return [ ("create_user", f_str(32, "创建人", editable=False)), ("create_time", f_dt("创建时间")), ("update_user", f_str(32, "更新人", editable=False)), ("update_time", f_dt("更新时间")), ("deleted", f_int("逻辑删除标记 0正常 1删除", default=0, editable=False)), ] def sub_head(extra=None): """子对象公共头：id / tenant_id / blueprint_id / version_no / sort_no""" out = [ ("id", f_id()), ("tenant_id", f_tenant()), ("blueprint_id", f_str(32, "所属蓝图ID（聚合根外键，逻辑关联无物理FK）", query=True)), ("version_no", f_int("所属蓝图版本号（子对象随版本快照）", default=1, query=True)), ("sort_no", f_int("排序号", default=0)), ] if extra: out += extra return out # ------------------------------------------------------------------ 11 表定义 TABLES = {} TABLES["pbl_blueprint"] = { "summary": "PBL 蓝图聚合根（B1）。一条记录=一个可编译的项目式学习蓝图，" "承载标题/学科/年级/班级/负责教师/状态/质量等级/当前版本号，" "并记录生成来源（人工/模板/AI生成）与 fork 溯源。" "所有读写 tenant_id 强制打头，跨租户一律 404。", "fields": [ ("id", f_id()), ("tenant_id", f_tenant()), ("code", f_str(64, "蓝图编码（租户内唯一，创建自动生成）", query=True)), ("title", f_str(200, "蓝图标题", query=True)), ("summary", f_str(1000, "蓝图摘要")), ("subject", f_str(64, "学科", query=True, code="pbl_subject")), ("grade_level", f_str(32, "适用年级", query=True, code="pbl_grade_level")), ("class_id", f_str(32, "关联班级ID（可空=未绑定班级）", query=True, notnull=False)), ("owner_teacher_id", f_str(32, "负责教师ID", query=True)), ("status", f_str(32, "蓝图状态", default="draft", query=True, code="pbl_blueprint_status")), ("quality_level", f_str(32, "质量等级（5级，禁跳级上升）", default="draft", query=True, code="pbl_quality_level")), ("current_version", f_int("当前生效版本号", default=1, query=True)), ("generation_source", f_str(32, "生成来源", default="manual", query=True, code="pbl_generation_source")), ("source_template_id", f_str(32, "来源模板ID（generation_source=template 时必填）", notnull=False, query=True)), ("fork_from_id", f_str(32, "fork 源蓝图ID（溯源，可空）", notnull=False, query=True)), ("cover_url", f_str(512, "封面图URL", notnull=False)), ("tags", f_str(512, "标签（逗号分隔）", query=True)), ("duration_hours", f_int("预计总课时（小时）", default=0, query=True)), ("budget_amount", f_money("蓝图预算金额（元，double(18,2)）")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_code", "type": "unique", "fields": ["tenant_id", "code", "deleted"]}, {"name": "idx_tenant_status", "type": "normal", "fields": ["tenant_id", "status"]}, {"name": "idx_tenant_owner", "type": "normal", "fields": ["tenant_id", "owner_teacher_id"]}, {"name": "idx_tenant_class", "type": "normal", "fields": ["tenant_id", "class_id"]}, ], "codes": ["pbl_blueprint_status", "pbl_quality_level", "pbl_generation_source", "pbl_subject", "pbl_grade_level"], } TABLES["pbl_blueprint_version"] = { "summary": "蓝图版本快照（B2，append-only）。每次 update/commit 生成一条新版本，" "存全量 snapshot_json + sha256 快照哈希 + change_delta 三分类变更" "（added/modified/removed），is_current 标记当前生效版本。" "历史版本永不修改，rollback 采用软删重建后提交新版本。", "fields": [ ("id", f_id()), ("tenant_id", f_tenant()), ("blueprint_id", f_str(32, "所属蓝图ID", query=True)), ("version_no", f_int("版本号（蓝图内递增）", default=1, query=True)), ("version_type", f_str(16, "版本类型", default="draft", query=True, code="pbl_version_type")), ("change_summary", f_str(512, "变更说明")), ("change_reason", f_str(512, "变更原因（审计留痕）")), ("snapshot_json", f_text("全量快照JSON（聚合根+7类子对象+关系）", ttype="longtext")), ("snapshot_hash", f_str(64, "快照sha256哈希（确定性序列化 sort_keys+紧凑分隔符）", query=True)), ("change_delta_json", f_text("变更增量JSON（added/modified/removed 三分类 + stat_*）")), ("stat_added", f_int("新增对象数", default=0)), ("stat_modified", f_int("修改对象数", default=0)), ("stat_removed", f_int("删除对象数", default=0)), ("is_current", f_int("是否当前生效版本 0否 1是", default=0, query=True)), ("commit_user", f_str(32, "提交人")), ("commit_time", f_dt("提交时间", editable=True, query=True)), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_ver", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "deleted"]}, {"name": "idx_tenant_current", "type": "normal", "fields": ["tenant_id", "blueprint_id", "is_current"]}, ], "codes": ["pbl_version_type"], } TABLES["pbl_blueprint_approval"] = { "summary": "蓝图审批记录（B3）。记录提交人/审批人/审批状态/审批意见与时间，" "支撑 draft→submitted→approved/rejected→published 状态流转的审批留痕。" "一次提交一条记录，驳回后可再次提交形成多条。", "fields": [ ("id", f_id()), ("tenant_id", f_tenant()), ("blueprint_id", f_str(32, "所属蓝图ID", query=True)), ("version_no", f_int("被审批的版本号", default=1, query=True)), ("submit_user", f_str(32, "提交人ID", query=True)), ("submit_time", f_dt("提交时间", editable=True, query=True)), ("approver_id", f_str(32, "审批人ID", query=True, notnull=False)), ("approval_status", f_str(32, "审批状态", default="pending", query=True, code="pbl_approval_status")), ("approval_opinion", f_str(1000, "审批意见")), ("approval_time", f_dt("审批时间", editable=True, query=True)), ("approval_round", f_int("审批轮次（驳回重提递增）", default=1, query=True)), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, {"name": "idx_tenant_approver", "type": "normal", "fields": ["tenant_id", "approver_id"]}, {"name": "idx_tenant_status", "type": "normal", "fields": ["tenant_id", "approval_status"]}, ], "codes": ["pbl_approval_status"], } TABLES["pbl_learner"] = { "summary": "学习者画像子对象（B4）。描述本蓝图面向的学习者群体特征：" "先备知识、技能基线、兴趣标签、人数区间与互动方式，" "供 Compiler 生成差异化任务与 Assessment 设定评估基线。", "fields": sub_head() + [ ("name", f_str(200, "画像名称", query=True)), ("description", f_str(2000, "画像描述")), ("prior_knowledge", f_str(2000, "先备知识")), ("skill_baseline", f_str(2000, "技能基线")), ("interest_tags", f_str(512, "兴趣标签（逗号分隔）", query=True)), ("learner_count", f_int("预计学习者人数", default=0, query=True)), ("min_count", f_int("最小人数", default=1)), ("max_count", f_int("最大人数", default=40)), ("interaction_type", f_str(32, "互动方式", default="individual", query=True, code="pbl_interaction_type")), ("group_mode", f_str(32, "分组模式", default="free", query=True, code="pbl_group_mode")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_sort", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "sort_no", "deleted"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, ], "codes": ["pbl_interaction_type", "pbl_group_mode"], } TABLES["pbl_learning_goal"] = { "summary": "学习目标子对象（B5）。可测量的学习目标定义：目标类型、布卢姆认知层级、" "素养维度、是否可测量、测量方式与目标值、解锁条件。" "是 Assessment Rubric 权重分配的锚点。", "fields": sub_head() + [ ("goal_code", f_str(64, "目标编码（蓝图内唯一）", query=True)), ("title", f_str(200, "目标标题", query=True)), ("description", f_str(2000, "目标描述")), ("goal_type", f_str(32, "目标类型", default="knowledge", query=True, code="pbl_goal_type")), ("measurable", f_int("是否可测量 0否 1是", default=1, query=True)), ("measure_method", f_str(512, "测量方式")), ("target_value", f_str(128, "目标值（如 ≥80分 / 完成3次）")), ("bloom_level", f_str(32, "布卢姆认知层级", default="understand", query=True, code="pbl_bloom_level")), ("competency_dim", f_str(64, "素养维度", query=True, code="pbl_competency_dim")), ("weight", f_int("评估权重（百分比 0-100）", default=0, query=True)), ("unlock_condition_json", f_text("解锁条件JSON（前置目标/任务达成后解锁）")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_code", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "goal_code", "deleted"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, ], "codes": ["pbl_goal_type", "pbl_bloom_level", "pbl_competency_dim"], } TABLES["pbl_problem"] = { "summary": "真实世界问题子对象（B6）。PBL 的核心锚点——定义待解决的真实问题：" "真实性等级、现实情境、利益相关方、数据来源与复杂度，" "驱动问题（B7）由本表派生。", "fields": sub_head() + [ ("title", f_str(200, "问题标题", query=True)), ("description", f_str(4000, "问题描述")), ("authenticity_level", f_str(32, "真实性等级", default="simulated", query=True, code="pbl_authenticity_level")), ("real_world_context", f_str(2000, "现实情境描述")), ("stakeholder", f_str(512, "利益相关方")), ("data_source", f_str(1000, "数据来源（真实数据集/调研/文献）")), ("complexity_level", f_str(32, "复杂度等级", default="medium", query=True, code="pbl_complexity_level")), ("domain", f_str(64, "所属领域", query=True, code="pbl_domain")), ("sdg_goals", f_str(256, "关联联合国可持续发展目标（逗号分隔）")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_sort", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "sort_no", "deleted"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, ], "codes": ["pbl_authenticity_level", "pbl_complexity_level", "pbl_domain"], } TABLES["pbl_driving_question"] = { "summary": "驱动问题子对象（B7）。由真实问题（B6）提炼的可探究开放性问题：" "问题文本、类型、是否开放、关联问题与关联目标、难度。" "是 Compiler 生成任务链的起点。", "fields": sub_head() + [ ("question_text", f_str(1000, "驱动问题文本", query=True)), ("question_type", f_str(32, "问题类型", default="open", query=True, code="pbl_question_type")), ("is_open_ended", f_int("是否开放式问题 0否 1是", default=1, query=True)), ("linked_problem_id", f_str(32, "关联真实问题ID（B6）", notnull=False, query=True)), ("linked_goal_ids", f_str(512, "关联学习目标ID列表（逗号分隔，B5）")), ("difficulty", f_str(32, "难度", default="medium", query=True, code="pbl_difficulty")), ("expected_outcome", f_str(2000, "预期探究成果")), ("sub_questions", f_text("子问题列表JSON（分解探究路径）")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_sort", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "sort_no", "deleted"]}, {"name": "idx_tenant_problem", "type": "normal", "fields": ["tenant_id", "linked_problem_id"]}, ], "codes": ["pbl_question_type", "pbl_difficulty"], } TABLES["pbl_project"] = { "summary": "项目子对象（B8）。蓝图的执行约束定义：周期、起止日期、团队规模区间、" "约束条件JSON、资源清单JSON、项目预算金额（double(18,2)）与交付物摘要。" "Compiler 据此生成 Game Definition 的时间与资源边界。", "fields": sub_head() + [ ("name", f_str(200, "项目名称", query=True)), ("description", f_str(2000, "项目描述")), ("duration_hours", f_int("项目总课时（小时）", default=0, query=True)), ("start_date", {"type": "date", "notnull": False, "default": None, "editable": True, "query": True, "summary": "计划开始日期"}), ("end_date", {"type": "date", "notnull": False, "default": None, "editable": True, "query": True, "summary": "计划结束日期"}), ("team_size_min", f_int("团队最小人数", default=1)), ("team_size_max", f_int("团队最大人数", default=6)), ("team_count", f_int("团队数量", default=1, query=True)), ("constraint_json", f_text("约束条件JSON（时间/资源/规则边界）")), ("resource_json", f_text("资源清单JSON（场地/设备/材料/经费）")), ("budget_amount", f_money("项目预算金额（元，double(18,2)）", query=True)), ("deliverable_summary", f_str(2000, "交付物摘要")), ("milestone_json", f_text("里程碑JSON（阶段节点+验收标准）")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_sort", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "sort_no", "deleted"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, ], "codes": [], } TABLES["pbl_role"] = { "summary": "角色子对象（B9）。项目内学习者角色定义：角色编码/名称/职责/所需技能/" "人数区间/归属权（ownership：个人/团队/全班）。" "scense_runtime 据此分配玩家角色与权限。", "fields": sub_head() + [ ("role_code", f_str(64, "角色编码（蓝图内唯一）", query=True)), ("role_name", f_str(128, "角色名称", query=True)), ("description", f_str(2000, "角色描述")), ("responsibility", f_str(2000, "角色职责")), ("required_skills", f_str(1000, "所需技能（逗号分隔）")), ("min_count", f_int("最小人数", default=1)), ("max_count", f_int("最大人数", default=1)), ("ownership", f_str(32, "归属权", default="team", query=True, code="pbl_ownership")), ("permission_json", f_text("角色权限JSON（可操作对象与动作白名单）")), ("avatar_hint", f_str(512, "形象提示（供文生图/角色卡渲染）")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_code", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "role_code", "deleted"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, ], "codes": ["pbl_ownership"], } TABLES["pbl_mission"] = { "summary": "任务子对象（B10）。可执行的任务/关卡定义：任务编码/标题/类型/解锁条件/" "前置任务/预计耗时/是否可评估/关联目标。" "Compiler 按 prerequisite 拓扑排序生成任务链，runtime 按 unlock_condition 放行。", "fields": sub_head() + [ ("mission_code", f_str(64, "任务编码（蓝图内唯一）", query=True)), ("title", f_str(200, "任务标题", query=True)), ("description", f_str(4000, "任务描述")), ("mission_type", f_str(32, "任务类型", default="explore", query=True, code="pbl_mission_type")), ("unlock_condition_json", f_text("解锁条件JSON（前置任务/时间/评分门槛）")), ("prerequisite_ids", f_str(512, "前置任务ID列表（逗号分隔，拓扑排序依据）")), ("duration_hours", f_int("预计耗时（小时）", default=0, query=True)), ("assessable", f_int("是否可评估 0否 1是", default=1, query=True)), ("linked_goal_ids", f_str(512, "关联学习目标ID列表（逗号分隔，B5）")), ("linked_role_ids", f_str(512, "关联角色ID列表（逗号分隔，B9）")), ("reward_json", f_text("奖励JSON（积分/徽章/解锁物）")), ("score_weight", f_int("评分权重（百分比 0-100）", default=0, query=True)), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_code", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "mission_code", "deleted"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, {"name": "idx_tenant_type", "type": "normal", "fields": ["tenant_id", "mission_type"]}, ], "codes": ["pbl_mission_type"], } TABLES["pbl_artifact_def"] = { "summary": "产出物定义子对象（B11）。学习者需提交的产出物规格：编码/名称/类型/" "文件格式/是否可评估/关联Rubric/归属权/数量区间/是否必须留证。" "pbl_evidence（M5）按本表采集与校验证据。", "fields": sub_head() + [ ("artifact_code", f_str(64, "产出物编码（蓝图内唯一）", query=True)), ("name", f_str(200, "产出物名称", query=True)), ("description", f_str(2000, "产出物描述")), ("artifact_type", f_str(32, "产出物类型", default="document", query=True, code="pbl_artifact_type")), ("file_format", f_str(64, "允许文件格式（逗号分隔，如 mp4,pdf,png）", query=True)), ("max_size_mb", f_int("单文件最大体积（MB）", default=100)), ("assessable", f_int("是否可评估 0否 1是", default=1, query=True)), ("rubric_id", f_str(32, "关联评分量规ID（pbl_assessment M6）", notnull=False, query=True)), ("ownership", f_str(32, "归属权", default="team", query=True, code="pbl_ownership")), ("min_count", f_int("最少提交数", default=1)), ("max_count", f_int("最多提交数", default=1)), ("evidence_required", f_int("是否必须留证 0否 1是", default=1, query=True)), ("linked_mission_ids", f_str(512, "关联任务ID列表（逗号分隔，B10）")), ("acceptance_criteria", f_str(2000, "验收标准")), ] + audit_tail(), "indexes": [ {"name": "PRIMARY", "type": "primary", "fields": ["id"]}, {"name": "uk_tenant_bp_code", "type": "unique", "fields": ["tenant_id", "blueprint_id", "version_no", "artifact_code", "deleted"]}, {"name": "idx_tenant_bp", "type": "normal", "fields": ["tenant_id", "blueprint_id"]}, {"name": "idx_tenant_type", "type": "normal", "fields": ["tenant_id", "artifact_type"]}, ], "codes": ["pbl_artifact_type", "pbl_ownership"], } # 支撑表（在途旧定义，归档保留；不计入 11 表主清单） SUPPORT_SUMMARY = { "pbl_blueprint_audit": "蓝图操作审计日志（append-only，支撑表）", "pbl_blueprint_edge": "子对象有向关系边（支撑表，weight+关系类型）", "pbl_blueprint_fork": "蓝图 fork 溯源链（支撑表，双向血缘）", "pbl_blueprint_lock": "蓝图编辑锁（支撑表，lock_token+expire 过期判定）", "pbl_blueprint_node": "子对象泛化节点（支撑表，obj_type 判别+payload JSON）", "pbl_blueprint_offline": "离线包导出/导入记录（支撑表，三级回落兜底）", "pbl_blueprint_publish": "发布记录（支撑表，published 后禁删）", "pbl_blueprint_template": "蓝图模板（支撑表，body+body_hash+builtin 平台内置）", "pbl_blueprint_version_delta": "版本变更增量明细（支撑表，added/modified/removed）", } # ------------------------------------------------------------------ 写 models def write_models(): os.makedirs(MODELS, exist_ok=True) sup_dir = os.path.join(MODELS, "support") os.makedirs(sup_dir, exist_ok=True) # 归档在途旧表定义（若存在于主目录） archived = [] for t in SUPPORT_TABLES: src = os.path.join(MODELS, t + ".json") if os.path.isfile(src): shutil.move(src, os.path.join(sup_dir, t + ".json")) archived.append(t) # 主目录只保留设计定稿 11 表 for fn in os.listdir(MODELS): if fn.endswith(".json") and fn[:-5] not in MAIN_TABLES: shutil.move(os.path.join(MODELS, fn), os.path.join(sup_dir, fn)) archived.append(fn[:-5]) for t in MAIN_TABLES: spec = TABLES[t] doc = { "summary": spec["summary"], "primary": ["id"], "fields": {k: v for k, v in spec["fields"]}, "indexes": spec["indexes"], "codes": spec["codes"], } with open(os.path.join(MODELS, t + ".json"), "w", encoding="utf-8") as f: json.dump(doc, f, ensure_ascii=False, indent=2) f.write("\n") print("[models] 主清单 11 表已写入；归档支撑表 %d 个: %s" % (len(archived), archived)) # ------------------------------------------------------------------ 写 json CRUD BROWSER_PICK = { "pbl_blueprint": ["code", "title", "subject", "grade_level", "status", "quality_level", "owner_teacher_id", "current_version"], "pbl_blueprint_version": ["blueprint_id", "version_no", "version_type", "is_current", "snapshot_hash", "commit_time"], "pbl_blueprint_approval": ["blueprint_id", "version_no", "approval_status", "submit_user", "approver_id", "approval_time"], "pbl_learner": ["blueprint_id", "name", "interaction_type", "group_mode", "learner_count", "sort_no"], "pbl_learning_goal": ["blueprint_id", "goal_code", "title", "goal_type", "bloom_level", "measurable", "weight"], "pbl_problem": ["blueprint_id", "title", "authenticity_level", "complexity_level", "domain"], "pbl_driving_question": ["blueprint_id", "question_text", "question_type", "is_open_ended", "difficulty"], "pbl_project": ["blueprint_id", "name", "duration_hours", "start_date", "end_date", "team_count", "budget_amount"], "pbl_role": ["blueprint_id", "role_code", "role_name", "ownership", "min_count", "max_count"], "pbl_mission": ["blueprint_id", "mission_code", "title", "mission_type", "assessable", "duration_hours", "score_weight"], "pbl_artifact_def": ["blueprint_id", "artifact_code", "name", "artifact_type", "file_format", "assessable", "ownership"], } def write_json(): os.makedirs(JSOND, exist_ok=True) sup_dir = os.path.join(JSOND, "support") os.makedirs(sup_dir, exist_ok=True) archived = [] for fn in os.listdir(JSOND): if fn.endswith(".json") and fn[:-5] not in MAIN_TABLES: shutil.move(os.path.join(JSOND, fn), os.path.join(sup_dir, fn)) archived.append(fn[:-5]) for t in MAIN_TABLES: spec = TABLES[t] params = {k: v for k, v in spec["fields"]} doc = { "tblname": t, "params": params, "editable": [ "api/%s/list.dspy" % t, "api/%s/edit.dspy" % t, "api/%s/view.dspy" % t, ], "browserfields": BROWSER_PICK.get(t, ["id", "tenant_id"]), } with open(os.path.join(JSOND, t + ".json"), "w", encoding="utf-8") as f: json.dump(doc, f, ensure_ascii=False, indent=2) f.write("\n") print("[json] 主清单 11 个 CRUD 契约已写入；归档支撑契约 %d 个: %s" % (len(archived), archived)) # ------------------------------------------------------------------ 写 DDL TYPE_MAP = { "str": lambda v: "varchar(%d)" % (v.get("size") or 255), "int": lambda v: "int", "double": lambda v: "double(%d,%d)" % ( (v.get("size")[0] if isinstance(v.get("size"), list) else (v.get("size") or 18)), (v.get("size")[1] if isinstance(v.get("size"), list) else (v.get("scale") or 2))), "datetime": lambda v: "datetime", "date": lambda v: "date", "text": lambda v: "text", "longtext": lambda v: "longtext", } def col_ddl(name, spec): t = spec.get("type", "str") sqltype = TYPE_MAP.get(t, lambda v: "varchar(255)")(spec) parts = [" `%s` %s" % (name, sqltype)] if spec.get("notnull"): parts.append("NOT NULL") d = spec.get("default") if d is not None: if isinstance(d, str): parts.append("DEFAULT '%s'" % d.replace("'", "''")) else: parts.append("DEFAULT %s" % d) elif not spec.get("notnull"): parts.append("DEFAULT NULL") c = spec.get("summary", "") parts.append("COMMENT '%s'" % c.replace("'", "''")) return " ".join(parts) def table_ddl(t, spec): lines = ["-- %s" % t, "CREATE TABLE IF NOT EXISTS `%s` (" % t] cols = [col_ddl(k, v) for k, v in spec["fields"]] for idx in spec["indexes"]: kind = idx.get("type", "normal") flds = ", ".join("`%s`" % x for x in idx["fields"]) if kind == "primary": cols.append(" PRIMARY KEY (%s)" % flds) elif kind == "unique": cols.append(" UNIQUE KEY `%s` (%s)" % (idx["name"], flds)) else: cols.append(" KEY `%s` (%s)" % (idx["name"], flds)) lines.append(",\n".join(cols)) lines.append(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci" " COMMENT='%s';" % spec["summary"].split("。")[0].replace("'", "''")) lines.append("") return "\n".join(lines) HEADER = """-- ============================================================================= -- pbl_blueprint %s -- 方言：mariadb（无 FOREIGN KEY / 无 ENUM / 无 TIMESTAMP，时间统一 datetime） -- 权威来源：projects/pbls/docs/01-design/modules/pbl_blueprint.md §2 表清单 B1~B11 -- 铁律：每表首业务字段 tenant_id varchar(32) NOT NULL，所有读写强制打头； -- 金额字段 double(18,2)；主键 id varchar(32)（UUID 去横线）。 -- ============================================================================= """ def write_sql(): os.makedirs(SQLD, exist_ok=True) core = ["pbl_blueprint", "pbl_blueprint_version", "pbl_blueprint_approval"] subs = [t for t in MAIN_TABLES if t not in core] with open(os.path.join(SQLD, "pbl_blueprint.core.sql"), "w", encoding="utf-8") as f: f.write(HEADER % "核心表 DDL（B1 聚合根 / B2 版本 / B3 审批）") for t in core: f.write(table_ddl(t, TABLES[t])) with open(os.path.join(SQLD, "pbl_blueprint.subobjects.sql"), "w", encoding="utf-8") as f: f.write(HEADER % "7 类子对象表 DDL（B4~B11，独立建表，blueprint_id+version_no 归属聚合根）") for t in subs: f.write(table_ddl(t, TABLES[t])) # 支撑表 DDL：从归档的 models/support 读取（若在途有定义） sup_dir = os.path.join(MODELS, "support") sup_lines = [] if os.path.isdir(sup_dir): for fn in sorted(os.listdir(sup_dir)): if not fn.endswith(".json"): continue try: with open(os.path.join(sup_dir, fn), "r", encoding="utf-8") as f: d = json.load(f) flds = d.get("fields") or {} idxs = d.get("indexes") or [] if not flds: continue cols = [] for k, v in flds.items(): if isinstance(v, dict): cols.append(col_ddl(k, v)) pk = d.get("primary") or ["id"] cols.append(" PRIMARY KEY (%s)" % ", ".join("`%s`" % x for x in pk)) for idx in idxs: if not isinstance(idx, dict): continue fldl = ", ".join("`%s`" % x for x in (idx.get("fields") or [])) if not fldl: continue kind = idx.get("type", "normal") nm = idx.get("name", "idx") if kind == "unique": cols.append(" UNIQUE KEY `%s` (%s)" % (nm, fldl)) elif kind != "primary": cols.append(" KEY `%s` (%s)" % (nm, fldl)) sup_lines.append("-- %s：%s" % (fn[:-5], SUPPORT_SUMMARY.get(fn[:-5], "支撑表"))) sup_lines.append("CREATE TABLE IF NOT EXISTS `%s` (" % fn[:-5]) sup_lines.append(",\n".join(cols)) sup_lines.append(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 " "COLLATE=utf8mb4_general_ci COMMENT='%s';\n" % SUPPORT_SUMMARY.get(fn[:-5], "支撑表")) except Exception as e: # noqa: BLE001 sup_lines.append("-- [WARN] %s 生成失败: %s\n" % (fn, e)) with open(os.path.join(SQLD, "pbl_blueprint.support.sql"), "w", encoding="utf-8") as f: f.write(HEADER % "运维支撑表 DDL（锁/fork溯源/审计/离线包/发布/模板/关系边/泛化节点/版本增量）") f.write("\n".join(sup_lines) if sup_lines else "-- （无支撑表定义）\n") print("[sql] core(%d表) + subobjects(%d表) + support(%d表) 已写入" % (len(core), len(subs), len(sup_lines) // 4 if sup_lines else 0)) def main(): write_models() write_json() write_sql() print("=" * 70) print("M1a 表定义已对齐设计定稿 B1~B11：models 11 + json 11 + sql 3 文件") return 0 if __name__ == "__main__": raise SystemExit(main())