#!/usr/bin/env python3 # -*- coding: utf-8 -*- """核心货号合流：官网目录 × 应用场景推荐清单 × 竞品大表 × 内部产品清单 → 数据/diag_sku_core.json 设计原则：**只合并不生成**。任何字段若来源里没有，就留空并记入「缺口」，绝不编造。 """ import json import os import re from collections import defaultdict import pandas as pd ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' D = os.path.join(ROOT, '数据') OUT = os.path.join(D, 'diag_sku_core.json') # 诊断原料线：官网分类关键词（其余属生命科学/生物医药等其他线） DIAG_CATS = ['DNA聚合酶', 'Tth双功能酶', '反转录酶', 'UNG酶', '单管全预混PCR', '直扩PCR', '多重/PCR扩增系列', '单管全预混qRT-PCR', '多重qRT-PCR', '快速qRT-PCR', 'LAMP-DNA', 'LAMP-RNA', '恒温扩增', '其他辅助原料、试剂', '冻干保护剂', 'qPCR', 'qRT-PCR', 'PCR相关'] def norm_sku(s): s = str(s or '').upper().strip() s = re.sub(r'\s+', '', s) s = re.sub(r'^BR\d?', '', s) # BR1A401 → A401（科研线） return s def parent_of(sku): """取父货号：M2181-4-095203 → M2181-4；MD2104-000812 → MD2104；E04-2 → E04""" s = sku m = re.match(r'^([A-Z]+\d+(?:-\d)?)(?:-\d{3,})?$', s) return m.group(1) if m else s def sku_tokens(text): """从自由文本里抓货号候选（含 ① ② 编号前缀、换行）。""" t = str(text or '') out = [] for m in re.findall(r'[A-Za-z]{1,4}\d{1,4}(?:-\d{1,6}){0,3}(?:-[A-Za-z0-9]{1,8})?', t): u = norm_sku(m) if len(u) >= 3 and any(ch.isdigit() for ch in u): out.append(u) return out def load_official(): d = json.load(open(os.path.join(D, 'biori_official_catalog.json'), encoding='utf-8')) recs = [] for p in d['items']: docs = p.get('说明书与资料') or [] lab = (docs[0].get('label') or '') if docs else '' sku = norm_sku(re.sub(r'\s*说明书.*$', '', lab)) if lab else '' recs.append({'sku': sku, '品名': p['官网品名'], '官网分类': p.get('官网标注分类') or '', '官网详情页': p.get('详情页'), '说明书URL': (docs[0].get('file_url') if docs else ''), '说明书文件名': lab}) return recs def load_recommend(): """202607/202608/202611 产品推荐清单 → 场景记录。""" scen = [] for f in ('资料/IMA知识库同步/诊断原料/202611-产品推荐清单.xlsx', '资料/IMA知识库同步/诊断原料/202608-产品推荐清单.xlsx', '资料/IMA知识库同步/诊断原料/202607-产品推荐清单.xlsx', '产品验证资料/PCR-LAMP/202607-PCR-LAMP产品推荐清单.xlsx'): p = os.path.join(ROOT, f) if not os.path.exists(p): continue for sheet in pd.ExcelFile(p).sheet_names: df = pd.read_excel(p, sheet_name=sheet) if '技术方法' not in df.columns: continue cur_proj = '' for _, r in df.iterrows(): proj = str(r.get('项目类型') or '').strip() sub = str(r.get('Unnamed: 1') or '').strip() if proj and proj != 'nan': cur_proj = proj tech = str(r.get('技术方法') or '').strip() if not tech or tech == 'nan': continue scen.append({ '项目类型': cur_proj, '细分场景': (sub if sub != 'nan' else ''), '技术方法': tech, 'DNA_RNA': str(r.get('DNA/RNA') or '').replace('nan', '').strip(), '推荐体系原文': str(r.get('推荐体系') or '').strip(), '推荐货号': [s for s in {norm_sku(x) for x in sku_tokens(r.get('推荐体系'))}], '特征': str(r.get('特征') or '').replace('nan', '').strip(), '冻干推荐原文': str(r.get('推荐体系及优先级-冻干') or '').strip(), '冻干货号': [s for s in {norm_sku(x) for x in sku_tokens(r.get('推荐体系及优先级-冻干'))}], '样本类型': str(r.get('样本类型') or '').replace('nan', '').strip(), '快速程序': str(r.get('快速程序\n未排序') or '').replace('nan', '').strip(), '是否可全预混': str(r.get('是否可全预混\n未排序') or '').replace('nan', '').strip(), '来源文件': os.path.basename(p), }) return scen def load_competitors(): """生命科学产品线-竞品大表 → 宝锐货号级竞品对标。""" p = os.path.join(ROOT, '资料/生命科学产品线-竞品大表-V1.0.md') if not os.path.exists(p): return {} rows = [] for line in open(p, encoding='utf-8'): if not line.startswith('|'): continue cells = [c.strip() for c in line.strip().strip('|').split('|')] if len(cells) < 11 or cells[0] in ('关键步骤', '---') or set(cells[0]) <= set('-: '): continue rows.append({'关键步骤': cells[0], '具体操作': cells[1], '区域': cells[2], '宝锐货号': norms(cells[3]), '宝锐品名': cells[4], '对标竞品': cells[5], '竞品到手价': cells[6], '竞品优势': cells[7], '客户反馈': cells[8], '竞争策略': cells[9], '备注': cells[10]}) out = defaultdict(list) for r in rows: if r['宝锐货号']: out[r['宝锐货号']].append(r) return dict(out) def norms(s): return norm_sku(s) def load_internal_products(): """分子检测类试剂产品清单 → 货号 → 品名/规格/销售分组。""" p = os.path.join(ROOT, '经营分析/数据分析/分子检测类试剂产品清单-20260919.xlsx') out = {} if not os.path.exists(p): return out x = pd.ExcelFile(p) for s in ('产品清单-生产部内容', '产品清单-分子试剂产品部内容'): if s not in x.sheet_names: continue d = x.parse(s) for _, r in d.iterrows(): sku = norm_sku(r.get('产品货号')) if not sku: continue e = out.setdefault(sku, {'货号': sku, '品名': str(r.get('产品名称') or '').strip(), '规格': str(r.get('产品规格型号') or '').strip(), '对内外': str(r.get('产品对内货号') or '').strip(), '销售分组': str(r.get('销售分组') or '').strip(), '组分': [], '来源Sheet': s}) cn = str(r.get('组分名称') or '').strip() if cn and cn != 'nan' and len(e['组分']) < 12: e['组分'].append({'组分': cn, '用量': str(r.get('子项用量') or '').strip(), '父项用量': str(r.get('父项用量') or '').strip()}) return out def main(): official = load_official() scen = load_recommend() comp = load_competitors() prod = load_internal_products() # 官方目录：只留诊断原料线，且必须有说明书 diag = [o for o in official if o['sku'] and o['说明书URL'] and any(k in o['官网分类'] for k in DIAG_CATS)] print('官网产品 %d ｜ 诊断原料线且有说明书 %d' % (len(official), len(diag))) scen_skus = {s for x in scen for s in x['推荐货号']} | {s for x in scen for s in x['冻干货号']} comp_skus = set(comp.keys()) print('推荐清单涉及货号 %d ｜ 竞品大表涉及货号 %d' % (len(scen_skus), len(comp_skus))) core = {} for o in diag: core[o['sku']] = {'货号': o['sku'], '品名': o['品名'], '官网分类': o['官网分类'], '官网详情页': o['官网详情页'], '说明书URL': o['说明书URL'], '说明书文件名': o['说明书文件名'], '来源': ['官网说明书']} for s in scen_skus: core.setdefault(s, {'货号': s, '品名': '', '官网分类': '', '官网详情页': '', '说明书URL': '', '说明书文件名': '', '来源': []}) core[s].setdefault('来源', []) if '应用场景' not in core[s]['来源']: core[s]['来源'].append('应用场景') # 合并内部产品清单（补齐品名/规格/分组/组分） for s, r in core.items(): hit = prod.get(s) or prod.get(parent_of(s)) if hit: r.setdefault('品名内', hit['品名']) r['规格'] = hit['规格'] r['销售分组'] = hit['销售分组'] r['组分'] = hit['组分'] r['内部货号'] = hit['对内外'] # 场景关联 r['应用场景'] = [{'项目类型': x['项目类型'], '细分场景': x['细分场景'], '技术方法': x['技术方法'], 'DNA_RNA': x['DNA_RNA'], '特征': x['特征'], '样本类型': x['样本类型'], '快速程序': x['快速程序'], '是否可全预混': x['是否可全预混'], '来源': x['来源文件']} for x in scen if s in x['推荐货号'] or s in x['冻干货号']] # 竞品对标 c = comp.get(s) r['竞品对标'] = c or [] # 分类归一（用户口径） txt = (r.get('官网分类', '') + ' ' + r.get('销售分组', '') + ' ' + r.get('品名', '') + r.get('品名内', '')) labs = [] if re.search(r'LAMP|恒温|Bst|HW2', txt, re.I): labs.append('LAMP') if re.search(r'反转录|RTase|RT-|qRT|RNA', txt, re.I): labs.append('RNA') if re.search(r'DNA聚合酶|Taq|Tth|聚合酶|Pfu|高保真', txt, re.I): labs.append('单酶' if not re.search(r'Premix|预混|Mix|Master', txt, re.I) else '预混液') if re.search(r'预混|Premix|Mix|Master', txt, re.I) and '预混液' not in labs: labs.append('预混液') if re.search(r'PCR|扩增|UNG|dNTP|Buffer|BSA|RNase', txt, re.I) and 'DNA' not in labs: labs.append('DNA') r['分类'] = labs or ['未归类'] # 缺口统计 gaps = defaultdict(int) for s, r in core.items(): if not r.get('说明书URL'): gaps['无说明书'] += 1 if not r.get('应用场景'): gaps['无应用场景'] += 1 if not r.get('竞品对标'): gaps['无竞品对标'] += 1 have_all = [s for s, r in core.items() if r.get('说明书URL') and r.get('应用场景') and r.get('竞品对标')] res = {'生成时间': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M'), '口径': '只合并不生成；字段为空表示来源里没有，缺口逐项记录', '统计': {'核心货号数': len(core), '三字段齐全': len(have_all), '有说明书': sum(1 for r in core.values() if r.get('说明书URL')), '有应用场景': sum(1 for r in core.values() if r.get('应用场景')), '有竞品对标': sum(1 for r in core.values() if r.get('竞品对标'))}, '缺口': dict(gaps), '推荐场景数': len(scen), 'items': [core[k] for k in sorted(core)]} json.dump(res, open(OUT, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) print('统计:', res['统计']) print('缺口:', res['缺口']) print('→ %s（%.1f KB）' % (OUT, os.path.getsize(OUT) / 1024.0)) if __name__ == '__main__': main()