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

"""合同产品在运输计划上的安排数量汇总。"""
from __future__ import annotations

from collections import defaultdict
from typing import Any, DefaultDict, Dict, Iterable, Optional, Set, Tuple

from sqlalchemy.orm import Session

from app.models.pl_models import (
    PlPlan,
    PlPlanContract,
    PlPlanDelivery,
    PlPlanDeliveryProduct,
    PlPlanLoad,
    PlPlanLoadProduct,
)
from app.services.pl_plan_service import (
    PLAN_CONTRACT_SIDE_DEMANDER,
    PLAN_CONTRACT_SIDE_SUPPLIER,
)
from app.services.pl_plan_stop_service import (
    PLAN_CONTRACT_SIDE_DELIVERY,
    PLAN_CONTRACT_SIDE_LOAD,
)

# 待安排 / 等通知 / 作废：不计入「已安排」（见 _bucket_for_plan）

ProductPackageKey = Tuple[str, str]


def normalize_package_spec(value: Any) -> str:
    if value is None:
        return ""
    return str(value).strip()


def product_package_key(product_id: Any, package_spec: Any = None) -> Optional[ProductPackageKey]:
    pid = str(product_id or "").strip()
    if not pid:
        return None
    return (pid, normalize_package_spec(package_spec))


def match_contract_product_for_key(
    products: Iterable[Any],
    key: Optional[ProductPackageKey],
) -> Optional[Any]:
    """按复合键匹配合同产品行；同键多行时 quantity 相加（防御）。"""
    if not key:
        return None
    matched = []
    for p in products or []:
        k = product_package_key(
            getattr(p, "product_id", None),
            getattr(p, "package_spec", None),
        )
        if k == key:
            matched.append(p)
    if not matched:
        return None
    if len(matched) == 1:
        return matched[0]
    from types import SimpleNamespace

    total_qty = sum(_to_float(getattr(p, "quantity", None)) for p in matched)
    base = matched[0]
    return SimpleNamespace(
        product_id=getattr(base, "product_id", None),
        package_spec=getattr(base, "package_spec", None),
        product_name=getattr(base, "product_name", None),
        model=getattr(base, "model", None),
        unit=getattr(base, "unit", None),
        quantity=total_qty,
        quantity_mode=getattr(base, "quantity_mode", None),
    )


def _to_float(value: Any) -> float:
    if value is None:
        return 0.0
    try:
        return float(value)
    except (TypeError, ValueError):
        return 0.0


def _bucket_for_plan(*, archive_status: Optional[str], follow_status: Optional[str], transport_status: Optional[str]) -> Optional[str]:
    """返回 'archived' | 'arranged_unarchived' | None（不计入）。"""
    follow = (follow_status or "").strip()
    if follow == "cancelled" or follow in ("wait_notice", "pending_arrange"):
        return None
    archive = (archive_status or "").strip() or "unarchived"
    if archive == "archived":
        return "archived"
    # 已安排未归档（含 archive draft 暂存）
    transport = (transport_status or "").strip()
    if transport == "arranged" or follow in ("arranged", "completed"):
        return "arranged_unarchived"
    return None


def merge_product_qty_maps(
    *,
    archived: Dict[ProductPackageKey, float],
    arranged_unarchived: Dict[ProductPackageKey, float],
) -> Dict[ProductPackageKey, Dict[str, float]]:
    """(product_id, package_spec) -> {archivedQty, arrangedUnarchivedQty, arrangedTotalQty}"""
    keys = set(archived) | set(arranged_unarchived)
    out: Dict[ProductPackageKey, Dict[str, float]] = {}
    for key in keys:
        a = round(float(archived.get(key, 0.0)), 3)
        u = round(float(arranged_unarchived.get(key, 0.0)), 3)
        out[key] = {
            "archivedQty": a,
            "arrangedUnarchivedQty": u,
            "arrangedTotalQty": round(a + u, 3),
        }
    return out


# plan：计划数量；archive_actual：装货 inbound / 送货 outbound，null 回退 plan_qty
QTY_MODE_PLAN = "plan"
QTY_MODE_ARCHIVE_ACTUAL = "archive_actual"


def _resolve_stop_product_qty(
    row: Any,
    *,
    stop_kind: str,
    qty_mode: str,
) -> float:
    """按模式解析段产品计入量。"""
    if qty_mode == QTY_MODE_ARCHIVE_ACTUAL:
        attr = "inbound_qty" if stop_kind == "load" else "outbound_qty"
        raw = getattr(row, attr, None)
        if raw is not None:
            return _to_float(raw)
        return _to_float(getattr(row, "plan_qty", None))
    return _to_float(getattr(row, "plan_qty", None))


