"""Bind only saved OQC name selections to products identified by source code.
|
Never re-draw templates, create materials, modify types/status, or import orders.
|
"""
|
import sys,json,csv,collections
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
rows=[x for x in json.loads((P/'mapping-plan.json').read_text(encoding='utf8')) if x['kind']=='oqc']
|
assert len(rows)==496
|
saved=json.loads((P/'random-selections.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,connect_timeout=10,read_timeout=45)
|
log=[];before={};pairrows=collections.defaultdict(list);review=[];new=[];existing=[]
|
try:
|
with c.cursor() as q:
|
q.execute('SELECT DATABASE() db');assert q.fetchone()['db']=='mom-hm'
|
q.execute("SELECT ENGINE e FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name='mes_qc_template_item'");assert q.fetchone()['e']=='InnoDB'
|
c.begin();q.execute("SELECT GET_LOCK('oqc-product-bind-20261012',10) ok");assert q.fetchone()['ok']==1
|
for table in ['mes_qc_oqc','mes_qc_ipqc']:
|
q.execute('SELECT COUNT(*) n FROM '+table);before[table+'_count']=q.fetchone()['n']
|
q.execute('SELECT * FROM mdm_item WHERE deleted=0 FOR UPDATE');items={str(x['code']):x for x in q.fetchall()}
|
q.execute('SELECT * FROM mes_qc_template WHERE deleted=0 FOR UPDATE');templates={x['id']:x for x in q.fetchall()}
|
before['templates']=[{k:t[k] for k in ['id','code','name','types','status']} for t in templates.values()]
|
q.execute('SELECT * FROM mes_qc_template_item WHERE deleted=0 FOR UPDATE');oldlinks=q.fetchall();before['relations']=oldlinks
|
oldpairs=collections.defaultdict(list)
|
for x in oldlinks:oldpairs[(x['template_id'],x['item_id'])].append(x['id'])
|
for x in rows:
|
r=x['source'];m=x['mapping'];selected=saved['oqc|'+r['检验单名称']]['selected_id'];assert selected==m['selected_id']
|
code=str(r['产品物料编码']) if r['产品物料编码'] is not None else None
|
product=items.get(code);template=templates.get(selected);issues=[]
|
if not code:issues.append('原表物料编码为空，未随机选择产品')
|
elif not product:issues.append('原表物料编码未找到，未新建物料：'+code)
|
elif product['status']!=0:issues.append('产品物料已停用')
|
if selected is None:issues.append('无名称匹配方案')
|
elif not template:issues.append('保存的方案已不存在')
|
else:assert template['name']==m['selected_name'],'Saved template name changed'
|
if product and x['item_id'] is not None:assert product['id']==x['item_id'] and product['name']==x['item_name'],'Source material changed'
|
record={'excel_row':x['excel_row'],'inspection_code':r['检验单编号'],'inspection_name':r['检验单名称'],'source_item_code':code,'item_id':product['id'] if product else None,'item_name':product['name'] if product else None,'template_id':selected,'template_name':template['name'] if template else m['selected_name'],'template_types':template['types'] if template else None,'template_status':template['status'] if template else None,'issues':issues}
|
if not issues:pairrows[(selected,product['id'])].append(record)
|
review.append(record)
|
for pair,source in pairrows.items():
|
if pair in oldpairs:existing.append({'template_id':pair[0],'item_id':pair[1],'relation_ids':oldpairs[pair]});continue
|
sql="INSERT INTO mes_qc_template_item(template_id,item_id,remark,creator,updater) VALUES (%s,%s,%s,'1','1')"
|
args=(pair[0],pair[1],'来源：出货检验导入表；按检验单名称与方案名称模糊映射，多候选依用户要求随机选择并留档。')
|
log.append(q.mogrify(sql,args)+';');q.execute(sql,args)
|
new.append({'template_id':pair[0],'item_id':pair[1],'relation_ids':[q.lastrowid]})
|
allpairs={**{(x['template_id'],x['item_id']):x for x in existing},**{(x['template_id'],x['item_id']):x for x in new}}
|
for r in review:
|
if r['issues']:r['action']='暂缓绑定';continue
|
pair=(r['template_id'],r['item_id']);r['action']='已存在，复用' if pair in oldpairs else '新增绑定';r['relation_ids']=allpairs[pair]['relation_ids']
|
(P/'before-oqc-product-bind.json').write_text(json.dumps(before,ensure_ascii=False,default=str,indent=2),encoding='utf8')
|
(P/'oqc-product-bind-executed.sql').write_text('START TRANSACTION;\n'+'\n'.join(log)+'\nCOMMIT;\n',encoding='utf8')
|
for pair in pairrows:
|
q.execute('SELECT COUNT(*) n FROM mes_qc_template_item WHERE template_id=%s AND item_id=%s AND deleted=0',pair);assert q.fetchone()['n']>=1
|
for table in ['mes_qc_oqc','mes_qc_ipqc']:
|
q.execute('SELECT COUNT(*) n FROM '+table);assert q.fetchone()['n']==before[table+'_count']
|
q.execute('SELECT id,code,name,types,status FROM mes_qc_template WHERE deleted=0 ORDER BY id');assert q.fetchall()==sorted(before['templates'],key=lambda x:x['id'])
|
c.commit()
|
# Verify committed associations and preserve any pre-existing duplicate links.
|
for pair in pairrows:
|
q.execute('SELECT id FROM mes_qc_template_item WHERE template_id=%s AND item_id=%s AND deleted=0',pair);found=q.fetchall();assert len(found)==(len(oldpairs[pair]) if pair in oldpairs else 1)
|
q.execute('SELECT COUNT(*) n FROM mes_qc_template_item WHERE deleted=0');assert q.fetchone()['n']==len(oldlinks)+len(new)
|
q.execute("SELECT RELEASE_LOCK('oqc-product-bind-20261012')")
|
result={'target':'36.213.150.237:10000/mom-hm','source_rows':len(rows),'bound_source_rows':sum(not r['issues'] for r in review),'pending_source_rows':sum(bool(r['issues']) for r in review),'distinct_products':len({pair[1] for pair in pairrows}),'distinct_templates':len({pair[0] for pair in pairrows}),'new_bindings':len(new),'already_existing_bindings':len(existing),'new_relations':new,'existing_relations':existing,'scope':'仅出货表映射对应产品关系，未导入检验单，未变更方案类型和启用状态，未处理过程检验','post_commit_verified':True}
|
(P/'oqc-product-bind-result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
(P/'oqc-product-bind-review.json').write_text(json.dumps(review,ensure_ascii=False,indent=2),encoding='utf8')
|
with (P/'出货检验-产品方案绑定结果.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['原表行号','检验单编号','检验单名称','原表物料编码','物料ID','产品名称','方案ID','方案名称','方案类型','方案状态','处理结果','关系ID','暂缓原因'])
|
for r in sorted(review,key=lambda x:x['excel_row']):w.writerow([r['excel_row'],r['inspection_code'],r['inspection_name'],r['source_item_code'],r['item_id'],r['item_name'],r['template_id'],r['template_name'],r['template_types'],r['template_status'],r['action'],','.join(map(str,r.get('relation_ids',[]))),'；'.join(r['issues'])])
|
print(json.dumps({k:v for k,v in result.items() if k not in ['new_relations','existing_relations']},ensure_ascii=False))
|
except Exception:c.rollback();raise
|
finally:c.close()
|
