cfc/cfc-langgraph/app/api/adapter.py

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel, ConfigDict
from typing import Optional
from app.graphs.chat_graph import create_chat_graph
from app.graphs.analysis_graph import create_analysis_graph
from app.graphs.health_coach_graph import create_health_coach_graph
from app.graphs.health_butler_graph import create_health_butler_graph
from app.graphs.nutrition_graph import create_nutrition_graph
from app.tools.java_client import JavaClient
from app.rag.retriever import RagRetriever
from app.config import settings
from app.prompt_service import get_prompt
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage
from app.models.health_plan import HealthPlanResponse, PlanTask
import uuid
import time
import re
import logging
import json

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/v1", tags=["adapter"])

# ---- Dify-compatible request/response models ----

class DifyMessage(BaseModel):
    role: str
    content: str

class DifyChatRequest(BaseModel):
    # 前端 userId/conversationId 可能为数字，允许 int/float 自动转 str，避免 422
    model_config = ConfigDict(coerce_numbers_to_str=True)
    query: str = ""
    user: str = "anonymous"
    conversation_id: str = ""
    messages: list[DifyMessage] = []
    inputs: dict = {}
    response_mode: str = "blocking"  # blocking / streaming
    user_id: str = ""
    bot_name: str = "AI管家"

class DifyAnalysisRequest(BaseModel):
    # 前端 userId 可能为数字，允许 int/float 自动转 str，避免 422
    model_config = ConfigDict(coerce_numbers_to_str=True)
    report_id: Optional[int] = None
    user_id: Optional[str] = None
    focus: Optional[str] = None
    messages: list[DifyMessage] = []
    inputs: dict = {}

class HealthCoachGenerateRequest(BaseModel):
    """Java 后端健康方案生成专用请求体"""
    family_id: Optional[int] = None
    member_ids: Optional[str] = None
    dimensions: Optional[str] = None
    goal: str = ""

class DifyChoice(BaseModel):
    index: int
    message: dict
    finish_reason: str = "stop"

class DifyUsage(BaseModel):
    prompt_tokens: int = 0
    completion_tokens: int = 0
    total_tokens: int = 0

class DifyResponse(BaseModel):
    id: str
    object: str = "chat.completion"
    created: int
    model: str = "langgraph-cfc"
    choices: list[DifyChoice]
    usage: DifyUsage = DifyUsage()
    metadata: dict = {}

# ---- Helpers ----

def _now_ts() -> int:
    return int(time.time())

def _extract_query(req: DifyChatRequest) -> str:
    if req.query:
        return req.query
    # fallback: take last user message
    for msg in reversed(req.messages):
        if msg.role == "user":
            return msg.content
    return ""

def _to_langgraph_context(req: DifyChatRequest) -> dict:
    ctx = {}
    if isinstance(req.inputs, dict):
        ctx["child_id"] = req.inputs.get("child_id")
        ctx["report_id"] = req.inputs.get("report_id")
        ctx["family_id"] = req.inputs.get("family_id")
    return ctx

# ---- Dify-compatible endpoints ----

@router.post("/chat/completion", response_model=DifyResponse)
async def chat_completion(req: DifyChatRequest):
    query = _extract_query(req)
    if not query:
        raise HTTPException(status_code=400, detail="query 为空")

    graph = create_chat_graph()
    initial_state = {
        "query": query,
        "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0,
        "conversation_id": req.conversation_id or None,
        "intent": None,
        "context": _to_langgraph_context(req),
        "messages": None,
        "answer": None,
        "tasks": [],
        "sources": [],
    }
    config = {
        "configurable": {"thread_id": req.conversation_id or str(req.user_id or "default")},
    }
    result = await graph.ainvoke(initial_state, config)
    answer = result.get("answer") or ""
    return DifyResponse(
        id=f"chatcmpl-{uuid.uuid4().hex[:24]}",
        created=_now_ts(),
        choices=[
            DifyChoice(
                index=0,
                message={"role": "assistant", "content": answer},
                finish_reason="stop",
            )
        ],
        usage=DifyUsage(
            prompt_tokens=len(query.split()),
            completion_tokens=len(answer.split()),
            total_tokens=len(query.split()) + len(answer.split()),
        ),
    )


@router.post("/analysis/run", response_model=DifyResponse)
async def analysis_run(req: DifyAnalysisRequest):
    graph = create_analysis_graph()
    report_id = req.report_id
    if report_id is None and isinstance(req.inputs, dict):
        report_id = req.inputs.get("report_id")
    initial_state = {
        "report_id": report_id,
        "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0,
        "focus": req.focus,
        "report_data": None,
        "survey_data": None,
        "dimension_scores": None,
        "analysis": None,
        "recommendations": [],
    }
    result = await graph.ainvoke(initial_state, {})
    analysis = result.get("analysis") or ""
    recommendations = result.get("recommendations") or []
    content = analysis
    if recommendations:
        content += "\n\n建议:\n" + "\n".join(f"- {r}" for r in recommendations)
    return DifyResponse(
        id=f"analysis-{uuid.uuid4().hex[:24]}",
        created=_now_ts(),
        choices=[
            DifyChoice(
                index=0,
                message={"role": "assistant", "content": content},
                finish_reason="stop",
            )
        ],
        usage=DifyUsage(
            prompt_tokens=len((req.focus or "").split()),
            completion_tokens=len(content.split()),
            total_tokens=len((req.focus or "").split()) + len(content.split()),
        ),
    )