def accumulate_stop_products(
    target: DefaultDict[ProductPackageKey, float],
    products: Iterable[Any],
    *,
    stop_kind: str = "load",
    qty_mode: str = QTY_MODE_PLAN,
) -> None:
    for row in products or []:
        key = product_package_key(
            getattr(row, "product_id", None),
            getattr(row, "package_spec", None),
        )
        if not key:
            continue
        qty = _resolve_stop_product_qty(row, stop_kind=stop_kind, qty_mode=qty_mode)
        if qty:
            target[key] += qty


def resolve_linked_plan_scopes(
    link_rows: Iterable[PlPlanContract],
    legacy_plan_ids: Iterable[str],
) -> Dict[str, Dict[str, Any]]:
    """
    plan_id -> {
      load_ids: set[str] | None,   # None = 该侧全部装货段
      delivery_ids: set[str] | None,
      use_all_stops: bool,         # 遗留主表合同：优先送货段，否则装货段，各产品只计一次
    }
    """
    scopes: Dict[str, Dict[str, Any]] = {}

    def _ensure(plan_id: str) -> Dict[str, Any]:
        if plan_id not in scopes:
            scopes[plan_id] = {
                "load_ids": set(),
                "delivery_ids": set(),
                "use_all_stops": False,
                "has_explicit": False,
            }
        return scopes[plan_id]

    for row in link_rows or []:
        plan_id = str(getattr(row, "plan_id", None) or "").strip()
        if not plan_id:
            continue
        scope = _ensure(plan_id)
        side = (getattr(row, "side", None) or "").strip()
        stop_id = str(getattr(row, "stop_id", None) or "").strip()
        scope["has_explicit"] = True
        if side in (PLAN_CONTRACT_SIDE_LOAD, PLAN_CONTRACT_SIDE_SUPPLIER):
            if stop_id:
                scope["load_ids"].add(stop_id)
            else:
                scope["load_ids"] = None  # type: ignore[assignment]
        elif side in (PLAN_CONTRACT_SIDE_DELIVERY, PLAN_CONTRACT_SIDE_DEMANDER):
            if stop_id:
                scope["delivery_ids"].add(stop_id)
            else:
                scope["delivery_ids"] = None  # type: ignore[assignment]
        elif stop_id:
            # 未知 side 但有 stop：两边都尝试匹配时由调用方按 id 过滤
            scope["load_ids"].add(stop_id)
            scope["delivery_ids"].add(stop_id)

    for plan_id in legacy_plan_ids or []:
        pid = str(plan_id or "").strip()
        if not pid:
            continue
        scope = _ensure(pid)
        if not scope["has_explicit"]:
            scope["use_all_stops"] = True

    return scopes


def sum_qtys_for_scope(
    *,
    load_products_by_load: Dict[str, list],
    delivery_products_by_delivery: Dict[str, list],
    all_load_ids: Set[str],
    all_delivery_ids: Set[str],
    scope: Dict[str, Any],
    qty_mode: str = QTY_MODE_PLAN,
) -> Dict[ProductPackageKey, float]:
    acc: DefaultDict[ProductPackageKey, float] = defaultdict(float)

    if scope.get("use_all_stops"):
        # 遗留：优先送货段数量；无送货则用装货，避免装+送双计
        if all_delivery_ids:
            for did in all_delivery_ids:
                accumulate_stop_products(
                    acc,
                    delivery_products_by_delivery.get(did, []),
                    stop_kind="delivery",
                    qty_mode=qty_mode,
                )
        else:
            for lid in all_load_ids:
                accumulate_stop_products(
                    acc,
                    load_products_by_load.get(lid, []),
                    stop_kind="load",
                    qty_mode=qty_mode,
                )
        return dict(acc)

    load_ids = scope.get("load_ids")
    delivery_ids = scope.get("delivery_ids")

    if load_ids is None:
        for lid in all_load_ids:
            accumulate_stop_products(
                acc,
                load_products_by_load.get(lid, []),
                stop_kind="load",
                qty_mode=qty_mode,
            )
    elif load_ids:
        for lid in load_ids:
            accumulate_stop_products(
                acc,
                load_products_by_load.get(lid, []),
                stop_kind="load",
                qty_mode=qty_mode,
            )

    if delivery_ids is None:
        for did in all_delivery_ids:
            accumulate_stop_products(
                acc,
                delivery_products_by_delivery.get(did, []),
                stop_kind="delivery",
                qty_mode=qty_mode,
            )
    elif delivery_ids:
        for did in delivery_ids:
            accumulate_stop_products(
                acc,
                delivery_products_by_delivery.get(did, []),
                stop_kind="delivery",
                qty_mode=qty_mode,
            )

    return dict(acc)


