geo-setp/schemas.py

from pydantic import BaseModel, Field, field_validator, model_validator
from typing import Literal


# ---------------------------------------------------------------------------
# Input
# ---------------------------------------------------------------------------

class RawStop(BaseModel):
    name: str | None = None
    address: str
    city: str | None = None
    contact: str | None = None

    @field_validator("name", "address", "city", "contact", mode="before")
    @classmethod
    def strip_optional_strings(cls, value: str | None) -> str | None:
        if value is None:
            return None
        text = value.strip()
        return text or None

    @field_validator("address")
    @classmethod
    def validate_address(cls, value: str) -> str:
        if not value:
            raise ValueError("stop address cannot be empty")
        return value


class RoutePlanRequest(BaseModel):
    task_name: str = Field(default="multi-destination-route-planning")
    origin_mode: Literal["fixed", "current_location"]
    origin_name: str | None = None
    origin_address: str | None = None
    origin_city: str | None = None
    destination_name: str | None = None
    destination_address: str
    destination_city: str | None = None
    stops: list[RawStop]
    route_strategy: Literal["shortest_distance", "fastest_time", "balanced"] = "shortest_distance"
    transport_mode: Literal["driving"] = "driving"
    need_deep_link: bool = True
    deep_link_mode: Literal["personal_map", "route_plan", "auto"] = "auto"
    need_html: bool = False
    max_permutations: int | None = None

    @field_validator(
        "task_name",
        "origin_name",
        "origin_address",
        "origin_city",
        "destination_name",
        "destination_address",
        "destination_city",
        mode="before",
    )
    @classmethod
    def strip_request_strings(cls, value: str | None) -> str | None:
        if value is None:
            return None
        text = value.strip()
        return text or None

    @field_validator("destination_address")
    @classmethod
    def validate_destination_address(cls, value: str) -> str:
        if not value:
            raise ValueError("destination_address cannot be empty")
        return value

    @field_validator("max_permutations")
    @classmethod
    def validate_max_permutations(cls, value: int | None) -> int | None:
        if value is not None and value <= 0:
            raise ValueError("max_permutations must be greater than 0")
        return value

    @model_validator(mode="after")
    def validate_route_request(self) -> "RoutePlanRequest":
        if not self.stops:
            raise ValueError("stops must contain at least one stop")

        if self.origin_mode == "fixed" and not self.origin_address:
            raise ValueError("origin_address is required when origin_mode='fixed'")

        destination_key = self.destination_address.casefold()
        duplicate_stop = next(
            (stop.address for stop in self.stops if stop.address.casefold() == destination_key),
            None,
        )
        if duplicate_stop is not None:
            raise ValueError("destination_address cannot also appear in stops")

        return self


# ---------------------------------------------------------------------------
# Output
# ---------------------------------------------------------------------------

class ResolvedPoint(BaseModel):
    role: Literal["origin", "stop", "destination"]
    input_name: str | None = None
    input_address: str
    resolved_name: str
    city: str | None = None
    district: str | None = None
    location: str  # "lon,lat" format as returned by Amap
    lon: float
    lat: float
    poi_id: str | None = None
    source: Literal["geo", "text_search", "search_detail", "manual_fallback"]
    confidence_note: str | None = None


class RouteLeg(BaseModel):
    from_label: str
    to_label: str
    origin_location: str
    destination_location: str
    distance_m: int
    duration_s: int


class CandidateRoute(BaseModel):
    stop_order_labels: list[str]
    full_order_labels: list[str]
    legs: list[RouteLeg]
    total_distance_m: int
    total_duration_s: int
    ranking_reason: str | None = None


class DeepLinks(BaseModel):
    personal_map: str | None = None
    android_route_plan: str | None = None
    ios_route_plan: str | None = None


class RoutePlanResult(BaseModel):
    success: bool
    origin_mode: Literal["fixed", "current_location"]
    resolved_origin: ResolvedPoint | None = None
    resolved_destination: ResolvedPoint
    resolved_stops: list[ResolvedPoint]
    candidates: list[CandidateRoute]
    best_route: CandidateRoute
    deep_links: DeepLinks | None = None
    summary: str
    warnings: list[str]


class LoadPlanRequest(BaseModel):
    merchant_id: int
    area: str
    license_plate: str


class LoadPlanResult(BaseModel):
    success: bool
    license_plate: str
    selected_shipment_ids: list[int] = Field(default_factory=list)
    summary: str
    warnings: list[str] = Field(default_factory=list)


# ---------------------------------------------------------------------------
# Load planning API integration
# ---------------------------------------------------------------------------

class TransportVehicleMaterialCapacity(BaseModel):
    id: int
    material_name: str
    capacity: int


class TransportVehicle(BaseModel):
    id: int
    merchant_id: int
    name: str
    license_plate: str
    material_capacities: list[TransportVehicleMaterialCapacity] = Field(default_factory=list)
    created_at: str
    updated_at: str


class ShipmentGeoCoordinates(BaseModel):
    lat: float
    lng: float


class ShipmentSalesItem(BaseModel):
    id: int
    name: str
    quantity: str
    unit: int | None = None
    unit_display: str | None = None
    position: str | None = None
    remark: str | None = None
    printing_job_id: int | None = None
    printing_job_width: str | None = None


class Shipment(BaseModel):
    id: int
    merchant_id: int
    customer: int | None = None
    customer_name: str | None = None
    shipment_date: str
    address: str
    contact_name: str | None = None
    contact_phone: str | None = None
    area: str | None = None
    remark: str | None = None
    status: int | str
    status_display: str | None = None
    external_id: str | None = None
    geo_coordinates: ShipmentGeoCoordinates | None = None
    delivery: int | None = None
    sales_items: list[ShipmentSalesItem] = Field(default_factory=list)
    created_at: str
    updated_at: str


class ShipmentPage(BaseModel):
    count: int | None = None
    next: str | None = None
    previous: str | None = None
    results: list[Shipment] = Field(default_factory=list)