shanxia2.0/backend/app/services/pl_plan_stop_rules.py

"""运输计划多段：内部客户地磅默认值与数量平衡校验"""
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)})不一致，请核对"
    ]