from pathlib import Path
|
import sys,json
|
from decimal import Decimal as D
|
sys.stdout.reconfigure(encoding='utf-8');sys.path.insert(0,str(Path('scripts/iqc-import-20261012/deps').resolve()))
|
import pymysql
|
P=Path(__file__).parent;r=json.loads((P/'remote-result-applied.json').read_text(encoding='utf-8'));e=json.loads((P/'sync-export.json').read_text(encoding='utf-8'))
|
assert r['applied']
|
def mid(table,value):return r['maps'][table][str(value)]
|
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=15,read_timeout=60)
|
try:
|
with c.cursor() as q:
|
for table,ids in r['created'].items():
|
q.execute('select count(*) n from `'+table+'` where id in ('+','.join(['%s']*len(ids))+')',ids);assert q.fetchone()['n']==len(ids),(table,'missing committed rows')
|
for item in e['data']['mes_pro_mps']:
|
q.execute('select m.*,w.product_id wo_product,w.quantity wo_qty,w.parent_id,w.order_source_code,l.count sale_qty,l.product_id sale_product,l.mps_id,l.product_unit_id from mes_pro_mps m join mes_pro_work_order w on w.id=m.work_order_id join erp_sale_order_items l on l.id=m.sale_order_item_id where m.id=%s',(mid('mes_pro_mps',item['id']),));v=q.fetchone()
|
assert v['quantity']==v['wo_qty']==v['sale_qty']==D(str(item['quantity']))
|
assert v['wo_product']==v['product_id']==mid('mes_md_item',item['product_id'])
|
q.execute('select mdm_item_id from mes_md_item where id=%s',(v['product_id'],));assert q.fetchone()['mdm_item_id']==v['sale_product']
|
assert v['mps_id']==v['id'] and v['sale_order_no']==v['order_source_code']==item['sale_order_no'] and v['parent_id']==0
|
assert v['quantity_produced']==0 and v['request_date'].weekday()<5
|
q.execute('select name from mes_md_unit_measure where id=%s',(v['unit_measure_id'],));mu=q.fetchone()['name'].strip()
|
q.execute('select name from mdm_unit_measure where id=%s',(v['product_unit_id'],));assert q.fetchone()['name'].strip()==mu
|
total=D(0)
|
for g in e['local_result']['groups'].values():
|
q.execute('select * from crm_contract where id=%s',(mid('crm_contract',g['contract_id']),));ct=q.fetchone()
|
q.execute('select * from crm_sale_quotation where id=%s',(mid('crm_sale_quotation',g['quotation_id']),));qt=q.fetchone()
|
q.execute('select * from erp_sale_order where id=%s',(mid('erp_sale_order',g['order_id']),));so=q.fetchone()
|
q.execute('select * from crm_receivable_plan where id=%s',(mid('crm_receivable_plan',g['receivable_plan_id']),));rp=q.fetchone()
|
assert ct['id']==qt['contract_id']==so['contract_id']==rp['contract_id'] and ct['order_id']==so['id']
|
assert all(x['customer_id']==mid('crm_customer',1613) for x in [ct,qt,so,rp])
|
assert ct['total_price']==qt['total_price']==so['total_price']==rp['price']==D(g['amount']);total+=ct['total_price']
|
assert rp['return_time'].strftime('%Y-%m-%d')=='2026-11-30' and rp['receivable_id'] is None
|
assert so['out_count']==0 and so['production_status']==0
|
assert total==D('2062353.18')
|
for item in e['data']['mes_pro_task']:
|
q.execute('select * from mes_pro_task where id=%s',(mid('mes_pro_task',item['id']),));t=q.fetchone()
|
assert t['status']==item['status'] and t['quantity']==D(str(item['quantity'])) and t['produced_quantity']==D(str(item['produced_quantity']))
|
assert t['work_order_id']==mid('mes_pro_work_order',item['work_order_id']) and t['start_time'].strftime('%Y-%m-%d %H:%M:%S')==item['start_time']
|
for table,count in r['protected_counts'].items():q.execute('select count(*) n from `'+table+'`');assert q.fetchone()['n']==count
|
checks={'passed':True,'host':r['host'],'database':r['database'],'total':str(total),'production_plans':13,'work_orders':13,'quotations':4,'contracts':4,'sales_orders':4,'receivable_plans':4,'packaging_partial_quantity':800,'new_actual_receipts':0,'new_stock_or_sales_transactions':0}
|
(P/'remote-verification.json').write_text(json.dumps(checks,ensure_ascii=False,indent=2),encoding='utf-8');print(json.dumps(checks,ensure_ascii=False))
|
finally:c.close()
|
