"""Name-only fuzzy candidates. Source scheme is retained but never used to score.
|
Random draws are persisted per product so re-running cannot silently remap rows.
|
Candidates are distinct from usable schemes: scope/type/status checks still apply.
|
"""
|
import json,re,csv,collections,secrets,difflib,unicodedata
|
from pathlib import Path
|
P=Path(__file__).parent
|
src=json.loads((P/'source.json').read_text(encoding='utf8'))
|
db=json.loads((P/'remote-snapshot.json').read_text(encoding='utf8'))
|
def norm(v):
|
s=unicodedata.normalize('NFKC',v or '').lower().replace('桑椹','桑葚').replace('黑枸杞','黑果枸杞')
|
s=re.sub(r'stp-qa-\([a-z]+\)\d+|sop-qc-[a-z]+-\d+','',s)
|
s=re.sub(r'\d{4}s|原料质检方案.*$|原料质量标准.*$|质量标准.*$|半成品内控标准.*$','',s)
|
s=s.replace('洞燕窝','燕窝').replace('洞燕盏','燕窝')
|
s=re.sub(r'出货检验|过程检验|来料检验|检验单|检验|即食|罐头|总统牌|小罐|碗装','',s)
|
s=s.replace('燕窝燕窝','燕窝')
|
s=re.sub(r'\((汁|尊享款|怡碗燕|青源堂|造养青年|自热型|代用茶)\)','',s)
|
return re.sub(r'[^\w\u4e00-\u9fffγ]','',s)
|
def score(a,b):
|
if not a or not b:return 0
|
if a==b:return 1.0
|
# Preserve meaningful product forms; ingredient containment stays a weak
|
# candidate and cannot establish applicability for a compound product.
|
if b in a or a in b:return .72+.18*min(len(a),len(b))/max(len(a),len(b))
|
return difflib.SequenceMatcher(None,a,b).ratio()
|
templates=db['mes_qc_template']
|
by_id={x['id']:x for x in templates}
|
selection_file=P/'random-selections.json'
|
saved=json.loads(selection_file.read_text(encoding='utf8')) if selection_file.exists() else {}
|
summary=[];plan=[];allcandidates=[]
|
items={str(x['code']):x for x in db['mdm_item']}
|
users={x['nickname'].strip():x for x in db['system_users']}
|
for kind,data in src.items():
|
typ='3' if kind=='oqc' else '2';codecounts=collections.Counter(x['检验单编号'] for x in data['rows'] if x['检验单编号'])
|
for name in dict.fromkeys(x['检验单名称'] for x in data['rows']):
|
core=norm(name)
|
ranked=sorted([(score(core,norm(t['name'])),t) for t in templates],key=lambda x:(-x[0],x[1]['id']))
|
best=ranked[0][0] if ranked else 0
|
# Only candidates close to the best name score; don't choose unrelated
|
# products merely to fill a field. An exact name takes precedence.
|
candidates=[t for v,t in ranked if v>=.67 and v>=best-.04]
|
eligible=[t for t in candidates if t['status']==0 and typ in t['types'].split(',')]
|
pool=eligible or candidates
|
key=kind+'|'+name
|
candidate_ids=[t['id'] for t in pool]
|
old=saved.get(key)
|
if old and old['candidate_ids']==candidate_ids and old['selected_id'] in candidate_ids:pick=by_id[old['selected_id']]
|
else:
|
pick=secrets.choice(pool) if pool else None
|
saved[key]={'candidate_ids':candidate_ids,'selected_id':pick['id'] if pick else None,'method':'同名或近似名称候选随机选择一次并留档；原表质检方案未用于匹配'}
|
info={'kind':kind,'name':name,'core':core,'candidate_ids':candidate_ids,'candidate_names':[t['name'] for t in pool],'selected_id':pick['id'] if pick else None,'selected_name':pick['name'] if pick else None,'score':round(best,4),'eligible':bool(eligible),'count':sum(x['检验单名称']==name for x in data['rows'])}
|
summary.append(info)
|
for v,t in ranked[:5]:allcandidates.append({'kind':kind,'name':name,'template_id':t['id'],'template_name':t['name'],'score':round(v,4),'candidate':t['id'] in candidate_ids,'types':t['types'],'status':t['status']})
|
for row in [x for x in data['rows'] if x['检验单名称']==name]:
|
issues=[];sourcecode=row['检验单编号']
|
if not sourcecode:issues.append('检验单编号为空：可在导入时生成，尚未分配')
|
elif codecounts[sourcecode]>1:issues.append('原表编号重复：需确认保留/重新编号规则')
|
if pick:
|
if typ not in pick['types'].split(','):issues.append('候选方案类型仅为来料检验，不支持'+('出货检验' if kind=='oqc' else '过程检验'))
|
if pick['status']!=0:issues.append('候选方案停用')
|
if core!=norm(pick['name']):issues.append('候选名称近似但产品形态/配方不同，待确认适用范围')
|
else:issues.append('无达到名称匹配阈值的方案')
|
item=None
|
if kind=='oqc':
|
ic=row['产品物料编码']
|
if ic is None:issues.append('产品物料编码为空')
|
else:
|
item=items.get(str(ic))
|
if not item:issues.append('产品物料编码在库中不存在：'+str(ic))
|
elif item['status']!=0:issues.append('物料停用')
|
if not row['客户名称']:issues.append('客户名称为空；无出货单关联数据可反查')
|
if row['本次出货数量'] is None:issues.append('本次出货数量为空')
|
else:
|
if not row['生产工单号']:issues.append('生产工单号为空；库中无工单记录')
|
if not row['工位名称']:issues.append('工位名称为空；库中无工位记录')
|
if row['合格品数量'] is None:issues.append('合格品数量为空')
|
if row['不合格品数量'] is None:issues.append('不合格品数量为空')
|
if row['检测数量'] is not None and row['合格品数量'] is not None and row['不合格品数量'] is not None and row['检测数量']!=row['合格品数量']+row['不合格品数量']:issues.append('检测数量与合格/不合格数量之和不符')
|
if row['检测人员'].strip() not in users:issues.append('检测人员未匹配')
|
plan.append({'kind':kind,'excel_row':row['excel_row'],'source':row,'mapping':info,'item_id':item['id'] if item else None,'item_name':item['name'] if item else None,'issues':issues,'ready':not issues})
|
(P/'random-selections.json').write_text(json.dumps(saved,ensure_ascii=False,indent=2),encoding='utf8')
|
(P/'mapping-plan.json').write_text(json.dumps(plan,ensure_ascii=False,indent=2),encoding='utf8')
|
(P/'name-summary.json').write_text(json.dumps(summary,ensure_ascii=False,indent=2),encoding='utf8')
|
for kind,label in [('oqc','出货检验'),('ipqc','过程检验')]:
|
with (P/(label+'-方案模糊映射及缺项.csv')).open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['原表行号','检验单编号','检验单名称','原表质检方案（仅留档）','随机候选方案ID','随机候选方案名称','候选数','候选ID列表','名称相似度','方案检验类型','候选方案状态','物料ID','物料名称','导入状态','暂缓原因'])
|
for x in sorted([x for x in plan if x['kind']==kind],key=lambda x:x['excel_row']):
|
r=x['source'];m=x['mapping'];t=by_id.get(m['selected_id'])
|
w.writerow([x['excel_row'],r['检验单编号'],r['检验单名称'],r['质检方案'],m['selected_id'],m['selected_name'],len(m['candidate_ids']),','.join(map(str,m['candidate_ids'])),m['score'],t['types'] if t else '',('启用' if t['status']==0 else '停用') if t else '',x['item_id'],x['item_name'],'可导入' if x['ready'] else '暂缓','；'.join(x['issues'])])
|
with (P/(label+'-名称候选.csv')).open('w',encoding='utf-8-sig',newline='') as f:
|
w=csv.writer(f);w.writerow(['检验单名称','方案ID','方案名称','相似度','是否随机候选','方案类型','方案状态'])
|
for x in allcandidates:
|
if x['kind']==kind:w.writerow([x['name'],x['template_id'],x['template_name'],x['score'],x['candidate'],x['types'],x['status']])
|
stats={kind:{'rows':len([x for x in plan if x['kind']==kind]),'ready':sum(x['ready'] for x in plan if x['kind']==kind),'name_candidate_rows':sum(bool(x['mapping']['candidate_ids']) for x in plan if x['kind']==kind),'no_name_candidate_rows':sum(not x['mapping']['candidate_ids'] for x in plan if x['kind']==kind),'multiple_name_candidate_rows':sum(len(x['mapping']['candidate_ids'])>1 for x in plan if x['kind']==kind),'issues':dict(collections.Counter(i for x in plan if x['kind']==kind for i in x['issues']))} for kind in src}
|
(P/'result.json').write_text(json.dumps(stats,ensure_ascii=False,indent=2),encoding='utf8')
|
print(json.dumps(stats,ensure_ascii=False))
|
for x in summary:print(x['kind'],x['name'],x['selected_name'],x['score'],x['candidate_ids'])
|
