"""Import reviewed native MES process masters and disabled routes, no backflush."""
|
import sys,json,datetime,unicodedata,re,csv
|
from pathlib import Path
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
plan=json.loads((P/'route-plan.json').read_text(encoding='utf8'))
|
snapshot=json.loads((P/'remote-snapshot.json').read_text(encoding='utf8'))
|
assert len(plan['processes'])==46 and len(plan['routes'])==28
|
assert all(not st['backflush_flag'] and (not st['starred'] or st['check_flag']) for r in plan['routes'] for st in r['steps'])
|
assert all(sum(st['key_flag'] for st in r['steps'])==1 for r in plan['routes'])
|
def dump(n,v): (P/n).write_text(json.dumps(v,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
def normal(n):return re.sub(r'\s+','',unicodedata.normalize('NFKC',n))
|
def same(old,new):
|
for k,v in new.items():
|
actual=old[k]
|
if isinstance(actual,bytes):actual=int.from_bytes(actual,'big')
|
if actual!=v:return False
|
return True
|
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=[];newprocess=[];newroutes=[];newcontents=0;newsteps=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_pro_process','mes_pro_process_content','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())
|
dump('before-route-import.json',before)
|
for t in tables:assert [(x['id'],x.get('code')) for x in before[t]]==[(x['id'],x.get('code')) for x in sorted(snapshot[t],key=lambda x:x['id'])], 'Remote data changed; review again'
|
def insert(t,values):
|
cols=list(values);sql='INSERT INTO '+t+' (`'+'`,`'.join(cols)+"`,creator,updater) VALUES ("+','.join(['%s']*len(cols))+",'1','1')"
|
args=tuple(values[k] for k in cols);log.append(q.mogrify(sql,args)+';');q.execute(sql,args);return q.lastrowid
|
def insert_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)
|
processes={normal(x['name']):x for x in before['mes_pro_process']}
|
codes={x['code']:x for x in before['mes_pro_process']}
|
nameids={}
|
for m in plan['processes']:
|
existing=processes.get(normal(m['name']))
|
if existing:
|
m['id']=existing['id'];m['action']='复用现有工序'
|
else:
|
assert m['code'] not in codes,'Process code occupied'
|
m['id']=insert('mes_pro_process',dict(code=m['code'],name=m['name'],attention=m['attention'],status=0,
|
remark='来源：生产工艺规程工序去重；参数按原文产品和版本执行。'))
|
newprocess.append(m['id']);m['action']='新增工序'
|
contentrows=[]
|
for idx,ev in enumerate(m['contents'],1):
|
prefix='【'+ev['document_code']+'；段落'+str(ev['paragraph'])+'】'
|
raw=ev['text'];parts=[raw[i:i+500-len(prefix)] for i in range(0,len(raw),500-len(prefix))]
|
for part in parts:contentrows.append(dict(process_id=m['id'],sort=len(contentrows)+1,content=prefix+part,
|
remark='原文摘录；仅适用所列产品规程。共同工序的不同版本参数不合并。'))
|
if not contentrows:contentrows.append(dict(process_id=m['id'],sort=1,content='来源流程图：'+m['name']+'。详细参数以对应原规程确认。',remark='来源：'+','.join(m['sources'])))
|
insert_many('mes_pro_process_content',contentrows);newcontents+=len(contentrows)
|
nameids[m['name']]=m['id']
|
oldroutes={x['code']:x for x in before['mes_pro_route']}
|
for r in plan['routes']:
|
notes='；'.join(r['issues'])
|
description='来源：'+r['document_code']+'；保留原文工序顺序，全部非倒冲、手动领料；未启用，待核实产品/BOM和参数。'
|
if r['document_id']==15:description+='红、黑枸杞分支独立制浆，在搅拌乳化处合流。'
|
header=dict(code=r['code'],name=r['name'],description=description,status=1,remark=(notes or '按规程生成；系统唯一产出关键工序用于数量统计，原文*控制点逐项设置需要质检。'))
|
# Long exception text stays complete in JSON/CSV; database remark gives a pointer.
|
if len(header['remark'])>255:header['remark']=header['remark'][:220]+'；完整事项见来源规程及生成清单。'
|
assert len(header['name'])<=100 and len(header['description'])<=255
|
old=oldroutes.get(r['code'])
|
if old:assert same(old,header);r['id']=old['id'];r['action']='复用现有路线'
|
else:r['id']=insert('mes_pro_route',header);newroutes.append(r['id']);r['action']='新增未启用路线'
|
values=[]
|
for i,st in enumerate(r['steps']):
|
st['process_id']=nameids[st['name']]
|
nextpid=nameids[r['steps'][i+1]['name']] if i+1<len(r['steps']) else None
|
remark=st['remark']
|
if r['document_id']==15 and st['name'] in ['红枸杞原浆制备','黑果枸杞原浆制备']:
|
remark+='；原料支线独立加工，顺序仅用于系统编排，合流于搅拌乳化剪切。'
|
if r['document_id']==7 and st['name']=='混合':remark+='；时间冲突：图/监控表40min、正文30min，待核实。'
|
if r['document_id']==10 and st['name'] in ['浸泡','淋干','高温杀菌','粉碎','过筛']:remark+='；此段仅见原文流程图，详细操作范围待核实。'
|
if st['name'] in ['微波杀菌','水浴灭菌','均质','瞬时灭菌','热灌装'] and st['evidence']:
|
parameter=next((x['text'] for x in st['evidence'] if re.search(r'\d+(?:[.～~\-±]|℃|min|s|Mpa|ml)',x['text'])),None)
|
if parameter:
|
# Retain the original paragraph as source, without selecting a universal setpoint.
|
room=255-len(remark)-2
|
if room>20:remark+='；'+parameter[:room]
|
assert len(remark)<=255
|
value=dict(route_id=r['id'],process_id=st['process_id'],sort=st['sort'],next_process_id=nextpid,
|
link_type=3,prepare_time=None,wait_time=None,key_flag=int(st['key_flag']),check_flag=int(st['check_flag']),
|
backflush_flag=0,remark=remark)
|
values.append(value);st['stored']=value
|
if old:
|
q.execute('SELECT * FROM mes_pro_route_process WHERE route_id=%s AND deleted=0 ORDER BY sort',(r['id'],));existing=q.fetchall()
|
assert len(existing)==len(values) and all(same(a,b) for a,b in zip(existing,values))
|
else:insert_many('mes_pro_route_process',values);newsteps+=len(values)
|
# No products, BOM, or existing route rows altered by this transaction.
|
for t in ['mes_pro_route_product','mes_pro_route_product_bom']:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');assert list(q.fetchall())==before[t]
|
for row in before['mes_pro_route']:
|
q.execute('SELECT * FROM mes_pro_route WHERE id=%s',(row['id'],));assert q.fetchone()==row
|
dump('route-plan-with-ids.json',plan)
|
(P/'route-import-executed.sql').write_text('START TRANSACTION;\n'+'\n'.join(log)+'\nCOMMIT;\n',encoding='utf8')
|
c.commit()
|
for r in plan['routes']:
|
q.execute('SELECT * FROM mes_pro_route WHERE id=%s',(r['id'],));row=q.fetchone();assert row['status']==1 and row['code']==r['code']
|
q.execute('SELECT * FROM mes_pro_route_process WHERE route_id=%s AND deleted=0 ORDER BY sort',(r['id'],));found=q.fetchall()
|
assert len(found)==len(r['steps']) and all(same(a,b['stored']) for a,b in zip(found,r['steps']))
|
for t,n in [('mes_pro_process',len(newprocess)),('mes_pro_process_content',newcontents),('mes_pro_route',len(newroutes)),('mes_pro_route_process',newsteps)]:
|
q.execute('SELECT COUNT(*) n FROM '+t+' WHERE deleted=0');assert q.fetchone()['n']==len(before[t])+n
|
result=dict(target='36.213.150.237:10000/mom-hm',source_files=26,unique_documents=24,duplicates=plan['duplicate_documents'],
|
new_processes=len(newprocess),new_process_contents=newcontents,new_routes=len(newroutes),new_route_steps=newsteps,
|
starred_check_steps=sum(st['starred'] for r in plan['routes'] for st in r['steps']),
|
check_steps=sum(st['check_flag'] for r in plan['routes'] for st in r['steps']),all_backflush_false=True,
|
route_status='未启用',products_bound=0,existing_test_route_preserved=True,post_commit_verified=True)
|
dump('route-import-result.json',result)
|
with (P/'工艺路线生成结果.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['路线ID','路线编码','路线名称','来源文件编号','工序数','需质检数','星号质检数','倒冲','状态','工序顺序','待核实事项'])
|
for r in plan['routes']:w.writerow([r['id'],r['code'],r['name'],r['document_code'],len(r['steps']),sum(st['check_flag'] for st in r['steps']),sum(st['starred'] for st in r['steps']),False,'未启用',' → '.join(st['name']+('*' if st['starred'] else '') for st in r['steps']),'；'.join(r['issues']+r['binding_issues'])])
|
print(json.dumps(result,ensure_ascii=False))
|
q.execute("SELECT RELEASE_LOCK('stp-routes-20261012')")
|
except Exception:c.rollback();raise
|
finally:c.close()
|