def aggregate_contract_product_plan_qtys(
    db: Session,
    *,
    tenant_no: str,
    contract_id: str,
) -> Dict[ProductPackageKey, Dict[str, float]]:
    """返回 (productId, packageSpec) -> 三数量字段。"""
    cid = (contract_id or "").strip()
    if not cid:
        return {}

    link_rows = (
        db.query(PlPlanContract)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.contract_id == cid,
        )
        .all()
    )
    link_plan_ids = {str(r.plan_id) for r in link_rows if r.plan_id}

    legacy_plans = (
        db.query(PlPlan)
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.contract_id == cid,
        )
        .all()
    )
    legacy_only_ids = {p.id for p in legacy_plans if p.id not in link_plan_ids}

    scopes = resolve_linked_plan_scopes(link_rows, legacy_only_ids)
    plan_ids = list(scopes.keys())
    if not plan_ids:
        return {}

    plans = (
        db.query(PlPlan)
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.id.in_(plan_ids),
            PlPlan.follow_status != "cancelled",
        )
        .all()
    )
    plan_by_id = {p.id: p for p in plans}

    loads = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
            PlPlanLoad.plan_id.in_(plan_ids),
        )
        .all()
    )
    deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
            PlPlanDelivery.plan_id.in_(plan_ids),
        )
        .all()
    )
    load_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
    delivery_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
    for row in loads:
        load_ids_by_plan[row.plan_id].add(row.id)
    for row in deliveries:
        delivery_ids_by_plan[row.plan_id].add(row.id)

    all_load_ids = {row.id for row in loads}
    all_delivery_ids = {row.id for row in deliveries}

    load_products_by_load: DefaultDict[str, list] = defaultdict(list)
    if all_load_ids:
        for prod in (
            db.query(PlPlanLoadProduct)
            .filter(
                PlPlanLoadProduct.tenant_no == tenant_no,
                PlPlanLoadProduct.del_flag == 0,
                PlPlanLoadProduct.load_id.in_(all_load_ids),
            )
            .all()
        ):
            load_products_by_load[prod.load_id].append(prod)

    delivery_products_by_delivery: DefaultDict[str, list] = defaultdict(list)
    if all_delivery_ids:
        for prod in (
            db.query(PlPlanDeliveryProduct)
            .filter(
                PlPlanDeliveryProduct.tenant_no == tenant_no,
                PlPlanDeliveryProduct.del_flag == 0,
                PlPlanDeliveryProduct.delivery_id.in_(all_delivery_ids),
            )
            .all()
        ):
            delivery_products_by_delivery[prod.delivery_id].append(prod)

    archived: DefaultDict[ProductPackageKey, float] = defaultdict(float)
    arranged_unarchived: DefaultDict[ProductPackageKey, float] = defaultdict(float)

    for plan_id, scope in scopes.items():
        plan = plan_by_id.get(plan_id)
        if not plan:
            continue
        bucket = _bucket_for_plan(
            archive_status=plan.archive_status,
            follow_status=plan.follow_status,
            transport_status=plan.transport_status,
        )
        if not bucket:
            continue
        qty_mode = (
            QTY_MODE_ARCHIVE_ACTUAL if bucket == "archived" else QTY_MODE_PLAN
        )
        qtys = sum_qtys_for_scope(
            load_products_by_load=load_products_by_load,
            delivery_products_by_delivery=delivery_products_by_delivery,
            all_load_ids=load_ids_by_plan.get(plan_id, set()),
            all_delivery_ids=delivery_ids_by_plan.get(plan_id, set()),
            scope=scope,
            qty_mode=qty_mode,
        )
        target = archived if bucket == "archived" else arranged_unarchived
        for pid, qty in qtys.items():
            target[pid] += qty

    return merge_product_qty_maps(archived=dict(archived), arranged_unarchived=dict(arranged_unarchived))


def attach_plan_qty_fields_to_products(
    products: list,
    qty_by_product_id: Dict[ProductPackageKey, Dict[str, float]],
) -> None:
    for item in products or []:
        if not isinstance(item, dict):
            continue
        key = product_package_key(
            item.get("productId"),
            item.get("packageSpec") or item.get("package_spec"),
        )
        stats = (qty_by_product_id.get(key) if key else None) or {
            "archivedQty": 0.0,
            "arrangedUnarchivedQty": 0.0,
            "arrangedTotalQty": 0.0,
        }
        item["archivedQty"] = stats["archivedQty"]
        item["arrangedUnarchivedQty"] = stats["arrangedUnarchivedQty"]
        item["arrangedTotalQty"] = stats["arrangedTotalQty"]


