402 lines
19 KiB
Python
402 lines
19 KiB
Python
#!/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())
|