chendandan_homework/app/schemas/flight.py

from __future__ import annotations

"""航班查询与结果模型：FlightSearchQuery查询参数与FlightOption搜索结果。"""

from datetime import datetime,date
from typing import Literal
from pydantic import field_validator,Field,model_validator
from app.schemas.common import SourceMetadata,AppModel

FlightType = Literal["round_trip","one_way"]

TravelClass = Literal["economy","business","first","premium_economy"]

FlightSort = Literal["top","price","departure_time","arrival_time","duration","emissions"]

StopsFilter = Literal["any","nonstop","one_or_fewer","two_or_fewer"]

class FlightSearchQuery(AppModel):
    """
    航班查询参数
    当前所有筛选条件默认不启用。
    后续由需求解析agent根据用户真实需求赋值
    """
    departure_airports:list[str] = Field(min_length=1)
    arrival_airports:list[str] = Field(min_length=1)

    outbound_date:date
    return_date:date | None = None

    flight_type:FlightType="one_way"

    adults:int = Field(default=1, ge=1, le=9)
    children:int = Field(default=0, ge=0, le=8)

    travel_class:TravelClass="economy"
    currency:str = "CNY"

    # 以下全部是可选条件，不设置就不限制
    sort_by:FlightSort | None=None
    stops:StopsFilter | None=None
    max_price:int | None = Field(default=None,gt=0)

    # SerpApi 时间范围格式，例如：
    # "9,20" 或 "9,20,10,22"
    outbound_times:str | None=None
    language:str = "zh-cn"
    country:str = "cn"

    no_cache:bool = False
    # 是否加载 Google Flights 的“查看更多航班”
    show_hidden: bool = False
    # 是否启用更完整但响应更慢的深度查询
    deep_search: bool = False

    @field_validator("departure_airports","arrival_airports")
    @classmethod
    def normalize_airports(cls,value:list[str])->list[str]:
        """机场代码统一转为大写并去重"""
        normalized:list[str] = []

        for item in value:
            code = item.strip().upper()
            if not code:
                continue
            if code not in normalized:
                normalized.append(code)

        if not normalized:
            raise ValueError("机场代码不能为空")
        return normalized
    
    @field_validator("currency")
    @classmethod
    def normalize_currency(cls,value:str)->str:
        return value.upper()
    
    @model_validator(mode="after")
    def validate_dates(self)->"FlightSearchQuery":
        if self.flight_type == "round_trip":
            if self.return_date is None:
                raise ValueError("往返航班必须提供return_date。")
            if self.return_date <= self.outbound_date:
                raise ValueError("返程日期必须晚于出发日期。")
        return self
    

class FlightSegment(AppModel):
    """一个具体航段，中转航班会包含FlightSegment。"""
    flight_number:str | None=None
    airline:str | None=None
    departure_airport_code:str
    departure_airport_name:str | None=None
    departure_time:datetime
    arrival_airport_code:str
    arrival_airport_name:str | None=None
    arrival_time:datetime

    duration_minutes:int | None=Field(default=None,ge=0)
    airplane:str | None=None
    travel_class:str | None=None
    legroom:str | None=None

    overnight:bool = False
    often_delayed_by_over_30_min:bool = False
    planed_and_crew_by:str | None=None
    extensions:list[str] = Field(default_factory=list)
    ticket_also_sold_by:list[str] = Field(default_factory=list)

class Layover(AppModel):
    """经停或中转信息。"""
    airport_code:str
    airport_name:str | None=None

    duration_minutes:int | None=Field(default=None,ge=0)
    overnight:bool = False

class CarbonEmissions(AppModel):
    """航班碳排放数据。"""
    this_flight_grams:int | None=None
    typical_for_route_grams:int | None=None
    difference_percent:int | None=None

# SerpApi 返回的航班分组:best_flights / other_flights 等
FlightSourceGroup = Literal["best","other","best_flights","other_flights"]

class FlightOption(AppModel):
    """一个完整的去程或者返程候选方案。"""
    option_id:str
    source_group:FlightSourceGroup
    provider_rank:int = Field(ge=0)

    segments:list[FlightSegment] = Field(min_length=1)
    layovers:list[Layover] = Field(default_factory=list)

    departure_airport_code:str
    final_arrival_airport_code:str

    departure_time:datetime
    arrival_time:datetime

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

    stop_count:int = Field(default=0,ge=0)

    price:int | None=Field(default=None,gt=0)
    currency:str = "CNY"

    flight_type:str | None=None

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

    is_overnight:bool = False

    carbon_emissions:CarbonEmissions | None=None

    # 往返查询选择去程后，用于继续查询返程
    departure_token:str | None=None

    # 获取预定选项是可能使用
    booking_token:str | None=None

class FlightSearchResult(AppModel):
    """标准化后的航班查询结果。"""
    
    query:FlightSearchQuery
    flights:list[FlightOption] = Field(default_factory=list)
    total_count:int = Field(default=0,ge=0)

    source:SourceMetadata

    price_insights:dict | None=None
    warnings:list[str] = Field(default_factory=list)