"""Consolidate indicator master names; never delete inspection lines or results."""
|
import sys,json,csv
|
from pathlib import Path
|
from collections import defaultdict,Counter
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
encode=lambda r:json.dumps(r,ensure_ascii=False,default=str,sort_keys=True)
|
s=json.loads((P/'before-indicator-dedup.json').read_text(encoding='utf8'))
|
keys={'mes_qc_template_indicator':['template_id'],'mes_qc_indicator_result_detail':['result_id','measure_no'],
|
'mes_qc_iqc_line':['iqc_id'],'mes_qc_oqc_line':['oqc_id'],'mes_qc_ipqc_line':['ipqc_id'],'mes_qc_rqc_line':['rqc_id']}
|
mapping={};plan=[]
|
for group in s['duplicates']:
|
rows=sorted(group['rows'],key=lambda r:r['id']);keep=rows[0]
|
assert {(r['item_type'],r['result_type']) for r in rows}=={(1,3)}, 'Incompatible result types'
|
assert not any(r['parent_id'] in {x['id'] for x in rows} for r in rows)
|
for duplicate in rows[1:]:
|
mapping[duplicate['id']]=keep['id']
|
plan.append({'name':group['name'],'retained_id':keep['id'],'retained_code':keep['code'],
|
'duplicate_id':duplicate['id'],'duplicate_code':duplicate['code'],
|
'original_unit_text':duplicate['unit_text'],'original_result_specification':duplicate['result_specification']})
|
assert len(mapping)==80 and len(s['duplicates'])==29
|
ids=[r['id'] for group in s['duplicates'] for r in group['rows']]
|
marks=','.join('%s' for _ in ids)
|
updater='qc-name-dedup-20261013'
|
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,autocommit=False)
|
counts={};backups={}
|
try:
|
with c.cursor() as q:
|
q.execute("SELECT DATABASE() db,NOW() now");state=q.fetchone();assert state['db']=='mom-hm';now=state['now']
|
q.execute("SELECT GET_LOCK('codex-qc-name-dedup',0) locked");assert q.fetchone()['locked']==1
|
q.execute('SELECT * FROM mes_qc_indicator ORDER BY id FOR UPDATE');masters=q.fetchall()
|
assert encode(masters)==encode(s['indicators']), 'Indicator masters changed since inspection'
|
# No group master is being removed, hence no children need moving.
|
assert not any(r['parent_id'] in mapping for r in masters)
|
for table in s['references']:
|
assert table in keys
|
q.execute('SELECT * FROM `'+table+'` WHERE indicator_id IN ('+marks+') ORDER BY id FOR UPDATE',ids)
|
rows=q.fetchall();backups[table]=rows
|
assert encode(rows)==encode(s['references'][table]['rows']), 'References changed: '+table
|
# Consolidation must not create ambiguous repeated measurements or template rows.
|
grouped=defaultdict(list)
|
for row in rows:
|
if row['deleted'] in (0,False,b'\x00'):
|
grouped[tuple(row[k] for k in keys[table])+(mapping.get(row['indicator_id'],row['indicator_id']),)].append(row['id'])
|
assert not [v for v in grouped.values() if len(v)>1], 'Duplicate target rows: '+table
|
(P/'indicator-dedup-locked-backup.json').write_text(encode({'masters':masters,'references':backups,'mapping':mapping,'at':str(now)}),encoding='utf8')
|
case='CASE indicator_id '+' '.join('WHEN %s THEN %s' for _ in mapping)+' ELSE indicator_id END'
|
pairs=[v for pair in mapping.items() for v in pair];oldmarks=','.join('%s' for _ in mapping)
|
for table,rows in backups.items():
|
q.execute('UPDATE `'+table+'` SET indicator_id='+case+',updater=%s,update_time=%s WHERE indicator_id IN ('+oldmarks+')',pairs+[updater,now]+list(mapping))
|
counts[table]=q.rowcount
|
assert counts[table]==sum(r['indicator_id'] in mapping for r in rows)
|
q.execute('UPDATE mes_qc_indicator SET deleted=1,updater=%s,update_time=%s WHERE deleted=0 AND id IN ('+oldmarks+')',[updater,now]+list(mapping))
|
assert q.rowcount==len(mapping)
|
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(), 'Duplicate names remain'
|
q.execute('SELECT * FROM mes_qc_indicator ORDER BY id');aftermasters={r['id']:r for r in q.fetchall()}
|
for original in masters:
|
new=aftermasters[original['id']]
|
if original['id'] in mapping:
|
assert new['deleted']==b'\x01'
|
assert all(new[k]==v for k,v in original.items() if k not in {'deleted','updater','update_time'})
|
else:assert encode(new)==encode(original)
|
for table,rows in backups.items():
|
q.execute('SELECT * FROM `'+table+'` WHERE indicator_id IN ('+marks+') ORDER BY id',ids)
|
current={r['id']:r for r in q.fetchall()}
|
assert len(current)==len(rows)
|
for original in rows:
|
new=current[original['id']]
|
assert new['indicator_id']==mapping.get(original['indicator_id'],original['indicator_id'])
|
allowed={'indicator_id','updater','update_time'} if original['indicator_id'] in mapping else set()
|
assert all(new[k]==v for k,v in original.items() if k not in allowed), 'Requirement or measured value changed'
|
q.execute('SELECT COUNT(*) n FROM `'+table+'` WHERE indicator_id IN ('+oldmarks+')',list(mapping))
|
assert q.fetchone()['n']==0
|
c.commit()
|
active_before=sum(r['deleted'] in (0,False,b'\x00') for r in masters)
|
result={'duplicate_name_groups':len(s['duplicates']),'logically_deleted_duplicates':len(mapping),
|
'active_before':active_before,'active_after':active_before-len(mapping),'reference_rows_migrated':counts,
|
'total_reference_rows_migrated':sum(counts.values()),'requirements_and_measured_values_unchanged':True,
|
'all_retained_master_fields_unchanged':True,'executed_at':str(now)}
|
(P/'indicator-name-dedup-result.json').write_text(encode(result),encoding='utf8')
|
(P/'indicator-name-dedup-mapping.json').write_text(encode(plan),encoding='utf8')
|
with (P/'检验项名称去重对应.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['检验项名称','保留ID','保留编码','合并ID','合并编码','原单位文本','原结果值属性（历史定义保留在逻辑删除项）'])
|
for r in plan:w.writerow([r['name'],r['retained_id'],r['retained_code'],r['duplicate_id'],r['duplicate_code'],r['original_unit_text'],r['original_result_specification']])
|
print(encode(result))
|
finally:c.close()
|
