legacy_db.py
5.59 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
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
老库（只读）连库探测工具 —— P2 六库合一前的结构比对用。
为什么不用 mysql CLI：本机 Homebrew 的 `mysql` 客户端是 26.7，**已移除
`mysql_native_password` 插件**，而老库账号用的正是这个认证方式，直接连会报
`ERROR 2059 (HY000): Authentication plugin 'mysql_native_password' cannot be loaded`。
PyMySQL 用纯 Python 实现了该插件，所以这里统一走 PyMySQL。
⚠️ 安全约定：本脚本**只发 SELECT / SHOW / 读 information_schema**，
不提供任何写操作的入口。凭据从 `私密/legacy-ro.env` 读，不落日志。
用法：
python3 legacy_db.py --env 私密/legacy-ro.env ping
python3 legacy_db.py --env 私密/legacy-ro.env databases
python3 legacy_db.py --env 私密/legacy-ro.env tables --db zt_park
python3 legacy_db.py --env 私密/legacy-ro.env show-create --db zt_park --table tp_p_lot
"""
import argparse
import os
import re
import sys
try:
import pymysql
except ImportError:
print('缺少 pymysql：python3 -m pip install pymysql', file=sys.stderr)
sys.exit(2)
# 模块名 -> env 文件里的键名（键名是历史约定，与实际库名不一定对得上）
MODULE_KEYS = ['SYS', 'PARK', 'PAY', 'ORDER', 'BILL', 'CHARGE']
def load_env(path):
"""读 KEY=VALUE 形式的 env 文件。不 export、不执行、不插值——纯文本解析。"""
cfg = {}
with open(path, encoding='utf-8') as fh:
for raw in fh:
line = raw.strip()
if not line or line.startswith('#') or '=' not in line:
continue
k, _, v = line.partition('=')
cfg[k.strip()] = v.strip()
return cfg
def db_names(cfg):
"""{模块名: 库名}。键名是 LEGACY_DB_<MODULE>（注意 PAY 那一项历史上被填成了 zt_order）。"""
out = {}
for mod in MODULE_KEYS:
name = cfg.get('LEGACY_DB_' + mod, '').strip()
if name:
out[mod] = name
return out
def connect(cfg, db=None, read_timeout=120):
"""read_timeout 默认 120s。大表的逐行探测（如 park_in_out_record 的 115 万行回查
9600 万行）会超，调用方按需放大——见 check_subset.py 的 --timeout。"""
return pymysql.connect(
host=cfg['LEGACY_DB_HOST'],
port=int(cfg.get('LEGACY_DB_PORT', 3306)),
user=cfg['LEGACY_DB_USER'],
password=cfg['LEGACY_DB_PASSWORD'],
database=db,
charset='utf8mb4',
connect_timeout=20,
read_timeout=read_timeout,
cursorclass=pymysql.cursors.Cursor,
# 老库是 MySQL 5.7/8.0 混合，关闭 SSL 强制，避免握手差异
ssl_disabled=True,
)
def query(cfg, sql, db=None, args=None):
conn = connect(cfg, db)
try:
with conn.cursor() as cur:
cur.execute(sql, args)
return cur.fetchall()
finally:
conn.close()
def cmd_ping(cfg, _a):
rows = query(cfg, "SELECT VERSION() AS ver, CURRENT_USER() AS who, "
"@@hostname AS hostname, @@character_set_server AS cs, "
"@@lower_case_table_names AS lctn")
r = rows[0]
print('版本 : %s' % r[0])
print('当前账号 : %s' % r[1])
print('主机 : %s' % r[2])
print('服务端字符集: %s' % r[3])
print('lower_case_table_names: %s (0=区分大小写，1=不区分；合库前必须知道这个值)' % r[4])
print('')
return r
def cmd_databases(cfg, _a):
rows = query(cfg, "SELECT SCHEMA_NAME, DEFAULT_CHARACTER_SET_NAME, DEFAULT_COLLATION_NAME "
"FROM information_schema.SCHEMATA ORDER BY SCHEMA_NAME")
print('%-40s %-16s %s' % ('库名', '字符集', '排序规则'))
print('-' * 80)
for r in rows:
print('%-40s %-16s %s' % r)
return rows
def cmd_tables(cfg, a):
wanted = db_names(cfg)
mods = [a.module] if a.module else sorted(wanted)
for mod in mods:
if mod not in wanted:
print('env 里没有 LEGACY_DB_%s' % mod, file=sys.stderr)
continue
db = wanted[mod]
try:
rows = query(cfg, "SELECT TABLE_NAME FROM information_schema.TABLES "
"WHERE TABLE_SCHEMA=%s ORDER BY TABLE_NAME", db=(db), args=(db,))
except Exception as e: # noqa: BLE001
print('[%s] 库 %s 连不上/不存在：%s' % (mod, db, e))
continue
print('[%s] 库 %s 共 %d 张表' % (mod, db, len(rows)))
for r in rows:
print(' %s' % r[0])
def cmd_show_create(cfg, a):
rows = query(cfg, "SHOW CREATE TABLE `%s`" % a.table, db=a.db)
print(rows[0][1])
def main():
ap = argparse.ArgumentParser()
ap.add_argument('--env', default=os.path.join(
os.path.dirname(os.path.abspath(__file__)), '..', '..', '私密', 'legacy-ro.env'))
sub = ap.add_subparsers(dest='cmd', required=True)
sub.add_parser('ping').set_defaults(fn=cmd_ping)
sub.add_parser('databases').set_defaults(fn=cmd_databases)
p = sub.add_parser('tables')
p.add_argument('--module', choices=MODULE_KEYS)
p.set_defaults(fn=cmd_tables)
p = sub.add_parser('show-create')
p.add_argument('--db', required=True)
p.add_argument('--table', required=True)
p.set_defaults(fn=cmd_show_create)
a = ap.parse_args()
cfg = load_env(a.env)
missing = [k for k in ('LEGACY_DB_HOST', 'LEGACY_DB_USER', 'LEGACY_DB_PASSWORD') if not cfg.get(k)]
if missing:
print('env 文件缺字段：%s' % ', '.join(missing), file=sys.stderr)
return 2
a.fn(cfg, a)
return 0
if __name__ == '__main__':
sys.exit(main())