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())
|