chendandan_homework/app/services/candidate_selection_service.py

from __future__ import annotations

"""候选筛选服务：对航班/酒店/往返组合做确定性规则筛选与加权评分排序，不调用大模型或外部接口。"""

import hashlib
from datetime import datetime, time
from typing import Any, Iterable

from app.schemas.selection import (
    RankedFlightCandidate,
    RankedHotelCandidate,
    RoundTripOption,
)
from app.schemas.travel_request import (
    TravelRequest,
)


class CandidateSelectionService:
    """使用确定性规则筛选和排序旅行候选。

    该服务不调用大模型，也不调用外部接口。
    """

    def rank_flights(
        self,
        flights: list[dict[str, Any]],
        request: TravelRequest,
        *,
        limit: int = 10,
        reverse_direction: bool = False,
        apply_time_preferences: bool = True,
    ) -> tuple[
        list[RankedFlightCandidate],
        list[str],
    ]:
        """对单程方向的航班候选进行排序。"""

        valid_flights = [
            flight
            for flight in flights
            if isinstance(flight, dict)
            and flight.get("option_id")
        ]

        warnings: list[str] = []

        if not valid_flights:
            return [], ["没有有效航班候选。"]

        filtered_flights = self._filter_flights(
            valid_flights,
            request,
        )

        if not filtered_flights:
            filtered_flights = valid_flights

            warnings.append(
                "没有航班完全满足中转限制，"
                "已回退到全部候选并进行排序。"
            )

        prices = [
            price
            for flight in filtered_flights
            if (
                price := self._to_float(
                    flight.get("price")
                )
            )
            is not None
        ]

        durations = [
            duration
            for flight in filtered_flights
            if (
                duration := self._to_float(
                    flight.get(
                        "duration_minutes"
                    )
                )
            )
            is not None
        ]

        priority = (
            request.flight_preferences.priority
        )

        weights = self._flight_weights(priority)

        ranked: list[RankedFlightCandidate] = []

        for flight in filtered_flights:
            price = self._to_float(
                flight.get("price")
            )

            duration = self._to_float(
                flight.get("duration_minutes")
            )

            stop_count = self._to_int(
                flight.get("stop_count")
            )

            price_score = self._inverse_score(
                price,
                prices,
            )

            duration_score = self._inverse_score(
                duration,
                durations,
            )

            stop_score = self._stop_score(
                stop_count
            )

            airport_score = (
                self._airport_match_score(
                    flight,
                    request,
                    reverse_direction=(
                        reverse_direction
                    ),
                )
            )

            time_score = 100.0

            candidate_warnings: list[str] = []

            if apply_time_preferences:
                (
                    time_score,
                    time_warnings,
                ) = self._time_match_score(
                    flight,
                    request,
                )

                candidate_warnings.extend(
                    time_warnings
                )

            score = (
                price_score * weights["price"]
                + duration_score
                * weights["duration"]
                + stop_score
                * weights["stops"]
                + airport_score
                * weights["airport"]
                + time_score
                * weights["time"]
            )

            reasons = self._flight_reasons(
                flight=flight,
                price_score=price_score,
                duration_score=duration_score,
                airport_score=airport_score,
                time_score=time_score,
                apply_time_preferences=(
                    apply_time_preferences
                ),
            )

            ranked.append(
                RankedFlightCandidate(
                    option_id=str(
                        flight["option_id"]
                    ),
                    score=round(score, 2),
                    reasons=reasons,
                    warnings=candidate_warnings,
                    flight=flight,
                )
            )

        ranked.sort(
            key=lambda candidate: (
                -candidate.score,
                self._to_float(
                    candidate.flight.get(
                        "price"
                    )
                )
                or float("inf"),
            )
        )

        return ranked[:limit], warnings

    def rank_hotels(
        self,
        hotels: list[dict[str, Any]],
        request: TravelRequest,
        *,
        limit: int = 10,
    ) -> tuple[
        list[RankedHotelCandidate],
        list[str],
    ]:
        """根据酒店预算、评分和客观属性排序。"""

        valid_hotels = [
            hotel
            for hotel in hotels
            if isinstance(hotel, dict)
            and hotel.get("hotel_id")
        ]

        warnings: list[str] = []

        if not valid_hotels:
            return [], ["没有有效酒店候选。"]

        filtered_hotels = self._filter_hotels(
            valid_hotels,
            request,
        )

        if not filtered_hotels:
            filtered_hotels = valid_hotels

            warnings.append(
                "没有酒店同时满足全部预算、"
                "评分或星级限制，"
                "已回退到全部候选。"
            )

        prices = [
            price
            for hotel in filtered_hotels
            if (
                price := self._to_float(
                    hotel.get("price_per_night")
                )
            )
            is not None
        ]

        ranked: list[RankedHotelCandidate] = []

        hotel_preferences = (
            request.hotel_preferences
        )

        for hotel in filtered_hotels:
            price = self._to_float(
                hotel.get("price_per_night")
            )

            rating = self._to_float(
                hotel.get("overall_rating")
            )

            location_rating = self._to_float(
                hotel.get("location_rating")
            )

            price_score = self._inverse_score(
                price,
                prices,
            )

            rating_score = self._rating_score(
                rating
            )

            location_score = self._rating_score(
                location_rating
            )

            cancellation_bonus = (
                5.0
                if hotel.get(
                    "free_cancellation"
                )
                is True
                else 0.0
            )

            score = (
                price_score * 0.45
                + rating_score * 0.35
                + location_score * 0.20
                + cancellation_bonus
            )

            score = min(score, 100.0)

            reasons: list[str] = []
            candidate_warnings: list[str] = []

            if price_score >= 70:
                reasons.append(
                    "每晚价格在候选中相对较低"
                )

            if rating is not None and rating >= 4.5:
                reasons.append(
                    f"用户评分较高（{rating:.1f}）"
                )

            if (
                location_rating is not None
                and location_rating >= 4.0
            ):
                reasons.append(
                    "位置评分较高"
                )

            if hotel.get("free_cancellation"):
                reasons.append(
                    "支持免费取消"
                )

            if hotel.get("deal"):
                reasons.append(
                    "接口返回优惠信息"
                )

            requested_amenities = {
                item.strip().lower()
                for item in hotel_preferences.amenities
                if item.strip()
            }

            available_amenities = {
                str(item).strip().lower()
                for item in hotel.get(
                    "amenities",
                    []
                )
                if str(item).strip()
            }

            if requested_amenities:
                matched = (
                    requested_amenities
                    & available_amenities
                )

                if matched:
                    reasons.append(
                        "匹配部分指定设施"
                    )

                missing = (
                    requested_amenities
                    - available_amenities
                )

                if missing:
                    candidate_warnings.append(
                        "未确认全部指定设施："
                        + "、".join(sorted(missing))
                    )

            if not reasons:
                reasons.append(
                    "综合价格、评分和位置排序"
                )

            ranked.append(
                RankedHotelCandidate(
                    hotel_id=str(
                        hotel["hotel_id"]
                    ),
                    score=round(score, 2),
                    reasons=reasons,
                    warnings=candidate_warnings,
                    hotel=hotel,
                )
            )

        ranked.sort(
            key=lambda candidate: (
                -candidate.score,
                self._to_float(
                    candidate.hotel.get(
                        "price_per_night"
                    )
                )
                or float("inf"),
            )
        )

        if hotel_preferences.near_subway:
            warnings.append(
                "酒店靠近地铁的要求尚未完成"
                "真实路线校验，将在高德路线阶段处理。"
            )

        if hotel_preferences.preferred_areas:
            warnings.append(
                "酒店商圈偏好将在地图距离阶段"
                "进一步验证。"
            )

        return ranked[:limit], warnings

    def rank_round_trip_combinations(
        self,
        combinations: list[
            dict[str, dict[str, Any]]
        ],
        request: TravelRequest,
        *,
        limit: int = 10,
    ) -> list[RoundTripOption]:
        """对完整往返组合进行统一排序。"""

        valid_combinations = [
            combination
            for combination in combinations
            if isinstance(
                combination.get("outbound"),
                dict,
            )
            and isinstance(
                combination.get(
                    "return_flight"
                ),
                dict,
            )
        ]

        prices: list[float] = []
        durations: list[float] = []

        for combination in valid_combinations:
            outbound = combination["outbound"]
            return_flight = combination[
                "return_flight"
            ]

            # 返程查询中的price通常代表选择该去程后
            # 得到的完整接口报价，因此不与去程相加。
            price = self._to_float(
                return_flight.get("price")
            )

            if price is None:
                price = self._to_float(
                    outbound.get("price")
                )

            if price is not None:
                prices.append(price)

            total_duration = (
                (
                    self._to_float(
                        outbound.get(
                            "duration_minutes"
                        )
                    )
                    or 0.0
                )
                + (
                    self._to_float(
                        return_flight.get(
                            "duration_minutes"
                        )
                    )
                    or 0.0
                )
            )

            durations.append(total_duration)

        weights = self._flight_weights(
            request.flight_preferences.priority
        )

        results: list[RoundTripOption] = []

        seen_ids: set[str] = set()

        for combination in valid_combinations:
            outbound = combination["outbound"]
            return_flight = combination[
                "return_flight"
            ]

            outbound_id = str(
                outbound.get("option_id", "")
            )

            return_id = str(
                return_flight.get(
                    "option_id",
                    "",
                )
            )

            combination_id = self._stable_id(
                outbound_id,
                return_id,
            )

            if combination_id in seen_ids:
                continue

            seen_ids.add(combination_id)

            quoted_price = self._to_float(
                return_flight.get("price")
            )

            if quoted_price is None:
                quoted_price = self._to_float(
                    outbound.get("price")
                )

            outbound_duration = (
                self._to_float(
                    outbound.get(
                        "duration_minutes"
                    )
                )
                or 0.0
            )

            return_duration = (
                self._to_float(
                    return_flight.get(
                        "duration_minutes"
                    )
                )
                or 0.0
            )

            total_duration = (
                outbound_duration
                + return_duration
            )

            outbound_stops = (
                self._to_int(
                    outbound.get("stop_count")
                )
                or 0
            )

            return_stops = (
                self._to_int(
                    return_flight.get(
                        "stop_count"
                    )
                )
                or 0
            )

            total_stops = (
                outbound_stops
                + return_stops
            )

            price_score = self._inverse_score(
                quoted_price,
                prices,
            )

            duration_score = (
                self._inverse_score(
                    total_duration,
                    durations,
                )
            )

            stop_score = max(
                0.0,
                100.0 - total_stops * 25.0,
            )

            outbound_airport_score = (
                self._airport_match_score(
                    outbound,
                    request,
                    reverse_direction=False,
                )
            )

            return_airport_score = (
                self._airport_match_score(
                    return_flight,
                    request,
                    reverse_direction=True,
                )
            )

            airport_score = (
                outbound_airport_score
                + return_airport_score
            ) / 2.0

            # 往返组合中不再次使用去程时间偏好
            # 评价返程，因为当前模型没有独立返程时间字段。
            score = (
                price_score * weights["price"]
                + duration_score
                * weights["duration"]
                + stop_score
                * weights["stops"]
                + airport_score
                * (
                    weights["airport"]
                    + weights["time"]
                )
            )

            reasons: list[str] = []

            if price_score >= 70:
                reasons.append(
                    "往返接口报价在组合中相对较低"
                )

            if total_stops == 0:
                reasons.append(
                    "去程和返程均为直飞"
                )

            elif total_stops <= 2:
                reasons.append(
                    "往返中转次数较少"
                )

            if duration_score >= 70:
                reasons.append(
                    "往返总飞行时长相对较短"
                )

            if airport_score >= 90:
                reasons.append(
                    "匹配用户机场偏好"
                )

            if not reasons:
                reasons.append(
                    "综合价格、时长和中转次数排序"
                )

            currency = str(
                return_flight.get("currency")
                or outbound.get("currency")
                or request.currency
            )

            results.append(
                RoundTripOption(
                    combination_id=combination_id,
                    score=round(score, 2),
                    quoted_price=quoted_price,
                    currency=currency,
                    outbound=outbound,
                    return_flight=return_flight,
                    reasons=reasons,
                    warnings=[
                        "组合价格采用返程查询返回的"
                        "接口报价，未将两个方向报价相加。"
                    ],
                )
            )

        results.sort(
            key=lambda option: (
                -option.score,
                option.quoted_price
                if option.quoted_price is not None
                else float("inf"),
            )
        )

        return results[:limit]

    def _filter_flights(
        self,
        flights: list[dict[str, Any]],
        request: TravelRequest,
    ) -> list[dict[str, Any]]:
        """按中转次数上限过滤航班列表。"""
        max_stops = (
            request.flight_preferences.max_stops
        )

        if max_stops is None:
            return flights

        return [
            flight
            for flight in flights
            if (
                self._to_int(
                    flight.get("stop_count")
                )
                is not None
                and self._to_int(
                    flight.get("stop_count")
                )
                <= max_stops
            )
        ]

    def _filter_hotels(
        self,
        hotels: list[dict[str, Any]],
        request: TravelRequest,
    ) -> list[dict[str, Any]]:
        """按价格上限、最低评分、星级过滤酒店列表。"""
        preferences = request.hotel_preferences

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

        for hotel in hotels:
            price = self._to_float(
                hotel.get("price_per_night")
            )

            rating = self._to_float(
                hotel.get("overall_rating")
            )

            hotel_class = self._to_float(
                hotel.get("hotel_class")
            )

            if (
                preferences.max_price_per_night
                is not None
                and (
                    price is None
                    or price
                    > preferences.max_price_per_night
                )
            ):
                continue

            if (
                preferences.minimum_rating
                is not None
                and (
                    rating is None
                    or rating
                    < preferences.minimum_rating
                )
            ):
                continue

            if preferences.hotel_classes:
                allowed_classes = {
                    float(value)
                    for value
                    in preferences.hotel_classes
                }

                if (
                    hotel_class is None
                    or hotel_class
                    not in allowed_classes
                ):
                    continue

            result.append(hotel)

        return result

    def _flight_weights(
        self,
        priority: str,
    ) -> dict[str, float]:
        """根据用户偏好优先级返回各评分维度的权重字典。"""
        if priority == "price":
            return {
                "price": 0.60,
                "duration": 0.15,
                "stops": 0.15,
                "airport": 0.05,
                "time": 0.05,
            }

        if priority == "convenience":
            return {
                "price": 0.15,
                "duration": 0.30,
                "stops": 0.25,
                "airport": 0.15,
                "time": 0.15,
            }

        return {
            "price": 0.35,
            "duration": 0.25,
            "stops": 0.20,
            "airport": 0.10,
            "time": 0.10,
        }

    def _flight_reasons(
        self,
        *,
        flight: dict[str, Any],
        price_score: float,
        duration_score: float,
        airport_score: float,
        time_score: float,
        apply_time_preferences: bool,
    ) -> list[str]:
        """根据各维度得分生成中文排序理由片段列表。"""
        reasons: list[str] = []

        stop_count = (
            self._to_int(
                flight.get("stop_count")
            )
            or 0
        )

        if stop_count == 0:
            reasons.append("直飞")

        elif stop_count == 1:
            reasons.append("仅中转一次")

        if price_score >= 70:
            reasons.append(
                "报价在候选中相对较低"
            )

        if duration_score >= 70:
            reasons.append(
                "飞行时长相对较短"
            )

        if airport_score >= 90:
            reasons.append(
                "匹配用户机场偏好"
            )

        if (
            apply_time_preferences
            and time_score >= 90
        ):
            reasons.append(
                "起降时间符合用户偏好"
            )

        if not reasons:
            reasons.append(
                "综合价格、时长和中转次数排序"
            )

        return reasons

    def _airport_match_score(
        self,
        flight: dict[str, Any],
        request: TravelRequest,
        *,
        reverse_direction: bool,
    ) -> float:
        """计算机场偏好匹配得分：命中偏好机场满分，否则低分。"""
        preferences = request.flight_preferences

        if reverse_direction:
            preferred_departure = (
                preferences
                .preferred_arrival_airports
            )

            preferred_arrival = (
                preferences
                .preferred_departure_airports
            )
        else:
            preferred_departure = (
                preferences
                .preferred_departure_airports
            )

            preferred_arrival = (
                preferences
                .preferred_arrival_airports
            )

        scores: list[float] = []

        departure_airport = str(
            flight.get("departure_airport", "")
        ).upper()

        arrival_airport = str(
            flight.get("arrival_airport", "")
        ).upper()

        if preferred_departure:
            scores.append(
                100.0
                if departure_airport
                in preferred_departure
                else 20.0
            )

        if preferred_arrival:
            scores.append(
                100.0
                if arrival_airport
                in preferred_arrival
                else 20.0
            )

        if not scores:
            return 100.0

        return sum(scores) / len(scores)

    def _time_match_score(
        self,
        flight: dict[str, Any],
        request: TravelRequest,
    ) -> tuple[float, list[str]]:
        """计算出发/到达时间窗匹配得分，返回得分与警告列表。"""
        preferences = request.flight_preferences

        departure_datetime = (
            self._parse_datetime(
                flight.get("departure_time")
            )
        )

        arrival_datetime = (
            self._parse_datetime(
                flight.get("arrival_time")
            )
        )

        scores: list[float] = []
        warnings: list[str] = []

        if (
            preferences.earliest_departure_time
            is not None
            or preferences.latest_departure_time
            is not None
        ):
            if departure_datetime is None:
                scores.append(0.0)
                warnings.append(
                    "无法解析航班起飞时间。"
                )

            elif self._within_time_window(
                departure_datetime.time(),
                preferences
                .earliest_departure_time,
                preferences
                .latest_departure_time,
            ):
                scores.append(100.0)

            else:
                scores.append(20.0)
                warnings.append(
                    "起飞时间不在用户偏好范围内。"
                )

        if (
            preferences.earliest_arrival_time
            is not None
            or preferences.latest_arrival_time
            is not None
        ):
            if arrival_datetime is None:
                scores.append(0.0)
                warnings.append(
                    "无法解析航班到达时间。"
                )

            elif self._within_time_window(
                arrival_datetime.time(),
                preferences
                .earliest_arrival_time,
                preferences
                .latest_arrival_time,
            ):
                scores.append(100.0)

            else:
                scores.append(20.0)
                warnings.append(
                    "到达时间不在用户偏好范围内。"
                )

        if not scores:
            return 100.0, warnings

        return sum(scores) / len(scores), warnings

    @staticmethod
    def _within_time_window(
        value: time,
        earliest: time | None,
        latest: time | None,
    ) -> bool:
        """判断给定时间是否落在可选的起止时间窗口内。"""
        if earliest is not None and value < earliest:
            return False

        if latest is not None and value > latest:
            return False

        return True

    @staticmethod
    def _stop_score(
        stop_count: int | None,
    ) -> float:
        """经停次数得分：0经停=100分，每多1次减40分。"""
        if stop_count is None:
            return 0.0

        return max(
            0.0,
            100.0 - stop_count * 40.0,
        )

    @staticmethod
    def _rating_score(
        value: float | None,
    ) -> float:
        """将5分制评分/星级线性映射为百分制得分。"""
        if value is None:
            return 0.0

        return max(
            0.0,
            min(100.0, value / 5.0 * 100.0),
        )

    @staticmethod
    def _inverse_score(
        value: float | None,
        values: Iterable[float],
    ) -> float:
        """将成本类数值反转为得分（值越小得分越高），在候选中做min-max归一化。"""
        if value is None:
            return 0.0

        valid_values = list(values)

        if not valid_values:
            return 0.0

        minimum = min(valid_values)
        maximum = max(valid_values)

        if maximum == minimum:
            return 100.0

        score = (
            maximum - value
        ) / (
            maximum - minimum
        ) * 100.0

        return max(0.0, min(100.0, score))

    @staticmethod
    def _parse_datetime(
        value: object,
    ) -> datetime | None:
        """安全解析ISO格式时间字符串，解析失败返回None。"""
        if value is None:
            return None

        try:
            return datetime.fromisoformat(
                str(value)
            )
        except ValueError:
            return None

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

        if isinstance(value, bool):
            return None

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

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

        if isinstance(value, bool):
            return None

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

    @staticmethod
    def _stable_id(
        *parts: str,
    ) -> str:
        """基于不定个字符串片段生成稳定的SHA256派生ID。"""
        raw_value = "|".join(parts)

        digest = hashlib.sha256(
            raw_value.encode("utf-8")
        ).hexdigest()

        return f"round-{digest[:16]}"