"""Atomic import of reviewed route bindings, direct BOM inputs and safe outputs."""
|
import sys,json,csv,collections
|
from pathlib import Path
|
from decimal import Decimal
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
plan=json.loads((P/'binding-plan.json').read_text(encoding='utf8'))
|
snapshot=json.loads((P/'before-binding-snapshot.json').read_text(encoding='utf8'))
|
def dump(n,v):(P/n).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
def normalized(rows):return json.loads(json.dumps(rows,default=str))
|
def flag(value):return int.from_bytes(value,'big') if isinstance(value,bytes) else int(value or 0)
|
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=[];newroutes=[];updatedsteps=0
|
try:
|
with c.cursor() as q:
|
q.execute('SELECT DATABASE() db');assert q.fetchone()['db']=='mom-hm'
|
q.execute("SELECT GET_LOCK('stp-routes-20261012',10) ok");assert q.fetchone()['ok']==1
|
c.begin();before={}
|
tables=['mes_md_item','mes_md_product_bom','mes_pro_route','mes_pro_route_process','mes_pro_route_product','mes_pro_route_product_bom']
|
for t in tables:
|
q.execute('SELECT ENGINE FROM information_schema.tables WHERE table_schema=DATABASE() AND table_name=%s',(t,));assert q.fetchone()['ENGINE']=='InnoDB'
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id FOR UPDATE');before[t]=list(q.fetchall())
|
if normalized(before[t])!=sorted(snapshot[t],key=lambda x:x['id']):
|
dump('changed-'+t+'.json',before[t])
|
raise AssertionError(f'{t} changed; re-review required')
|
dump('before-binding-import.json',before)
|
def execute(sql,args):log.append(q.mogrify(sql,args)+';');q.execute(sql,args)
|
def insert(t,v):
|
cols=list(v);sql='INSERT INTO '+t+' (`'+'`,`'.join(cols)+"`,creator,updater) VALUES ("+','.join(['%s']*len(cols))+",'1','1')"
|
execute(sql,tuple(v[k] for k in cols));return q.lastrowid
|
rt={}
|
for r in plan['routes']:
|
if r['id'] is None:
|
assert not any(x['code']==r['code'] for x in before['mes_pro_route'])
|
r['id']=insert('mes_pro_route',{'code':r['code'],'name':r['name'],'description':r.get('description','来源：'+r['document_code']+'；按产品小包装规格细分产出；非倒冲、手动领料。'),'status':1,'remark':'产品名称、规格及BOM匹配；投入保留原BOM比例；原料BOM/参数未补齐前保持停用。'})
|
newroutes.append(r['id']);isnew=True
|
else:isnew=False
|
rt[r['route_key']]=r['id']
|
for idx,st in enumerate(r['steps']):
|
v=dict(route_id=r['id'],process_id=st['process_id'],sort=st['sort'],next_process_id=r['steps'][idx+1]['process_id'] if idx+1<len(r['steps']) else None,link_type=3,prepare_time=None,wait_time=None,key_flag=int(st['key_flag']),check_flag=int(st['check_flag']),backflush_flag=0,output_item_id=st['output_item_id'],remark=st['stored']['remark'])
|
if isnew:insert('mes_pro_route_process',v)
|
else:
|
old=next(x for x in before['mes_pro_route_process'] if x['route_id']==r['id'] and x['process_id']==st['process_id'])
|
assert flag(old['check_flag'])==v['check_flag'] and flag(old['backflush_flag'])==0
|
# Only production-count key and output change; retain source flags, parameters, ordering and text.
|
execute("UPDATE mes_pro_route_process SET key_flag=%s,output_item_id=%s,updater='1' WHERE id=%s",(v['key_flag'],v['output_item_id'],old['id']));updatedsteps+=1
|
r['persisted_product_ids']=r['product_ids']
|
for b in plan['bindings']:
|
b['route_id']=rt[b['route_key']]
|
b['binding_id']=insert('mes_pro_route_product',dict(route_id=b['route_id'],item_id=b['product_id'],quantity=None,production_time=None,time_unit_type=None,remark=('来源：'+b['document_code']+'；'+b['evidence']+'；非倒冲、手动领料')[:255]))
|
for a in plan['inputs']:
|
a['route_id']=rt[a['route_key']]
|
a['id']=insert('mes_pro_route_product_bom',dict(route_id=a['route_id'],process_id=a['process_id'],product_id=a['product_id'],item_id=a['item_id'],quantity=Decimal(a['quantity']),remark=('来源产品BOM行'+str(a['bom_id'])+'；保留原比例；'+a['reason']+'；手动领料、非倒冲')[:255]))
|
# Verify database relations and every original ratio before commit.
|
q.execute('SELECT * FROM mes_pro_route_product WHERE deleted=0');products=list(q.fetchall())
|
assert len(products)==len(before['mes_pro_route_product'])+len(plan['bindings'])
|
assert len({x['item_id'] for x in products})==len(products)
|
for old in before['mes_pro_route_product']:assert next(x for x in products if x['id']==old['id'])==old
|
q.execute('SELECT * FROM mes_pro_route_product_bom WHERE deleted=0');inputs=list(q.fetchall())
|
assert len(inputs)==len(plan['inputs'])
|
assert len({(x['route_id'],x['product_id'],x['item_id']) for x in inputs})==len(inputs)
|
for a in plan['inputs']:
|
row=next(x for x in inputs if x['id']==a['id'])
|
assert row['route_id']==a['route_id'] and row['process_id']==a['process_id'] and row['product_id']==a['product_id'] and row['item_id']==a['item_id'] and row['quantity']==Decimal(a['quantity'])
|
q.execute('SELECT * FROM mes_pro_route_process WHERE deleted=0');steps=list(q.fetchall())
|
for r in plan['routes']:
|
rows=sorted([x for x in steps if x['route_id']==r['id']],key=lambda x:x['sort'])
|
assert len(rows)==len(r['steps']) and sum(flag(x['key_flag']) for x in rows)==1
|
for actual,st in zip(rows,r['steps']):
|
assert actual['process_id']==st['process_id'] and actual['output_item_id']==st['output_item_id']
|
assert flag(actual['check_flag'])==int(st['check_flag']) and flag(actual['backflush_flag'])==0
|
assert not st['starred'] or flag(actual['check_flag'])==1
|
assert actual['output_item_id'] is None or (',' not in actual['output_item_id'] and int(actual['output_item_id']) in {x['id'] for x in before['mes_md_item']})
|
# Complete source CCPS remain unchanged, including original unbound routes.
|
for old in before['mes_pro_route_process']:
|
actual=next(x for x in steps if x['id']==old['id'])
|
assert actual['check_flag']==old['check_flag'] and actual['backflush_flag']==old['backflush_flag'] and actual['remark']==old['remark']
|
for t in ['mes_md_item','mes_md_product_bom']:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');assert list(q.fetchall())==before[t]
|
q.execute('SELECT * FROM mes_pro_route WHERE deleted=0');after_routes=list(q.fetchall())
|
for old in before['mes_pro_route']:assert next(x for x in after_routes if x['id']==old['id'])==old
|
assert all(next(x for x in after_routes if x['id']==r['id'])['status']==1 for r in plan['routes'])
|
result={'database':'mom-hm','products_bound':len(plan['bindings']),'routes_used':len(plan['routes']),'derived_routes_created':len(newroutes),'derived_route_ids':newroutes,'direct_bom_inputs':len(inputs),'configured_output_rows':len(plan['outputs']),'pending_unbound_products':len(plan['exceptions']),'products_without_bom':sum(b['bom_count']==0 for b in plan['bindings']),'packaging_only_products':sum(b['packaging_only'] for b in plan['bindings']),'original_ccp_flags_preserved':True,'all_backflush_zero':True,'existing_binding_484_preserved':True,'all_routes_disabled':True,'verified_before_commit':True}
|
c.commit()
|
dump('binding-import-result.json',result);dump('binding-plan-with-ids.json',plan)
|
(P/'binding-import-executed.sql').write_text('START TRANSACTION;\n'+'\n'.join(log)+'\nCOMMIT;\n',encoding='utf8')
|
print(json.dumps(result,ensure_ascii=False))
|
q.execute("SELECT RELEASE_LOCK('stp-routes-20261012')")
|
except:
|
c.rollback();raise
|
finally:c.close()
|
