cfc/cfc-langgraph/app/tools/java_client.py

import httpx
from typing import Optional
from app.config import settings
import logging

logger = logging.getLogger(__name__)


class JavaClient:
    """Java 后端 HTTP 客户端 (单例, 复用连接池)"""

    _instance: Optional["JavaClient"] = None

    def __new__(cls):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._client = None
            cls._instance._base_url = settings.java_base_url
        return cls._instance

    async def _get_client(self) -> httpx.AsyncClient:
        if self._client is None:
            limits = httpx.Limits(
                max_connections=10,
                max_keepalive_connections=5,
                keepalive_expiry=30,
            )
            self._client = httpx.AsyncClient(
                base_url=self._base_url,
                timeout=httpx.Timeout(120.0, connect=3.0),
                limits=limits,
            )
        return self._client

    async def close(self):
        if self._client:
            await self._client.aclose()
            self._client = None
            JavaClient._instance = None

    async def get_published_articles(self) -> list[dict]:
        """拉取已发布知识库文章（分页 + records 解析，避免单次响应过大 / 返回 dict 结构）"""
        client = await self._get_client()
        page = 1
        size = 50
        records = []
        while True:
            resp = await client.post("/api/article/list", json={"status": "published", "page": page, "size": size})
            data = resp.json()
            if data.get("code") != 200:
                break
            body = data.get("data", {}) or {}
            items = body.get("records", []) or []
            records.extend(items)
            total = body.get("total", 0)
            if page * size >= int(total or 0) or not items:
                break
            page += 1
        return records

    async def search_products(self, keyword: str, limit: int = 5) -> list[dict]:
        client = await self._get_client()
        resp = await client.post("/api/product/search", json={
            "keyword": keyword, "status": "上架", "limit": limit
        })
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", [])
        return []

    async def search_activities(self, keyword: str, limit: int = 5) -> list[dict]:
        client = await self._get_client()
        resp = await client.post("/api/activity/search", json={
            "keyword": keyword, "status": "published", "limit": limit
        })
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", [])
        return []

    async def search_articles(self, keyword: str, limit: int = 5) -> list[dict]:
        client = await self._get_client()
        resp = await client.post("/api/article/search", json={
            "keyword": keyword, "status": "published", "limit": limit
        })
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", [])
        return []

    async def get_user_context(self, user_id: int, params: Optional[dict] = None) -> dict:
        client = await self._get_client()
        resp = await client.post(settings.effective_java_context_url, json={
            "user_id": str(user_id),
            "params": params or {},
        })
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", {})
        return {}

    @staticmethod
    def _join_metadata(value) -> str:
        """把后端返回的标签/维度数组或字符串统一为 Chroma metadata 字符串。"""
        if value is None:
            return ""
        if isinstance(value, str):
            return value
        if isinstance(value, (list, tuple, set)):
            return ",".join(str(item) for item in value if item)
        return str(value)

    @classmethod
    def _map_dan_knowledge(cls, record: dict) -> dict:
        """dan_knowledge_base 字段(camelCase) → loader/_build_chunks 所需结构。"""
        return {
            "id": record.get("id"),
            "title": record.get("title") or "",
            "content": record.get("content") or "",
            "category": "",
            "tags": cls._join_metadata(record.get("tagNames") or record.get("tags")),
            "dimensions": cls._join_metadata(
                record.get("dimensionCodes") or record.get("dimensions")),
            "source_url": record.get("sourceUrl") or "",
            "pub_date": str(record.get("createdAt") or record.get("updatedAt") or ""),
        }

    async def get_microbiome_articles(self) -> list[dict]:
        """拉取菌群知识库文章（统一数据源 dan_knowledge_base，source_type='article'）

        2026-10-20 迁移：microbiome_article 表废弃，809 篇菌群知识统一到
        dan_knowledge_base 管理。分页拉 /api/admin/knowledge-base/list，
        ORDER BY id 稳定分页，避免跨页重复。标签/维度由后端批量回填。
        """
        client = await self._get_client()
        page = 1
        size = 50
        records = []
        while True:
            resp = await client.post("/api/admin/knowledge-base/list",
                                     json={"page": page, "size": size, "status": 1, "sourceType": "article"})
            data = resp.json()
            if data.get("code") != 200:
                break
            body = data.get("data", {}) or {}
            items = body.get("records", []) or []
            records.extend(items)
            total = body.get("total", 0)
            if page * size >= int(total or 0) or not items:
                break
            page += 1
        return [
            self._map_dan_knowledge(r) for r in records if r.get("id") is not None
        ]

    async def get_dan_knowledge_base(self) -> list[dict]:
        """拉取案例报告源（dan_knowledge_base 表，status=1，排除 article 菌群文章）

        若 article 为 0 条（INSERT 前确实为 0），则该源拉 0 条，正常。
        返回结构包含 content/tags/dimensions，供 Chroma metadata 使用。
        """
        client = await self._get_client()
        page = 1
        size = 1000
        records = []
        while True:
            resp = await client.post("/api/admin/knowledge-base/list", json={"page": page, "size": size, "status": 1, "excludeSourceType": "article"})
            data = resp.json()
            if data.get("code") != 200:
                break
            body = data.get("data", {}) or {}
            items = body.get("records", []) or []
            records.extend(items)
            total = body.get("total", 0)
            if page * size >= int(total or 0) or not items:
                break
            page += 1
        return [
            self._map_dan_knowledge(r) for r in records if r.get("id") is not None
        ]

    async def get_member_reports(self, member_id: int) -> list[dict]:
        """获取某个家庭成员的健康报告列表"""
        client = await self._get_client()
        resp = await client.post("/api/health/report/list", json={"memberId": member_id})
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", [])
        return []

    async def get_report_indicators(self, report_id: int) -> list[dict]:
        """获取健康报告的所有指标明细"""
        client = await self._get_client()
        resp = await client.post("/api/health/report/indicator/list", json={"reportId": report_id})
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", [])
        return []

    async def query_health_knowledge(self, item_type: str, item_name: str) -> dict | None:
        """查询健康知识库中某个指标/菌属/营养素的定义"""
        client = await self._get_client()
        resp = await client.post("/api/health/knowledge/query", json={
            "itemType": item_type, "itemName": item_name,
        })
        data = resp.json()
        if data.get("code") == 200 and data.get("data"):
            return data["data"]
        return None

    async def batch_query_health_knowledge(self, queries: list[dict]) -> list[dict]:
        """批量查询健康知识库"""
        client = await self._get_client()
        resp = await client.post("/api/health/knowledge/batch-query", json={"queries": queries})
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", [])
        return []

    async def get_family_context(self, user_id: int, intent_type: str = "child_info", params: dict | None = None) -> dict:
        """获取家庭上下文数据（通过公开的 /api/ai/context 端点）"""
        client = await self._get_client()
        body = {"user_id": str(user_id), "intent_type": intent_type}
        if params:
            body["params"] = params
        resp = await client.post("/api/ai/context", json=body)
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", {})
        return {}

    async def get_member_profile(self, member_id: int) -> dict:
        """获取家庭成员画像快照"""
        client = await self._get_client()
        resp = await client.post("/api/profile/my", json={"memberId": member_id})
        data = resp.json()
        if data.get("code") == 200 and data.get("data"):
            return data["data"]
        return {}

    async def get_member_health_status(self, member_id: int) -> dict:
        """获取家庭成员健康自检问卷（health_status，内部端点无用户归属校验）"""
        client = await self._get_client()
        resp = await client.post("/api/health-status/internal/get", json={"memberId": member_id})
        data = resp.json()
        if data.get("code") == 200 and data.get("data"):
            return data["data"]
        return {}

    async def get_member_diet_preferences(self, member_id: int) -> dict:
        """获取家庭成员饮食偏好（diet_preferences，含饮食结构现状）"""
        client = await self._get_client()
        resp = await client.post("/api/diet/preferences/current-member", json={"memberId": member_id})
        data = resp.json()
        if data.get("code") == 200 and data.get("data"):
            return data["data"]
        return {}

    async def get_member_indicator_current(self, member_id: int) -> list[dict]:
        """获取成员指标当前值+相对上次变化（member_indicator_current，含 delta/trend）"""
        client = await self._get_client()
        resp = await client.post("/api/health/indicators/current", json={"memberId": member_id})
        data = resp.json()
        if data.get("code") == 200:
            return data.get("data", [])
        return []