pipeline-app/deploy/ddiff.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""ddiff.py — 测试/生产环境差异比对（增量部署的"对账单"）。

用法：
  # 在生产机（或测试机）执行，拉取本环境快照后与对照快照比较：
  ./py3/bin/python deploy/ddiff.py snapshot > /tmp/schema_here.json    # 导出本环境快照
  ./py3/bin/python deploy/ddiff.py compare --against /tmp/schema_test.json
       # 打印差异清单：缺表/缺列/缺索引/参数差异/模块git版本差异

  # 一步到位：先在对照机跑 snapshot 存文件，scp 过来，再 compare。
  # 每次生产上线前：先 snapshot 测试环境 → compare 生产，差异必须全部有对应迁移或豁免。

设计原则：
  - 只读：本工具只查 information_schema / params / git log，绝不写库。
  - 输出面向人工核对：按 类型 分组，标注 [需迁移] / [可忽略]。
"""

import json
import os
import subprocess
import sys

APP_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
PKGS_DIR = os.path.join(APP_ROOT, "pkgs")


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 mysql(conf, sql):
    host, port, user, pwd, db = conf
    r = subprocess.run(["mysql", "-h", host, "-P", port, "-u", user, "-p" + pwd,
                        db, "-N", "-e", sql], capture_output=True, text=True)
    return r.stdout if r.returncode == 0 else ""


def snapshot(conf):
    _, _, _, _, db = conf
    s = {}
    s["tables"] = sorted(mysql(conf, "SHOW TABLES").split())
    s["cols"] = sorted(x for x in mysql(conf,
        "SELECT CONCAT(table_name,'|',column_name,'|',column_type) FROM information_schema.columns "
        "WHERE table_schema='%s'" % db).split("\n") if x)
    s["idx"] = sorted(x for x in mysql(conf,
        "SELECT CONCAT(table_name,'|',index_name,'|',GROUP_CONCAT(column_name ORDER BY seq_in_index)) "
        "FROM information_schema.statistics WHERE table_schema='%s' "
        "GROUP BY table_name,index_name" % db).split("\n") if x)
    s["params"] = sorted(x for x in mysql(conf,
        "SELECT CONCAT(params_name,'=',LEFT(params_value,60)) FROM params").split("\n") if x)
    # 模块版本（pkgs 下各仓库的 HEAD）
    mods = {}
    if os.path.isdir(PKGS_DIR):
        for d in sorted(os.listdir(PKGS_DIR)):
            gitdir = os.path.join(PKGS_DIR, d, ".git")
            if os.path.isdir(gitdir):
                r = subprocess.run(["git", "-C", os.path.join(PKGS_DIR, d),
                                    "log", "--oneline", "-1"], capture_output=True, text=True)
                mods[d] = r.stdout.strip()
    s["modules"] = mods
    return s


def compare(base, target):
    """base = 对照环境(测试), target = 本环境(生产)。打印 base 有 / target 无 的缺口。"""
    print("═══ 差异清单（对照环境有 / 本环境缺）═══\n")

    bt, tt = set(base["tables"]), set(target["tables"])
    missing_tables = sorted(bt - tt)
    extra_tables = sorted(tt - bt)
    print("── 缺表 [%d] %s" % (len(missing_tables), "【需迁移】" if missing_tables else ""))
    for t in missing_tables:
        print("   -", t)
    if extra_tables:
        print("   （本环境多出的表，通常可忽略）:", ", ".join(extra_tables[:10]))

    bc, tc = set(base["cols"]), set(target["cols"])
    from collections import defaultdict
    # 列类型不同（同名不同型）先算出来——这些不算"缺列"，避免重复报告
    base_colmap = {}
    for c in base["cols"]:
        p = c.split("|")
        if len(p) == 3:
            base_colmap[(p[0], p[1])] = p[2]
    target_colmap = {}
    for c in target["cols"]:
        p = c.split("|")
        if len(p) == 3:
            target_colmap[(p[0], p[1])] = p[2]
    type_diff = []
    for (t, col), bt2 in base_colmap.items():
        if (t, col) in target_colmap and target_colmap[(t, col)] != bt2:
            type_diff.append("%s.%s: 对照=%s 本环境=%s" % (t, col, bt2, target_colmap[(t, col)]))
    type_diff_keys = set((d.split(".")[0], d.split(".")[1].split(":")[0]) for d in type_diff)

    miss_cols = defaultdict(list)
    for c in bc - tc:
        p = c.split("|")
        if len(p) == 3 and (p[0], p[1]) not in type_diff_keys:
            miss_cols[p[0]].append(p[1] + " " + p[2])
    print("\n── 缺列 [%d 表] %s" % (len(miss_cols), "【需迁移】" if miss_cols else ""))
    for t in sorted(miss_cols):
        print("   %s: %s" % (t, ", ".join(sorted(miss_cols[t]))))
    print("\n── 列类型不同 [%d] %s" % (len(type_diff), "【需迁移】" if type_diff else ""))
    for d in type_diff:
        print("   ", d)

    bi, ti = set(base["idx"]), set(target["idx"])
    miss_idx = defaultdict(list)
    for i in bi - ti:
        p = i.split("|")
        if len(p) == 3:
            miss_idx[p[0]].append(p[1] + "(" + p[2] + ")")
    # 名字不同但列相同的索引视为等价（如 uk_project_role vs uk_sd_project_role_models_pr）
    target_idxcols = {}
    for i in target["idx"]:
        p = i.split("|")
        if len(p) == 3:
            target_idxcols.setdefault(p[0], set()).add(p[2])
    real_miss = {}
    for t, lst in miss_idx.items():
        real = [x for x in lst if x.split("(")[1].rstrip(")") not in target_idxcols.get(t, set())]
        if real:
            real_miss[t] = real
    print("\n── 缺索引 [%d 表]（已排除列等价的改名索引）%s"
          % (len(real_miss), "【建议迁移】" if real_miss else ""))
    for t in sorted(real_miss):
        print("   %s: %s" % (t, ", ".join(sorted(real_miss[t])[:8])))

    bp, tp = set(base["params"]), set(target["params"])
    miss_params = sorted(bp - tp)
    print("\n── 缺参数/参数值不同 [%d] %s" % (len(miss_params), "【需迁移】" if miss_params else ""))
    for p in miss_params:
        print("   -", p[:80])

    bm, tm = base.get("modules", {}), target.get("modules", {})
    mod_diff = []
    for k in sorted(set(bm) | set(tm)):
        if bm.get(k) != tm.get(k):
            mod_diff.append("   %s:\n      对照: %s\n      本环境: %s" % (k, bm.get(k, "(无)"), tm.get(k, "(无)")))
    print("\n── 模块版本不同 [%d] %s" % (len(mod_diff), "【需 pull+install】" if mod_diff else ""))
    for d in mod_diff:
        print(d)

    total = len(missing_tables) + len(miss_cols) + len(type_diff) + len(miss_params) + len(mod_diff)
    print("\n═══ 汇总：需处理 %d 项（缺表 %d / 缺列 %d / 类型不同 %d / 参数 %d / 模块 %d），"
          "缺索引 %d 建议项 ═══"
          % (total, len(missing_tables), len(miss_cols), len(type_diff),
             len(miss_params), len(mod_diff), len(real_miss)))
    return total


def main():
    if len(sys.argv) < 2 or sys.argv[1] not in ("snapshot", "compare"):
        print(__doc__)
        sys.exit(1)
    conf = get_db_conf()
    if sys.argv[1] == "snapshot":
        print(json.dumps(snapshot(conf), ensure_ascii=False, indent=1))
        return
    # compare
    if "--against" not in sys.argv:
        print("compare 需要 --against <对照快照.json>")
        sys.exit(1)
    path = sys.argv[sys.argv.index("--against") + 1]
    base = json.load(open(path))
    target = snapshot(conf)
    total = compare(base, target)
    sys.exit(0 if total == 0 else 2)


if __name__ == "__main__":
    main()