"""Independent read-only validation of posted documents and inventory conservation."""
|
import sys,json,csv
|
from pathlib import Path
|
from decimal import Decimal
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
D=Decimal
|
canon=lambda x:json.loads(json.dumps(x,default=str))
|
r=json.loads((P/'shipping-complete-result.json').read_text(encoding='utf8'))
|
b=json.loads((P/'shipping-complete-locked-before.json').read_text(encoding='utf8'))
|
name_fix_path=P/'contract-name-prefix-result.json'
|
name_fixes={x['id']:x for x in json.loads(name_fix_path.read_text(encoding='utf8'))['changes']} if name_fix_path.exists() else {}
|
quote_fix_path=P/'quotation-name-sync-result.json'
|
quote_name_fixes={x['id']:x for x in json.loads(quote_fix_path.read_text(encoding='utf8'))['changes']} if quote_fix_path.exists() else {}
|
date_fix_path=P/'business-create-times-result.json'
|
date_fixes={(x['table'],x['id']):x for x in json.loads(date_fix_path.read_text(encoding='utf8'))['changes']} if date_fix_path.exists() else {}
|
remark_fix_path=P/'contract-order-remarks-cleaned-result.json'
|
remark_fixes={(x['table'],x['id']):x for x in json.loads(remark_fix_path.read_text(encoding='utf8'))['changes']} if remark_fix_path.exists() else {}
|
if date_fix_path.exists():
|
date_result=json.loads(date_fix_path.read_text(encoding='utf8'))
|
latest_backup=json.loads((P/date_result['backup']).read_text(encoding='utf8'))
|
# Verify against customer state immediately before this authorized adjustment.
|
b['tables']['crm_customer']=latest_backup['tables']['crm_customer']
|
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)
|
with c.cursor() as q:
|
data={}
|
for t in b['tables']:
|
q.execute('SELECT * FROM `'+t+'` WHERE deleted=0 ORDER BY id');data[t]={x['id']:x for x in q.fetchall()}
|
receipt=data['mes_wm_misc_receipt'][r['receipt_id']];assert receipt['status']==4 and receipt['code']==r['receipt_code']
|
expected_new={'mes_wm_misc_receipt':{r['receipt_id']},'mes_wm_misc_receipt_line':{a['receipt_line_id'] for a in r['allocations']},'mes_wm_product_sales_detail':{a['detail_id'] for a in r['allocations']},'mes_wm_stock_reserve':{a['reserve_id'] for a in r['allocations']},'mes_wm_transaction':{a[k] for a in r['allocations'] for k in ['in_transaction_id','out_transaction_id']},'mes_wm_material_stock':set(r['new_stock_ids']),'mes_wm_batch':set(r['new_batch_ids'])}
|
for t,ids in expected_new.items():assert set(data[t])-{x['id'] for x in b['tables'][t]}==ids,(t,'Unexpected added rows')
|
for old in b['tables']['mes_wm_material_stock']:
|
actual=canon(data['mes_wm_material_stock'][old['id']]);diff={k for k in old if old[k]!=actual[k]};assert diff<={'updater','update_time'},(old['id'],diff)
|
for id in r['new_stock_ids']:
|
s=data['mes_wm_material_stock'][id];assert all(s[k]==0 for k in ['quantity','frozen_quantity','reserved_quantity','in_transit_quantity'])
|
for a in r['allocations']:
|
qty=D(a['quantity']);inc=data['mes_wm_transaction'][a['in_transaction_id']];out=data['mes_wm_transaction'][a['out_transaction_id']]
|
assert inc['quantity']==qty and out['quantity']==-qty and inc['type']==1 and out['type']==2 and inc['biz_type']==115 and out['biz_type']==119
|
for k in ['item_id','material_stock_id','batch_id','batch_code','warehouse_id','location_id','area_id']:assert inc[k]==out[k],(a['no'],k)
|
assert inc['material_stock_id']==a['stock_id'] and inc['biz_id']==r['receipt_id'] and inc['biz_line_id']==a['receipt_line_id'] and out['biz_id']==a['sales_id'] and out['biz_line_id']==a['sales_line_id']
|
reserve=data['mes_wm_stock_reserve'][a['reserve_id']];assert reserve['status']==3 and reserve['quantity']==qty and reserve['material_stock_id']==a['stock_id']
|
detail=data['mes_wm_product_sales_detail'][a['detail_id']];assert detail['quantity']==qty and detail['material_stock_id']==a['stock_id']
|
sl=data['mes_wm_product_sales_line'][a['sales_line_id']];nl=data['mes_wm_sales_notice_line'][a['notice_line_id']];ol=data['erp_sale_order_items'][a['order_item_id']]
|
assert sl['quantity']==nl['out_count']==ol['out_count']==ol['count']==qty and sl['batch_id']==nl['batch_id']==a['batch_id']
|
assert sl['material_stock_id']==a['stock_id'] and sl['quality_status'] is None and sl['oqc_id'] is None
|
for o in r['orders']:
|
erp=data['erp_sale_order'][o['order_id']];notice=data['mes_wm_sales_notice'][o['notice_id']];sales=data['mes_wm_product_sales'][o['sales_id']]
|
assert erp['out_count']==erp['total_count'] and erp['out_status']==3 and notice['status']==4 and notice['out_status']==2 and sales['status']==4
|
assert sales['carrier']==o['carrier'] and sales['shipping_number']==o['shipping_number'] and sales['client_id']==erp['customer_id']==o['customer_id']
|
if (P/'shipping-remarks-cleared-result.json').exists():assert sales['remark'] is None and notice['remark'] is None
|
else:assert ('模拟生成' in sales['remark']) if o['jd'] else ('青H' in sales['carrier'])
|
assert len({o['shipping_number'] for o in r['orders']})==23
|
affected_orders={o['order_id'] for o in r['orders']};affected_lines={a['order_item_id'] for a in r['allocations']}
|
for t,rows in b['tables'].items():
|
if t in ['erp_sale_order','erp_sale_order_items']:
|
ids=affected_orders if t=='erp_sale_order' else affected_lines
|
allowed={'out_count','updater','update_time'}|({'out_status'} if t=='erp_sale_order' else set())
|
for old in rows:
|
new=canon(data[t][old['id']]);diff={k for k in old if old[k]!=new[k]};allowed_row=allowed if old['id'] in ids else set()
|
if (t,old['id']) in date_fixes:
|
allowed_row=allowed_row|{'create_time','updater','update_time'}
|
assert new['create_time']==date_fixes[(t,old['id'])]['create_time']
|
if (t,old['id']) in remark_fixes:
|
allowed_row|={'remark','updater'}
|
assert new['remark']==remark_fixes[(t,old['id'])]['remark']
|
assert diff<=allowed_row,(t,old['id'],diff)
|
elif t.startswith('crm_') or t in ['mdm_item','mes_qc_oqc','mes_wm_warehouse','mes_wm_warehouse_location','mes_wm_warehouse_area']:
|
fixes=name_fixes if t=='crm_contract' else quote_name_fixes if t=='crm_sale_quotation' else {}
|
if fixes or any(k[0]==t for k in date_fixes) or any(k[0]==t for k in remark_fixes):
|
for old in rows:
|
new=canon(data[t][old['id']]);diff={k for k in old if old[k]!=new[k]}
|
allowed_row={'name','updater','update_time'} if old['id'] in fixes else set()
|
if (t,old['id']) in date_fixes:
|
allowed_row|={'create_time','updater','update_time'}
|
assert new['create_time']==date_fixes[(t,old['id'])]['create_time']
|
if (t,old['id']) in remark_fixes:
|
allowed_row|={'remark','updater'}
|
assert new['remark']==remark_fixes[(t,old['id'])]['remark']
|
assert diff<=allowed_row
|
if old['id'] in fixes:assert new['name']==fixes[old['id']]['new_name']
|
else:assert rows==canon(list(data[t].values())),('Protected table changed',t)
|
for t in ['mes_wm_transaction','mes_wm_stock_reserve','mes_wm_product_sales_detail','mes_wm_batch','mes_wm_misc_receipt','mes_wm_misc_receipt_line','mes_wm_misc_receipt_detail']:
|
for old in b['tables'][t]:assert old==canon(data[t][old['id']]),('Existing row changed',t,old['id'])
|
negative=next(x for x in data['erp_sale_order'].values() if x['no']=='26000369');assert negative['out_count']==0
|
summary={'passed':True,**r['counts'],'original_stock_rows_unchanged':len(b['tables']['mes_wm_material_stock']),'new_zero_stock_rows':len(r['new_stock_ids']),'stock_net_change':'0','financial_data_unchanged':True,'negative_order_not_shipped':'26000369','tracking_numbers':'模拟生成；自送保留青H前缀'}
|
(P/'shipping-complete-verification.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2),encoding='utf8')
|
with (P/'九月销售出库发货核对清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['销售订单号','客户','发货通知单号','出库单号','状态','承运商','运单或内部配送号','说明'])
|
for o in r['orders']:w.writerow([o['no'],o['customer'],o['notice_code'],o['sales_code'],'已完成',o['carrier'],o['shipping_number'],'模拟运单，非真实物流凭证' if o['jd'] else '自有车辆；车牌前缀青H；编号为内部配送号'])
|
print(json.dumps(summary,ensure_ascii=False))
|
c.close()
|
