63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
"""运输计划多段：内部客户地磅默认值与数量平衡校验"""
|
|
from __future__ import annotations
|
|
|
|
from decimal import Decimal
|
|
from typing import Any, Iterable, Literal, Optional, Tuple
|
|
|
|
from app.services.pl_internal_customer import is_internal_customer_name
|
|
|
|
StopType = Literal["load", "delivery"]
|
|
|
|
|
|
def resolve_stop_dibang_defaults(
|
|
load_customer_name: Optional[str],
|
|
stop_type: StopType,
|
|
) -> Tuple[bool, str]:
|
|
"""地磅默认勾选以装货客户是否为内部客户为准。"""
|
|
if not is_internal_customer_name(load_customer_name):
|
|
return False, "none"
|
|
if stop_type == "load":
|
|
return True, "outbound"
|
|
if stop_type == "delivery":
|
|
return True, "inbound"
|
|
return False, "none"
|
|
|
|
|
|
def _extract_qty(item: Any, *field_names: str) -> Decimal:
|
|
for name in field_names:
|
|
if isinstance(item, dict):
|
|
value = item.get(name)
|
|
else:
|
|
value = getattr(item, name, None)
|
|
if value is not None:
|
|
return Decimal(str(value))
|
|
return Decimal("0")
|
|
|
|
|
|
def _sum_qty(items: Iterable[Any], *field_names: str) -> Decimal:
|
|
total = Decimal("0")
|
|
for item in items or []:
|
|
total += _extract_qty(item, *field_names)
|
|
return total
|
|
|
|
|
|
def _fmt_qty(qty: Decimal) -> str:
|
|
if qty == qty.to_integral_value():
|
|
return str(int(qty))
|
|
return format(qty.normalize(), "f")
|
|
|
|
|
|
def warn_qty_imbalance(
|
|
plan_products: Iterable[Any],
|
|
load_products: Iterable[Any],
|
|
delivery_products: Iterable[Any],
|
|
) -> list[str]:
|
|
load_total = _sum_qty(load_products, "plan_qty")
|
|
delivery_total = _sum_qty(delivery_products, "plan_qty")
|
|
|
|
if load_total == delivery_total:
|
|
return []
|
|
return [
|
|
f"装货合计({_fmt_qty(load_total)})与送货合计({_fmt_qty(delivery_total)})不一致，请核对"
|
|
]
|