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
| import sys,json,csv,hashlib
| from pathlib import Path
| P=Path(__file__).parent
| sys.path.insert(0,str(P/'deps'))
| import pymysql
| plan=json.loads((P/'national-standards-plan.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',connect_timeout=10,read_timeout=45,cursorclass=pymysql.cursors.DictCursor)
| log=[];before={};created=0;links=0;verified=[]
| 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 e FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name IN ('mes_qc_template','mes_qc_indicator','mes_qc_template_indicator')")
| engines=q.fetchall();assert len(engines)==3 and all(x['e']=='InnoDB' for x in engines)
| c.begin();q.execute("SELECT GET_LOCK('iqc-national-standards-20261012',10) ok");assert q.fetchone()['ok']==1
| for s in plan['schemes']:
| tid=s['template_id']
| q.execute('SELECT * FROM mes_qc_template WHERE id=%s AND deleted=0 FOR UPDATE',(tid,));t=q.fetchone()
| assert t and t['code'].startswith('STP-QA-TJ-') and s['product'] in t['name'],(tid,t)
| before[str(tid)]={'template':t}
| q.execute('SELECT * FROM mes_qc_template_indicator WHERE template_id=%s AND deleted=0 FOR UPDATE',(tid,));old=q.fetchall();before[str(tid)]['items']=old
| assert not old or len(old)==len(s['items']),('Unexpected template content',tid)
| for sort,i in enumerate(s['items']):
| key=json.dumps([i['name'],i['unit'],i['requirement']],ensure_ascii=False,separators=(',',':'))
| code='GB-IQC-'+hashlib.sha256(key.encode()).hexdigest()[:24]
| q.execute('SELECT * FROM mes_qc_indicator WHERE code=%s FOR UPDATE',(code,));found=q.fetchall();assert len(found)<=1
| if found:
| ind=found[0];assert ind['deleted'] in (0,b'\x00') and (ind['name'],ind['unit_text'],ind['result_specification'],ind['result_type'])==(i['name'],i['unit'],i['requirement'],3)
| iid=ind['id']
| else:
| assert len(i['name'])<=100 and len(i['unit'])<=50 and len(i['requirement'])<=255
| mutate(q,"INSERT INTO mes_qc_indicator(code,name,parent_id,item_type,unit_text,type,result_type,result_specification,remark,creator,updater) VALUES (%s,%s,0,1,%s,4,3,%s,%s,'1','1')",(code,i['name'],i['unit'],i['requirement'],'来源：食品安全国家标准；文本录入并人工判定，不自动生成实测结果。'))
| iid=q.lastrowid;created+=1
| method='要求：'+i['requirement']+(' '+i['unit'] if i['unit'] else '')+'；方法：'+i['method']
| remark=s['standard']+'；'+i['clause']+'；须按适用形态和检验日期人工判定，不默认合格。'
| assert len(method)<=255 and len(remark)<=255 and len(s['source_url'])<=255
| q.execute('SELECT * FROM mes_qc_template_indicator WHERE template_id=%s AND sort_order=%s AND deleted=0',(tid,sort*10));present=q.fetchall()
| if present:
| assert len(present)==1 and (present[0]['indicator_id'],present[0]['check_method'],present[0]['doc_url'])==(iid,method,s['source_url'])
| else:
| mutate(q,"INSERT INTO mes_qc_template_indicator(template_id,indicator_id,sort_order,judge_flag,check_method,doc_url,remark,creator,updater) VALUES (%s,%s,%s,1,%s,%s,%s,'1','1')",(tid,iid,sort*10,method,s['source_url'],remark));links+=1
| note='依据官方国标正文补充完整感官、理化及适用微生物项；条件项按形态/COA及检验日期人工判断。'+s['note']
| assert len(note)<=255
| mutate(q,"UPDATE mes_qc_template SET default_execute_standard=%s,status=%s,remark=%s,updater='1' WHERE id=%s AND deleted=0",(s['standard'],s['status'],note,tid))
| q.execute('SELECT COUNT(*) n FROM mes_qc_template_indicator WHERE template_id=%s AND deleted=0',(tid,));assert q.fetchone()['n']==len(s['items'])
| verified.append({'template_id':tid,'product':s['product'],'standard':s['standard'],'status':s['status'],'items':len(s['items'])})
| for x in plan['pending']:
| q.execute('SELECT * FROM mes_qc_template WHERE id=%s AND deleted=0 FOR UPDATE',(x['template_id'],));t=q.fetchone();assert t and t['status']==1
| before[str(x['template_id'])]={'template':t}
| q.execute('SELECT COUNT(*) n FROM mes_qc_template_indicator WHERE template_id=%s AND deleted=0',(x['template_id'],));assert q.fetchone()['n']==0
| mutate(q,"UPDATE mes_qc_template SET status=1,remark=%s,updater='1' WHERE id=%s",('国标补充核对：'+x['reason'],x['template_id']))
| (P/'before-national-standards.json').write_text(json.dumps(before,ensure_ascii=False,default=str,indent=2),encoding='utf8')
| (P/'national-standards-executed.sql').write_text('START TRANSACTION;\n'+'\n'.join(log)+'\nCOMMIT;\n',encoding='utf8')
| c.commit()
| # Re-read committed records, including manual text handling and source links.
| for s in plan['schemes']:
| q.execute('SELECT default_execute_standard,status FROM mes_qc_template WHERE id=%s',(s['template_id'],));t=q.fetchone();assert t=={'default_execute_standard':s['standard'],'status':s['status']}
| q.execute('SELECT ti.*,i.result_type,i.result_specification,i.name,i.unit_text FROM mes_qc_template_indicator ti JOIN mes_qc_indicator i ON i.id=ti.indicator_id WHERE ti.template_id=%s AND ti.deleted=0 ORDER BY ti.sort_order',(s['template_id'],))
| rows=q.fetchall();assert len(rows)==len(s['items'])
| for r,i in zip(rows,s['items']):
| assert r['result_type']==3 and r['standard_value'] is None and r['threshold_min'] is None and r['threshold_max'] is None
| assert (r['name'],r['unit_text'],r['result_specification'])==(i['name'],i['unit'],i['requirement'])
| q.execute("SELECT RELEASE_LOCK('iqc-national-standards-20261012')")
| result={'new_indicators':created,'new_template_items':links,'schemes':verified,'pending':plan['pending'],'post_commit_verified':True}
| (P/'national-standards-result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
| print(json.dumps(result,ensure_ascii=False))
| except Exception:
| c.rollback();raise
| finally:c.close()
|
|
