import sys,json
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
b=json.loads((P/'indicator-dedup-locked-backup.json').read_text(encoding='utf8'))
|
mapping={int(k):v for k,v in b['mapping'].items()}
|
normalize=lambda r:json.loads(json.dumps(r,ensure_ascii=False,default=str))
|
c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',cursorclass=pymysql.cursors.DictCursor)
|
with c.cursor() as q:
|
q.execute('SELECT TRIM(name) name,COUNT(*) n FROM mes_qc_indicator WHERE deleted=0 GROUP BY TRIM(name) HAVING COUNT(*)>1')
|
assert not q.fetchall()
|
q.execute('SELECT * FROM mes_qc_indicator ORDER BY id');masters={r['id']:normalize(r) for r in q.fetchall()}
|
for old in b['masters']:
|
current=masters[old['id']]
|
if old['id'] in mapping:
|
assert current['deleted']=="b'\\x01'"
|
else:assert current==old
|
verified=0
|
for table,rows in b['references'].items():
|
if not rows:continue
|
ids=[r['id'] for r in rows]
|
q.execute('SELECT * FROM `'+table+'` WHERE id IN ('+','.join('%s' for _ in ids)+')',ids)
|
current={r['id']:normalize(r) for r in q.fetchall()};assert len(current)==len(rows)
|
for old in rows:
|
new=current[old['id']]
|
assert new['indicator_id']==mapping.get(old['indicator_id'],old['indicator_id'])
|
allowed={'indicator_id','updater','update_time'} if old['indicator_id'] in mapping else set()
|
assert all(new[k]==v for k,v in old.items() if k not in allowed)
|
verified+=1
|
print(json.dumps({'duplicate_names_remaining':0,'duplicate_masters_logically_deleted':len(mapping),'reference_rows_verified':verified,'requirements_and_results_preserved':True},ensure_ascii=False))
|
c.close()
|
