pipeline-app/deploy/dbackup.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""dbackup.py — 增量表备份与恢复命令生成（四步流程第 1 步/第 4 步配套）。

用法（在应用根目录执行）：
  ./py3/bin/python deploy/dbackup.py backup --tables sd_projects,pipeline_tasks
  ./py3/bin/python deploy/dbackup.py backup --batch          # 备份待执行迁移批次声明的所有表
  ./py3/bin/python deploy/dbackup.py verify <备份文件>        # 校验备份文件完整性
  ./py3/bin/python deploy/dbackup.py restore --file <备份文件> --confirm   # 生成并执行恢复（危险）

备份位置：<APP_ROOT>/backups/<库名>_<表名>_<YYYYmmdd_HHMMSS>.sql
恢复 = DROP TABLE + 导入，执行前打印 [NEEDS_APPROVAL]，必须 --confirm 才真执行。
"""

import json
import os
import subprocess
import sys
from datetime import datetime

APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
BACKUP_DIR = os.path.join(APP_ROOT, "backups")
MIG_DIR = os.path.join(APP_ROOT, "deploy", "migrations")


def get_db_conf():
    sys.path.insert(0, APP_ROOT)
    try:
        from appPublic.jsonConfig import getConfig
        from appPublic.aes import aes_decode_b64
        cfg = getConfig(APP_ROOT, {"workdir": APP_ROOT})
        kw = cfg.databases["pipeline"].kwargs
        pwd = aes_decode_b64(cfg.password_key, kw.password)
        return str(kw.host), str(kw.port), str(kw.user), pwd, str(kw.db)
    except Exception as e:
        print("[FATAL] 读配置失败:", e)
        sys.exit(1)


def table_exists(conf, name):
    host, port, user, pwd, db = conf
    r = subprocess.run(["mysql", "-h", host, "-P", port, "-u", user, "-p" + pwd,
                        db, "-N", "-e", "SHOW TABLES LIKE '%s'" % name],
                       capture_output=True, text=True)
    return bool((r.stdout or "").strip())


def batch_tables():
    """收集所有待执行迁移批次声明的 backup_tables（去重）。"""
    # 读台账判断哪些批次已执行（简化：直接扫所有迁移文件的 backup_tables 并集）
    tables = []
    if os.path.isdir(MIG_DIR):
        for f in sorted(os.listdir(MIG_DIR)):
            if not f.endswith(".json"):
                continue
            try:
                d = json.load(open(os.path.join(MIG_DIR, f)))
                for t in d.get("backup_tables") or []:
                    if t not in tables:
                        tables.append(t)
            except Exception:
                pass
    return tables


def backup(conf, tables):
    host, port, user, pwd, db = conf
    os.makedirs(BACKUP_DIR, exist_ok=True)
    ts = datetime.now().strftime("%Y%m%d_%H%M%S")
    done = []
    for t in tables:
        if not table_exists(conf, t):
            print("  跳过(表不存在): %s" % t)
            continue
        out_file = os.path.join(BACKUP_DIR, "%s_%s_%s.sql" % (db, t, ts))
        r = subprocess.run(
            ["mysqldump", "-h", host, "-P", port, "-u", user, "-p" + pwd,
             db, t, "--single-transaction", "--skip-lock-tables",
             "--result-file", out_file],
            capture_output=True, text=True)
        err = (r.stderr or "").replace(
            "mysqldump: [Warning] Using a password on the command line interface can be insecure.\n", "")
        if r.returncode != 0 or not os.path.isfile(out_file) or os.path.getsize(out_file) == 0:
            print("[FAIL] %s: %s" % (t, err[:200]))
            sys.exit(1)
        size = os.path.getsize(out_file)
        print("  OK %s (%d bytes)" % (out_file, size))
        done.append(out_file)
    # 生成对应的恢复命令文件
    if done:
        restore_file = os.path.join(BACKUP_DIR, "RESTORE_%s.sh" % ts)
        with open(restore_file, "w") as f:
            f.write("#!/bin/bash\n# 恢复命令（危险！先人工核对）\n")
            for bf in done:
                f.write('mysql -h%s -P%s -u%s -p"$PASS" %s < %s\n' % (host, port, user, db, bf))
        os.chmod(restore_file, 0o700)
        print("  恢复命令: %s" % restore_file)
    return done


def verify(conf, path):
    if not os.path.isfile(path):
        print("[FAIL] 文件不存在: %s" % path)
        sys.exit(1)
    size = os.path.getsize(path)
    head = open(path, encoding="utf-8", errors="replace").read(4000)
    has_create = "CREATE TABLE" in head
    # 尾部完整性标记
    tail = subprocess.run(["tail", "-c", "200", path], capture_output=True, text=True).stdout
    has_dump_info = "Dump completed" in tail
    print("文件: %s (%d bytes)" % (path, size))
    print("  含 CREATE TABLE: %s" % ("是" if has_create else "否[异常]"))
    print("  含 Dump completed: %s" % ("是" if has_dump_info else "否[可能被截断]"))
    if not (has_create and has_dump_info and size > 0):
        sys.exit(1)
    print("  ✅ 备份文件完整")


def restore(conf, path):
    """恢复 = 从备份文件重建表。打印命令，--confirm 才执行。"""
    host, port, user, pwd, db = conf
    if not os.path.isfile(path):
        print("[FAIL] 文件不存在: %s" % path)
        sys.exit(1)
    print("[NEEDS_APPROVAL] 恢复操作将 DROP 并重建备份中的表！")
    print("  备份文件: %s" % path)
    print("  目标库: %s@%s" % (db, host))
    if "--confirm" not in sys.argv:
        print("  （预演模式，未执行。确认后追加 --confirm）")
        print("  手工命令: mysql -h%s -P%s -u%s -p<密码> %s < %s" % (host, port, user, db, path))
        return
    r = subprocess.run(["mysql", "-h", host, "-P", port, "-u", user, "-p" + pwd, db],
                       stdin=open(path), capture_output=True, text=True)
    err = (r.stderr or "").replace(
        "mysql: [Warning] Using a password on the command line interface can be insecure.\n", "")
    if r.returncode != 0:
        print("[FAIL] 恢复失败: %s" % err[:300])
        sys.exit(1)
    print("  ✅ 恢复完成")


def main():
    if len(sys.argv) < 2 or sys.argv[1] not in ("backup", "verify", "restore"):
        print(__doc__)
        sys.exit(1)
    conf = get_db_conf()
    cmd = sys.argv[1]
    if cmd == "backup":
        tables = []
        if "--tables" in sys.argv:
            tables = [t.strip() for t in sys.argv[sys.argv.index("--tables") + 1].split(",") if t.strip()]
        elif "--batch" in sys.argv:
            tables = batch_tables()
        if not tables:
            print("无表可备份（--tables a,b,c 或 --batch）")
            sys.exit(1)
        print("备份 %d 张表: %s" % (len(tables), ",".join(tables)))
        backup(conf, tables)
    elif cmd == "verify":
        if len(sys.argv) < 3:
            print("verify 需要备份文件路径")
            sys.exit(1)
        verify(conf, sys.argv[2])
    elif cmd == "restore":
        if "--file" not in sys.argv:
            print("restore 需要 --file <备份文件>")
            sys.exit(1)
        restore(conf, sys.argv[sys.argv.index("--file") + 1])


if __name__ == "__main__":
    main()