import httpx import asyncio from typing import List from config import settings class ChatlogHTTPError(RuntimeError): def __init__(self, status_code: int, method: str, path: str, detail: str): self.status_code = status_code self.method = method self.path = path self.detail = detail super().__init__(f"chatlog HTTP {status_code}: {method} {path} body={detail!r}") class MessageIndexNotReady(RuntimeError): """Raised when chatlog has sessions but its message time index is not usable yet.""" class ChatlogClient: def __init__(self): self.base = settings.chatlog_base_url self._contact_db_file = None async def _get(self, path: str, params: dict, timeout: float = 30.0) -> dict: try: async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client: r = await client.get(f"{self.base}{path}", params=params) r.raise_for_status() return r.json() except httpx.TimeoutException: raise RuntimeError(f"chatlog timeout: GET {path}") except httpx.HTTPStatusError as e: detail = self._response_detail(e.response) raise ChatlogHTTPError(e.response.status_code, "GET", path, detail) except Exception as e: raise RuntimeError(f"chatlog request failed: {e}") async def _post(self, path: str, body: dict, timeout: float = 30.0) -> dict: try: async with httpx.AsyncClient(timeout=timeout, trust_env=False) as client: r = await client.post(f"{self.base}{path}", json=body) r.raise_for_status() return r.json() except httpx.TimeoutException: raise RuntimeError(f"chatlog timeout: POST {path}") except httpx.HTTPStatusError as e: detail = self._response_detail(e.response) raise ChatlogHTTPError(e.response.status_code, "POST", path, detail) except Exception as e: raise RuntimeError(f"chatlog request failed: {e}") def _response_detail(self, response: httpx.Response) -> str: try: body = response.json() if isinstance(body, dict): return str(body.get("error") or body.get("detail") or body) return str(body) except Exception: return response.text async def get_messages( self, talker: str, time: str = "", sender: str = "", keyword: str = "", min_seq: int = 0, limit: int = 100, offset: int = 0, ) -> dict: params: dict = { "talker": talker, "limit": limit, "offset": offset, "format": "json", } if time: params["time"] = time else: params["time"] = "1970-01-01,2099-12-31" if sender: params["sender"] = sender if keyword: params["keyword"] = keyword if min_seq > 0: params["min_seq"] = min_seq try: data = await self._get("/api/v1/chatlog", params) except ChatlogHTTPError as e: detail = e.detail.lower() if e.status_code == 404 and "time range not found" in detail: await asyncio.sleep(0.2) try: data = await self._get("/api/v1/chatlog", params) except ChatlogHTTPError as retry_error: if ( retry_error.status_code == 404 and "time range not found" in retry_error.detail.lower() ): raise MessageIndexNotReady( "自动解密仍在处理消息库，请稍后刷新聊天记录；如果长时间为空，请在微信里打开该聊天并翻看历史消息。" ) from retry_error raise elif e.status_code == 404 and "not found" in detail: # chatlog sometimes reports a valid date window as missing while it is warming/querying. await asyncio.sleep(0.2) try: data = await self._get("/api/v1/chatlog", params) except ChatlogHTTPError as retry_error: retry_detail = retry_error.detail.lower() if ( retry_error.status_code == 404 and "time range not found" in retry_detail ): raise MessageIndexNotReady( "自动解密仍在处理消息库，请稍后刷新聊天记录；如果长时间为空，请在微信里打开该聊天并翻看历史消息。" ) from retry_error if retry_error.status_code == 404 and "not found" in retry_detail: return {"total": 0, "items": []} raise else: raise if isinstance(data, dict): return data return {"total": len(data), "items": data} async def get_message(self, talker: str, seq: int) -> dict | None: try: async with httpx.AsyncClient(timeout=10.0, trust_env=False) as client: r = await client.get( f"{self.base}/api/v1/chatlog/message", params={"talker": talker, "seq": seq}, ) if r.status_code == 404: return None r.raise_for_status() return r.json() except httpx.TimeoutException: raise RuntimeError("chatlog timeout: get_message") except Exception as e: raise RuntimeError(f"chatlog request failed: {e}") async def get_messages_batch(self, talker: str, seqs: List[int]) -> dict: return await self._post("/api/v1/chatlog/batch", {"talker": talker, "seqs": seqs}) async def get_chatrooms(self, keyword: str = "", limit: int = 100, offset: int = 0) -> dict: params: dict = {"limit": limit, "offset": offset, "format": "json"} if keyword: params["keyword"] = keyword return await self._get("/api/v1/chatroom", params, timeout=10.0) async def get_contacts(self, keyword: str = "", limit: int = 100, offset: int = 0) -> dict: params: dict = {"limit": limit, "offset": offset, "format": "json"} if keyword: params["keyword"] = keyword return await self._get("/api/v1/contact", params, timeout=10.0) async def get_chatroom_members(self, talker: str, time: str = "") -> dict: params: dict = {"talker": talker} if time: params["time"] = time return await self._get("/api/v1/chatroom/members", params) async def get_sessions(self, keyword: str = "", limit: int = 500) -> list: params: dict = {"limit": limit, "format": "json"} if keyword: params["keyword"] = keyword data = await self._get("/api/v1/session", params, timeout=15.0) if isinstance(data, list): return data return data.get("items", data.get("data", [])) async def get_avatar_url(self, wxid: str) -> str: if self._contact_db_file is None: try: db_list = await self._get("/api/v1/db", {}) self._contact_db_file = (db_list.get("contact") or [""])[0] except Exception: self._contact_db_file = "" if not self._contact_db_file: return "" safe_wxid = wxid.replace("'", "''") sql = f"SELECT small_head_url, big_head_url FROM contact WHERE username='{safe_wxid}' LIMIT 1" params = {"group": "contact", "file": self._contact_db_file, "sql": sql} try: rows = await self._get("/api/v1/db/query", params, timeout=5.0) if rows: url = rows[0].get("small_head_url") or rows[0].get("big_head_url") or "" if url: return url except Exception: pass return "" async def get_db_paths(self) -> dict: data = await self._get("/api/v1/db", {}, timeout=10.0) return data if isinstance(data, dict) else {} chatlog_client = ChatlogClient()