import sys,json
|
from pathlib import Path
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
P=Path(__file__).parent
|
r=json.loads((P/'normalize-applied.json').read_text(encoding='utf8'))
|
before=json.loads((P/'before-normalize-apply.json').read_text(encoding='utf8'))
|
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)
|
canonical=lambda v:json.loads(json.dumps(v,ensure_ascii=False,default=str))
|
with c.cursor() as q:
|
# A placeholder supplier descriptor is a display name too; do not invent a real supplier.
|
for row in before['mes_qc_iqc']:
|
if row['vendor_name']=='演示批次来料（不代表真实供应商交付）':
|
q.execute('SELECT vendor_name FROM mes_qc_iqc WHERE id=%s',(row['id'],));v=q.fetchone()['vendor_name']
|
assert v in [row['vendor_name'],'批次来料']
|
q.execute('UPDATE mes_qc_iqc SET vendor_name=%s,update_time=update_time WHERE id=%s',('批次来料',row['id']))
|
change=next(x for x in r['changes'] if x['table']=='mes_qc_iqc' and x['id']==row['id'])
|
change['before']['vendor_name']=row['vendor_name'];change['after']['vendor_name']='批次来料'
|
changes={(x['table'],x['id']):x['after'] for x in r['changes']}
|
checked=0
|
for t,rows in before.items():
|
for row in rows:
|
q.execute('SELECT * FROM `'+t+'` WHERE id=%s',(row['id'],));actual=q.fetchone()
|
expected={**row,**changes.get((t,row['id']),{})}
|
assert canonical(actual)==expected,(t,row['id'])
|
for k,v in actual.items():
|
if not isinstance(v,str):continue
|
if k=='name' or k.endswith('_name'):assert '演示' not in v,(t,row['id'],k)
|
if k in ['code','no','record_no','lot_number','shipping_number'] or k.endswith('_code') or k.endswith('_no'):assert 'OCT0814-DEMO' not in v,(t,row['id'],k)
|
if t in ['crm_sale_quotation','crm_contract','erp_sale_order'] or t.startswith('mes_') and 'code' in actual:
|
col='no' if t.startswith('crm_') or t=='erp_sale_order' else 'code'
|
if actual.get(col):
|
q.execute('SELECT COUNT(*) n FROM `'+t+'` WHERE `'+col+'`=%s',(actual[col],));assert q.fetchone()['n']==1,(t,actual[col])
|
checked+=1
|
c.commit()
|
(P/'normalize-applied.json').write_text(json.dumps(r,ensure_ascii=False,indent=2),encoding='utf8')
|
result={'verified':True,'rows_checked':checked,'codes_unique':True,'names_normalized':True,'all_other_fields_unchanged':True,'measurement_provenance_retained':True}
|
(P/'normalize-verification.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
print(json.dumps(result,ensure_ascii=False));c.close()
|
