#!/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())