chendandan_homework/app/services/plan_validation_service.py

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
                ),
            )
        )