import json,xml.etree.ElementTree as ET,collections,re
|
from pathlib import Path
|
from decimal import Decimal,ROUND_HALF_UP
|
root=Path('scripts/stock-import-202609')
|
before=ET.parse(root/'remote-snapshot.xml').getroot()
|
after=ET.parse(root/'remote-post-snapshot.xml').getroot()
|
def rows(db,key):return [{c.tag:c.text or '' for c in r} for r in db.find(key)]
|
source=json.loads((root/'source.json').read_text(encoding='utf-8'),parse_float=Decimal)
|
ledger=next(s for s in source if s['sheet']=='Sheet1')['rows']
|
original={str(r[4]).strip():r for r in ledger if len(r)>20 and isinstance(r[4],(str,int)) and re.fullmatch('[A-Za-z0-9]+',str(r[4]).strip())}
|
assert len(original)==843
|
units={r['id']:r['name'].strip() for r in rows(after,'units')}
|
materials=collections.defaultdict(list)
|
for m in rows(after,'materials'):materials[m['code']].append(m)
|
warehouses={w['id']:w for w in rows(after,'warehouses')}
|
batches={b['id']:b for b in rows(after,'batches')}
|
stock=[s for s in rows(after,'stock') if s['creator']=='stock-import-202609']
|
assert len(stock)==855
|
assert len(rows(after,'warehouses'))==len(rows(before,'warehouses'))+9
|
oldmdm={m['id']:m for m in rows(before,'materials')}
|
newmdm={m['id']:m for m in rows(after,'materials')}
|
assert all(newmdm[k]==m for k,m in oldmdm.items()),'Existing material was modified'
|
assert len(newmdm)-len(oldmdm)==582
|
assert rows(before,'deletedMaterials')==rows(after,'deletedMaterials')
|
actual=collections.defaultdict(Decimal)
|
distribution=collections.Counter()
|
stockids=set()
|
for s in stock:
|
assert s['id'] not in stockids;stockids.add(s['id'])
|
m=newmdm[s['item_id']]
|
assert m['code'] in original
|
b=batches[s['batch_id']]
|
assert b['item_id']==s['item_id'] and b['code']=='INIT202609-'+m['code']==s['batch_code']
|
assert b['produce_date']==b['expire_date']==''
|
assert s['frozen']=='0' and Decimal(s['frozen_quantity'])==Decimal(s['reserved_quantity'])==Decimal(s['in_transit_quantity'])==0
|
assert Decimal(s['quantity'])>=0
|
actual[m['code']]+=Decimal(s['quantity'])
|
distribution[warehouses[s['warehouse_id']]['name']]+=1
|
assert s['location_id']==s['area_id']==s['vendor_id']==''
|
assert s['receipt_time']=='2026-09-30 23:59:59'
|
assert set(actual)==set(original)
|
for code,r in original.items():
|
assert len(materials[code])==1
|
m=materials[code][0];sourceunit=r[5];unit=units[m['unit_measure_id']]
|
if code=='81013938':sourceunit='克'
|
quantity=Decimal(r[20])
|
if sourceunit=='公斤' and unit=='克':quantity*=1000
|
elif sourceunit=='克' and unit=='公斤':quantity/=1000
|
elif sourceunit!=unit:assert sourceunit in ('盒','罐','袋','套','个') and unit in ('个','张')
|
assert actual[code]==quantity.quantize(Decimal('.000001'),rounding=ROUND_HALF_UP),(code,quantity,actual[code])
|
transactions=[t for t in rows(after,'transactions') if t['creator']=='stock-import-202609']
|
positive={s['id']:s for s in stock if Decimal(s['quantity'])>0}
|
assert len(transactions)==len(positive)==755
|
seen=set()
|
for t in transactions:
|
sid=t['material_stock_id'];assert sid not in seen;seen.add(sid)
|
s=positive[sid]
|
assert t['item_id']==s['item_id'] and t['batch_id']==s['batch_id'] and t['warehouse_id']==s['warehouse_id']
|
assert Decimal(t['quantity'])==Decimal(s['quantity']) and t['type']=='1' and t['biz_type']=='126'
|
assert seen==set(positive)
|
assert len(rows(after,'batches'))==843
|
text='线上36 mom-hm独立回读校验通过：843个台账物料、855条分仓库存、582个新物料、9个仓库、843个期初汇总批次、755条库存流水；100条零结存；原有物料和已删除记录未改动。\n'+json.dumps(dict(distribution),ensure_ascii=False)+'\n'
|
(root/'verification-result.txt').write_text(text,encoding='utf-8')
|
print(text)
|
