"""Reviewed product aliases/specifications and direct BOM allocations; no invented material."""
|
import json,csv,re,copy,collections
|
from pathlib import Path
|
from decimal import Decimal
|
P=Path(__file__).parent
|
s=json.loads((P/'before-binding-snapshot.json').read_text(encoding='utf8'))
|
source=json.loads((P/'route-plan-with-ids.json').read_text(encoding='utf8'))
|
items={x['id']:x for x in s['mes_md_item']}
|
boms=collections.defaultdict(list)
|
for b in s['mes_md_product_bom']:boms[b['item_id']].append(b)
|
routes={r['id']:copy.deepcopy(r) for r in source['routes']}
|
# IDs reviewed against name + specification + packaging/formula components, not string score alone.
|
matches={
|
2:[524,525,526,987,988,989,1013],
|
3:[527,540,541,962,979,1151,1152,1153],
|
4:[523,974,992,1122],
|
5:[505,506,510],6:[500,503,509,533,534,985,986],
|
7:[507,508],8:[529,530,984,1157,1540],
|
9:[535,1155],10:[528,973],
|
11:[497,498,504,531,532,536,976,983,1508,1509],
|
13:[485,486,487,488,489,490,491,493,496,538,539,543,975,1544],
|
14:[537,1154],15:[494,495],16:[501],
|
18:[514],19:[511],20:[512],21:[515],22:[516],23:[517],24:[518],25:[519],26:[520,1158,1543],28:[521],29:[522]
|
}
|
exceptions=[
|
{'product_id':484,'reason':'已绑定现有路线1；保留既有绑定，候选为STP-SG-001-09。'},
|
{'product_id':502,'reason':'BOM是红、黑两种30ml袋拼装，不是文档红黑混合原浆；缺拼装规程。'},
|
{'product_id':513,'reason':'BOM分别使用红、黑包装膜，属于两种原浆拼装；缺拼装规程。'},
|
{'product_id':1510,'reason':'规格明确红原浆30ml和黑原浆30ml两种，不是混合原浆。'},
|
{'product_id':990,'reason':'有机黑果枸杞原浆缺有机原浆规程，不能套用有机干果或非有机原浆。'},
|
{'product_id':1121,'reason':'240g可匹配两份黑果枸杞规程，缺小包装和产品执行标准，暂不选版本。'},
|
{'product_id':1123,'reason':'枸杞沙棘混合原浆无对应配方规程。'},
|
]
|
# These items are already filled/sealed food despite several wrong MES categories.
|
filled={613,628,639,641,643,649,654,663,669,776,807,811,815,835,860}
|
pending=[];binding=[];allocations=[];output_rows=[]
|
def text(i):return (i['name'] or '')+' '+(i['specification'] or '')
|
def packaging(i):
|
n=text(i)
|
if i['id'] in filled:return False
|
return i['item_type_id'] in [20,21,22] or bool(re.search('复合膜|内袋|滤膜|瓶贴|瓶签|瓶身|瓶盖|胶盖|干燥剂|罐子|物流箱|垫板|防伪|标签|礼盒|外盒|外箱|封套|内托|格挡|套装勿拆|装饰贴|气泡袋|勺子|保护盒|纸箱|桶装',n))
|
def pick(r,names):
|
for n in names:
|
st=next((x for x in r['steps'] if x['name']==n),None)
|
if st:return st
|
raise AssertionError((r['code'],names))
|
def allocate(r,pid,b):
|
i=items[b['bom_item_id']];n=text(i)
|
if i['id'] in filled:return pick(r,['贴标','外包装']),'BOM已灌装/封装半成品；进入包装段'
|
if packaging(i):
|
if re.search('复合膜|内袋|滤膜|瓶身|瓶盖|胶盖|干燥剂|罐子|金属小罐|塑料小罐|亚克力瓶|直筒罐',n) or (i['item_type_id']==20 and not re.search('卡纸|桶装|款|纸盒|礼盒|外箱|垫板|内托|封套|保护盒',n)):
|
return pick(r,['热灌装','内包装','外包装']),'内包装材料随灌装/内包装投料'
|
if re.search('瓶贴|瓶签|前贴|后贴',n):return pick(r,['贴标','内包装','外包装']),'瓶身贴标材料'
|
return pick(r,['外包装']),'盒箱、内托、防伪及包装附件'
|
if r['document_id'] in [1,2,8,11]:
|
return pick(r,['人工选别','风选']),'干果原料/散装半成品按选别段接入；不得反推上游原料BOM'
|
if r['document_id']==3:return pick(r,['人工选别']),'已冻干枸杞按文档选别'
|
if r['document_id']==13:
|
return pick(r,['过筛','人工选别']),'现有BOM用料已为山药粉，接入粉体筛选段；非生山药'
|
if r['document_id'] in [7,10]:return pick(r,['混合']),'粉体/添加剂进入混合'
|
if r['document_id']==12:
|
return (pick(r,['提取过滤']),'黑果枸杞原料提取') if i['id']==836 else (pick(r,['混合定容','配制']),'复合果膏配制/定容')
|
if r['document_id']==23:
|
if re.search('^(人参|葛根|枳椇子)',i['name']) and not re.search('浓缩|提取|粉|汁|浸膏',n):return pick(r,['浸泡','煮提']),'植物原料浸泡煮提'
|
return pick(r,['配制高剪切','定容']),'浓缩液、低聚肽及辅料配制'
|
if r['document_id'] in [4,5,9,14,15,19,20,24]:
|
# BOM ingredients are prepared juices/concentrates, not unprocessed fruit.
|
return pick(r,['搅拌乳化剪切','预热保温']),'BOM为已制备原浆/浓缩汁，直接进入配制乳化段'
|
return pick(r,['配制','混合','搅拌乳化剪切']),'配方原辅料按文档配制'
|
|
existing={x['item_id']:x['route_id'] for x in s['mes_pro_route_product']}
|
for rid,pids in matches.items():
|
base=routes[rid]
|
for pid in pids:
|
assert pid in items and pid not in existing
|
bs=boms[pid]
|
assert len({x['bom_item_id'] for x in bs})==len(bs),'Duplicate source BOM must be reviewed'
|
# Assembly uses the actual stocked sealed unit; preserve complete source route separately.
|
pack=bool(bs) and any(b['bom_item_id'] in filled for b in bs) and all(b['bom_item_id'] in filled or packaging(items[b['bom_item_id']]) for b in bs)
|
if pack:
|
key=f'{rid}-PACK'
|
if key not in routes:
|
r=copy.deepcopy(base);r['id']=None;r['base_id']=rid;r['code']=base['code']+'-PACK'
|
r['name']=base['product']+('（'+base['variant']+'）' if base['variant'] else '')+'包装路线（'+base['document_code']+'）'
|
# Do not relabel sealed bottles unless the product BOM actually contains bottle labels.
|
use_label=any(re.search('瓶贴|瓶签',text(items[b['bom_item_id']])) for p in pids for b in boms[p])
|
r['steps']=[copy.deepcopy(st) for st in base['steps'] if st['name'] in (['贴标'] if use_label else [])+['外包装','成品检验','成品入库']]
|
r['description']='来源：'+base['document_code']+'；BOM投入已封装半成品，提取规程包装段；前段生产见完整来源路线；非倒冲、手动领料。'
|
routes[key]=r
|
r=routes[key]
|
else:r=base;key=rid
|
note='名称、规格与原文产品相符'
|
if pid in [528,535]:note='名称较泛；BOM包装明确黑果枸杞袋泡茶/即食藜麦粉R型'
|
if pid==508:note='名称原件疑为录入误字；BOM黑果枸杞原浆及包装明确罐头'
|
if pid==517:note='产品名称酸枣仁蜜桃味与营养素不一致；营养素包装及钙、蛋白、牛磺酸等BOM配方对应STP-YL-009'
|
if pid==521:note='名称未含燕窝；BOM燕窝肽、鱼胶原蛋白肽、GABA及瓶装对应STP-YL-007'
|
if pid==522:note='DHA海藻油名称别名；BOM藻油粉、磷脂酰丝氨酸等对应STP-YL-013'
|
row={'product_id':pid,'route_key':key,'base_id':rid,'document_code':base['document_code'],'evidence':note,'packaging_only':pack,'bom_count':len(bs)}
|
binding.append(row)
|
if not bs:pending.append({'product_id':pid,'reason':'已绑定产品及产出，未找到产品BOM，未虚构投料。'})
|
if pack:pending.append({'product_id':pid,'reason':'直接BOM仅含已封装半成品和包材，绑定包装段；半成品生产原料BOM缺失，未反推或重复投料。'})
|
for b in bs:
|
st,why=allocate(r,pid,b)
|
assert Decimal(b['quantity'])>0
|
allocations.append({'route_key':key,'process_id':st['process_id'],'process_name':st['name'],'product_id':pid,'item_id':b['bom_item_id'],'quantity':b['quantity'],'bom_id':b['id'],'reason':why})
|
output_rows.append({'route_key':key,'product_id':pid,'process_name':'外包装','output_item_id':pid,'basis':'随生产工单产品；系统空产出配置原生回退为工单产品，适用于同路线多SKU'})
|
|
# Manufacturing inputs with a definite packaged output from existing material master.
|
# Each route below has one matching container output; never use a comma list whose first SKU would win.
|
intermediate={2:('内包装',776),3:('内包装',860),7:('冷却风干',643),8:('水浴灭菌',807),14:('冷却风干',963),19:('冷却风干',669),26:('冷却风干',1158)}
|
# Bottle route8 has both 50ml and 60ml manufactured products with no BOM. Separate that container variant.
|
for row in binding:
|
if row['product_id'] in [1157,1540] and not row['packaging_only']:
|
key='8-50ML'
|
if key not in routes:
|
r=copy.deepcopy(routes[8]);r['id']=None;r['base_id']=8;r['code']+='-50ML';r['name']=r['product']+'50ml瓶装工艺路线（'+r['document_code']+'）';routes[key]=r
|
row['route_key']=key
|
for o in output_rows:
|
if o['product_id']==row['product_id']:o['route_key']=key
|
intermediate['8-50ML']=('水浴灭菌',1157)
|
# Native tasks copy workorder quantity without box-to-unit conversion. Manufacture stocked
|
# small units on their own UNIT routes; never switch a boxed workorder to a unit-pack ID.
|
intermediate={}
|
# Route for stocked unit packs: stop at the last unit-pack operation, without inventing an outer box.
|
# Existing filled component products are needed to manufacture the components consumed by assembly BOMs.
|
semi_groups={2:[776,1013],3:[860],4:[992],6:[639,663,815],7:[643],8:[807,1157],11:[628,649,811,835],14:[963],15:[613],16:[641],19:[669],26:[1158]}
|
for rid,pids in semi_groups.items():
|
base=routes[rid];key=f'{rid}-UNIT';r=copy.deepcopy(base)
|
r['id']=None;r['base_id']=rid;r['code']=base['code']+'-UNIT'
|
r['name']=base['product']+('（'+base['variant']+'）' if base['variant'] else '')+'小包装产出路线（'+base['document_code']+'）'
|
outer=next(i for i,st in enumerate(r['steps']) if st['name']=='外包装')
|
r['steps']=r['steps'][:outer]+[st for st in r['steps'][outer:] if st['name'] in ['成品检验','成品入库']]
|
r['description']='来源：'+base['document_code']+'；生产已封装小包装，止于内包/灌装及后处理，省略外盒包装；产出随工单产品；非倒冲、手动领料。'
|
routes[key]=r
|
for pid in pids:
|
old=next((b for b in binding if b['product_id']==pid),None)
|
if old:
|
assert old['bom_count']==0
|
old['route_key']=key
|
output_rows=[o for o in output_rows if o['product_id']!=pid]
|
else:
|
assert pid not in existing and not boms[pid]
|
binding.append({'product_id':pid,'route_key':key,'base_id':rid,'document_code':base['document_code'],'evidence':'现有小包装食品物料名称及规格；对应成品BOM将其作为封装半成品使用','packaging_only':False,'bom_count':0})
|
pending.append({'product_id':pid,'reason':'小包装产品/半成品已匹配路线和产出，但自身缺生产BOM；未从上级成品BOM反推制造投料。'})
|
output_rows.append({'route_key':key,'product_id':pid,'process_name':'内包装' if any(st['name']=='内包装' for st in r['steps']) else '热灌装','output_item_id':pid,'basis':'随生产工单小包装产品；省略外盒段，后续灭菌/冷却/检验仍保留，多个规格保持工单物料动态对应'})
|
used={b['route_key'] for b in binding}
|
finalroutes=[]
|
for key,r in routes.items():
|
if key not in used:continue
|
r['route_key']=key
|
for idx,st in enumerate(r['steps'],1):
|
key_stage='外包装' if any(x['name']=='外包装' for x in r['steps']) else ('内包装' if any(x['name']=='内包装' for x in r['steps']) else '热灌装')
|
st['sort']=idx*10;st['key_flag']=st['name']==key_stage;st['output_item_id']=None
|
if key in intermediate:
|
stage,outid=intermediate[key]
|
# Small unit workorders must not output a second differently coded stock item.
|
small=[b for b in binding if b['route_key']==key and b['product_id'] in [1013,1157,1158]]
|
if small:
|
# Finished small units still use dynamic current product; intermediate not fabricated.
|
pending.extend({'product_id':b['product_id'],'reason':'小包装本身是产出产品，采用工单产品动态产出；未另外生成不同编码半成品。'} for b in small)
|
else:
|
pick(r,[stage])['output_item_id']=str(outid)
|
output_rows.append({'route_key':key,'product_id':None,'process_name':stage,'output_item_id':outid,'basis':'原文包装规格与现有半成品物料一致；成品BOM或产品规格佐证'})
|
if any(st['output_item_id'] for st in r['steps']) and key==3:
|
# 979 is 90g bottles, whereas the other group are 4g cans.
|
if any(b['product_id']==979 and b['route_key']==key for b in binding):
|
pick(r,['内包装'])['output_item_id']=None
|
output_rows=[o for o in output_rows if not(o['route_key']==key and o['product_id'] is None)]
|
r['product_ids']=[b['product_id'] for b in binding if b['route_key']==key]
|
assert sum(st['key_flag'] for st in r['steps'])==1
|
assert all(not st['backflush_flag'] and (not st['starred'] or st['check_flag']) for st in r['steps'])
|
finalroutes.append(r)
|
assert len(binding)==len({b['product_id'] for b in binding})
|
assert len(allocations)==sum(b['bom_count'] for b in binding)
|
plan={'bindings':binding,'routes':finalroutes,'inputs':allocations,'outputs':output_rows,'exceptions':exceptions,'pending':pending}
|
(P/'binding-plan.json').write_text(json.dumps(plan,ensure_ascii=False,indent=2),encoding='utf8')
|
def write(name,headers,rows):
|
with (P/name).open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(headers);w.writerows(rows)
|
rt={r['route_key']:r for r in finalroutes}
|
write('产品路线匹配明细.csv',['产品ID','产品编码','产品名称','规格','路线编码','来源规程','仅包装段','BOM条数','匹配依据'],[[b['product_id'],items[b['product_id']]['code'],items[b['product_id']]['name'],items[b['product_id']]['specification'],rt[b['route_key']]['code'],b['document_code'],b['packaging_only'],b['bom_count'],b['evidence']] for b in binding])
|
write('工序投入匹配明细.csv',['路线编码','产品编码','产品名称','工序','投入编码','投入名称','规格','原BOM比例','BOM行ID','匹配依据'],[[rt[a['route_key']]['code'],items[a['product_id']]['code'],items[a['product_id']]['name'],a['process_name'],items[a['item_id']]['code'],items[a['item_id']]['name'],items[a['item_id']]['specification'],a['quantity'],a['bom_id'],a['reason']] for a in allocations])
|
write('工序产出匹配明细.csv',['路线编码','生产产品编码','生产产品名称','工序','产出编码','产出名称','规格','配置依据'],[[rt[o['route_key']]['code'],items[o['product_id']]['code'] if o['product_id'] else '适用路线产品',items[o['product_id']]['name'] if o['product_id'] else '',o['process_name'],items[o['output_item_id']]['code'],items[o['output_item_id']]['name'],items[o['output_item_id']]['specification'],o['basis']] for o in output_rows])
|
write('产品绑定及投入待核实清单.csv',['产品编码','产品名称','规格','事项'],[[items[e['product_id']]['code'],items[e['product_id']]['name'],items[e['product_id']]['specification'],e['reason']] for e in exceptions+pending])
|
print(json.dumps({'products':len(binding),'routes':len(finalroutes),'derived_routes':sum(r['id'] is None for r in finalroutes),'inputs':len(allocations),'outputs':len(output_rows),'unbound':len(exceptions),'no_bom':sum(b['bom_count']==0 for b in binding),'packaging_products':sum(b['packaging_only'] for b in binding)},ensure_ascii=False))
|
