95 lines
2.9 KiB
Python
95 lines
2.9 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(
|
|
customer_name: Optional[str],
|
|
stop_type: StopType,
|
|
) -> Tuple[bool, str]:
|
|
"""地磅默认勾选以本段客户是否为内部客户（山峡/云栋）为准。"""
|
|
if not is_internal_customer_name(customer_name):
|
|
return False, "none"
|
|
if stop_type == "load":
|
|
return True, "outbound"
|
|
if stop_type == "delivery":
|
|
return True, "inbound"
|
|
return False, "none"
|
|
|
|
|
|
def normalize_stop_dibang_fields(
|
|
customer_name: Optional[str],
|
|
stop_type: StopType,
|
|
generate_dibang: Optional[int],
|
|
dibang_direction: Optional[str],
|
|
) -> Tuple[int, str]:
|
|
"""归一化段次地磅勾选与方向。
|
|
|
|
- 非内部客户：强制不生成地磅单
|
|
- 内部客户：尊重显式勾选值（含可选车取消勾选传 0）；仅 None 时用默认
|
|
- 勾选但方向缺失/none 时按本段客户补默认方向
|
|
"""
|
|
if not is_internal_customer_name(customer_name):
|
|
return 0, "none"
|
|
|
|
default_gen, default_dir = resolve_stop_dibang_defaults(customer_name, stop_type)
|
|
if generate_dibang is None:
|
|
gen = 1 if default_gen else 0
|
|
else:
|
|
gen = 1 if int(generate_dibang) == 1 else 0
|
|
|
|
direction = (dibang_direction or "").strip()
|
|
if gen:
|
|
if direction not in ("outbound", "inbound"):
|
|
direction = default_dir if default_dir != "none" else (
|
|
"outbound" if stop_type == "load" else "inbound"
|
|
)
|
|
else:
|
|
direction = "none"
|
|
return gen, direction
|
|
|
|
|
|
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)})不一致，请核对"
|
|
]
|