"""Simplify inspection-item display labels/remarks; preserve all technical fields."""
|
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/'result.json').exists(),'Already completed; inspect instead of repeating.'
|
remarks={
|
'来源：SOP-QC-YL；分组项':'原料检验项目分组。',
|
'SOP原料检验；标准以方案条款为准，人工判定。':'按质检方案要求检验并人工判定。',
|
'上一检验项的判定要求续文，不参与独立判定，无需重复检验。':'前项检验要求的补充说明，不单独判定或重复检验。',
|
'来源：食品安全国家标准；文本录入并人工判定，不自动生成实测结果。':'依据适用食品安全国家标准检验；检验结果由人员录入并判定。'}
|
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
|
q.execute("SELECT table_name FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name LIKE 'mes_qc_%'");tabs=[x['TABLE_NAME'] for x in q.fetchall()];before={}
|
for t in tabs:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id'+(' FOR UPDATE' if t=='mes_qc_indicator' else ''));before[t]=q.fetchall()
|
q.execute('SELECT status FROM system_users WHERE id=289 AND deleted=0');assert q.fetchone()['status']==0
|
backup=P/('locked-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')
|
rows=before['mes_qc_indicator'];changes=[];names=[]
|
for old in rows:
|
name=old['name'][3:] if old['name'].startswith('SOP原料检验-') else old['name']
|
remark=remarks.get(old['remark'],old['remark']);names.append(name)
|
if name==old['name'] and remark==old['remark']:continue
|
q.execute('UPDATE mes_qc_indicator SET name=%s,remark=%s,updater=%s,update_time=update_time WHERE id=%s AND deleted=0',(name,remark,'289',old['id']));assert q.rowcount==1
|
changes.append({'id':old['id'],'old_name':old['name'],'name':name,'old_remark':old['remark'],'remark':remark})
|
assert len(names)==len(set(names)),'Name normalization must not introduce duplicate names'
|
expected={x['id']:x for x in changes};after={}
|
for t,oldrows in before.items():
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');newrows=q.fetchall();after[t]=newrows;assert len(oldrows)==len(newrows)
|
for old,new in zip(oldrows,newrows):
|
assert old['id']==new['id'];diff={k for k in old if old[k]!=new[k]}
|
assert diff<=({'name','remark','updater'} if t=='mes_qc_indicator' and old['id'] in expected else set()),(t,old['id'],diff)
|
if t=='mes_qc_indicator' and old['id'] in expected:assert new['name']==expected[old['id']]['name'] and new['remark']==expected[old['id']]['remark']
|
c.commit()
|
for t,rows in after.items():
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');assert q.fetchall()==rows,('Post-commit mismatch',t)
|
result={'committed':True,'verified':True,'at':str(state['now']),'remarks_changed':sum(x['old_remark']!=x['remark'] for x in changes),'names_changed':sum(x['old_name']!=x['name'] for x in changes),'changes':changes,'backup':backup.name,'limits_tools_units_formulas_results_and_associations_unchanged':True,'duplicate_names':0}
|
(P/'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()
|
