"""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()
|
