import sys,json,csv
|
from pathlib import Path
|
from decimal import Decimal
|
from collections import defaultdict
|
P=Path(__file__).parent
|
if (P/'shipping-complete-result.json').exists():
|
import runpy
|
runpy.run_path(str(P/'verify_shipping_complete.py'),run_name='__main__')
|
sys.exit(0)
|
sys.path.insert(0,str(P.parent/'iqc-import-20261012/deps'))
|
import pymysql
|
before=json.loads((P/'shipping-drafts-locked-before.json').read_text(encoding='utf8'))
|
r=json.loads((P/'shipping-drafts-result.json').read_text(encoding='utf8'))
|
canonical=lambda x:json.loads(json.dumps(x,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:
|
tables={}
|
for t in before['tables']:
|
q.execute('SELECT * FROM `'+t+'` WHERE deleted=0 ORDER BY id');tables[t]={x['id']:x for x in q.fetchall()}
|
draftids={o['order_id'] for o in r['orders']}
|
for t,rows in before['tables'].items():
|
for old in rows:
|
new=canonical(tables[t][old['id']]);diff={k for k in old if old[k]!=new[k]}
|
assert not diff or t=='erp_sale_order' and old['id'] in draftids and diff<={'out_status','updater','update_time'},(t,old['id'],diff)
|
if t not in ['mes_wm_sales_notice','mes_wm_sales_notice_line','mes_wm_product_sales','mes_wm_product_sales_line']:
|
assert len(rows)==len(tables[t]),('Unexpected records',t)
|
assert len(tables['mes_wm_sales_notice'])==23 and len(tables['mes_wm_product_sales'])==23
|
assert len(tables['mes_wm_sales_notice_line'])==75 and len(tables['mes_wm_product_sales_line'])==75
|
for o in r['orders']:
|
order=tables['erp_sale_order'][o['order_id']];notice=tables['mes_wm_sales_notice'][o['notice_id']];sales=tables['mes_wm_product_sales'][o['sales_id']]
|
assert notice['status']==3 and notice['out_status']==0 and sales['status']==0 and order['out_status']==1 and order['out_count']==0
|
assert notice['sale_order_id']==sales['sale_order_id']==order['id'] and sales['notice_id']==notice['id'] and notice['client_id']==sales['client_id']==order['customer_id']
|
assert not sales['carrier'] and not sales['shipping_number']
|
for l in o['lines']:
|
nl=tables['mes_wm_sales_notice_line'][l['notice_line_id']];sl=tables['mes_wm_product_sales_line'][l['sales_line_id']];src=tables['erp_sale_order_items'][l['order_item_id']]
|
assert nl['sale_order_item_id']==sl['sale_order_item_id']==src['id'] and sl['notice_line_id']==nl['id'] and sl['sales_id']==sales['id']
|
assert nl['quantity']==sl['quantity']==src['count'] and sl['enter_quantity']==src['count'] and sl['enter_unit_measure_id']==src['product_unit_id']
|
assert src['out_count']==0 and nl['out_count']==0 and sl['quality_status'] is None and sl['material_stock_id'] is None
|
needs=defaultdict(Decimal);units={};nos=defaultdict(set)
|
for o in r['orders']:
|
for l in o['lines']:
|
key=(l['item_id'],l['sale_unit_id'],l['primary_unit_id']);needs[key]+=Decimal(l['quantity']);nos[key].add(o['no'])
|
q.execute('SELECT id,name FROM mes_md_unit_measure WHERE deleted=0');units={x['id']:x['name'] for x in q.fetchall()}
|
checks=[]
|
for (id,su,pu),qty in needs.items():
|
item=tables['mdm_item'][id];stocks=[x for x in tables['mes_wm_material_stock'].values() if x['item_id']==id]
|
available=sum(max(Decimal(0),(x['quantity'] or Decimal(0))-(x['frozen_quantity'] or Decimal(0))-(x['reserved_quantity'] or Decimal(0))) for x in stocks if x['frozen'] in [0,b'\x00'])
|
checks.append({'item_id':id,'code':item['code'],'name':item['name'],'spec':item['specification'],'sale_unit':units.get(su),'primary_unit':units.get(pu),'need':str(qty),'available_primary':str(available),'shortfall_same_unit':str(max(Decimal(0),qty-available)) if su==pu else '需先确认单位换算','orders':'/'.join(sorted(nos[(id,su,pu)])),'issue':'单位换算待确认' if su!=pu else ('库存不足' if available<qty else '数量足够；库存来自9月结存，扣库口径待确认')})
|
report={'passed':True,'notices':23,'outbound_drafts':23,'lines':75,'stock_rows_unchanged':len(tables['mes_wm_material_stock']),'stock_transactions_created':0,'quality_records_unchanged':True,'carrier_and_tracking_not_invented':True,'excluded_return_no':'26000369','unit_mismatch_lines':r['counts']['unit_mismatch_lines'],'shortage_items_same_unit':sum(x['issue']=='库存不足' for x in checks),'workflow_complete':False,'pending':'等待历史补录或按当前库存实际出库口径确认'}
|
(P/'shipping-drafts-verification.json').write_text(json.dumps(report,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','编码','名称','规格','销售单位','主单位','待发数量','现存主单位数量','同单位缺口','订单号','处理事项'])
|
for x in checks:w.writerow([x['item_id'],x['code'],x['name'],x['spec'],x['sale_unit'],x['primary_unit'],x['need'],x['available_primary'],x['shortfall_same_unit'],x['orders'],x['issue']])
|
c.close();print(json.dumps(report,ensure_ascii=False))
|