@router.post("/health/coach", response_model=DifyResponse)
async def health_coach(req: DifyChatRequest):
    query = _extract_query(req)
    if not query:
        raise HTTPException(status_code=400, detail="query 为空")

    graph = create_health_coach_graph()
    initial_state = {
        "query": query,
        "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0,
        "child_id": int(req.inputs.get("child_id")) if req.inputs.get("child_id") else None,
        "conversation_id": req.conversation_id or None,
        "context": _to_langgraph_context(req),
        "answer": None,
        "sources": [],
        "memory_messages": None,
    }
    config = {
        "configurable": {"thread_id": req.conversation_id or f"health_{req.user_id}"},
    }
    result = await graph.ainvoke(initial_state, config)
    answer = result.get("answer") or ""

    sources = result.get("sources") or []
    metadata = {}
    if sources:
        metadata["sources"] = [
            {"title": s.get("title", ""), "type": s.get("type", "knowledge")}
            for s in sources
        ]

    return DifyResponse(
        id=f"health-{uuid.uuid4().hex[:24]}",
        created=_now_ts(),
        model="langgraph-health-coach",
        choices=[
            DifyChoice(
                index=0,
                message={"role": "assistant", "content": answer},
                finish_reason="stop",
            )
        ],
        usage=DifyUsage(
            prompt_tokens=len(query.split()),
            completion_tokens=len(answer.split()),
            total_tokens=len(query.split()) + len(answer.split()),
        ),
        metadata=metadata,
    )


@router.post("/health/butler", response_model=DifyResponse)
async def health_butler(req: DifyChatRequest):
    """AI 健康管家 — 基于健康知识库检索 + 个性化建议 + 任务生成"""
    query = _extract_query(req)
    if not query:
        raise HTTPException(status_code=400, detail="query 为空")

    graph = create_health_butler_graph()

    # 从 inputs 中提取健康管家需要的上下文
    inputs = req.inputs or {}
    family_id = inputs.get("family_id")
    child_id = inputs.get("child_id")
    report_id = inputs.get("report_id")
    focus = inputs.get("focus")

    # conversation_id 用于 checkpointer thread
    thread_id = req.conversation_id or f"butler_{req.user_id or 'anon'}"

    initial_state = {
        "query": query,
        "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0,
        "conversation_id": thread_id,
        "family_id": int(family_id) if family_id else None,
        "child_id": int(child_id) if child_id else None,
        "report_id": int(report_id) if report_id else None,
        "focus": focus,
        "kb_context": None,
        "knowledge_results": [],
        "answer": None,
        "tasks": [],
        "sources": [],
        "messages": None,
    }
    config = {
        "configurable": {"thread_id": thread_id},
    }
    result = await graph.ainvoke(initial_state, config)
    answer = result.get("answer") or ""

    # 从回答中提取 TASK 标记
    tasks = result.get("tasks") or []
    sources = result.get("sources") or []
    metadata = {"tasks": tasks, "sources": sources}

    return DifyResponse(
        id=f"butler-{uuid.uuid4().hex[:24]}",
        created=_now_ts(),
        model="langgraph-health-butler",
        choices=[
            DifyChoice(
                index=0,
                message={"role": "assistant", "content": answer},
                finish_reason="stop",
            )
        ],
        usage=DifyUsage(
            prompt_tokens=len(query.split()),
            completion_tokens=len(answer.split()),
            total_tokens=len(query.split()) + len(answer.split()),
        ),
        metadata=metadata,
    )


def _parse_normal_range(ref_range: str) -> tuple[float | None, float | None]:
    """解析正常范围字符串，返回 (下限, 上限)，如 '30-100' → (30, 100), '<5' → (None, 5)"""
    if not ref_range:
        return None, None
    ref_range = ref_range.strip()
    m = re.match(r'([<>]=?)\s*([\d.]+)', ref_range)
    if m:
        op, val = m.group(1), float(m.group(2))
        if op.startswith('>'):
            return (val, None)
        else:
            return (None, val)
    m = re.match(r'([\d.]+)\s*[-~]\s*([\d.]+)', ref_range)
    if m:
        return (float(m.group(1)), float(m.group(2)))
    return None, None


def _is_abnormal(status: str, value: float | None, low: float | None, high: float | None) -> bool:
    """判断指标是否异常：优先用 status 字段，否则用数值与范围比较"""
    if status and status not in ("正常", "正常范围", "未检出", ""):
        return True
    if value is not None and low is not None and high is not None:
        return value < low or value > high
    return False


def _try_parse_value(raw: str) -> float | None:
    if not raw:
        return None
    raw = raw.strip().replace(",", "").replace(" ", "")
    try:
        return float(raw)
    except ValueError:
        return None


async def _search_knowledge(retriever: RagRetriever, query: str, k: int = 3) -> list[dict]:
    """从知识库检索相关内容"""
    try:
        return await retriever.retrieve(query, k=k)
    except Exception as e:
        logger.warning("知识库检索失败: %s", e)
        return []


