#!/usr/bin/env python3 """Copy the legacy PostgreSQL-schema Data Center into real MySQL databases. The command is dry-run by default. It never changes PostgreSQL source data and will not overwrite an existing MySQL target unless both ``--execute`` and ``--replace-target`` are supplied. """ from __future__ import annotations import argparse import asyncio from datetime import datetime, timezone import json from pathlib import Path import sys from typing import Any import uuid from psycopg import sql REPOSITORY_ROOT = Path(__file__).resolve().parents[1] if str(REPOSITORY_ROOT) not in sys.path: sys.path.insert(0, str(REPOSITORY_ROOT)) from app import db from app.config import settings from app.data_platform.mysql_db import close_data_pool, get_data_conn, init_data_pool from app.data_platform.mysql_service import ( _create_physical_table, delete_project_database, ensure_platform_registry, ensure_project_database, get_project_database, ) from app.data_platform.registry import TABLE_DEFINITIONS from app.data_platform.schema import ( _custom_table_from_row, _field_definitions_from_rows, _field_row, _table_with_identity, ) SYSTEM_COLUMNS = ( "id", "tenant_id", "project_id", "created_at", "updated_at", "deleted_at", "deleted_by", ) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--project-id", action="append", help="仅迁移指定数据库编码，可重复") parser.add_argument("--execute", action="store_true", help="实际执行；未提供时只输出迁移计划") parser.add_argument( "--replace-target", action="store_true", help="目标 MySQL 数据库已存在时先删除后重建（必须同时提供 --execute）", ) parser.add_argument("--batch-size", type=int, default=1000) return parser.parse_args() def mysql_value(value: Any, *, json_field: bool = False) -> Any: if value is None: return None if isinstance(value, uuid.UUID): return str(value) if isinstance(value, datetime) and value.tzinfo is not None: return value.astimezone(timezone.utc).replace(tzinfo=None) if json_field and not isinstance(value, str): return json.dumps(value, ensure_ascii=False, separators=(",", ":")) return value async def legacy_databases(project_ids: set[str] | None) -> list[dict[str, Any]]: async with db.get_conn() as conn: async with conn.cursor() as cur: query = sql.SQL( "SELECT * FROM {}.project_databases WHERE status='ready'" ).format(sql.Identifier(settings.db_schema)) params: tuple[Any, ...] = () if project_ids: query += sql.SQL(" AND project_id = ANY(%s)") params = (list(project_ids),) query += sql.SQL(" ORDER BY created_at, project_id") await cur.execute(query, params) return [dict(row) for row in await cur.fetchall()] async def legacy_table_entries( project_id: str, ) -> tuple[tuple[Any, str, str], ...]: """Read legacy metadata directly without running PostgreSQL DDL.""" async with db.get_conn() as conn: async with conn.cursor() as cur: await cur.execute( sql.SQL( "SELECT * FROM {}.project_table_overrides WHERE project_id=%s" ).format(sql.Identifier(settings.db_schema)), (project_id,), ) overrides = { str(row["source_code"]): dict(row) for row in await cur.fetchall() } await cur.execute( sql.SQL( """ SELECT * FROM {}.project_table_definitions WHERE project_id=%s AND status='active' ORDER BY created_at, table_code """ ).format(sql.Identifier(settings.db_schema)), (project_id,), ) custom_rows = [dict(row) for row in await cur.fetchall()] entries: list[tuple[Any, str, str]] = [] for definition in TABLE_DEFINITIONS: override = overrides.get(definition.code) if override and str(override["status"]) == "deleted": continue effective = ( _table_with_identity( definition, code=str(override["table_code"]), label=str(override["label"]), fields=( _field_definitions_from_rows(override["fields_jsonb"]) if override.get("fields_jsonb") is not None else definition.fields ), ) if override else definition ) entries.append((effective, "builtin", definition.code)) entries.extend( (_custom_table_from_row(row), "custom", str(row["table_code"])) for row in custom_rows ) return tuple(entries) async def target_has_database(project_id: str) -> bool: return await get_project_database(project_id) is not None async def rebuild_target_structure( database: dict[str, Any], entries: tuple[tuple[Any, str, str], ...], ) -> None: project_id = str(database["project_id"]) database_name = str(database["database_name"]) async with get_data_conn(database_name) as conn: async with conn.cursor() as cur: await cur.execute("SET FOREIGN_KEY_CHECKS=0") await cur.execute("SHOW TABLES") for row in await cur.fetchall(): table_name = str(next(iter(row.values()))) if not table_name.replace("_", "a").isalnum(): raise RuntimeError(f"目标表名不安全：{table_name}") await cur.execute(f"DROP TABLE `{table_name}`") await cur.execute("SET FOREIGN_KEY_CHECKS=1") await conn.commit() async with get_data_conn() as conn: async with conn.cursor() as cur: await cur.execute( "DELETE FROM project_table_definitions WHERE project_id=%s", (project_id,), ) await cur.execute( "DELETE FROM data_change_logs WHERE project_id=%s", (project_id,), ) for order, (definition, origin, source_code) in enumerate(entries): await cur.execute( """ INSERT INTO project_table_definitions ( project_id, source_code, table_code, label, group_name, description, fields_json, origin, display_order, allow_create, allow_update, allow_delete, created_by ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, 'migration') """, ( project_id, source_code, definition.code, definition.label, definition.group, definition.description, json.dumps([_field_row(field) for field in definition.fields], ensure_ascii=False), origin, order, int(definition.allow_create), int(definition.allow_update), int(definition.allow_delete), ), ) await conn.commit() for definition, _origin, _source_code in entries: await _create_physical_table(database_name, definition) async def copy_table_rows( source_schema: str, target_database: str, definition: Any, batch_size: int, ) -> int: business_columns = [field.code for field in definition.fields] columns = [ "id", "tenant_id", "project_id", *business_columns, "created_at", "updated_at", "deleted_at", "deleted_by", ] json_fields = {field.code for field in definition.fields if field.data_type == "json"} placeholders = ", ".join(["%s"] * len(columns)) target_query = ( f"INSERT INTO `{definition.code}` " f"({', '.join(f'`{column}`' for column in columns)}) VALUES ({placeholders})" ) copied = 0 async with db.get_conn() as source_conn: async with source_conn.cursor() as source_cur: await source_cur.execute( sql.SQL("SELECT {} FROM {}.{} ORDER BY created_at, id").format( sql.SQL(", ").join(sql.Identifier(column) for column in columns), sql.Identifier(source_schema), sql.Identifier(definition.code), ) ) async with get_data_conn(target_database) as target_conn: async with target_conn.cursor() as target_cur: while True: rows = await source_cur.fetchmany(batch_size) if not rows: break values = [ tuple( mysql_value(row[column], json_field=column in json_fields) for column in columns ) for row in rows ] await target_cur.executemany(target_query, values) copied += len(values) await target_conn.commit() return copied async def copy_audit_logs(source_schema: str, project_id: str, batch_size: int) -> int: columns = ( "tenant_id", "project_id", "table_code", "record_id", "operation", "before_data", "after_data", "actor", "created_at", ) copied = 0 async with db.get_conn() as source_conn: async with source_conn.cursor() as source_cur: await source_cur.execute( sql.SQL("SELECT {} FROM {}.data_change_logs ORDER BY created_at, id").format( sql.SQL(", ").join(sql.Identifier(column) for column in columns), sql.Identifier(source_schema), ) ) async with get_data_conn() as target_conn: async with target_conn.cursor() as target_cur: while True: rows = await source_cur.fetchmany(batch_size) if not rows: break values = [ ( row["tenant_id"], project_id, row["table_code"], str(row["record_id"]), row["operation"], mysql_value(row["before_data"], json_field=True), mysql_value(row["after_data"], json_field=True), row["actor"], mysql_value(row["created_at"]), ) for row in rows ] await target_cur.executemany( """ INSERT INTO data_change_logs ( tenant_id, project_id, table_code, record_id, operation, before_data, after_data, actor, created_at ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) """, values, ) copied += len(values) await target_conn.commit() return copied async def migrate(args: argparse.Namespace) -> None: await db.init_pool() if not await init_data_pool(): raise RuntimeError("无法连接 DATA_MYSQL_URL 指向的 MySQL 服务") await ensure_platform_registry() try: requested = set(args.project_id or []) or None databases = await legacy_databases(requested) if not databases: print("没有找到待迁移的 PostgreSQL 数据中心数据库。") return print(f"发现 {len(databases)} 个数据库；模式：{'执行' if args.execute else '仅预览'}") for source in databases: project_id = str(source["project_id"]) entries = await legacy_table_entries(project_id) print( f"- {source['display_name']} ({project_id})：{len(entries)} 张表，" f"{source['schema_name']} → {project_id}" ) if not args.execute: continue exists = await target_has_database(project_id) if exists and not args.replace_target: raise RuntimeError( f"目标 {project_id} 已存在；确认可覆盖后追加 --replace-target" ) if exists: target = await get_project_database(project_id) await delete_project_database(project_id, str(target["database_name"])) target = await ensure_project_database( project_id, str(source["tenant_id"]), str(source["display_name"]), ) await rebuild_target_structure(target, entries) total = 0 for definition, _origin, _source_code in entries: count = await copy_table_rows( str(source["schema_name"]), str(target["database_name"]), definition, max(100, min(10_000, args.batch_size)), ) total += count print(f" · {definition.code}: {count:,} 条") audit_count = await copy_audit_logs( str(source["schema_name"]), project_id, max(100, min(10_000, args.batch_size)), ) print(f" 完成：业务记录 {total:,} 条，审计记录 {audit_count:,} 条") finally: await close_data_pool() await db.close_pool() if __name__ == "__main__": asyncio.run(migrate(parse_args()))