def plan_ids_linked_to_contract(
    db: Session,
    *,
    tenant_no: str,
    contract_id: str,
) -> Set[str]:
    cid = (contract_id or "").strip()
    if not cid:
        return set()
    ids: Set[str] = set()
    for (plan_id,) in (
        db.query(PlPlanContract.plan_id)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.contract_id == cid,
        )
        .distinct()
        .all()
    ):
        if plan_id:
            ids.add(str(plan_id))
    for (plan_id,) in (
        db.query(PlPlan.id)
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.contract_id == cid,
        )
        .all()
    ):
        if plan_id:
            ids.add(str(plan_id))
    return ids


def contract_ids_linked_to_plan(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
) -> Set[str]:
    ids: Set[str] = set()
    if plan.contract_id:
        ids.add(str(plan.contract_id).strip())
    for (cid,) in (
        db.query(PlPlanContract.contract_id)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.plan_id == plan.id,
        )
        .distinct()
        .all()
    ):
        if cid:
            ids.add(str(cid).strip())
    return {x for x in ids if x}


def auto_start_execute_approved_contracts(
    db: Session,
    *,
    tenant_no: str,
    plan: PlPlan,
    user_id: Optional[str] = None,
) -> list[str]:
    """将关联且状态为 approved 的合同置为 executing（不 commit）。返回已推进的合同 id。"""
    from datetime import datetime

    from app.models.crm_models import CrmContract

    started: list[str] = []
    for cid in contract_ids_linked_to_plan(db, tenant_no=tenant_no, plan=plan):
        row = (
            db.query(CrmContract)
            .filter(
                CrmContract.id == cid,
                CrmContract.tenant_no == tenant_no,
                CrmContract.del_flag == 0,
            )
            .first()
        )
        if not row or row.status != "approved":
            continue
        row.status = "executing"
        row.update_by = user_id
        row.update_time = datetime.now()
        started.append(cid)
    return started


# =============================================================================
# 安排运输：合同数量硬拦 / 成功后提醒
# =============================================================================

FRAMEWORK_CONTRACT_TYPES = frozenset({"framework", "framework_sale", "framework_purchase"})


def effective_arranged_total(
    *,
    arranged_total: float,
    old_plan_qty: float,
    new_plan_qty: float,
) -> float:
    """已安排合计 − 本计划旧量 + 本计划新量。"""
    return round(
        _to_float(arranged_total) - _to_float(old_plan_qty) + _to_float(new_plan_qty),
        3,
    )


def contract_product_has_fixed_qty(
    *,
    contract_type: Optional[str],
    quantity: Any,
    quantity_mode: Optional[str],
) -> bool:
    """合同产品行是否按固定数量做超量硬拦。"""
    ctype = (contract_type or "").strip()
    mode = (quantity_mode or "").strip() or "fixed"
    if ctype in FRAMEWORK_CONTRACT_TYPES and mode == "customer_demand":
        return False
    qty = _to_float(quantity)
    return qty > 0


def evaluate_product_qty_for_arrange(
    *,
    has_fixed_qty: bool,
    contract_qty: Optional[float],
    arranged_total: float,
    old_plan_qty: float,
    new_plan_qty: float,
) -> Optional[Dict[str, Any]]:
    """
    返回:
      - None: 跳过（本次无量）
      - kind=violation | remaining | framework_cumulative
    """
    new_qty = _to_float(new_plan_qty)
    if new_qty <= 0:
        return None

    old_qty = _to_float(old_plan_qty)
    arranged = _to_float(arranged_total)
    effective = effective_arranged_total(
        arranged_total=arranged,
        old_plan_qty=old_qty,
        new_plan_qty=new_qty,
    )

    if not has_fixed_qty:
        return {
            "kind": "framework_cumulative",
            "arrangedTotalQty": effective,
            "thisPlanQty": round(new_qty, 3),
            "effectiveTotal": effective,
        }

    cq = _to_float(contract_qty)
    overflow = round(effective - cq, 3)
    base = {
        "contractQty": round(cq, 3),
        "arrangedTotalQty": effective,
        "thisPlanQty": round(new_qty, 3),
        "oldPlanQty": round(old_qty, 3),
        "newPlanQty": round(new_qty, 3),
        "effectiveTotal": effective,
    }
    if overflow > 0:
        return {**base, "kind": "violation", "overflowQty": overflow}
    return {
        **base,
        "kind": "remaining",
        "remainingQty": round(cq - effective, 3),
    }


CONTRACT_QTY_VIOLATION_CODE = "CONTRACT_QTY_OVERFLOW"


