""" 菌群报告 PDF 解析器 从 extract_full_report_v5.py 提取核心逻辑，封装为可调用函数 """ import os import re from PyPDF2 import PdfReader # === 常量 === # Kangxi 部首 → CJK 统一汉字（与 extract_full_report_v5.py 完全一致） RADICAL_MAP = { '\u2f18': '卜', '\u2f1f': '土', '\u2f24': '大', '\u2f26': '子', '\u2f29': '小', '\u2f2d': '山', '\u2f32': '干', '\u2f3c': '心', '\u2f42': '文', '\u2f46': '无', '\u2f4a': '木', '\u2f50': '比', '\u2f54': '水', '\u2f55': '火', '\u2f5c': '牛', '\u2f5f': '玉', '\u2f60': '瓜', '\u2f62': '甘', '\u2f63': '生', '\u2f64': '用', '\u2f69': '白', '\u2f6a': '皮', '\u2f6c': '目', '\u2f6f': '石', '\u2f75': '竹', '\u2f76': '米', '\u2f7a': '羊', '\u2f7b': '羽', '\u2f7c': '老', '\u2f7f': '耳', '\u2f81': '肉', '\u2f90': '衣', '\u2f95': '谷', '\u2f96': '豆', '\u2f9d': '身', '\u2fa6': '金', '\u2faf': '面', '\u2fb2': '韭', '\u2fb9': '香', '\u2fca': '黑', '\u2ec9': '贝', '\u2edd': '食', '\u2ee2': '马', '\u2ee5': '鱼', '\u2ee8': '麦', '\u2ee9': '黄', '\u2ef0': '龙', # 氏 radical U+2F52 → U+6C0F '\u2f52': '氏', # 以下为本项目报告 PDF 中实际出现的部首字形（U+2F00-Kangxi / U+2E80-CJK Radicals） '\u2f00': '一', '\u2f04': '乙', '\u2f06': '二', '\u2f08': '人', '\u2f09': '儿', '\u2f0a': '入', '\u2f0f': '几', '\u2f12': '力', '\u2f17': '十', '\u2f1c': '又', '\u2f1d': '口', '\u2f20': '士', '\u2f25': '女', '\u2f2b': '尸', '\u2f2f': '工', '\u2f34': '广', '\u2f38': '弓', '\u2f3f': '手', '\u2f40': '支', '\u2f45': '方', '\u2f47': '日', '\u2f49': '月', '\u2f4c': '止', '\u2f51': '毛', '\u2f53': '气', '\u2f5a': '片', '\u2f5b': '牙', '\u2f5d': '犬', '\u2f70': '示', '\u2f74': '立', '\u2f79': '网', '\u2f7d': '而', '\u2f83': '自', '\u2f84': '至', '\u2f86': '舌', '\u2f8a': '色', '\u2f8d': '虫', '\u2f8e': '血', '\u2f8f': '行', '\u2f94': '言', '\u2f9b': '走', '\u2f9c': '足', '\u2f9f': '辛', '\u2fa5': '里', '\u2fae': '非', '\u2fb0': '革', '\u2fb3': '音', '\u2fb8': '首', '\u2fbc': '高', '\u2fcf': '鼠', '\u2fd0': '鼻', '\u2e9f': '母', '\u2ec5': '见', '\u2ec6': '角', '\u2ed3': '长', '\u2ed4': '门', '\u2ed8': '青', '\u2ed9': '韦', '\u2edb': '风', '\u2ee3': '骨', '\u2eec': '齐', } KNOWN_MACRO = ['碳水化合物', '蛋白质', '脂肪', '纤维素', '乳制品'] KNOWN_AMINO = ['苏氨酸', '异亮氨酸', '亮氨酸', '赖氨酸', '蛋氨酸', '胱氨酸', '苯丙氨酸', '酪氨酸', '缬氨酸', '组氨酸', '丙氨酸', '丝氨酸', '甘氨酸', '脯氨酸', '谷氨酸', '天门冬氨酸', '天冬氨酸', '天冬酰胺', '谷氨酰胺', '精氨酸', '色氨酸'] KNOWN_VITAMINS = ['维生素A', '维生素B1', '维生素B2', '维生素B5', '维生素B6', '叶酸', '维生素B12', '维生素C', '维生素D', '维生素K2', '维生素E'] KNOWN_TRACE = ['铁', '锌'] KNOWN_DISEASE_RISKS = ['炎症性肠炎', '肠易激综合征', '感染性腹泻', '自闭症', '抑郁症', '甲状腺疾病', '肺部感染或疾病', '自体免疫病', '结直肠癌', '肥胖', '便秘', '过敏', '失眠', '肝病', '肾病', '胃病', '胆病', '心脑血管疾病', 'II型糖尿病'] KNOWN_BARRIER = ['肠道炎症水平', '肠道产气', '肠道屏障', '脂多糖LPS', '次级胆汁酸', '对甲酚（p-Cresol）', '吲哚', '苯酚', '腐胺', '硫化氢', '尸胺'] KNOWN_SCFA = ['丁酸盐（Butyrate）', '丙酸盐（Propionate）', '乙酸盐（Acetate）', '异戊酸盐（Isovaleric）'] KNOWN_NEURO = ['血清素（5-HT）', 'γ-氨基丁酸（GABA）', '谷氨酸（Glutamate）', '色氨酸（Tryptophan）', 'DOPAC', '多巴胺', '组胺（Histamine）', '一氧化氮', '喹啉（Quinolinic）', '维生素K2', '肌醇（Inositol）', '肾上腺素', '去甲肾上腺素', '乙酰胆碱', '皮质醇'] KNOWN_ANTIBIOTICS = ['β-内酰胺酶类', '氨基糖苷类', '大环内酯类', '呋喃类', '喹诺酮类', '磺胺类', '甲氧苄啶类', '氯霉素类', '四环素类'] KNOWN_PATHOGENS = ['幽门螺杆菌', '艰难梭菌', '沙门氏菌', '志贺氏菌', '弯曲杆菌'] # === 菌群表标题（完整版 extract_full_report_v5.py 移植） === BACTERIA_TABLE_TITLES = [ '核心菌属构成表', '益生菌', '有害菌属构成表', '其它重要菌属构成表', '病原菌属构成表', ] PHYLUM_TABLE_TITLES = ['菌门构成表', '菌群门水平构成表', '门水平菌群构成'] CLASS_TABLE_TITLES = ['菌纲构成表', '菌群纲水平构成表', '纲水平菌群构成'] ORDER_TABLE_TITLES = ['菌目构成表', '菌群目水平构成表', '目水平菌群构成'] FAMILY_TABLE_TITLES = ['菌科构成表', '菌群科水平构成表', '科水平菌群构成'] GENUS_TABLE_TITLES = ['菌属构成表', '菌群属水平构成表', '属水平菌群构成'] SPECIES_TABLE_TITLES = ['菌种构成表', '菌群种水平构成表', '种水平菌群构成'] DISEASE_BACTERIA_TITLES = [ '肥胖相关菌', '便秘相关菌', '抑郁相关菌', '过敏相关菌', '腹胀相关菌', '失眠相关菌', '肠道健康相关菌', '多动症相关菌', '自闭症相关菌', ] # === 食物推荐表（完整版 extract_full_report_v5.py 移植） === COLUMNS_FOOD = ['名称', '分类', '推荐指数', '能量KJ', '蛋白g', '脂肪g', '碳水化合物g', '淀粉g', '总膳食纤维g', '胆固醇mg'] KNOWN_CATS = ['主食', '乳制品', '干果', '坚果', '快餐', '水产品', '水果', '汤', '肉类', '蔬菜', '豆类及豆制品', '蛋类', '饮料'] FOOD_SKIP_TEXTS = [ '根据您的肠道菌群', '分值从-100', '食物推荐考虑', '食物推荐是综合', '需要注意的是', '本饮食推荐', '该饮食推荐根据', '后续表格中的营养', '16S 高通量测序', '基于机器学习和', '肠道菌群健康检测报告说明', '检测方法及局限性', '数据分析及模型', '结果解读及使用', '影响因素说明', '建议将检测结果', '营养建议说明', '重要提示', '推荐食物清单', '实际食用时需结合', '如有特殊疾病', '免责声明', '本检测报告仅供', '以上模型预测', '正常范围的定义', '募极生物', ] def norm(s): return ''.join(RADICAL_MAP.get(c, c) for c in s) def detect_format(pages_text): """检测 triplet / inline 格式（v5 detect_format 移植，页面级判定） 仅当同一页同时含'疾病风险评估'+'指标范围'且存在内嵌数字的行时才算 inline， 否则按 triplet 处理（标准模板报告为 triplet，报告类 PDF 为 inline）。 """ for pt in pages_text: t = norm(pt) if '疾病风险评估' in t and '指标范围' in t: for line in t.split('\n'): ls = line.strip() if not ls: continue if re.search(r'[\u4e00-\u9fff]+\d+\.?\d*[\u4e00-\u9fff]+', ls): return 'inline' return 'triplet' return 'triplet' def extract_text(file_path): """读取 PDF 并提取文本""" reader = PdfReader(file_path) lines = [] for page in reader.pages: text = norm(page.extract_text() or '') for line in text.split('\n'): ls = line.strip() if ls: lines.append(ls) return lines def parse_overview(lines): """提取报告概述""" text = '\n'.join(lines) r = {} m = re.search(r'编号[：:\s]*(\d+)', text) if m: r['report_number'] = m.group(1) m = re.search(r'姓名[：:\s]*([\u4e00-\u9fff]{2,10})', text) if m: r['person_name'] = re.sub(r'(编号|年龄|性别|备注|肠道).*', '', m.group(1))[:4] m = re.search(r'年龄[：:\s]*(\d+)', text) if m: r['age'] = int(m.group(1)) m = re.search(r'性别[：:\s]*([\u4e00-\u9fff])', text) if m: r['gender'] = 'male' if m.group(1) == '男' else 'female' for kw in ['健康总分', '菌群健康', '慢病控制', '营养均衡', '肠道菌群平衡', '菌群多样性', '有益菌', '有害菌', '核心菌属']: m = re.search(rf'{kw}\s*(\d+)', text) if m: r[kw] = int(m.group(1)) m = re.search(r'肠道预测年龄[：:\s]*([\d.]+)', text) if m: r['gut_age'] = m.group(1) m = re.search(r'肠型[：:\s]*(\S+)', text) if m: r['gut_type'] = m.group(1) return r def parse_triplet_until(lines, stop_markers): """三元组解析：3行一组 名称/数值/状态""" results = [] i = 0 while i < len(lines): if any(lines[i] == sm or lines[i].startswith(sm) for sm in stop_markers): break if lines[i] in ('指标范围', '名称', '丰度', '评估'): i += 1 continue name = lines[i] if i + 2 >= len(lines): break val = lines[i + 1] status = lines[i + 2] if re.match(r'^-?\d+\.?\d*$', val): results.append({'name': name, 'value': val, 'status': status}) i += 3 else: i += 1 return results # ========================================== # 完整分区提取（移植自 extract_full_report_v5.py，修复分区错位） # 原问题：疾病风险 stop 标记缺'营养状况评估'导致吞并营养/氨基酸； # 维生素/微量元素标题在数据后（Case B）导致为空；屏障/SCFA/神经递质/病原菌/抗生素缺失 # ========================================== def _extract_all_triplet(all_lines): """按'指标范围'数据页标记分段提取（v5 extract_all_triplet 移植） 数据页索引约定（v5 实证）： data_pages[0] = 疾病风险页（'疾病风险评估'标题前也有'指标范围'） data_pages[1] = 营养状况评估页（含前5个氨基酸：苏氨酸/异亮氨酸/亮氨酸/赖氨酸/蛋氨酸） data_pages[2] = 剩余氨基酸页 data_pages[3] = 维生素评估页（含微量元素） """ data_pages = [] for i, line in enumerate(all_lines): if line == '指标范围': data_pages.append(i) result = {} # 数据页1: 疾病风险评估（按已知疾病名过滤在调用方做） if len(data_pages) >= 1: i = data_pages[0] + 1 if i < len(all_lines) and all_lines[i] == '疾病风险评估': i += 1 rows = parse_triplet_until(all_lines[i:], ['主要营养评估', '氨基酸评估', '维生素评估', '微量元素评估', '主要消化道致病菌', '抗生素风险评估', '抗生素耐药风险']) result['disease_risks'] = [r for r in rows if '注：' not in r['name'] and '注:' not in r['name']] # 数据页2: 营养状况评估（含前5个氨基酸） if len(data_pages) >= 2: i = data_pages[1] + 1 if i < len(all_lines) and all_lines[i] == '营养状况评估': i += 1 rows = parse_triplet_until(all_lines[i:], ['主要营养评估', '氨基酸评估']) result['nutrition_rows'] = rows[:5] result['amino_p2'] = rows[5:10] if len(rows) > 5 else [] # 数据页3: 剩余氨基酸 if len(data_pages) >= 3: i = data_pages[2] + 1 if i < len(all_lines) and '氨基酸' in all_lines[i]: while i < len(all_lines) and '氨基酸' in all_lines[i]: i += 1 remaining = parse_triplet_until(all_lines[i:], ['氨基酸评估']) result['amino_p3'] = remaining # 数据页4: 维生素评估 + 微量元素 if len(data_pages) >= 4: i = data_pages[3] + 1 if i < len(all_lines) and all_lines[i] == '维生素评估': i += 1 vit_rows = parse_triplet_until(all_lines[i:], ['微量元素评估', '主要消化道致病菌']) vit, trace = [], [] for r in vit_rows: if '维生素' in r['name']: vit.append(r) else: trace.append(r) result['vitamin_rows'] = vit result['trace_rows'] = trace # 主要消化道致病菌（特殊格式：致病菌/丰度%/评估） for i, line in enumerate(all_lines): if line == '主要消化道致病菌': j = i + 1 while j < len(all_lines) and all_lines[j] in ('致病菌', '丰度', '评估'): j += 1 path_rows = [] while j < len(all_lines) and not all_lines[j].startswith('肠道屏障'): name = all_lines[j] if j + 2 < len(all_lines) and re.match(r'^\d+%$', all_lines[j+1]): path_rows.append({'name': name, 'value': all_lines[j+1], 'status': all_lines[j+2]}) j += 3 else: j += 1 result['pathogens'] = path_rows break # 抗生素风险评估（去重：同名只保留第一条） for i, line in enumerate(all_lines): if line == '抗生素风险评估': abx_rows = parse_triplet_until(all_lines[i+1:], ['抗生素耐药风险', '个体化食物推荐表']) seen = set() deduped = [] for r in abx_rows: n = r.get('name', '') if n and n not in seen: seen.add(n) deduped.append(r) result['antibiotics'] = deduped break return result def _extract_barrier_and_scfa_triplet(all_lines): """肠道屏障及代谢物/短链脂肪酸/神经递质及激素 - 已知指标名精确匹配（v5 移植）""" results = {'barrier': [], 'scfa': [], 'neurotransmitter': []} current_section = None i = 0 while i < len(all_lines): line = all_lines[i] if line == '肠道屏障及菌群代谢物': current_section = 'barrier' i += 1 continue if '短链脂肪酸' in line: current_section = 'scfa' i += 1 continue if '神经递质' in line: current_section = 'neurotransmitter' i += 1 continue if current_section is None: i += 1 continue # 跳过头行 if line in ('名称', '评估值', '正常范围', '过量', '缺乏', '相关症状') or \ '过量 /' in line or '缺乏' in line: i += 1 continue # 提前终止 if current_section == 'barrier' and line in ('短链脂肪酸', '抗生素风险', '抗生素耐药风险'): break if current_section == 'scfa' and line in ('神经递质', '神经递质及激素指标', '神经递质及激素', '抗生素风险'): break if current_section == 'neurotransmitter' and line in ('抗生素风险', '个体化食物推荐表'): break # 只处理已知指标名 known = {'barrier': KNOWN_BARRIER, 'scfa': KNOWN_SCFA, 'neurotransmitter': KNOWN_NEURO} known_list = known[current_section] matched_name = None for kn in sorted(known_list, key=len, reverse=True): if line.startswith(kn) or line == kn: matched_name = kn break if matched_name is None: i += 1 continue # 模式1：整行解析 名称+数值+状态+范围（inline 单行） parts = line.split() if len(parts) >= 2: vi = None for pi, p in enumerate(parts): if re.match(r'^\d+$', p) and pi >= 1: vi = pi break if vi: val = parts[vi] status = '' range_ = '' for rp in parts[vi+1:]: if rp in ('正常', '过多', '轻度产气', '过低', '不足', '缺乏', '不⾜'): status = rp elif re.match(r'^\d+-\d+$', rp): range_ = rp results[current_section].append({ 'name': matched_name, 'value': val, 'status': status, 'refRange': range_ }) i += 1 continue # 模式2：下一行是数值（三元组） if i + 1 < len(all_lines) and re.match(r'^\d+$', all_lines[i+1]): val = all_lines[i+1] status = '' range_ = '' for j in range(2, min(6, len(all_lines)-i)): if all_lines[i+j] in ('正常', '过多', '轻度产气', '过低', '不足', '缺乏', '不⾜'): status = all_lines[i+j] elif re.match(r'^\d+-\d+$', all_lines[i+j]): range_ = all_lines[i+j] results[current_section].append({ 'name': matched_name, 'value': val, 'status': status, 'refRange': range_ }) i += 2 continue i += 1 # 去重（每个 section 内每个指标名只保留一条） for section in ('barrier', 'scfa', 'neurotransmitter'): seen = set() deduped = [] for item in results[section]: n = item.get('name', '') if n and n not in seen: seen.add(n) deduped.append(item) results[section] = deduped return results def _extract_names_from_region(region_str, known_names, end_pos): """从区域文本中提取已知名称+数值+状态（v5 移植）""" for h in ['肠道屏障及菌群代谢物名称评估值正常范围过量 / 缺乏相关症状', '名称评估值正常范围过量 / 缺乏相关症状', '短链脂肪酸评估值正常范围过量 / 缺乏相关症状', '神经递质及激素评估值正常范围过量 / 缺乏相关症状', '过量 / 缺乏相关症状']: region_str = region_str.replace(h, '') positions = [] for kn in known_names: # 找所有匹配位置，选第一个后面有数字（允许空格）且不是头部标记的 search_start = 0 while True: ii = region_str.find(kn, search_start) if ii == -1: break after = region_str[ii+len(kn):ii+len(kn)+10].lstrip() if re.match(r'\d', after): positions.append((ii, kn)) break # 找到第一个有效匹配 search_start = ii + 1 if not positions: return [] # 去重：同名先后取前面（较精准的），不同名重叠取较长的 positions.sort() filtered = [] for i, (pos, name) in enumerate(positions): has_longer = False for j, (pos2, name2) in enumerate(positions): if i != j and pos2 <= pos and pos + len(name) <= pos2 + len(name2) and len(name2) > len(name): has_longer = True break if not has_longer: filtered.append((pos, name)) positions = filtered seen_names = {} for pos, name in positions: if name not in seen_names: seen_names[name] = pos positions = [(pos, name) for name, pos in seen_names.items()] positions.sort() local = [] for i, (ii, kn) in enumerate(positions): start_after = ii + len(kn) after_text = region_str[start_after:].lstrip() vm = re.match(r'(\d+(?:\.\d+)?)', after_text) if not vm: continue val = vm.group(1) whitespace_skipped = len(region_str[start_after:]) - len(after_text) val_start = start_after + whitespace_skipped val_end = val_start + len(val) if i + 1 < len(positions): next_idx = positions[i + 1][0] status = region_str[val_end:next_idx].strip() else: status = region_str[val_end:end_pos].strip() # 清理状态中的残留标记 status = re.sub( r'注意大环内酯类|注意呋喃类|注意氯霉素类|注意喹诺酮类|注意磺胺类|' r'注意甲氧苄啶类|注意四环素类|正常氨基酸评估|正常维生素评估|' r'正常微量元素评估|指标范围|疾病风险评估|营养状况评估|' r'抗生素风险评估|抗生素耐药风险|主要消化道致病菌', '', status).strip() # 截断：状态中出现'注'（如'注：本检测仅用于...'）只取之前部分 status = re.split(r'注[:：]', status)[0].strip() if len(status) > 20: status = status[:20] local.append({'name': kn, 'value': val, 'status': status}) return local def _extract_inline_module(all_text, start_marker, end_markers, known_names): """从 inline 文本中提取已知名称的指标（v5 移植） 支持两种布局： A) 标题在前、数据在后（标准） B) 标题在后、数据在前（'指标范围'与标题之间） """ sm_idx = all_text.find(start_marker) if sm_idx == -1: return [] end_pos = len(all_text) for em in end_markers: ei = all_text.find(em, sm_idx + len(start_marker)) if ei != -1 and ei < end_pos: end_pos = ei # Case A: 标题在前，数据在标题和 end_marker 之间 region_a = all_text[sm_idx:end_pos] result_a = _extract_names_from_region(region_a, known_names, len(region_a)) # Case B: 标题在后，数据在'指标范围'和标题之间 zb_idx = all_text.rfind('指标范围', 0, sm_idx) if zb_idx == -1: zb_idx = max(0, sm_idx - 500) region_b = all_text[zb_idx:sm_idx] result_b = _extract_names_from_region(region_b, known_names, len(region_b)) # 取匹配多的那个 if len(result_a) >= len(result_b): return result_a return result_b def _extract_pathogens_inline(text): """提取 inline 格式中主要消化道致病菌数据（幽门螺杆菌0%未检出...）""" results = [] sidx = text.find('主要消化道致病菌') if sidx == -1: return results eidx = text.find('肠道屏障', sidx) if eidx == -1: eidx = len(text) region = text[sidx:eidx] for name in KNOWN_PATHOGENS: idx = region.find(name) if idx == -1: continue after = region[idx + len(name):] m = re.match(r'\s*(\d+%)', after) if not m: continue abundance = m.group(1) assessment_start = idx + len(name) + m.end() assessment_end = len(region) for next_name in KNOWN_PATHOGENS: if next_name == name: continue ni = region.find(next_name, assessment_start) if ni != -1 and ni < assessment_end: assessment_end = ni assessment = region[assessment_start:assessment_end].strip() results.append({'name': name, 'value': abundance, 'status': assessment}) return results def _final_fallback_scan(full_text, result, section_known_map): """最终的全文正则扫描——弥补所有提取器漏掉的已知指标项（v5 移植）""" for section, known_list in section_known_map.items(): existing = {r['name'] for r in result.get(section, [])} for name in known_list: if name in existing: continue # 搜索多个变体：完整名、去掉括号的后缀 search_names = [name] paren = name.find('（') if paren != -1: search_names.append(name[:paren]) found = False for sname in search_names: idx = full_text.find(sname) if idx == -1: continue # 跳过名称之后紧跟着的重复名称（正字+部首两次出现的情况） skip = idx + len(sname) while skip < len(full_text): ch = full_text[skip] if ch == '\n' or ch == ' ': skip += 1 elif '\u4e00' <= ch <= '\u9fff' or '\u2f00' <= ch <= '\u2fdf': skip += 1 else: break after = full_text[skip:].lstrip() m = re.match(r'(\d+(?:\.\d+)?)', after) if not m: continue val = m.group(1) val_end_in_after = m.end() status_raw = after[val_end_in_after:].lstrip() status_m = re.search(r'([\u4e00-\u9fff/]+)', status_raw) status = status_m.group(1).strip() if status_m else status_raw[:20] if len(status) > 20: status = status[:20] result.setdefault(section, []).append({'name': name, 'value': val, 'status': status}) found = True break return result def parse_report_pdf(file_path: str) -> dict: """主函数：解析 PDF 返回结构化数据（v5 完整提取逻辑移植）""" reader = PdfReader(file_path) pages_text = [p.extract_text() or '' for p in reader.pages] lines = extract_text(file_path) full_text = '\n'.join(lines) # 指纹检测：北京菌群报告 if detect_beijing_format(full_text): return _parse_beijing_report(file_path) fmt = detect_format(pages_text) result = {'format': fmt, 'overview': parse_overview(lines)} _SECTION_KNOWN_MAP = { 'disease_risks': KNOWN_DISEASE_RISKS, 'nutrition': KNOWN_MACRO, 'amino_acids': KNOWN_AMINO, 'vitamins': KNOWN_VITAMINS, 'trace_elements': KNOWN_TRACE, '抗生素风险评估': KNOWN_ANTIBIOTICS, '肠道屏障及代谢物': KNOWN_BARRIER, '短链脂肪酸': KNOWN_SCFA, '神经递质及激素': KNOWN_NEURO, } if fmt == 'triplet': parsed = _extract_all_triplet(lines) # 疾病风险评估（已知疾病名词表过滤） disease_set = set(KNOWN_DISEASE_RISKS) result['disease_risks'] = [r for r in parsed.get('disease_risks', []) if r['name'] in disease_set] # 兜底：inline 方式补充 triplet 漏掉的项目 for r in _extract_inline_module(full_text, '疾病风险评估', ['主要营养评估', '主要消化道致病菌'], KNOWN_DISEASE_RISKS): if r['name'] not in {x['name'] for x in result['disease_risks']}: result['disease_risks'].append(r) # 主要营养评估 nutri_set = set(KNOWN_MACRO) result['nutrition'] = [r for r in parsed.get('nutrition_rows', []) if r['name'] in nutri_set] for r in _extract_inline_module(full_text, '营养状况评估', ['主要营养评估', '氨基酸评估'], KNOWN_MACRO): if r['name'] not in {x['name'] for x in result['nutrition']}: result['nutrition'].append(r) # 氨基酸评估（营养页内嵌 p2 + 独立页 p3 合并 + inline 兜底） amino_rows = parsed.get('amino_p2', []) + parsed.get('amino_p3', []) amino_set = set(KNOWN_AMINO) result['amino_acids'] = [r for r in amino_rows if r['name'] in amino_set] for r in _extract_inline_module(full_text, '氨基酸评估', ['维生素评估', '主要消化道致病菌'], KNOWN_AMINO): if r['name'] not in {x['name'] for x in result['amino_acids']}: result['amino_acids'].append(r) # 维生素评估 / 微量元素评估 vit_set = set(KNOWN_VITAMINS) trace_set = set(KNOWN_TRACE) result['vitamins'] = [r for r in parsed.get('vitamin_rows', []) if r['name'] in vit_set] result['trace_elements'] = [r for r in parsed.get('trace_rows', []) if r['name'] in trace_set] for r in _extract_inline_module(full_text, '维生素评估', ['微量元素评估', '主要消化道致病菌'], KNOWN_VITAMINS): if r['name'] not in {x['name'] for x in result['vitamins']}: result['vitamins'].append(r) for r in _extract_inline_module(full_text, '微量元素评估', ['主要消化道致病菌'], KNOWN_TRACE): if r['name'] not in {x['name'] for x in result['trace_elements']}: result['trace_elements'].append(r) # 主要消化道致病菌 result['主要消化道致病菌'] = parsed.get('pathogens', []) # 抗生素风险评估（已知抗生素词表过滤） abx_set = set(KNOWN_ANTIBIOTICS) result['抗生素风险评估'] = [r for r in parsed.get('antibiotics', []) if r['name'] in abx_set] # 肠道屏障 + 短链脂肪酸 + 神经递质（已知指标名精确匹配） bs = _extract_barrier_and_scfa_triplet(lines) result['肠道屏障及代谢物'] = bs['barrier'] result['短链脂肪酸'] = bs['scfa'] result['神经递质及激素'] = bs['neurotransmitter'] # inline 兜底补充 for section_key, start_mk, end_mks, known_list in [ ('肠道屏障及代谢物', '肠道屏障及菌群代谢物', ['短链脂肪酸', '神经递质', '抗生素风险'], KNOWN_BARRIER), ('短链脂肪酸', '短链脂肪酸', ['神经递质', '抗生素风险'], KNOWN_SCFA), ('神经递质及激素', '神经递质', ['抗生素风险', '个体化食物推荐表'], KNOWN_NEURO), ]: existing = {r['name'] for r in result.get(section_key, [])} for r in _extract_inline_module(full_text, start_mk, end_mks, known_list): if r['name'] not in existing: result.setdefault(section_key, []).append(r) existing.add(r['name']) # 最终兜底：全文正则扫描所有已知指标名 _final_fallback_scan(full_text, result, _SECTION_KNOWN_MAP) else: # inline 格式（v5 extract_pdf_to_json inline 分支移植） result['disease_risks'] = _extract_inline_module(full_text, '疾病风险评估', ['主要营养评估', '主要消化道致病菌'], KNOWN_DISEASE_RISKS) result['nutrition'] = _extract_inline_module(full_text, '营养状况评估', ['主要营养评估', '氨基酸评估'], KNOWN_MACRO) result['amino_acids'] = _extract_inline_module(full_text, '营养状况评估', ['主要营养评估', '氨基酸评估'], KNOWN_AMINO) amino_p2 = _extract_inline_module(full_text, '氨基酸评估', ['维生素评估', '主要消化道致病菌'], KNOWN_AMINO) for a in amino_p2: if a['name'] not in {x['name'] for x in result['amino_acids']}: result['amino_acids'].append(a) result['vitamins'] = _extract_inline_module(full_text, '维生素评估', ['微量元素评估', '主要消化道致病菌'], KNOWN_VITAMINS) result['trace_elements'] = _extract_inline_module(full_text, '微量元素评估', ['主要消化道致病菌'], KNOWN_TRACE) result['主要消化道致病菌'] = _extract_pathogens_inline(full_text) result['抗生素风险评估'] = _extract_inline_module(full_text, '抗生素风险评估', ['抗生素耐药风险', '个体化食物推荐表'], KNOWN_ANTIBIOTICS) result['肠道屏障及代谢物'] = _extract_inline_module(full_text, '肠道屏障及菌群代谢物', ['短链脂肪酸', '神经递质', '抗生素风险'], KNOWN_BARRIER) result['短链脂肪酸'] = _extract_inline_module(full_text, '短链脂肪酸', ['神经递质', '抗生素风险'], KNOWN_SCFA) result['神经递质及激素'] = _extract_inline_module(full_text, '神经递质', ['抗生素风险', '个体化食物推荐表'], KNOWN_NEURO) _final_fallback_scan(full_text, result, _SECTION_KNOWN_MAP) # 菌群检出详细列表（核心/益生菌/有害菌/病原菌 + 门纲目科属种） bacteria_tables = _extract_bacteria_tables(file_path) result['菌群检出详细列表'] = bacteria_tables # 个体化食物推荐表（保留原始键名与结构） food_rows, food_fmt = _extract_food_rows(file_path) result['个体化食物推荐表'] = { '格式': food_fmt, '条目数': len(food_rows), '数据': food_rows } # 统一结构：北京报告特有字段（标准报告为空） result['肠道微生物健康指数'] = {} result['菌群多样性'] = {} result['肠道黏膜屏障'] = {} result['菌群表型评估'] = {} result['短链脂肪酸合成能力'] = {} result['肠道菌群精准分布'] = [] result['营养物质及营养素代谢评估'] = {} result.setdefault('抗生素风险评估', {}) result['毒性物质清除能力评估'] = {} result['趣味肠菌评估'] = {} result['胃肠道感染病原体'] = [] result['健康整体评估'] = {} result['肠道菌群主要检测结果'] = {} return result def parse_report_pdf_with_fallback(file_path: str) -> dict: """算法解析 + 简单校验，返回结构化数据""" result = parse_report_pdf(file_path) # 简单校验：如果关键字段缺失，标记为解析不完整 if not result.get('overview', {}).get('overallScore') and \ not result.get('overview', {}).get('健康总分'): result['_parse_incomplete'] = True return result def _parse_bacteria_table(reader, pages_text, full_text, title, fmt, skip_header=False): """从PDF中解析一个菌群表格（移植自 extract_full_report_v5.py）""" results = [] sidx = full_text.find(title) if sidx == -1: return results # 病原菌检出表特殊处理：找"仅列出检出的病原菌"（跳过前面的说明文字中的"病原菌"） if title == '病原菌': better_sidx = full_text.find('仅列出检出的病原菌') if better_sidx != -1: sidx = better_sidx # 找表格结束位置（下一个标题或页尾） end_pos = len(full_text) for t in (BACTERIA_TABLE_TITLES + PHYLUM_TABLE_TITLES + CLASS_TABLE_TITLES + ORDER_TABLE_TITLES + FAMILY_TABLE_TITLES + GENUS_TABLE_TITLES + SPECIES_TABLE_TITLES + DISEASE_BACTERIA_TITLES + ['指标范围', '个体化食物推荐表', '报告总结', '健康总分']): if t == title: continue ei = full_text.find(t, sidx + len(title)) if ei != -1 and ei < end_pos: end_pos = ei region = full_text[sidx:end_pos] # 检测区域实际格式：如果换行数很少(<3)则是inline格式，即使全局fmt=triplet lines_from_region = [l.strip() for l in region.split('\n') if l.strip()] actual_fmt = fmt if fmt == 'triplet' and len(lines_from_region) <= 5: actual_fmt = 'inline' if actual_fmt == 'triplet': # 三元组格式：每个字段单独一行 lines = [l.strip() for l in region.split('\n') if l.strip()] start = 0 for i, line in enumerate(lines): if line == '名称': start = i + 1 break if line.startswith('名称'): if skip_header: start = i + 1 break i = start while i < len(lines): name = lines[i] if not name or len(name) <= 1 or name in ['说明', '检测结果', '结果解释', '建议']: i += 1 continue if name.startswith('说明：') or name.startswith('改善方式'): i += 1 continue if len(name) > 80: i += 1 continue # 找丰度% if i + 1 < len(lines) and re.match(r'^[\d]+\.?[\d]*%?$|^ND$', lines[i + 1]): pct = lines[i + 1] normal_range = '' pop_level = '' detection_rate = '' desc = '' j = i + 2 # 正常范围 (允许小数，如 0.06-6.96, 0.03-3.07) if j < len(lines) and re.match(r'^[\d]+\.?[\d]*-[\d]+\.?[\d]*$', lines[j]): normal_range = lines[j] j += 1 # 人群水平% (允许小数，如 15.66%) if j < len(lines) and re.match(r'^\d+\.?\d*%$', lines[j]): pop_level = lines[j] j += 1 # 检出率% if j < len(lines) and re.match(r'^\d+\.?\d*%$', lines[j]): detection_rate = lines[j] j += 1 # 说明 if j < len(lines) and lines[j].startswith('说明'): desc = lines[j] j += 1 # 改善方式 if j < len(lines) and lines[j].startswith('改善方式'): if desc: desc += ' | ' + lines[j] else: desc = lines[j] j += 1 entry = {'名称': name, '丰度%': pct} if normal_range: entry['正常范围%'] = normal_range if pop_level: entry['人群水平%'] = pop_level if detection_rate: entry['检出率%'] = detection_rate if desc: entry['说明'] = desc results.append(entry) i = j else: i += 1 else: # inline 格式 sample = region[:500] truly_compressed = bool(re.search(r'[a-z]\d', sample, re.IGNORECASE)) if not truly_compressed: # 有空格分隔的inline格式，使用re.finditer region_clean = region for hdr in ['名称', '丰度%', '正常范围%', '处于人群%水平', '%正常人有检出', '人群水平%', '%人检出']: region_clean = region_clean.replace(hdr, '') region_clean = re.sub( r'说明：[\u4e00-\u9fff\s，。、；：,.;:()（）、/a-zA-Z0-9\-]{10,}?(?=[\u4e00-\u9fff]|$)', '', region_clean) # 扫描所有匹配的数据行 for m in re.finditer( r'([\u4e00-\u9fff]{2,12}(?:[（(][\u4e00-\u9fff\w]+[)）])?)\s+' # 中文名 r'(?:[A-Z][a-z]+(?:\s[A-Z][a-z]+)*\s+)?' # 可选英文名 r'(ND|[\d]+\.?[\d]*%?)\s+' # 丰度 r'([\d]+\.?[\d]*-[\d]+\.?[\d]*)?\s*' # 可选正常范围 r'(\d+\.?\d*%?)\s+' # 人群水平% r'(\d+\.?\d*%)', # 检出率% region_clean): name = m.group(1).strip() pct = m.group(2) if '病原菌' in name or '构成表' in name or '说明' in name or len(name) <= 1: continue normal_range = m.group(3) or '' pop_level = m.group(4) if not pop_level.endswith('%'): pop_level += '%' detection_rate = m.group(5) entry = {'名称': name, '丰度%': pct} if normal_range: entry['正常范围%'] = normal_range entry['人群水平%'] = pop_level entry['检出率%'] = detection_rate results.append(entry) # 模式1没有匹配时：仅中文名+丰度+人群水平(+检出率) if not results: for m in re.finditer( r'([\u4e00-\u9fff]{2,10}[\u4e00-\u9fff]?)\s+' r'(ND|[\d]+\.?[\d]*%?)\s+' r'(\d+\.?\d*%)\s+' r'(\d+\.?\d*%)?', region_clean): name = m.group(1).strip() pct = m.group(2) if '病原菌' in name or '构成表' in name or '说明' in name or len(name) <= 1: continue pop_level = m.group(3) detection_rate = m.group(4) or '' entry = {'名称': name, '丰度%': pct, '人群水平%': pop_level} if detection_rate: entry['检出率%'] = detection_rate results.append(entry) return results def _parse_phylum_tables(reader, full_text, fmt, title_list=None): """提取菌群层级构成表（门/纲/目/科/属/种 level）（移植自 extract_full_report_v5.py）""" if title_list is None: title_list = PHYLUM_TABLE_TITLES results = [] for phylum_title in title_list: rows = _parse_bacteria_table(reader, None, full_text, phylum_title, fmt) results.extend(rows) return results def _parse_taxonomy_levels(reader, full_text, fmt): """从"菌群检出详细列表"中提取纲目科属种各级数据（移植自 extract_full_report_v5.py）""" results = {} sidx = full_text.find('菌群检出详细列表') if sidx == -1: return results end_pos = len(full_text) for t in ['个体化食物推荐表', '报告总结', '健康总分']: ei = full_text.find(t, sidx) if ei != -1 and ei < end_pos: end_pos = ei section = full_text[sidx:end_pos] for level, level_name in [('纲', '菌纲构成'), ('目', '菌目构成'), ('科', '菌科构成'), ('属', '菌属构成'), ('种', '菌种构成')]: marker = f'\n{level}\n名称\n丰度%' marker2 = f'{level} 名称 丰度%' li = section.find(marker) level_start = None if li == -1: li2 = section.find(marker2) if li2 != -1: li = li2 level_start = li2 + len(marker2) else: compressed_marker = f'{level}名称丰度%人群水平%%人检出' cli = section.find(compressed_marker) if cli == -1: continue # 压缩格式解析：用正则提取数据 level_start = cli + len(compressed_marker) level_end = len(section) for next_level in ['目', '科', '属', '种']: if next_level == level: continue ni = section.find(f'{next_level}名称丰度%人群水平%%人检出', level_start) if ni != -1 and ni < level_end: level_end = ni break level_region = section[level_start:level_end] compressed_pattern = re.compile( r'([\u4e00-\u9fff·]+(?:\s[\u4e00-\u9fff·]+)?\s+)?' # 可选中文名 r'([A-Za-z][A-Za-z\s.\-]*?)' # 拉丁名（可能含空格） r'(\d+\.?\d*%)(\d+\.?\d*%)(\d+\.?\d*%)' # 三连百分比 ) rows = [] for m in compressed_pattern.finditer(level_region): cn_name = (m.group(1) or '').strip() latin_name = m.group(2).strip() pct = m.group(3) pop_level = m.group(4) detection = m.group(5) name = cn_name if cn_name else latin_name entry = {'名称': name, '丰度%': pct, '人群水平%': pop_level, '检出率%': detection} rows.append(entry) if rows: results[level_name] = rows continue else: level_start = li + len(marker) level_end = len(section) for next_level in ['纲', '目', '科', '属', '种']: if next_level == level: continue ni = section.find(f'\n{next_level}\n名称', level_start) if ni != -1 and ni < level_end: level_end = ni break level_region = section[level_start:level_end] lines = [l.strip() for l in level_region.split('\n') if l.strip()] rows = [] i = 0 while i < len(lines): if lines[i] in ['名称', '丰度%', '人群水平%', '%人检出']: i += 1 continue name = lines[i] if i + 2 < len(lines) and re.match(r'^[\d]+\.?[\d]*%?$', lines[i + 1]): pct = lines[i + 1] pop_level = lines[i + 2] if i + 2 < len(lines) else '' detection = lines[i + 3] if i + 3 < len(lines) and re.match(r'^[\d.]+%$', lines[i + 3]) else '' entry = {'名称': name, '丰度%': pct, '人群水平%': pop_level} if detection: entry['检出率%'] = detection rows.append(entry) i += 4 if detection else 3 else: i += 1 if rows: results[level_name] = rows return results def _extract_bacteria_tables(pdf_path): """提取菌群检出详细列表，返回 {中文分组名: [行]}（移植自 extract_full_report_v5.py） 与原始脚本保持一致的文本构造与格式检测：full_text 使用 '\\n'.join(norm(p.extract_text()) for p in reader.pages)（不 strip、不过滤空行）， fmt 使用原脚本 L672-683 的独立检测逻辑。 """ reader = PdfReader(pdf_path) full_text = '\n'.join(norm(p.extract_text()) for p in reader.pages) # 原脚本 extract_bacteria_tables 的格式检测（L674-683） fmt = 'triplet' if '指标范围' in full_text and '疾病风险评估' in full_text else 'inline' for pt in [p.extract_text() for p in reader.pages]: t = norm(pt) if '疾病风险评估' in t and '指标范围' in t: for line in t.split('\n'): if re.search(r'[\u4e00-\u9fff]+\d+\.?\d*[\u4e00-\u9fff]+', line.strip()): fmt = 'inline' break break all_tables = {} # 核心菌属构成表1-3 core_genus = [] for i in range(1, 4): title = f'核心菌属构成表{i}' rows = _parse_bacteria_table(reader, None, full_text, title, fmt) core_genus.extend(rows) all_tables['核心菌属'] = core_genus # 益生菌（使用更精确的表头定位，跳过前面的说明文字） prob_marker = '仅列出丰度前22的益生菌种' prob_sidx = full_text.find(prob_marker) if prob_sidx != -1: prob_rows = _parse_bacteria_table(reader, None, full_text, prob_marker, fmt, skip_header=True) else: prob_rows = _parse_bacteria_table(reader, None, full_text, '益生菌', fmt, skip_header=True) all_tables['益生菌'] = [r for r in prob_rows if r.get('名称') and r['名称'] not in ['我的益生菌都为ND', '仅列出丰度前22的益生菌种。']] # 有害菌属构成表1-2 harmful = [] for i in range(1, 3): title = f'有害菌属构成表{i}' rows = _parse_bacteria_table(reader, None, full_text, title, fmt) harmful.extend(rows) all_tables['有害菌属'] = harmful # 其它重要菌属 other_rows = _parse_bacteria_table(reader, None, full_text, '其它重要菌属构成表', fmt) all_tables['其它重要菌属'] = other_rows # 病原菌属构成表 patho_genus = _parse_bacteria_table(reader, None, full_text, '病原菌属构成表', fmt) all_tables['病原菌属'] = patho_genus # 病原菌（检出列表） patho_rows = _parse_bacteria_table(reader, None, full_text, '病原菌', fmt, skip_header=True) all_tables['病原菌检出'] = [r for r in patho_rows if r.get('名称') and len(r['名称']) >= 2 and '仅列出' not in r['名称'] and '说明' not in r['名称']] # 菌门构成表（phylum level） phylum_rows = _parse_phylum_tables(reader, full_text, fmt) if phylum_rows: all_tables['菌门构成'] = phylum_rows # 菌纲构成表 class_rows = _parse_phylum_tables(reader, full_text, fmt, CLASS_TABLE_TITLES) if class_rows: all_tables['菌纲构成'] = class_rows # 菌目构成表 order_rows = _parse_phylum_tables(reader, full_text, fmt, ORDER_TABLE_TITLES) if order_rows: all_tables['菌目构成'] = order_rows # 菌科构成表 family_rows = _parse_phylum_tables(reader, full_text, fmt, FAMILY_TABLE_TITLES) if family_rows: all_tables['菌科构成'] = family_rows # 菌属构成表 genus_rows = _parse_phylum_tables(reader, full_text, fmt, GENUS_TABLE_TITLES) if genus_rows: all_tables['菌属构成'] = genus_rows # 菌种构成表 species_rows = _parse_phylum_tables(reader, full_text, fmt, SPECIES_TABLE_TITLES) if species_rows: all_tables['菌种构成'] = species_rows # 菌群层级（纲目科属种）- 从"菌群检出详细列表"统一入口提取 taxonomy_rows = _parse_taxonomy_levels(reader, full_text, fmt) for key, rows in taxonomy_rows.items(): if rows: all_tables[key] = rows return all_tables def split_7_fields(s): """将压缩数字串切分为 7 个字段（移植自 extract_full_report_v5.py）""" results = [] ranges = [(2, 4), (1, 2), (1, 2), (1, 2), (1, 2), (1, 2), (1, 4)] def backtrack(pos, idx, nums): if idx == 7: if pos == len(s): results.append(list(nums)) return if pos >= len(s): return lo, hi = ranges[idx] for w in range(lo, min(hi + 1, len(s) - pos + 1)): chunk = s[pos:pos + w] if chunk.isdigit(): backtrack(pos + w, idx + 1, nums + [int(chunk)]) backtrack(0, 0, []) return results def decode_compressed(name, num_str, ref_vals=None): """解码压缩食物数字串（移植自 extract_full_report_v5.py）""" raw = num_str.lstrip('-') has_neg = num_str.startswith('-') candidates = [] for rec_len in range(1, 3): if rec_len > len(raw): continue rec = ('-' if has_neg else '') + raw[:rec_len] try: rec_val = int(rec) if not (-100 <= rec_val <= 100): continue except Exception: continue remain = raw[rec_len:] for nums in split_7_fields(remain): if ref_vals: matches = sum(1 for i in range(7) if ref_vals[i] == nums[i]) if matches >= 6: candidates.append([rec_val] + nums) else: candidates.append([rec_val] + nums) if not candidates: return None if ref_vals: candidates.sort(key=lambda r: (sum(1 for i in range(7) if ref_vals[i] == r[1:][i]), -len(str(abs(r[0])))), reverse=True) if sum(1 for i in range(7) if ref_vals[i] == candidates[0][1:][i]) < 6: return None else: candidates.sort(key=lambda r: (len(str(abs(r[0]))), -sum(1 for i in range(7) if r[1:][i] == 0))) return candidates[0] def extract_food_table(pdf_path, ref_lookup=None): """提取个体化食物推荐表（移植自 extract_full_report_v5.py）""" reader = PdfReader(pdf_path) food_start = None for i, page in enumerate(reader.pages): if '个体化食物推荐表' in page.extract_text(): food_start = i break if food_start is None: return [], 'not_found' first_text = norm(reader.pages[food_start + 1].extract_text()) lines = [l.strip() for l in first_text.split('\n') if l.strip() and not re.match(r'\d+/\d+', l)] # 判断压缩格式：任何一行超过100字符（单行密集格式），或前10行中超过3行长行 is_compressed = (len(lines) >= 1 and any(len(l) > 100 for l in lines[:10])) or \ sum(1 for l in lines[:10] if len(l) > 100) >= 2 rows = [] if is_compressed: fmt = 'compressed' for i in range(food_start + 1, len(reader.pages)): text = norm(reader.pages[i].extract_text()) text = re.sub(r'\d+/\d+', '', text) header = '名称分类推荐指数能量KJ蛋白g脂肪g碳水化合物g淀粉g总膳食纤维g胆固醇mg' text = text.replace(header, '') for kw in FOOD_SKIP_TEXTS: text = text.replace(kw, '') while text: best_cat, best_idx = None, len(text) for cat in KNOWN_CATS: idx = text.find(cat) if idx != -1 and idx < best_idx: best_idx, best_cat = idx, cat if best_cat is None: break name = text[:best_idx] text = text[best_idx + len(best_cat):] num_str = '' while text and (text[0].isdigit() or text[0] in '-\u2212\u2014'): c = '-' if text[0] in '\u2212\u2014' else text[0] num_str += c text = text[1:] if not name or not num_str: continue ref_vals = ref_lookup.get(name) if ref_lookup else None decoded = decode_compressed(name, num_str, ref_vals) if decoded: rows.append(dict(zip(COLUMNS_FOOD, [name, best_cat] + [str(v) for v in decoded]))) else: fmt = 'vertical' all_lines = [] for i in range(food_start + 1, len(reader.pages)): for line in norm(reader.pages[i].extract_text()).split('\n'): lt = line.strip() if not lt or re.match(r'\d+/\d+', lt) or lt in COLUMNS_FOOD: continue if len(lt) > 60 and any(k in lt for k in FOOD_SKIP_TEXTS): continue all_lines.append(lt) i = 0 while i + 9 < len(all_lines): name = all_lines[i].strip() cat = all_lines[i + 1].strip() if cat not in KNOWN_CATS: i += 1 continue nums = [] ok = True for j in range(2, 10): v = all_lines[i + j].replace('\u2212', '-').replace('\u2014', '-').strip() try: int(v) nums.append(v) except Exception: ok = False break if ok and len(nums) == 8: rows.append(dict(zip(COLUMNS_FOOD, [name, cat] + nums))) i += 1 return rows, fmt def _extract_food_rows(pdf_path): """食物推荐表解析：优先复用同目录其他报告的参考营养表做压缩格式解码""" ref_nutrition = {} base_dir = os.path.dirname(pdf_path) or '.' for fname in sorted(os.listdir(base_dir)): if fname.lower().endswith('.pdf') and fname != os.path.basename(pdf_path): try: tr, _ = extract_food_table(os.path.join(base_dir, fname)) if len(tr) > 100: for r in tr: ref_nutrition[r['名称']] = [int(r[k]) for k in ['能量KJ', '蛋白g', '脂肪g', '碳水化合物g', '淀粉g', '总膳食纤维g', '胆固醇mg']] break except Exception: continue food_rows, food_fmt = extract_food_table(pdf_path, ref_nutrition or None) return food_rows, food_fmt # ============================================================ # 北京菌群报告 — 指纹检测 # ============================================================ FINGERPRINTS_BEIJING = [ "肠道微生物健康指数", "GMHI", "肠道菌群主要检测结果", "核心菌属综合评分", "有益菌综合评分", "有害菌综合评分", "肠道菌群精准分布", "营养物质及营养素代谢评估", "趣味肠菌评估", "高通量测序", ] def detect_beijing_format(text): """检测是否为北京菌群报告格式""" beijing_matches = sum(1 for f in FINGERPRINTS_BEIJING if f in text) if beijing_matches >= 3: return True return False # ============================================================ # 北京菌群报告 — 章节提取（PyPDF2 版） # ============================================================ # 疾病风险关键词（北京报告特有） DISEASE_KEYWORDS_BJ = [ "溃疡性结肠炎", "克罗恩病", "腹泻型肠易激综合征", "硬化性胆管炎", "非酒精性脂肪性肝病", "肝硬化", "便秘", "IgA肾病", "肾结石", "格雷夫斯病", "慢性淋巴细胞性甲状腺炎", "强直性脊柱炎", "精神分裂症", "痛经", "结直肠癌", "胃癌", "甲状腺癌", "肺癌", "乳腺癌", "高血压", "冠心病", "2型糖尿病", "高脂血症", "食管癌", "胆结石", "哮喘", "心脏病", "膜性肾病", "帕金森病", "类风湿关节炎", ] # 抗生素分类（北京报告格式） ANTIBIOTIC_CATEGORIES_BJ = [ "β内酰胺类", "氨基糖苷类", "大环内酯类", "四环素类", "氯霉素类", "磺酰胺类", "喹诺酮类", "万古霉素类", "头孢菌素类", "磷霉素类", "甲氧苄氨嘧啶类", "N-乙酰基转移酶类", "O-磷酸转移酶类", "16S甲基转移酶", ] # 核心菌属 22 种（北京报告） CORE_GENERA_BJ = [ ("阿克曼菌属", "Akkermansia"), ("另枝菌属", "Alistipes"), ("拟杆菌属", "Bacteroides"), ("双歧杆菌属", "Bifidobacterium"), ("布劳特氏菌属", "Blautia"), ("梭菌属", "Clostridium"), ("粪球菌属", "Coprococcus"), ("戴阿利斯特杆菌属", "Dialister"), ("多尔氏菌属", "Dorea"), ("真杆菌属", "Eubacterium"), ("粪杆菌属", "Faecalibacterium"), ("Lachnoclostridium属", "Lachnoclostridium"), ("毛螺菌属", "Lachnospira"), ("乳杆菌属", "Lactobacillus"), ("巨单胞菌属", "Megamonas"), ("颤螺旋菌属", "Oscillospira"), ("副拟杆菌属", "Parabacteroides"), ("考拉杆菌属", "Phascolarctobacterium"), ("普雷沃氏菌属", "Prevotella"), ("罗氏菌属", "Roseburia"), ("瘤胃球菌属", "Ruminococcus"), ("萨特氏菌属", "Sutterella"), ] # 病原体列表（北京报告） PATHOGEN_LIST_BJ = [ "气单胞菌属", "蜡样芽孢杆菌", "唐菖蒲伯克霍尔德氏菌", "弯曲杆菌属", "艰难梭菌", "肉毒梭菌", "产气荚膜梭菌", "幽门螺杆菌", "肺炎克雷伯菌", "类志贺邻单胞菌", "沙门氏菌属", "志贺氏菌属", "金黄色葡萄球菌", "肺炎链球菌", "霍乱弧菌", "拟态弧菌", "副溶血弧菌", "小肠结肠炎耶尔森氏菌", "假结核耶尔森氏菌", ] def _find_content_after_toc(text): """找到北京报告中 TOC 之后的内容起始位置""" # 找"第一部分 健康整体评估"后的内容，或者"健康整体评估"正文 markers = ['第一部分 健康整体评估', '菌群得分总览', '您属于肠'] for m in markers: idx = text.find(m) if idx != -1: return idx return 0 def _find_last_section(text, section_title): """找到章节标题的最后一次出现（实际内容，非目录）""" last = -1 start = 0 while True: idx = text.find(section_title, start) if idx == -1: break last = idx start = idx + 1 return last def _parse_bj_gmhi(text): """提取 GMHI 肠道微生物健康指数（从内容区提取）""" gmhi = {} # 找最后一次出现的 GMHI 值（内容区，非目录） content_start = _find_content_after_toc(text) content = text[content_start:] m = re.search(r'GMHI[）\)]?[为是](\d+)', content) if m: gmhi['GMHI'] = float(m.group(1)) m = re.search(r'肠道微生物健康指数[为](\d+)', content) if m: gmhi['肠道微生物健康指数'] = float(m.group(1)) m = re.search(r'肠道菌群状态[为：:\s]*(\S+?)[。，\n]', content) if m: gmhi['肠道菌群状态'] = m.group(1) # 健康状态推断 if gmhi.get('肠道微生物健康指数') or gmhi.get('GMHI'): score = gmhi.get('肠道微生物健康指数') or gmhi.get('GMHI', 0) if score >= 90: gmhi['健康状态'] = '健康' elif score >= 80: gmhi['健康状态'] = '健康倾向' elif score >= 40: gmhi['健康状态'] = '亚健康' elif score >= 20: gmhi['健康状态'] = '不健康倾向' else: gmhi['健康状态'] = '不健康' return gmhi def _parse_bj_gut_type(text): """提取肠型""" content_start = _find_content_after_toc(text) content = text[content_start:] m = re.search(r'属于\s*肠[型道]\s*([IVX]+)', content) if m: return f'肠型{m.group(1)}' m = re.search(r'属于\s*肠[型道]\s*(\S+?)[，。,.\n]', content) if m: return m.group(1).strip() return None def _parse_bj_balance(text): """提取肠道菌群平衡评估""" m = re.search(r'(I{1,3}度失衡)', text) if m: return m.group(1) return None def _parse_bj_barrier(text): """提取肠道黏膜屏障/BE比值""" barrier = {} # 在内容区找表格 content_start = _find_content_after_toc(text) content = text[content_start:] # 找 BE比值 表格行 m = re.search(r'BE比[值例][：:\s]*(\d+\.?\d*)', text) if m: barrier['BE比值'] = float(m.group(1)) m = re.search(r'双歧杆菌[属菌][：:\s]*(\d+\.?\d*)', text) if m: barrier['双歧杆菌属'] = float(m.group(1)) m = re.search(r'肠杆菌[科属][：:\s]*(\d+\.?\d*)', text) if m: barrier['肠杆菌科'] = float(m.group(1)) # 综合评估在"BE比值"附近找 be_idx = content.find('BE比') if be_idx != -1: chunk = content[be_idx:be_idx + 300] m = re.search(r'(正常|轻度损伤|中度损伤|重度损伤|损伤)', chunk) if m: barrier['综合评估'] = m.group(1) return barrier def _parse_bj_diversity(text): """提取菌群多样性""" div = {} content_start = _find_content_after_toc(text) content = text[content_start:] m = re.search(r'多样性指数[为](\d+\.?\d*)', content) if m: div['Shannon指数'] = float(m.group(1)) m = re.search(r'高于[：:\s]*(\d+)%', content) if m: div['人群百分位'] = f'高于{m.group(1)}%' m = re.search(r'菌种总数[：:\s]*(\d+)', content) if m: div['菌种总数'] = int(m.group(1)) return div def _parse_bj_phenotype(text): """提取菌群表型评估""" pheno = {} # 找表型评估表格区域 last_idx = _find_last_section(text, '菌群表型评估') if last_idx == -1: last_idx = _find_last_section(text, '肠道菌群表型评估') if last_idx == -1: return pheno # 取表格区域 region = text[last_idx:last_idx + 1500] # 解析表格行: 指标名 评估值 结果 for label, kw in [('革兰氏阳性菌', '阳性菌'), ('革兰氏阴性菌', '阴性菌'), ('生物膜合成', '生物膜'), ('好氧菌', '好氧'), ('厌氧菌', '厌氧'), ('兼性厌氧菌', '兼性'), ('氧化胁迫耐受', '氧化'), ('致病潜力', '致病')]: # 找该行: 值在指标名附近 idx = region.find(kw) if idx == -1: continue chunk = region[idx:idx + 100] val_m = re.search(r'(\d+\.?\d*)', chunk) if val_m: entry = {'评估值': float(val_m.group(1))} for status in ['正常', '异常']: if status in chunk: entry['结果'] = status break pheno[label] = entry return pheno def _parse_bj_scfa(text): """提取短链脂肪酸合成能力""" scfa = {} # 找 SCFA 表格区域（最后一次出现） last_idx = _find_last_section(text, '短链脂肪酸合成能力') if last_idx == -1: last_idx = _find_last_section(text, '短链脂肪酸') if last_idx == -1: return scfa region = text[last_idx:last_idx + 800] for acid in ['甲酸', '乙酸', '丙酸', '丁酸', '戊酸', '己酸']: m = re.search(rf'{acid}[^a-zA-Z]*?(\d+)', region) if m: scfa[acid] = int(m.group(1)) return scfa def _parse_bj_core_bacteria(text): """提取核心菌属精准分布""" bacteria = [] for cn_name, lat_name in CORE_GENERA_BJ: idx = text.find(lat_name) if idx == -1: idx = text.find(cn_name) if idx == -1: continue chunk = text[max(0, idx - 50):idx + 200] val_m = re.search(r'(\d+\.?\d*)', chunk) status_m = re.search(r'(缺失|严重超标|超标|偏低|正常)', chunk) pct_m = re.search(r'(\d+\.?\d*)%', chunk) range_m = re.search(r'([\d.]+-[\d.]+)', chunk) entry = { '拉丁名': lat_name, '中文名': cn_name, '检测结果': float(val_m.group(1)) if val_m else None, '状态': status_m.group(1) if status_m else '未知', '人群百分位': float(pct_m.group(1)) if pct_m else None, '参考范围': range_m.group(1) if range_m else None, } bacteria.append(entry) return bacteria def _parse_bj_disease_risks(text): """提取疾病风险评估""" diseases = [] for keyword in DISEASE_KEYWORDS_BJ: idx = text.find(keyword) if idx == -1: continue chunk = text[idx:idx + 200] val_m = re.search(r'(\d+\.\d+)', chunk) level_m = re.search(r'(低风险|较低风险|中度风险|较高风险|高风险)', chunk) if val_m: diseases.append({ '疾病': keyword, '风险指数': float(val_m.group(1)), '风险等级': level_m.group(1) if level_m else '', }) return diseases def _parse_bj_nutrient_metabolism(text): """提取营养物质及营养素代谢评估""" metabolism = {} # 主要营养物质 for n in ['碳水化合物', '蛋白质', '脂肪', '矿物质']: m = re.search(rf'{n}[^估]*?(\d+)', text) if m: metabolism.setdefault('主要营养物质代谢', {})[n] = int(m.group(1)) # 糖类 for item in ['葡萄糖（糖酵解途径）', '葡萄糖（磷酸戊糖途径）', '果糖和甘露糖', '半乳糖', '淀粉和蔗糖', '氨基糖和核苷酸糖', '丙酮酸']: m = re.search(rf'{item[:4]}.*?(\d+)', text) if m: metabolism.setdefault('糖类代谢', {})[item] = int(m.group(1)) # 脂类 for item in ['甘油酯', '甘油磷脂', '甘油三酯', '鞘脂类', '脂肪酸', '硬脂酸', '棕榈酸', '花生四烯酸', '二十二碳六烯酸', 'Omega-3', 'Omega-6']: m = re.search(rf'{item}.*?(\d+)', text) if m: metabolism.setdefault('脂类代谢', {})[item] = int(m.group(1)) # 维生素 for item in ['维生素A', '维生素B1', '维生素B2', '维生素B3', '维生素B5', '维生素B6', '维生素B7', '维生素B12', '维生素C', '维生素D', '维生素E', '维生素K1', '维生素K2', '维生素K3', '叶酸']: m = re.search(rf'{item}.*?(\d+)', text) if m: metabolism.setdefault('维生素', {})[item] = int(m.group(1)) # 微量元素 for item in ['铁', '锌', '钙', '镁', '硒', '锰', '铜', '钴', '镍', '钼', '铬', '钒']: m = re.search(rf'{item}[^估]*?(\d+)', text) if m: metabolism.setdefault('微量元素', {})[item] = int(m.group(1)) # 氨基酸 for item in ['赖氨酸', '丝氨酸', '亮氨酸', '色氨酸', '苯丙氨酸', '缬氨酸', '组氨酸', '半胱氨酸', '酪氨酸', '丙氨酸', '脯氨酸', '苏氨酸', '谷氨酸', '异亮氨酸', '精氨酸', '蛋氨酸', '甘氨酸', '天冬氨酸', '牛磺酸']: m = re.search(rf'{item}[^估]*?(\d+)', text) if m: metabolism.setdefault('氨基酸', {})[item] = int(m.group(1)) # 神经递质 for item in ['5-羟色胺', 'γ-氨基丁酸', '多巴胺', '乙酰胆碱', '组胺', '去甲肾上腺素']: m = re.search(rf'{item}.*?(\d+)', text) if m: metabolism.setdefault('神经递质', {})[item] = int(m.group(1)) # 其他单项 for pattern, key in [ (r'嘌呤代谢.*?(\d+)', '嘌呤代谢'), (r'谷胱甘肽.*?(\d+)', '三肽（谷胱甘肽）'), (r'胆汁酸.*?(\d+)', '胆汁酸代谢'), (r'硫辛酸.*?(\d+)', '抗自由基（硫辛酸）'), (r'辅酶Q.*?(\d+)', '抗自由基（辅酶Q）'), ]: m = re.search(pattern, text) if m: metabolism[key] = int(m.group(1)) return metabolism def _parse_bj_antibiotics(text): """提取抗生素风险评估""" ab = {} for cat in ANTIBIOTIC_CATEGORIES_BJ: m = re.search(rf'{cat}[：:\s]*(\d+)', text) if m: ab[cat] = int(m.group(1)) return ab def _parse_bj_toxins(text): """提取毒性物质清除能力评估""" toxins = {} toxin_items = [ "苯甲酸酯", "对氨基苯甲酸乙酯", "对氟苯甲酸乙酯", "氯烷烃和氯烯烃", "氯代环己烷", "氯苯", "甲苯", "二甲苯", "硝基甲苯", "乙苯", "苯乙烯", "阿特拉津", "己内酰胺", "双酚", "二恶英", "萘", "多环芳烃", ] for item in toxin_items: m = re.search(rf'{item}.*?(\d+)', text) if m: toxins[item] = int(m.group(1)) return toxins def _parse_bj_pathogens(text): """提取胃肠道感染病原体评估""" pathogens = [] for p in PATHOGEN_LIST_BJ: idx = text.find(p) if idx == -1: continue chunk = text[idx:idx + 150] val_m = re.search(r'(\d+\.?\d*)', chunk) status_m = re.search(r'(检出|超标|未检出|正常)', chunk) if val_m: entry = {'名称': p, '检测值': float(val_m.group(1)) if '.' in val_m.group(1) else int(val_m.group(1))} if status_m: entry['状态'] = status_m.group(1) pathogens.append(entry) return pathogens def _parse_bj_psych_eval(text, find_table_val=None): """提取趣味肠菌评估""" psych = {} m = re.search(r'最高概率[：:\s]*(\S+)', text) if m: psych['人格特征'] = {'最高概率': m.group(1)} # 认知功能（从表格） if find_table_val: cog = find_table_val('认知能力', val_col=1) if cog: psych['认知功能'] = {'认知能力评分': int(cog)} return psych def _parse_bj_scfa_from_tables(find_table_rows): """从表格提取短链脂肪酸合成能力""" scfa = {} for acid in ['甲酸', '乙酸', '丙酸', '丁酸', '戊酸', '己酸']: rows = find_table_rows(acid) for name, val in rows: if acid in name: scfa[acid] = int(val) break return scfa def _parse_bj_core_bacteria_from_tables(all_tables): """从表格提取核心菌属精准分布""" bacteria = [] for cn_name, lat_name in CORE_GENERA_BJ: for table in all_tables: for row in table: row_str = ' '.join(str(c) for c in row if c) if lat_name in row_str or cn_name in row_str: vals = [str(c).strip() for c in row if c and str(c).strip()] entry = {'拉丁名': lat_name, '中文名': cn_name} for v in vals: v_clean = v.replace('"', '').replace('#', '').strip() # 判断是否为范围值（如 0.0041-0.0339） if '-' in v_clean and v_clean.replace('.','').replace('-',' ').replace(' ','').isdigit(): entry['参考范围'] = v_clean elif v_clean == 'ND': if '检测结果' not in entry: entry['检测结果'] = 0.0 elif v_clean.replace('.','').replace('-','').isdigit() and v_clean != '': val = float(v_clean) if '检测结果' not in entry: entry['检测结果'] = val elif '人群百分位' not in entry: entry['人群百分位'] = val if '检测结果' in entry: bacteria.append(entry) break if bacteria and bacteria[-1].get('拉丁名') == lat_name: break return bacteria def _parse_bj_nutrient_from_tables(find_table_rows, find_table_val): """从表格提取营养物质及营养素代谢评估""" metabolism = {} # 主要营养物质 for n in ['碳水化合物', '蛋白质', '脂肪', '矿物质']: rows = find_table_rows(n) if rows: for name, val in rows: if n in name: metabolism.setdefault('主要营养物质代谢', {})[n] = int(val) # 糖类 sugar_items = ['葡萄糖（糖酵解途径）', '葡萄糖（磷酸戊糖途径）', '果糖和甘露糖', '半乳糖', '淀粉和蔗糖', '氨基糖和核苷酸糖', '丙酮酸'] for item in sugar_items: rows = find_table_rows(item[:3]) if rows: for name, val in rows: metabolism.setdefault('糖类代谢', {})[item] = int(val) # 脂类 for item in ['甘油酯', '甘油磷脂', '甘油三酯', '鞘脂类', '脂肪酸', '硬脂酸', '棕榈酸', '花生四烯酸', '二十二碳六烯酸', 'Omega-3', 'Omega-6']: rows = find_table_rows(item[:3]) if rows: for name, val in rows: metabolism.setdefault('脂类代谢', {})[item] = int(val) # 维生素 for item in ['维生素A', '维生素B1', '维生素B2', '维生素B3', '维生素B5', '维生素B6', '维生素B7', '维生素B12', '维生素C', '维生素D', '维生素E', '维生素K1', '维生素K2', '维生素K3', '叶酸']: rows = find_table_rows(item) if rows: for name, val in rows: metabolism.setdefault('维生素', {})[item] = int(val) # 微量元素 for item in ['铁', '锌', '钙', '镁', '硒', '锰', '铜', '钴', '镍', '钼', '铬', '钒']: rows = find_table_rows(item) if rows: for name, val in rows: if name == item: metabolism.setdefault('微量元素', {})[item] = int(val) # 氨基酸 for item in ['赖氨酸', '丝氨酸', '亮氨酸', '色氨酸', '苯丙氨酸', '缬氨酸', '组氨酸', '半胱氨酸', '酪氨酸', '丙氨酸', '脯氨酸', '苏氨酸', '谷氨酸', '异亮氨酸', '精氨酸', '蛋氨酸', '甘氨酸', '天冬氨酸', '牛磺酸']: rows = find_table_rows(item[:2]) if rows: for name, val in rows: metabolism.setdefault('氨基酸', {})[item] = int(val) # 神经递质 for item in ['5-羟色胺', 'γ-氨基丁酸', '多巴胺', '乙酰胆碱', '组胺', '去甲肾上腺素']: rows = find_table_rows(item[:3]) if rows: for name, val in rows: metabolism.setdefault('神经递质', {})[item] = int(val) # 其他单项 for item, key in [('嘌呤代谢', '嘌呤代谢'), ('谷胱甘肽', '三肽（谷胱甘肽）'), ('胆汁酸', '胆汁酸代谢'), ('硫辛酸', '抗自由基（硫辛酸）'), ('辅酶Q', '抗自由基（辅酶Q）')]: v = find_table_val(item) if v: metabolism[key] = int(v) return metabolism def _parse_bj_antibiotics_from_tables(find_table_rows): """从表格提取抗生素风险评估""" ab = {} for cat in ANTIBIOTIC_CATEGORIES_BJ: rows = find_table_rows(cat, val_col=2) for name, val in rows: ab[cat] = int(val) return ab def _parse_bj_toxins_from_tables(all_tables, find_table_rows): """从表格提取毒性物质清除能力评估""" toxins = {} toxin_items = [ "苯甲酸酯", "对氨基苯甲酸乙酯", "对氟苯甲酸乙酯", "氯烷烃和氯烯烃", "氯代环己烷", "氯苯", "甲苯", "二甲苯", "硝基甲苯", "乙苯", "苯乙烯", "阿特拉津", "己内酰胺", "双酚", "二恶英", "萘", "多环芳烃", ] for item in toxin_items: rows = find_table_rows(item[:3], name_col=1, val_col=2) for name, val in rows: toxins[item] = int(val) return toxins def _parse_bj_overview(text): """提取北京报告概述（统一到标准报告 overview 结构）""" r = {} m = re.search(r'检测编号[：:\s]*(\S+)', text) if m: r['report_number'] = m.group(1) m = re.search(r'姓\s*名[：:\s]*(\S+)', text) if m: r['person_name'] = m.group(1) m = re.search(r'年\s*龄[：:\s]*(\S+)', text) if m: r['age'] = m.group(1) m = re.search(r'性\s*别[：:\s]*(\S+)', text) if m: r['gender'] = 'male' if m.group(1) in ('男', 'M') else 'female' # 肠型 gut_type = _parse_bj_gut_type(text) if gut_type: r['gut_type'] = gut_type # GMHI → 健康总分 m = re.search(r'肠道微生物健康指数[：:\s]*(\d+)', text) if m: r['overallScore'] = int(m.group(1)) return r def _parse_beijing_report(file_path: str) -> dict: """解析北京菌群报告 PDF，输出统一结构（pdfplumber 表格+文本混合提取）""" try: import pdfplumber with pdfplumber.open(file_path) as pdf: text = '\n'.join(page.extract_text() or '' for page in pdf.pages) # 提取所有表格 all_tables = [] for page in pdf.pages: tables = page.extract_tables() for table in tables: if table and len(table) >= 2: all_tables.append(table) except Exception as e: from PyPDF2 import PdfReader reader = PdfReader(file_path) text = '\n'.join((page.extract_text() or '') for page in reader.pages) all_tables = [] def find_table_rows(keyword, val_col=1, name_col=0, try_all=False): """从表格中找指定关键词的数据行""" rows = [] for table in all_tables: for row in table: row_str = ' '.join(str(c) for c in row if c) if keyword in row_str: if name_col < len(row) and val_col < len(row): name = str(row[name_col] or '').strip() val = str(row[val_col] or '').strip() if name and val and val.replace('.','').replace('-','').isdigit(): rows.append((name, val)) if rows and not try_all: break return rows def find_table_val(keyword, val_col=1, name_col=0): """从表格中找单个值""" for table in all_tables: for row in table: row_str = ' '.join(str(c) for c in row if c) if keyword in row_str: if val_col < len(row): val = str(row[val_col] or '').strip() if val and val.replace('.','').isdigit(): return val return None result = { 'format': 'beijing', 'overview': _parse_bj_overview(text), 'disease_risks': [], 'nutrition': [], 'amino_acids': [], 'vitamins': [], 'trace_elements': [], '菌群检出详细列表': {}, '个体化食物推荐表': {'格式': 'not_found', '条目数': 0, '数据': []}, # 北京报告特有指标 '肠道微生物健康指数': _parse_bj_gmhi(text), '菌群多样性': _parse_bj_diversity(text), '肠道黏膜屏障': _parse_bj_barrier(text), '菌群表型评估': _parse_bj_phenotype(text), '短链脂肪酸合成能力': _parse_bj_scfa_from_tables(find_table_rows), '肠道菌群精准分布': _parse_bj_core_bacteria_from_tables(all_tables), '营养物质及营养素代谢评估': _parse_bj_nutrient_from_tables(find_table_rows, find_table_val), '抗生素风险评估': _parse_bj_antibiotics_from_tables(find_table_rows), '毒性物质清除能力评估': _parse_bj_toxins_from_tables(all_tables, find_table_rows), '趣味肠菌评估': _parse_bj_psych_eval(text, find_table_val), '胃肠道感染病原体': [], '健康整体评估': {}, '肠道菌群主要检测结果': {}, } # 从表格更新 GMHI/抗炎/免疫/纤维/多样性/屏障 gmhi_val = find_table_val('肠道微生物健康指数', val_col=1) if gmhi_val: result['肠道微生物健康指数']['肠道微生物健康指数'] = float(gmhi_val) score = float(gmhi_val) if score >= 90: result['肠道微生物健康指数']['健康状态'] = '健康' elif score >= 80: result['肠道微生物健康指数']['健康状态'] = '健康倾向' elif score >= 40: result['肠道微生物健康指数']['健康状态'] = '亚健康' elif score >= 20: result['肠道微生物健康指数']['健康状态'] = '不健康倾向' else: result['肠道微生物健康指数']['健康状态'] = '不健康' for attr, kw in [('肠道抗炎能力', '抗炎能力'), ('肠道免疫力', '免疫力'), ('肠道膳食纤维需求', '膳食纤维需求')]: v = find_table_val(kw) if v: result.setdefault('肠道菌群主要检测结果', {})[attr] = int(v) if '抗炎' in kw: result['overview']['inflammationScore'] = int(v) elif '免疫' in kw: result['overview']['immunityScore'] = int(v) # 多样性 div_val = find_table_val('肠道微生物多样性', val_col=1) if div_val: result['菌群多样性']['Shannon指数'] = float(div_val) # BE比值 be_val = find_table_val('B/E比值', val_col=1) if be_val: result['肠道黏膜屏障']['BE比值'] = float(be_val) bi_val = find_table_val('双歧杆菌属', val_col=1) if bi_val: result['肠道黏膜屏障']['双歧杆菌属'] = float(bi_val) ent_val = find_table_val('肠杆菌科', val_col=1) if ent_val: result['肠道黏膜屏障']['肠杆菌科'] = float(ent_val) # 肠型 gut_type = _parse_bj_gut_type(text) balance = _parse_bj_balance(text) if gut_type or balance: result['健康整体评估'] = {} if gut_type: result['健康整体评估']['肠型'] = gut_type if balance: result['健康整体评估']['菌群平衡评估'] = balance # 疾病风险评估（文本提取） disease_risks = _parse_bj_disease_risks(text) if disease_risks: result['disease_risks'] = disease_risks # 病原体 pathogens = _parse_bj_pathogens(text) if pathogens: result['胃肠道感染病原体'] = pathogens # 营养物质 → 标准字段映射 metabolism = result.get('营养物质及营养素代谢评估', {}) if metabolism.get('主要营养物质代谢'): result['nutrition'] = [ {'name': k, 'value': str(v), 'status': '正常' if v >= 60 else '偏低'} for k, v in metabolism['主要营养物质代谢'].items() ] if metabolism.get('氨基酸'): result['amino_acids'] = [ {'name': k, 'value': str(v), 'status': '正常' if v >= 60 else '偏低'} for k, v in metabolism['氨基酸'].items() ] if metabolism.get('维生素'): result['vitamins'] = [ {'name': k, 'value': str(v), 'status': '正常' if v >= 60 else '偏低'} for k, v in metabolism['维生素'].items() ] if metabolism.get('微量元素'): result['trace_elements'] = [ {'name': k, 'value': str(v), 'status': '正常' if v >= 60 else '偏低'} for k, v in metabolism['微量元素'].items() ] return result