157 lines
7.1 KiB
Python
157 lines
7.1 KiB
Python
|
"""Export a read-only MySQL snapshot and a row-count/hash manifest.
|
|||
|
|
|
|||
|
|
Pause application writes before exporting. Connection settings come from
|
|||
|
|
database.engine.url (DB_* or DATABASE_URL in the process environment).
|
|||
|
|
This script never initializes, resets, or deletes database tables.
|
|||
|
|
"""
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import argparse
|
|||
|
|
import hashlib
|
|||
|
|
import json
|
|||
|
|
import os
|
|||
|
|
from pathlib import Path
|
|||
|
|
import shutil
|
|||
|
|
import subprocess
|
|||
|
|
import sys
|
|||
|
|
import tempfile
|
|||
|
|
from datetime import datetime, timezone
|
|||
|
|
|
|||
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|||
|
|
MYSQLDUMP_FALLBACK = Path(r"C:\tools\mysql\mysql-8.0.27-winx64\bin\mysqldump.exe")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class MigrationError(Exception):
|
|||
|
|
pass
|
|||
|
|
|
|||
|
|
|
|||
|
|
def option_value(value: object) -> str:
|
|||
|
|
"""Quote MySQL option-file values without exposing them in argv."""
|
|||
|
|
value = str(value)
|
|||
|
|
if "\x00" in value:
|
|||
|
|
raise MigrationError("数据库连接参数不能包含 NUL 字符。")
|
|||
|
|
return '"' + (value.replace("\\", "\\\\").replace('"', '\\"')
|
|||
|
|
.replace("\n", "\\n").replace("\r", "\\r")
|
|||
|
|
.replace("\t", "\\t")) + '"'
|
|||
|
|
|
|||
|
|
|
|||
|
|
def source_counts(engine) -> dict[str, int]:
|
|||
|
|
from sqlalchemy import inspect, text
|
|||
|
|
|
|||
|
|
database = engine.url.database
|
|||
|
|
with engine.connect() as connection:
|
|||
|
|
inspector = inspect(connection)
|
|||
|
|
names = sorted(inspector.get_table_names(schema=database)
|
|||
|
|
+ inspector.get_view_names(schema=database))
|
|||
|
|
storage = connection.execute(text(
|
|||
|
|
"SELECT TABLE_NAME, ENGINE FROM information_schema.TABLES "
|
|||
|
|
"WHERE TABLE_SCHEMA = :database AND TABLE_TYPE = 'BASE TABLE'"
|
|||
|
|
), {"database": database}).all()
|
|||
|
|
if any(str(row[1]).upper() != "INNODB" for row in storage):
|
|||
|
|
raise MigrationError("源库存在非 InnoDB 表，不能保证事务一致性；请先检查存储引擎。")
|
|||
|
|
quote = engine.dialect.identifier_preparer.quote_identifier
|
|||
|
|
return {name: int(connection.execute(text(
|
|||
|
|
f"SELECT COUNT(*) FROM {quote(database)}.{quote(name)}"
|
|||
|
|
)).scalar_one()) for name in names}
|
|||
|
|
|
|||
|
|
|
|||
|
|
def export_database(args) -> tuple[Path, Path, dict[str, int]]:
|
|||
|
|
executable = args.mysqldump or shutil.which("mysqldump")
|
|||
|
|
if not executable and MYSQLDUMP_FALLBACK.is_file():
|
|||
|
|
executable = str(MYSQLDUMP_FALLBACK)
|
|||
|
|
if not executable or not Path(executable).is_file():
|
|||
|
|
raise MigrationError("找不到 mysqldump，请使用 --mysqldump 指定其完整路径。")
|
|||
|
|
|
|||
|
|
sys.path.insert(0, str(ROOT))
|
|||
|
|
from database import engine
|
|||
|
|
|
|||
|
|
url = engine.url
|
|||
|
|
if url.get_backend_name() != "mysql" or not url.database or not url.username:
|
|||
|
|
raise MigrationError("database.engine.url 必须是包含数据库名和用户名的 MySQL 连接。")
|
|||
|
|
stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S%fZ")
|
|||
|
|
output = (args.output or ROOT / "artifacts" / "migration" / f"wolin-{stamp}.sql").resolve()
|
|||
|
|
manifest_path = output.with_suffix(".manifest.json")
|
|||
|
|
if output.exists() or manifest_path.exists():
|
|||
|
|
raise MigrationError("导出文件或 manifest 已存在；请指定新的 --output，程序不会覆盖它们。")
|
|||
|
|
output.parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
|
|||
|
|
try:
|
|||
|
|
before = source_counts(engine)
|
|||
|
|
if not before:
|
|||
|
|
raise MigrationError("源数据库没有表，已取消导出以避免误迁移空库。")
|
|||
|
|
with tempfile.TemporaryDirectory(prefix="wolin-mysql-export-") as temp_dir:
|
|||
|
|
config_path = Path(temp_dir) / "client.cnf"
|
|||
|
|
config = "[client]\n" + "\n".join(
|
|||
|
|
f"{key}={option_value(value)}" for key, value in {
|
|||
|
|
"host": url.host or "localhost", "port": url.port or 3306,
|
|||
|
|
"user": url.username, "password": url.password or "",
|
|||
|
|
"protocol": "TCP", "default-character-set": "utf8mb4",
|
|||
|
|
}.items()
|
|||
|
|
) + "\n"
|
|||
|
|
descriptor = os.open(config_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|||
|
|
with os.fdopen(descriptor, "w", encoding="utf-8", newline="\n") as config_file:
|
|||
|
|
config_file.write(config)
|
|||
|
|
command = [str(executable), f"--defaults-file={config_path}",
|
|||
|
|
"--single-transaction", "--no-tablespaces", "--set-gtid-purged=OFF",
|
|||
|
|
"--column-statistics=0", "--skip-add-drop-table", "--skip-add-locks",
|
|||
|
|
"--hex-blob", "--default-character-set=utf8mb4", "--", url.database]
|
|||
|
|
# An incomplete SQL file deliberately has no manifest and cannot be restored.
|
|||
|
|
with output.open("xb") as sql_file:
|
|||
|
|
result = subprocess.run(command, stdout=sql_file, stderr=subprocess.PIPE,
|
|||
|
|
check=False)
|
|||
|
|
sql_file.flush()
|
|||
|
|
os.fsync(sql_file.fileno())
|
|||
|
|
if result.returncode:
|
|||
|
|
raise MigrationError(
|
|||
|
|
f"mysqldump 失败（退出码 {result.returncode}）；请检查连接和导出权限。"
|
|||
|
|
f"未生成 manifest，部分文件不能用于迁移：{output}"
|
|||
|
|
)
|
|||
|
|
after = source_counts(engine)
|
|||
|
|
if before != after:
|
|||
|
|
raise MigrationError(
|
|||
|
|
"导出期间表结构或行数发生变化。请暂停应用写入后重新导出；本次不生成 manifest。"
|
|||
|
|
)
|
|||
|
|
digest = hashlib.sha256()
|
|||
|
|
with output.open("rb") as sql_file:
|
|||
|
|
for block in iter(lambda: sql_file.read(1024 * 1024), b""):
|
|||
|
|
digest.update(block)
|
|||
|
|
manifest = {
|
|||
|
|
"format_version": 1,
|
|||
|
|
"created_at_utc": datetime.now(timezone.utc).isoformat(),
|
|||
|
|
"database": url.database,
|
|||
|
|
"sql_file": output.name,
|
|||
|
|
"sha256": digest.hexdigest(),
|
|||
|
|
"tables": before,
|
|||
|
|
"consistency": "single-transaction; row counts checked before and after dump",
|
|||
|
|
}
|
|||
|
|
with manifest_path.open("x", encoding="utf-8", newline="\n") as manifest_file:
|
|||
|
|
json.dump(manifest, manifest_file, ensure_ascii=False, indent=2)
|
|||
|
|
manifest_file.write("\n")
|
|||
|
|
return output, manifest_path, before
|
|||
|
|
finally:
|
|||
|
|
engine.dispose()
|
|||
|
|
|
|||
|
|
|
|||
|
|
def main() -> int:
|
|||
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|||
|
|
parser.add_argument("--output", type=Path, help="新的 SQL 文件路径；默认 artifacts/migration/wolin-时间戳.sql")
|
|||
|
|
parser.add_argument("--mysqldump", help="mysqldump 可执行文件的完整路径")
|
|||
|
|
args = parser.parse_args()
|
|||
|
|
try:
|
|||
|
|
output, manifest, counts = export_database(args)
|
|||
|
|
except MigrationError as error:
|
|||
|
|
print(f"导出失败：{error}", file=sys.stderr)
|
|||
|
|
return 1
|
|||
|
|
except Exception as error:
|
|||
|
|
# Driver/subprocess exception text may contain connection or record data.
|
|||
|
|
print(f"导出失败（{type(error).__name__}）：请检查数据库连接、权限及目标目录。"
|
|||
|
|
"未完成的 SQL 文件没有有效 manifest，不能用于导入。", file=sys.stderr)
|
|||
|
|
return 1
|
|||
|
|
print(f"SQL：{output}\nManifest：{manifest}")
|
|||
|
|
print(f"已导出 {len(counts)} 个表/视图，共 {sum(counts.values())} 行；未修改源数据库。")
|
|||
|
|
return 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
raise SystemExit(main())
|