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