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
| import json, hashlib, collections, csv, re, sys
| from pathlib import Path
| sys.stdout.reconfigure(encoding='utf-8')
| root=Path('scripts/sop-qc')
| plan=json.loads((root/'inspection-plan.json').read_text(encoding='utf-8'))
| def quote(v):
| if v is None:return 'NULL'
| if isinstance(v,int):return str(v)
| return "CONVERT(0x"+str(v).encode('utf-8').hex()+" USING utf8mb4)" if v else "''"
| def code(name):return 'SOP-QC-YL-I-'+hashlib.sha256(name.encode()).hexdigest()[:16]
| groups=['感官','理化','成分','微生物','污染物','农兽药残留']
| types={'感官':2,'理化':4,'成分':5,'微生物':4,'污染物':5,'农兽药残留':5}
| indicators={g:{'code':'SOP-QC-YL-G-'+str(n+1),'name':'SOP原料检验-'+g,'parent_code':None,'item_type':2,'type':types[g],'result_type':None,'sort_order':n*10,'remark':'来源：SOP-QC-YL；分组项'} for n,g in enumerate(groups)}
| rows=[];templates=[];materials=[];continuation_names=set()
| for s in plan['schemes']:
| review=s['review_flags']
| remark='来源：'+s['source_code']+'；按原文人工判定。'+('待复核：'+'；'.join(review) if review else '')
| templates.append([s['code'],s['name'],'1',s['status'],remark[:255]])
| seen_groups=set();order=0
| for item in s['items']:
| name=item['name'];group=item['group'];ic=code(name)
| indicators.setdefault(name,{'code':ic,'name':name,'parent_code':indicators[group]['code'],'item_type':1,'type':types[group],'result_type':3,'sort_order':len(indicators),'remark':'SOP原料检验；标准以方案条款为准，人工判定。'})
| if group not in seen_groups:
| rows.append([s['code'],indicators[group]['code'],order,0,'分组标题，不参与判定',None]);order+=10;seen_groups.add(group)
| requirement=item['requirement'] or '待核对原文判定标准，详见 '+s['source_code']+' 第'+item['section']+'条；不得据此自动判定合格。'
| method='来源：'+s['source_code']+' §'+item['section']+'；'+'；'.join(item['method_references'])
| chunks=[requirement[i:i+255] for i in range(0,len(requirement),255)]
| rows.append([s['code'],ic,order,1,method[:255],chunks[0]]);order+=10
| # The schema only allows 255 characters per requirement; continuation rows preserve
| # complete source text without truncation. They do not require a second test result.
| for n,chunk in enumerate(chunks[1:],start=2):
| cn=name+'（标准续文'+str(n)+'）';cc=code(cn);continuation_names.add(cn)
| indicators.setdefault(cn,{'code':cc,'name':cn,'parent_code':indicators[group]['code'],'item_type':1,'type':types[group],'result_type':3,'sort_order':len(indicators),'remark':'上一检验项的判定要求续文，不参与独立判定，无需重复检验。'})
| rows.append([s['code'],cc,order,0,'来源：'+s['source_code']+' §'+item['section']+'；判定要求续文，无需重复检验',chunk]);order+=10
| for m in s['remote_material_matches']:
| materials.append([s['code'],m['code'],m['name'],'精确名称匹配；抽样数及缺陷率原文未统一规定，留空待设置。'])
| columns={
| 'stage_sop_indicator':'code varchar(100),name varchar(100),parent_code varchar(100),item_type tinyint,type int,result_type int,sort_order int,remark varchar(255)',
| 'stage_sop_template':'code varchar(100),name varchar(100),types varchar(255),status int,remark varchar(255)',
| 'stage_sop_line':'template_code varchar(100),indicator_code varchar(100),sort_order int,judge_flag tinyint,check_method varchar(255),remark varchar(255)',
| 'stage_sop_item':'template_code varchar(100),item_code varchar(50),item_name varchar(100),remark varchar(255)'}
| data={ 'stage_sop_indicator':[[i[k] for k in ['code','name','parent_code','item_type','type','result_type','sort_order','remark']] for i in indicators.values()], 'stage_sop_template':templates,'stage_sop_line':rows,'stage_sop_item':materials}
| sql=['-- 来源 SOP-QC-YL；增量幂等导入，保留现有记录。须在同一连接的事务中执行。']
| for table,definition in columns.items():
| sql.append(f'CREATE TEMPORARY TABLE `mom-hm`.{table} ({definition}) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci')
| for offset in range(0,len(data[table]),100):
| sql.append(f'INSERT INTO `mom-hm`.{table} VALUES '+','.join('('+','.join(quote(v) for v in row)+')' for row in data[table][offset:offset+100]))
| audit="creator,create_time,updater,update_time,deleted"
| values="'1',NOW(),'1',NOW(),b'0'"
| sql.append(f"INSERT INTO `mom-hm`.mes_qc_indicator (code,name,parent_id,item_type,type,result_type,sort_order,remark,{audit}) SELECT s.code,s.name,0,s.item_type,s.type,s.result_type,s.sort_order,s.remark,{values} FROM `mom-hm`.stage_sop_indicator s WHERE s.parent_code IS NULL AND NOT EXISTS (SELECT 1 FROM `mom-hm`.mes_qc_indicator e WHERE e.code=s.code)")
| sql.append(f"INSERT INTO `mom-hm`.mes_qc_indicator (code,name,parent_id,item_type,type,result_type,sort_order,remark,{audit}) SELECT s.code,s.name,p.id,s.item_type,s.type,s.result_type,s.sort_order,s.remark,{values} FROM `mom-hm`.stage_sop_indicator s JOIN `mom-hm`.mes_qc_indicator p ON p.code=s.parent_code AND p.deleted=0 WHERE NOT EXISTS (SELECT 1 FROM `mom-hm`.mes_qc_indicator e WHERE e.code=s.code)")
| sql.append(f"INSERT INTO `mom-hm`.mes_qc_template (code,name,types,status,remark,{audit}) SELECT s.code,s.name,s.types,s.status,s.remark,{values} FROM `mom-hm`.stage_sop_template s WHERE NOT EXISTS (SELECT 1 FROM `mom-hm`.mes_qc_template e WHERE e.code=s.code)")
| sql.append(f"INSERT INTO `mom-hm`.mes_qc_template_indicator (template_id,indicator_id,sort_order,judge_flag,check_method,remark,{audit}) SELECT t.id,i.id,s.sort_order,s.judge_flag,s.check_method,s.remark,{values} FROM `mom-hm`.stage_sop_line s JOIN `mom-hm`.mes_qc_template t ON t.code=s.template_code AND t.deleted=0 JOIN `mom-hm`.mes_qc_indicator i ON i.code=s.indicator_code AND i.deleted=0 WHERE NOT EXISTS (SELECT 1 FROM `mom-hm`.mes_qc_template_indicator e WHERE e.template_id=t.id AND e.indicator_id=i.id)")
| sql.append(f"INSERT INTO `mom-hm`.mes_qc_template_item (template_id,item_id,remark,{audit}) SELECT t.id,m.id,s.remark,{values} FROM `mom-hm`.stage_sop_item s JOIN `mom-hm`.mes_qc_template t ON t.code=s.template_code AND t.deleted=0 JOIN `mom-hm`.mdm_item m ON m.code=s.item_code AND m.name=s.item_name AND m.deleted=0 AND m.status=0 WHERE NOT EXISTS (SELECT 1 FROM `mom-hm`.mes_qc_template_item e WHERE e.template_id=t.id AND e.item_id=m.id)")
| file=Path('sql/mysql/sop_qc_yl_import_20261012.sql');file.write_text(';\n\n'.join(sql)+';\n',encoding='utf-8')
| summary={'source_documents':len(templates),'inspection_items':len(indicators)-len(groups)-len(continuation_names),'groups':len(groups),'continuation_items':len(continuation_names),'scheme_lines':len(rows),'material_links':len(materials),'enabled_schemes':sum(s['status']==0 for s in plan['schemes']),'review_schemes':sum(s['status']==1 for s in plan['schemes']),'staging_counts':{t:len(rs) for t,rs in data.items()},'sql_file':str(file)}
| (root/'import-summary.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2),encoding='utf-8')
| with (root/'待复核方案清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
| w=csv.writer(f);w.writerow(['方案编号','名称','来源','待复核原因','物料匹配数'])
| for s in plan['schemes']:
| if s['review_flags'] or not s['remote_material_matches']:w.writerow([s['code'],s['name'],s['file'],'；'.join(s['review_flags']) or '系统无准确名称物料匹配',len(s['remote_material_matches'])])
| print(json.dumps(summary,ensure_ascii=False))
|
|