def _stop_kind_from_scope(scope: Dict[str, Any]) -> str:
    load_ids = scope.get("load_ids")
    delivery_ids = scope.get("delivery_ids")
    has_load = load_ids is None or bool(load_ids)
    has_delivery = delivery_ids is None or bool(delivery_ids)
    if scope.get("use_all_stops"):
        return "delivery"
    if has_load and not has_delivery:
        return "load"
    if has_delivery and not has_load:
        return "delivery"
    if has_load:
        return "load"
    return "delivery"


def _plan_scope_for_contract(
    *,
    plan_id: str,
    contract_id: str,
    link_rows: list[PlPlanContract],
    legacy_contract_id: Optional[str],
) -> Dict[str, Any]:
    cid = (contract_id or "").strip()
    links = [r for r in link_rows if str(r.contract_id or "").strip() == cid]
    legacy_ids: list[str] = []
    if not links and legacy_contract_id and str(legacy_contract_id).strip() == cid:
        legacy_ids = [plan_id]
    scopes = resolve_linked_plan_scopes(links, legacy_ids)
    return scopes.get(plan_id) or {
        "load_ids": set(),
        "delivery_ids": set(),
        "use_all_stops": False,
        "has_explicit": False,
    }


def _load_plan_stop_product_maps(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
) -> tuple[Set[str], Set[str], Dict[str, list], Dict[str, list]]:
    loads = (
        db.query(PlPlanLoad)
        .filter(
            PlPlanLoad.tenant_no == tenant_no,
            PlPlanLoad.del_flag == 0,
            PlPlanLoad.plan_id == plan_id,
        )
        .all()
    )
    deliveries = (
        db.query(PlPlanDelivery)
        .filter(
            PlPlanDelivery.tenant_no == tenant_no,
            PlPlanDelivery.del_flag == 0,
            PlPlanDelivery.plan_id == plan_id,
        )
        .all()
    )
    load_ids = {row.id for row in loads}
    delivery_ids = {row.id for row in deliveries}
    load_products_by_load: DefaultDict[str, list] = defaultdict(list)
    if load_ids:
        for prod in (
            db.query(PlPlanLoadProduct)
            .filter(
                PlPlanLoadProduct.tenant_no == tenant_no,
                PlPlanLoadProduct.del_flag == 0,
                PlPlanLoadProduct.load_id.in_(load_ids),
            )
            .all()
        ):
            load_products_by_load[prod.load_id].append(prod)
    delivery_products_by_delivery: DefaultDict[str, list] = defaultdict(list)
    if delivery_ids:
        for prod in (
            db.query(PlPlanDeliveryProduct)
            .filter(
                PlPlanDeliveryProduct.tenant_no == tenant_no,
                PlPlanDeliveryProduct.del_flag == 0,
                PlPlanDeliveryProduct.delivery_id.in_(delivery_ids),
            )
            .all()
        ):
            delivery_products_by_delivery[prod.delivery_id].append(prod)
    return load_ids, delivery_ids, dict(load_products_by_load), dict(delivery_products_by_delivery)