@router.post("/health/coach/generate")
async def health_coach_generate(req: HealthCoachGenerateRequest):
    """健康方案生成 — 选人→拉指标→查知识库→LLM"""
    goal = req.goal or "改善健康状况"
    member_ids_str = req.member_ids or ""
    dimensions = req.dimensions or ""
    member_ids = [m.strip() for m in member_ids_str.split(",") if m.strip()]

    java = JavaClient()
    retriever = RagRetriever(collection_name="cfc_knowledge")
    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=0.3,
    )

    # ====== 1. 获取家庭成员信息 ======
    members_info = []
    if member_ids:
        # 从 context 获取家庭信息
        for uid_str in member_ids:
            uid = int(uid_str)
            ctx = await java.get_family_context(uid, "child_info")
            children = ctx.get("children", []) if isinstance(ctx, dict) else []
            for child in children:
                child_id = str(child.get("用户ID", ""))
                if child_id in member_ids:
                    members_info.append({
                        "id": child_id,
                        "name": child.get("姓名", f"成员{child_id}"),
                        "age": child.get("年龄", "未知"),
                        "energy": child.get("能量", 0),
                    })
                    break
            # 如果 context 没找到，用基本信息兜底
            if not any(m["id"] == uid_str for m in members_info):
                members_info.append({"id": uid_str, "name": f"成员{uid_str}", "age": "未知", "energy": 0})
    else:
        members_info.append({"id": "0", "name": "用户", "age": "未知", "energy": 0})

    # ====== 2. 获取每个成员的指标数据 ======
    all_indicators = []
    for member in members_info:
        uid = int(member["id"])
        reports = await java.get_member_reports(uid)
        if not reports:
            logger.info("成员 %s 无健康报告", member["id"])
            continue
        latest = max(reports, key=lambda r: r.get("reportDate", ""))
        report_id = latest.get("id")
        member["latest_report_id"] = report_id
        member["report_date"] = latest.get("reportDate", "")
        member["overall_score"] = latest.get("overallScore", "未知")

        indicators = await java.get_report_indicators(report_id)
        for ind in indicators:
            ind["_member_id"] = member["id"]
            ind["_member_name"] = member["name"]
            all_indicators.append(ind)

    # ====== 3. 获取指标定义+正常范围 ======
    known_indicators = {}
    for ind in all_indicators:
        name = ind.get("indicatorName", "").strip()
        if not name or name in known_indicators:
            continue
        kb = await java.query_health_knowledge("indicator", name)
        if not kb:
            kb = await java.query_health_knowledge("bacteria", name)
        if not kb:
            kb = await java.query_health_knowledge("nutrient", name)
        if kb:
            known_indicators[name] = kb

    # ====== 4. 识别异常指标 ======
    abnormal_list = []
    normal_list = []
    for ind in all_indicators:
        name = ind.get("indicatorName", "")
        raw_val = ind.get("indicatorValue", "")
        status = ind.get("status", "")
        unit = ind.get("unit", "")
        ref_range = ind.get("refRange", "")

        # 优先用知识库中的正常范围
        kb = known_indicators.get(name)
        if kb and kb.get("normalRange"):
            ref_range = kb.get("normalRange", ref_range)

        low, high = _parse_normal_range(ref_range)
        value = _try_parse_value(raw_val)
        is_abnormal = _is_abnormal(status, value, low, high)

        entry = {
            "member": ind.get("_member_name", ""),
            "indicator": name,
            "value": raw_val,
            "unit": unit,
            "ref_range": ref_range,
            "status": status,
            "is_abnormal": is_abnormal,
            "description": kb.get("description", "") if kb else "",
            "suggestion": kb.get("suggestion", "") if kb else "",
        }
        if is_abnormal:
            abnormal_list.append(entry)
        else:
            normal_list.append(entry)

    # ====== 5. 检索知识库 ======
    kb_results = []

    # 5a. 异常指标检索
    abnormal_queries = set()
    for ind in abnormal_list:
        abnormal_queries.add(ind["indicator"])
    for q in list(abnormal_queries)[:5]:
        results = await _search_knowledge(retriever, f"{q} 改善建议", k=3)
        kb_results.extend(results)

    # 5b. 用户需求检索
    goal_results = await _search_knowledge(retriever, goal, k=5)
    kb_results.extend(goal_results)

    # 5c. 维度检索
    if dimensions:
        dim_results = await _search_knowledge(retriever, dimensions, k=3)
        kb_results.extend(dim_results)

    # 去重
    seen_content = set()
    deduped_kb = []
    for r in kb_results:
        h = r.get("content", "")[:100]
        if h not in seen_content:
            seen_content.add(h)
            deduped_kb.append(r)

    # ====== 6. 组装结构化 Prompt ======
    prompt_parts = []

    # 系统提示
    prompt_parts.append(await get_prompt("health_plan_text") or TEXT_PLAN_SYSTEM_PROMPT)

    # 目标与维度
    prompt_parts.append(f"\n## 用户目标\n{goal}")
    if dimensions:
        prompt_parts.append(f"\n## 重点关注维度\n{dimensions}")

    # 成员信息
    prompt_parts.append("\n## 家庭成员")
    for m in members_info:
        scores = f"健康评分: {m.get('overall_score', '未知')}" if m.get('overall_score') else ""
        report = f"最近报告: {m.get('report_date', '无')}" if m.get('report_date') else ""
        prompt_parts.append(f"- {m['name']} (年龄: {m['age']}) {scores} {report}")

    # 异常指标
    if abnormal_list:
        prompt_parts.append("\n## 异常指标")
        for ind in abnormal_list:
            parts = [f"- {ind['member']} - {ind['indicator']}: {ind['value']}{ind['unit']} (参考范围: {ind['ref_range']})"]
            if ind['description']:
                parts.append(f"  说明: {ind['description']}")
            if ind['suggestion']:
                parts.append(f"  建议: {ind['suggestion']}")
            prompt_parts.append("\n".join(parts))

    # 正常指标
    if normal_list:
        prompt_parts.append("\n## 正常指标（参考）")
        normal_summary = [f"- {ind['indicator']}: {ind['value']}{ind['unit']} (正常)" for ind in normal_list[:10]]
        prompt_parts.extend(normal_summary)

    # 知识库参考
    if deduped_kb:
        prompt_parts.append("\n## 知识库参考（可引用）")
        for r in deduped_kb[:8]:
            title = r.get("metadata", {}).get("title", "")
            content = r.get("content", "")[:300]
            prompt_parts.append(f"---\n{title}\n{content}")

    full_prompt = "\n".join(prompt_parts)

    # ====== 7. 调用 LLM ======
    request_template = await get_prompt("health_plan_text_request") or "请基于以上数据，生成一份针对{goal}的健康改善方案。"
    try:
        request_text = request_template.format(goal=goal)
    except (KeyError, IndexError, ValueError):
        logger.warning("Java 配置的 health_plan_text_request 模板缺少占位符，回退本地模板")
        request_text = "请基于以上数据，生成一份针对{goal}的健康改善方案。".format(goal=goal)
    messages = [
        SystemMessage(content=full_prompt),
        HumanMessage(content=request_text),
    ]
    response = await llm.ainvoke(messages)
    answer = response.content

    return answer


