import sys,json,csv
|
from pathlib import Path
|
from decimal import Decimal
|
P=Path(__file__).parent
|
if (P/'ecommerce-customer-correction-result.json').exists():
|
import runpy
|
runpy.run_path(str(P/'verify_current_sales.py'),run_name='__main__')
|
sys.exit(0)
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
plan=json.loads((P/'import-plan.json').read_text(encoding='utf8'))
|
r=json.loads((P/'source-order-regroup-result.json').read_text(encoding='utf8'))
|
before=json.loads((P/'source-order-regroup-before.json').read_text(encoding='utf8'))
|
D=Decimal
|
canonical=lambda v:json.loads(json.dumps(v,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)
|
with c.cursor() as q:
|
current={}
|
for t in before['tables']:
|
q.execute('SELECT * FROM `'+t+'` ORDER BY id');current[t]={x['id']:x for x in q.fetchall()}
|
protected=[t for t in before['tables'] if t not in ['erp_sale_order','erp_sale_order_items','crm_contract']]
|
for t in protected:assert canonical(list(current[t].values()))==before['tables'][t],t
|
orderids={x['order_id'] for x in r['orders']};lineids={id for x in r['orders'] for id in x['order_item_ids']};contractids={x['contract_id'] for x in r['orders']}
|
allowed={'erp_sale_order':{'no','total_count','total_price','total_product_price','total_tax_price','discount_price','remark','updater','update_time'},'erp_sale_order_items':{'order_id','updater','update_time'},'crm_contract':{'order_no','updater','update_time'}}
|
relevant={'erp_sale_order':orderids,'erp_sale_order_items':lineids,'crm_contract':contractids}
|
for t in allowed:
|
for old in before['tables'][t]:
|
new=canonical(current[t][old['id']]);different={k for k in old if old[k]!=new[k]}
|
assert not different or old['id'] in relevant[t] and different<=allowed[t],(t,old['id'],different)
|
seen=set();total=D(0);detail=[]
|
assert len(r['orders'])==len({x['source_no'] for x in r['orders']})==24
|
for group,g in zip(r['groups'],plan['groups']):
|
ct=current['crm_contract'][group['contract_id']]
|
assert ct['order_id']==group['primary_order_id'] and ct['order_no']==group['orders'][0]['source_no']
|
assert sum(D(o['gross']) for o in group['orders'])==ct['total_price']==D(g['gross'])
|
for o in group['orders']:
|
order=current['erp_sale_order'][o['order_id']]
|
assert order['deleted'] in [0,b'\x00'] and order['no']==o['source_no'] and order['contract_id']==ct['id'] and order['customer_id']==ct['customer_id']
|
assert order['total_price']==D(o['gross']) and order['total_product_price']+order['total_tax_price']-order['discount_price']==order['total_price']
|
lines=[l for l in current['erp_sale_order_items'].values() if l['order_id']==o['order_id'] and l['deleted'] in [0,b'\x00']]
|
actualids={l['id'] for l in lines};assert actualids==set(o['order_item_ids']) and not actualids&seen;seen|=actualids
|
assert sum(l['count'] for l in lines)==order['total_count'] and sum(l['total_price'] for l in lines)==order['total_product_price'] and sum(l['tax_price'] for l in lines)==order['total_tax_price']
|
for row,id in zip(o['source_rows'],o['order_item_ids']):
|
source=next(l for l in g['lines'] if l['row']==row);assert source['source_no']==o['source_no'] and current['erp_sale_order_items'][id]['count']==D(source['quantity'])
|
total+=order['total_price'];detail.append({'source_no':o['source_no'],'order_id':o['order_id'],'customer':g['customer_name'],'lines':len(lines),'gross':o['gross'],'contract_id':ct['id'],'invoice_id':group['invoice_id']})
|
assert len(seen)==76 and total==D(plan['totals']['gross'])
|
# No duplicate active or deleted order number may shadow these imported source numbers.
|
for no in {o['source_no'] for o in r['orders']}:assert sum(o['no']==no for o in current['erp_sale_order'].values())==1
|
out={'passed':True,'orders':24,'lines':76,'total':str(total),'protected_tables_unchanged':protected,'detail':detail}
|
(P/'source-order-verification.json').write_text(json.dumps(out,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(['原单据号/订单号','系统订单ID','客户','明细行数','含税金额','合同ID','合并开票ID'])
|
for x in detail:w.writerow([x['source_no'],x['order_id'],x['customer'],x['lines'],x['gross'],x['contract_id'],x['invoice_id']])
|
c.close();print(json.dumps({k:v for k,v in out.items() if k!='detail'},ensure_ascii=False))
|
