cfc/cfc-langgraph/app/agents/report_parse_agent.py

"""
报告解析 Agent：算法解析 + LLM 兜底
"""
import json
import logging
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