# ===== 健康方案生成（结构化 JSON）=====

class HealthPlanRequest(BaseModel):
    member_ids: Optional[str] = None
    dimensions: Optional[str] = None
    goal: str = ""
    family_id: Optional[int] = None


class HealthPlanRegenerateRequest(BaseModel):
    section: str  # nutrition | diet | exercise
    feedback: str = ""
    existing_section_content: str = ""
    member_ids: Optional[str] = None
    dimensions: Optional[str] = None
    goal: str = ""
    family_id: Optional[int] = None


TEXT_PLAN_SYSTEM_PROMPT = """你是一个专业的家庭健康方案生成器。请根据用户提供的健康数据，生成一份结构化的健康改善方案。

输出格式要求：
## 方案概述
[简要说明方案的总体目标和适用对象]

## 成员健康概况
[每个成员的关键指标摘要]

## 需要关注的异常指标
[列出异常指标及对应的知识库建议]

## 改善方案
### 1. 饮食调整
[具体、可执行的饮食建议]

### 2. 生活习惯
[具体、可执行的生活习惯建议]

### 3. 补充建议
[如需补充营养素或益生菌，给出具体建议]

### 4. 跟踪建议
[建议定期复查的指标和频率]

## 注意事项
[禁忌、提醒等]

请基于实际数据给出建议，不要编造科学依据。引用知识库内容时标注来源。"""

PLAN_SYSTEM_PROMPT = """你是一个专业的家庭健康方案规划师。根据用户提供的健康数据和目标，生成结构化的健康改善方案。

## 输出格式（必须输出合法 JSON，不要有其他内容）

{
  "overview": "总体概述（100字以内，说明方案目标和核心策略）",
  "sections": [
    {
      "key": "nutrition",
      "title": "营养补充建议",
      "content": "Markdown 格式的详细内容",
      "items": [
        {"name": "产品名", "dosage": "用量", "timing": "服用时间", "reason": "推荐理由"}
      ],
      "tasks": [
        {
          "action_type": "buy",
          "title": "购买维生素D3补充剂",
          "dimension": "wealth",
          "frequency": "once",
          "notes": "每日一粒，随餐服用"
        }
      ]
    },
    {
      "key": "diet",
      "title": "饮食建议",
      "content": "Markdown 格式的餐饮建议",
      "items": [{"meal": "餐型", "food": "食物建议", "notes": "注意事项"}],
      "tasks": [
        {
          "action_type": "diet",
          "title": "早餐增加高蛋白与膳食纤维",
          "dimension": "body",
          "frequency": "daily",
          "notes": ""
        }
      ]
    },
    {
      "key": "exercise",
      "title": "运动计划",
      "content": "Markdown 格式的运动建议",
      "items": [{"type": "运动类型", "duration": "时长", "frequency": "频率", "notes": "注意事项"}],
      "tasks": [
        {
          "action_type": "exercise",
          "title": "每周3次有氧运动，每次30分钟",
          "dimension": "body",
          "frequency": "daily",
          "notes": ""
        }
      ]
    }
  ],
  "abnormal_indicators": [
    {"member": "姓名", "indicator": "指标名", "value": "值", "unit": "单位", "suggestion": "建议"}
  ]
}

## tasks 字段约定
- 每个 section 的 `tasks` 是该 section 中"可执行的行动项"列表，与 `content`（人类可读 Markdown）分离。
- `action_type` 取值仅限：`buy`（购买/补充产品）、`read`（阅读）、`exercise`（运动）、`checkin`（打卡/记录）、`diet`（饮食）、`activity`（活动/社交）。
- `dimension` 取值仅限五维：`body`/`mind`/`wisdom`/`action`/`wealth`。
- `frequency`：`once`=一次性任务；`daily`=每日重复任务。
- `title` 是最终写入任务系统的标题，必须是**具体可执行的动作**，不要写纯原理/机制描述。
- 若某 section 没有可执行的行动项，`tasks` 输出空数组 `[]`。

## 原则
1. 基于实际数据给出建议，不编造
2. 引用知识库内容时标注来源
3. 建议要具体可执行，避免空泛
4. 营养补充部分要具体到产品类型和用量
5. 严重健康问题建议咨询医生
6. 当问卷中某项关键信息缺失或不确定（如抗生素用药史、排便性状、过敏表现、饮食结构不明确）时，应在方案末尾明确提示用户：「为了获得更精准的饮食建议，建议您进行肠道菌群检测（GI Navigator 或其他权威检测），以获取确切的菌群组成数据」。

## 画像数据使用指南
如果提供了用户的画像数据（五维评分、身体指标、心理指标等），请结合这些真实数据给出更有针对性的建议。特别关注异常指标（如睡眠不足、压力偏高、运动频率低等），在方案中明确说明这些指标的现状和改善方向。
"""

