extract_table_inventory.py
12.8 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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
从老仓库 zteits-clouds 的 mapper XML + 实体注解里静态抽取「表清单」。
为什么需要它：老库是**生产库**，P2 之前不能连；而 `doc/sql/*.sql` 只是 ALTER 变更日志，
不含完整建表语句。所以合并六库前，唯一可离线获得的全量表清单来源是**代码里真正用到的表名**
（XML 的 FROM/JOIN/INTO/UPDATE + tk.mybatis 实体的 @Table）。
产出：
table-inventory.csv 表名, 所属模块(可能有多个), 引用文件数, 来源(xml/entity), 是否跨模块
table-inventory.md 人读的汇总：冲突表、无前缀表、跨模块访问表
用法：
python3 extract_table_inventory.py <老仓库根目录> <输出目录>
"""
import csv
import os
import re
import sys
from collections import defaultdict
# SQL 关键字：这些词长得像表名但不能当表名
SQL_KEYWORDS = {
'select', 'from', 'where', 'join', 'inner', 'left', 'right', 'outer', 'full',
'on', 'as', 'and', 'or', 'set', 'values', 'into', 'update', 'delete', 'insert',
'group', 'order', 'by', 'having', 'limit', 'union', 'all', 'distinct', 'dual',
'case', 'when', 'then', 'else', 'end', 'if', 'exists', 'not', 'in', 'is', 'null',
'count', 'sum', 'max', 'min', 'avg', 'asc', 'desc', 'table', 'cross', 'using',
}
# 抓 from/join/into/update/delete from 后面的表名（允许 `db`.`table` 或 db.table）
# ⚠️ 字符类里**不能**含 `$`：老代码有 `FROM pay_order${tableName}` 这种按后缀分表的写法，
# 含 `$` 会把它当成表名 `pay_order$`，凭空多出两个不存在的表。
TABLE_REF = re.compile(
r'\b(?:from|join|into|update)\s+'
r'(?:`?([A-Za-z_]\w*)`?\s*\.\s*)?' # 可选 schema 前缀
r'`?([A-Za-z_]\w*)`?',
re.IGNORECASE,
)
TABLE_ANNO = re.compile(r'@Table\s*\(\s*name\s*=\s*"([^"]+)"')
# 动态表名：`FROM tp_p_eqp_inout_record${tableName}` / `from pay_order${year}`
# 这类表**静态抽不出真实名字**（后缀是调用方在运行时拼的，如 yyyyMM），必须单独列出来
DYNAMIC_TABLE = re.compile(
r'\b(?:from|join|into|update)\s+([A-Za-z_]\w*)\s*\$\{(\w+)\}',
re.IGNORECASE,
)
TABLENAME_ANNO = re.compile(r'@TableName\s*\(\s*(?:value\s*=\s*)?"([^"]+)"')
def module_of(rel_path):
"""取相对路径的第一段作为模块名，如 zteits-park/src/... -> zteits-park"""
parts = rel_path.replace(os.sep, '/').split('/')
return parts[0] if parts else '(root)'
def scan(root):
tables = defaultdict(lambda: {
'modules': set(), 'files': set(), 'kinds': set(), 'schemas': set(),
})
# (表名前缀, 变量名) -> {模块}
dynamic = {}
def record(table, module, rel_path, kind, schema=None):
table = table.lower()
if table in SQL_KEYWORDS or len(table) < 2:
return
if table.startswith('$') or table.startswith('{'):
return
entry = tables[table]
entry['modules'].add(module)
entry['files'].add(rel_path)
entry['kinds'].add(kind)
if schema:
entry['schemas'].add(schema.lower())
for dirpath, dirnames, filenames in os.walk(root):
# 跳过构建产物与前端依赖
dirnames[:] = [
d for d in dirnames
if d not in ('node_modules', 'target', '.git', 'dist', '.idea')
]
for fn in filenames:
full = os.path.join(dirpath, fn)
rel = os.path.relpath(full, root)
module = module_of(rel)
if fn.endswith('.xml') and ('mapper' in rel or 'smapper' in rel):
try:
with open(full, encoding='utf-8', errors='ignore') as fh:
text = fh.read()
except OSError:
continue
if '<mapper' not in text:
continue
# 先去注释，避免注释里的 SQL 片段污染
text = re.sub(r'<!--.*?-->', ' ', text, flags=re.DOTALL)
for m in TABLE_REF.finditer(text):
# ⚠️ `<update id="updateXxx">` / `<insert id=...>` 这类**标签属性**也会命中
# （标签名恰好是 update/insert，属性名恰好是 id），全仓 1613 处。
# 属性后面紧跟 `=`，真实的表名不会，据此排掉。
tail = text[m.end():m.end() + 8].lstrip()
if tail.startswith('='):
continue
schema, table = m.group(1), m.group(2)
record(table, module, rel, 'xml', schema)
for m in DYNAMIC_TABLE.finditer(text):
dynamic.setdefault((m.group(1).lower(), m.group(2)), set()).add(module)
elif fn.endswith('.java'):
try:
with open(full, encoding='utf-8', errors='ignore') as fh:
text = fh.read()
except OSError:
continue
for rx in (TABLE_ANNO, TABLENAME_ANNO):
for m in rx.finditer(text):
record(m.group(1), module, rel, 'entity')
return tables, dynamic
def main():
if len(sys.argv) != 3:
print(__doc__)
return 2
root, outdir = sys.argv[1], sys.argv[2]
if not os.path.isdir(root):
print('不是目录: %s' % root)
return 2
os.makedirs(outdir, exist_ok=True)
tables, dynamic = scan(root)
print('抽出表名 %d 个' % len(tables))
print('动态表名（运行时拼后缀）%d 族' % len(dynamic))
csv_path = os.path.join(outdir, 'table-inventory.csv')
with open(csv_path, 'w', newline='', encoding='utf-8') as fh:
w = csv.writer(fh)
w.writerow(['表名', '模块', '模块数', '引用文件数', '来源', 'schema', '跨模块'])
for name in sorted(tables):
e = tables[name]
mods = sorted(e['modules'])
w.writerow([
name, '|'.join(mods), len(mods), len(e['files']),
'|'.join(sorted(e['kinds'])), '|'.join(sorted(e['schemas'])),
'Y' if len(mods) > 1 else '',
])
print('写出 %s' % csv_path)
md_path = os.path.join(outdir, 'table-inventory.md')
with open(md_path, 'w', encoding='utf-8') as fh:
write_report(fh, tables, root, dynamic)
print('写出 %s' % md_path)
return 0
def write_report(fh, tables, root, dynamic=None):
w = fh.write
# 模块 → 单模块归属的表
owned = defaultdict(list)
cross = []
for name in sorted(tables):
mods = sorted(tables[name]['modules'])
if len(mods) == 1:
owned[mods[0]].append(name)
else:
cross.append((name, mods))
# 前缀 → 表
prefix = defaultdict(list)
for name in tables:
prefix[name.split('_')[0]].append(name)
w('# 表清单（静态抽取自 `%s`）\n\n' % root)
w('> 生成方式：`sql/tools/extract_table_inventory.py`，扫全仓 mapper XML 的\n')
w('> `FROM/JOIN/INTO/UPDATE` 与实体的 `@Table`/`@TableName`。\n')
w('> 不含：只由 DDL 变更日志新增、代码从未引用的表。\n\n')
w('## 1. 总量与模块归属\n\n')
w('共 **%d** 张表：单模块归属 %d 张，被多个模块触碰 %d 张。\n\n'
% (len(tables), sum(len(v) for v in owned.values()), len(cross)))
w('| 模块 | 表数 | 说明 |\n|---|---:|---|\n')
notes = {
'zteits-park': '停车场、车位、设备、进出场',
'zteits-order': '停车订单、共享订单',
'zteits-pay': '账户、客户、结算',
'zteits-bill': '计费规则、活动优惠、黑白名单、会员卡 —— `t_*` 前缀几乎全是它的',
'zteits-sys': '组织、用户、权限、字典',
'zteits-business': '商户/业务域',
'zteits-charge': '充电桩',
}
for mod, names in sorted(owned.items(), key=lambda kv: -len(kv[1])):
w('| `%s` | %d | %s |\n' % (mod, len(names), notes.get(mod, '')))
w('\n## 2. ⚠️ 跨模块表（合并后必须逐个定归属）\n\n')
w('这 %d 张表被**多个模块的 SQL 直接引用**。老系统靠"各模块连自己的库"侥幸隔离，\n'
% len(cross))
w('合库后它们变成真正可写，必须指定唯一归属模块，其余模块改走 `parking-module-api`：\n\n')
w('| 表 | 引用它的模块 | 建议归属 |\n|---|---|---|\n')
for name, mods in cross:
# 归属建议：取前缀所属模块；tp_ -> park，td_* -> 看具体业务
if name.startswith('tp_'):
suggest = '`zteits-park`'
elif name.startswith('td_b_') or name.startswith('td_p_'):
suggest = '`zteits-pay`（结算/账户）'
elif name.startswith('td_c_'):
suggest = '`zteits-pay`（客户）'
else:
suggest = '待定'
w('| `%s` | %s | %s |\n' % (name, '、'.join('`%s`' % m for m in mods), suggest))
w('\n## 3. 前缀 → 归属对照\n\n')
w('合并六库时**表名不改**，靠前缀区分业务域。前缀与模块**不是**一一对应，此表是证据：\n\n')
w('| 前缀 | 表数 | 主要归属模块 |\n|---|---:|---|\n')
for pfx, names in sorted(prefix.items(), key=lambda kv: -len(kv[1])):
mods = defaultdict(int)
for n in names:
for m in tables[n]['modules']:
mods[m] += 1
top = sorted(mods.items(), key=lambda kv: -kv[1])
w('| `%s_` | %d | %s |\n' % (
pfx, len(names),
'、'.join('`%s`(%d)' % (m, c) for m, c in top[:3])))
w('\n## 4. 疑似测试/遗留表（迁移时确认后丢弃）\n\n')
# ⚠️ 必须按整词匹配：用 `'temp' in name` 会把 `charger_assess_template`、
# `sms_template_config` 这类**正经的模板表**也当成临时表误报。
suspect_rx = re.compile(r'(^|_)(temp|test|nbtemp|tmp)($|_)')
suspects = [n for n in sorted(tables) if suspect_rx.search(n)]
for n in suspects:
w('- `%s` —— %s\n' % (n, '、'.join('`%s`' % m for m in sorted(tables[n]['modules']))))
w('\n## 5. ⚠️ 分表族（运行期拼后缀，静态数不出真实表名）\n\n')
if not dynamic:
w('无。\n')
else:
w('下列 SQL 的表名后缀是**调用方在运行时拼出来的**，所以：\n\n')
w('- 真实物理表**比本清单的表名多得多**（每月/每年一张），`SHOW TABLES` 会看到一批\n')
w(' "代码里搜不到"的表——它们正是这里拼出来的\n')
w('- 建表/清理脚本在仓库外（或定时任务里），合库时要连着这些**分表族**一起搬\n')
w('- 迁移改造时**不能**把 `${...}` 写死成某个月，要保留动态拼名逻辑\n\n')
w('| 表名前缀 | 变量 | 后缀格式 | 涉及模块 |\n|---|---|---|---|\n')
# 后缀格式均已回查调用方核实，不是猜的
hints = {
'tableName': '`yyyyMM`（`SimpleDateFormat("yyyyMM")`）',
'monthStr': '`yyyyMM`（`DateUtil.format(queryMonth,"yyyyMM")`）',
'tableStr': '`yyyyMM`（`DateUtil.format(endTime,"yyyyMM")`）',
'recordDateStr': '`yyyyMM`（`DateUtil.date().toString("yyyyMM")`）',
'year': '`yyyy`（`DateUtil.year(startDate)`）',
}
for (prefix, var), mods in sorted(dynamic.items()):
w('| `%s` | `%s` | %s | %s |\n'
% (prefix, var, hints.get(var, '需查调用方'),
'、'.join('`%s`' % m for m in sorted(mods))))
w('\n> ⚠️ 顺带查出的**存量缺陷**：`ParkingLotUseStatisticDaoImpl:169` 的\n')
w('> `recordStr` 用的是 `DateUtil.date()`（**当前时间**）而不是入参 `endDate` 所在月份。\n')
w('> 也就是说它查"任意历史区间的路侧 20 分钟数据"时，实际只查**本月**那张分表 ——\n')
w('> 跨月/历史区间查询会静默返回空或错数据。迁移时**不要原样搬**，要么修，\n')
w('> 要么明确记录为已知缺陷（修它属于业务逻辑变更，需单独确认）。\n')
w('\n## 6. 本抽取**不能**回答的问题（需只读连生产库）\n\n')
w('| 问题 | 为什么代码答不了 |\n|---|---|\n')
w('| 六库各有哪些表（含代码从未引用的） | 代码只暴露"用到过的表" |\n')
w('| 重名表的结构是否一致 | 需要 `SHOW CREATE TABLE` 逐个比对 |\n')
w('| 每张表的字符集是否 `utf8mb4` | 非 utf8mb4 的表合并后中文/emoji 会截断 |\n')
w('| `dev_park` / `dev_charge` 与正式库的差异 | 需要两个库各导一份结构 |\n')
w('| 各表的实际行数（决定迁移窗口） | 需要 `information_schema.TABLES` |\n')
w('\n> ⚠️ 老库是**生产库**。上表一律走**只读账号**导出，禁止任何写操作。\n')
if __name__ == '__main__':
sys.exit(main())