""" 菌群报告 PDF 解析器 从 extract_full_report_v5.py 提取核心逻辑,封装为可调用函数 """ import re from PyPDF2 import PdfReader # === 常量 === RADICAL_MAP = { '\u2f52': '\u6c0f', '\u2f51': '\u6bcd', '\u2f59': '\u6b6f', '\u2f04': '\u4e59', '\u2f20': '\u4e00', '\u2f21': '\u4e28', '\u2f22': '\u4e3f', '\u2f23': '\u4e39', '\u2f2b': '\u5c38', '\u2f2d': '\u5c71', '\u2f44': '\u4e59', '\u2f53': '\u6c14', '\u2f55': '\u6c34', '\u2f5c': '\u725b', '\u2f5f': '\u7389', '\u2f7a': '\u7f8a', '\u2f81': '\u8089', '\u2f83': '\u81ea', '\u2f8a': '\u8272', '\u2f8e': '\u8840', '\u2f95': '\u79be', '\u2f96': '\u8c46', '\u2faf': '\u9762', '\u2fb9': '\u9999', '\u2fca': '\u9ed1', '\u2ec9': '\u8d1d', '\u2edd': '\u98df', '\u2ee2': '\u9a6c', '\u2ee9': '\u9ec4', } 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 = ['幽门螺杆菌', '艰难梭菌', '沙门氏菌', '志贺氏菌', '弯曲杆菌'] def norm(s): return ''.join(RADICAL_MAP.get(c, c) for c in s) def detect_format(lines): """检测 triplet / inline 格式""" text = '\n'.join(lines) has_triplet = '指标范围' in text and '疾病风险评估' in text for line in lines: if len(line) > 15 and re.search(r'[\u4e00-\u9fff]+[\d.]+[\u4e00-\u9fff/]+', line): for known in KNOWN_DISEASE_RISKS: if known in line: return 'inline' return 'triplet' if has_triplet else 'inline' 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 def parse_report_pdf(file_path: str) -> dict: """主函数:解析 PDF 返回结构化数据""" lines = extract_text(file_path) fmt = detect_format(lines) result = {'format': fmt, 'overview': parse_overview(lines)} # 疾病风险评估 for i, line in enumerate(lines): if '疾病风险评估' in line and '注' not in line: risks = parse_triplet_until(lines[i+1:], ['主要营养评估', '氨基酸评估', '维生素评估', '微量元素评估', '抗生素风险评估']) result['disease_risks'] = [r for r in risks if '注' not in r['name']] break # 主要营养评估 for i, line in enumerate(lines): if '主要营养评估' in line: nutrients = parse_triplet_until(lines[i+1:], ['氨基酸评估']) result['nutrition'] = nutrients[:5] break # 氨基酸评估 for i, line in enumerate(lines): if '氨基酸评估' in line: aminos = parse_triplet_until(lines[i+1:], ['维生素评估', '微量元素评估']) result['amino_acids'] = aminos break # 维生素评估 for i, line in enumerate(lines): if '维生素评估' in line: vits = parse_triplet_until(lines[i+1:], ['微量元素评估', '抗生素风险评估']) result['vitamins'] = [r for r in vits if '维生素' in r['name']] result['trace_elements'] = [r for r in vits if '维生素' not in r['name']] break 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