REGENERATE_SECTION_SYSTEM_PROMPT = """你是一个家庭健康方案规划师。根据用户反馈重新生成指定部分内容。

## 输出格式（必须输出合法 JSON，不要有其他内容）
{
  "content": "重新生成的 Markdown 内容",
  "tasks": [
    {
      "action_type": "buy|read|exercise|checkin|diet|activity",
      "title": "可执行任务标题",
      "dimension": "body|mind|wisdom|action|wealth",
      "frequency": "once|daily",
      "notes": "补充说明"
    }
  ]
}

## tasks 约定
- action_type 取值：buy/read/exercise/checkin/diet/activity
- dimension 取值：body/mind/wisdom/action/wealth
- frequency：once=一次性；daily=每日重复
- 无行动项时 tasks 输出 []

## 原则
- 保持与原格式一致
- 结合用户反馈进行修改
- 建议要具体可执行"""


def _format_diet_preferences(dp: dict) -> str:
    if not dp:
        return ""
    parts = []
    def _list(v):
        if not v:
            return ""
        if isinstance(v, list):
            return ", ".join(str(x) for x in v)
        try:
            parsed = json.loads(v)
            if isinstance(parsed, list):
                return ", ".join(str(x) for x in parsed)
        except Exception:
            pass
        return str(v)
    if dp.get("allergies"):
        parts.append(f"过敏原:{_list(dp['allergies'])}")
    if dp.get("absoluteAvoid"):
        parts.append(f"忌口:{_list(dp['absoluteAvoid'])}")
    if dp.get("mustEat"):
        parts.append(f"必吃:{_list(dp['mustEat'])}")
    if dp.get("religiousDiet") and dp["religiousDiet"] not in ("none", None, ""):
        parts.append(f"宗教饮食:{dp['religiousDiet']}")
    if dp.get("spiceLevel"):
        parts.append(f"辣度:{dp['spiceLevel']}")
    if dp.get("noPreference"):
        parts.append("无特殊饮食偏好")
    if dp.get("dietStructure"):
        try:
            ds = json.loads(dp["dietStructure"])
            if isinstance(ds, dict):
                if ds.get("vegetableFruit"):
                    parts.append(f"每日蔬果量:{ds['vegetableFruit']}")
                if ds.get("dairy"):
                    parts.append(f"乳制品:{ds['dairy']}")
                if ds.get("sugaryDrinks"):
                    parts.append(f"含糖饮料:{ds['sugaryDrinks']}")
        except Exception:
            parts.append(f"饮食结构:{dp['dietStructure']}")
    return "; ".join(parts)


