1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
| import sys,json,re,csv
| from pathlib import Path
| sys.path.insert(0,str(Path(__file__).parent/'deps'))
| import pymysql
| P=Path(__file__).parent
| src=json.loads((P/'source.json').read_text(encoding='utf8'))[0]['rows'][1:]
| c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',connect_timeout=10,read_timeout=45,cursorclass=pymysql.cursors.DictCursor)
| def clean(v):return re.sub(r'\s+','',v or '').replace('（','(').replace('）',')')
| def basename(v):
| v=clean(v)
| return re.sub(r'[-(]?\d+(?:\.\d+)?(?:kg|g|公斤|克|ml|mL|L)(?:[/／](?:袋|箱|桶|瓶|包))?\)?$','',v,flags=re.I)
| def doccode(v):
| m=re.search(r'(YL|TJ)[)）]\s*(\d{3})',v);assert m,v
| return 'STP-QA-'+m[1]+'-'+m[2],m[1],m[2]
| log=[];mapping=[];newm=[];newt=[];links=0
| def mutate(q,sql,args):
| log.append(q.mogrify(sql,args)+';');q.execute(sql,args)
| try:
| with c.cursor() as q:
| q.execute('SELECT DATABASE() db');assert q.fetchone()['db']=='mom-hm'
| q.execute("SELECT ENGINE AS e FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name IN ('mdm_item','mes_qc_template','mes_qc_template_item')");engines=q.fetchall();assert len(engines)==3 and all(x['e']=='InnoDB' for x in engines)
| c.begin();q.execute("SELECT GET_LOCK('iqc-supplement-20261012',10) ok");assert q.fetchone()['ok']==1
| q.execute('SELECT * FROM mdm_item WHERE deleted=0 FOR UPDATE');materials=list(q.fetchall())
| q.execute('SELECT * FROM mes_qc_template WHERE deleted=0 FOR UPDATE');templates=list(q.fetchall())
| q.execute("SELECT id FROM mdm_unit_measure WHERE code='KG' AND deleted=0");kg=q.fetchone()['id']
| for r in src:
| code,name,ic,vendor,batch,tn,*rest=r
| product=name.removesuffix('来料检验')
| matches=[x for x in materials if x['code']==ic]
| if not matches:matches=[x for x in materials if basename(x['name'])==clean(product)]
| assert len(matches)<=1,('Ambiguous material',ic,product)
| action='编码匹配'
| if matches:
| m=matches[0];action='复用名称/名称加规格' if m['code']!=ic else action
| else:
| assert product in ['丹参粉','山梨糖醇'],('Unexpected new material',product)
| q.execute('SELECT id FROM mdm_item WHERE code=%s',(ic,));assert not q.fetchone(),'Deleted code conflict'
| spec=re.search(r'规格：([^；]+)',r[13] or '')
| mutate(q,"INSERT INTO mdm_item(code,name,specification,unit_measure_id,item_type,status,is_batch_managed,remark,creator,updater) VALUES (%s,%s,%s,%s,1,0,b'1',%s,'1','1')",(ic,product,spec[1] if spec else None,kg,'来源：来料检验导入表；已核对现有名称及名称加规格，无相同物料。'))
| m={'id':q.lastrowid,'code':ic,'name':product,'specification':spec[1] if spec else None,'status':0};materials.append(m);newm.append(m);action='新增'
| assert m['status']==0,('Disabled material',m['code'])
| tc,kind,num=doccode(tn)
| # Reuse only a compatible material name and the document number stated in the source.
| ts=[x for x in templates if x['code']==tc]
| if not ts and kind=='YL':
| candidates=[x for x in templates if x['code']=='SOP-QC-YL-'+num]
| if candidates:
| t=candidates[0];base=re.sub(r'原料质检方案.*$','',clean(t['name']))
| if base==clean(product) or (product=='红豆粉' and base=='红豆粉(熟)'):ts=[t]
| if not ts:
| q.execute('SELECT id FROM mes_qc_template WHERE code=%s',(tc,));assert not q.fetchone(),'Deleted template conflict'
| mutate(q,"INSERT INTO mes_qc_template(code,name,types,status,remark,creator,updater) VALUES (%s,%s,'1',1,%s,'1','1')",(tc,tn,'来源：来料检验导入表所列质量标准；未提供标准正文，待补充检验指标及复核后启用。'))
| t={'id':q.lastrowid,'code':tc,'name':tn,'status':1,'types':'1'};templates.append(t);newt.append(t)
| else:assert len(ts)==1;t=ts[0]
| q.execute('SELECT id FROM mes_qc_template_item WHERE template_id=%s AND item_id=%s AND deleted=0',(t['id'],m['id']))
| if not q.fetchone():
| mutate(q,"INSERT INTO mes_qc_template_item(template_id,item_id,remark,creator,updater) VALUES (%s,%s,%s,'1','1')",(t['id'],m['id'],'来源：来料检验导入表；按物料编码/名称加规格及标准编号匹配。'));links+=1
| mapping.append({'row_code':code,'source_item_code':ic,'item_code':m['code'],'item_id':m['id'],'item_name':m['name'],'material_action':action,'source_template':tn,'template_id':t['id'],'template_code':t['code'],'template_name':t['name'],'template_status':t['status']})
| assert len(newm)==2 and len({x['code'] for x in newm})==2
| for m in newm:
| q.execute('SELECT COUNT(*) n FROM mdm_item WHERE code=%s AND deleted=0',(m['code'],));assert q.fetchone()['n']==1
| for t in newt:
| q.execute('SELECT status FROM mes_qc_template WHERE id=%s',(t['id'],));assert q.fetchone()['status']==1
| (P/'supplement-executed.sql').write_text('START TRANSACTION;\n'+'\n'.join(log)+'\nCOMMIT;\n',encoding='utf8')
| c.commit();q.execute("SELECT RELEASE_LOCK('iqc-supplement-20261012')")
| except Exception:c.rollback();raise
| finally:c.close()
| (P/'supplement-mapping.json').write_text(json.dumps(mapping,ensure_ascii=False,indent=2),encoding='utf8')
| with (P/'supplement-review.csv').open('w',encoding='utf-8-sig',newline='') as f:
| w=csv.writer(f);w.writerow(['单号','源物料编码','实际物料编码','物料名称','物料处理','原标准','关联方案编码','方案名称','方案状态'])
| for x in mapping:w.writerow([x['row_code'],x['source_item_code'],x['item_code'],x['item_name'],x['material_action'],x['source_template'],x['template_code'],x['template_name'],'启用' if x['template_status']==0 else '停用待补充标准'])
| result={'new_materials':newm,'reused_missing_code_materials':len({x['source_item_code'] for x in mapping if x['material_action']=='复用名称/名称加规格'}),'new_disabled_templates':len(newt),'new_material_template_links':links,'verified':True}
| (P/'supplement-result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8');print(json.dumps(result,ensure_ascii=False))
|
|
