chendandan_homework/app/clients/serpapi_client.py

"""
这个 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)