""" 跨库数据迁移脚本(MySQL → PostgreSQL / SQL Server / SQLite / Oracle 等) ============================================================= 把「源库」的数据整体搬到「目标库」: - 目标库结构用模型 metadata 生成(db.metadata.create_all,方言可移植, 模型即 head,与 src/models_registry.py / setup_db.py 同源); - 数据按拓扑序逐表 SELECT ... ORDER BY id,显式带 id 批量插入,保住 无外键裸列引用关系(user.dept_id、role_menu.role_id/menu_id 等); - 最后按方言重置自增序列(PG setval / MSSQL IDENTITY_INSERT + DBCC CHECKIDENT / Oracle 补 sequence + BEFORE INSERT 触发器; MySQL / SQLite 显式插 id 后自动跟随,无需处理)。 与 init_db.py / setup_db.py 的分工: - init_db.py / setup_db.py 单库初始化(当前 .env 目标库 create_all + 种子数据) - migrate_db.py 双库迁移(源库 → 目标库,CLI 显式参数,不依赖 .env 连接串) 用法: python scripts/migrate_db.py \ --src-driver=mysql --src-host=127.0.0.1 --src-port=3306 \ --src-db=djangoadmin.flask.elevue --src-user=root --src-pass=root \ --dst-driver=postgresql --dst-host=127.0.0.1 --dst-port=5432 \ --dst-db=djangoadmin.flask.elevue --dst-user=postgres --dst-pass=123456 \ --drop-target-first 先干跑核对计划:加 --dry-run(只连源、逐表数行数、打印计划,不碰目标库)。 """ import argparse import os import sys import time from contextlib import contextmanager from pathlib import Path from urllib.parse import quote_plus # 将项目根加入 sys.path(脚本以 `python scripts/xxx.py` 运行时 cwd 是 scripts/) ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if ROOT not in sys.path: sys.path.insert(0, ROOT) # Windows 控制台默认 GBK(cp936):✓ 等字符无法编码会崩溃、中文按终端编码显示成乱码。 # 统一把 stdout/stderr 切到 UTF-8,Git Bash / 重定向均正常;无 reconfigure 的环境跳过。 if sys.stdout and hasattr(sys.stdout, 'reconfigure'): sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') from sqlalchemy import create_engine, inspect, select, text # noqa: E402 from sqlalchemy.pool import NullPool # noqa: E402 from config import DB_PREFIX # noqa: E402 from config.database import _normalize_driver, _sqlite_url # noqa: E402 import src.core.database as _dbm # noqa: E402 import src.models_registry as _models_registry # noqa: E402,F401 # 模型注册中心:填充 db.metadata db = _dbm.db # 各驱动默认端口(sqlite 无端口概念) _DEFAULT_PORTS = {"mysql": 3306, "postgresql": 5432, "mssql": 1433, "oracle": 1521, "sqlite": None} # 迁移顺序:基础/被引用表在前,依赖表在后。项目无外键,表间关联是裸 Integer 列 # 由应用层维护,显式带 id 插入即保住引用关系;保序只是锦上添花(父先子后)。 _BASE_TABLES = [ "dept", "menu", "role", "level", "position", "city", "category", "dict", "config", "file_template", "param", "job", "link", ] _DEP_TABLES = [ "role_menu", "dict_item", "config_item", "job_log", "user", "user_role", "article", "notice", "operation_log", "login_log", ] def build_parser() -> argparse.ArgumentParser: """CLI 参数:源/目标两端连接显式给出,不依赖 .env。""" p = argparse.ArgumentParser(description="跨库数据迁移:源库 → 目标库") def add_side(side: str, default_user: str): p.add_argument(f"--{side}-driver", choices=("mysql", "postgresql", "mssql", "oracle", "sqlite"), default="mysql", help="驱动(oracle 时 --{side}-db 即 service_name;sqlite 时即数据库文件路径)") p.add_argument(f"--{side}-host", default="127.0.0.1") p.add_argument(f"--{side}-port", type=int, default=None, help="端口,缺省按驱动默认(mysql 3306 / pg 5432 / mssql 1433 / oracle 1521)") p.add_argument(f"--{side}-db", required=True, help="数据库名(sqlite 时为文件路径)") p.add_argument(f"--{side}-user", default=default_user) p.add_argument(f"--{side}-pass", default="", help=f"口令,留空回落环境变量 {side.upper()}_DB_PASS(避免进 shell 历史)") add_side("src", "root") add_side("dst", "postgres") p.add_argument("--drop-target-first", action="store_true", help="先删目标库再重建(默认关)") p.add_argument("--dry-run", action="store_true", help="只连源、逐表数行数、打印计划,不碰目标库") p.add_argument("--chunk-size", type=int, default=1000, help="每批插入行数(默认 1000)") p.add_argument("--only-tables", default="", help="逗号分隔的表名子集(全名或去掉前缀的短名),续传用") p.add_argument("--skip-existing", action="store_true", help="目标已等于源行数的表跳过") p.add_argument("--skip-verify", action="store_true", help="跳过迁移后的逐表行数校验") p.add_argument("--single-transaction", action="store_true", help="全部表包进一个大事务(默认每表一个事务)") return p # ============================================================ # URL / 引擎 # ============================================================ def make_url(driver: str, host: str, port: int, database: str, username: str, password: str) -> str: """带显式参数的连接串构建器,URL 形态与 config.database._build_database_url 逐字等价 (后者读 .env 全局,脚本端需要显式参数版)。""" drv = _normalize_driver(driver) if drv == 'sqlite': return _sqlite_url(database) # 此时 database 即文件路径,忽略 host/port/user/pass base = quote_plus(username) + ':' + quote_plus(password) + '@' + host + ':' + str(port) if drv == 'oracle': # Oracle 的"库"即连接的服务名(service_name),schema 是连接用户 return 'oracle+oracledb://' + base + '/?service_name=' + quote_plus(database) base = base + '/' + database if drv == 'mysql': return 'mysql+pymysql://' + base + '?charset=utf8mb4' if drv == 'postgresql': return 'postgresql+psycopg2://' + base return 'mssql+pymssql://' + base + '?charset=utf8' def make_engine(driver: str, host: str, port: int, database: str, username: str, password: str): """迁移专用引擎:NullPool,避免连接池把目标/源的会话状态串起来。""" return create_engine(make_url(driver, host, port, database, username, password), poolclass=NullPool) def _resolve_port(driver: str, port: int | None) -> int | None: return port if port is not None else _DEFAULT_PORTS[driver] def _fmt_duration(sec: float) -> str: """耗时格式化:<1s 显示毫秒,<60s 显示秒,否则分+秒。""" if sec < 1: return f"{sec * 1000:.0f}ms" if sec < 60: return f"{sec:.1f}s" return f"{int(sec // 60)}m{int(sec % 60)}s" def _busy(msg: str) -> None: """缓冲提示:耗时操作开始前打印一行,让用户感知进程仍在工作(而非卡死)。""" print(f" · {msg}...", flush=True) @contextmanager def step(title: str): """耗时步骤缓冲包装:进入打印"正在...(请勿中断)",退出打印"✓ 完成(耗时)"。 用于删库/建库/建表/大表迁移等可能耗时较长的操作,避免用户误以为进程卡死。 通过 `with step(...) as h: h['detail'] = ...` 可给完成行追加明细(如行数/失败数)。 """ t0 = time.perf_counter() print(f"[·] {title}...(可能耗时较长,请勿中断)", flush=True) holder: dict = {} try: yield holder finally: detail = f",{holder['detail']}" if holder.get('detail') else "" print(f"[✓] {title} 完成(耗时 {_fmt_duration(time.perf_counter() - t0)}){detail}", flush=True) # ============================================================ # 建目标库(镜像 init_db.py 的分支) # ============================================================ def create_database_if_missing(driver: str, host: str, port: int, database: str, username: str, password: str, drop_first: bool = False, tables: tuple = ()): """按方言连维护库建目标库(不存在才建)。sqlite 无建库概念,只确保父目录并可选删旧文件。 - mysql: 连无库连接 CREATE DATABASE ... utf8mb4 - postgresql: 连 postgres 维护库(autocommit),库名含点号必须双引号 - mssql: 连 master 维护库,标识符用方括号 - oracle: 服务由 DBA 管理,schema 即连接用户,无需/无法"建库";仅校验连通性, drop_first 时尽力 DROP TABLE PURGE 清掉 tables 里已存在的项目表 """ if driver == 'sqlite': file = Path(database).expanduser() if file.parent.as_posix() != '.': file.parent.mkdir(parents=True, exist_ok=True) if drop_first and file.exists(): _busy(f"正在删除旧目标文件 {file}") file.unlink() print(f"removed existing sqlite file {file}") return if driver == 'mysql': import pymysql _busy("正在连接 MySQL(维护实例)") conn = pymysql.connect(host=host, port=int(port), user=username, password=password, autocommit=True) cur = conn.cursor() if drop_first: _busy(f"正在删除旧目标库 `{database}`") cur.execute(f"DROP DATABASE IF EXISTS `{database}`") _busy(f"正在创建目标库 `{database}`(utf8mb4)") cur.execute( f"CREATE DATABASE IF NOT EXISTS `{database}` " f"DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_general_ci" ) cur.close() conn.close() return if driver == 'postgresql': import psycopg2 _busy("正在连接 PostgreSQL(postgres 维护库)") # 连 postgres 维护库创建目标库;DB_DATABASE 含点号,标识符必须双引号 conn = psycopg2.connect( host=host, port=int(port), user=username, password=password, dbname="postgres", ) conn.autocommit = True with conn.cursor() as cur: cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (database,)) exists = cur.fetchone() is not None if drop_first and exists: _busy(f'正在删除旧目标库 "{database}"(DROP ... WITH FORCE)') cur.execute(f'DROP DATABASE "{database}" WITH (FORCE)') exists = False if not exists: _busy(f'正在创建目标库 "{database}"(UTF8)') # 强制 UTF8 编码(TEMPLATE template0 避免继承模板库非 UTF8 编码导致 CJK 乱码) cur.execute(f'CREATE DATABASE "{database}" ENCODING \'UTF8\' TEMPLATE template0') conn.close() return if driver == 'oracle': # 服务由 DBA 管理,schema 就是连接用户;连上即视为"目标库就绪" import oracledb _busy(f"正在连接 Oracle 服务 {database}") conn = oracledb.connect(user=username, password=password, dsn=f"{host}:{port}/{database}") cur = conn.cursor() cur.execute("SELECT 1 FROM dual") cur.fetchone() if drop_first: _busy(f"正在清理 {len(tables)} 张旧项目表") for t in tables: try: cur.execute(f"DROP TABLE {t} PURGE") except Exception: pass # 表不存在则跳过 conn.close() print(f"[OK] Oracle 服务 {database} 连接成功(schema={username})") if drop_first: print(f" drop-first:已尽力清理 {len(tables)} 张项目表") return # mssql:连 master 维护库创建目标库 import pymssql _busy("正在连接 SQL Server(master 维护库)") conn = pymssql.connect( server=host, user=username, password=password, port=int(port), database="master", autocommit=True, ) cur = conn.cursor() if drop_first: _busy(f"正在删除旧目标库 [{database}](SINGLE_USER + DROP)") cur.execute( f"IF DB_ID(N'{database}') IS NOT NULL " f"ALTER DATABASE [{database}] SET SINGLE_USER WITH ROLLBACK IMMEDIATE" ) cur.execute(f"IF DB_ID(N'{database}') IS NOT NULL DROP DATABASE [{database}]") _busy(f"正在创建目标库 [{database}]") cur.execute(f"IF DB_ID(N'{database}') IS NULL CREATE DATABASE [{database}]") cur.close() conn.close() # ============================================================ # 迁移顺序 # ============================================================ def ordered_tables(): """按拓扑序返回要迁移的 Table 对象(base 先、dep 后;metadata 中未列出的追加末尾)。""" out, seen = [], set() for name in _BASE_TABLES + _DEP_TABLES: tbl = db.metadata.tables.get(DB_PREFIX + name) if tbl is not None: out.append(tbl) seen.add(tbl.name) for tbl in db.metadata.tables.values(): if tbl.name not in seen: out.append(tbl) return out # ============================================================ # 泵数据 # ============================================================ def sanitize(payload: list[dict], table) -> list[dict]: """把驱动读回的零日期/非法日期字符串统一置 None(防御性兜底)。 PyMySQL 默认已把 MySQL 的 '0000-00-00 00:00:00' 转成 None;此处兜底处理 个别驱动返回原始字符串的情况,避免 PostgreSQL/MSSQL 拒绝非法日期。 本项目时间列全部 nullable,置 None 安全。 """ for row in payload: for col in table.columns: v = row.get(col.name) if isinstance(v, str) and v.startswith('0000-00-00'): row[col.name] = None return payload def pump_table(src, dst, table, chunk_size: int, driver: str) -> int: """逐 chunk 批量插入一张表,显式带 id 以保住裸列引用关系。 src/dst 为已就绪的 Connection;dst 处于调用方开启的事务内 (默认每表一个事务;--single-transaction 时全库一个大事务)。 mssql 必须 SET IDENTITY_INSERT ON 才能显式插自增 id。 """ pk = next(iter(table.primary_key.columns)) sel = select(*table.columns).order_by(pk) # ORDER BY id:自引用父先子后 use_identity = (driver == 'mssql') total = 0 t0 = time.perf_counter() if use_identity: dst.execute(text(f"SET IDENTITY_INSERT [{table.name}] ON")) try: result = src.execute(sel) while True: rows = result.fetchmany(chunk_size) if not rows: break payload = sanitize([dict(r._mapping) for r in rows], table) dst.execute(table.insert(), payload) total += len(payload) print(f" [{table.name}] +{len(payload)}(累计 {total})") finally: if use_identity: dst.execute(text(f"SET IDENTITY_INSERT [{table.name}] OFF")) print(f" [✓ {table.name}] 共 {total} 行(耗时 {_fmt_duration(time.perf_counter() - t0)})") return total # ============================================================ # 自增序列重置 # ============================================================ def pg_reset_sql(table_name: str) -> str: """PostgreSQL:把 SERIAL/IDENTITY 序列推进到 MAX(id),否则显式插 id 后新插入会主键冲突。 - 非空表:setval(seq, MAX(id)),is_called=true → 下一 id = MAX(id)+1 - 空表: setval(seq, 1, false),is_called=false → 下一 id = 1 (不能 setval(seq, 0):PG 序列最小值是 1,会抛 NumericValueOutOfRange) """ return ( f"SELECT setval(pg_get_serial_sequence('{table_name}', 'id'), " f"COALESCE((SELECT MAX(id) FROM \"{table_name}\"), 1), " f"(SELECT MAX(id) FROM \"{table_name}\") IS NOT NULL)" ) def mssql_reset_sql(table_name: str, max_id) -> str: """SQL Server:RESEED 必须是字面量,先查 MAX(id) 再拼。空表 RESEED 0 → 下一 id=1。""" return f"DBCC CHECKIDENT ('{table_name}', RESEED, {max_id or 0})" def oracle_sequence_sql(table_name: str, start_with: int) -> str: """Oracle:为无自增机制的 INTEGER 主键建 sequence,START WITH 决定下一 id。""" return f"CREATE SEQUENCE {table_name}_id_seq START WITH {start_with} INCREMENT BY 1 NOCACHE" def oracle_trigger_sql(table_name: str) -> str: """Oracle:BEFORE INSERT 触发器,id 为 NULL 时从 sequence 取 NEXTVAL。""" seq = f"{table_name}_id_seq" trg = f"{table_name}_id_trg" return ( f"CREATE OR REPLACE TRIGGER {trg}\n" f"BEFORE INSERT ON {table_name} FOR EACH ROW\n" f"WHEN (NEW.id IS NULL)\n" f"BEGIN SELECT {seq}.NEXTVAL INTO :NEW.id FROM dual; END;" ) def ensure_oracle_autoincrement(conn, table_name: str): """Oracle 迁移后补自增机制:重建 sequence(START WITH MAX(id)+1)+ BEFORE INSERT 触发器。 Oracle 的 create_all 对 Integer 自增主键只生成 INTEGER NOT NULL(无 identity/sequence), 不补的话应用 save() 拿不到新 id。空表 MAX(id) 为 NULL → COALESCE 0 → START WITH 1。 Oracle 无 DROP SEQUENCE IF EXISTS,先查 user_sequences 再删。 """ max_id = conn.execute(text(f"SELECT COALESCE(MAX(id), 0) FROM {table_name}")).scalar() seq = f"{table_name}_id_seq" exists = conn.execute(text( "SELECT COUNT(*) FROM user_sequences WHERE sequence_name = UPPER(:n)" ), {"n": seq}).scalar() if exists: conn.execute(text(f"DROP SEQUENCE {seq}")) conn.execute(text(oracle_sequence_sql(table_name, max_id + 1))) conn.execute(text(oracle_trigger_sql(table_name))) def reset_sequence(driver: str, conn, table): """迁移后重置自增序列。MySQL / SQLite 显式插 id 后 auto_increment/rowid 自动 跟随 MAX(id),无需处理;PostgreSQL / SQL Server / Oracle 需要。""" if driver == 'postgresql': conn.execute(text(pg_reset_sql(table.name))) elif driver == 'mssql': max_id = conn.execute(text(f"SELECT MAX(id) FROM [{table.name}]")).scalar() conn.execute(text(mssql_reset_sql(table.name, max_id))) elif driver == 'oracle': ensure_oracle_autoincrement(conn, table.name) # ============================================================ # 校验 # ============================================================ def verify_counts(src, dst, tables) -> list[tuple[str, int, int]]: """逐表 count 对比,返回不一致的表列表((表名, 源行数, 目标行数))。""" failed = [] for t in tables: s = src.execute(text(f"SELECT COUNT(*) FROM {t.name}")).scalar() d = dst.execute(text(f"SELECT COUNT(*) FROM {t.name}")).scalar() mark = "PASS" if s == d else "FAIL" print(f" [{t.name}] source={s} target={d} {mark}") if s != d: failed.append((t.name, s, d)) return failed def _existing_equal(src, dst, table) -> bool: s = src.execute(text(f"SELECT COUNT(*) FROM {table.name}")).scalar() d = dst.execute(text(f"SELECT COUNT(*) FROM {table.name}")).scalar() return s == d # ============================================================ # 主流程 # ============================================================ def _dry_run(src_engine, tables, dst_driver: str, dst_db: str, t_start: float) -> int: print("=" * 62) print(f"[DRY-RUN 开始] 计划迁移 {len(tables)} 张表 → {dst_driver}:{dst_db}") print("=" * 62) with src_engine.connect() as src: for t in tables: n = src.execute(text(f"SELECT COUNT(*) FROM {t.name}")).scalar() print(f" - {t.name}: {n} 行") print(f"[DRY-RUN 完成] {len(tables)} 张表核对完毕(耗时 {_fmt_duration(time.perf_counter() - t_start)}),未触碰目标库") return 0 def run(args) -> int: """执行一次迁移。返回进程退出码(0 成功 / 1 校验失败)。""" src_driver = _normalize_driver(args.src_driver) dst_driver = _normalize_driver(args.dst_driver) src_engine = make_engine( src_driver, args.src_host, _resolve_port(src_driver, args.src_port), args.src_db, args.src_user, args.src_pass or os.environ.get("SRC_DB_PASS", ""), ) dst_engine = make_engine( dst_driver, args.dst_host, _resolve_port(dst_driver, args.dst_port), args.dst_db, args.dst_user, args.dst_pass or os.environ.get("DST_DB_PASS", ""), ) tables = ordered_tables() if args.only_tables: wanted = {x.strip() for x in args.only_tables.split(",") if x.strip()} tables = [t for t in tables if t.name in wanted or t.name[len(DB_PREFIX):] in wanted] try: # 源端缺表容错(老库可能缺整表) src_insp = inspect(src_engine) present, missing = [], [] for t in tables: if src_insp.has_table(t.name): present.append(t) else: missing.append(t) for t in missing: print(f"[SKIP] 源库不存在表 {t.name},跳过") t_start = time.perf_counter() if args.dry_run: return _dry_run(src_engine, present, dst_driver, args.dst_db, t_start) print("=" * 62) print(f"[迁移开始] 源: {src_driver}:{args.src_db} → 目标: {dst_driver}:{args.dst_db}") note = "" if not missing else f",源端缺失 {len(missing)} 张已跳过" print(f"[迁移开始] 待迁移 {len(present)} 张表{note}") print("=" * 62) # 1) 建目标库(删库/建库可能耗时较长,step() 会给出缓冲提示) with step("[1/5] 建目标库"): create_database_if_missing( dst_driver, args.dst_host, _resolve_port(dst_driver, args.dst_port), args.dst_db, args.dst_user, args.dst_pass or os.environ.get("DST_DB_PASS", ""), drop_first=args.drop_target_first, tables=tuple(t.name for t in present), # Oracle drop_first 时清表用 ) # 2) 建结构(模型即 head,方言可移植) with step(f"[2/5] 建表结构({len(present)} 张表 create_all)") as h: db.metadata.create_all(dst_engine) h['detail'] = f"{len(present)} 张表" # 3) 泵数据 total_rows = 0 with step("[3/5] 逐表迁移数据") as h: if args.single_transaction: with dst_engine.begin() as dst, src_engine.connect() as src: for idx, t in enumerate(present, 1): if args.skip_existing and _existing_equal(src, dst, t): print(f" [skip] {t.name}(源=目标)") continue _busy(f"[{idx}/{len(present)}] 正在迁移 {t.name}") total_rows += pump_table(src, dst, t, args.chunk_size, dst_driver) else: with src_engine.connect() as src: for idx, t in enumerate(present, 1): with dst_engine.begin() as dst: if args.skip_existing and _existing_equal(src, dst, t): print(f" [skip] {t.name}(源=目标)") continue _busy(f"[{idx}/{len(present)}] 正在迁移 {t.name}") total_rows += pump_table(src, dst, t, args.chunk_size, dst_driver) h['detail'] = f"共 {total_rows} 行" # 4) 重置自增序列(PG/MSSQL/Oracle;MySQL/SQLite 自动跟随) # 单表失败不中断其余表,避免一张空表序列异常连累后续全部序列 reset_errors = [] if dst_driver in ("postgresql", "mssql", "oracle"): with step("[4/5] 重置自增序列") as h: for t in present: try: # 每表独立事务:Oracle 的 CREATE/DROP SEQUENCE / TRIGGER 是 DDL 会隐式提交, # 不能与其它表共享一个大事务 with dst_engine.begin() as dst: reset_sequence(dst_driver, dst, t) except Exception as e: reset_errors.append((t.name, str(e))) print(f" [WARN] 序列重置失败 {t.name}: {e}") if reset_errors: h['detail'] = f"{len(reset_errors)} 张失败" else: print("[4/5 重置序列] MySQL/SQLite 自动跟随,无需处理") # 5) 校验 if not args.skip_verify: with step("[5/5] 逐表校验行数") as h: with src_engine.connect() as src, dst_engine.connect() as dst: failed = verify_counts(src, dst, present) h['detail'] = f"{len(failed)} 张表不一致" if failed else f"{len(present)} 张表全部 PASS" if failed: for name, s, d in failed: print(f" - {name}: source={s} target={d}") return 1 else: print("[5/5 校验] 已跳过(--skip-verify)") # 汇总 print("=" * 62) print(f"[迁移完成] 成功:{total_rows} 行 / {len(present)} 张表 / 总耗时 {_fmt_duration(time.perf_counter() - t_start)}") print("=" * 62) return 0 finally: src_engine.dispose() dst_engine.dispose() def main(): args = build_parser().parse_args() try: sys.exit(run(args)) except KeyboardInterrupt: sys.exit(130) if __name__ == "__main__": main()