""" 这个 Client 只负责： Pydantic查询参数 → SerpApi HTTP参数 → 发送请求 → 返回原始JSON """ from __future__ import annotations from typing import Any import httpx from app.schemas.flight import FlightSearchQuery from app.schemas.hotel import HotelSearchQuery SERPAPI_BASE_URL = "https://serpapi.com/search.json" FLIGHT_TYPE_MAP = { "round_trip":1, "one_way":2, } TRAVEL_CLASS_MAP = { "economy":1, "business":3, "first":4, "premium_economy":2, } FLIGHT_SORT_MAP = { "top":1, "price":2, "departure_time":3, "arrival_time":4, "duration":5, "emissions":6, } STOPS_MAP = { "any":0, "nonstop":1, "one_or_fewer":2, "two_or_fewer":3, } HOTEL_SORT_MAP = { "lowest_price":3, "highest_rating":8, "most_reviewed":13, } HOTEL_RATING_MAP = { "3.5+":7, "4.0+":8, "4.5+":9, } class SerpApiError(RuntimeError): """SerApi调用错误。""" class SerpApiClient: """SerpApi异步HTTP客户端""" def __init__(self,api_key:str,timeout_seconds:int | float=30.0)->None: """初始化SerpApiClient，验证API Key并提供可选的自定义base_url。""" if not api_key.strip(): raise ValueError("SerpApi API Key不能为空。") self.api_key = api_key self.timeout_seconds = timeout_seconds self._client:httpx.AsyncClient | None = None async def __aenter__(self)->"SerpApiClient": """异步上下文管理器协议，支持async with用法。""" self._ensure_client() return self async def __aexit__(self, exc_type:object, exc_value:object, traceback:object)->None: """异步上下文管理器协议，支持async with用法。""" await self.aclose() def _ensure_client(self)->httpx.AsyncClient: """惰性创建并缓存httpx.AsyncClient实例。""" if self._client is None: timeout = httpx.Timeout( timeout=self.timeout_seconds, connect=min(10.0,self.timeout_seconds), ) self._client=httpx.AsyncClient( timeout=timeout, follow_redirects=True ) return self._client async def aclose(self)->None: """关闭底层httpx连接，释放资源。""" if self._client is not None: await self._client.aclose() self._client = None async def _request(self,params:dict[str,Any],)->dict[str,Any]: """ 执行SerpApi请求并统一处理错误。 错误消息中不会打印完整请求URL， 避免API Key通过查询参数泄漏。 """ client = self._ensure_client() request_params = { **params, "api_key":self.api_key, } try: response = await client.get( SERPAPI_BASE_URL, params=request_params ) except httpx.TimeoutException as e: raise SerpApiError( f"SerpApi请求超时，超时时间为{self.timeout_seconds}秒" ) from e except httpx.RequestError as e: raise SerpApiError( f"SerpApi网络请求失败：{type(e).__name__}" ) from e if response.status_code >= 400: error_text = response.text[:500] raise SerpApiError(f"SerpApi返回HTTP错误：{response.status_code},响应内容：{error_text}") try: data:dict[str, Any] = response.json() except ValueError as e: raise SerpApiError("SerpApi返回内容不是合法JSON。") from e if error :=data.get("error"): raise SerpApiError(f"SerpApi查询失败：{error}") metadata = data.get("search_metadata",{}) status = metadata.get("status") if status is not None and str(status).lower() != "success": raise SerpApiError(f"SerpApi返回异常状态：{status}") return data def _build_flight_params( self, query: FlightSearchQuery, ) -> dict[str, Any]: """把内部航班查询模型转换为 SerpApi 参数。""" params: dict[str, Any] = { "engine": "google_flights", "departure_id": ",".join( query.departure_airports ), "arrival_id": ",".join( query.arrival_airports ), "outbound_date": query.outbound_date.isoformat(), "type": FLIGHT_TYPE_MAP[query.flight_type], "adults": query.adults, "children": query.children, "travel_class": TRAVEL_CLASS_MAP[ query.travel_class ], "currency": query.currency, "hl": query.language, "gl": query.country, } if query.return_date is not None: params["return_date"] = ( query.return_date.isoformat() ) if query.sort_by is not None: params["sort_by"] = FLIGHT_SORT_MAP[ query.sort_by ] if query.stops is not None: params["stops"] = STOPS_MAP[query.stops] if query.max_price is not None: params["max_price"] = query.max_price if query.outbound_times is not None: params["outbound_times"] = ( query.outbound_times ) if query.show_hidden: params["show_hidden"] = True if query.deep_search: params["deep_search"] = True if query.no_cache: params["no_cache"] = True return params async def search_flights_raw( self, query: FlightSearchQuery, ) -> dict[str, Any]: """查询首次航班结果。 单程搜索返回完整单程候选。 往返搜索首先返回去程候选。 """ params = self._build_flight_params(query) return await self._request(params) async def search_return_flights_raw( self, query: FlightSearchQuery, departure_token: str, ) -> dict[str, Any]: """根据选定去程查询对应的返程航班。""" if query.flight_type != "round_trip": raise ValueError( "只有 round_trip 查询才能获取返程航班。" ) token = departure_token.strip() if not token: raise ValueError( "departure_token 不能为空。" ) params = self._build_flight_params(query) params["departure_token"] = token return await self._request(params) async def search_hotels_raw(self, query: HotelSearchQuery) -> dict[str, Any]: """查询酒店并返回SerpApi原始JSON。""" params:dict[str,Any] = { "engine": "google_hotels", "q": query.query, "check_in_date": query.check_in_date.isoformat(), "check_out_date": query.check_out_date.isoformat(), "adults": query.adults, "children": len(query.children_ages), "currency": query.currency, "hl": query.language, "gl": query.country, } if query.children_ages: params["children_ages"] = ",".join( str(age) for age in query.children_ages ) if query.min_price is not None: params["min_price"] = query.min_price if query.max_price is not None: params["max_price"] = query.max_price if query.rating_filter is not None: params["rating"] = HOTEL_RATING_MAP[query.rating_filter] if query.hotel_class: params["hotel_class"] = ",".join( str(hotel_class) for hotel_class in query.hotel_class ) if query.free_cancellation is not None: params["free_cancellation"] = query.free_cancellation if query.sort is not None: params["sort"] = HOTEL_SORT_MAP[query.sort] if query.no_cache: params["no_cache"] = True return await self._request(params) async def search_airports_raw( self, query: str, language: str = "zh-cn", ) -> dict[str, Any]: """根据城市或机场关键词查询机场信息。""" keyword = query.strip() if not keyword: raise ValueError("机场搜索关键词不能为空。") params: dict[str, Any] = { "engine": "google_flights_autocomplete", "q": keyword, "hl": language, } return await self._request(params)