#!/usr/bin/env python3 # -*- coding: utf-8 -*- """M1b DDL 生成器：由 m1b/tables.py 单一事实来源产出双方言 DDL + models JSON。 对应 QC 退回意见 #5/#7：需给出字段/表/关联关系映射 + DDL 产物 + 落库证据。 产物（全部写入仓库，可复核）： modules/pbl_blueprint/pbl_blueprint/sql/m1b_ddl.sql mariadb 方言（生产） modules/pbl_blueprint/pbl_blueprint/sql/m1b_ddl_sqlite.sql sqlite 方言（取证/测试） modules/pbl_blueprint/pbl_blueprint/models/m1b/*.json 4 个表定义（四段式） modules/pbl_blueprint/pbl_blueprint/json/m1b/pbl_blueprint_template.json 离线模板包 projects/pbls/deliverables/m1b/ddl_report.json 生成报告（字段/索引统计） 用法： python3 modules/pbl_blueprint/tools/m1b_gen_ddl.py # 生成全部产物 python3 modules/pbl_blueprint/tools/m1b_gen_ddl.py --check # 只校验一致性，不写盘 一致性保证：mariadb DDL、sqlite DDL、models JSON 三者都由同一份 TABLES 生成， 字段名/顺序/类型语义一一对应；--check 模式会逐表逐字段比对 models JSON 与 TABLES，任何漂移都返回非 0 退出码。 """ import argparse import json import os import sys # --- M1b sys.path bootstrap: modules/ 下各包互为兄弟仓库，需逐个入 path --- _M1B_MOD_ROOT = os.path.abspath(os.path.join(os.path.dirname(os.path.abspath(__file__)), "..")) _M1B_MODULES_DIR = os.path.abspath(os.path.join(_M1B_MOD_ROOT, "..")) _M1B_CANDIDATES = [_M1B_MOD_ROOT, _M1B_MODULES_DIR] try: for _d in sorted(os.listdir(_M1B_MODULES_DIR)): _sub = os.path.join(_M1B_MODULES_DIR, _d) if os.path.isdir(_sub) and not _d.startswith("."): _M1B_CANDIDATES.append(_sub) except OSError: pass for _p in _M1B_CANDIDATES: if _p not in sys.path: sys.path.insert(0, _p) # --- end bootstrap --- HERE = os.path.dirname(os.path.abspath(__file__)) MOD_ROOT = os.path.dirname(HERE) # modules/pbl_blueprint PKG = os.path.join(MOD_ROOT, "pbl_blueprint") REPO_ROOT = os.path.abspath(os.path.join(MOD_ROOT, "..", "..")) if REPO_ROOT not in sys.path: sys.path.insert(0, REPO_ROOT) if MOD_ROOT not in sys.path: sys.path.insert(0, MOD_ROOT) from pbl_blueprint.m1b.tables import TABLES, to_model_json # noqa: E402 from pbl_blueprint.m1b.init import build_sqlite_ddl # noqa: E402 SQL_DIR = os.path.join(PKG, "sql") MODELS_DIR = os.path.join(PKG, "models", "m1b") JSON_DIR = os.path.join(PKG, "json", "m1b") REPORT_DIR = os.path.join(REPO_ROOT, "projects", "pbls", "deliverables", "m1b") #: 抽象类型 -> mariadb 类型 MYSQL_TYPES = { "text": "LONGTEXT", "json": "LONGTEXT", "int": "BIGINT", "bool": "TINYINT(1)", "datetime": "VARCHAR(32)", "float": "DECIMAL(18,4)", } def mysql_type(field): """抽象类型 -> mariadb 列类型（string(N) -> VARCHAR(N)）。""" raw = field.get("type") or "string(255)" base = raw.split("(")[0].strip().lower() if base == "string": size = raw.split("(")[1].rstrip(")") if "(" in raw else "255" return "VARCHAR(%s)" % size return MYSQL_TYPES.get(base, "VARCHAR(255)") def gen_mysql_ddl(tables=None): """生成 mariadb 方言 DDL（InnoDB / utf8mb4 / 无 FOREIGN KEY，Q-OPEN-3）。""" out = [ "-- ============================================================", "-- PBL M1b DDL (mariadb dialect) - generated by tools/m1b_gen_ddl.py", "-- Source of truth: pbl_blueprint/m1b/tables.py", "-- Q-OPEN-3: NO FOREIGN KEY; base tables (world/scene/entity/script)", "-- are NEVER created/altered here (read-only soft refs).", "-- Idempotent: CREATE TABLE IF NOT EXISTS / CREATE INDEX guarded.", "-- ============================================================", "", ] idx_stmts = [] for t in (tables or TABLES): cols = [] for f in t["fields"]: if f.get("pk"): cols.append(" `%s` VARCHAR(40) NOT NULL COMMENT '%s'" % (f["name"], _esc(f.get("comment")))) continue notnull = " NOT NULL" if f.get("required") else " NULL" default = "" if f["name"] == "is_deleted": default = " DEFAULT 0" elif f["name"] == "enabled": default = " DEFAULT 1" elif f["name"] == "resolve_status": default = " DEFAULT 'unresolved'" elif f["name"] == "ownership": default = " DEFAULT 'read_only'" elif f["name"] == "tenant_key": default = " DEFAULT '__tenant__'" cols.append(" `%s` %s%s%s COMMENT '%s'" % ( f["name"], mysql_type(f), notnull, default, _esc(f.get("comment")))) pks = [f["name"] for f in t["fields"] if f.get("pk")] if pks: cols.append(" PRIMARY KEY (`%s`)" % "`, `".join(pks)) out.append("-- %s" % t["summary"].replace("\n", " ")) out.append("CREATE TABLE IF NOT EXISTS `%s` (" % t["name"]) out.append(",\n".join(cols)) out.append(") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 " "COLLATE=utf8mb4_general_ci COMMENT='%s';" % _esc(t["summary"][:60])) out.append("") for ix in t["indexes"]: kw = "UNIQUE INDEX" if ix.get("unique") else "INDEX" idx_stmts.append( "-- %s\nALTER TABLE `%s` ADD %s `%s` (%s);" % ( ix.get("comment", ""), t["name"], kw, ix["name"], ", ".join(["`%s`" % c for c in ix["columns"]]))) out.append("-- ---------- indexes (MySQL 不支持 CREATE INDEX IF NOT EXISTS，") out.append("-- 重复执行会报 1061 Duplicate key name，可忽略；") out.append("-- 首次建库执行本段即可) ----------") out.extend(idx_stmts) out.append("") return "\n".join(out) def _esc(text): """SQL 注释里的单引号转义。""" return (text or "").replace("'", "''").replace("\n", " ") OFFLINE_TEMPLATES = { "templates": [ { "id": "tpl_platform_pbl_stem_v1", "tenant_id": None, "tenant_key": "__platform__", "code": "platform.pbl.stem", "name": "平台公共·STEM 项目式学习模板", "description": "平台公共模板（tenant_id NULL，租户只读）：STEM 主题 PBL 骨架，" "含 7 类子对象默认结构与扩展字段默认值。", "category": "STEM", "scope": "platform", "version": "1.0.0", "status": "published", "source": "offline_fallback", "tags": ["platform", "stem", "public"], "ext_schema": { "difficulty": {"type": "enum", "options": ["入门", "进阶", "高阶"], "default": "进阶"}, "duration_hours": {"type": "int", "default": 12, "min": 1, "max": 200}, "subject": {"type": "string", "default": "综合实践"}, }, "default_values": { "difficulty": "进阶", "duration_hours": 12, "subject": "综合实践", "assessment_mode": "rubric", }, "subobject_policy": { "driving_question": {"min": 1, "default_item": { "question": "我们如何用工程方法解决身边的真实问题？", "cognitive_level": "应用"}}, "learning_goal": {"min": 2, "default_item": { "goal": "能运用跨学科知识完成一个可交付的作品", "bloom_level": "L3", "assessable": True}}, "mission": {"min": 3, "default_item": { "name": "阶段任务", "estimated_minutes": 45}}, "problem": {"min": 1, "default_item": { "title": "核心问题", "difficulty": "中"}}, "role": {"min": 2, "default_item": { "name": "团队成员", "team_size": 4}}, "learner": {"min": 0, "default_item": {"name": "学习者"}}, "artifact_def": {"min": 1, "default_item": { "name": "作品与报告", "artifact_type": "模型"}}, }, "structure": { "blueprint": {"name": "STEM 项目式学习（模板实例）", "status": "draft", "category": "STEM"}, "subobjects": { "driving_question": [ {"question": "我们如何用工程方法解决身边的真实问题？", "sort_no": 1, "ext": {"cognitive_level": "应用", "weight": 0.4}}, ], "learning_goal": [ {"goal": "掌握需求分析与方案设计流程", "sort_no": 1, "ext": {"bloom_level": "L3", "assessable": True}}, {"goal": "完成可演示的原型作品", "sort_no": 2, "ext": {"bloom_level": "L4", "assessable": True}}, ], "mission": [ {"name": "M1 需求调研", "sort_no": 1, "ext": {"estimated_minutes": 45}}, {"name": "M2 方案设计", "sort_no": 2, "ext": {"estimated_minutes": 90}}, {"name": "M3 原型实现与展示", "sort_no": 3, "ext": {"estimated_minutes": 120}}, ], "problem": [ {"title": "如何在有限材料下提升结构强度？", "sort_no": 1, "ext": {"difficulty": "中"}}, ], "role": [ {"name": "项目经理", "sort_no": 1, "ext": {"team_size": 1, "responsibility": "统筹与汇报"}}, {"name": "工程师", "sort_no": 2, "ext": {"team_size": 3, "responsibility": "设计与实现"}}, ], "learner": [], "artifact_def": [ {"name": "设计文档", "sort_no": 1, "ext": {"artifact_type": "文档"}}, {"name": "原型作品", "sort_no": 2, "ext": {"artifact_type": "模型"}}, ], }, }, }, { "id": "tpl_platform_pbl_humanities_v1", "tenant_id": None, "tenant_key": "__platform__", "code": "platform.pbl.humanities", "name": "平台公共·人文社科探究模板", "description": "平台公共模板（tenant_id NULL）：人文社科主题探究骨架。", "category": "人文社科", "scope": "platform", "version": "1.0.0", "status": "published", "source": "offline_fallback", "tags": ["platform", "humanities", "public"], "ext_schema": { "difficulty": {"type": "enum", "options": ["入门", "进阶", "高阶"], "default": "入门"}, "duration_hours": {"type": "int", "default": 8}, "subject": {"type": "string", "default": "社会"}, }, "default_values": {"difficulty": "入门", "duration_hours": 8, "subject": "社会", "assessment_mode": "peer"}, "subobject_policy": { "driving_question": {"min": 1, "default_item": { "question": "社区中的公共议题如何影响我们的生活？"}}, "learning_goal": {"min": 1, "default_item": { "goal": "能用证据支持自己的观点"}}, "mission": {"min": 2, "default_item": {"name": "探究阶段"}}, "problem": {"min": 1, "default_item": {"title": "议题界定"}}, "role": {"min": 1, "default_item": {"name": "调研员"}}, "learner": {"min": 0, "default_item": {"name": "学习者"}}, "artifact_def": {"min": 1, "default_item": {"name": "调研报告"}}, }, "structure": { "blueprint": {"name": "人文社科探究（模板实例）", "status": "draft", "category": "人文社科"}, "subobjects": { "driving_question": [ {"question": "社区中的公共议题如何影响我们的生活？", "sort_no": 1}], "learning_goal": [ {"goal": "能用证据支持自己的观点", "sort_no": 1}], "mission": [{"name": "M1 议题选择", "sort_no": 1}, {"name": "M2 田野调查与汇报", "sort_no": 2}], "problem": [{"title": "如何界定一个可研究的公共议题？", "sort_no": 1}], "role": [{"name": "调研员", "sort_no": 1}, {"name": "记录员", "sort_no": 2}], "learner": [], "artifact_def": [{"name": "调研报告", "sort_no": 1}, {"name": "公开展示", "sort_no": 2}], }, }, }, ] } def write(path, text): """写文件（自动建目录），返回字节数。""" d = os.path.dirname(path) if d: os.makedirs(d, exist_ok=True) with open(path, "w", encoding="utf-8") as fh: fh.write(text) return len(text.encode("utf-8")) def check_models_consistency(): """校验 models/m1b/*.json 与 TABLES 一致（字段名/顺序/索引/唯一键）。""" problems = [] for t in TABLES: path = os.path.join(MODELS_DIR, "%s.json" % t["name"]) if not os.path.exists(path): problems.append({"table": t["name"], "code": "MODEL_MISSING", "path": path}) continue with open(path, "r", encoding="utf-8") as fh: try: model = json.load(fh) except ValueError as exc: problems.append({"table": t["name"], "code": "MODEL_INVALID_JSON", "error": str(exc)}) continue expect = to_model_json(t["name"]) got_fields = [f["name"] for f in (model.get("fields") or [])] exp_fields = [f["name"] for f in expect["fields"]] if got_fields != exp_fields: problems.append({"table": t["name"], "code": "FIELD_MISMATCH", "missing": [f for f in exp_fields if f not in got_fields], "extra": [f for f in got_fields if f not in exp_fields]}) got_ix = sorted([i["name"] for i in (model.get("indexes") or [])]) exp_ix = sorted([i["name"] for i in expect["indexes"]]) if got_ix != exp_ix: problems.append({"table": t["name"], "code": "INDEX_MISMATCH", "got": got_ix, "expect": exp_ix}) for sec in ("summary", "fields", "indexes", "codes"): if sec not in model: problems.append({"table": t["name"], "code": "SECTION_MISSING", "section": sec}) return problems def main(argv=None): ap = argparse.ArgumentParser(description="M1b DDL / models generator") ap.add_argument("--check", action="store_true", help="只校验 models 与 TABLES 一致性，不写盘") ap.add_argument("--no-offline", action="store_true", help="不生成离线模板包") args = ap.parse_args(argv) if args.check: problems = check_models_consistency() print(json.dumps({"mode": "check", "tables": [t["name"] for t in TABLES], "problems": problems, "consistent": not problems}, ensure_ascii=False, indent=2)) return 1 if problems else 0 mysql_ddl = gen_mysql_ddl() sqlite_ddl = build_sqlite_ddl() written = [] written.append({"path": os.path.relpath(os.path.join(SQL_DIR, "m1b_ddl.sql"), REPO_ROOT), "bytes": write(os.path.join(SQL_DIR, "m1b_ddl.sql"), mysql_ddl), "dialect": "mariadb", "lines": mysql_ddl.count("\n") + 1}) written.append({"path": os.path.relpath(os.path.join(SQL_DIR, "m1b_ddl_sqlite.sql"), REPO_ROOT), "bytes": write(os.path.join(SQL_DIR, "m1b_ddl_sqlite.sql"), sqlite_ddl), "dialect": "sqlite", "lines": sqlite_ddl.count("\n") + 1}) for t in TABLES: p = os.path.join(MODELS_DIR, "%s.json" % t["name"]) txt = json.dumps(to_model_json(t["name"]), ensure_ascii=False, indent=2) + "\n" written.append({"path": os.path.relpath(p, REPO_ROOT), "bytes": write(p, txt), "dialect": "model-json", "table": t["name"], "fields": len(t["fields"]), "indexes": len(t["indexes"])}) if not args.no_offline: p = os.path.join(JSON_DIR, "pbl_blueprint_template.json") txt = json.dumps(OFFLINE_TEMPLATES, ensure_ascii=False, indent=2) + "\n" written.append({"path": os.path.relpath(p, REPO_ROOT), "bytes": write(p, txt), "dialect": "offline-package", "templates": len(OFFLINE_TEMPLATES["templates"])}) report = { "generator": "tools/m1b_gen_ddl.py", "source_of_truth": "pbl_blueprint/m1b/tables.py", "tables": [{"name": t["name"], "summary": t["summary"], "field_count": len(t["fields"]), "fields": [f["name"] for f in t["fields"]], "indexes": [{"name": i["name"], "unique": bool(i.get("unique")), "columns": i["columns"], "comment": i.get("comment", "")} for i in t["indexes"]], "codes": t.get("codes") or {}} for t in TABLES], "q_open_3": {"foreign_keys": 0, "base_tables_touched": [], "note": "无 FOREIGN KEY；不 CREATE/ALTER world/scene/entity/script"}, "unique_keys_tenant_first": all( (i.get("unique") and i["columns"][0] == "tenant_key") for t in TABLES for i in t["indexes"]), "written": written, "consistency_problems": check_models_consistency(), } rp = os.path.join(REPORT_DIR, "ddl_report.json") write(rp, json.dumps(report, ensure_ascii=False, indent=2) + "\n") print(json.dumps({"ok": True, "written": [w["path"] for w in written], "report": os.path.relpath(rp, REPO_ROOT), "tables": [t["name"] for t in TABLES], "consistency_problems": report["consistency_problems"]}, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": sys.exit(main())