feat(pl): add transport history helpers and schemas

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
zlwuandCursor committed 2026-08-21 11:18:34 +08:00
1 parent 74146de156
commit 0e90b1d949
3 files changed
+97

No files matched your search

+36
View File
@@ -87,6 +87,42 @@ def _format_plan_date(value) -> Optional[str]:
return str(value)
def _normalize_main_plate(value: Optional[str]) -> str:
return (value or "").strip()
def _parse_history_date(raw: Optional[str]) -> Optional[date]:
text = (raw or "").strip()
if not text:
return None
try:
return date.fromisoformat(text)
except ValueError:
return None
def _as_date(value) -> Optional[date]:
if value is None:
return None
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
return None
def _history_matches_date(pickup_time, delivery_time, day: date) -> bool:
return _as_date(pickup_time) == day or _as_date(delivery_time) == day
def _history_status_label(plan_voided: bool, plan_archived: bool) -> str:
if plan_voided:
return "已作废"
if plan_archived:
return "已归档"
return "进行中"
def _resolve_driver_link_flags(
plan: Optional[PlPlan], transport: PlTransport
) -> tuple[bool, bool]:
+24
View File
@@ -283,6 +283,30 @@ class PublicTransportPageRead(BaseModel):
model_config = {"populate_by_name": True}
class PublicTransportHistoryRecordRead(BaseModel):
plan_id: str = Field(alias="planId")
plan_no: str = Field(alias="planNo")
transport_id: str = Field(alias="transportId")
driver_access_token: Optional[str] = Field(default=None, alias="driverAccessToken")
driver_url: Optional[str] = Field(default=None, alias="driverUrl")
license_plate: Optional[str] = Field(default=None, alias="licensePlate")
pickup_time: Optional[datetime] = Field(default=None, alias="pickupTime")
delivery_time: Optional[datetime] = Field(default=None, alias="deliveryTime")
pickup_address: Optional[str] = Field(default=None, alias="pickupAddress")
destination_address: Optional[str] = Field(default=None, alias="destinationAddress")
plan_voided: bool = Field(default=False, alias="planVoided")
plan_archived: bool = Field(default=False, alias="planArchived")
status_label: str = Field(default="进行中", alias="statusLabel")
model_config = {"populate_by_name": True}
class PublicTransportHistoryPageRead(BaseModel):
license_plate: Optional[str] = Field(default=None, alias="licensePlate")
date: str
records: list[PublicTransportHistoryRecordRead] = Field(default_factory=list)
model_config = {"populate_by_name": True}
class PublicTransportSitePageRead(BaseModel):
business_type: str = Field(alias="businessType")
date: Optional[str] = None
@@ -0,0 +1,37 @@
from datetime import date, datetime
from app.api.pl.public_transport import (
_history_matches_date,
_normalize_main_plate,
_parse_history_date,
_history_status_label,
)
def test_normalize_main_plate_trims():
assert _normalize_main_plate(" 苏AD12345 ") == "苏AD12345"
assert _normalize_main_plate(None) == ""
assert _normalize_main_plate("") == ""
def test_parse_history_date_ok_and_bad():
assert _parse_history_date("2026-08-21") == date(2026, 8, 21)
assert _parse_history_date("2026/08/21") is None
assert _parse_history_date("") is None
def test_history_matches_date_or_logic():
day = date(2026, 8, 21)
assert _history_matches_date(datetime(2026, 8, 21, 9, 0), None, day) is True
assert _history_matches_date(None, datetime(2026, 8, 21, 18, 0), day) is True
assert _history_matches_date(
datetime(2026, 8, 20, 9, 0), datetime(2026, 8, 22, 18, 0), day
) is False
assert _history_matches_date(None, None, day) is False
def test_history_status_label():
assert _history_status_label(True, False) == "已作废"
assert _history_status_label(True, True) == "已作废"
assert _history_status_label(False, True) == "已归档"
assert _history_status_label(False, False) == "进行中"