#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 四季调养药膳 — 跨库交叉关联脚本 方向: A. 四季调养药膳 × 中华药膳全书(调理药膳336/配方库301): 名称精确+包含匹配 B. 四季调养药膳 × 大医网中药材1000: 配方中的药材名匹配(最长前缀+别名表) C. 四季调养药膳 × 大医网方剂1000: 名称精确+包含匹配 输出: 03_关联融合/四季调养-药膳全书_交叉关联.json 03_关联融合/四季调养-大医网_交叉关联.json """ import json, os, re, collections BASE = os.path.expanduser('~/Documents/ai_agent_scraper_study/data') SJSY = os.path.join(BASE, '四季调养药膳', '02_加工数据', '四季调养药膳.json') FUSION = os.path.join(BASE, '四季调养药膳', '03_关联融合') sjsy = json.load(open(SJSY, encoding='utf-8'))['recipes'] # ============ B 前置: 大医网药材名与别名表 ============ herb_dir = os.path.join(BASE, '大医网', '01_来源数据', '中药材') herb_names = [] # (名称, file_id) herb_alias = collections.defaultdict(set) # 别名 -> {正名} for fn in sorted(os.listdir(herb_dir)): if not fn.endswith('.json'): continue try: d = json.load(open(os.path.join(herb_dir, fn), encoding='utf-8')) except Exception: continue name = d.get('名称', '').strip() if not name: continue fid = fn.split('_')[0] herb_names.append((name, fid)) for al in re.split(r'[，,、;；\s]+', d.get('别名', '') or ''): al = al.strip() if 1 < len(al) <= 6: herb_alias[al].add(name) # 别名补充(药膳常用): EXTRA_ALIAS = {'杞子': '枸杞子', '红枣': '大枣', '圆肉': '龙眼肉', '苡仁': '薏苡仁', '薏仁': '薏苡仁', '苡米': '薏苡仁', '薏苡米': '薏苡仁', '麦门冬': '麦冬', '天门冬': '天冬', '云苓': '茯苓', '白苓': '茯苓', '银花': '金银花', '双花': '金银花', '虫草': '冬虫夏草', '广木香': '木香', '首乌': '何首乌', '制首乌': '何首乌', '白果': '银杏叶', '燕菜': '燕窝'} for a, n in EXTRA_ALIAS.items(): herb_alias[a].add(n) # 非药材词(调味/厨料, 不做药材匹配; 大医网无对应条目或属食材常识) NON_HERB = {'料酒', '绍酒', '黄酒', '米酒', '白酒', '素油', '菜油', '麻油', '香油', '味精', '食盐', '精盐', '白糖', '冰糖', '红糖', '蜂蜜', '酱油', '醋', '淀粉', '水淀粉', '玉米粉', '面粉', '米粉', '糯米粉', '药包', '网油', '清汤', '高汤', '鸡汤', '上汤', '开水', '温水', '凉水', '淘米水'} # 食材撞名别名黑名单: 大医网民间别名与食材同名, 禁止用作匹配 # (牛耳枫子别名"猪肚"、黄芪别名"羊肉"、车前草别名"猪肚子"、鱼胶别名"鱼肚/鱼白"、鲃鱼别名"青鱼") FOOD_COLLISION_ALIAS = {'猪肚', '猪肚子', '羊肉', '鱼肚', '鱼白', '青鱼', '蜂乳'} # 处理前缀(匹配时剥去): 炙/制/炒/煨/煅/焦/霜/法/飞/熟/生 PROC_PREFIX = re.compile(r'^(炙|制|炒|煨|煅|焦|霜|法|飞|熟|生)(?=[\u4e00-\u9fff]{2,})') # 最长前缀排序列表(含别名), 排除非药材词 all_terms = sorted(set([n for n, _ in herb_names] + list(herb_alias.keys())) - NON_HERB - FOOD_COLLISION_ALIAS, key=len, reverse=True) term2names = collections.defaultdict(set) for n, _ in herb_names: term2names[n].add(n) for a, names in herb_alias.items(): for n in names: term2names[a].add(n) # 炮制前缀(允许出现在药材名左侧)与形态后缀(允许出现在右侧) LEFT_OK = set('炙制炒煨煅焦法霜飞生熟粉鲜嫩干白') RIGHT_OK = set('片末粉霜茸丝') def extract_herbs(material_text): """从配方文本抽取药材名(最长前缀+span防重叠+炮制前缀/形态后缀宽容)""" found = collections.defaultdict(set) # 大医网正名 -> 原文写法 occupied = [] # 已命中的字符区间, 防止短词嵌在长词/调味词内部 def taken(a, b): return any(not (b <= s or a >= e) for s, e in occupied) ONE_CHAR_OK = {'葱', '梨', '藕'} # 大医网仅有的3个安全1字药名(边界检查防雪梨/料酒类误配) for term in all_terms: if len(term) < 2 and term not in ONE_CHAR_OK: continue # 1字词仅放行白名单 for m in re.finditer(re.escape(term), material_text): a, b = m.start(), m.end() if taken(a, b): continue # 左界: 前一字符须非汉字, 或为炮制前缀(且前缀+词不是更长的药材名) if a > 0 and '\u4e00' <= material_text[a-1] <= '\u9fff': if material_text[a-1] not in LEFT_OK or (material_text[a-1] + term) in all_terms: continue # 右界: 后一字符须非汉字, 或为形态后缀(且词+后缀不是更长的药材名) if b < len(material_text) and '\u4e00' <= material_text[b] <= '\u9fff': if material_text[b] not in RIGHT_OK or (term + material_text[b]) in all_terms: continue occupied.append((a, b)) for canon in term2names[term]: found[canon].add(m.group(0)) return found # ============ A: × 中华药膳全书 ============ book_files = { '调理药膳': os.path.join(BASE, '中华药膳全书学做药膳不生病', '01_来源数据', '调理药膳.json'), '药膳配方库': os.path.join(BASE, '中华药膳全书学做药膳不生病', '01_来源数据', '药膳配方库_recipes.json'), } book_records = [] for src, fp in book_files.items(): if os.path.exists(fp): for rec in json.load(open(fp, encoding='utf-8')): book_records.append({'来源': src, **{k: rec.get(k, '') for k in ('id', '名称', 'name', '症状', 'category')}}) norm = lambda s: re.sub(r'[\s（）()。,，、的]', '', s or '') book_by_norm = collections.defaultdict(list) for rec in book_records: nm = rec.get('名称') or rec.get('name') if nm: book_by_norm[norm(nm)].append(rec) link_book = [] for r in sjsy: hits = [] key = norm(r['名称']) if key in book_by_norm: for rec in book_by_norm[key]: hits.append({'匹配类型': '名称精确', '对方库': rec['来源'], '对方id': rec.get('id') or rec.get('category', ''), '对方名称': rec.get('名称') or rec.get('name')}) else: for rec in book_records: other = rec.get('名称') or rec.get('name') or '' if not other: continue no, nk = norm(other), key if len(no) >= 3 and (no in nk or nk in no): hits.append({'匹配类型': '名称包含', '对方库': rec['来源'], '对方id': rec.get('id') or rec.get('category', ''), '对方名称': other}) if hits: link_book.append({'本库id': r['id'], '本库名称': r['名称'], '季节': r['季节'], '关联数': len(hits), '关联': hits}) # ============ D: × 大医网药膳食疗1000 ============ diet_dir = os.path.join(BASE, '大医网', '01_来源数据', '药膳食疗') diet_idx = [] for fn in sorted(os.listdir(diet_dir)): if fn.endswith('.json'): try: d = json.load(open(os.path.join(diet_dir, fn), encoding='utf-8')) nm = (d.get('名称') or '').strip() if nm: diet_idx.append({'名称': nm, 'file_id': fn.split('_')[0]}) except Exception: pass # ============ C: × 大医网方剂 ============ formula_dir = os.path.join(BASE, '大医网', '01_来源数据', '方剂') formula_idx = {} for fn in sorted(os.listdir(formula_dir)): if fn.endswith('.json'): try: d = json.load(open(os.path.join(formula_dir, fn), encoding='utf-8')) nm = (d.get('名称') or d.get('方名') or '').strip() if nm: formula_idx[nm] = fn.split('_')[0] except Exception: pass link_dayi = [] herb_hit_count = collections.Counter() for r in sjsy: rec = {'本库id': r['id'], '本库名称': r['名称'], '季节': r['季节'], '性味': r['性味']} key = norm(r['名称']) # C: 方剂名称匹配 f_hits = [] for nm, fid in formula_idx.items(): nk = norm(nm) if nk == key or (len(nk) >= 3 and (nk in key or key in nk)): f_hits.append({'方剂': nm, 'file_id': fid}) rec['关联方剂'] = f_hits # D: 药膳食疗名称匹配(本库名称↔药膳食疗名称, 单向包含) d_hits = [] for it in diet_idx: nk = norm(it['名称']) if nk == key or (len(nk) >= 3 and (nk in key or key in nk)): d_hits.append({'药膳食疗': it['名称'], 'file_id': it['file_id']}) rec['关联药膳食疗'] = d_hits # B: 药材匹配(配方文本) herbs = extract_herbs(r['配方']) rec['关联药材'] = sorted(herbs.keys()) rec['药材原文写法'] = {k: sorted(v) for k, v in sorted(herbs.items())} for h in herbs: herb_hit_count[h] += 1 link_dayi.append(rec) os.makedirs(FUSION, exist_ok=True) with open(os.path.join(FUSION, '四季调养-药膳全书_交叉关联.json'), 'w', encoding='utf-8') as f: json.dump({'metadata': {'本库条数': len(sjsy), '对方库': '中华药膳全书(调理药膳336+配方库301)', '关联条数': len(link_book), '生成日期': '2026-09-08'}, 'cross_references': link_book}, f, ensure_ascii=False, indent=1) with open(os.path.join(FUSION, '四季调养-大医网_交叉关联.json'), 'w', encoding='utf-8') as f: json.dump({'metadata': {'本库条数': len(sjsy), '对方库': '大医网(方剂1000+药膳食疗1000+中药材1000)', '药材命中TOP30': herb_hit_count.most_common(30), '有药材关联条数': sum(1 for x in link_dayi if x['关联药材']), '有方剂关联条数': sum(1 for x in link_dayi if x['关联方剂']), '有药膳食疗关联条数': sum(1 for x in link_dayi if x['关联药膳食疗']), '生成日期': '2026-09-08'}, 'cross_references': link_dayi}, f, ensure_ascii=False, indent=1) print(f"药膳全书关联: {len(link_book)}/{len(sjsy)} 条") print(f"大医网: 有药材关联 {sum(1 for x in link_dayi if x['关联药材'])} 条, 有方剂关联 {sum(1 for x in link_dayi if x['关联方剂'])} 条, 有药膳食疗关联 {sum(1 for x in link_dayi if x['关联药膳食疗'])} 条") print("药材命中TOP15:", herb_hit_count.most_common(15)) print("\n抽样(苡仁炖猪蹄):") for x in link_dayi: if x['本库名称'] == '苡仁炖猪蹄': print(json.dumps(x, ensure_ascii=False, indent=1)) print("\n抽样(当归生姜羊肉汤):") for x in link_dayi: if x['本库名称'] == '当归生姜羊肉汤': print(json.dumps({k: x[k] for k in ('本库id','关联方剂','关联药膳食疗','关联药材')}, ensure_ascii=False, indent=1))