from __future__ import annotations """用户旅行需求模型:从自然语言解析出的结构化TravelRequest,包含偏好、日期、预算等所有维度。""" from datetime import date, time from typing import Literal from pydantic import Field, field_validator from app.schemas.common import AppModel TravelPace = Literal[ "relaxed", "normal", "intensive", ] TravelPriority = Literal[ "price", "convenience", "balanced", ] class FlightPreferences(AppModel): """用户对航班的偏好。 这些字段全部是软偏好或可选限制。 用户没有明确表达时,不主动添加限制。 """ priority: TravelPriority = "balanced" preferred_departure_airports: list[str] = Field( default_factory=list ) preferred_arrival_airports: list[str] = Field( default_factory=list ) earliest_departure_time: time | None = None latest_departure_time: time | None = None earliest_arrival_time: time | None = None latest_arrival_time: time | None = None # 0表示只接受直飞;None表示不限制。 max_stops: int | None = Field( default=None, ge=0, le=3, ) @field_validator( "preferred_departure_airports", "preferred_arrival_airports", ) @classmethod def normalize_airports( cls, value: list[str], ) -> list[str]: result: list[str] = [] for airport in value: code = airport.strip().upper() if code and code not in result: result.append(code) return result class HotelPreferences(AppModel): """用户对酒店的偏好。""" max_price_per_night: float | None = Field( default=None, gt=0, ) minimum_rating: float | None = Field( default=None, ge=0, le=5, ) hotel_classes: list[int] = Field( default_factory=list, ) @field_validator("hotel_classes") @classmethod def validate_hotel_classes( cls, value: list[int], ) -> list[int]: for item in value: if item not in (2, 3, 4, 5): raise ValueError( f"酒店星级必须在 2-5 之间,实际: {item}" ) return value near_subway: bool | None = None preferred_areas: list[str] = Field( default_factory=list ) amenities: list[str] = Field( default_factory=list ) class TravelRequest(AppModel): """从用户自然语言中提取的完整旅行需求。""" origin_city: str | None = None destination_city: str | None = None departure_date: date | None = None return_date: date | None = None # “4天3晚”等信息先保留下来, # 后续确定性代码可以据此推算返程日期。 trip_days: int | None = Field( default=None, ge=1, le=30, ) nights: int | None = Field( default=None, ge=0, le=29, ) adults: int = Field( default=1, ge=1, le=9, ) children: int = Field( default=0, ge=0, le=8, ) @field_validator( "children", mode="before", ) @classmethod def normalize_children( cls, value: object, ) -> int: """LLM 可能误将 children 输出为 [],防御性转换为 int。""" if value is None: return 0 if isinstance(value, bool): return 1 if value else 0 if isinstance(value, (list, tuple)): return len(value) return int(value) total_budget: float | None = Field( default=None, gt=0, ) currency: str = "CNY" interests: list[str] = Field( default_factory=list ) pace: TravelPace = "normal" flight_preferences: FlightPreferences = Field( default_factory=FlightPreferences ) hotel_preferences: HotelPreferences = Field( default_factory=HotelPreferences ) special_requirements: list[str] = Field( default_factory=list ) @field_validator( "origin_city", "destination_city", mode="before", ) @classmethod def normalize_city( cls, value: object, ) -> str | None: if value is None: return None text = str(value).strip() return text or None @field_validator("currency") @classmethod def normalize_currency( cls, value: str, ) -> str: return value.strip().upper()