chendandan_homework/app/services/route_evaluation_service.py

from __future__ import annotations

"""路线评估服务：对酒店按距离/地铁/评分做加权排序，使用高德距离或球面余弦公式计算路线衔接。不调用大模型。"""

import math
from statistics import mean
from typing import Iterable

from app.schemas.route import (
    HotelRouteEvaluation,
    RouteLeg,
)
from app.schemas.selection import (
    RankedHotelCandidate,
)


class RouteEvaluationService:
    """根据真实距离重新评价酒店候选。

    本服务不调用大模型和外部接口。
    """

    def rank_hotels(
        self,
        hotels: list[RankedHotelCandidate],
        legs_by_hotel: dict[
            str,
            list[RouteLeg],
        ],
        subway_distance_by_hotel: dict[
            str,
            float | None,
        ],
        *,
        require_near_subway: bool,
        limit: int = 3,
    ) -> list[HotelRouteEvaluation]:
        """综合原酒店得分、景点距离和地铁距离。"""

        average_distances: list[float] = []

        average_distance_by_hotel: dict[
            str,
            float | None,
        ] = {}

        for hotel_candidate in hotels:
            hotel_id = hotel_candidate.hotel_id

            legs = legs_by_hotel.get(
                hotel_id,
                [],
            )

            distances = [
                leg.distance_meters
                for leg in legs
            ]

            average_distance = (
                mean(distances)
                if distances
                else None
            )

            average_distance_by_hotel[
                hotel_id
            ] = average_distance

            if average_distance is not None:
                average_distances.append(
                    average_distance
                )

        results: list[HotelRouteEvaluation] = []

        for hotel_candidate in hotels:
            hotel_id = hotel_candidate.hotel_id
            hotel = hotel_candidate.hotel

            legs = legs_by_hotel.get(
                hotel_id,
                [],
            )

            distances = [
                leg.distance_meters
                for leg in legs
            ]

            durations = [
                leg.duration_seconds
                for leg in legs
                if leg.duration_seconds is not None
            ]

            average_distance = (
                mean(distances)
                if distances
                else None
            )

            maximum_distance = (
                max(distances)
                if distances
                else None
            )

            average_duration = (
                mean(durations)
                if durations
                else None
            )

            centrality_score = self._inverse_score(
                average_distance,
                average_distances,
            )

            subway_distance = (
                subway_distance_by_hotel.get(
                    hotel_id
                )
            )

            subway_score = self._subway_score(
                subway_distance
            )

            base_score = hotel_candidate.score

            if require_near_subway:
                final_score = (
                    base_score * 0.45
                    + centrality_score * 0.40
                    + subway_score * 0.15
                )
            else:
                final_score = (
                    base_score * 0.55
                    + centrality_score * 0.45
                )

            reasons = list(
                hotel_candidate.reasons
            )

            warnings = list(
                hotel_candidate.warnings
            )

            if (
                average_distance is not None
                and centrality_score >= 70
            ):
                reasons.append(
                    "到主要景点的平均距离较短"
                )

            if (
                average_duration is not None
                and centrality_score >= 70
            ):
                reasons.append(
                    "前往主要景点的预计耗时较低"
                )

            if require_near_subway:
                if (
                    subway_distance is not None
                    and subway_distance <= 800
                ):
                    reasons.append(
                        "周边800米内查询到地铁站"
                    )

                elif subway_distance is None:
                    warnings.append(
                        "未获得可解析的地铁站距离"
                    )

                else:
                    warnings.append(
                        "距离最近地铁站超过"
                        f"{subway_distance:.0f}米"
                    )

            if not legs:
                warnings.append(
                    "酒店缺少有效路线距离结果"
                )

            hotel_name = str(
                hotel.get("name")
                or hotel_candidate.hotel_id
            )

            results.append(
                HotelRouteEvaluation(
                    hotel_id=hotel_id,
                    hotel_name=hotel_name,
                    final_score=round(
                        min(final_score, 100.0),
                        2,
                    ),
                    base_hotel_score=(
                        hotel_candidate.score
                    ),
                    centrality_score=round(
                        centrality_score,
                        2,
                    ),
                    subway_score=round(
                        subway_score,
                        2,
                    ),
                    average_distance_meters=(
                        round(average_distance, 2)
                        if average_distance
                        is not None
                        else None
                    ),
                    maximum_distance_meters=(
                        round(maximum_distance, 2)
                        if maximum_distance
                        is not None
                        else None
                    ),
                    average_duration_seconds=(
                        round(average_duration, 2)
                        if average_duration
                        is not None
                        else None
                    ),
                    nearest_subway_distance_meters=(
                        round(subway_distance, 2)
                        if subway_distance
                        is not None
                        else None
                    ),
                    route_legs=legs,
                    reasons=self._unique(reasons),
                    warnings=self._unique(
                        warnings
                    ),
                    hotel=hotel,
                )
            )

        results.sort(
            key=lambda item: (
                -item.final_score,
                (
                    item.average_distance_meters
                    if item.average_distance_meters
                    is not None
                    else float("inf")
                ),
            )
        )

        return results[:limit]

    def build_haversine_leg(
        self,
        *,
        origin_name: str,
        origin_location: str,
        destination_name: str,
        destination_location: str,
    ) -> RouteLeg:
        """高德距离查询失败时的直线距离降级。"""

        distance = self.haversine_distance_meters(
            origin_location,
            destination_location,
        )

        return RouteLeg(
            origin_name=origin_name,
            origin_location=origin_location,
            destination_name=destination_name,
            destination_location=(
                destination_location
            ),
            distance_meters=distance,
            duration_seconds=None,
            source="haversine_fallback",
        )

    @classmethod
    def haversine_distance_meters(
        cls,
        first_location: str,
        second_location: str,
    ) -> float:
        """根据两组经纬度计算球面直线距离。"""

        first_lon, first_lat = (
            cls.parse_location(first_location)
        )

        second_lon, second_lat = (
            cls.parse_location(second_location)
        )

        earth_radius = 6_371_000.0

        lat1 = math.radians(first_lat)
        lat2 = math.radians(second_lat)

        delta_lat = math.radians(
            second_lat - first_lat
        )

        delta_lon = math.radians(
            second_lon - first_lon
        )

        value = (
            math.sin(delta_lat / 2) ** 2
            + math.cos(lat1)
            * math.cos(lat2)
            * math.sin(delta_lon / 2) ** 2
        )

        central_angle = 2 * math.atan2(
            math.sqrt(value),
            math.sqrt(1 - value),
        )

        return round(
            earth_radius * central_angle,
            2,
        )

    @staticmethod
    def parse_location(
        location: str,
    ) -> tuple[float, float]:
        """解析高德使用的 经度,纬度 格式。"""

        parts = [
            part.strip()
            for part in location.split(",")
        ]

        if len(parts) != 2:
            raise ValueError(
                "坐标必须使用“经度,纬度”格式："
                f"{location}"
            )

        longitude = float(parts[0])
        latitude = float(parts[1])

        if not -180 <= longitude <= 180:
            raise ValueError(
                f"经度超出范围：{longitude}"
            )

        if not -90 <= latitude <= 90:
            raise ValueError(
                f"纬度超出范围：{latitude}"
            )

        return longitude, latitude

    @staticmethod
    def _inverse_score(
        value: float | None,
        values: Iterable[float],
    ) -> float:
        """将距离类数值反转为得分（越近得分越高）。"""
        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 _subway_score(
        distance_meters: float | None,
    ) -> float:
        """根据酒店到最近地铁站的步行距离计算地铁便利得分。"""
        if distance_meters is None:
            return 0.0

        if distance_meters <= 500:
            return 100.0

        if distance_meters <= 800:
            return 85.0

        if distance_meters <= 1200:
            return 60.0

        if distance_meters <= 2000:
            return 30.0

        return 10.0

    @staticmethod
    def _unique(
        values: list[str],
    ) -> list[str]:
        """按元素值去重并保持首次出现顺序。"""
        result: list[str] = []

        for value in values:
            if value and value not in result:
                result.append(value)

        return result