import csv,json,collections,re,xml.etree.ElementTree as ET
|
from pathlib import Path
|
from decimal import Decimal
|
root=Path('scripts/stock-import-202609')
|
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')
|
merged_names={}
|
for merged in ledger['merged']:
|
match=re.fullmatch(r'D(\d+):D(\d+)',merged)
|
if match:
|
first,last=map(int,match.groups())
|
for number in range(first,last+1):merged_names[number]=ledger['rows'][first-1][3]
|
rows=[]
|
for index,r in enumerate(ledger['rows'],1):
|
if len(r)>20 and isinstance(r[4],(int,str)) and re.fullmatch(r'[A-Za-z0-9]+',str(r[4]).strip()):
|
name=r[3] or merged_names.get(index)
|
assert name,('Missing material name',index,r[4])
|
rows.append(dict(row=index,code=str(r[4]).strip(),name=name,unit=r[5],quantity=str(r[20]),stock_unit=r[6],stock_quantity=str(r[21])))
|
assert len(rows)==len({r['code'] for r in rows})
|
for target in ('local','remote'):
|
db=ET.parse(root/f'{target}-snapshot.xml').getroot()
|
materials=collections.defaultdict(list)
|
for m in db.find('materials'): materials[m.findtext('code')].append({c.tag:c.text for c in m})
|
units={u.findtext('id'):u.findtext('name').strip() for u in db.find('units')}
|
names=collections.defaultdict(list)
|
for entries in materials.values():
|
for m in entries:names[m['name']].append(m)
|
plan=[]
|
for r in rows:
|
out=dict(r)
|
matches=materials[r['code']]
|
out['material_id']=matches[0]['id'] if len(matches)==1 else ''
|
out['system_name']=matches[0]['name'] if len(matches)==1 else ''
|
out['system_unit']=units.get(matches[0]['unit_measure_id'],'') if len(matches)==1 else ''
|
out['state']='编码匹配' if len(matches)==1 else ('编码重复' if matches else '缺少物料')
|
if not matches and names[r['name']]:out['state']='同名不同编码，待核对';out['candidates']=';'.join(m['code'] for m in names[r['name']])
|
if len(matches)==1 and out['system_unit']!=r['unit']:out['state']='单位待换算'
|
plan.append(out)
|
(root/f'{target}-plan.json').write_text(json.dumps(plan,ensure_ascii=False,indent=2),encoding='utf-8')
|
with (root/f'{target}-库存核对清单.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
keys=['row','code','name','unit','quantity','stock_unit','stock_quantity','material_id','system_name','system_unit','state','candidates']
|
writer=csv.DictWriter(f,fieldnames=keys);writer.writeheader();writer.writerows(plan)
|
print(target,len(plan),dict(collections.Counter(r['state'] for r in plan)))
|
