110 lines
3.1 KiB
Python
110 lines
3.1 KiB
Python
"""运输计划公共页共享纯函数（司机/现场 history 等）。"""
|
|
from __future__ import annotations
|
|
|
|
import calendar
|
|
import re
|
|
from datetime import date, datetime
|
|
from typing import Any, Iterable, Optional
|
|
|
|
from app.models.pl_models import PlPlan, PlTransport
|
|
|
|
CRM_PRODUCT_UNIT = "crm_product_unit"
|
|
|
|
_MONTH_RE = re.compile(r"^(\d{4})-(\d{2})$")
|
|
|
|
|
|
def _resolve_unit_label(unit_map: dict[str, str], unit_code: Optional[str]) -> str:
|
|
code = (unit_code or "").strip()
|
|
if not code:
|
|
return "吨"
|
|
return unit_map.get(code, code)
|
|
|
|
|
|
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 _parse_calendar_month(raw: Optional[str]) -> Optional[tuple[date, date]]:
|
|
"""解析 YYYY-MM，返回当月 [start, end]；无效返回 None。"""
|
|
text = (raw or "").strip()
|
|
m = _MONTH_RE.match(text)
|
|
if not m:
|
|
return None
|
|
year, month = int(m.group(1)), int(m.group(2))
|
|
if month < 1 or month > 12:
|
|
return None
|
|
last_day = calendar.monthrange(year, month)[1]
|
|
return date(year, month, 1), date(year, month, last_day)
|
|
|
|
|
|
def _as_plain_date(value: Any) -> Optional[date]:
|
|
if value is None:
|
|
return None
|
|
if isinstance(value, datetime):
|
|
return value.date()
|
|
if isinstance(value, date):
|
|
return value
|
|
return None
|
|
|
|
|
|
def _plan_hit_dates(plan: Any, *, include_plan_date: bool) -> set[date]:
|
|
"""计划在 history 中可能命中的日期集合。"""
|
|
days: set[date] = set()
|
|
if include_plan_date:
|
|
d = _as_plain_date(getattr(plan, "plan_date", None))
|
|
if d is not None:
|
|
days.add(d)
|
|
for attr in ("pickup_time", "delivery_time"):
|
|
d = _as_plain_date(getattr(plan, attr, None))
|
|
if d is not None:
|
|
days.add(d)
|
|
return days
|
|
|
|
|
|
def _accumulate_calendar_day_counts(
|
|
plans: Iterable[Any],
|
|
*,
|
|
month_start: date,
|
|
month_end: date,
|
|
include_plan_date: bool,
|
|
) -> dict[str, int]:
|
|
"""按日累计计划条数（仅统计落在当月内的命中日）。"""
|
|
counts: dict[str, int] = {}
|
|
for plan in plans:
|
|
for day in _plan_hit_dates(plan, include_plan_date=include_plan_date):
|
|
if day < month_start or day > month_end:
|
|
continue
|
|
key = day.isoformat()
|
|
counts[key] = counts.get(key, 0) + 1
|
|
return counts
|
|
|
|
|
|
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]:
|
|
"""返回 (作废, 归档)。"""
|
|
voided = (
|
|
(plan is not None and plan.follow_status == "cancelled")
|
|
or transport.transport_status == "cancelled"
|
|
)
|
|
archived = bool(plan is not None and plan.archive_status == "archived")
|
|
return voided, archived
|