""" 报告解析 Agent：算法解析 + LLM 兜底 + 多类型报告支持 """ import json import logging import base64 from typing import Optional from app.config import settings from app.parsers.pdf_parser import parse_report_pdf_with_fallback logger = logging.getLogger(__name__) # 算法解析器使用中文键名（与 extract_full_report_v5.py 一致）， # Java 消费端期望英文键名；在此做双向映射，Java 读取英文键即可。 _CN_TO_EN_OVERVIEW: dict[str, str] = { '健康总分': 'overallScore', '菌群健康': 'gutHealthScore', '慢病控制': 'chronicDiseaseScore', '营养均衡': 'nutritionScore', '肠道菌群平衡': 'balanceScore', '菌群多样性': 'diversityScore', '有益菌': 'beneficialScore', '有害菌': 'harmfulScore', '核心菌属': 'coreGenusScore', } # 指标分区 → Java 消费端 indicators 的 category（与 Java PdfParseService 兜底路径保持一致） _INDICATOR_SECTION_CATEGORY: list[tuple[str, str]] = [ ('nutrition', '主要营养评估'), ('amino_acids', '氨基酸评估'), ('vitamins', '维生素评估'), ('trace_elements', '微量元素评估'), ('抗生素风险评估', '抗生素耐药'), ('肠道屏障及代谢物', '肠道屏障功能'), ('短链脂肪酸', '短链脂肪酸'), ('神经递质及激素', '神经递质与激素'), ] # 菌群检出详细列表分组 → Java 消费端键 _FLORA_GROUP_TO_KEY: dict[str, str] = { '核心菌属': 'gut_flora', '益生菌': 'probiotic_species', '菌纲构成': 'taxonomy_class', '菌目构成': 'taxonomy_order', '菌科构成': 'taxonomy_family', '菌属构成': 'taxonomy_genus', '菌种构成': 'taxonomy_species', '病原菌属': 'pathogen_genus', '病原菌检出': 'pathogen_detection', } # Java 端无独立字段、需并入 gut_flora（带 category 区分）的分组 _FLORA_EXTRA_GROUPS: list[str] = ['有害菌属', '其它重要菌属'] class ReportParseAgent: """报告解析 Agent""" def __init__(self): self.llm_api_key = getattr(settings, 'llm_api_key', '') @staticmethod def _normalize_overview_keys(result: dict) -> None: """将 overview 中的中文键名映射为英文键名（双向写入），兼容 Java 消费端。""" overview = result.get('overview') if not overview: return for _cn, _en in _CN_TO_EN_OVERVIEW.items(): if _cn in overview and _en not in overview: overview[_en] = overview[_cn] @staticmethod def _normalize_for_java(result: dict) -> None: """将算法解析结果归一化为 Java 消费端 snake_case 键。 保留中文键供调试/LLM 兜底；仅当构建出非空列表时才覆盖 result 对应键， 避免覆盖 LLM 兜底直接产出的 indicators/flora/foods。 """ # 1. indicators：合并各指标分区（含主要消化道致病菌特殊键名） indicators = [] for key, category in _INDICATOR_SECTION_CATEGORY: items = result.get(key) if not isinstance(items, list): continue for it in items: if not isinstance(it, dict) or not it.get('name'): continue indicators.append({ 'category': category, 'indicatorName': it['name'], 'indicatorValue': it.get('value', ''), 'unit': '', 'refRange': it.get('refRange', ''), 'status': it.get('status', ''), 'symptoms': '', }) for it in result.get('主要消化道致病菌') or []: if not isinstance(it, dict): continue name = it.get('name') or it.get('致病菌') if name: indicators.append({ 'category': '主要消化道致病菌', 'indicatorName': name, 'indicatorValue': it.get('value') or it.get('丰度', ''), 'unit': '', 'refRange': '', 'status': it.get('status') or it.get('评估', ''), 'symptoms': '', }) if indicators: result['indicators'] = indicators # 2. 菌群分组 → snake_case 键（条目字段中文 → 英文） flora = result.get('菌群检出详细列表') if isinstance(flora, dict): for group, en_key in _FLORA_GROUP_TO_KEY.items(): converted = [] for it in flora.get(group) or []: if not isinstance(it, dict) or not it.get('名称'): continue converted.append({ 'name': it['名称'], 'value': it.get('丰度%', ''), 'normal_range': it.get('正常范围%', ''), 'population_level': it.get('人群水平%', ''), 'detection_rate': it.get('检出率%', ''), 'description': it.get('说明', ''), 'category': group, 'level': it.get('水平', ''), }) if converted: result[en_key] = converted extra = [] for group in _FLORA_EXTRA_GROUPS: for it in flora.get(group) or []: if not isinstance(it, dict) or not it.get('名称'): continue extra.append({ 'name': it['名称'], 'value': it.get('丰度%', ''), 'normal_range': it.get('正常范围%', ''), 'population_level': it.get('人群水平%', ''), 'detection_rate': it.get('检出率%', ''), 'description': it.get('说明', ''), 'category': group, 'level': it.get('水平', ''), }) if extra: result['gut_flora'] = (result.get('gut_flora') or []) + extra # 3. foods：个体化食物推荐表 → snake_case food_table = result.get('个体化食物推荐表') rows = food_table.get('数据') if isinstance(food_table, dict) else food_table foods = [] if isinstance(rows, list): for it in rows: if not isinstance(it, dict) or not it.get('名称'): continue foods.append({ 'name': it['名称'], 'category': it.get('分类', ''), 'score': it.get('推荐指数'), 'energy_kj': it.get('能量KJ'), 'protein': it.get('蛋白g'), 'fat': it.get('脂肪g'), 'carbs': it.get('碳水化合物g'), 'starch': it.get('淀粉g'), 'fiber': it.get('总膳食纤维g'), 'cholesterol': it.get('胆固醇mg'), }) if foods: result['foods'] = foods async def parse(self, file_path: str) -> dict: """解析 PDF 报告，算法解析 + LLM 兜底""" # 1. 算法解析 result = parse_report_pdf_with_fallback(file_path) logger.info("算法解析完成: format=%s, overview_keys=%d", result.get('format'), len(result.get('overview', {}))) # 1.5 归一化 overview 键名：中文 → 英文（Java 消费端兼容） self._normalize_overview_keys(result) # 1.6 归一化指标/菌群/食物为 Java 消费端 snake_case 键 self._normalize_for_java(result) # 2. 如果解析不完整，LLM 兜底 if result.get('_parse_incomplete') or not result.get('disease_risks'): logger.info("算法解析不完整，尝试 LLM 兜底") llm_result = await self._parse_with_llm(file_path) if llm_result: # 合并 LLM 结果到算法结果上（LLM 覆盖缺失字段） for key in ['disease_risks', 'nutrition', 'amino_acids', 'vitamins', 'trace_elements', 'indicators']: if key in llm_result and not result.get(key): result[key] = llm_result[key] if llm_result.get('overview'): for k, v in llm_result['overview'].items(): if k not in result.get('overview', {}): result.setdefault('overview', {})[k] = v # 重新归一化：LLM 补齐的分区也要并入 indicators self._normalize_for_java(result) # 清理内部标记 result.pop('_parse_incomplete', None) return result async def parse_generic(self, file_path: str, extra_context: Optional[dict] = None) -> dict: """通用报告 LLM 解析（不经过算法解析，直接走 LLM）""" try: from PyPDF2 import PdfReader reader = PdfReader(file_path) text = '\n'.join(page.extract_text() or '' for page in reader.pages) ctx_str = "" if extra_context: ctx_str = f"\n额外上下文：{json.dumps(extra_context, ensure_ascii=False)}" prompt = f"""你是一个通用报告解析专家。请从以下PDF文本中提取结构化数据，返回JSON格式。 报告文本内容： {text[:12000]}{ctx_str} 请分析这份报告，推断它的类型和内容，然后按以下JSON Schema返回： {{ "reportType": "推断的报告类型名称", "reportTypeFamily": "报告家族分类（如: dan/cognitive/gut_flora/health_check/other）", "confidence": "high/medium/low", "summary": {{ "personName": "姓名", "reportDate": "报告日期", "reportNumber": "报告编号", "overallScore": "总分（如果有）", "interpretation": "报告整体解读摘要" }}, "indicators": [ {{"name": "指标名称", "value": "数值", "category": "分类", "status": "状态"}} ], "sections": [ {{"title": "段落标题", "content": "段落内容摘要", "items": [{{"name": "...", "value": "..."}}]}} ], "textFeatures": ["文本特征1", "文本特征2", ...] }} 只返回JSON，不要其他文字。""" if self.llm_api_key: import httpx async with httpx.AsyncClient(timeout=120) as client: resp = await client.post( f"{settings.llm_base_url}/chat/completions", json={ "model": settings.llm_model or "gpt-4o", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, }, headers={"Authorization": f"Bearer {self.llm_api_key}"}, ) resp.raise_for_status() data = resp.json() content = data['choices'][0]['message']['content'] content = content.replace('```json', '').replace('```', '').strip() return json.loads(content) else: logger.warning("LLM 未配置，返回空") return {"reportType": "unknown", "summary": {}, "indicators": [], "sections": []} except Exception as e: logger.error("通用LLM解析失败: %s", e) return {"reportType": "unknown", "error": str(e), "summary": {}, "indicators": [], "sections": []} async def _parse_with_llm(self, file_path: str) -> Optional[dict]: """LLM 兜底解析""" try: from PyPDF2 import PdfReader reader = PdfReader(file_path) text = '\n'.join(page.extract_text() or '' for page in reader.pages) # 构造 prompt prompt = f"""你是一个肠道菌群检测报告解析专家。请从以下PDF文本中提取结构化数据，返回JSON格式。 文本内容： {text[:8000]} 请按以下JSON Schema返回： {{ "overview": {{ "person_name": "", "report_number": "", "age": 0, "gender": "male/female", "overallScore": 0, "gutHealthScore": 0, "chronicDiseaseScore": 0, "nutritionScore": 0, "gutAge": "", "gutType": "" }}, "disease_risks": [{{"name": "", "value": "", "status": ""}}], "nutrition": [{{"name": "", "value": "", "status": ""}}], "amino_acids": [{{"name": "", "value": "", "status": ""}}], "vitamins": [{{"name": "", "value": "", "status": ""}}], "trace_elements": [{{"name": "", "value": "", "status": ""}}] }} 只返回JSON，不要其他文字。""" if self.llm_api_key: # 调 OpenAI 兼容 API import httpx async with httpx.AsyncClient(timeout=60) as client: resp = await client.post( f"{settings.llm_base_url}/chat/completions", json={ "model": settings.llm_model or "gpt-4o", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, }, headers={"Authorization": f"Bearer {self.llm_api_key}"}, ) resp.raise_for_status() data = resp.json() content = data['choices'][0]['message']['content'] content = content.replace('```json', '').replace('```', '').strip() return json.loads(content) else: logger.warning("LLM 未配置，跳过 LLM 兜底") return None except Exception as e: logger.warning("LLM 解析失败: %s", e) return None # ================================================================ # 多类型报告解析 # ================================================================ async def parse_by_type(self, file_path: str, report_type: str, extra_context: Optional[dict] = None) -> dict: """按报告类型路由到专用解析器。 :param file_path: PDF 文件路径 :param report_type: 报告类型标识: - "brain_status": 脑状态测量报告 - "cognitive_aptitude": 先天智力潜能/皮纹学测评报告 - "scanned_image": 扫描图片 PDF（无文字层，需多模态 LLM） - "auto": 自动检测类型 :return: 结构化解析结果 dict """ if report_type == "auto": report_type = self._detect_report_type(file_path) logger.info("自动检测报告类型: %s", report_type) if report_type == "brain_status": return await self._parse_brain_status(file_path) elif report_type == "cognitive_aptitude": return await self._parse_cognitive_aptitude(file_path) elif report_type == "scanned_image": return await self._parse_scanned_image(file_path) else: logger.info("未识别的报告类型 %s，走通用解析", report_type) return await self.parse_generic(file_path, extra_context) # ---- 格式检测 ---- def _detect_report_type(self, file_path: str) -> str: """根据 PDF 文本内容自动检测报告类型。""" try: from PyPDF2 import PdfReader reader = PdfReader(file_path) # 检测是否有可提取文本 total_text = "" for i in range(min(12, len(reader.pages))): t = reader.pages[i].extract_text() or "" total_text += t if len(total_text.strip()) < 20: # 前几页几乎无文字 → 可能是扫描图片 PDF return "scanned_image" # 脑状态测量报告 if "脑状态测量报告" in total_text or "大脑综合状态" in total_text: return "brain_status" # 智力潜能 / 皮纹学报告 if ("先天数据" in total_text or "智力潜能" in total_text or "皮纹" in total_text or "ATD" in total_text or "trc" in total_text.lower()): return "cognitive_aptitude" # 北京肠道菌群报告 if ("肠道菌群检测" in total_text and "高通量测序" in total_text): return "gut_flora_beijing" return "generic" except Exception as e: logger.warning("报告类型检测失败: %s", e) return "generic" # ---- 通用 LLM 调用 ---- async def _call_llm(self, prompt: str, timeout: int = 120) -> str: """调用 LLM 并返回文本响应。 :raises RuntimeError: LLM 未配置或调用失败 """ if not self.llm_api_key: raise RuntimeError("LLM 未配置") import httpx async with httpx.AsyncClient(timeout=timeout) as client: resp = await client.post( f"{settings.llm_base_url}/chat/completions", json={ "model": settings.llm_model or "gpt-4o", "messages": [{"role": "user", "content": prompt}], "temperature": 0.1, }, headers={"Authorization": f"Bearer {self.llm_api_key}"}, ) resp.raise_for_status() data = resp.json() content = data['choices'][0]['message']['content'] return content.replace('```json', '').replace('```', '').strip() async def _call_llm_vision(self, prompt: str, image_base64: str, timeout: int = 180) -> str: """调用多模态 LLM（vision），传入图片 + 文字提示。 :raises RuntimeError: LLM 未配置或调用失败 """ if not self.llm_api_key: raise RuntimeError("LLM 未配置") import httpx async with httpx.AsyncClient(timeout=timeout) as client: resp = await client.post( f"{settings.llm_base_url}/chat/completions", json={ "model": settings.llm_model or "gpt-4o", "messages": [ { "role": "user", "content": [ {"type": "text", "text": prompt}, { "type": "image_url", "image_url": { "url": f"data:image/png;base64,{image_base64}", }, }, ], } ], "temperature": 0.1, }, headers={"Authorization": f"Bearer {self.llm_api_key}"}, ) resp.raise_for_status() data = resp.json() content = data['choices'][0]['message']['content'] return content.replace('```json', '').replace('```', '').strip() def _pdf_to_text(self, file_path: str, max_pages: int = 0) -> str: """用 PyPDF2 提取 PDF 全文。max_pages=0 表示全部。""" from PyPDF2 import PdfReader reader = PdfReader(file_path) n = len(reader.pages) if max_pages == 0 else min(max_pages, len(reader.pages)) return '\n'.join(reader.pages[i].extract_text() or '' for i in range(n)) # ---- 1. 脑状态测量报告 ---- async def _parse_brain_status(self, file_path: str) -> dict: """解析脑状态测量报告。 文本可提取，数据清晰。使用 LLM 提取结构化数据。 返回结构: { "reportType": "brain_status", "reportTypeFamily": "cognitive", "summary": { "personName": "张文远", "age": 44, "gender": "male", "reportDate": "2026-05-03", "reportNumber": "CL79474", "overallScore": 82.1, "overallLevel": "良好", }, "indicators": [ {"name": "大脑综合状态得分", "value": "82.1", "category": "综合评分", "status": "良好"}, {"name": "脑供血问题风险评估", "value": "低风险", "category": "风险评估", "status": "低风险"}, ... ], "sections": [ {"title": "疲劳评估", "content": "...", "items": [...]}, {"title": "情绪评估", "content": "...", "items": [...]}, {"title": "睡眠评估", "content": "...", "items": [...]}, ] } """ logger.info("开始解析脑状态测量报告: %s", file_path) text = self._pdf_to_text(file_path) prompt = f"""你是一个脑状态测量报告解析专家。请从以下文本中提取结构化数据，返回JSON格式。 报告文本内容： {text} 请按以下JSON Schema返回： {{ "reportType": "brain_status", "reportTypeFamily": "cognitive", "summary": {{ "personName": "姓名", "age": 年龄数字, "gender": "male/female", "reportDate": "报告日期 yyyy-MM-dd", "reportNumber": "报告编号", "overallScore": 综合状态得分数值, "overallLevel": "良好/一般/较差等文字描述" }}, "indicators": [ {{"name": "指标名称", "value": "数值或等级", "category": "分类", "status": "状态描述"}} ], "sections": [ {{ "title": "段落标题（如：健康风险评估/疲劳评估/情绪评估/睡眠评估等）", "content": "段落摘要", "items": [ {{"name": "子项名称", "value": "数值", "status": "状态/等级"}} ] }} ] }} 注意： 1. 尽量提取所有出现的评估指标，包括：大脑综合状态得分、大脑健康状态得分、大脑能力状态得分、 脑供血问题风险、脑供氧问题风险、思维负荷、思维状态风险、大脑疲劳评估、用脑模式、 焦虑情绪评估、抑郁情绪评估、抵触情绪评估、安全感评估、情绪管理评估、综合情绪评估、 睡眠效果评估等 2. 数值尽量提取数字，状态文字原样保留 3. 只返回JSON，不要其他文字。""" try: content = await self._call_llm(prompt, timeout=120) result = json.loads(content) logger.info("脑状态报告解析完成: indicators=%d, sections=%d", len(result.get('indicators', [])), len(result.get('sections', []))) return result except Exception as e: logger.error("脑状态报告解析失败: %s", e) return { "reportType": "brain_status", "reportTypeFamily": "cognitive", "error": str(e), "summary": {}, "indicators": [], "sections": [], } # ---- 2. 智力潜能 / 皮纹学测评报告 ---- async def _parse_cognitive_aptitude(self, file_path: str) -> dict: """解析先天智力潜能/皮纹学测评报告。 115页，文本碎片化严重。分块提取 + LLM 合并。 返回结构: { "reportType": "cognitive_aptitude", "reportTypeFamily": "cognitive", "summary": { "personName": "张老师", "gender": "male", "region": "北京", "phone": "15901552192", "testDate": "2026-04-03", "birthday": "1982-02-06", }, "indicators": [ {"name": "TRC", "value": "107+X+M", "category": "先天数据", "status": ""}, {"name": "ATD", "value": "35.5", "category": "思维敏捷性", "status": "超级敏感型"}, ... ], "sections": [ {"title": "智力潜能测评", "content": "...", "items": [...]}, {"title": "学习风格测评", "content": "...", "items": [...]}, {"title": "八大智能测试", "content": "...", "items": [...]}, ... ] } """ logger.info("开始解析智力潜能测评报告: %s", file_path) from PyPDF2 import PdfReader reader = PdfReader(file_path) total_pages = len(reader.pages) # 分块提取文本 — 每块约 8000 字符 all_pages = [] for i in range(total_pages): t = reader.pages[i].extract_text() or '' all_pages.append((i + 1, t)) # 合并前 5 页（基本信息区）作为第一块 first_chunk = '\n'.join(t for _, t in all_pages[:5]) # 合并中间数据页（6-60页）作为第二块 mid_chunk = '\n'.join(t for _, t in all_pages[5:min(30, len(all_pages))]) # 合并后续页（30-62页有文字的） later_chunk = '\n'.join(t for _, t in all_pages[30:min(62, len(all_pages))]) # 第一块: 基本信息 + 目录 prompt1 = f"""你是一个先天智力潜能（皮纹学）测评报告解析专家。请从以下文本中提取基本信息和目录结构，返回JSON格式。 报告文本内容（前5页）： {first_chunk} 请按以下JSON Schema返回： {{ "reportType": "cognitive_aptitude", "reportTypeFamily": "cognitive", "summary": {{ "personName": "姓名", "gender": "male/female", "region": "地区", "phone": "电话", "testDate": "测评日期", "birthday": "出生日期", "trc": "TRC值（如107+X+M）", "atd": "ATD角度值", "learningType": "学习类型（听觉型/体觉型/视觉型）", "motivationType": "动机类型", "cognitiveType": "认知类型", "brainDominance": "左脑型/右脑型/全脑型" }}, "toc": ["章节1标题", "章节2标题", ...] }} 只返回JSON，不要其他文字。""" # 第二块: 智力潜能/学习风格/八大智能等核心数据 prompt2 = f"""你是一个先天智力潜能（皮纹学）测评报告解析专家。请从以下文本中提取测评指标，返回JSON格式。 报告文本内容（数据页）： {mid_chunk} 请按以下JSON Schema返回： {{ "indicators": [ {{"name": "指标名称", "value": "数值", "category": "分类", "status": "状态/描述"}} ], "sections": [ {{ "title": "段落标题（如：智力潜能测评/学习风格测评/左右脑功能/先天性格/八大智能等）", "content": "段落摘要", "items": [{{"name": "子项名称", "value": "数值或描述", "status": "状态"}}] }} ] }} 注意： 1. 指标包括但不限于：TRC(总脊纹数)、ATD(思维敏捷性角)、各脑区指标、 八大智能（语言/逻辑数学/空间/身体动觉/音乐/人际/内省/自然观察）、 先天学习潜能、先天行为导向、先天学习管道等 2. 如果某个指标的数值看起来是百分比或数值，直接提取 3. 只返回JSON，不要其他文字。""" # 第三块: 后续章节（学科建议/职业建议/心理学测试等） prompt3 = f"""你是一个先天智力潜能（皮纹学）测评报告解析专家。请从以下文本中提取测评建议和心理学测试结果，返回JSON格式。 报告文本内容（后续页面）： {later_chunk} 请按以下JSON Schema返回： {{ "sections2": [ {{ "title": "段落标题（如：性格色彩/大五人格/MBTI/学科选择/职业能力/感觉统合等）", "content": "段落摘要", "items": [{{"name": "子项名称", "value": "数值或描述", "status": "状态"}}] }} ], "recommendations": [ {{"category": "建议类别", "content": "建议内容"}} ] }} 只返回JSON，不要其他文字。""" result = { "reportType": "cognitive_aptitude", "reportTypeFamily": "cognitive", "summary": {}, "indicators": [], "sections": [], } try: # 并行请求三块 import asyncio tasks = [] if first_chunk.strip(): tasks.append(self._call_llm(prompt1, timeout=120)) if mid_chunk.strip(): tasks.append(self._call_llm(prompt2, timeout=120)) if later_chunk.strip(): tasks.append(self._call_llm(prompt3, timeout=120)) responses = await asyncio.gather(*tasks, return_exceptions=True) # 合并结果 for i, resp in enumerate(responses): if isinstance(resp, Exception): logger.warning("智力潜能报告第%d块解析失败: %s", i + 1, resp) continue try: chunk_result = json.loads(str(resp)) if i == 0: # 第一块: summary + toc result['summary'] = chunk_result.get('summary', {}) result['toc'] = chunk_result.get('toc', []) elif i == 1: # 第二块: indicators + sections result['indicators'] = chunk_result.get('indicators', []) result['sections'] = chunk_result.get('sections', []) elif i == 2: # 第三块: sections2 + recommendations if chunk_result.get('sections2'): result['sections'].extend(chunk_result['sections2']) if chunk_result.get('recommendations'): result['recommendations'] = chunk_result['recommendations'] except (json.JSONDecodeError, KeyError) as e: logger.warning("智力潜能报告第%d块JSON解析失败: %s", i + 1, e) logger.info("智力潜能报告解析完成: indicators=%d, sections=%d", len(result.get('indicators', [])), len(result.get('sections', []))) return result except Exception as e: logger.error("智力潜能报告解析失败: %s", e) return result # ---- 3. 扫描图片 PDF（多模态 LLM） ---- async def _parse_scanned_image(self, file_path: str) -> dict: """解析扫描图片 PDF（无文字层）。 流程: 1. 用 PyPDF2 提取页面图片（base64 编码） 2. 逐页发送给多模态 LLM 进行 OCR + 结构化提取 3. 合并各页结果 返回结构与 parse_generic 一致: { "reportType": "推断的类型", "reportTypeFamily": "报告家族", "summary": {...}, "indicators": [...], "sections": [...] } """ logger.info("开始解析扫描图片PDF: %s", file_path) from PyPDF2 import PdfReader reader = PdfReader(file_path) total_pages = len(reader.pages) all_indicators = [] all_sections = [] summary = {} detected_type = "unknown" for page_idx in range(total_pages): page = reader.pages[page_idx] image_b64 = self._extract_page_image_base64(page) if not image_b64: logger.warning("第%d页无图片可提取", page_idx + 1) continue prompt = f"""请分析这张报告图片（第{page_idx + 1}页，共{total_pages}页），提取其中的所有结构化数据。 请按以下JSON Schema返回： {{ "reportType": "推断的报告类型名称", "reportTypeFamily": "报告家族分类（如: gut_flora/health_check/cognitive/brain_status/other）", "summary": {{ "personName": "姓名（如果能识别）", "reportDate": "报告日期（如果能识别）", "reportNumber": "报告编号（如果能识别）", "overallScore": "总分（如果可见）", "interpretation": "本页内容摘要" }}, "indicators": [ {{"name": "指标名称", "value": "数值", "category": "分类", "status": "状态"}} ], "sections": [ {{"title": "段落标题", "content": "段落内容摘要", "items": [{{"name": "...", "value": "..."}}]}} ] }} 注意： 1. 请仔细阅读图片中的所有文字，包括表格数据、数值、状态描述 2. 如果是菌群报告，提取菌属名称、丰度值、参考范围等 3. 如果是体检报告，提取各项检查指标、数值、参考范围、异常标记 4. 数值尽量提取精确数字，状态文字原样保留 5. 只返回JSON，不要其他文字。""" try: content = await self._call_llm_vision(prompt, image_b64, timeout=180) page_result = json.loads(content) # 合并指标 if page_result.get('indicators'): all_indicators.extend(page_result['indicators']) # 合并段落 if page_result.get('sections'): all_sections.extend(page_result['sections']) # 合并基本信息（第一页优先） if page_idx == 0 and page_result.get('summary'): summary = page_result['summary'] # 更新检测到的类型 if page_result.get('reportType') and detected_type == "unknown": detected_type = page_result['reportType'] logger.info("扫描PDF第%d页解析完成: indicators=%d", page_idx + 1, len(page_result.get('indicators', []))) except Exception as e: logger.warning("扫描PDF第%d页解析失败: %s", page_idx + 1, e) result = { "reportType": detected_type, "reportTypeFamily": "other", "summary": summary, "indicators": all_indicators, "sections": all_sections, } logger.info("扫描图片PDF解析完成: pages=%d, indicators=%d, sections=%d", total_pages, len(all_indicators), len(all_sections)) return result def _extract_page_image_base64(self, page) -> str: """从 PDF 页面对象中提取图片，返回 base64 编码字符串。 支持以下情况： - 页面包含 /XObject 中的 /Image 类型对象 - 页面是整页图片（常见于扫描件） :return: base64 编码的 PNG 图片，或空字符串（无图片） """ try: from PyPDF2 import PdfReader import io # 获取页面资源 resources = page.get('/Resources') if not resources: return "" x_objects = resources.get('/XObject') if not x_objects: return "" x_obj = x_objects.get_object() # 找最大的图片对象 best_image = None best_size = 0 for name in x_obj: obj = x_obj[name] obj_resolved = obj.get_object() subtype = obj_resolved.get('/Subtype') if subtype != '/Image': continue width = int(obj_resolved.get('/Width', 0)) height = int(obj_resolved.get('/Height', 0)) size = width * height if size > best_size: best_size = size best_image = obj_resolved if not best_image: return "" # 提取图片数据 color_space = best_image.get('/ColorSpace', '/DeviceRGB') bits_per_component = int(best_image.get('/BitsPerComponent', 8)) width = int(best_image.get('/Width', 0)) height = int(best_image.get('/Height', 0)) raw_data = best_image.get_data() # 转换为 PIL Image → PNG → base64 # 尝试安装 Pillow（如果没有） try: from PIL import Image except ImportError: logger.warning("Pillow 未安装，尝试基础 base64 编码") # 如果没有 Pillow，直接 base64 编码原始数据 return base64.b64encode(raw_data).decode('utf-8') # 根据 ColorSpace 确定模式 if isinstance(color_space, str): if 'RGB' in color_space or 'RGB' in str(color_space): mode = 'RGB' elif 'Gray' in color_space or 'Gray' in str(color_space): mode = 'L' elif 'CMYK' in color_space: mode = 'CMYK' else: mode = 'RGB' else: mode = 'RGB' if width > 0 and height > 0: img = Image.frombytes(mode, (width, height), raw_data) # CMYK → RGB if mode == 'CMYK': img = img.convert('RGB') # 缩小图片以减少 token 消耗（最大 2000px 边） max_dim = 2000 if max(width, height) > max_dim: ratio = max_dim / max(width, height) new_size = (int(width * ratio), int(height * ratio)) img = img.resize(new_size, Image.LANCZOS) # 转 PNG buf = io.BytesIO() img.save(buf, format='PNG', optimize=True) return base64.b64encode(buf.getvalue()).decode('utf-8') return "" except Exception as e: logger.warning("提取页面图片失败: %s", e) return ""