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 {} async def get_microbiome_articles(self) -> list[dict]: """拉取菌群知识库文章(分页,避免单次响应过大导致 Broken pipe)""" client = await self._get_client() page = 1 size = 50 records = [] while True: resp = await client.post("/api/microbiome/article/list", json={"page": page, "size": size, "keyword": "", "category": ""}) 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 get_dan_knowledge_base(self) -> list[dict]: """拉取知识库内容(dan_knowledge_base 表,仅启用 status=1)""" 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}) 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 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 {}