206 lines
8.8 KiB
Python
206 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""M5a-DISC.1：以 ``models/*.json`` 为**单一真源**再生成 MariaDB DDL。
|
||
|
||
背景（QC 退回 #4）：仓库内 ``sql/pbl_evidence.sql`` 是旧模型残留——把 ``id`` 渲染成
|
||
``BIGINT UNSIGNED NOT NULL AUTO_INCREMENT``（与 models 的 ``str(32)`` 主键冲突），
|
||
引用字段 ``artifact_id/session_id/blueprint_id`` 渲染成 ``BIGINT UNSIGNED``（与 models
|
||
的 ``str(32)`` 冲突），并且存在 ``NOT NULL NOT NULL`` / ``NULL NULL`` 重复渲染、
|
||
``COMMENT`` 无引号等生成缺陷。宿主 ``xls2ddl/json2ddl`` 在本开发机未安装
|
||
（``import xls2ddl`` ModuleNotFoundError），故本脚本作为**仓库内可复现**的等价生成器：
|
||
纯标准库、无第三方依赖，读 models 四段式 JSON → 输出 MariaDB 方言 DDL。
|
||
|
||
生成规则（与 database-table-definition-spec 对齐）：
|
||
* 抽象类型 → MariaDB 原生类型映射表见 ``TYPE_MAP``；``text`` → ``LONGTEXT``。
|
||
* ``nullable == "no"`` → ``NOT NULL``（**只渲染一次**，其他写法一律按可空，不重复输出）。
|
||
* ``default`` 存在 → ``DEFAULT`` 字面量（字符串加单引号并转义，数字/None 原样）。
|
||
* 主键取 ``summary[0].primary``（数组），**绝不输出 AUTO_INCREMENT**——主键由应用层
|
||
``pbl_evidence/pk.py:gen_pk()`` 生成（DISC.1 落定方案 a）。
|
||
* ``indexes`` → ``UNIQUE KEY`` / ``KEY``，列顺序保序（``idxfields`` 数组）。
|
||
* ``title`` → ``COMMENT '...'``（带引号）。
|
||
|
||
用法::
|
||
|
||
python3 scripts/gen_ddl_from_models.py # 生成 sql/pbl_evidence.sql
|
||
python3 scripts/gen_ddl_from_models.py --check # 只校验磁盘产物与 models 一致
|
||
python3 scripts/gen_ddl_from_models.py --stdout # 打印不落盘
|
||
|
||
退出码：0 成功；1 生成物与磁盘不一致（--check）或 models 解析失败。
|
||
"""
|
||
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
|
||
ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||
MODELS_DIR = os.path.join(ROOT, 'models')
|
||
SQL_DIR = os.path.join(ROOT, 'sql')
|
||
|
||
# 表顺序（与模块内数据流一致：先产出物后证据）
|
||
TABLE_ORDER = ('pbl_artifact', 'pbl_evidence')
|
||
|
||
# 抽象类型 → MariaDB 原生类型（str/char/float/double/ddouble/decimal 需带 length/dec）
|
||
TYPE_MAP = {
|
||
'str': lambda f: 'VARCHAR(%d)' % int(f['length']),
|
||
'char': lambda f: 'CHAR(%d)' % int(f['length']),
|
||
'short': lambda f: 'SMALLINT',
|
||
'int': lambda f: 'INT',
|
||
'long': lambda f: 'BIGINT',
|
||
'float': lambda f: 'DECIMAL(%d,%d)' % (int(f['length']), int(f['dec'])),
|
||
'double': lambda f: 'DECIMAL(%d,%d)' % (int(f['length']), int(f['dec'])),
|
||
'ddouble': lambda f: 'DECIMAL(%d,%d)' % (int(f['length']), int(f['dec'])),
|
||
'decimal': lambda f: 'DECIMAL(%d,%d)' % (int(f['length']), int(f['dec'])),
|
||
'date': lambda f: 'DATE',
|
||
'time': lambda f: 'TIME',
|
||
'datetime': lambda f: 'DATETIME',
|
||
'timestamp': lambda f: 'DATETIME',
|
||
'text': lambda f: 'LONGTEXT',
|
||
'bin': lambda f: 'LONGBLOB',
|
||
}
|
||
|
||
GEN_BANNER = (
|
||
'-- 本文件由 scripts/gen_ddl_from_models.py 自动生成，**单一真源 = models/*.json**。\n'
|
||
'-- 禁止手工编辑本文件（改表结构请改 models/{table}.json 后重跑本脚本）。\n'
|
||
'-- 生成命令：python3 scripts/gen_ddl_from_models.py\n'
|
||
'--\n'
|
||
'-- M5a-DISC.1 主键口径（models / DDL / collector 三方对齐）：\n'
|
||
'-- id = VARCHAR(32) NOT NULL，无 AUTO_INCREMENT、无 DEFAULT；\n'
|
||
'-- 主键值由应用层 pbl_evidence/pk.py:gen_pk()（appPublic.uniqueID.getID，\n'
|
||
'-- 离线降级 uuid4().hex，长度均 <= 32）在 INSERT 前生成并显式写入列集合。\n'
|
||
'-- 引用字段 tenant_id/artifact_id/session_id/blueprint_id/learner_id/\n'
|
||
'-- source_event_id/dedup_key/collect_batch_id 一律 VARCHAR(32)，与主键同型，\n'
|
||
'-- JOIN / 等值比较不存在类型不匹配。\n'
|
||
)
|
||
|
||
|
||
def _q(text):
|
||
"""SQL 单引号字面量（转义内部单引号，去掉换行避免破坏 DDL 行结构）。"""
|
||
s = str(text if text is not None else '')
|
||
s = s.replace('\\', '\\\\').replace("'", "''")
|
||
s = s.replace('\r', ' ').replace('\n', ' ')
|
||
return "'%s'" % s
|
||
|
||
|
||
def load_model(table):
|
||
path = os.path.join(MODELS_DIR, '%s.json' % table)
|
||
if not os.path.isfile(path):
|
||
raise RuntimeError('缺少表定义 %s' % path)
|
||
with open(path, 'r', encoding='utf-8') as fh:
|
||
model = json.load(fh)
|
||
for section in ('summary', 'fields'):
|
||
if not model.get(section):
|
||
raise RuntimeError('%s: 四段式缺少/空 %s 段' % (table, section))
|
||
return model
|
||
|
||
|
||
def render_column(field):
|
||
name = field['name']
|
||
ftype = (field.get('type') or '').strip()
|
||
if ftype not in TYPE_MAP:
|
||
raise RuntimeError('%s: 未支持抽象类型 %r' % (name, ftype))
|
||
spec = TYPE_MAP[ftype](field)
|
||
|
||
parts = [' `%s` %s' % (name, spec)]
|
||
# NOT NULL 只判定一次（nullable 必须是字符串 'no'，其他写法按可空处理）
|
||
parts.append('NOT NULL' if field.get('nullable') == 'no' else 'NULL')
|
||
if 'default' in field and field['default'] is not None:
|
||
dv = field['default']
|
||
if isinstance(dv, bool):
|
||
dv = int(dv)
|
||
if isinstance(dv, (int, float)):
|
||
parts.append('DEFAULT %s' % dv)
|
||
elif ftype in ('int', 'long', 'short', 'float', 'double', 'ddouble', 'decimal') \
|
||
and str(dv).replace('.', '', 1).replace('-', '', 1).isdigit():
|
||
parts.append('DEFAULT %s' % dv)
|
||
else:
|
||
parts.append('DEFAULT %s' % _q(dv))
|
||
if field.get('title'):
|
||
parts.append('COMMENT %s' % _q(field['title']))
|
||
return ' '.join(parts)
|
||
|
||
|
||
def render_table(table):
|
||
model = load_model(table)
|
||
summary = model['summary'][0]
|
||
pk = summary.get('primary') or []
|
||
if not isinstance(pk, list) or not pk:
|
||
raise RuntimeError('%s: summary[0].primary 必须是非空数组' % table)
|
||
fields = model['fields']
|
||
names = [f['name'] for f in fields]
|
||
missing = [c for c in pk if c not in names]
|
||
if missing:
|
||
raise RuntimeError('%s: 主键列不在 fields 中: %s' % (table, missing))
|
||
|
||
lines = ['CREATE TABLE IF NOT EXISTS `%s` (' % table]
|
||
body = [render_column(f) for f in fields]
|
||
body.append(' PRIMARY KEY (%s)' % ', '.join('`%s`' % c for c in pk))
|
||
for idx in (model.get('indexes') or []):
|
||
idxfields = idx.get('idxfields')
|
||
if not isinstance(idxfields, list) or not idxfields:
|
||
raise RuntimeError('%s: 索引 %s 的 idxfields 必须是非空数组' % (table, idx.get('name')))
|
||
bad = [c for c in idxfields if c not in names]
|
||
if bad:
|
||
raise RuntimeError('%s: 索引 %s 引用了不存在的列 %s' % (table, idx.get('name'), bad))
|
||
kw = 'UNIQUE KEY' if idx.get('idxtype') == 'unique' else 'KEY'
|
||
body.append(' %s `%s` (%s)' % (kw, idx.get('name'),
|
||
', '.join('`%s`' % c for c in idxfields)))
|
||
lines.append(',\n'.join(body))
|
||
lines.append(') ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci COMMENT=%s;'
|
||
% _q(summary.get('title') or table))
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def build_sql(tables=TABLE_ORDER):
|
||
chunks = [GEN_BANNER]
|
||
for t in tables:
|
||
chunks.append('-- 表 %s（真源 models/%s.json）' % (t, t))
|
||
chunks.append(render_table(t))
|
||
chunks.append('')
|
||
return '\n'.join(chunks).rstrip() + '\n'
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument('--check', action='store_true', help='只比对磁盘产物，不写文件')
|
||
ap.add_argument('--stdout', action='store_true', help='打印到标准输出，不写文件')
|
||
ap.add_argument('--out', default=os.path.join(SQL_DIR, 'pbl_evidence.sql'))
|
||
args = ap.parse_args()
|
||
|
||
try:
|
||
sql = build_sql()
|
||
except Exception as exc:
|
||
print('GEN DDL FAILED: %s' % exc)
|
||
return 1
|
||
|
||
if args.stdout:
|
||
sys.stdout.write(sql)
|
||
return 0
|
||
|
||
old = ''
|
||
if os.path.isfile(args.out):
|
||
with open(args.out, 'r', encoding='utf-8') as fh:
|
||
old = fh.read()
|
||
|
||
if args.check:
|
||
if old == sql:
|
||
print('[PASS] %s 与 models/*.json 再生成结果一致（%d 字节）'
|
||
% (os.path.relpath(args.out, ROOT).replace(os.sep, '/'), len(sql)))
|
||
return 0
|
||
print('[FAIL] %s 与 models 再生成结果不一致（磁盘 %d 字节 / 应为 %d 字节）'
|
||
% (os.path.relpath(args.out, ROOT).replace(os.sep, '/'), len(old), len(sql)))
|
||
return 1
|
||
|
||
d = os.path.dirname(args.out)
|
||
if not os.path.isdir(d):
|
||
os.makedirs(d)
|
||
with open(args.out, 'w', encoding='utf-8') as fh:
|
||
fh.write(sql)
|
||
print('[OK] 已生成 %s（%d 字节，真源 models/%s.json）'
|
||
% (os.path.relpath(args.out, ROOT).replace(os.sep, '/'), len(sql), '/'.join(TABLE_ORDER)))
|
||
return 0
|
||
|
||
|
||
if __name__ == '__main__':
|
||
sys.exit(main())
|