chendandan_homework/app/services/travel_search_service.py

from __future__ import annotations
"""旅行搜索服务：封装SerpApiClient，把原始搜索响应规范化为Pydantic结果模型，提供统一的航班/酒店/机场搜索入口。"""
import hashlib
from datetime import datetime
from typing import Any
from app.clients.serpapi_client import SerpApiClient
from app.schemas.common import SourceMetadata
from app.schemas.flight import (
    CarbonEmissions,
    FlightOption,
    FlightSearchQuery,
    FlightSearchResult,
    FlightSegment,
    FlightSourceGroup,
    Layover,
)

from app.schemas.hotel import (
    GeoPoint,
    HotelOption,
    HotelPriceOffer,
    HotelSearchQuery,
    HotelSearchResult,
)

class TravelSearchService:
    """
    旅行资源查询服务
    职责：
    1.调用SerpApiClient 获取原始数据
    2.合并不同结果分组
    3.标准化字段
    4.去除完成重复的候选
    5.返回项目内部统一数据模型
    不负责：
    1.根据用户偏好筛选
    2.对候选进行个性化排序
    3.生成推荐理由
    4.阶段候选数量。
    """

    def __init__(self,client:SerpApiClient)->None:
        """初始化搜索服务，传入配置好的SerpApiClient实例。"""
        self._client = client

    async def search_flights(
        self,
        query: FlightSearchQuery,
    ) -> FlightSearchResult:
        """查询航班（单程或去程）。

        单程搜索返回完整单程候选。
        往返搜索首先返回去程候选。
        """
        raw_data = await self._client.search_flights_raw(query)

        return self.normalize_flights_response(
            query=query,
            raw_data=raw_data,
        )

    async def search_return_flights(
        self,
        query: FlightSearchQuery,
        departure_token: str,
    ) -> FlightSearchResult:
        """查询与选定去程匹配的返程候选。"""

        raw_data = (
            await self._client.search_return_flights_raw(
                query=query,
                departure_token=departure_token,
            )
        )

        return self.normalize_flights_response(
            query=query,
            raw_data=raw_data,
        )
    
    async def search_hotels(self, query: HotelSearchQuery) -> HotelSearchResult:
        """按查询条件搜索酒店并对结果做规范化处理。"""
        raw_data = await self._client.search_hotels_raw(query)

        return self.normalize_hotels_response(query=query, raw_data=raw_data)
    
    @classmethod
    def normalize_flights_response(cls,query:FlightSearchQuery,raw_data:dict[str,Any])->FlightSearchResult:
        """
        把SerpApi航班原始响应转换成统一结构。
        best_flights和other_flights会全部进入标准化流程。
        这里只去除完全重复的候选，不限制返回数量
        """
        normalized_flights:list[FlightOption]=[]
        seen_option_ids:set[str]=set()
        warnings:list[str]=[]

        groups: tuple[
            tuple[FlightSourceGroup,list[dict[str,Any]]],
            ...,
        ] = (
            (
                "best_flights",
                raw_data.get("best_flights",[]),
            ),
            (
                "other_flights",
                raw_data.get("other_flights",[]),
            )
        )

        for source_group, raw_options in groups:
            if not isinstance(raw_options,list):
                warnings.append(f"{source_group}不是列表，已忽略。")
                continue

            for provider_rank,raw_option in enumerate(
                raw_options,
                start=1
            ):
                if not isinstance(raw_option,dict):
                    warnings.append(f"{source_group}第{provider_rank}条。不是有效对象，已忽略")
                    continue

                try:
                    option = cls._normalize_flight_option(
                        raw_option = raw_option,
                        source_group = source_group,
                        provider_rank = provider_rank,
                        currency = query.currency,
                        adults = query.adults,
                    )
                except (KeyError,TypeError,ValueError) as e:
                    warnings.append(f"{source_group}第{provider_rank}条。航班解析失败：{e}")
                    continue

                if option.option_id in seen_option_ids:
                    continue

                seen_option_ids.add(option.option_id)
                normalized_flights.append(option)

        metadata = raw_data.get("search_metadata",{})

        warnings.append(
            "price 字段已按单人（每成人）票价标准化，原始 SerpApi 总价已除以 adults 人数"
        )

        return FlightSearchResult(
            query=query,
            flights=normalized_flights,
            total_count=len(normalized_flights),
            source=SourceMetadata(
                provider="serpapi_google_flights",
                search_id=metadata.get("id"),
                status=metadata.get("status","Success"),
                currency=query.currency,
                is_cached=None
            ),
            price_insights=raw_data.get("price_insights"),
            warnings=warnings
        )
    
    @classmethod
    def _normalize_flight_option(
        cls,
        raw_option:dict[str,Any],
        source_group:FlightSourceGroup,
        provider_rank:int,
        currency:str,
        adults:int = 1,
        )->FlightOption:
        """把单条航班原始dict规范化为Pydantic FlightOption。"""
        raw_segments = raw_option.get("flights")

        if not isinstance(raw_segments,list) or not raw_segments:
            raise ValueError("缺少flights航段信息。")
        
        segments = [
            cls._normalize_flight_segment(raw_segment)
            for raw_segment in raw_segments
        ]

        raw_layovers = raw_option.get("layovers", [])

        layovers = [
            cls._normalize_layover(raw_layover)
            for raw_layover in raw_layovers
            if isinstance(raw_layover, dict)
        ]

        first_segment = segments[0]
        last_segment = segments[-1]

        total_duration = cls._to_int(
            raw_option.get("total_duration")
        )

        if total_duration is None:
            total_duration = sum(
                segment.duration_minutes or 0
                for segment in segments
            ) + sum(
                layover.duration_minutes or 0
                for layover in layovers
            )

        airlines = cls._unique_strings(
            segment.airline
            for segment in segments
            if segment.airline
        )

        price = cls._normalize_price_per_adult(
            raw_option.get("price"),
            adults=adults,
        )

        option_id = cls._build_flight_option_id(
            segments=segments,
            price=price,
        )

        raw_carbon = raw_option.get("carbon_emissions")

        carbon_emissions = None

        if isinstance(raw_carbon, dict):
            carbon_emissions = CarbonEmissions(
                this_flight_grams=cls._to_int(
                    raw_carbon.get("this_flight")
                ),
                typical_for_route_grams=cls._to_int(
                    raw_carbon.get(
                        "typical_for_this_route"
                    )
                ),
                difference_percent=cls._to_int(
                    raw_carbon.get(
                        "difference_percent"
                    )
                ),
            )

        return FlightOption(
            option_id=option_id,
            source_group=source_group,
            provider_rank=provider_rank,
            segments=segments,
            layovers=layovers,
            departure_airport_code=(
                first_segment.departure_airport_code
            ),
            final_arrival_airport_code=(
                last_segment.arrival_airport_code
            ),
            departure_time=first_segment.departure_time,
            arrival_time=last_segment.arrival_time,
            total_duration_minutes=total_duration,
            stop_count=max(len(segments) - 1, 0),
            price=price,
            currency=currency,
            flight_type=raw_option.get("type"),
            airlines=airlines,
            is_overnight=(
                any(segment.overnight for segment in segments)
                or any(layover.overnight for layover in layovers)
            ),
            carbon_emissions=carbon_emissions,
            departure_token=raw_option.get(
                "departure_token"
            ),
            booking_token=raw_option.get(
                "booking_token"
            ),
        )

    @classmethod
    def _normalize_flight_segment(
        cls,
        raw_segment: dict[str, Any],
    ) -> FlightSegment:
        """规范化单个航段的起降信息。"""
        if not isinstance(raw_segment, dict):
            raise TypeError("航段不是有效对象。")

        departure = raw_segment.get(
            "departure_airport",
            {},
        )
        arrival = raw_segment.get(
            "arrival_airport",
            {},
        )

        departure_code = str(
            departure.get("id", "")
        ).strip().upper()

        arrival_code = str(
            arrival.get("id", "")
        ).strip().upper()

        if not departure_code or not arrival_code:
            raise ValueError("航段缺少机场代码。")

        departure_time = cls._parse_local_datetime(
            departure.get("time")
        )
        arrival_time = cls._parse_local_datetime(
            arrival.get("time")
        )

        return FlightSegment(
            flight_number=raw_segment.get(
                "flight_number"
            ),
            airline=raw_segment.get("airline"),
            departure_airport_code=departure_code,
            departure_airport_name=departure.get("name"),
            departure_time=departure_time,
            arrival_airport_code=arrival_code,
            arrival_airport_name=arrival.get("name"),
            arrival_time=arrival_time,
            duration_minutes=cls._to_int(
                raw_segment.get("duration")
            ),
            airplane=raw_segment.get("airplane"),
            travel_class=raw_segment.get(
                "travel_class"
            ),
            legroom=raw_segment.get("legroom"),
            overnight=bool(
                raw_segment.get("overnight", False)
            ),
            often_delayed_by_over_30_min=bool(
                raw_segment.get(
                    "often_delayed_by_over_30_min",
                    False,
                )
            ),
            plane_and_crew_by=raw_segment.get(
                "plane_and_crew_by"
            ),
            extensions=cls._string_list(
                raw_segment.get("extensions")
            ),
            ticket_also_sold_by=cls._string_list(
                raw_segment.get(
                    "ticket_also_sold_by"
                )
            ),
        )

    @classmethod
    def _normalize_layover(
        cls,
        raw_layover: dict[str, Any],
    ) -> Layover:
        """规范化中转停留信息。"""
        airport_code = str(
            raw_layover.get("id", "")
        ).strip().upper()

        if not airport_code:
            raise ValueError("经停信息缺少机场代码。")

        return Layover(
            airport_code=airport_code,
            airport_name=raw_layover.get("name"),
            duration_minutes=cls._to_int(
                raw_layover.get("duration")
            ),
            overnight=bool(
                raw_layover.get("overnight", False)
            ),
        )

    @classmethod
    def normalize_hotels_response(
        cls,
        query: HotelSearchQuery,
        raw_data: dict[str, Any],
    ) -> HotelSearchResult:
        """把酒店原始响应转换为统一结构。"""

        raw_properties = raw_data.get("properties", [])

        normalized_hotels: list[HotelOption] = []
        seen_hotel_ids: set[str] = set()
        warnings: list[str] = []

        if not isinstance(raw_properties, list):
            raw_properties = []
            warnings.append(
                "properties 不是列表，已按空结果处理。"
            )

        for provider_rank, raw_hotel in enumerate(
            raw_properties,
            start=1,
        ):
            if not isinstance(raw_hotel, dict):
                warnings.append(
                    f"第 {provider_rank} 条酒店不是有效对象，"
                    "已忽略。"
                )
                continue

            try:
                hotel = cls._normalize_hotel_option(
                    raw_hotel=raw_hotel,
                    provider_rank=provider_rank,
                    currency=query.currency,
                )
            except (KeyError, TypeError, ValueError) as exc:
                warnings.append(
                    f"第 {provider_rank} 条酒店解析失败："
                    f"{exc}"
                )
                continue

            if hotel.hotel_id in seen_hotel_ids:
                continue

            seen_hotel_ids.add(hotel.hotel_id)
            normalized_hotels.append(hotel)

        metadata = raw_data.get("search_metadata", {})
        pagination = raw_data.get(
            "serpapi_pagination",
            {},
        )

        return HotelSearchResult(
            query=query,
            hotels=normalized_hotels,
            total_count=len(normalized_hotels),
            source=SourceMetadata(
                provider="serpapi_google_hotels",
                search_id=metadata.get("id"),
                status=metadata.get("status", "Success"),
                currency=query.currency,
                is_cached=None,
            ),
            next_page_token=(
                pagination.get("next_page_token")
                or raw_data.get("next_page_token")
            ),
            warnings=warnings,
        )

    @classmethod
    def _normalize_hotel_option(
        cls,
        raw_hotel: dict[str, Any],
        provider_rank: int,
        currency: str,
    ) -> HotelOption:
        """把单条酒店原始dict规范化为Pydantic HotelOption。"""
        name = str(raw_hotel.get("name", "")).strip()

        if not name:
            raise ValueError("酒店名称为空。")

        raw_coordinates = raw_hotel.get(
            "gps_coordinates",
            {},
        )

        coordinates = None

        if isinstance(raw_coordinates, dict):
            latitude = cls._to_float(
                raw_coordinates.get("latitude")
            )
            longitude = cls._to_float(
                raw_coordinates.get("longitude")
            )

            if latitude is not None and longitude is not None:
                coordinates = GeoPoint(
                    latitude=latitude,
                    longitude=longitude,
                )

        rate_per_night = raw_hotel.get(
            "rate_per_night",
            {},
        )
        total_rate = raw_hotel.get(
            "total_rate",
            {},
        )

        price_per_night = None
        total_price = None

        if isinstance(rate_per_night, dict):
            price_per_night = cls._to_float(
                rate_per_night.get(
                    "extracted_lowest"
                )
            )

        if isinstance(total_rate, dict):
            total_price = cls._to_float(
                total_rate.get(
                    "extracted_lowest"
                )
            )

        price_offers = cls._normalize_hotel_prices(
            raw_hotel.get("prices", [])
        )

        property_token = raw_hotel.get(
            "property_token"
        )

        hotel_id = (
            str(property_token)
            if property_token
            else cls._stable_id(
                prefix="hotel",
                values=[
                    name,
                    coordinates.latitude
                    if coordinates
                    else "",
                    coordinates.longitude
                    if coordinates
                    else "",
                ],
            )
        )

        images = raw_hotel.get("images", [])
        thumbnail = raw_hotel.get("thumbnail")

        if (
            thumbnail is None
            and isinstance(images, list)
            and images
            and isinstance(images[0], dict)
        ):
            thumbnail = images[0].get("thumbnail")

        free_cancellation_raw = raw_hotel.get(
            "free_cancellation"
        )

        free_cancellation = (
            free_cancellation_raw
            if isinstance(
                free_cancellation_raw,
                bool,
            )
            else None
        )

        return HotelOption(
            hotel_id=hotel_id,
            provider_rank=provider_rank,
            name=name,
            property_type=raw_hotel.get("type"),
            description=raw_hotel.get("description"),
            coordinates=coordinates,
            check_in_time=raw_hotel.get(
                "check_in_time"
            ),
            check_out_time=raw_hotel.get(
                "check_out_time"
            ),
            hotel_class=cls._to_int(
                raw_hotel.get(
                    "extracted_hotel_class"
                )
            ),
            overall_rating=cls._to_float(
                raw_hotel.get("overall_rating")
            ),
            review_count=cls._to_int(
                raw_hotel.get("reviews")
            ),
            location_rating=cls._to_float(
                raw_hotel.get("location_rating")
            ),
            price_per_night=price_per_night,
            total_price=total_price,
            currency=currency,
            amenities=cls._string_list(
                raw_hotel.get("amenities")
            ),
            excluded_amenities=cls._string_list(
                raw_hotel.get(
                    "excluded_amenities"
                )
            ),
            free_cancellation=free_cancellation,
            sponsored=bool(
                raw_hotel.get("sponsored", False)
            ),
            eco_certified=bool(
                raw_hotel.get(
                    "eco_certified",
                    False,
                )
            ),
            property_token=property_token,
            thumbnail_url=thumbnail,
            price_offers=price_offers,
            deal=raw_hotel.get("deal"),
            deal_description=raw_hotel.get(
                "deal_description"
            ),
        )

    @classmethod
    def _normalize_hotel_prices(
        cls,
        raw_prices: Any,
    ) -> list[HotelPriceOffer]:
        """规范化酒店价格列表。"""
        if not isinstance(raw_prices, list):
            return []

        offers: list[HotelPriceOffer] = []

        for raw_price in raw_prices:
            if not isinstance(raw_price, dict):
                continue

            raw_nightly = raw_price.get(
                "rate_per_night",
                {},
            )
            raw_total = raw_price.get(
                "total_rate",
                {},
            )

            nightly_price = None
            total_price = None

            if isinstance(raw_nightly, dict):
                nightly_price = cls._to_float(
                    raw_nightly.get(
                        "extracted_lowest"
                    )
                )

            if isinstance(raw_total, dict):
                total_price = cls._to_float(
                    raw_total.get(
                        "extracted_lowest"
                    )
                )

            free_cancellation_raw = raw_price.get(
                "free_cancellation"
            )

            offers.append(
                HotelPriceOffer(
                    source=raw_price.get("source"),
                    price_per_night=nightly_price,
                    total_price=total_price,
                    free_cancellation=(
                        free_cancellation_raw
                        if isinstance(
                            free_cancellation_raw,
                            bool,
                        )
                        else None
                    ),
                )
            )

        return offers

    @staticmethod
    def _build_flight_option_id(
        segments: list[FlightSegment],
        price: int | None,
    ) -> str:
        """根据关键字段生成稳定的航班选项ID。"""
        values: list[Any] = [price]

        for segment in segments:
            values.extend(
                [
                    segment.flight_number,
                    segment.departure_airport_code,
                    segment.departure_time.isoformat(),
                    segment.arrival_airport_code,
                    segment.arrival_time.isoformat(),
                ]
            )

        return TravelSearchService._stable_id(
            prefix="flight",
            values=values,
        )

    @staticmethod
    def _stable_id(
        prefix: str,
        values: list[Any],
    ) -> str:
        """生成SHA256派生的稳定字符串ID。"""
        raw_value = "|".join(
            str(value)
            for value in values
        )

        digest = hashlib.sha256(
            raw_value.encode("utf-8")
        ).hexdigest()[:20]

        return f"{prefix}_{digest}"

    @staticmethod
    def _parse_local_datetime(
        value: Any,
    ) -> datetime:
        """解析第三方接口返回的机场当地时间。

        SerpApi 返回值没有明确时区偏移，
        因此这里保留为无时区 datetime。
        """

        if not isinstance(value, str) or not value.strip():
            raise ValueError("航班时间为空。")

        try:
            return datetime.fromisoformat(value.strip())
        except ValueError as exc:
            raise ValueError(
                f"无法解析航班时间：{value}"
            ) from exc

    @staticmethod
    def _to_int(value: Any) -> int | None:
        """安全转换为int，不可转换返回None。"""
        if value is None or isinstance(value, bool):
            return None

        try:
            return int(value)
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _normalize_price_per_adult(
        raw_price: Any,
        adults: int,
    ) -> int | None:
        """将 SerpApi 原始总价除以成人数，返回单人票价。

        SerpApi / Google Flights 返回的 price 是所选乘客的
        订单总价（含所有成人），这里统一折算为每成人单价，
        避免前端误解为「一个人」或「全部人」的价格。
        """
        if raw_price is None or isinstance(raw_price, bool):
            return None

        if adults < 1:
            adults = 1

        try:
            total = float(raw_price)
        except (TypeError, ValueError):
            return None

        return round(total / adults)

    @staticmethod
    def _to_float(value: Any) -> float | None:
        """安全转换为float，不可转换返回None。"""
        if value is None or isinstance(value, bool):
            return None

        try:
            return float(value)
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _string_list(value: Any) -> list[str]:
        """把字符串或列表统一转换为字符串列表。"""
        if not isinstance(value, list):
            return []

        return [
            str(item).strip()
            for item in value
            if str(item).strip()
        ]

    @staticmethod
    def _unique_strings(
        values: Any,
    ) -> list[str]:
        """字符串列表去重并保持顺序。"""
        result: list[str] = []

        for value in values:
            text = str(value).strip()

            if text and text not in result:
                result.append(text)

        return result
    

    async def search_airports(
        self,
        query: str,
    ) -> dict[str, Any]:
        """查询并标准化城市对应的机场信息。"""

        raw_data = await self._client.search_airports_raw(
            query=query,
        )

        raw_suggestions = raw_data.get(
            "suggestions",
            [],
        )

        suggestions: list[dict[str, Any]] = []

        if not isinstance(raw_suggestions, list):
            raw_suggestions = []

        for raw_suggestion in raw_suggestions:
            if not isinstance(raw_suggestion, dict):
                continue

            raw_airports = raw_suggestion.get(
                "airports",
                [],
            )

            airports: list[dict[str, Any]] = []

            if isinstance(raw_airports, list):
                for raw_airport in raw_airports:
                    if not isinstance(
                        raw_airport,
                        dict,
                    ):
                        continue

                    airport_code = str(
                        raw_airport.get("id", "")
                    ).strip().upper()

                    if not airport_code:
                        continue

                    airports.append(
                        {
                            "code": airport_code,
                            "name": raw_airport.get(
                                "name"
                            ),
                            "city": raw_airport.get(
                                "city"
                            ),
                            "city_id": raw_airport.get(
                                "city_id"
                            ),
                            "distance": raw_airport.get(
                                "distance"
                            ),
                        }
                    )

            suggestions.append(
                {
                    "position": raw_suggestion.get(
                        "position"
                    ),
                    "name": raw_suggestion.get("name"),
                    "type": raw_suggestion.get("type"),
                    "description": (
                        raw_suggestion.get(
                            "description"
                        )
                    ),
                    "location_id": raw_suggestion.get(
                        "id"
                    ),
                    "airports": airports,
                }
            )

        metadata = raw_data.get(
            "search_metadata",
            {},
        )

        return {
            "query": query,
            "total_count": len(suggestions),
            "suggestions": suggestions,
            "source": {
                "provider": (
                    "serpapi_google_flights_autocomplete"
                ),
                "search_id": metadata.get("id"),
                "status": metadata.get(
                    "status",
                    "Success",
                ),
            },
        }