"""Remove placeholder labels without inventing measurement or sampling facts."""
|
import sys,json
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.stdout.reconfigure(encoding='utf8')
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
assert not (P/'display-labels-result.json').exists(),'Already processed.'
|
preview=json.loads((P/'display-labels-preview.json').read_text(encoding='utf8'))
|
mapping={'演示：符合现有方案描述（非真实实测）':'待复核','演示样品，非真实质检记录':None,'演示外观、数量核验，不作为真实检验标准':'外观、数量核验；检验方法适用性待确认','演示数量一致性核验，不作为真实检验标准':'数量一致性核验；检验方法适用性待确认'}
|
c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',autocommit=False,cursorclass=pymysql.cursors.DictCursor)
|
try:
|
with c.cursor() as q:
|
q.execute("SET time_zone='+08:00'")
|
q.execute('SELECT DATABASE() db,NOW() now,GET_LOCK(%s,10) locked',('quality-indicator-20261014',));state=q.fetchone();assert state['db']=='mom-hm' and state['locked']==1
|
tables=sorted({x['table'] for x in preview});before={}
|
for t in tables:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id FOR UPDATE');before[t]=q.fetchall()
|
backup=P/('display-labels-before-'+state['now'].strftime('%Y%m%d-%H%M%S')+'.json')
|
backup.write_text(json.dumps({'state':state,'tables':before},ensure_ascii=False,default=str,indent=2),encoding='utf8')
|
expected={};counts={};changes=[]
|
for batch in preview:
|
t=batch['table'];col=batch['column'];live={x['id']:x for x in before[t]};groups={}
|
assert col in {'value','sampling_location','check_method'}
|
for p in batch['rows']:
|
old=live[p['id']];assert old[col]==p['value'] and p['value'] in mapping
|
new=mapping[p['value']];groups.setdefault(new,[]).append(p['id']);expected.setdefault((t,p['id']),{})[col]=new
|
changes.append({'table':t,'id':p['id'],'field':col,'old':p['value'],'new':new})
|
counts[t+'.'+col]=len(batch['rows'])
|
for new,ids in groups.items():
|
q.execute('UPDATE '+t+' SET `'+col+'`=%s,updater=%s,update_time=update_time WHERE deleted=0 AND id IN ('+','.join(['%s']*len(ids))+')',[new,'289',*ids]);assert q.rowcount==len(ids)
|
after={}
|
for t,rows in before.items():
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');after[t]=q.fetchall();assert len(rows)==len(after[t])
|
for old,new in zip(rows,after[t]):
|
assert old['id']==new['id'];updates=expected.get((t,old['id']),{});diff={k for k in old if old[k]!=new[k]};assert diff<=(set(updates)|{'updater'} if updates else set())
|
for col,val in updates.items():assert new[col]==val
|
c.commit()
|
for t,rows in after.items():
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');assert q.fetchall()==rows
|
result={'committed':True,'verified':True,'at':str(state['now']),'count':len(changes),'counts':counts,'changes':changes,'backup':backup.name,'numeric_measurements_states_relations_and_times_unchanged':True}
|
(P/'display-labels-result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
print(json.dumps({k:v for k,v in result.items() if k!='changes'},ensure_ascii=False))
|
except Exception:c.rollback();raise
|
finally:
|
with c.cursor() as q:q.execute('SELECT RELEASE_LOCK(%s)',('quality-indicator-20261014',))
|
c.close()
|
