from typing import Any import requests from config import Config from task_service import TaskError def _headers() -> dict[str, str]: if not Config.SEARCH_QA_API_KEY: raise TaskError( "SEARCH_QA_NOT_CONFIGURED", "搜索问一问开放 API 尚未配置", 503, ) return {"X-API-Key": Config.SEARCH_QA_API_KEY} def _request(method: str, path: str, **kwargs) -> requests.Response: try: session = requests.Session() # 搜索问一问开放 API 必须直连;本机或服务环境的全局代理可能 # 指向不可用端口,且不应影响这一条受控的服务端对服务端链路。 session.trust_env = False response = session.request( method, f"{Config.SEARCH_QA_API_BASE_URL}{path}", headers={**_headers(), **kwargs.pop("headers", {})}, timeout=Config.SEARCH_QA_TIMEOUT_SECONDS, **kwargs, ) except requests.RequestException as exc: raise TaskError("SEARCH_QA_UNAVAILABLE", "搜索问一问服务暂时不可用", 502) from exc if response.status_code == 429: raise TaskError("SEARCH_QA_RATE_LIMITED", "搜索问一问接口请求过于频繁,请稍后重试", 429) if response.status_code >= 400: message = "搜索问一问服务请求失败" try: payload = response.json() message = str(payload.get("message") or payload.get("error") or message) except ValueError: pass raise TaskError("SEARCH_QA_UPSTREAM_ERROR", message, 502) return response def list_runs(task_id: str = "", xhs_account_id: str = "", limit: int = 20) -> dict[str, Any]: limit = max(1, min(int(limit), 100)) params: dict[str, Any] = {"limit": limit} if task_id: params = {"task_id": task_id} elif xhs_account_id: params["xhs_account_id"] = xhs_account_id payload = _request("GET", "/api/open/v1/search-qa/runs", params=params).json() if not payload.get("ok"): raise TaskError("SEARCH_QA_UPSTREAM_ERROR", "搜索问一问服务返回失败", 502) runs = payload.get("runs") or ([payload["run"]] if payload.get("run") else []) return {"items": runs, "total": len(runs), "query": params}