import json,csv,re,sys,collections
|
from pathlib import Path
|
from decimal import Decimal,ROUND_HALF_UP
|
import xml.etree.ElementTree as ET
|
P=Path('scripts/bom-import')
|
def load(target):
|
r=ET.parse(P/f'{target}-snapshot.xml')
|
return {tag:[{e.tag:e.text or '' for e in x} for x in r.findall(f'./{tag}/row')] for tag in ['materials','units','bom']}
|
def norm(s):return re.sub(r'\s+','',s or '').replace('（','(').replace('）',')')
|
source=[]
|
for s in json.loads((P/'xls-source.json').read_text(encoding='utf-8')):
|
parent=name=None
|
for n,r in enumerate(s['rows'],1):
|
if n<=2:continue
|
if r[0]:name=str(r[0]).strip();parent=str(r[1]).strip()
|
assert parent and len(r)>=6,(s['sheet'],n,r)
|
source.append(dict(file='电商产品BOM(1).xls',sheet=s['sheet'],row=n,parent_code=parent,parent_name=name,child_code=str(r[2]).strip(),child_name=str(r[3]).strip(),quantity=str(r[4]),unit=str(r[5] or '').strip(),section=''))
|
sections={'红枣单碗':'71500052','鱼胶单碗':'71500051'}
|
for s in json.loads((P/'xlsx-source.json').read_text(encoding='utf-8')):
|
parent_name=s['sheet'];parent_code='';section='';cols={}
|
prior=[]
|
for n,r in enumerate(s['rows'],1):
|
if not any(str(x or '').strip() for x in r):continue
|
if norm(str(r[0]))=='项目号':
|
cols={norm(str(v)):i for i,v in enumerate(r) if v is not None};continue
|
if not str(r[0] or '').isdigit():
|
section=str(r[0]);candidates=[x['child_code'] for x in prior if x['child_code'].startswith(('715','740'))]
|
if section in sections:parent_code=sections[section]
|
else:
|
assert len(set(candidates))==1,(s['sheet'],section,candidates)
|
parent_code=candidates[0]
|
continue
|
row=dict(file='传渠产品BOMxml(1).xlsx',sheet=s['sheet'],row=n,parent_code=parent_code,parent_name=parent_name if not section else section,child_code=str(r[0]),child_name=str(r[1]).strip(),quantity=str(r[cols['数量']]),unit=str(r[cols['计量单位']] or '').strip() if '计量单位' in cols else '',section=section)
|
source.append(row)
|
if not section:prior.append(row)
|
(P/'source-lines.json').write_text(json.dumps(source,ensure_ascii=False,indent=2),encoding='utf-8')
|
unit_evidence=collections.defaultdict(set)
|
for x in source:
|
if x['unit']:unit_evidence[x['child_code']].add(x['unit'])
|
for target in ['local','remote']:
|
db=load(target);codes=collections.defaultdict(list);names=collections.defaultdict(list);units={x['id']:x for x in db['units']}
|
for x in db['materials']:codes[x['code']].append(x);names[norm(x['name'])].append(x)
|
accepted=[];issues=[];notes=[]
|
for raw in source:
|
x=dict(raw)
|
if not x['parent_code'] and x['file'].endswith('.xlsx'):
|
# Each top-level sheet corresponds to the imported TC product series, in sheet order.
|
sheets=json.loads((P/'xlsx-source.json').read_text(encoding='utf-8'))
|
tc='TC'+str(next(i for i,s in enumerate(sheets,1) if s['sheet']==x['sheet'])).zfill(4)
|
x['parent_code']=tc
|
parents=codes[x['parent_code']] if x['parent_code'] else names[norm(x['parent_name'])];children=codes[x['child_code']]
|
reason=[]
|
if len(parents)!=1:reason.append('父物料未找到或不唯一')
|
if len(children)!=1:reason.append('子物料未找到或不唯一')
|
if reason:issues.append(dict(x,reason='；'.join(reason)));continue
|
p,c=parents[0],children[0];x.update(parent_code=p['code'],parent_name=p['name'],child_name=c['name'],parent_id=int(p['id']),child_id=int(c['id']))
|
if p['status']!='0' or c['status']!='0':reason.append('物料未启用')
|
if p['id']==c['id']:reason.append('BOM自引用')
|
u=units.get(c['unit_measure_id']);base=u['code'] if u else ''
|
x['system_unit']=base
|
if not x['unit']:
|
if len(unit_evidence[x['child_code']])==1:
|
x['unit']=next(iter(unit_evidence[x['child_code']]));notes.append(dict(x,reason='单位取自同编码其他源表行'))
|
elif '--use-system-unit' in sys.argv:
|
x['unit']=base;notes.append(dict(x,reason='源表缺少单位，经用户确认按系统基本单位录入'))
|
else:reason.append('源表缺少单位且无同编码单位依据')
|
q=Decimal(x['quantity']);factor=Decimal(1)
|
if q<=0:reason.append('用量非正数')
|
if x['unit'] and x['unit']!=base:
|
masses={'KG':Decimal(1000),'GM':Decimal(1),'MG':Decimal('0.001')}
|
if x['unit'] in masses and base in masses:factor=masses[x['unit']]/masses[base]
|
else:reason.append('源表单位与系统单位不一致，缺少换算依据')
|
if reason:issues.append(dict(x,reason='；'.join(reason)));continue
|
exact=q*factor;stored=exact.quantize(Decimal('0.000001'),rounding=ROUND_HALF_UP)
|
if exact!=stored:notes.append(dict(x,reason=f'单位换算后为 {exact} {base}，按数据库6位小数保存为 {stored}；原始用量保留在备注'))
|
x['system_quantity']=str(stored);x['factor']=str(factor);accepted.append(x)
|
groups=collections.defaultdict(list)
|
for x in accepted:groups[(x['parent_id'],x['child_id'])].append(x)
|
unique=[];duplicates=[]
|
for pair,rows in groups.items():
|
if len({Decimal(x['system_quantity']) for x in rows})>1:
|
issues.extend(dict(x,reason='同一父子项存在不同用量') for x in rows);continue
|
x=rows[0];x['sources']=[f"{r['file']}/{r['sheet']}/{r['row']}" for r in rows]
|
if len(rows)>1:duplicates.extend(dict(r,reason='相同父子项、用量的重复行，不重复累加') for r in rows[1:])
|
unique.append(x)
|
existing=collections.defaultdict(list)
|
for x in db['bom']:existing[(int(x['item_id']),int(x['bom_item_id']))].append(x)
|
final=[]
|
for x in unique:
|
old=existing[(x['parent_id'],x['child_id'])]
|
if old and len(old)!=1:
|
issues.append(dict(x,reason='已有BOM有重复记录，保留已有数据'));continue
|
if old and Decimal(old[0]['quantity'])!=Decimal(x['system_quantity']):
|
if '--update-existing' not in sys.argv:
|
issues.append(dict(x,reason='已有BOM用量不同，保留已有数据'));continue
|
x['action']='更新';x['existing_id']=int(old[0]['id']);x['old_quantity']=old[0]['quantity']
|
else:x['action']='已存在' if old else '新增'
|
x['parent_unit_id']=codes[x['parent_code']][0]['unit_measure_id'];x['child_unit_id']=codes[x['child_code']][0]['unit_measure_id']
|
final.append(x)
|
graph=collections.defaultdict(set)
|
for x in db['bom']:graph[int(x['item_id'])].add(int(x['bom_item_id']))
|
for x in final:graph[x['parent_id']].add(x['child_id'])
|
def visit(n,stack,done):
|
if n in stack:raise ValueError(f'BOM环路 {n}')
|
if n in done:return
|
stack.add(n)
|
for child in graph.get(n,set()):visit(child,stack,done)
|
stack.remove(n);done.add(n)
|
done=set()
|
for n in list(graph):visit(n,set(),done)
|
plan=dict(target=target,source_count=len(source),lines=final,issues=issues,duplicates=duplicates,notes=notes)
|
(P/f'{target}-plan.json').write_text(json.dumps(plan,ensure_ascii=False,indent=2),encoding='utf-8')
|
fields=['file','sheet','row','parent_code','parent_name','child_code','child_name','quantity','unit','system_quantity','system_unit','action','reason']
|
for kind,rows in [('导入清单',final),('待核对',issues),('重复行',duplicates),('单位补全',notes)]:
|
with (P/f'{target}-{kind}.csv').open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.DictWriter(f,fields,extrasaction='ignore');w.writeheader();w.writerows(rows)
|
print(target,'源行',len(source),'BOM关系',len(final),'父项',len({x['parent_id'] for x in final}),'重复',len(duplicates),'待核对',len(issues),'单位补全',len(notes),collections.Counter(x['reason'] for x in issues))
|