def collect_plan_contract_qty_checks(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
) -> tuple[list[Dict[str, Any]], list[Dict[str, Any]]]:
    """
    返回 (violations, reminders)。
    在写运输前调用；reminders 可缓存到成功响应，勿在状态变更后再算。
    """
    from app.models.crm_models import CrmContract, CrmContractProduct

    plan = (
        db.query(PlPlan)
        .filter(
            PlPlan.id == plan_id,
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
        )
        .first()
    )
    if not plan:
        return [], []

    link_rows = (
        db.query(PlPlanContract)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.plan_id == plan_id,
        )
        .all()
    )
    contract_ids: Set[str] = set()
    for row in link_rows:
        cid = str(row.contract_id or "").strip()
        if cid:
            contract_ids.add(cid)
    legacy_cid = str(plan.contract_id or "").strip()
    if legacy_cid:
        contract_ids.add(legacy_cid)
    if not contract_ids:
        return [], []

    load_ids, delivery_ids, load_products_by_load, delivery_products_by_delivery = (
        _load_plan_stop_product_maps(db, tenant_no=tenant_no, plan_id=plan_id)
    )
    in_bucket = (
        _bucket_for_plan(
            archive_status=plan.archive_status,
            follow_status=plan.follow_status,
            transport_status=plan.transport_status,
        )
        is not None
    )

    violations: list[Dict[str, Any]] = []
    reminders: list[Dict[str, Any]] = []

    for cid in sorted(contract_ids):
        contract = (
            db.query(CrmContract)
            .filter(
                CrmContract.id == cid,
                CrmContract.tenant_no == tenant_no,
                CrmContract.del_flag == 0,
            )
            .first()
        )
        if not contract:
            continue

        scope = _plan_scope_for_contract(
            plan_id=plan_id,
            contract_id=cid,
            link_rows=link_rows,
            legacy_contract_id=legacy_cid,
        )

        plan_qtys = sum_qtys_for_scope(
            load_products_by_load=load_products_by_load,
            delivery_products_by_delivery=delivery_products_by_delivery,
            all_load_ids=load_ids,
            all_delivery_ids=delivery_ids,
            scope=scope,
        )
        if not plan_qtys:
            continue

        qty_map = aggregate_contract_product_plan_qtys(
            db, tenant_no=tenant_no, contract_id=cid
        )
        products = (
            db.query(CrmContractProduct)
            .filter(
                CrmContractProduct.contract_id == cid,
                CrmContractProduct.tenant_no == tenant_no,
                CrmContractProduct.del_flag == 0,
            )
            .all()
        )
        stop_kind = _stop_kind_from_scope(scope)
        contract_type = (contract.contract_type or "").strip()
        contract_no = contract.contract_no or ""
        customer_name = (contract.customer_name or "").strip()

        for key, new_qty_raw in plan_qtys.items():
            new_qty = _to_float(new_qty_raw)
            if new_qty <= 0:
                continue
            if not isinstance(key, tuple) or len(key) != 2:
                continue
            pid, pkg = key[0], key[1]
            cp = match_contract_product_for_key(products, key)
            # 计划产品不在合同明细：不校验数量（归属另有规则）
            if not cp:
                continue

            arranged_total = _to_float((qty_map.get(key) or {}).get("arrangedTotalQty"))
            if in_bucket:
                old_qty = new_qty
                # arranged 已含当前库内量；effective = arranged - old + new = arranged
            else:
                old_qty = 0.0

            has_fixed = contract_product_has_fixed_qty(
                contract_type=contract_type,
                quantity=cp.quantity,
                quantity_mode=cp.quantity_mode,
            )
            evaluated = evaluate_product_qty_for_arrange(
                has_fixed_qty=has_fixed,
                contract_qty=_to_float(cp.quantity) if has_fixed else None,
                arranged_total=arranged_total,
                old_plan_qty=old_qty,
                new_plan_qty=new_qty,
            )
            if not evaluated:
                continue

            item = {
                "contractId": cid,
                "contractNo": contract_no,
                "contractType": contract_type,
                "customerName": customer_name,
                "stopKind": stop_kind,
                "productId": pid,
                "packageSpec": pkg or None,
                "productName": cp.product_name or "",
                "model": (cp.model or "").strip() or None,
                "unit": (cp.unit or "").strip() or None,
                "thisPlanQty": evaluated.get("thisPlanQty", round(new_qty, 3)),
                "arrangedTotalQty": evaluated["arrangedTotalQty"],
            }
            kind = evaluated["kind"]
            if kind == "violation":
                violations.append(
                    {
                        **item,
                        "contractQty": evaluated["contractQty"],
                        "oldPlanQty": evaluated["oldPlanQty"],
                        "newPlanQty": evaluated["newPlanQty"],
                        "effectiveTotal": evaluated["effectiveTotal"],
                        "overflowQty": evaluated["overflowQty"],
                        # 与提醒表同列：剩余可为负，表示超量
                        "remainingQty": round(
                            evaluated["contractQty"] - evaluated["effectiveTotal"], 3
                        ),
                    }
                )
            elif kind == "remaining":
                reminders.append(
                    {
                        **item,
                        "mode": "remaining",
                        "contractQty": evaluated["contractQty"],
                        "remainingQty": evaluated["remainingQty"],
                    }
                )
            elif kind == "framework_cumulative":
                reminders.append(
                    {
                        **item,
                        "mode": "framework_cumulative",
                    }
                )

    return violations, reminders


def assert_contract_qty_for_arrange(
    db: Session,
    *,
    tenant_no: str,
    plan_id: str,
) -> list[Dict[str, Any]]:
    """超限则 422；通过则返回 reminders（供成功响应复用）。"""
    from fastapi import HTTPException

    violations, reminders = collect_plan_contract_qty_checks(
        db, tenant_no=tenant_no, plan_id=plan_id
    )
    if violations:
        raise HTTPException(
            status_code=422,
            detail={
                "code": CONTRACT_QTY_VIOLATION_CODE,
                "message": "合同产品安排数量超过合同数量",
                "items": violations,
            },
        )
    return reminders


def build_contract_qty_reminder_payload(
    reminders: list[Dict[str, Any]],
) -> Optional[Dict[str, Any]]:
    if not reminders:
        return None
    return {"items": reminders}


