61 lines
3.4 KiB
Python
61 lines
3.4 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""只读探测：12 个 SQL 脚本在三个库上的执行状态。"""
|
||
import io, sys, pymysql
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
||
CONNS = {
|
||
'200/zentao_dev (生产)': dict(host='192.168.3.200', user='root', password='PX4fTAAsJ#T!1',
|
||
database='zentao_dev', charset='utf8mb4', connect_timeout=15),
|
||
'161/zentao_dev': dict(host='192.168.1.161', user='devgps', password='dev@2021GPS',
|
||
database='zentao_dev', charset='utf8mb4', connect_timeout=15),
|
||
'161/zentao_dev_2026': dict(host='192.168.1.161', user='devgps', password='dev@2021GPS',
|
||
database='zentao_dev_2026', charset='utf8mb4', connect_timeout=15),
|
||
}
|
||
|
||
# (SQL文件, [探测项]) 探测项: ('col', 表, 列) / ('table', 表) / ('row', SQL, 期望值>0)
|
||
CHECKS = [
|
||
('20260724_alter_zt_file_add_url', [('col', 'zt_file', 'url')]),
|
||
('20260724_alter_zt_meeting_add_url', [('col', 'zt_meeting', 'url')]),
|
||
('20260724_alter_zt_story_add_ai_doc_urls', [('col', 'zt_story', c) for c in
|
||
['prd_url', 'code_review_url', 'work_log_url', 'test_case_url',
|
||
'test_report_download_url', 'test_report_submit_url', 'test_other_url', 'code_review_status']]),
|
||
('20260724_alter_zt_story_expand_add_ai_rate', [('col', 'zt_story_expand', 'ai_participation_rate')]),
|
||
('20260724_insert_ai_user', [('row', "SELECT COUNT(*) FROM zt_user WHERE account='ai'")]),
|
||
('20260728_create_zt_doc_check', [('table', 'zt_doc_check')]),
|
||
('20260728_create_zt_perf_config', [('table', 'zt_perf_config')]),
|
||
('20260728_init_zt_perf_config', [('row', "SELECT COUNT(*) FROM zt_perf_config")]),
|
||
('20260729_alter_zt_story_expand_acceptance', [('col', 'zt_story_expand', 'acceptance_criteria')]),
|
||
('20260729_insert_perf_menu', [('row', "SELECT COUNT(*) FROM base_menu WHERE id BETWEEN 1539 AND 1551")]),
|
||
('20260730_alter_zt_user_dev_direction', [('col', 'zt_user', 'dev_direction')]),
|
||
('20260731_alter_zt_story_doc_update', [('col', 'zt_story', 'ai_doc_update_url')]),
|
||
]
|
||
|
||
for label, cfg in CONNS.items():
|
||
try:
|
||
conn = pymysql.connect(**cfg)
|
||
except Exception as e:
|
||
print(f'\n===== {label}: 连接失败 {e}'); continue
|
||
cur = conn.cursor()
|
||
print(f'\n===== {label} =====')
|
||
for fname, probes in CHECKS:
|
||
marks = []
|
||
for p in probes:
|
||
try:
|
||
if p[0] == 'col':
|
||
cur.execute("""SELECT COUNT(*) FROM information_schema.columns
|
||
WHERE table_schema=DATABASE() AND table_name=%s AND column_name=%s""",
|
||
(p[1], p[2]))
|
||
marks.append(f'{p[2]}={"√" if cur.fetchone()[0] else "×"}')
|
||
elif p[0] == 'table':
|
||
cur.execute("""SELECT COUNT(*) FROM information_schema.tables
|
||
WHERE table_schema=DATABASE() AND table_name=%s""", (p[1],))
|
||
marks.append(f'{p[1]}={"√" if cur.fetchone()[0] else "×"}')
|
||
else:
|
||
cur.execute(p[1])
|
||
marks.append(f'行数={cur.fetchone()[0]}')
|
||
except Exception as e:
|
||
marks.append(f'ERR:{e}')
|
||
ok = all('×' not in m and 'ERR' not in m and '行数=0' not in m for m in marks)
|
||
print(f' {"✅" if ok else "❌"} {fname}: {"; ".join(marks)}')
|
||
conn.close()
|