from __future__ import annotations """方案校验服务：用确定性规则校验行程方案的航班/酒店/活动安排/预算合理性，产出结构化问题清单与预算估算。不调用大模型。""" from datetime import datetime, timedelta from typing import Any from app.schemas.itinerary import ( ItineraryActivity, ItineraryPlan, ) from app.schemas.place import ( MapResearchResult, ) from app.schemas.route import ( RouteEvaluationResult, ) from app.schemas.selection import ( CandidateSelectionResult, ) from app.schemas.travel_request import ( TravelRequest, ) from app.schemas.validation import ( BudgetEstimate, PlanValidationResult, ValidationIssue, ) class PlanValidationService: """对Planner生成的行程进行确定性校验。 不调用大模型，不调用外部API。 """ def validate( self, *, request: TravelRequest, plan: ItineraryPlan, selection: CandidateSelectionResult, map_result: MapResearchResult, route_result: RouteEvaluationResult, ) -> PlanValidationResult: """校验完整行程方案：验证航班与酒店ID是否来自候选列表、活动地点是否存在、预算是否合理，返回校验结果与问题清单。""" issues: list[ValidationIssue] = [] flight_data = self._validate_flight( plan=plan, selection=selection, issues=issues, ) hotel_data = self._validate_hotel( request=request, plan=plan, selection=selection, route_result=route_result, issues=issues, ) self._validate_days( request=request, plan=plan, issues=issues, ) self._validate_activities( request=request, plan=plan, map_result=map_result, issues=issues, ) self._validate_boundary_days( request=request, plan=plan, flight_data=flight_data, issues=issues, ) budget = self._build_budget( request=request, plan=plan, flight_quote=flight_data.get( "price" ), hotel_estimate=hotel_data.get( "hotel_estimate" ), issues=issues, ) error_issues = [ issue for issue in issues if issue.severity == "error" ] warning_issues = [ issue for issue in issues if issue.severity == "warning" ] revision_feedback = [ issue.message for issue in error_issues ] return PlanValidationResult( is_valid=not error_issues, error_count=len(error_issues), warning_count=len(warning_issues), issues=issues, budget=budget, revision_feedback=revision_feedback, ) def _validate_flight( self, *, plan: ItineraryPlan, selection: CandidateSelectionResult, issues: list[ValidationIssue], ) -> dict[str, Any]: """确认Planner选择的航班来自真实候选。""" selected = plan.selected_flight if ( selected.flight_type != selection.flight_type ): self._add_issue( issues, code="FLIGHT_TYPE_MISMATCH", severity="error", message=( "Planner选择的航班类型与" "候选筛选结果不一致。" ), ) if selected.flight_type == "round_trip": combination = next( ( option for option in selection.round_trip_options if option.combination_id == selected.combination_id ), None, ) if combination is None: self._add_issue( issues, code=( "FLIGHT_COMBINATION_NOT_FOUND" ), severity="error", message=( "Planner选择的往返航班组合" "不存在于真实候选中：" f"{selected.combination_id}" ), ) return { "price": None, "outbound": {}, "return_flight": {}, } outbound_id = str( combination.outbound.get( "option_id", "", ) ) return_id = str( combination.return_flight.get( "option_id", "", ) ) if ( outbound_id != selected.outbound_option_id ): self._add_issue( issues, code=( "OUTBOUND_FLIGHT_ID_MISMATCH" ), severity="error", message=( "Planner填写的去程航班ID" "与所选往返组合不一致。" ), ) if ( return_id != selected.return_option_id ): self._add_issue( issues, code=( "RETURN_FLIGHT_ID_MISMATCH" ), severity="error", message=( "Planner填写的返程航班ID" "与所选往返组合不一致。" ), ) if ( selected.quoted_price is not None and combination.quoted_price is not None and not self._numbers_close( selected.quoted_price, combination.quoted_price, ) ): self._add_issue( issues, code="FLIGHT_PRICE_MISMATCH", severity="warning", message=( "Planner填写的航班报价与" "真实候选接口报价不一致，" "预算将采用候选接口报价。" ), ) return { "price": combination.quoted_price, "outbound": combination.outbound, "return_flight": ( combination.return_flight ), } candidate = next( ( item for item in selection.one_way_options if item.option_id == selected.outbound_option_id ), None, ) if candidate is None: self._add_issue( issues, code="ONE_WAY_FLIGHT_NOT_FOUND", severity="error", message=( "Planner选择的单程航班" "不存在于真实候选中。" ), ) return { "price": None, "outbound": {}, "return_flight": {}, } candidate_price = self._to_float( candidate.flight.get("price") ) return { "price": candidate_price, "outbound": candidate.flight, "return_flight": {}, } def _validate_hotel( self, *, request: TravelRequest, plan: ItineraryPlan, selection: CandidateSelectionResult, route_result: RouteEvaluationResult, issues: list[ValidationIssue], ) -> dict[str, Any]: """确认酒店真实存在，并重新计算住宿费用。""" selected = plan.selected_hotel evaluated_hotel = next( ( item for item in route_result.evaluated_hotels if item.hotel_id == selected.hotel_id ), None, ) ranked_hotel = next( ( item for item in selection.hotels if item.hotel_id == selected.hotel_id ), None, ) if ( evaluated_hotel is None and ranked_hotel is None ): self._add_issue( issues, code="HOTEL_NOT_FOUND", severity="error", message=( "Planner选择的酒店不存在于" "真实酒店候选中：" f"{selected.hotel_id}" ), ) return { "hotel_estimate": None, "hotel": {}, } source_hotel = ( evaluated_hotel.hotel if evaluated_hotel is not None else ranked_hotel.hotel ) source_name = str( source_hotel.get("name", "") ) if ( source_name and source_name != selected.name ): self._add_issue( issues, code="HOTEL_NAME_MISMATCH", severity="error", message=( "Planner填写的酒店名称与" "所选酒店ID对应名称不一致。" ), ) if ( request.departure_date is not None and selected.check_in_date != request.departure_date ): # 红眼航班（如20:50出发、次日00:05到达） # 允许入住日期延后一天。 next_day = ( request.departure_date + timedelta(days=1) ) if selected.check_in_date != next_day: self._add_issue( issues, code="HOTEL_CHECK_IN_MISMATCH", severity="error", message=( "酒店入住日期必须与" "旅行出发日期一致；" "红眼航班最多允许延后一天。" ), ) else: self._add_issue( issues, code="HOTEL_CHECK_IN_REDEYE", severity="warning", message=( "酒店入住日期比出发日期晚一天，" "疑似红眼航班。" "Validator已接受此日期，" "但请确认住宿晚数仍正确。" ), ) if ( request.return_date is not None and selected.check_out_date != request.return_date ): self._add_issue( issues, code="HOTEL_CHECK_OUT_MISMATCH", severity="error", message=( "酒店退房日期必须与" "旅行返程日期一致。" ), ) expected_nights: int | None = None if ( request.departure_date is not None and request.return_date is not None ): expected_nights = ( request.return_date - request.departure_date ).days elif request.nights is not None: expected_nights = request.nights if ( expected_nights is not None and selected.nights != expected_nights ): self._add_issue( issues, code="HOTEL_NIGHTS_MISMATCH", severity="error", message=( "酒店住宿晚数不正确：" f"应为{expected_nights}晚，" f"Planner填写为" f"{selected.nights}晚。" ), ) source_price_per_night = self._to_float( source_hotel.get( "price_per_night" ) ) if source_price_per_night is None: source_price_per_night = ( selected.price_per_night ) self._add_issue( issues, code="HOTEL_PRICE_SOURCE_MISSING", severity="warning", message=( "酒店候选缺少可解析的" "每晚价格，预算暂时采用" "Planner填写的价格。" ), ) if ( source_price_per_night is not None and selected.price_per_night is not None and not self._numbers_close( source_price_per_night, selected.price_per_night, ) ): self._add_issue( issues, code=( "HOTEL_PRICE_PER_NIGHT_MISMATCH" ), severity="warning", message=( "Planner填写的每晚酒店价格" "与候选数据不一致，预算将采用" "候选价格重新计算。" ), ) hotel_estimate: float | None = None if ( source_price_per_night is not None and expected_nights is not None ): hotel_estimate = ( source_price_per_night * expected_nights ) if ( hotel_estimate is not None and selected.estimated_total_price is not None and not self._numbers_close( hotel_estimate, selected.estimated_total_price, ) ): self._add_issue( issues, code="HOTEL_TOTAL_PRICE_MISMATCH", severity="warning", message=( "Planner填写的酒店总价估算" "不正确，已按每晚价格乘" "住宿晚数重新计算。" ), ) if ( route_result.selected_hotel_id and selected.hotel_id != route_result.selected_hotel_id ): self._add_issue( issues, code="NON_TOP_ROUTE_HOTEL", severity="warning", message=( "Planner未选择路线综合排名" "第一的酒店，需要Reviewer检查" "选择理由是否充分。" ), ) return { "hotel_estimate": hotel_estimate, "hotel": source_hotel, } def _validate_days( self, *, request: TravelRequest, plan: ItineraryPlan, issues: list[ValidationIssue], ) -> None: """检查行程日期是否完整且连续。""" if ( request.departure_date is None or request.return_date is None ): return expected_day_count = ( request.return_date - request.departure_date ).days + 1 expected_dates = [ request.departure_date + timedelta(days=index) for index in range(expected_day_count) ] actual_dates = [ day.date for day in plan.days ] actual_indexes = [ day.day_index for day in plan.days ] if len(plan.days) != expected_day_count: self._add_issue( issues, code="DAY_COUNT_MISMATCH", severity="error", message=( "行程天数不正确：" f"应生成{expected_day_count}天，" f"实际生成{len(plan.days)}天。" ), ) if actual_dates != expected_dates: missing_dates = [ item.isoformat() for item in expected_dates if item not in actual_dates ] extra_dates = [ item.isoformat() for item in actual_dates if item not in expected_dates ] details: list[str] = [] if missing_dates: details.append( "缺少日期：" + "、".join(missing_dates) ) if extra_dates: details.append( "多余日期：" + "、".join(extra_dates) ) self._add_issue( issues, code="DAY_DATE_MISMATCH", severity="error", message=( "每日行程日期没有完整覆盖" "旅行期间。" + ( " ".join(details) if details else "" ) ), ) expected_indexes = list( range(1, len(plan.days) + 1) ) if actual_indexes != expected_indexes: self._add_issue( issues, code="DAY_INDEX_MISMATCH", severity="error", message=( "每日行程的day_index必须" "从1开始连续递增。" ), ) def _validate_activities( self, *, request: TravelRequest, plan: ItineraryPlan, map_result: MapResearchResult, issues: list[ValidationIssue], ) -> None: """检查时间冲突、POI真实性和旅行节奏。""" attractions_by_id = { place.poi_id: place for place in map_result.attractions if place.poi_id } attractions_by_name = { place.name: place for place in map_result.attractions } restaurants_by_id = { place.poi_id: place for place in map_result.restaurants if place.poi_id } restaurants_by_name = { place.name: place for place in map_result.restaurants } used_attractions: set[str] = set() pace_limits = { "relaxed": 2, "normal": 3, "intensive": 4, } attraction_limit = pace_limits.get( request.pace, 3, ) for day in plan.days: sequences = [ activity.sequence for activity in day.activities ] if len(sequences) != len(set(sequences)): self._add_issue( issues, code=( "DUPLICATE_ACTIVITY_SEQUENCE" ), severity="error", message=( f"第{day.day_index}天存在" "重复的活动sequence。" ), day_index=day.day_index, ) expected_sequences = list( range(1, len(day.activities) + 1) ) if sequences != expected_sequences: self._add_issue( issues, code="ACTIVITY_SEQUENCE_GAP", severity="warning", message=( f"第{day.day_index}天的活动" "sequence没有从1连续递增。" ), day_index=day.day_index, ) ordered_activities = sorted( day.activities, key=lambda item: item.start_time, ) previous: ItineraryActivity | None = None attraction_count = 0 for activity in ordered_activities: # 航班、交通类活动允许跨午夜 # （如 20:50->次日00:05）。 if ( activity.end_time <= activity.start_time ): is_cross_midnight = ( activity.activity_type in {"flight", "transport"} ) self._add_issue( issues, code=( "INVALID_ACTIVITY_TIME" ), severity=( "warning" if is_cross_midnight else "error" ), message=( f"第{day.day_index}天活动" f"'{activity.name}'的" "结束时间早于或等于开始时间" "（跨午夜活动请确认时长正确）。" if is_cross_midnight else ( f"第{day.day_index}天活动" f"'{activity.name}'的" "结束时间必须晚于开始时间。" ) ), day_index=day.day_index, activity_sequence=( activity.sequence ), ) if ( previous is not None and activity.start_time < previous.end_time # 跨午夜活动的时间比较 # 无法反映实际先后关系，跳过。 and previous.end_time > previous.start_time and activity.end_time > activity.start_time ): self._add_issue( issues, code="ACTIVITY_OVERLAP", severity="error", message=( f"第{day.day_index}天活动" f"'{previous.name}'与" f"'{activity.name}'" "存在时间重叠。" ), day_index=day.day_index, activity_sequence=( activity.sequence ), ) previous = activity if ( activity.activity_type == "attraction" ): attraction_count += 1 attraction_key = ( activity.reference_id or activity.name ) if attraction_key in used_attractions: self._add_issue( issues, code=( "DUPLICATE_ATTRACTION" ), severity="error", message=( f"景点'{activity.name}'" "在行程中被重复安排。" ), day_index=day.day_index, activity_sequence=( activity.sequence ), ) used_attractions.add( attraction_key ) self._validate_place_reference( activity=activity, places_by_id=( attractions_by_id ), places_by_name=( attractions_by_name ), place_label="景点", issues=issues, day_index=day.day_index, ) elif ( activity.activity_type == "restaurant" ): self._validate_place_reference( activity=activity, places_by_id=( restaurants_by_id ), places_by_name=( restaurants_by_name ), place_label="餐厅", issues=issues, day_index=day.day_index, ) if attraction_count > attraction_limit: self._add_issue( issues, code="PACE_LIMIT_EXCEEDED", severity="warning", message=( f"第{day.day_index}天安排了" f"{attraction_count}个主要景点，" f"超过{request.pace}节奏建议的" f"{attraction_limit}个。" ), day_index=day.day_index, ) def _validate_place_reference( self, *, activity: ItineraryActivity, places_by_id: dict[str, Any], places_by_name: dict[str, Any], place_label: str, issues: list[ValidationIssue], day_index: int, ) -> None: """确认活动引用的POI来自高德查询结果。""" if activity.reference_id: place = places_by_id.get( activity.reference_id ) if place is None: self._add_issue( issues, code=f"UNKNOWN_{place_label}_ID", severity="error", message=( f"{place_label}" f"'{activity.name}'的POI ID" "不存在于高德查询候选中。" ), day_index=day_index, activity_sequence=( activity.sequence ), ) return if place.name != activity.name: self._add_issue( issues, code=f"{place_label}_NAME_MISMATCH", severity="error", message=( f"{place_label}POI ID对应名称" f"为'{place.name}'，" f"Planner却填写为" f"'{activity.name}'。" ), day_index=day_index, activity_sequence=( activity.sequence ), ) return if activity.name not in places_by_name: self._add_issue( issues, code=f"UNKNOWN_{place_label}_NAME", severity="error", message=( f"{place_label}" f"'{activity.name}'" "不存在于高德查询候选中。" ), day_index=day_index, activity_sequence=( activity.sequence ), ) def _validate_boundary_days( self, *, request: TravelRequest, plan: ItineraryPlan, flight_data: dict[str, Any], issues: list[ValidationIssue], ) -> None: """检查到达日和返程日安排是否合理。""" outbound = flight_data.get( "outbound", {}, ) return_flight = flight_data.get( "return_flight", {}, ) if not isinstance(outbound, dict): outbound = {} if not isinstance(return_flight, dict): return_flight = {} outbound_departure = self._parse_datetime( outbound.get("departure_time") ) outbound_arrival = self._parse_datetime( outbound.get("arrival_time") ) return_departure = self._parse_datetime( return_flight.get( "departure_time" ) ) # 检查去程日期。 if ( outbound_departure is not None and request.departure_date is not None and outbound_departure.date() != request.departure_date ): self._add_issue( issues, code="OUTBOUND_DATE_MISMATCH", severity="error", message=( "所选去程航班日期与" "用户出发日期不一致。" ), ) # 检查返程日期。 if ( return_departure is not None and request.return_date is not None and return_departure.date() != request.return_date ): self._add_issue( issues, code="RETURN_DATE_MISMATCH", severity="error", message=( "所选返程航班日期与" "用户返程日期不一致。" ), ) departure_day = next( ( day for day in plan.days if day.date == request.departure_date ), None, ) return_day = next( ( day for day in plan.days if day.date == request.return_date ), None, ) if departure_day is not None: has_outbound_flight = any( activity.activity_type == "flight" for activity in departure_day.activities ) if not has_outbound_flight: self._add_issue( issues, code=( "OUTBOUND_FLIGHT_ACTIVITY_MISSING" ), severity="warning", message=( "到达日行程中没有明确记录" "去程航班活动。" ), day_index=( departure_day.day_index ), ) if ( departure_day is not None and outbound_arrival is not None and outbound_arrival.date() == departure_day.date ): # 默认预留90分钟完成下机、 # 取行李和离开机场。 recommended_ready_time = ( outbound_arrival + timedelta(minutes=90) ) for activity in ( departure_day.activities ): if activity.activity_type in { "flight", "transport", }: continue activity_start = datetime.combine( departure_day.date, activity.start_time, ) if activity_start < outbound_arrival: self._add_issue( issues, code=( "ACTIVITY_BEFORE_ARRIVAL" ), severity="error", message=( f"到达日活动" f"'{activity.name}'" "安排在航班到达之前。" ), day_index=( departure_day.day_index ), activity_sequence=( activity.sequence ), ) elif ( activity_start < recommended_ready_time ): self._add_issue( issues, code=( "ARRIVAL_BUFFER_TOO_SHORT" ), severity="warning", message=( f"到达日活动" f"'{activity.name}'" "距离航班到达时间过近，" "建议预留下机、行李和" "机场离场时间。" ), day_index=( departure_day.day_index ), activity_sequence=( activity.sequence ), ) if return_day is not None: has_return_flight = any( activity.activity_type == "flight" for activity in return_day.activities ) if ( return_flight and not has_return_flight ): self._add_issue( issues, code=( "RETURN_FLIGHT_ACTIVITY_MISSING" ), severity="warning", message=( "返程日行程中没有明确记录" "返程航班活动。" ), day_index=return_day.day_index, ) if ( return_day is not None and return_departure is not None and return_departure.date() == return_day.date ): # 默认国内航班提前180分钟结束 # 景点、餐饮等非交通活动。 recommended_leave_time = ( return_departure - timedelta(minutes=180) ) for activity in return_day.activities: if activity.activity_type in { "flight", "transport", }: continue activity_end = datetime.combine( return_day.date, activity.end_time, ) if activity_end > return_departure: self._add_issue( issues, code=( "ACTIVITY_AFTER_FLIGHT" ), severity="error", message=( f"返程日活动" f"'{activity.name}'" "结束时间晚于返程航班" "起飞时间。" ), day_index=return_day.day_index, activity_sequence=( activity.sequence ), ) elif ( activity_end > recommended_leave_time ): self._add_issue( issues, code=( "RETURN_BUFFER_TOO_SHORT" ), severity="warning", message=( f"返程日活动" f"'{activity.name}'" "距离航班起飞时间较近，" "可能没有充分预留" "前往机场和办理登机的时间。" ), day_index=return_day.day_index, activity_sequence=( activity.sequence ), ) def _build_budget( self, *, request: TravelRequest, plan: ItineraryPlan, flight_quote: float | None, hotel_estimate: float | None, issues: list[ValidationIssue], ) -> BudgetEstimate: """计算目前能够确定的旅行费用。""" explicit_activity_costs = 0.0 for day in plan.days: for activity in day.activities: if activity.activity_type in { "flight", "hotel_check_in", "hotel_check_out", }: # 航班和酒店已经单独统计， # 防止重复计费。 continue if activity.estimated_cost is not None: explicit_activity_costs += ( activity.estimated_cost ) known_components: list[float] = [] if flight_quote is not None: known_components.append(flight_quote) else: self._add_issue( issues, code="FLIGHT_PRICE_UNKNOWN", severity="warning", message=( "所选航班缺少可解析报价，" "无法完整估算预算。" ), ) if hotel_estimate is not None: known_components.append( hotel_estimate ) else: self._add_issue( issues, code="HOTEL_PRICE_UNKNOWN", severity="warning", message=( "所选酒店缺少可解析价格，" "无法完整估算预算。" ), ) known_components.append( explicit_activity_costs ) known_total = ( round(sum(known_components), 2) if known_components else None ) user_budget = request.total_budget remaining_known_budget: ( float | None ) = None status = "unknown" if ( user_budget is not None and known_total is not None ): remaining_known_budget = round( user_budget - known_total, 2, ) if known_total > user_budget: status = "over_budget" self._add_issue( issues, code="KNOWN_BUDGET_EXCEEDED", severity="error", message=( "当前可确定费用已超过" f"用户总预算：已知费用" f"{known_total:.0f}" f"{request.currency}，" f"预算为" f"{user_budget:.0f}" f"{request.currency}。" ), ) else: status = "within_known_budget" coverage_note = ( "当前预算仅包含航班接口报价、" "酒店估算和行程中明确填写的活动费用；" "未明确查询的餐饮、门票、市内交通、" "保险及其他费用未计入。" "航班接口报价的计价范围仍以" "实际供应商页面为准。" ) return BudgetEstimate( currency=request.currency, flight_quote=flight_quote, hotel_estimate=hotel_estimate, explicit_activity_costs=round( explicit_activity_costs, 2, ), known_total=known_total, user_budget=user_budget, remaining_known_budget=( remaining_known_budget ), status=status, coverage_note=coverage_note, ) @staticmethod def _parse_datetime( value: object, ) -> datetime | None: """安全解析时间字符串，解析失败返回None。""" if value is None: return None if isinstance(value, datetime): return value try: return datetime.fromisoformat( str(value) ) except ValueError: return None @staticmethod def _to_float( value: object, ) -> float | None: """安全转换为float，不可转换值返回None。""" if value is None: return None if isinstance(value, bool): return None try: return float(value) except (TypeError, ValueError): return None @staticmethod def _numbers_close( first: float, second: float, ) -> bool: """允许1元或1%的价格误差。""" tolerance = max( 1.0, abs(second) * 0.01, ) return abs(first - second) <= tolerance @staticmethod def _add_issue( issues: list[ValidationIssue], *, code: str, severity: str, message: str, day_index: int | None = None, activity_sequence: int | None = None, ) -> None: """按严重级别（error/warning）向问题列表追加一条校验问题。""" issues.append( ValidationIssue( code=code, severity=severity, message=message, day_index=day_index, activity_sequence=( activity_sequence ), ) )