yumoqing
dd40df6337
- METHODOLOGY.md: 备份/操作/验证/回退四大步操作规范 - dmig.py: 迁移引擎(台账表pipeline_deploy_ledger+幂等+破坏性守卫+down回退) - ddiff.py: 环境schema/参数/模块版本差异对账 - dbackup.py: mysqldump增量表备份+恢复命令生成 - migrations/m0001-m0004: 首批迁移(今日发现的生产库缺口)
181 lines
7.3 KiB
Python
Executable File
181 lines
7.3 KiB
Python
Executable File
#!/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
|
||
miss_cols = defaultdict(list)
|
||
for c in bc - tc:
|
||
p = c.split("|")
|
||
if len(p) == 3:
|
||
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]))))
|
||
# 列宽/类型不同（同名不同型）
|
||
base_colmap = {}
|
||
for c in base["cols"]:
|
||
p = c.split("|")
|
||
if len(p) == 3:
|
||
base_colmap[(p[0], p[1])] = p[2]
|
||
type_diff = []
|
||
for c in target["cols"]:
|
||
p = c.split("|")
|
||
if len(p) == 3 and (p[0], p[1]) in base_colmap:
|
||
bt2 = base_colmap[(p[0], p[1])]
|
||
if bt2 != p[2]:
|
||
type_diff.append("%s.%s: 对照=%s 本环境=%s" % (p[0], p[1], bt2, p[2]))
|
||
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()
|