import sys,json
|
from pathlib import Path
|
from collections import defaultdict
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path(__file__).parent/'iqc-import-20261012/deps'))
|
import pymysql
|
out=Path(__file__).parent/'schedule-diagnosis-20261015'
|
before=json.loads(sorted(out.glob('complete-route-before-*.json'))[-1].read_text(encoding='utf-8'))
|
data={}
|
with pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',connect_timeout=15,cursorclass=pymysql.cursors.DictCursor) as c:
|
with c.cursor() as q:
|
for t in before:
|
q.execute('SELECT * FROM '+t+' WHERE deleted=0 ORDER BY id');data[t]=q.fetchall()
|
q.execute('SELECT * FROM mes_pro_work_order_process WHERE deleted=0 ORDER BY id');data['mes_pro_work_order_process']=q.fetchall()
|
q.execute('SELECT id,status FROM mes_dv_machinery WHERE deleted=0');machines={m['id']:m['status'] for m in q.fetchall()}
|
q.execute('SELECT * FROM mes_pro_task WHERE deleted=0 AND status NOT IN (4,5)');active=q.fetchall()
|
q.execute('SELECT * FROM mes_md_workstation WHERE deleted=0 AND status=0');stations=q.fetchall()
|
q.execute('SELECT * FROM mes_md_workstation_machine WHERE deleted=0');links=q.fetchall()
|
normalize=lambda rows:json.loads(json.dumps(rows,default=str))
|
assert normalize(data['mes_pro_work_order'])==before['mes_pro_work_order'],'工单被改动'
|
for table,allowed in [('mes_pro_route',{'status','update_time'}),('mes_pro_route_process',{'output_item_id','update_time'})]:
|
now={r['id']:r for r in normalize(data[table])}
|
for row in before[table]:
|
unexpected={k:{'before':v,'after':now[row['id']].get(k)} for k,v in row.items() if k not in allowed and v!=now[row['id']].get(k)}
|
if unexpected:print(json.dumps({'concurrentChange':table,'id':row['id'],'fields':unexpected},ensure_ascii=False))
|
for table in ['mes_md_product_bom','mes_pro_route_product_bom']:
|
now={r['id']:r for r in normalize(data[table])}
|
assert all(now[r['id']]==r for r in before[table]),'原有BOM被改动'
|
assert all(r['status']==0 for r in data['mes_pro_route'])
|
alloc={(b['route_id'],b['product_id']) for b in data['mes_pro_route_product_bom']}
|
assert all((p['route_id'],p['item_id']) in alloc for p in data['mes_pro_route_product'])
|
station_machines=defaultdict(set)
|
for rel in links:station_machines[rel['workstation_id']].add(rel['machinery_id'])
|
occupied=set().union(*(station_machines[t['workstation_id']] for t in active)) if active else set()
|
station_available={s['id']:bool(station_machines[s['id']]) and all(machines.get(i)==1 or (machines.get(i)==2 and i in occupied) for i in station_machines[s['id']]) for s in stations}
|
order_checks=[]
|
for o in data['mes_pro_work_order']:
|
if o['status']!=1:continue
|
qps=[p for p in data['mes_pro_work_order_process'] if p['work_order_id']==o['id']]
|
blocked=[p['process_id'] for p in qps if not any(s['process_id']==p['process_id'] and station_available[s['id']] for s in stations)]
|
order_checks.append({'id':o['id'],'code':o['code'],'blockedEquipmentProcesses':blocked})
|
report={'allRoutesEnabled':len(data['mes_pro_route']),'boundProductsWithInput':len(data['mes_pro_route_product']),'explicitOutputProcesses':sum(bool(p['output_item_id']) for p in data['mes_pro_route_process']),'nativeDefaultOutputProcesses':sum(not p['output_item_id'] for p in data['mes_pro_route_process']),'existingBomAndWorkOrdersPreserved':True,'equipmentChecks':order_checks,'scope':'数据库读回及资源前置条件核验，未执行实际排产或线上HTTP预览'}
|
(out/'verification.json').write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding='utf-8')
|
print(json.dumps(report,ensure_ascii=False))
|
