chendandan_homework/04_work/app/schemas/itinerary.py

from __future__ import annotations

"""行程规划结果模型：ItineraryPlan多日行程、FlightSelection航班选择、HotelSelection酒店选择及每日活动。"""

from datetime import date, time
from typing import Literal

from pydantic import Field, field_validator, model_validator

from app.schemas.common import AppModel


ActivityType = Literal[
    "flight",
    "hotel_check_in",
    "hotel_check_out",
    "attraction",
    "restaurant",
    "transport",
    "free_time",
    "other",
]

TransportMode = Literal[
    "flight",
    "walking",
    "public_transit",
    "driving",
    "taxi",
    "unknown",
]


class FlightSelection(AppModel):
    """规划角色选择的航班方案。

    这里保存候选ID，后续Validator将检查这些ID
    是否真实存在于候选结果中。
    """

    flight_type: Literal[
        "one_way",
        "round_trip",
    ]

    combination_id: str | None = None

    outbound_option_id: str
    return_option_id: str | None = None

    # 接口报价（不是去程+返程的加总）。
    quoted_price: float | None = Field(
        default=None,
        ge=0,
    )

    currency: str = "CNY"

    outbound_flight_numbers: list[str] = Field(
        default_factory=list
    )

    return_flight_numbers: list[str] = Field(
        default_factory=list
    )

    outbound_departure_time: str | None = None
    outbound_arrival_time: str | None = None

    return_departure_time: str | None = None
    return_arrival_time: str | None = None

    selection_reason: str

    @model_validator(mode="after")
    def validate_round_trip(
        self,
    ) -> "FlightSelection":
        if self.flight_type == "round_trip":
            if not self.combination_id:
                raise ValueError(
                    "往返方案必须提供combination_id。"
                )

            if not self.return_option_id:
                raise ValueError(
                    "往返方案必须提供return_option_id。"
                )

        return self


class HotelSelection(AppModel):
    """规划角色选择的住宿方案。"""

    hotel_id: str
    name: str

    check_in_date: date
    check_out_date: date
    nights: int = Field(ge=1)

    price_per_night: float | None = Field(
        default=None,
        ge=0,
    )

    estimated_total_price: float | None = Field(
        default=None,
        ge=0,
    )

    currency: str = "CNY"

    address: str | None = None
    location: str | None = None

    selection_reason: str

    @field_validator("location", mode="before")
    @classmethod
    def _coerce_location_to_str(
        cls,
        value: object,
    ) -> object:
        """LLM 可能将 coordinates dict 误填到 location，
        自动转换为 "longitude,latitude" 字符串。"""
        if isinstance(value, dict):
            lat = value.get("latitude")
            lng = value.get("longitude")
            if lat is not None and lng is not None:
                return f"{lng},{lat}"
        return value


class ItineraryActivity(AppModel):
    """一天中的一个具体活动。"""

    sequence: int = Field(ge=1)

    activity_type: ActivityType

    # 景点或餐厅对应的真实POI ID。
    # 航班、入住等活动允许为空。
    reference_id: str | None = None

    name: str

    start_time: time
    end_time: time

    address: str | None = None
    location: str | None = None

    transport_mode: TransportMode | None = None

    estimated_transport_minutes: int | None = Field(
        default=None,
        ge=0,
    )

    estimated_cost: float | None = Field(
        default=None,
        ge=0,
    )

    notes: str | None = None


class DailyItinerary(AppModel):
    """单日行程。"""

    day_index: int = Field(
        ge=1,
        le=30,
    )

    date: date

    theme: str

    activities: list[ItineraryActivity] = Field(
        min_length=1
    )

    daily_notes: list[str] = Field(
        default_factory=list
    )


class ItineraryPlan(AppModel):
    """行程规划角色生成的完整结构化方案。"""

    title: str
    overview: str

    selected_flight: FlightSelection

    selected_hotel: HotelSelection

    days: list[DailyItinerary] = Field(
        min_length=1
    )

    highlights: list[str] = Field(
        default_factory=list
    )

    assumptions: list[str] = Field(
        default_factory=list
    )

    warnings: list[str] = Field(
        default_factory=list
    )

    @model_validator(mode="after")
    def validate_days(
        self,
    ) -> "ItineraryPlan":
        day_indexes = [
            item.day_index
            for item in self.days
        ]

        dates = [
            item.date
            for item in self.days
        ]

        if len(day_indexes) != len(set(day_indexes)):
            raise ValueError(
                "每日行程的day_index不能重复。"
            )

        if len(dates) != len(set(dates)):
            raise ValueError(
                "每日行程日期不能重复。"
            )

        return self