""" 菌群报告 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': '氏', } 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(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 # 菌群检出详细列表（核心/益生菌/有害菌/病原菌 + 门纲目科属种） 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 } 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