def _format_health_status(hs: dict) -> str:
    if not hs:
        return ""
    def _arr(v):
        if not v:
            return ""
        if isinstance(v, list):
            return ", ".join(str(x.get("name", x)) if isinstance(x, dict) else str(x) for x in v)
        try:
            parsed = json.loads(v)
            if isinstance(parsed, list):
                return ", ".join(str(x.get("name", x)) if isinstance(x, dict) else str(x) for x in parsed)
        except Exception:
            pass
        return v
    out = []
    if hs.get("heightCm"):
        out.append(f"身高{hs['heightCm']}cm")
    if hs.get("weightKg"):
        out.append(f"体重{hs['weightKg']}kg")
    if hs.get("bloodPressure"):
        out.append(f"血压{hs['bloodPressure']}")
    if hs.get("bloodGlucose"):
        out.append(f"空腹血糖{hs['bloodGlucose']}mmol/L")
    lipids = _arr(hs.get("bloodLipids"))
    if lipids:
        out.append(f"血脂:{lipids}")
    diseases = _arr(hs.get("diseaseHistory"))
    if diseases:
        out.append(f"疾病史:{diseases}")
    meds = _arr(hs.get("medications"))
    if meds:
        out.append(f"用药:{meds}")
    if hs.get("notes"):
        out.append(f"备注:{hs['notes']}")
    # 生活方式与肠道健康（心情/睡眠/排便/抗生素/过敏表现）
    if hs.get("lifestyle"):
        try:
            ls = json.loads(hs["lifestyle"])
            parts = []
            if isinstance(ls, dict):
                if ls.get("mood"):
                    parts.append(f"心情状态:{ls['mood']}")
                if ls.get("sleepDuration"):
                    parts.append(f"睡眠时长:{ls['sleepDuration']}")
                if ls.get("bedtime"):
                    parts.append(f"入睡时间:{ls['bedtime']}")
                if ls.get("bowelFrequency"):
                    parts.append(f"排便频率:{ls['bowelFrequency']}")
                if ls.get("bowelForm"):
                    parts.append(f"粪便性状:{ls['bowelForm']}")
                if ls.get("antibioticHistory"):
                    parts.append(f"近2年抗生素用药史:{ls['antibioticHistory']}")
                if ls.get("allergySymptoms") and ls["allergySymptoms"]:
                    sym = ", ".join(ls["allergySymptoms"]) if isinstance(ls["allergySymptoms"], list) else str(ls["allergySymptoms"])
                    parts.append(f"过敏表现:{sym}")
            if parts:
                out.append("生活方式:" + "; ".join(parts))
        except Exception:
            out.append(f"生活方式:{hs['lifestyle']}")
    return "; ".join(out)


async def _collect_plan_data(java: JavaClient, retriever: RagRetriever, member_ids_str: str, goal: str, dimensions: str):
    """统一数据收集逻辑"""
    member_ids = [m.strip() for m in member_ids_str.split(",") if m.strip()]

    # 1. 家庭成员信息
    members_info = []
    for uid_str in member_ids:
        ctx = await java.get_family_context(int(uid_str), "child_info")
        children = ctx.get("children", []) if isinstance(ctx, dict) else []
        for child in children:
            cid = str(child.get("用户ID", ""))
            if cid == uid_str:
                members_info.append({
                    "id": cid,
                    "name": child.get("姓名", f"成员{cid}"),
                    "age": child.get("年龄", "未知"),
                })
                break
        if not any(m["id"] == uid_str for m in members_info):
            members_info.append({"id": uid_str, "name": f"成员{uid_str}", "age": "未知"})

    # 2. 健康指标
    all_indicators = []
    abnormal_list = []
    for member in members_info:
        reports = await java.get_member_reports(int(member["id"]))
        if not reports:
            continue
        latest = max(reports, key=lambda r: r.get("reportDate", ""))
        indicators = await java.get_report_indicators(latest.get("id"))
        for ind in indicators:
            ind["_member_name"] = member["name"]
            all_indicators.append(ind)

    # 2.5 健康自检问卷（health_status；未填写为空 dict）
    for member in members_info:
        try:
            member["health_status"] = await java.get_member_health_status(int(member["id"]))
        except Exception:
            member["health_status"] = {}
        try:
            member["diet_preferences"] = await java.get_member_diet_preferences(int(member["id"]))
        except Exception:
            member["diet_preferences"] = {}

    # 3. 知识库
    kb_results = []
    queries = set()
    for ind in all_indicators:
        queries.add(ind.get("indicatorName", ""))
    queries.add(goal)
    if dimensions:
        queries.add(dimensions)
    for q in list(queries)[:8]:
        if q:
            results = await retriever.retrieve(q, k=3)
            kb_results.extend(results)

    return members_info, all_indicators, kb_results


# 从 LLM 原始输出中提取 tasks 的正则兜底：在 content 文本里找形如
# "1. 动作（动词/名词）..." 的行。以可识别的动作词开头视为潜在任务。
_TASK_FALLBACK_RE = re.compile(r"^\s*(?:\d+[\.、)]|\-\s*)\s*"
                               r"(?:(?:购买|购置|阅读|看|运动|锻炼|跑步|散步|打卡|记录|饮食|吃|少|多|活动|参加|亲子).*)$")


def _parse_plan_response(answer: str) -> dict:
    """解析 LLM 原始输出为 HealthPlanResponse；非法 JSON 或校验失败时用正则兜底提取 tasks。

    返回 dict：{"success": bool, "data": {...}, "error": str|None}
    """
    start = answer.find("{")
    end = answer.rfind("}") + 1
    if start >= 0 and end > start:
        try:
            parsed = json.loads(answer[start:end])
            resp = HealthPlanResponse.model_validate(parsed)
            return {"success": True, "data": resp.model_dump(), "error": None}
        except Exception as e:
            # 校验失败：尝试正则兜底
            fallback = _fallback_extract_tasks(parsed if isinstance(parsed, dict) else {})
            if fallback is not None:
                return {"success": True, "data": fallback, "error": str(e)}
            return {"success": False, "data": {"raw": answer, "overview": answer[:200]},
                    "error": str(e)}
    return {"success": False, "data": {"raw": answer, "overview": answer[:200]},
            "error": "no JSON found"}


