"""Atomic native parameter/template/workstation supplementation; preserve production settings."""
|
import json,sys,collections
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
p=json.loads((P/'parameter-station-plan.json').read_text(encoding='utf8'))
|
snap=json.loads((P/'before-parameter-snapshot.json').read_text(encoding='utf8'))
|
ss=json.loads((P/'before-station-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 norm(v):return json.loads(json.dumps(v,default=str))
|
def flag(v):return int.from_bytes(v,'big') if isinstance(v,bytes) else int(v 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=[];newunits=[];newparams=[];newtemplates=[];newstations=[];newbarcodes=[]
|
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=['mdm_unit_measure','mes_pd_process_param','mes_pd_process_param_template','mes_pd_process_param_template_detail','mes_pro_process_param','mes_pro_route_process','mes_md_workshop','mes_md_workstation']
|
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())
|
expected=snap.get(t,ss.get(t));assert norm(before[t])==expected,f'{t} changed; review required'
|
dump('before-parameter-station-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
|
def many(t,values):
|
if not values:return
|
cols=list(values[0]);sql='INSERT INTO '+t+' (`'+'`,`'.join(cols)+"`,creator,updater) VALUES ("+','.join(['%s']*len(cols))+",'1','1')"
|
args=[tuple(v[k] for k in cols) for v in values];log.extend(q.mogrify(sql,a)+';' for a in args);q.executemany(sql,args)
|
unitmap={u['code']:u['id'] for u in before['mdm_unit_measure']}
|
for code,name in p['units']:
|
if code not in unitmap:
|
assert not any(u['name'].strip()==name for u in before['mdm_unit_measure'])
|
unitmap[code]=insert('mdm_unit_measure',dict(code=code,name=name,primary_flag=0,primary_id=None,change_rate=1,status=0,remark='生产工艺参数记录单位；单位间未设换算关系。'));newunits.append(unitmap[code])
|
q.execute("SELECT value FROM system_dict_data WHERE deleted=0 AND status=0 AND dict_type='mes_pd_param_type'");assert '1' in {r['value'] for r in q.fetchall()}
|
q.execute("SELECT value FROM system_dict_data WHERE deleted=0 AND status=0 AND dict_type='mes_pd_template_type'");assert {'1','2'}<={r['value'] for r in q.fetchall()}
|
parammap={};existing={r['param_name']:r for r in before['mes_pd_process_param']}
|
for d in p['parameters']:
|
v=dict(param_code=d['param_code'],param_name=d['param_name'],param_type=1,unit_measure_id=unitmap[d['unit_code']],required=0,remark=d['remark'])
|
if d['param_name'] in existing:
|
old=existing[d['param_name']];assert old['param_type']==1 and old['unit_measure_id']==v['unit_measure_id'];pid=old['id']
|
else:pid=insert('mes_pd_process_param',v);newparams.append(pid)
|
d['id']=pid;d['unit_measure_id']=v['unit_measure_id'];parammap[d['key']]=d
|
templatemap={}
|
for t in p['templates']:
|
notes='；'.join(t['notes']);sources='、'.join(t['source_codes'])
|
remark=('来源：'+sources+'；工序记录模板；标准在明细备注，记录现场实际值，不自动判定超限。'+('；'+notes if notes else ''))[:500]
|
t['id']=insert('mes_pd_process_param_template',dict(template_code=t['template_code'],template_name=t['template_name'],template_type=t['template_type'],remark=remark));newtemplates.append(t['id']);templatemap[t['key']]=t['id']
|
details=[]
|
for d in t['details']:
|
param=parammap[d['param_key']]
|
standard=d['standard'] or '原文未指定统一数值，记录实际值；不得套用其他产品标准。'
|
rem=('要求：'+standard+'；'+d['note']+'；来源：'+sources+'；实际值由现场录入。')[:500]
|
details.append(dict(template_id=t['id'],process_param_id=param['id'],param_code=param['param_code'],param_name=param['param_name'],param_type=1,unit_measure_id=param['unit_measure_id'],required=int(d['required']),remark=rem))
|
many('mes_pd_process_param_template_detail',details)
|
many('mes_pro_process_param',[dict(process_id=b['process_id'],process_param_id=parammap[b['param_key']]['id'],required=0,sort=b['sort'],remark=b['remark']) for b in p['process_parameter_bindings']])
|
for b in p['route_template_bindings']:
|
row=next(r for r in before['mes_pro_route_process'] if r['route_id']==b['route_id'] and r['process_id']==b['process_id'])
|
assert row['param_template_id'] is None
|
b['param_template_id']=templatemap[b['template_key']]
|
execute("UPDATE mes_pro_route_process SET param_template_id=%s,updater='1' WHERE id=%s",(b['param_template_id'],row['id']))
|
# Use native virtual line-side hierarchy. No equipment/worker assignments are invented.
|
q.execute('SELECT * FROM mes_wm_warehouse WHERE id=23 AND deleted=0');assert q.fetchone()['code']=='WIP_VIRTUAL_WAREHOUSE'
|
q.execute('SELECT * FROM mes_wm_warehouse_location WHERE id=8 AND deleted=0');loc=q.fetchone();assert loc['warehouse_id']==23 and loc['code']=='WIP_VIRTUAL_LOCATION'
|
q.execute('SELECT * FROM mes_wm_warehouse_area WHERE id=8 AND deleted=0');area=q.fetchone();assert area['location_id']==8 and area['code']=='WIP_VIRTUAL_AREA'
|
q.execute('SELECT * FROM mes_wm_barcode_config WHERE biz_type=602 AND deleted=0');configs=list(q.fetchall());assert len(configs)<=1
|
config=configs[0] if configs else None
|
for st in p['stations']:
|
old=[r for r in before['mes_md_workstation'] if r['process_id']==st['process_id']]
|
assert len(old)<=1
|
if old:st['id']=old[0]['id']
|
else:
|
st['id']=insert('mes_md_workstation',st);newstations.append(st['id'])
|
if config and config['status']==0 and flag(config['auto_generate_flag']):
|
content=(config['content_format'] or '{BUSINESSCODE}').replace('{BUSINESSCODE}',st['code'])
|
q.execute('SELECT COUNT(*) n FROM mes_wm_barcode WHERE content=%s AND deleted=0',(content,));assert q.fetchone()['n']==0
|
newbarcodes.append(insert('mes_wm_barcode',dict(config_id=config['id'],format=config['format'],biz_type=602,content=content,biz_id=st['id'],biz_code=st['code'],biz_name=st['name'],status=0,remark='新增工作站按已有自动条码配置生成。')))
|
# Verify coverage, native snapshots, unit foreign keys and untouched route production flags.
|
q.execute('SELECT * FROM mes_md_workstation WHERE deleted=0');stations=list(q.fetchall())
|
assert len(stations)==46 and len({st['process_id'] for st in stations})==46
|
assert all(st['status']==0 and st['warehouse_id']==23 and st['location_id']==8 and st['area_id']==8 and st['workshop_id'] in {w['id'] for w in before['mes_md_workshop']} for st in stations)
|
q.execute('SELECT * FROM mes_pro_route_process WHERE deleted=0 ORDER BY id');actual=list(q.fetchall())
|
assert len(actual)==len(before['mes_pro_route_process'])
|
expected_bind={(b['route_id'],b['process_id']):b['param_template_id'] for b in p['route_template_bindings']}
|
for row,old in zip(actual,before['mes_pro_route_process']):
|
assert row['id']==old['id']
|
assert row['param_template_id']==expected_bind.get((row['route_id'],row['process_id']),old['param_template_id'])
|
for k in ['key_flag','check_flag','backflush_flag','output_item_id','sort','next_process_id','remark']:assert row[k]==old[k]
|
q.execute('SELECT * FROM mes_pd_process_param_template_detail WHERE deleted=0');details=list(q.fetchall())
|
assert len(details)==sum(len(t['details']) for t in p['templates'])
|
assert all(d['process_param_id'] in newparams and d['template_id'] in newtemplates and d['unit_measure_id'] in set(unitmap.values()) and d['param_type']==1 for d in details)
|
q.execute('SELECT * FROM mes_pro_process_param WHERE deleted=0');links=list(q.fetchall())
|
assert len(links)==len(p['process_parameter_bindings']) and len({(l['process_id'],l['process_param_id']) for l in links})==len(links)
|
q.execute('SELECT * FROM mes_md_workshop WHERE deleted=0 ORDER BY id');assert list(q.fetchall())==before['mes_md_workshop']
|
result={'database':'mom-hm','new_units':len(newunits),'new_parameters':len(newparams),'new_templates':len(newtemplates),'template_details':len(details),'route_process_template_bindings':len(expected_bind),'process_parameter_bindings':len(links),'new_workstations':len(newstations),'workstation_barcodes':len(newbarcodes),'one_station_per_process':True,'production_flags_and_outputs_preserved':True,'actual_production_values_not_prefilled':True,'existing_workshops_preserved':True,'verified':True}
|
c.commit();dump('parameter-station-import-result.json',result);dump('parameter-station-plan-with-ids.json',p)
|
(P/'parameter-station-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()
|
