#!/usr/bin/env python3 """Export the Python gateway SQLAlchemy schema as a reviewable Markdown dictionary. This is an offline migration-development tool. It is not part of the Go runtime. """ from __future__ import annotations import argparse import importlib import sys from pathlib import Path CURRENT_TARGETS = { "departments": "gateway.departments(已落地)", "providers": "gateway.providers + gateway.provider_models(已落地,models 拆表)", "users": "gateway.portal_users(已落地)", "virtual_keys": "gateway.api_keys(已落地,字段需转换)", "admin_accounts": "gateway.admin_accounts(已落地)", "audit_logs": "gateway.audit_events(已落地,不迁移无限长正文)", } def default_text(column) -> str: value = column.default if value is None: return "—" argument = value.arg if callable(argument): return getattr(argument, "__name__", "callable") return str(argument).replace("|", "\\|") def render(legacy_root: Path) -> str: sys.path.insert(0, str(legacy_root)) db = importlib.import_module("app.db") tables = [table for name, table in sorted(db.Base.metadata.tables.items()) if name != "audit_log_views"] lines = [ "# 旧 Python 网关数据字典与 ID 迁移映射", "", "> 由 `scripts/export_legacy_dictionary.py` 从旧工程 SQLAlchemy metadata 生成。", "> `audit_log_views` 是兼容查询投影,不作为独立持久化实体,因此这里统计 38 个实体。", "", "## 映射约定", "", "- 旧整数主键统一通过 `gateway.legacy_id_mappings` 映射为 UUID,不在新业务表保留双主键。", "- UUID 使用固定 namespace 的 UUIDv5 生成,输入为 `source_system/entity_type/legacy_id`,重复导入保持幂等。", "- 外键迁移必须先查映射表;缺失映射时整批失败,不允许写入悬空引用。", "- 旧密文不能直接复用:Provider、TOTP、内部服务 Key 等敏感字段在导入时解密后用新用途标签重新加密。", "- `audit_logs` 的完整请求/响应正文不进入新库,只迁移受长度约束的预览与结构化元数据。", "", "## 实体去向", "", "| 旧表 | 新域/目标 | 状态 |", "|---|---|---|", ] for table in tables: target = CURRENT_TARGETS.get(table.name, "对应 M2–M4 领域表") status = "已建表,待批量导入器" if table.name in CURRENT_TARGETS else "已登记,随对应里程碑建表" lines.append(f"| `{table.name}` | {target} | {status} |") lines.extend(["", "## 字段字典", ""]) for table in tables: lines.extend([ f"### `{table.name}`", "", "| 字段 | 旧类型 | 可空 | 默认值 | 键/引用 |", "|---|---|---:|---|---|", ]) for column in table.columns: flags = [] if column.primary_key: flags.append("PK") if column.unique: flags.append("UNIQUE") flags.extend(f"FK → `{fk.target_fullname}`" for fk in column.foreign_keys) lines.append( f"| `{column.name}` | `{column.type}` | {'是' if column.nullable else '否'} | " f"`{default_text(column)}` | {';'.join(flags) or '—'} |" ) lines.append("") return "\n".join(lines).rstrip() + "\n" def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--legacy-root", type=Path, required=True) parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(render(args.legacy_root.resolve()), encoding="utf-8") if __name__ == "__main__": main()