compare_dup_tables.py
6.16 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
六库重名表结构比对 —— P2 合库前的核心检查。
合库时"同名表直接搬"只在结构一致时成立。不一致就必须改名或建视图过渡，
否则数据会串。本脚本对每组重名表算一个**规范化结构签名**（列 + 类型 + 可空 +
默认值 + 键 + 排序规则 + 索引），签名相同的直接判定可合并，不同的展开差异明细。
只读：仅查 information_schema + SHOW CREATE TABLE（后者只在需要展示差异时调）。
"""
import hashlib
import os
import re
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from legacy_db import load_env, connect # noqa: E402
DBS = ['zt_sys', 'zt_park', 'zt_order', 'zt_order_n', 'zt_bill', 'zt_charge']
def col_sig(cur, db, table):
"""列定义规范化。忽略列注释/自增值这些不影响数据搬运的差异。"""
cur.execute(
"SELECT COLUMN_NAME, COLUMN_TYPE, IS_NULLABLE, IFNULL(COLUMN_DEFAULT,'~'), "
" COLUMN_KEY, EXTRA, IFNULL(COLLATION_NAME,'') "
"FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s "
"ORDER BY COLUMN_NAME", (db, table))
return [(r[0].lower(), r[1].lower(), r[2], str(r[3]), r[4],
re.sub(r'auto_increment', '', r[5]).strip(), r[6]) for r in cur.fetchall()]
def idx_sig(cur, db, table):
"""索引定义规范化（索引名忽略大小写；排除主键的自动命名差异不适用，主键名恒为 PRIMARY）。"""
cur.execute(
"SELECT INDEX_NAME, NON_UNIQUE, SEQ_IN_INDEX, COLUMN_NAME, IFNULL(SUB_PART,-1) "
"FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s "
"ORDER BY INDEX_NAME, SEQ_IN_INDEX", (db, table))
return [(r[0].lower(), r[1], r[2], (r[3] or '').lower(), r[4]) for r in cur.fetchall()]
def sig(cur, db, table):
payload = repr(col_sig(cur, db, table)) + '||' + repr(idx_sig(cur, db, table))
return hashlib.sha1(payload.encode('utf-8')).hexdigest()[:12], payload
COL_ATTRS = ['列名', '类型', '可空', '默认值', '键', 'EXTRA', '排序规则']
def col_diff(a_cols, b_cols):
"""
列级差异（a=左库，b=右库）。
⚠️ 必须展开到**属性级**：只打 `左 varchar(32) / 右 varchar(32)` 会让人以为"没差别"，
实际差异在默认值 / 排序规则 / EXTRA（on update CURRENT_TIMESTAMP）上——
这三类差异恰恰是合库时最容易漏、合完才发现数据不对的。
"""
da = {c[0]: c for c in a_cols}
db_ = {c[0]: c for c in b_cols}
out = []
for name in sorted(set(da) | set(db_)):
if name not in da:
out.append(' + %-30s 仅右边有: %s' % (name, db_[name][1]))
elif name not in db_:
out.append(' - %-30s 仅左边有: %s' % (name, da[name][1]))
elif da[name] != db_[name]:
det = []
for i, label in enumerate(COL_ATTRS):
if da[name][i] != db_[name][i]:
det.append('%s 左[%s]/右[%s]' % (label, da[name][i], db_[name][i]))
out.append(' ~ %-30s %s' % (name, '; '.join(det) if det else '未知差异'))
return out
def idx_diff(a_idx, b_idx):
"""索引差异：按索引名比对，缺索引 / 列顺序不同 / 唯一性不同。"""
da, db_ = {}, {}
for r in a_idx:
da.setdefault(r[0], []).append(r)
for r in b_idx:
db_.setdefault(r[0], []).append(r)
out = []
for n in sorted(set(da) | set(db_)):
if n not in da:
out.append(' + 索引 %s 仅右边有' % n)
elif n not in db_:
out.append(' - 索引 %s 仅左边有' % n)
elif da[n] != db_[n]:
out.append(' ~ 索引 %s 定义不同: 左%s / 右%s' % (
n,
[(x[3], x[1]) for x in da[n]],
[(x[3], x[1]) for x in db_[n]]))
return out
def main():
cfg = load_env(os.path.join(os.path.dirname(os.path.abspath(__file__)),
'..', '..', '私密', 'legacy-ro.env'))
conn = connect(cfg)
cur = conn.cursor()
inlist = ','.join("'" + d + "'" for d in DBS)
cur.execute("SELECT TABLE_SCHEMA, TABLE_NAME FROM information_schema.TABLES "
"WHERE TABLE_SCHEMA IN (" + inlist + ")")
where = {}
for s, t in cur.fetchall():
where.setdefault(t.lower(), {})[s] = t # 记原始大小写
dups = {n: v for n, v in where.items() if len(v) > 1}
print('跨库重名表 %d 组\n' % len(dups))
same, diff = [], []
for name in sorted(dups):
locs = dups[name]
sigs, cols, idxs = {}, {}, {}
for dbname, real in sorted(locs.items()):
s, _p = sig(cur, dbname, real)
sigs[dbname] = s
cols[dbname] = col_sig(cur, dbname, real)
idxs[dbname] = idx_sig(cur, dbname, real)
uniq = set(sigs.values())
if len(uniq) == 1:
same.append((name, sorted(locs)))
else:
diff.append((name, sorted(locs), sigs, cols, idxs))
print('=' * 78)
print('【A】结构一致（可直接合并）: %d 组' % len(same))
print('=' * 78)
for name, locs in same:
print(' %-40s %s' % (name, '、'.join(locs)))
print()
print('=' * 78)
print('【B】⚠️ 结构不一致（必须处理）: %d 组' % len(diff))
print('=' * 78)
for name, locs, sigs, cols, idxs in diff:
print('\n── %s %s' % (name, '、'.join('%s=%s' % (k, v) for k, v in sigs.items())))
base = locs[0]
for other in locs[1:]:
print(' [%s] vs [%s]' % (base, other))
d = col_diff(cols[base], cols[other])
if not d:
print(' 列完全一致 —— 差异只在索引层')
for line in d[:18]:
print(line)
if len(d) > 18:
print(' ...（另有 %d 处列差异）' % (len(d) - 18))
di = idx_diff(idxs[base], idxs[other])
for line in di[:8]:
print(line)
if len(di) > 8:
print(' ...（另有 %d 处索引差异）' % (len(di) - 8))
conn.close()
return 0
if __name__ == '__main__':
sys.exit(main())