"""Correct sales order grouping, without duplicating contracts, invoices or payments."""
|
import sys,json,copy
|
from pathlib import Path
|
from decimal import Decimal,ROUND_HALF_UP
|
P=Path(__file__).parent
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
D=Decimal
|
money=lambda v:D(v).quantize(D('.01'),rounding=ROUND_HALF_UP)
|
plan=json.loads((P/'import-plan.json').read_text(encoding='utf8'))
|
old=json.loads((P/'import-result.json').read_text(encoding='utf8'))
|
assert not (P/'source-order-regroup-result.json').exists(),'Already corrected; verify instead of rerunning.'
|
c=pymysql.connect(host='36.213.150.237',port=10000,user='root',password=sys.stdin.readline().strip(),database='mom-hm',charset='utf8mb4',autocommit=False,cursorclass=pymysql.cursors.DictCursor)
|
result={'groups':[],'orders':[],'rule':'相同原单据号一个订单；合同和合并开票保留；正合并金额内的退货原单独立保留；负合并金额两组仍排除'}
|
try:
|
with c.cursor() as q:
|
q.execute('SELECT GET_LOCK(%s,10) locked,NOW() now,DATABASE() db',('sales-ledger-202609',));state=q.fetchone();assert state['locked']==1 and state['db']=='mom-hm';now=state['now']
|
backup={'state':state,'tables':{}}
|
protected=['crm_sale_quotation','crm_sale_quotation_product','crm_contract_product','crm_invoice','crm_receivable','crm_receivable_plan','crm_permission','mdm_item','mes_md_item','crm_customer']
|
for t in ['erp_sale_order','erp_sale_order_items','crm_contract']+protected:
|
q.execute('SELECT * FROM `'+t+'` ORDER BY id');backup['tables'][t]=q.fetchall()
|
(P/'source-order-regroup-before.json').write_text(json.dumps(backup,ensure_ascii=False,default=str,indent=2),encoding='utf8')
|
orders={x['id']:x for x in backup['tables']['erp_sale_order']};items={x['id']:x for x in backup['tables']['erp_sale_order_items']};contracts={x['id']:x for x in backup['tables']['crm_contract']}
|
original_ids={g['order_id'] for g in old['groups']};all_source=set()
|
for g in plan['groups']:
|
assert not (all_source & set(g['source_numbers'])),'A source number crosses invoice groups; needs global consolidation.'
|
all_source.update(g['source_numbers'])
|
conflicts=[x for x in orders.values() if x['no'] in all_source and x['id'] not in original_ids];assert not conflicts,[(x['id'],x['no']) for x in conflicts]
|
for g,prior in zip(plan['groups'],old['groups']):
|
original=orders[prior['order_id']];contract=contracts[prior['contract_id']]
|
assert original['no']=='XSDD-'+g['key'] and original['total_price']==D(g['gross']) and original['out_count']==0 and original['production_status']==0
|
assert contract['order_id']==original['id'] and contract['total_price']==D(g['gross'])
|
for a,l in zip(g['lines'],prior['lines']):
|
line=items[l['order_item_id']];assert line['order_id']==original['id'] and line['count']==D(a['quantity']) and line['total_price']==D(a['net']) and line['tax_price']==D(a['tax']) and line['out_count']==0 and not line['mps_id']
|
buckets=[]
|
for no in g['source_numbers']:
|
ls=[(a,l) for a,l in zip(g['lines'],prior['lines']) if a['source_no']==no]
|
buckets.append({'source_no':no,'lines':ls,'gross':money(sum(D(a['gross_raw']) for a,l in ls))})
|
# Preserve the original invoice-group amount; document the one-cent allocation explicitly.
|
delta=D(g['gross'])-sum(b['gross'] for b in buckets)
|
recipient=next(b for b in reversed(buckets) if b['gross']>0);recipient['gross']+=delta
|
changed=[]
|
for i,b in enumerate(buckets):
|
no=b['source_no'];net=sum(D(a['net']) for a,l in b['lines']);tax=sum(D(a['tax']) for a,l in b['lines']);count=sum(D(a['quantity']) for a,l in b['lines']);gross=b['gross']
|
marker='退货原单，负数量负金额保留；' if gross<0 else ''
|
remark=f'九月台账{g["key"]}；原单据号{no}；同号明细归同一订单；{marker}关联合并合同{contract["no"]}，开票组含税合计{g["gross"]}元；本单净税额与含税金额尾差{net+tax-gross}元；合并组分配尾差{delta if b is recipient else 0}元。'
|
fields={'no':no,'total_count':count,'total_price':gross,'total_product_price':net,'total_tax_price':tax,'discount_price':net+tax-gross,'remark':remark[:255],'updater':'291','update_time':now}
|
if i==0:
|
oid=original['id'];q.execute('UPDATE erp_sale_order SET '+','.join('`'+k+'`=%s' for k in fields)+' WHERE id=%s',[*fields.values(),oid])
|
else:
|
fields={**original,**fields};fields.pop('id');fields['creator']='291';fields['create_time']=now
|
q.execute('INSERT INTO erp_sale_order ('+','.join('`'+k+'`' for k in fields)+') VALUES ('+','.join(['%s']*len(fields))+')',list(fields.values()));oid=q.lastrowid
|
ids=[l['order_item_id'] for a,l in b['lines']]
|
q.execute('UPDATE erp_sale_order_items SET order_id=%s,updater=%s,update_time=%s WHERE id IN ('+','.join(['%s']*len(ids))+')',[oid,'291',now,*ids])
|
entry={'source_no':no,'order_id':oid,'contract_id':contract['id'],'customer_id':original['customer_id'],'gross':str(gross),'net':str(net),'tax':str(tax),'count':str(count),'order_item_ids':[l['order_item_id'] for a,l in b['lines']],'source_rows':[a['row'] for a,l in b['lines']]};changed.append(entry);result['orders'].append(entry)
|
q.execute('UPDATE crm_contract SET order_no=%s,updater=%s,update_time=%s WHERE id=%s',(buckets[0]['source_no'],'291',now,contract['id']))
|
result['groups'].append({'key':g['key'],'primary_order_id':original['id'],'contract_id':contract['id'],'orders':changed,'invoice_id':prior['invoice_id'],'receivable_id':prior['receivable_id'],'plan_id':prior['plan_id']})
|
# Validate all balances and line partitioning before commit.
|
assert len(result['orders'])==24 and len({o['source_no'] for o in result['orders']})==24
|
seen=set();total=D(0)
|
for o in result['orders']:
|
q.execute('SELECT * FROM erp_sale_order WHERE id=%s',(o['order_id'],));actual=q.fetchone()
|
assert actual['no']==o['source_no'] and actual['total_price']==D(o['gross']) and actual['total_product_price']+actual['total_tax_price']-actual['discount_price']==actual['total_price']
|
q.execute('SELECT * FROM erp_sale_order_items WHERE order_id=%s AND deleted=0',(o['order_id'],));ls=q.fetchall();ids={x['id'] for x in ls};assert ids==set(o['order_item_ids']) and not ids & seen;seen|=ids
|
assert sum(x['count'] for x in ls)==actual['total_count'] and sum(x['total_price'] for x in ls)==actual['total_product_price'] and sum(x['tax_price'] for x in ls)==actual['total_tax_price']
|
total+=actual['total_price']
|
assert len(seen)==76 and total==D(plan['totals']['gross'])
|
for group,g in zip(result['groups'],plan['groups']):assert sum(D(x['gross']) for x in group['orders'])==D(g['gross'])
|
for t in protected:
|
q.execute('SELECT * FROM `'+t+'` ORDER BY id');assert q.fetchall()==backup['tables'][t],('Unexpected changes',t)
|
c.commit();result.update(committed=True,at=str(now),order_count=24,line_count=76,total=str(total))
|
(P/'source-order-regroup-result.json').write_text(json.dumps(result,ensure_ascii=False,indent=2),encoding='utf8')
|
print(json.dumps({'committed':True,'orders':24,'lines':76,'total':str(total),'split_orders':[o for o in result['orders'] if o['source_no'] in ['26000393','26000406','26000369']]},ensure_ascii=False))
|
except Exception:c.rollback();raise
|
finally:
|
with c.cursor() as q:q.execute('SELECT RELEASE_LOCK(%s)',('sales-ledger-202609',))
|
c.close()
|