def validate_stop_products_package_against_contract(
    *,
    contract_products: Iterable[Any],
    stop_products: Iterable[Any],
) -> tuple[bool, Optional[str]]:
    """
    绑合同时：计划段产品包装必填，且必须属于本合同该产品的包装集合。
    返回 (ok, error_message)。
    """
    allowed_by_pid: DefaultDict[str, Set[str]] = defaultdict(set)
    contract_pids: Set[str] = set()
    for row in contract_products or []:
        pid = str(getattr(row, "product_id", None) or "").strip()
        if not pid:
            continue
        contract_pids.add(pid)
        pkg = normalize_package_spec(getattr(row, "package_spec", None))
        if pkg:
            allowed_by_pid[pid].add(pkg)

    for row in stop_products or []:
        pid = str(getattr(row, "product_id", None) or "").strip()
        if not pid or pid not in contract_pids:
            continue
        allowed = allowed_by_pid.get(pid) or set()
        if not allowed:
            continue
        pkg = normalize_package_spec(getattr(row, "package_spec", None))
        name = (
            str(getattr(row, "product_name", None) or "").strip()
            or pid
        )
        if not pkg:
            return False, f"请选择包装规格：{name}"
        if pkg not in allowed:
            return False, f"包装规格不在合同明细中：{name}（{pkg}）"
    return True, None