def _fallback_extract_tasks(parsed: dict) -> Optional[dict]:
    """当 LLM 输出缺 tasks 或校验失败时，从各 section.content 用正则提取任务并回填。
    任一 section 无 tasks 才触发；全部已含 tasks 则返回 None（表示无需兜底）。"""
    if not isinstance(parsed, dict):
        return None
    sections = parsed.get("sections")
    if not isinstance(sections, list) or not sections:
        return None
    changed = False
    for sec in sections:
        if not isinstance(sec, dict):
            continue
        tasks = sec.get("tasks")
        if isinstance(tasks, list) and tasks:
            # 检查 tasks 是否全部有效；如有无效项则视为缺失，触发 fallback
            try:
                for t in tasks:
                    PlanTask.model_validate(t)
                continue  # 全部有效，跳过
            except Exception:
                pass  # 有无效项，继续执行 fallback
        content = sec.get("content", "")
        extracted = []
        for raw in content.split("\n"):
            line = raw.strip()
            if not line:
                continue
            m = _TASK_FALLBACK_RE.match(line)
            if not m:
                continue
            # 去掉行首编号/项目符号
            title = re.sub(r"^\s*(?:\d+[\.、)]|\-\s*)\s*", "", line).strip()
            if not title:
                continue
            action = _classify_action(title)
            if action is None:
                continue
            extracted.append({
                "action_type": action["action_type"],
                "title": title,
                "dimension": action["dimension"],
                "frequency": action["frequency"],
                "notes": "",
            })
        if extracted:
            sec["tasks"] = extracted
            changed = True
    if changed:
        return parsed
    return None


def _classify_action(title: str) -> Optional[dict]:
    """按动作词分类，映射到 action_type + 五维维度 + 频率（与 Java classifyTaskLine 对齐）。"""
    if re.search(r"购买|购置|采购|下单|买入|囤|选购", title):
        return {"action_type": "buy", "dimension": "wealth", "frequency": "once"}
    if re.search(r"阅读|看|读书", title):
        return {"action_type": "read", "dimension": "wisdom", "frequency": "daily"}
    if re.search(r"运动|锻炼|跑步|散步|健身|瑜伽|拉伸", title):
        return {"action_type": "exercise", "dimension": "body", "frequency": "daily"}
    if re.search(r"打卡|记录|复盘|记", title):
        return {"action_type": "checkin", "dimension": "mind", "frequency": "daily"}
    if re.search(r"饮食|吃|少|多|餐|营养|水", title):
        return {"action_type": "diet", "dimension": "body", "frequency": "daily"}
    if re.search(r"活动|参加|亲子|社交|户外|游戏", title):
        return {"action_type": "activity", "dimension": "action", "frequency": "daily"}
    return None


@router.post("/health/plan/generate", response_model=dict)
async def health_plan_generate(req: HealthPlanRequest):
    """健康方案生成 — 返回结构化 JSON（总览+营养+饮食+运动）"""
    goal = req.goal or "改善健康状况"
    java = JavaClient()
    retriever = RagRetriever(collection_name="cfc_knowledge")
    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=0.3,
    )

    members_info, all_indicators, kb_results = await _collect_plan_data(java, retriever, req.member_ids or "", goal, req.dimensions or "")

    # 获取每个成员的画像数据
    for member in members_info:
        try:
            profile = await java.get_member_profile(int(member["id"]))
            member["profile"] = profile
        except Exception:
            member["profile"] = {}

    # 构建 prompt
    parts = [await get_prompt("health_plan") or PLAN_SYSTEM_PROMPT]
    parts.append(f"\n## 用户目标\n{goal}")
    if req.dimensions:
        parts.append(f"\n## 重点关注维度\n{req.dimensions}")
    parts.append("\n## 家庭成员")
    for m in members_info:
        profile = m.get("profile", {})
        dims = profile.get("dimension_scores", {})
        body = profile.get("body_metrics", {})
        mind = profile.get("mind_metrics", {})
        parts.append(f"- {m['name']} (年龄: {m['age']})")
        hs_text = _format_health_status(m.get("health_status"))
        if hs_text:
            parts.append(f"  健康自检: {hs_text}")
        dp_text = _format_diet_preferences(m.get("diet_preferences"))
        if dp_text:
            parts.append(f"  饮食偏好: {dp_text}")
        if dims:
            parts.append(f"  五维评分: 身{dims.get('body','?')} 智{dims.get('wisdom','?')} 心{dims.get('mind','?')} 行{dims.get('action','?')} 富{dims.get('wealth','?')}")
        if body.get('sleep_dur_avg'):
            parts.append(f"  平均睡眠: {body['sleep_dur_avg']}小时/天")
        if mind.get('stress_avg'):
            parts.append(f"  平均压力: {mind['stress_avg']}/10")
        if body.get('exercise_count_week'):
            parts.append(f"  周运动: {body['exercise_count_week']}次")

    if all_indicators:
        parts.append("\n## 健康指标摘要")
        for ind in all_indicators[:15]:
            status = ind.get("status", "")
            if status in ("abnormal", "high", "low", "偏高", "偏低"):
                parts.append(f"- 【异常】{ind.get('_member_name','')} - {ind.get('indicatorName','')}: {ind.get('indicatorValue','')} {ind.get('unit','')} (状态: {status})")
            else:
                parts.append(f"- {ind.get('_member_name','')} - {ind.get('indicatorName','')}: {ind.get('indicatorValue','')} {ind.get('unit','')}")

    if kb_results:
        parts.append("\n## 知识库参考")
        for r in kb_results[:6]:
            title = r.get("metadata", {}).get("title", "")
            content = r.get("content", "")[:200]
            parts.append(f"---\n{title}\n{content}")

    full_prompt = "\n".join(parts)

    messages = [SystemMessage(content=full_prompt)]
    try:
        response = await llm.ainvoke(messages)
        answer = response.content
        result = _parse_plan_response(answer)
        if result["success"]:
            return {"success": True, "data": result["data"], "parse_error": result["error"]}
        return {"success": True, "data": result["data"], "parse_error": result["error"]}
    except Exception as e:
        logger.error("方案生成失败: %s", e)
        return {"success": False, "error": str(e)}


