import sys,json,collections
|
from pathlib import Path
|
from decimal import Decimal
|
sys.stdout.reconfigure(encoding='utf-8')
|
sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
P=Path(__file__).parent;D=Decimal
|
before=json.loads((P/'before-production-floor-apply.json').read_text(encoding='utf8'))
|
patch=json.loads((P/'production-floor-applied.json').read_text(encoding='utf8'));assert patch['applied']
|
changes={(x['table'],x['id']):x['after'] for x in patch['changes']}
|
canon=lambda v:json.loads(json.dumps(v,ensure_ascii=False,default=str))
|
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,autocommit=False)
|
allrows={};checked=0
|
with c.cursor() as q:
|
for t,rows in before.items():
|
ids=[x['id'] for x in rows];q.execute('SELECT * FROM `'+t+'` WHERE id IN ('+','.join(['%s']*len(ids))+')',tuple(ids));actual={x['id']:x for x in q.fetchall()}
|
assert len(actual)==len(rows)
|
for row in rows:
|
expected={**row,**changes.get((t,row['id']),{})}
|
for key,value in actual[row['id']].items():
|
if isinstance(value,D):assert value==D(expected[key]),(t,row['id'],key)
|
else:assert canon(value)==expected[key],(t,row['id'],key)
|
allrows[t]=actual;checked+=len(rows)
|
for prefix in ['item_receipt','product_issue','item_consume','product_produce','product_receipt','product_sales']:
|
lt='mes_wm_'+prefix+'_line';dt='mes_wm_'+prefix+'_detail';totals=collections.defaultdict(lambda:D(0))
|
for row in allrows[dt].values():totals[row['line_id']]+=row['quantity']
|
for row in allrows[lt].values():assert totals[row['id']]==row.get('quantity',row.get('enter_quantity')),(lt,row['id'])
|
for row in allrows['mes_pro_work_order_bom'].values():
|
issued=sum(x['issued_quantity'] for x in allrows['mes_pro_task_issue'].values() if x['work_order_id']==row['work_order_id'] and x['item_id']==row['item_id'] and allrows['mes_pro_task'][x['task_id']]['process_id']==row['process_id'])
|
assert issued==row['quantity'],('BOM and issue mismatch',row['id'])
|
for row in allrows['mes_pro_task_issue'].values():assert row['used_quantity']==row['issued_quantity'] and row['available_quantity']==0
|
for row in allrows['mes_qc_iqc'].values():
|
line=allrows['mes_wm_arrival_notice_line'][row['source_line_id']]
|
assert row['received_quantity']==row['check_quantity']==row['qualified_quantity']==line['arrival_quantity']==line['qualified_quantity']
|
for row in allrows['mes_qc_ipqc'].values():assert row['qualified_quantity']==row['check_quantity']==allrows['mes_wm_product_produce_line'][row['source_line_id']]['quantity']==200
|
for row in allrows['mes_qc_oqc'].values():assert row['out_quantity']==row['check_quantity']==row['qualified_quantity']==allrows['mes_wm_product_sales_line'][row['source_line_id']]['quantity']==200
|
balances=collections.defaultdict(lambda:D(0))
|
for row in allrows['mes_wm_transaction'].values():balances[row['material_stock_id']]+=row['quantity']
|
assert len(balances)==62 and all(v==0 for v in balances.values())
|
ids=list(allrows['mes_pro_work_order_param_record']);q.execute('SELECT * FROM mes_pro_work_order_param_record_detail WHERE record_id IN ('+','.join(['%s']*len(ids))+')',tuple(ids));params=q.fetchall()
|
quantity_params=[x for x in params if '数量' in x['param_name']]
|
if '--sync-params' in sys.argv:
|
assert not (P/'before-quantity-params.json').exists(),'Parameter correction already applied'
|
(P/'before-quantity-params.json').write_text(json.dumps(params,ensure_ascii=False,indent=2,default=str),encoding='utf8')
|
corrections=[]
|
for row in quantity_params:
|
record=allrows['mes_pro_work_order_param_record'][row['record_id']];task=allrows['mes_pro_task'][record['task_id']]
|
assert '演示参数值' in row['remark'] and task['quantity']==200 and row['param_value']=='003'
|
q.execute('UPDATE mes_pro_work_order_param_record_detail SET param_value=%s,update_time=update_time WHERE id=%s AND param_value=%s',('200',row['id'],row['param_value']));assert q.rowcount==1
|
corrections.append({'id':row['id'],'old_value':row['param_value'],'new_value':'200','param_name':row['param_name']})
|
c.commit()
|
(P/'quantity-params-corrections.json').write_text(json.dumps(corrections,ensure_ascii=False,indent=2),encoding='utf8')
|
q.execute('SELECT * FROM mes_pro_work_order_param_record_detail WHERE record_id IN ('+','.join(['%s']*len(ids))+')',tuple(ids));params=q.fetchall()
|
prior=json.loads((P/'before-quantity-params.json').read_text(encoding='utf8'))
|
old_params={x['id']:x for x in prior}
|
for row in params:
|
expected={**old_params[row['id']]}
|
if '数量' in row['param_name']:
|
expected['param_value']='200';assert D(row['param_value'])==200
|
assert canon(row)==expected,('Parameter unrelated fields changed',row['id'])
|
result={'verified':True,'rows_checked':checked+len(params),'chains':4,'minimum_quantity':200,'bom_issue_consistent':True,'quality_source_quantities_consistent':True,'warehouse_line_details_consistent':True,'process_quantity_parameters_consistent':True,'inventory_balanced':True,'all_other_fields_unchanged':True}
|
(P/'production-floor-verification.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
print(json.dumps(result,ensure_ascii=False));c.close()
|
