"""Frozen name mapping, optional customer, validated draft OQC import only. Default: local review; --apply: locked transaction and committed verification. Password is read from stdin, never written to audit files. """ import sys, json, csv, collections, datetime from decimal import Decimal 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'] snapshot = json.loads((P/'remote-snapshot.json').read_text(encoding='utf8')) saved = json.loads((P/'random-selections.json').read_text(encoding='utf8')) assert len(rows)==496 def build(s): templates = {t['id']:t for t in s['mes_qc_template']} items = {t['code']:t for t in s['mdm_item']} users = {t['nickname'].strip():t['id'] for t in s['system_users'] if t['status']==0} results = {t['label']:int(t['value']) for t in s['system_dict_data'] if t['dict_type']=='mes_qc_check_result' and t['status']==0} bindings = {(t['template_id'],t['item_id']) for t in s['mes_qc_template_item']} indicators = collections.defaultdict(list) for t in s['mes_qc_template_indicator']: indicators[t['template_id']].append(t) codes = collections.Counter(r['source']['检验单编号'] for r in rows) review = [] for x in rows: r=x['source']; tid=x['mapping']['selected_id']; issues=[] assert saved['oqc|'+r['检验单名称']]['selected_id']==tid t=templates.get(tid); item=items.get(r['产品物料编码']); code=r['检验单编号'] if not code:issues.append('检验单编号为空') elif codes[code]>1:issues.append('原表检验单编号重复，未擅自改号') if not item:issues.append('物料编码为空或系统不存在') elif item['status']!=0:issues.append('物料已停用') if not t:issues.append('无已保存的名称匹配方案') elif t['status']!=0:issues.append('质检方案停用，暂缓导入') elif not indicators[tid]:issues.append('质检方案缺少检验指标') if t and item and (tid,item['id']) not in bindings:issues.append('产品未绑定所选方案') quantities={} for key,col in [('out_quantity','本次出货数量'),('check_quantity','检测数量'),('qualified_quantity','合格品数量'),('unqualified_quantity','不合格品数量')]: value=r[col] if value is None:issues.append(col+'为空');quantities[key]=None else: try: value=Decimal(str(value));assert value.is_finite() and value>=0 quantities[key]=value except Exception:issues.append(col+'非合法非负数');quantities[key]=None if all(quantities[k] is not None for k in ['check_quantity','qualified_quantity','unqualified_quantity']): if quantities['check_quantity'] != quantities['qualified_quantity']+quantities['unqualified_quantity']:issues.append('合格与不合格数量合计不等于检测数量') dates={} for key,col in [('out_date','出货日期'),('inspect_date','检测日期')]: try:dates[key]=datetime.datetime.fromisoformat(r[col]).strftime('%Y-%m-%d %H:%M:%S') except Exception:issues.append(col+'为空或格式无效');dates[key]=None inspector=users.get((r['检测人员'] or '').strip()) if r['检测人员'] and not inspector:issues.append('检测人员未找到') result=results.get(r['检测结果']) if r['检测结果'] and result is None:issues.append('检测结果未匹配字典') header=dict(code=code,name=r['检验单名称'],template_id=tid,item_id=item['id'] if item else None, client_id=None,batch_code=r['批次号'],inspector_user_id=inspector,check_result=result, status=0,remark=r['备注'],**quantities,**dates) for key,size in [('code',100),('name',100),('batch_code',100),('remark',255)]: if header[key] is not None and len(str(header[key]))>size:issues.append(key+'超出字段长度') review.append(dict(excel_row=x['excel_row'],header=header,template_name=t['name'] if t else None, item_name=item['name'] if item else None,issues=issues)) keys=collections.defaultdict(list) for x in review: if not x['issues']: h=x['header'];keys[(h['item_id'],h['client_id'],h['batch_code'],h['out_date'])].append(x) for group in keys.values(): if len(group)>1: for x in group:x['issues'].append('原表物料、客户、批次、出货日期重复') return review def dump(name,value): (P/name).write_text(json.dumps(value,ensure_ascii=False,default=str,indent=2),encoding='utf8') review=build(snapshot) dump('oqc-import-review.json',review) summary={'source_rows':len(review),'eligible':sum(not x['issues'] for x in review),'pending':sum(bool(x['issues']) for x in review), 'reasons':dict(collections.Counter(i for x in review for i in x['issues']))} print(json.dumps(summary,ensure_ascii=False)) if '--apply' not in sys.argv:sys.exit() 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=60) log=[];inserted=[];reused=[];changes=[] try: with c.cursor() as q: q.execute('SELECT DATABASE() db');assert q.fetchone()['db']=='mom-hm' q.execute("SELECT GET_LOCK('oqc-import-20261012',10) ok");assert q.fetchone()['ok']==1 c.begin();live={} for table in ['mes_qc_oqc','mes_qc_oqc_line','mes_qc_iqc','mes_qc_ipqc','mes_qc_ipqc_line','mes_qc_template','mes_qc_template_indicator','mes_qc_template_item','mes_qc_indicator','mdm_item','system_users','system_dict_data']: q.execute('SELECT ENGINE FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=%s',(table,));assert q.fetchone()['ENGINE']=='InnoDB' q.execute('SELECT * FROM '+table+' WHERE deleted=0 FOR UPDATE');live[table]=list(q.fetchall()) dump('before-oqc-import.json',live) current=build(live) assert [(x['excel_row'],x['issues']) for x in current]==[(x['excel_row'],x['issues']) for x in review], 'Live validation changed; regenerate review' original_templates=[dict(t) for t in live['mes_qc_template']] templates={t['id']:t for t in live['mes_qc_template']};inds={t['id']:t for t in live['mes_qc_indicator']} criteria=collections.defaultdict(list) for t in live['mes_qc_template_indicator']:criteria[t['template_id']].append(t) def execute(sql,args): log.append(q.mogrify(sql,args)+';');q.execute(sql,args) for x in current: if x['issues']:continue h=x['header'];tid=h['template_id'] q.execute('SELECT * FROM mes_qc_oqc WHERE code=%s AND deleted=0',(h['code'],));existing=q.fetchall() if existing: assert len(existing)==1 e=existing[0] for key,val in h.items(): actual=e[key] if isinstance(actual,datetime.datetime):actual=actual.strftime('%Y-%m-%d %H:%M:%S') assert actual==val, 'Existing code has different data: '+h['code']+' '+key x['id']=e['id'];x['action']='已存在，复用';reused.append(x);continue q.execute('SELECT id FROM mes_qc_oqc WHERE item_id=%s AND client_id IS NULL AND batch_code <=> %s AND out_date=%s AND deleted=0',(h['item_id'],h['batch_code'],h['out_date'])) if q.fetchone():x['issues'].append('系统已存在同物料、空客户、批次和出货日期的检验单');continue t=templates[tid];types=(t['types'] or '').split(',') if '3' not in types: newtypes=','.join(types+['3']) execute("UPDATE mes_qc_template SET types=%s,updater='1' WHERE id=%s AND status=0 AND deleted=0",(newtypes,tid)) assert q.rowcount==1;changes.append(dict(id=tid,name=t['name'],before=t['types'],after=newtypes));t['types']=newtypes cols=list(h) execute('INSERT INTO mes_qc_oqc (`'+'`,`'.join(cols)+"`,creator,updater) VALUES ("+','.join(['%s']*len(cols))+",'1','1')",tuple(h[k] for k in cols)) oid=q.lastrowid;x['id']=oid;x['action']='新增草稿' for ti in criteria[tid]: indicator=inds.get(ti['indicator_id']) line=dict(oqc_id=oid,indicator_id=ti['indicator_id'],tool=indicator['tool'] if indicator else None, check_method=ti['check_method'],standard_value=ti['standard_value'],unit_measure_id=ti['unit_measure_id'], max_threshold=ti['threshold_max'],min_threshold=ti['threshold_min'],critical_quantity=0,major_quantity=0,minor_quantity=0) lc=list(line) execute('INSERT INTO mes_qc_oqc_line (`'+'`,`'.join(lc)+"`,creator,updater) VALUES ("+','.join(['%s']*len(lc))+",'1','1')",tuple(line[k] for k in lc)) x['lines']=len(criteria[tid]);inserted.append(x) for table in ['mes_qc_iqc','mes_qc_ipqc','mes_qc_ipqc_line']: q.execute('SELECT * FROM '+table+' WHERE deleted=0 ORDER BY id');assert list(q.fetchall())==sorted(live[table],key=lambda x:x['id']) q.execute('SELECT id,status,types FROM mes_qc_template WHERE deleted=0');after={x['id']:x for x in q.fetchall()} changed={x['id']:x for x in changes} for t in original_templates: assert after[t['id']]['status']==t['status'] assert after[t['id']]['types']==(changed[t['id']]['after'] if t['id'] in changed else t['types']) (P/'oqc-import-executed.sql').write_text('START TRANSACTION;\n'+'\n'.join(log)+'\nCOMMIT;\n',encoding='utf8') c.commit() for x in inserted+reused: q.execute('SELECT * FROM mes_qc_oqc WHERE id=%s',(x['id'],));e=q.fetchone() for key,val in x['header'].items(): actual=e[key] if isinstance(actual,datetime.datetime):actual=actual.strftime('%Y-%m-%d %H:%M:%S') assert actual==val,(x['id'],key) q.execute('SELECT COUNT(*) n FROM mes_qc_oqc_line WHERE oqc_id=%s AND deleted=0',(x['id'],));assert q.fetchone()['n']==len(criteria[x['header']['template_id']]) q.execute('SELECT COUNT(*) n FROM mes_qc_oqc WHERE deleted=0');assert q.fetchone()['n']==len(live['mes_qc_oqc'])+len(inserted) result=dict(target='36.213.150.237:10000/mom-hm',source_rows=496,inserted=len(inserted),reused=len(reused),pending=sum(bool(x['issues']) for x in current), inserted_lines=sum(x['lines'] for x in inserted),template_type_changes=changes,customer=None,status='草稿',post_commit_verified=True, reasons=dict(collections.Counter(i for x in current for i in x['issues']))) dump('oqc-import-result.json',result);dump('oqc-import-review.json',current) with (P/'出货检验-录入结果及暂缓清单.csv').open('w',encoding='utf-8-sig',newline='') as f: w=csv.writer(f);w.writerow(['原表行号','检验单编号','检验单名称','系统ID','方案ID','方案名称','物料ID','物料名称','客户ID','处理结果','暂缓原因']) for x in current: h=x['header'];w.writerow([x['excel_row'],h['code'],h['name'],x.get('id'),h['template_id'],x['template_name'],h['item_id'],x['item_name'],'',x.get('action','暂缓导入'),'；'.join(x['issues'])]) print(json.dumps(result,ensure_ascii=False)) q.execute("SELECT RELEASE_LOCK('oqc-import-20261012')") except Exception:c.rollback();raise finally:c.close()