@router.post("/health/plan/regenerate-section", response_model=dict)
async def health_plan_regenerate(req: HealthPlanRegenerateRequest):
    """重新生成方案的某一个 section"""
    java = JavaClient()
    llm = ChatOpenAI(
        model=settings.llm_model,
        api_key=settings.llm_api_key,
        base_url=settings.llm_base_url,
        temperature=0.3,
    )

    members_info, all_indicators, kb_results = await _collect_plan_data(java, None, req.member_ids or "", req.goal, req.dimensions or "")

    # 构建上下文
    ctx_parts = [f"目标: {req.goal}"]
    for m in members_info:
        ctx_parts.append(f"- {m['name']} (年龄: {m['age']})")
        hs_text = _format_health_status(m.get("health_status"))
        if hs_text:
            ctx_parts.append(f"  健康自检: {hs_text}")
        dp_text = _format_diet_preferences(m.get("diet_preferences"))
        if dp_text:
            ctx_parts.append(f"  饮食偏好: {dp_text}")
    for ind in all_indicators[:10]:
        if ind.get("status") in ("abnormal", "high", "low", "偏高", "偏低"):
            ctx_parts.append(f"- 【异常】{ind.get('_member_name','')} - {ind.get('indicatorName','')}: {ind.get('indicatorValue','')}")

    prompt = await get_prompt("health_plan_regenerate") or REGENERATE_SECTION_SYSTEM_PROMPT
    prompt += f"\n\n## 当前 {req.section} 内容\n{req.existing_section_content[:500]}"
    prompt += f"\n\n## 用户反馈\n{req.feedback}"
    prompt += f"\n\n## 相关背景\n" + "\n".join(ctx_parts[:10])

    try:
        response = await llm.ainvoke([SystemMessage(content=prompt)])
        answer = response.content
        start = answer.find("{")
        end = answer.rfind("}") + 1
        content = answer
        tasks = []
        if start >= 0 and end > start:
            try:
                parsed = json.loads(answer[start:end])
                content = parsed.get("content") or answer
                raw_tasks = parsed.get("tasks") or []
                # 用 PlanTask 校验，非法条目丢弃
                for t in raw_tasks:
                    try:
                        pt = PlanTask.model_validate(t)
                        tasks.append(pt.model_dump())
                    except Exception:
                        continue
            except Exception as e:
                logger.warning("解析重生成 section JSON 失败: %s", e)
        return {"success": True, "content": content, "tasks": tasks}
    except Exception as e:
        logger.error("重新生成 section 失败: %s", e)
        return {"success": False, "error": str(e)}


@router.post("/nutrition/send", response_model=DifyResponse)
async def nutrition_send(req: DifyChatRequest):
    """AI 营养助手 — 基于健康报告的个性化营养建议"""
    query = _extract_query(req)
    if not query:
        raise HTTPException(status_code=400, detail="query 为空")

    graph = create_nutrition_graph()
    initial_state = {
        "query": query,
        "user_id": int(req.user_id) if str(req.user_id).isdigit() else 0,
        "child_id": int(req.inputs.get("child_id")) if req.inputs.get("child_id") else None,
        "conversation_id": req.conversation_id or None,
        "context": _to_langgraph_context(req),
        "answer": None,
        "sources": [],
        "tasks": [],
        "messages": None,
    }
    config = {
        "configurable": {"thread_id": req.conversation_id or f"nutrition_{req.user_id}"},
    }
    result = await graph.ainvoke(initial_state, config)
    answer = result.get("answer") or ""

    sources = result.get("sources") or []
    metadata = {}
    if sources:
        metadata["sources"] = [
            {"title": s.get("name", ""), "type": s.get("type", "tool")}
            for s in sources
        ]

    return DifyResponse(
        id=f"nutrition-{uuid.uuid4().hex[:24]}",
        created=_now_ts(),
        model="langgraph-nutrition",
        choices=[
            DifyChoice(
                index=0,
                message={"role": "assistant", "content": answer},
                finish_reason="stop",
            )
        ],
        usage=DifyUsage(
            prompt_tokens=len(query.split()),
            completion_tokens=len(answer.split()),
            total_tokens=len(query.split()) + len(answer.split()),
        ),
        metadata=metadata,
    )