#!/usr/bin/env python3 # -*- coding: utf-8 -*- """宝锐分子诊断原料知识库 —— 数据层构建（合流，不生成）。 来源： 1) 数据/biori_official_catalog.json 官网 762 产品（品名/分类/详情页/说明书链接） 2) 数据/manual_tables.json 148 份说明书的体系配制表 + 反应程序（PDF 表格识别） 3) 资料/IMA知识库同步/诊断原料/*.xlsx 产品推荐清单（客户应用场景 × 推荐体系） 4) 经营分析/数据分析/分子检测类试剂产品清单-20260919.xlsx 内部货号/品名/规格/组分 5) 数据/activities_feishu.json 飞书活动（客户案例 + 竞品证据，原文可溯源） 产出：数据/diag_knowledge.json """ import json import os import re import sys from collections import Counter, defaultdict import pandas as pd sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) ROOT = '/Users/liuxinyuan/Desktop/Hermes输出-工作类' D = os.path.join(ROOT, '数据') OUT = os.path.join(D, 'diag_knowledge.json') DIAG_CATS = ['DNA聚合酶', 'Tth双功能酶', '反转录酶', 'UNG酶', '单管全预混PCR', '直扩PCR', '多重/PCR扩增系列', '单管全预混qRT-PCR', '多重qRT-PCR', '快速qRT-PCR', 'LAMP-DNA', 'LAMP-RNA', '恒温扩增', '其他辅助原料、试剂'] BRANDS = ['诺唯赞', '翌圣', '菲鹏', '瀚海新酶', '全式金', '康为世纪', '康为', '宝创', '致善', '迈迪安', '赛默飞', 'Thermo', 'Takara', '宝生物', '近岸', '义翘', '金斯瑞', '吐露港', '艾科瑞', 'ABclonal', '爱博泰克', 'Toyobo', 'NEB', 'Roche', '罗氏', 'Promega', 'Sigma', 'Merck', '默克', '碧云天', '生工', '擎科', '华大'] def N(s): return re.sub(r'\s+', '', str(s or '')).upper().strip() def parent(sku): s = N(sku) m = re.match(r'^([A-Z]{1,4}\d{1,4}(?:-\d)?)(?:-\d{3,})?$', s) return m.group(1) if m else s def toks(text): 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})?', str(text or '')): u = N(m) if len(u) >= 3 and any(c.isdigit() for c in u): out.append(u) return out def classify(rec): """确定性分类：官网标注分类为主，销售分组为辅。返回 (标签列表, 主分类)""" cat = rec.get('官网分类') or '' grp = rec.get('销售分组') or '' name = (rec.get('品名') or '') + (rec.get('品名内') or '') t = cat + ' ' + grp + ' ' + name labs = [] if re.search(r'LAMP|恒温', t, re.I): labs.append('LAMP') if re.search(r'DNA聚合酶|Tth双功能|反转录酶|UNG酶|化学修饰酶|抗体修饰|配体修饰|突变酶|高分辨酶|聚合酶', t) or re.search(r'^1\.01', grp): labs.append('单酶') if re.search(r'全预混|直扩|多重|Premix|Master', t, re.I) or re.search(r'^1\.0[34]', grp): labs.append('预混液') if re.search(r'反转录|qRT|RNA|RTase', t, re.I): labs.append('RNA') if re.search(r'PCR|DNA|UNG|扩增|dNTP', t, re.I): labs.append('DNA') if re.search(r'其他辅助|保护剂|Buffer|BSA|RNase Inhibitor|稀释液|dNTP|辅料', t, re.I): labs.append('辅料') order = ['LAMP', '单酶', '预混液', 'RNA', 'DNA', '辅料'] labs = [x for x in order if x in labs] or ['未归类'] return labs, labs[0] MARK = '①②③④⑤⑥⑦⑧⑨⑩⑪⑫⑬⑭⑮⑯⑰⑱⑲⑳' def split_numbered(text): """按 ①②③ 编号切分，返回 [段1, 段2, ...]；无编号则整段作为 1 段。""" t = str(text or '').replace('nan', '').strip() if not t: return [] idx = [(i, ch) for i, ch in enumerate(t) if ch in MARK] if not idx: return [t] out = [] for n, (i, _) in enumerate(idx): j = idx[n + 1][0] if n + 1 < len(idx) else len(t) out.append(t[i + 1:j].strip()) return out def main(): official = json.load(open(os.path.join(D, 'biori_official_catalog.json'), encoding='utf-8'))['items'] manuals = json.load(open(os.path.join(D, 'manual_tables.json'), encoding='utf-8'))['items'] # 产品资料（QA 正式版说明书）——本地目录，iCloud 物化后已解析 assets = {} _ap = os.path.join(D, 'asset_manuals.json') if os.path.exists(_ap): assets = json.load(open(_ap, encoding='utf-8'))['items'] # 宽松键（忽略 - _ 空格）：解决官网货号「HW206- M01」与磁盘文件名「HW206-_M01」不一致 loose = {re.sub(r'[^A-Z0-9]', '', k): k for k in manuals} acts = json.load(open(os.path.join(D, 'activities_feishu.json'), encoding='utf-8'))['items'] # ── 内部产品索引（按多个货号列建索引，提高命中率） prod = {} x = pd.ExcelFile(os.path.join(ROOT, '经营分析/数据分析/分子检测类试剂产品清单-20260919.xlsx')) for s in x.sheet_names: d = x.parse(s) if '产品货号' not in d.columns: continue for _, r in d.iterrows(): keys = {N(r.get('产品货号'))} for c in ('产品对内货号', '产品生产货号'): if c in d.columns: keys.add(N(r.get(c))) keys = {k for k in keys if k and k != 'NAN'} if not keys: continue rec = {'品名': str(r.get('产品名称') or '').strip(), '规格': str(r.get('产品规格型号') or '').strip(), '销售分组': str(r.get('销售分组') or '').strip(), 'BOM分组': str(r.get('BOM分组') or '').strip(), '货号': sorted(keys)[0], '组分': []} for k in keys: if k not in prod: prod[k] = dict(rec) for k in keys: cn = str(r.get('组分名称') or '').strip() if cn and cn != 'nan' and len(prod[k]['组分']) < 12: prod[k]['组分'].append({'组分': cn, '组分货号': N(r.get('组分货号')), '用量': str(r.get('子项用量') or '').strip()}) # ── 官网：诊断原料线（含说明书） off_by_sku = {} for p in official: docs = p.get('说明书与资料') or [] lab = (docs[0].get('label') or '') if docs else '' sku = N(re.sub(r'\s*说明书.*$', '', lab)) if lab else '' if not sku: continue e = off_by_sku.setdefault(sku, {'品名': p['官网品名'], '官网分类': p.get('官网标注分类') or '', '官网详情页': p.get('详情页'), '说明书URL': '', '说明书文件名': '', '说明书页数': 0}) if docs: e['说明书URL'] = docs[0].get('file_url') e['说明书文件名'] = lab # ── 场景（推荐清单） 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 sh in pd.ExcelFile(p).sheet_names: df = pd.read_excel(p, sheet_name=sh) if '技术方法' not in df.columns: continue cur = '' for _, r in df.iterrows(): pj = str(r.get('项目类型') or '').strip() if pj and pj != 'nan': cur = pj tech = str(r.get('技术方法') or '').strip() if not tech or tech == 'nan': continue # ①②③ 编号在「推荐体系」与「特征」间一一对应 → 精确落到货号 sys_items = split_numbered(r.get('推荐体系')) feat_items = split_numbered(r.get('特征')) dry_items = split_numbered(r.get('推荐体系及优先级-冻干')) pairs = [] for n, seg in enumerate(sys_items): ks = toks(seg) if not ks: continue pairs.append({'货号': ks[0], '推荐体系原文': seg, '特征': feat_items[n] if n < len(feat_items) else ''}) for seg in dry_items: ks = toks(seg) if not ks: continue pairs.append({'货号': ks[0], '推荐体系原文': seg, '特征': '（冻干体系）' + seg, '冻干': True}) scen.append({'项目类型': cur, '细分场景': str(r.get('Unnamed: 1') or '').replace('nan', '').strip(), '技术方法': tech, '核酸类型': str(r.get('DNA/RNA') or '').replace('nan', '').strip(), '推荐体系原文': str(r.get('推荐体系') or '').strip(), '推荐货号': sorted({p['货号'] for p in pairs}), '货号明细': pairs, '冻干推荐原文': str(r.get('推荐体系及优先级-冻干') or '').strip(), '特征': str(r.get('特征') or '').replace('nan', '').strip(), '样本类型': str(r.get('样本类型') or '').replace('nan', '').strip(), '来源': os.path.basename(p)}) scen_by_sku = defaultdict(list) for s in scen: for k in s['推荐货号']: scen_by_sku[k].append(s) # ── 案例/竞品证据（活动库，原文摘录） prep = [] for a in acts: t = str(a.get('text') or '') + ' ' + str(a.get('type') or '') prep.append({'日期': a.get('date'), '客户': a.get('customer'), '销售': a.get('sales'), 'low': t.lower(), 'raw': re.sub(r'\s+', ' ', str(a.get('text') or ''))[:200], 'brands': sorted({b for b in BRANDS if b in t})}) # ── 核心货号集合：官网诊断线（有说明书） ∪ 推荐清单货号 core = {} for sku, o in off_by_sku.items(): if any(k in o['官网分类'] for k in DIAG_CATS): core[sku] = {'货号': sku, '来源': ['官网']} for sku in scen_by_sku: core.setdefault(sku, {'货号': sku, '来源': []}) core[sku]['来源'].append('推荐清单') items = [] for sku, base in sorted(core.items()): o = off_by_sku.get(sku) or {} # 说明书三级匹配：全货号 → 父货号 → 去尾缀 mkey = next((k for k in (sku, parent(sku)) if k in manuals), None) if not mkey: mkey = loose.get(re.sub(r'[^A-Z0-9]', '', sku)) or loose.get(re.sub(r'[^A-Z0-9]', '', parent(sku))) if not mkey: cand = [k for k in manuals if k.startswith(sku[:5]) and len(sku) >= 5] mkey = cand[0] if len(cand) == 1 else None man = manuals.get(mkey) if mkey else None # 产品资料的说明书（QA 正式版）优先：更全、更新、含体系表 ast = assets.get(sku) or assets.get(parent(sku)) if not ast: _l = re.sub(r'[^A-Z0-9]', '', sku) ast = next((v for k, v in assets.items() if re.sub(r'[^A-Z0-9]', '', k) == _l), None) pr = prod.get(sku) or prod.get(parent(sku)) or {} rec = { '货号': sku, '品名': o.get('品名') or pr.get('品名') or '', '官网分类': o.get('官网分类') or '', '官网详情页': o.get('官网详情页') or '', '规格': pr.get('规格') or '', '销售分组': pr.get('销售分组') or '', 'BOM分组': pr.get('BOM分组') or '', '内部货号': pr.get('货号') or '', '组分': pr.get('组分') or [], '说明书URL': o.get('说明书URL') or '', '说明书文件名': (man or {}).get('说明书文件') or o.get('说明书文件名') or '', '说明书本地文件': (man or {}).get('说明书文件') or '', '说明书匹配键': mkey or '', '体系配制': (ast or man or {}).get('体系配制') or [], '反应程序': (ast or man or {}).get('反应程序') or [], '说明书_产品资料文件': (ast or {}).get('本地副本') or '', '说明书来源': ('产品资料(QA正式版)' if ast else ('官网' if man else '')), '应用场景': [], '案例与反馈': [], '竞品证据': [], '竞品品牌': {}, '来源': sorted(set(base['来源'] + (['说明书'] if man else []))), } for s in scen_by_sku.get(sku, []) + [s for k, v in scen_by_sku.items() if k == parent(sku) for s in v]: if s not in rec['应用场景']: rec['应用场景'].append(s) # 活动库：案例 + 竞品（原文可溯源） sk = sku.lower() par = parent(sku).lower() brand_cnt = Counter() for a in prep: hit = sk in a['low'] or (len(par) >= 5 and par in a['low']) if not hit: continue if a['brands']: brand_cnt.update(a['brands']) if len(rec['竞品证据']) < 8: rec['竞品证据'].append({'日期': a['日期'], '客户': a['客户'], '销售': a['销售'], '竞品': a['brands'], '原文': a['raw']}) elif len(rec['案例与反馈']) < 8: rec['案例与反馈'].append({'日期': a['日期'], '客户': a['客户'], '销售': a['销售'], '原文': a['raw']}) rec['竞品品牌'] = dict(brand_cnt.most_common(8)) rec['场景特征'] = [{'项目类型': s['项目类型'], '细分场景': s['细分场景'], '技术方法': s['技术方法'], '核酸类型': s['核酸类型'], '特征': p.get('特征', '')} for s in rec['应用场景'] for p in s.get('货号明细', []) if p['货号'] == sku] rec['冻干体系'] = any(p.get('冻干') for s in rec['应用场景'] for p in s.get('货号明细', []) if p['货号'] == sku) items.append(rec) # ── 后处理 0：产品资料说明书文件名里有完整品名 → 补空品名 for r in items: if not r['品名']: a = assets.get(r['货号']) or assets.get(parent(r['货号'])) if a and a.get('品名_自文件名'): r['品名'] = a['品名_自文件名'] r['品名来源'] = '产品资料说明书文件名(%s)' % a['本地副本'] # ── 后处理 1：同系列品名继承（明确标注来源，非编造） named = [r for r in items if r['品名']] for r in items: if r['品名']: r['品名来源'] = '官网' if r['说明书URL'] or r['官网分类'] else '内部产品清单' continue cands = [n for n in named if len(os.path.commonprefix([r['货号'], n['货号']])) >= 4] cands.sort(key=lambda n: -len(os.path.commonprefix([r['货号'], n['货号']]))) if cands: src = cands[0] pre = os.path.commonprefix([r['货号'], src['货号']]) r['品名'] = src['品名'] r['品名来源'] = '同系列继承(%s)' % src['货号'] r['同系列说明'] = '本货号为 %s 系列的规格/配方变体，品名取自同系列 %s（前缀 %s）' % ( pre, src['货号'], pre) if not r['官网分类']: r['官网分类'] = src['官网分类'] if not r['销售分组']: r['销售分组'] = src['销售分组'] else: r['品名来源'] = '待补（推荐清单未附品名，官网无此货号）' # ── 后处理 2：分类回退（前缀规则，确定性） for r in items: labs, main = classify(r) if labs == ['未归类'] or main == '未归类': s = r['货号'] if re.match(r'^FM', s): labs, main = (['预混液', '冻干'], '预混液') if 'FM' else labs elif re.match(r'^AS(212|412|44)', s): labs, main = ['辅料'], '辅料' elif re.match(r'^M[DM]?\d', s): labs, main = ['预混液'], '预混液' elif re.match(r'^E\d', s): labs, main = ['单酶'], '单酶' r['分类标签'], r['主分类'] = labs, main # ── 后处理 3：无品名货号补「同场景关联主货号」（仅关联提示，不认定同一产品） for r in items: if r['品名']: continue peers = [] for s in r['应用场景']: for p in s.get('货号明细', []): if p['货号'] != r['货号'] and not p.get('冻干'): peers.append({'货号': p['货号'], '项目类型': s['项目类型'], '技术方法': s['技术方法']}) seen, uniq = set(), [] for p in peers: if p['货号'] not in seen: seen.add(p['货号']) uniq.append(p) r['同场景关联货号'] = uniq[:6] stat = { '核心货号数': len(items), '有说明书': sum(1 for r in items if r['说明书URL'] or r['说明书文件名'] or r.get('说明书_产品资料文件')), '有体系配制表': sum(1 for r in items if sum(len(x['表格']) for x in r['体系配制'])), '有反应程序': sum(1 for r in items if r['反应程序']), '有应用场景': sum(1 for r in items if r['应用场景']), '有案例反馈': sum(1 for r in items if r['案例与反馈']), '有竞品证据': sum(1 for r in items if r['竞品证据']), '分类分布': dict(Counter(r['主分类'] for r in items)), } res = {'生成时间': pd.Timestamp.now().strftime('%Y-%m-%d %H:%M'), '口径': {'金额': '不涉及', '说明书': '官网产品页 PDF（阿里云 OSS），已下载 148 份', '体系配制与程序': '说明书 PDF 表格识别原文（find_tables），保留页码', '应用场景': '202607/202608 产品推荐清单（客户项目 × 技术方法）', '案例与竞品': '飞书活动库原文摘录（客户/日期可溯源）；诊断原料线无货号级竞品对标表', '分类': '官网标注分类为主 + 内部销售分组为辅，确定性规则'}, '统计': stat, 'items': items} json.dump(res, open(OUT, 'w', encoding='utf-8'), ensure_ascii=False, indent=1) print(json.dumps(stat, ensure_ascii=False, indent=1)) print('\n→ %s（%.0f KB）' % (OUT, os.path.getsize(OUT) / 1024.0)) if __name__ == '__main__': main()