#!/usr/bin/env python # -*- coding: utf-8 -*- """ DDL 同步:docs/db/schema-postgresql.sql → backend/.../db/migration/V1__business_tables.sql 为什么要有这个脚本 ------------------ 文档里的 DDL 是**交付物**(已实测、写进验收标准),Flyway 迁移是**工程产物**。 两者必须字字相同,否则"文档说一套、代码跑另一套"——这是最难查的一类事故。 所以:迁移文件由脚本生成,禁止手改;改 DDL 一律先改文档,再跑本脚本。 用法 ---- python scripts/sync-ddl.py # 生成/更新迁移文件 python scripts/sync-ddl.py --check # 只校验是否同步(CI 用,不同步则退出码 1) """ from __future__ import annotations import io import os import sys # Windows 控制台默认不是 UTF-8 if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) SOURCE = os.path.join(ROOT, "docs", "db", "schema-postgresql.sql") TARGET = os.path.join( ROOT, "backend", "xt-server", "src", "main", "resources", "db", "migration", "V1__business_tables.sql", ) HEADER = """\ -- ═══════════════════════════════════════════════════════════════════ -- V1 · 业务表(20 张) -- -- ⚠️ 本文件由脚本生成,请勿手改! -- 来源 :docs/db/schema-postgresql.sql -- 重新生成 :python scripts/sync-ddl.py -- 校验同步 :python scripts/sync-ddl.py --check -- -- 源文件已在 PostgreSQL 16.15 实测零错误: -- 20 表 / 177 列注释 / 65 索引 / 20 表注释 -- ═══════════════════════════════════════════════════════════════════ """ def read(path: str) -> str: with io.open(path, encoding="utf-8") as f: return f.read() def normalize(text: str) -> str: """统一行尾并去掉首尾空白,用于比对。""" return text.replace("\r\n", "\n").replace("\r", "\n").strip() def render(source_text: str) -> str: return HEADER + normalize(source_text) + "\n" def main() -> int: check_only = "--check" in sys.argv if not os.path.exists(SOURCE): print(f"[X] 找不到源文件:{SOURCE}") return 1 expected = render(read(SOURCE)) if check_only: if not os.path.exists(TARGET): print(f"[X] 迁移文件不存在:{TARGET}") return 1 actual = io.open(TARGET, encoding="utf-8").read() if normalize(actual) != normalize(expected): print("[X] DDL 已漂移:文档与迁移文件不一致") print(f" 文档 :{SOURCE}") print(f" 迁移 :{TARGET}") print(" 修复 :python scripts/sync-ddl.py") return 1 print("[OK] DDL 已同步") return 0 os.makedirs(os.path.dirname(TARGET), exist_ok=True) with io.open(TARGET, "w", encoding="utf-8", newline="\n") as f: f.write(expected) tables = expected.count("CREATE TABLE ") indexes = expected.count("CREATE INDEX ") + expected.count("CREATE UNIQUE INDEX ") comments = expected.count("COMMENT ON COLUMN ") print(f"[OK] 已生成 {os.path.relpath(TARGET, ROOT)}") print(f" {tables} 表 / {indexes} 索引 / {comments} 列注释 / " f"{len(expected.encode('utf-8'))} 字节") return 0 if __name__ == "__main__": raise SystemExit(main())