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
| import sys,json,csv
| from pathlib import Path
| from datetime import datetime,timedelta
| from decimal import Decimal
| sys.path.insert(0,str(Path(__file__).parent/'deps'))
| import pymysql
| P=Path(__file__).parent
| plan=json.loads((P/'plan.json').read_text(encoding='utf8'))
| ready=[x for x in plan if not x['issues'] and x['code']!='Y202608006']
| skipped=[x for x in plan if x not in ready]
| assert ready and len(ready)+len(skipped)==59
| 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)
| sql_log=[]
| def execute(q,sql,args=None):
| sql_log.append(q.mogrify(sql,args)+';')
| q.execute(sql,args)
| def same(a,b):
| if isinstance(a,Decimal):return a==Decimal(str(b))
| if isinstance(a,datetime):return str(a)==b
| return a==b
| columns=['code','name','template_id','vendor_name','vendor_batch','item_id','received_quantity','check_quantity','qualified_quantity','unqualified_quantity','check_result','receive_date','inspect_date','inspector_user_id','remark','status']
| inserted=[];existing=[];lines=0
| try:
| with c.cursor() as q:
| q.execute('SELECT DATABASE() db');assert q.fetchone()['db']=='mom-hm'
| q.execute("SELECT engine AS table_engine FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name IN ('mes_qc_iqc','mes_qc_iqc_line')")
| engines=q.fetchall();assert len(engines)==2 and all(x['table_engine']=='InnoDB' for x in engines)
| c.begin()
| q.execute("SELECT GET_LOCK('iqc-import-20261012',10) acquired");assert q.fetchone()['acquired']==1
| for x in ready:
| assert datetime.fromisoformat(x['receive_date'])==datetime.fromisoformat(x['inspect_date'])-timedelta(days=15)
| q.execute('SELECT id,code,status FROM mdm_item WHERE id=%s AND deleted=0 FOR UPDATE',(x['item_id'],));m=q.fetchone()
| assert m and m['code']==x['item_code'] and m['status']==0,'Material changed'
| q.execute('SELECT id,name,status,types FROM mes_qc_template WHERE id=%s AND deleted=0 FOR UPDATE',(x['template_id'],));t=q.fetchone()
| assert t and t['name']==x['template_name'] and t['status']==0 and '1' in t['types'].split(','),'Template changed'
| q.execute('SELECT id FROM mes_qc_template_item WHERE template_id=%s AND item_id=%s AND deleted=0 FOR UPDATE',(x['template_id'],x['item_id']));assert q.fetchall(),'Missing template relation'
| q.execute('SELECT id,status FROM system_users WHERE id=%s AND deleted=0',(x['inspector_user_id'],));u=q.fetchone();assert u and u['status']==0,'Inspector unavailable'
| q.execute('SELECT * FROM mes_qc_iqc WHERE code=%s FOR UPDATE',(x['code'],));old=q.fetchall()
| if old:
| assert len(old)==1 and old[0]['deleted']==b'\x00' and all(same(old[0][k],x[k]) for k in columns),'Existing code has different content'
| existing.append(old[0]['id']);continue
| q.execute('SELECT id FROM mes_qc_iqc WHERE item_id=%s AND COALESCE(vendor_name,\'\')=%s AND vendor_batch <=> %s AND receive_date=%s AND deleted=0 FOR UPDATE',(x['item_id'],x['vendor_name'],x['vendor_batch'],x['receive_date']));assert not q.fetchall(),'Existing import key'
| execute(q,'INSERT INTO mes_qc_iqc ('+','.join(columns)+",creator,updater) VALUES ("+','.join(['%s']*len(columns))+",'1','1')",tuple(x[k] for k in columns))
| iqc_id=q.lastrowid;inserted.append(iqc_id)
| execute(q,"INSERT INTO mes_qc_iqc_line (iqc_id,indicator_id,tool,check_method,standard_value,unit_measure_id,max_threshold,min_threshold,critical_quantity,major_quantity,minor_quantity,creator,updater) SELECT %s,ti.indicator_id,i.tool,ti.check_method,ti.standard_value,ti.unit_measure_id,ti.threshold_max,ti.threshold_min,0,0,0,'1','1' FROM mes_qc_template_indicator ti LEFT JOIN mes_qc_indicator i ON i.id=ti.indicator_id AND i.deleted=0 WHERE ti.template_id=%s AND ti.deleted=0",(iqc_id,x['template_id']))
| lines+=q.rowcount
| q.execute('SELECT * FROM mes_qc_iqc WHERE id=%s',(iqc_id,));saved=q.fetchone()
| assert all(same(saved[k],x[k]) for k in columns),'Content verification failed'
| q.execute('SELECT COUNT(*) n FROM mes_qc_template_indicator WHERE template_id=%s AND deleted=0',(x['template_id'],));expected=q.fetchone()['n']
| q.execute('SELECT COUNT(*) n FROM mes_qc_iqc_line WHERE iqc_id=%s AND deleted=0',(iqc_id,));assert q.fetchone()['n']==expected,'Line count mismatch'
| (P/'executed-import.sql').write_text('START TRANSACTION;\n'+'\n'.join(sql_log)+'\nCOMMIT;\n',encoding='utf8')
| c.commit()
| q.execute('SELECT * FROM mes_qc_iqc WHERE code IN ('+','.join(['%s']*len(ready))+') AND deleted=0',tuple(x['code'] for x in ready));post=q.fetchall()
| assert len(post)==len(ready)
| for x in ready:
| saved=next(z for z in post if z['code']==x['code']);assert all(same(saved[k],x[k]) for k in columns)
| (P/'post-import.json').write_text(json.dumps(post,default=str,ensure_ascii=False,indent=2),encoding='utf8')
| q.execute("SELECT RELEASE_LOCK('iqc-import-20261012')")
| except Exception:
| c.rollback();raise
| finally:c.close()
| with (P/'skipped-rows.csv').open('w',encoding='utf-8-sig',newline='') as f:
| q=csv.writer(f);q.writerow(['Excel行号','检验单编号','名称','物料编码','质检方案','检测日期','计划来料日期','跳过原因'])
| for x in skipped:q.writerow([x['row'],x['code'],x['name'],x['item_code'],x['source_template'],x['inspect_date'],x['receive_date'],'；'.join(x['issues'])])
| result={'target':'36.213.150.237:10000/mom-hm','source_rows':len(plan),'inserted':len(inserted),'already_present':len(existing),'inserted_lines':lines,'skipped':len(skipped),'inserted_ids':inserted,'verified':True,'date_rule':'检测日期前15天'}
| (P/'import-result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
| print(json.dumps(result,ensure_ascii=False))
|
|
