3452078359-xuexue
3dd5731751
- move the relational data center to MySQL and a standalone workbench\n- add Interface Center API credentials, policies, logs, and DBeaver SSH guidance\n- harden authentication and deployment while retiring unused management surfaces
375 lines
14 KiB
Python
375 lines
14 KiB
Python
#!/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()))
|