import collections, csv, json, re, unicodedata, xml.etree.ElementTree as ET
|
from decimal import Decimal
|
from pathlib import Path
|
|
ROOT = Path('scripts/stock-import-202609/material-repair')
|
ROOT.mkdir(exist_ok=True)
|
db = ET.parse(ROOT.parent / 'remote-duplicate-references-snapshot.xml').getroot()
|
def rows(tag): return [{x.tag: x.text for x in row} for row in db.find(tag)]
|
materials = rows('materials')
|
source = json.loads((ROOT.parent / 'source.json').read_text(encoding='utf-8'))
|
raw = {str(r[0]).strip(): r for r in source[0]['rows'][1:] if r[0]}
|
units = {u['id']: u for u in rows('units')}
|
warehouses = {w['id']: w['code'] for w in rows('warehouses')}
|
stock_warehouses = collections.defaultdict(set)
|
for s in rows('stock'): stock_warehouses[s['item_id']].add(warehouses[s['warehouse_id']])
|
def norm(value):
|
return re.sub(r'\s+', '', unicodedata.normalize('NFKC', value or '')).lower().replace('克', 'g').replace('×', '*')
|
def q(value):
|
if value is None: return 'NULL'
|
return "'" + str(value).replace('\\', '\\\\').replace("'", "''") + "'"
|
fixes = []
|
for m in materials:
|
if m['creator'] != 'stock-import-202609': continue
|
r = raw[m['code']]
|
name, spec = str(r[18] or '').strip(), str(r[19] or '').strip() or None
|
# The source's brand/name column contains inconsistent packaging labels in seven rows.
|
# Preserve the ledger description for these instead of changing a box/film into a product.
|
if not norm(str(r[1])).startswith(norm(name)): name = m['name']
|
wh = stock_warehouses[m['id']]
|
if m['code'].startswith('74') or 'BCPK' in wh: type_code = 'MAT06'
|
elif m['code'].startswith('81'):
|
type_code = 'MAT05' if any(t in name for t in ['标签', '底贴', '防伪', '二维码']) else ('MAT03' if 'NBCK' in wh else 'MAT04')
|
elif m['code'].startswith(('61', '62', '71')) or wh == {'YLK'}: type_code = 'MAT01'
|
else: type_code = 'FG'
|
fixes.append(dict(id=int(m['id']), code=m['code'], old_name=m['name'], old_spec=m['specification'], name=name, specification=spec, type_code=type_code))
|
m['name'], m['specification'] = name, spec
|
assert len(fixes) == 582
|
|
groups = collections.defaultdict(list)
|
for m in materials: groups[norm(m['name']), norm(m['specification'])].append(m)
|
merges = []
|
for group in groups.values():
|
if len(group) < 2 or not any(m['creator'] == 'stock-import-202609' for m in group): continue
|
group.sort(key=lambda m: (m['create_time'], int(m['id'])))
|
original = group[0]
|
for duplicate in group[1:]:
|
assert duplicate['creator'] == 'stock-import-202609', 'Do not merge pre-existing independent records'
|
from_unit, to_unit = units[duplicate['unit_measure_id']], units[original['unit_measure_id']]
|
mass = {'GM': Decimal('1'), 'KG': Decimal('1000')}
|
if from_unit['code'] in mass and to_unit['code'] in mass:
|
factor = mass[from_unit['code']] / mass[to_unit['code']]
|
else:
|
assert from_unit['code'] == to_unit['code'] or {from_unit['code'], to_unit['code']} <= {'BT', 'BX', 'EA'}, 'Unknown unit conversion'
|
factor = Decimal('1') # User instructed ignoring packaging-unit differences; identical name/spec.
|
r = raw[duplicate['code']]
|
provenance = f"归并前物料编码：{duplicate['code']}；名称：{duplicate['name']}；规格：{duplicate['specification'] or '原表未填写'}；单位：{from_unit['name'].strip()}；产地：{str(r[36] or '').strip()}；原描述：{str(r[1] or '').strip()}；归并至：{original['code']}；数量换算系数：{factor}"
|
merges.append(dict(old_id=int(duplicate['id']), old_code=duplicate['code'], new_id=int(original['id']), new_code=original['code'], name=original['name'], specification=original['specification'], old_unit=from_unit['name'].strip(), new_unit=to_unit['name'].strip(), factor=str(factor), provenance=provenance))
|
assert len(merges) == 10
|
assert {m['old_id'] for m in merges} == {969,970,971,972,1003,1004,1006,1009,1043,1528}
|
(ROOT / 'plan.json').write_text(json.dumps(dict(fixes=fixes, merges=merges), ensure_ascii=False, indent=2), encoding='utf-8')
|
for name, data in [('名称规格分类修正清单.csv', fixes), ('重复物料归并清单.csv', merges)]:
|
with (ROOT / name).open('w', encoding='utf-8-sig', newline='') as f:
|
writer = csv.DictWriter(f, fieldnames=list(data[0])); writer.writeheader(); writer.writerows(data)
|
sql = ["CREATE TEMPORARY TABLE stage_material_fix(id BIGINT PRIMARY KEY,code VARCHAR(100),old_name VARCHAR(200),old_spec VARCHAR(255),name VARCHAR(200),specification VARCHAR(255),type_code VARCHAR(64)) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"]
|
for m in fixes: sql.append('INSERT INTO stage_material_fix VALUES(' + ','.join(q(m[k]) for k in ['id','code','old_name','old_spec','name','specification','type_code']) + ')')
|
sql.append("CREATE TEMPORARY TABLE stage_material_merge(old_id BIGINT PRIMARY KEY,old_code VARCHAR(100),new_id BIGINT,new_code VARCHAR(100),factor DECIMAL(24,9),provenance VARCHAR(700)) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci")
|
for m in merges: sql.append('INSERT INTO stage_material_merge VALUES(' + ','.join(q(m[k]) for k in ['old_id','old_code','new_id','new_code','factor','provenance']) + ')')
|
(ROOT / 'repair-stage.sql').write_text(';\n'.join(sql) + ';\n', encoding='utf-8')
|
print(f'Prepared {len(fixes)} field/class repairs, {sum(bool(m["specification"]) for m in fixes)} source specifications, {len(merges)} duplicate merges; source batch provenance retained.')
|