def get_contract_product_arrange_qty_detail(
    db: Session,
    *,
    tenant_no: str,
    contract_id: str,
    product_id: str,
    package_spec: Optional[str] = None,
    exclude_plan_id: Optional[str] = None,
) -> Dict[str, Any]:
    """
    表单「可安排数量」与问号弹窗数据。
    arrangedQty / availableQty 均不含 exclude_plan_id（若该计划已在已安排桶中）。
    """
    from app.models.crm_models import CrmContract, CrmContractProduct

    cid = (contract_id or "").strip()
    pid = (product_id or "").strip()
    pkg = normalize_package_spec(package_spec)
    key = product_package_key(pid, pkg)
    exclude_id = (exclude_plan_id or "").strip() or None
    empty = {
        "contractId": cid,
        "contractNo": None,
        "customerName": None,
        "productId": pid,
        "packageSpec": pkg or None,
        "productName": None,
        "hasFixedQty": False,
        "contractQty": None,
        "arrangedQty": 0.0,
        "availableQty": None,
        "plans": [],
    }
    if not cid or not pid or not key:
        return empty

    contract = (
        db.query(CrmContract)
        .filter(
            CrmContract.id == cid,
            CrmContract.tenant_no == tenant_no,
            CrmContract.del_flag == 0,
        )
        .first()
    )
    if not contract:
        return empty

    candidates = (
        db.query(CrmContractProduct)
        .filter(
            CrmContractProduct.contract_id == cid,
            CrmContractProduct.product_id == pid,
            CrmContractProduct.tenant_no == tenant_no,
            CrmContractProduct.del_flag == 0,
        )
        .all()
    )
    cp = match_contract_product_for_key(candidates, key)
    contract_type = (contract.contract_type or "").strip()
    has_fixed = False
    contract_qty: Optional[float] = None
    if cp:
        has_fixed = contract_product_has_fixed_qty(
            contract_type=contract_type,
            quantity=cp.quantity,
            quantity_mode=cp.quantity_mode,
        )
        if has_fixed:
            contract_qty = round(_to_float(cp.quantity), 3)

    qty_map = aggregate_contract_product_plan_qtys(
        db, tenant_no=tenant_no, contract_id=cid
    )
    arranged_including = _to_float((qty_map.get(key) or {}).get("arrangedTotalQty"))

    link_rows = (
        db.query(PlPlanContract)
        .filter(
            PlPlanContract.tenant_no == tenant_no,
            PlPlanContract.del_flag == 0,
            PlPlanContract.contract_id == cid,
        )
        .all()
    )
    link_plan_ids = {str(r.plan_id) for r in link_rows if r.plan_id}
    legacy_plans = (
        db.query(PlPlan)
        .filter(
            PlPlan.tenant_no == tenant_no,
            PlPlan.del_flag == 0,
            PlPlan.contract_id == cid,
        )
        .all()
    )
    legacy_only_ids = {p.id for p in legacy_plans if p.id not in link_plan_ids}
    scopes = resolve_linked_plan_scopes(link_rows, legacy_only_ids)
    plan_ids = list(scopes.keys())

    exclude_qty = 0.0
    plans_out: list[Dict[str, Any]] = []

    if plan_ids:
        plans = (
            db.query(PlPlan)
            .filter(
                PlPlan.tenant_no == tenant_no,
                PlPlan.del_flag == 0,
                PlPlan.id.in_(plan_ids),
                PlPlan.follow_status != "cancelled",
            )
            .all()
        )
        plan_by_id = {p.id: p for p in plans}

        loads = (
            db.query(PlPlanLoad)
            .filter(
                PlPlanLoad.tenant_no == tenant_no,
                PlPlanLoad.del_flag == 0,
                PlPlanLoad.plan_id.in_(plan_ids),
            )
            .all()
        )
        deliveries = (
            db.query(PlPlanDelivery)
            .filter(
                PlPlanDelivery.tenant_no == tenant_no,
                PlPlanDelivery.del_flag == 0,
                PlPlanDelivery.plan_id.in_(plan_ids),
            )
            .all()
        )
        load_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
        delivery_ids_by_plan: DefaultDict[str, Set[str]] = defaultdict(set)
        for row in loads:
            load_ids_by_plan[row.plan_id].add(row.id)
        for row in deliveries:
            delivery_ids_by_plan[row.plan_id].add(row.id)

        all_load_ids = {row.id for row in loads}
        all_delivery_ids = {row.id for row in deliveries}
        load_products_by_load: DefaultDict[str, list] = defaultdict(list)
        if all_load_ids:
            for prod in (
                db.query(PlPlanLoadProduct)
                .filter(
                    PlPlanLoadProduct.tenant_no == tenant_no,
                    PlPlanLoadProduct.del_flag == 0,
                    PlPlanLoadProduct.load_id.in_(all_load_ids),
                )
                .all()
            ):
                load_products_by_load[prod.load_id].append(prod)
        delivery_products_by_delivery: DefaultDict[str, list] = defaultdict(list)
        if all_delivery_ids:
            for prod in (
                db.query(PlPlanDeliveryProduct)
                .filter(
                    PlPlanDeliveryProduct.tenant_no == tenant_no,
                    PlPlanDeliveryProduct.del_flag == 0,
                    PlPlanDeliveryProduct.delivery_id.in_(all_delivery_ids),
                )
                .all()
            ):
                delivery_products_by_delivery[prod.delivery_id].append(prod)

        for plan_id, scope in scopes.items():
            plan = plan_by_id.get(plan_id)
            if not plan:
                continue
            bucket = _bucket_for_plan(
                archive_status=plan.archive_status,
                follow_status=plan.follow_status,
                transport_status=plan.transport_status,
            )
            if not bucket:
                continue
            qty_mode = (
                QTY_MODE_ARCHIVE_ACTUAL if bucket == "archived" else QTY_MODE_PLAN
            )
            qtys = sum_qtys_for_scope(
                load_products_by_load=load_products_by_load,
                delivery_products_by_delivery=delivery_products_by_delivery,
                all_load_ids=load_ids_by_plan.get(plan_id, set()),
                all_delivery_ids=delivery_ids_by_plan.get(plan_id, set()),
                scope=scope,
                qty_mode=qty_mode,
            )
            pq = _to_float(qtys.get(key))
            if pq <= 0:
                continue
            if exclude_id and str(plan_id) == exclude_id:
                exclude_qty = pq
                continue
            plan_date = getattr(plan, "plan_date", None)
            plans_out.append(
                {
                    "planId": plan_id,
                    "planNo": plan.plan_no or "",
                    "planDate": plan_date.isoformat() if plan_date else None,
                    "planQty": round(pq, 3),
                    "planType": getattr(plan, "plan_type", None),
                    "belongsCompany": getattr(plan, "belongs_company", None),
                    "supplierName": getattr(plan, "supplier_name", None),
                    "customerName": getattr(plan, "customer_name", None),
                    "licensePlate": getattr(plan, "license_plate", None),
                    "driverName": getattr(plan, "driver_name", None),
                    "followUserName": getattr(plan, "follow_user_name", None),
                    "followStatus": plan.follow_status,
                    "transportStatus": plan.transport_status,
                    "archiveStatus": plan.archive_status,
                    "createTime": plan.create_time.isoformat(sep=" ", timespec="seconds")
                    if getattr(plan, "create_time", None)
                    else None,
                }
            )

        plans_out.sort(key=lambda x: (x.get("planDate") or "", x.get("planNo") or ""), reverse=True)

    arranged_qty = round(max(0.0, arranged_including - exclude_qty), 3)
    available_qty = None
    if has_fixed and contract_qty is not None:
        available_qty = round(contract_qty - arranged_qty, 3)

    return {
        "contractId": cid,
        "contractNo": contract.contract_no or "",
        "customerName": (contract.customer_name or "").strip() or None,
        "productId": pid,
        "packageSpec": pkg or None,
        "productName": (cp.product_name if cp else None) or None,
        "hasFixedQty": has_fixed,
        "contractQty": contract_qty,
        "arrangedQty": arranged_qty,
        "availableQty": available_qty,
        "plans": plans_out,
    }