import csv, json, xml.etree.ElementTree as ET
|
from decimal import Decimal as D
|
from pathlib import Path
|
|
ROOT = Path('scripts/stock-import-202609/material-repair')
|
def read(name):
|
root = ET.parse(ROOT.parent / name).getroot()
|
return {s.tag: [{x.tag: x.text for x in row} for row in s] for s in root}
|
before = read('remote-duplicate-references-snapshot.xml')
|
after = read('remote-material-repair-final-snapshot.xml')
|
plan = json.loads((ROOT / 'plan.json').read_text(encoding='utf-8'))
|
merge = {str(x['old_id']): x for x in plan['merges']}
|
materials = {x['id']: x for x in after['materials']}
|
mes = {x['mdm_item_id']: x for x in after['mesMaterials'] if x['deleted'] == '0'}
|
assert len(before['materials']) - len(materials) == 10
|
assert set(merge) <= {x['id'] for x in after['deletedMaterials']}
|
for fix in plan['fixes']:
|
if str(fix['id']) in merge: continue
|
actual = materials[str(fix['id'])]
|
assert (actual['name'], actual['specification']) == (fix['name'], fix['specification'])
|
assert actual['category_id'] and actual['is_batch_managed'] == '1'
|
|
numeric = {'quantity', 'frozen_quantity', 'reserved_quantity', 'in_transit_quantity'}
|
for section, allowed in [('stock', {'item_id','item_type_id','updater','update_time'} | numeric), ('transactions', {'item_id','quantity','updater','update_time'}), ('batches', {'item_id','remark','updater','update_time'})]:
|
current = {x['id']: x for x in after[section]}
|
assert len(before[section]) == len(current)
|
for old in before[section]:
|
new = current[old['id']]
|
mapping = merge.get(old['item_id'])
|
target = str(mapping['new_id']) if mapping else old['item_id']
|
factor = D(mapping['factor']) if mapping else D('1')
|
assert new['item_id'] == target
|
for field in old:
|
if field not in allowed: assert old[field] == new[field], (section, old['id'], field)
|
for field in numeric & old.keys():
|
if old[field] is None: assert new[field] is None
|
else: assert D(new[field]) == D(old[field]) * factor, (section,old['id'],field)
|
if section == 'batches' and mapping:
|
assert mapping['old_code'] in new['remark'] and mapping['provenance'] in new['remark']
|
|
stock = after['stock']
|
canonical = {x['item_id'] for x in stock if x['creator'] == 'stock-import-202609'}
|
assert len(canonical) == 837
|
for item_id in canonical:
|
m, i = materials[item_id], mes[item_id]
|
for a, b in [('name','name'), ('specification','specification'), ('unit_measure_id','unit_measure_id'), ('category_id','item_type_id'), ('is_batch_managed','batch_flag'), ('status','status')]: assert m[a] == i[b], (item_id,a)
|
assert i['batch_flag'] == '1'
|
type_plan = json.loads((ROOT / 'verified-type-plan.json').read_text(encoding='utf-8'))
|
for row in type_plan:
|
m, i = materials[row['id']], mes[row['id']]
|
assert m['item_type'] == row['target_type']
|
assert m['category_id'] == i['item_type_id'] == row['target_category_id']
|
batch = {x['id']: x for x in after['batches']}
|
for s in stock:
|
assert s['item_id'] in materials and s['item_id'] == batch[s['batch_id']]['item_id']
|
assert s['item_type_id'] == materials[s['item_id']]['category_id']
|
assert before['units'] == after['units']
|
assert before['warehouses'] == after['warehouses']
|
assert before['locations'] == after['locations']
|
assert before['areas'] == after['areas']
|
assert not after['duplicateReferences']
|
old_specs = json.loads((ROOT / 'existing-spec-plan.json').read_text(encoding='utf-8'))
|
for f in old_specs['fixes']: assert materials[f['id']]['specification'] == mes[f['id']]['specification'] == f['specification']
|
msg = '核验通过：582条新增物料字段及分类已整理，201条原表规格已提取，39条已有物料缺失规格已补齐；10条重复物料归并，837个库存物料分类和批次管理与MES一致，574条MES记录补建；855条库存、843个批次及755条流水记录保留，单位换算后实物数量守恒；仓库、库区、库位和单位主数据均未改变。\n'
|
(ROOT / 'verification.txt').write_text(msg, encoding='utf-8')
|
with (ROOT / '整理后MES物料清单.csv').open('w', encoding='utf-8-sig', newline='') as f:
|
w=csv.writer(f);w.writerow(['物料ID','编码','名称','规格','物料类型','分类ID','单位ID','批次管理','MES已同步'])
|
for item_id in sorted(canonical,key=int):
|
m=materials[item_id];w.writerow([item_id,m['code'],m['name'],m['specification'],{'1':'原料','2':'半成品','3':'成品','4':'辅料'}[m['item_type']],m['category_id'],m['unit_measure_id'],'启用','是'])
|
print(msg)
|
