54 lines
2.5 KiB
Python
54 lines
2.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""只读勘察第三轮：自定义表 zt_story_expand 及 story 相关自定义表。绝不写入。"""
|
|
import io, sys, pymysql
|
|
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
|
|
|
SRC = dict(host='192.168.3.200', user='root', password='PX4fTAAsJ#T!1',
|
|
database='zentao_dev', charset='utf8mb4', connect_timeout=15)
|
|
DST = dict(host='192.168.1.161', user='devgps', password='dev@2021GPS',
|
|
database='zentao_dev_2026', charset='utf8mb4', connect_timeout=15)
|
|
SID = 9197
|
|
|
|
src = pymysql.connect(**SRC); scur = src.cursor()
|
|
dst = pymysql.connect(**DST); dcur = dst.cursor()
|
|
|
|
def story_tables(cur, db):
|
|
cur.execute("""SELECT table_name FROM information_schema.tables
|
|
WHERE table_schema=%s AND (table_name LIKE '%%story%%' OR table_name LIKE '%%expand%%'
|
|
OR table_name LIKE '%%deliver%%' OR table_name LIKE '%%ys%%')""", (db,))
|
|
return sorted(r[0] for r in cur.fetchall())
|
|
|
|
print('200 上相关表:', story_tables(scur, 'zentao_dev'))
|
|
print('161_2026 相关表:', story_tables(dcur, 'zentao_dev_2026'))
|
|
|
|
for label, cur, db in [('200', scur, 'zentao_dev'), ('161_2026', dcur, 'zentao_dev_2026')]:
|
|
cur.execute("""SELECT column_name FROM information_schema.columns
|
|
WHERE table_schema=%s AND table_name='zt_story_expand'""", (db,))
|
|
cols = [r[0] for r in cur.fetchall()]
|
|
if not cols:
|
|
print(f'{label}: zt_story_expand 不存在'); continue
|
|
key = 'story_id' if 'story_id' in cols else ('story' if 'story' in cols else None)
|
|
if not key:
|
|
print(f'{label}: zt_story_expand 列={cols}'); continue
|
|
cur.execute(f"SELECT * FROM zt_story_expand WHERE `{key}`=%s", (SID,))
|
|
rows = cur.fetchall()
|
|
print(f'{label} zt_story_expand({key}=9197): {len(rows)} 行')
|
|
for r in rows:
|
|
print(' ', dict(zip(cols, r)))
|
|
|
|
# task 表相关：zt_team / zt_effort 是否涉及这两个任务
|
|
for t in ['zt_team', 'zt_effort', 'zt_taskestimate']:
|
|
for label, cur, db in [('200', scur, 'zentao_dev'), ('161_2026', dcur, 'zentao_dev_2026')]:
|
|
cur.execute("""SELECT table_name FROM information_schema.tables
|
|
WHERE table_schema=%s AND table_name=%s""", (db, t))
|
|
if not cur.fetchone():
|
|
continue
|
|
try:
|
|
cur.execute(f"SELECT COUNT(*) FROM `{t}` WHERE (account IN ('zhouwenjun','sunying') AND project IN (142,286)) OR (task IN (18841,18849))")
|
|
print(f'{label} {t} 相关行: {cur.fetchone()[0]}')
|
|
except Exception:
|
|
pass
|
|
|
|
src.close(); dst.close()
|
|
print('done')
|