#!/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()