feat(pl): hard-block over-limit contract qty on arrange and remind after success

Validate arranged+this plan qty against fixed contract product quantities on
transport arrange; return post-success remaining/framework cumulative reminders.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/api/pl/transport.py

@@ -492,6 +492,7 @@ def _build_arrange_result(
     site_records: list[TransportSiteLinkRead],
     inventory_records: list[TransportInventoryLinkRead],
     io_order: Optional[TransportIoOrderBrief] = None,
+    contract_qty_reminder: Optional[dict] = None,
 ) -> TransportArrangeResult:
     token = transport.driver_access_token or ""
     result = TransportArrangeResult.model_validate(transport)
@@ -500,6 +501,10 @@ def _build_arrange_result(
     result.site_records = site_records
     result.inventory_records = inventory_records
     result.io_order = io_order
+    if contract_qty_reminder:
+        from app.schemas.pl.transport import ContractQtyReminder
+
+        result.contract_qty_reminder = ContractQtyReminder.model_validate(contract_qty_reminder)
     return result
 
 
@@ -784,6 +789,16 @@ def arrange_transport(
 
     validate_hazard_cert_ready_for_transport(db, tenant_no, plan, products)
 
+    from app.services.pl_contract_plan_qty import (
+        assert_contract_qty_for_arrange,
+        build_contract_qty_reminder_payload,
+    )
+
+    contract_qty_reminders = assert_contract_qty_for_arrange(
+        db, tenant_no=tenant_no, plan_id=plan_id
+    )
+    contract_qty_reminder_payload = build_contract_qty_reminder_payload(contract_qty_reminders)
+
     # 计算运费
     freight_amount = payload.freight_amount
     if payload.freight_type == "by_weight" and payload.freight_price:
@@ -884,6 +899,13 @@ def arrange_transport(
     plan.freight_amount = freight_amount or 0
     plan.update_by = user.id
 
+    # 安排运输时：关联且仍为「已审批」的合同自动开始执行
+    from app.services.pl_contract_plan_qty import auto_start_execute_approved_contracts
+
+    auto_start_execute_approved_contracts(
+        db, tenant_no=tenant_no, plan=plan, user_id=user.id
+    )
+
     needs_site_link = plan_requires_site_link(db, tenant_no=tenant_no, plan=plan)
 
     if is_rearrange:
@@ -1030,7 +1052,13 @@ def arrange_transport(
 
     db.commit()
     db.refresh(transport)
-    return _build_arrange_result(transport, site_records, inventory_records, io_order_brief)
+    return _build_arrange_result(
+        transport,
+        site_records,
+        inventory_records,
+        io_order_brief,
+        contract_qty_reminder=contract_qty_reminder_payload,
+    )
 
 
 # =============================================================================

backend/app/schemas/pl/transport.py

@@ -217,6 +217,26 @@ class TransportIoOrderBrief(BaseModel):
     model_config = {"populate_by_name": True}
 
 
+class ContractQtyReminderItem(BaseModel):
+    contract_id: str = Field(..., alias="contractId")
+    contract_no: Optional[str] = Field(default=None, alias="contractNo")
+    contract_type: Optional[str] = Field(default=None, alias="contractType")
+    stop_kind: str = Field(..., alias="stopKind")
+    product_id: str = Field(..., alias="productId")
+    product_name: Optional[str] = Field(default=None, alias="productName")
+    mode: str
+    contract_qty: Optional[float] = Field(default=None, alias="contractQty")
+    arranged_total_qty: float = Field(..., alias="arrangedTotalQty")
+    remaining_qty: Optional[float] = Field(default=None, alias="remainingQty")
+    this_plan_qty: float = Field(..., alias="thisPlanQty")
+    model_config = {"populate_by_name": True}
+
+
+class ContractQtyReminder(BaseModel):
+    items: list[ContractQtyReminderItem] = Field(default_factory=list)
+    model_config = {"populate_by_name": True}
+
+
 class TransportArrangeResult(TransportRead):
     driver_access_token: Optional[str] = Field(default=None, alias="driverAccessToken")
     driver_url: Optional[str] = Field(default=None, alias="driverUrl")
@@ -225,6 +245,9 @@ class TransportArrangeResult(TransportRead):
         default_factory=list, alias="inventoryRecords"
     )
     io_order: Optional[TransportIoOrderBrief] = Field(default=None, alias="ioOrder")
+    contract_qty_reminder: Optional[ContractQtyReminder] = Field(
+        default=None, alias="contractQtyReminder"
+    )
     model_config = {"from_attributes": True, "populate_by_name": True}
 
 

backend/app/services/pl_contract_plan_qty.py

@@ -0,0 +1,814 @@
+"""合同产品在运输计划上的安排数量汇总。"""
+from __future__ import annotations
+
+from collections import defaultdict
+from typing import Any, DefaultDict, Dict, Iterable, Optional, Set
+
+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）
+
+
+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[str, float],
+    arranged_unarchived: Dict[str, float],
+) -> Dict[str, Dict[str, float]]:
+    """product_id -> {archivedQty, arrangedUnarchivedQty, arrangedTotalQty}"""
+    product_ids = set(archived) | set(arranged_unarchived)
+    out: Dict[str, Dict[str, float]] = {}
+    for pid in product_ids:
+        a = round(float(archived.get(pid, 0.0)), 3)
+        u = round(float(arranged_unarchived.get(pid, 0.0)), 3)
+        out[pid] = {
+            "archivedQty": a,
+            "arrangedUnarchivedQty": u,
+            "arrangedTotalQty": round(a + u, 3),
+        }
+    return out
+
+
+def accumulate_stop_products(
+    target: DefaultDict[str, float],
+    products: Iterable[Any],
+) -> None:
+    for row in products or []:
+        pid = str(getattr(row, "product_id", None) or "").strip()
+        if not pid:
+            continue
+        qty = _to_float(getattr(row, "plan_qty", None))
+        if qty:
+            target[pid] += 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],
+) -> Dict[str, float]:
+    acc: DefaultDict[str, 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, []))
+        else:
+            for lid in all_load_ids:
+                accumulate_stop_products(acc, load_products_by_load.get(lid, []))
+        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, []))
+    elif load_ids:
+        for lid in load_ids:
+            accumulate_stop_products(acc, load_products_by_load.get(lid, []))
+
+    if delivery_ids is None:
+        for did in all_delivery_ids:
+            accumulate_stop_products(acc, delivery_products_by_delivery.get(did, []))
+    elif delivery_ids:
+        for did in delivery_ids:
+            accumulate_stop_products(acc, delivery_products_by_delivery.get(did, []))
+
+    return dict(acc)
+
+
+def aggregate_contract_product_plan_qtys(
+    db: Session,
+    *,
+    tenant_no: str,
+    contract_id: str,
+) -> Dict[str, Dict[str, float]]:
+    """返回 productId -> 三数量字段。"""
+    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[str, float] = defaultdict(float)
+    arranged_unarchived: DefaultDict[str, 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
+        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,
+        )
+        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[str, Dict[str, float]],
+) -> None:
+    for item in products or []:
+        if not isinstance(item, dict):
+            continue
+        pid = str(item.get("productId") or "").strip()
+        stats = qty_by_product_id.get(pid) 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()
+        )
+        products_by_pid = {
+            str(p.product_id or "").strip(): p for p in products if p.product_id
+        }
+        stop_kind = _stop_kind_from_scope(scope)
+        contract_type = (contract.contract_type or "").strip()
+        contract_no = contract.contract_no or ""
+
+        for pid, new_qty_raw in plan_qtys.items():
+            new_qty = _to_float(new_qty_raw)
+            if new_qty <= 0:
+                continue
+            cp = products_by_pid.get(pid)
+            # 计划产品不在合同明细：不校验数量（归属另有规则）
+            if not cp:
+                continue
+
+            arranged_total = _to_float((qty_map.get(pid) 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,
+                "stopKind": stop_kind,
+                "productId": pid,
+                "productName": cp.product_name or "",
+                "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"],
+                    }
+                )
+            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}

backend/tests/test_pl_contract_qty_arrange_check.py

@@ -0,0 +1,123 @@
+"""安排运输时合同数量硬拦 / 提醒纯函数单测。"""
+
+from app.services.pl_contract_plan_qty import (
+    contract_product_has_fixed_qty,
+    effective_arranged_total,
+    evaluate_product_qty_for_arrange,
+)
+
+
+def test_effective_arranged_total_subtracts_old():
+    assert effective_arranged_total(arranged_total=100, old_plan_qty=30, new_plan_qty=40) == 110.0
+
+
+def test_effective_arranged_total_rounds():
+    assert effective_arranged_total(arranged_total=10.1234, old_plan_qty=1.1, new_plan_qty=2.2) == 11.223
+
+
+def test_framework_customer_demand_not_fixed():
+    assert (
+        contract_product_has_fixed_qty(
+            contract_type="framework_sale",
+            quantity=0,
+            quantity_mode="customer_demand",
+        )
+        is False
+    )
+
+
+def test_framework_fixed_mode_with_qty_is_fixed():
+    assert (
+        contract_product_has_fixed_qty(
+            contract_type="framework_purchase",
+            quantity=50,
+            quantity_mode="fixed",
+        )
+        is True
+    )
+
+
+def test_sale_positive_qty_is_fixed():
+    assert (
+        contract_product_has_fixed_qty(
+            contract_type="sale",
+            quantity=100,
+            quantity_mode="fixed",
+        )
+        is True
+    )
+
+
+def test_zero_or_empty_qty_not_fixed():
+    assert contract_product_has_fixed_qty(contract_type="sale", quantity=0, quantity_mode="fixed") is False
+    assert contract_product_has_fixed_qty(contract_type="sale", quantity=None, quantity_mode="fixed") is False
+
+
+def test_evaluate_skip_when_no_new_qty():
+    assert (
+        evaluate_product_qty_for_arrange(
+            has_fixed_qty=True,
+            contract_qty=100,
+            arranged_total=80,
+            old_plan_qty=0,
+            new_plan_qty=0,
+        )
+        is None
+    )
+
+
+def test_evaluate_overflow_violation():
+    r = evaluate_product_qty_for_arrange(
+        has_fixed_qty=True,
+        contract_qty=100,
+        arranged_total=80,
+        old_plan_qty=0,
+        new_plan_qty=30,
+    )
+    assert r is not None
+    assert r["kind"] == "violation"
+    assert r["overflowQty"] == 10.0
+    assert r["effectiveTotal"] == 110.0
+    assert r["arrangedTotalQty"] == 110.0
+
+
+def test_evaluate_remaining_reminder():
+    r = evaluate_product_qty_for_arrange(
+        has_fixed_qty=True,
+        contract_qty=200,
+        arranged_total=150,
+        old_plan_qty=0,
+        new_plan_qty=40,
+    )
+    assert r is not None
+    assert r["kind"] == "remaining"
+    assert r["remainingQty"] == 10.0
+    assert r["arrangedTotalQty"] == 190.0
+
+
+def test_evaluate_rearrange_deducts_old():
+    # arranged 已含旧 30；改为 40 → effective 110
+    r = evaluate_product_qty_for_arrange(
+        has_fixed_qty=True,
+        contract_qty=100,
+        arranged_total=100,
+        old_plan_qty=30,
+        new_plan_qty=40,
+    )
+    assert r is not None
+    assert r["kind"] == "violation"
+    assert r["overflowQty"] == 10.0
+
+
+def test_evaluate_framework_cumulative():
+    r = evaluate_product_qty_for_arrange(
+        has_fixed_qty=False,
+        contract_qty=None,
+        arranged_total=126.5,
+        old_plan_qty=0,
+        new_plan_qty=28,
+    )
+    assert r is not None
+    assert r["kind"] == "framework_cumulative"
+    assert r["arrangedTotalQty"] == 154.5
+    assert r["thisPlanQty"] == 28.0

docs/superpowers/plans/2026-08-30-pl-plan-contract-qty-check.md

@@ -0,0 +1,276 @@
+# 运输计划安排 · 合同数量硬拦与成功后提醒 Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** 安排运输时对有数量合同产品硬拦超量；安排成功后对未超限产品提醒剩余量、对框架无数量合同提醒累计已安排。
+
+**Architecture:** 在 `POST /transport/order/arrange/{plan_id}` 写运输前调用校验（复用 `pl_contract_plan_qty` 汇总口径）；超限 `422`；成功则在 `TransportArrangeResult.contractQtyReminder` 带回明细，前端成功后 `ElMessageBox.alert`。
+
+**Tech Stack:** FastAPI / SQLAlchemy / Pydantic v2；Vue 3 + Element Plus；pytest
+
+**Spec:** `docs/superpowers/specs/2026-08-30-pl-plan-contract-qty-check-design.md`  
+**Demo:** `docs/fresh-mint-pl-plan-contract-qty-check-demo.html`
+
+---
+
+## File map
+
+| File | Responsibility |
+|------|----------------|
+| `backend/app/services/pl_contract_plan_qty.py` | 纯函数：有效合计、是否固定量、评估 violations/reminders；编排：按计划收集合同校验 |
+| `backend/app/schemas/pl/transport.py` | `ContractQtyReminder*` + `TransportArrangeResult` 扩展 |
+| `backend/app/api/pl/transport.py` | arrange 前 assert；result 附带 reminder |
+| `backend/tests/test_pl_contract_qty_arrange_check.py` | 纯函数 +（可选）编排单测 |
+| `web-admin/src/api/pl/index.ts` | TS 类型 |
+| `web-admin/src/utils/pl-contract-qty-reminder.ts` | 格式化超额/提醒文案 |
+| `web-admin/src/composables/usePlPlanArrangeSubmit.ts` 或调用方 | 成功后弹提醒；422 超额弹窗 |
+
+---
+
+### Task 1: 纯函数 — 有效合计与产品判定
+
+**Files:**
+- Modify: `backend/app/services/pl_contract_plan_qty.py`
+- Test: `backend/tests/test_pl_contract_qty_arrange_check.py`
+
+- [ ] **Step 1: 写失败单测**
+
+```python
+from app.services.pl_contract_plan_qty import (
+    effective_arranged_total,
+    contract_product_has_fixed_qty,
+    evaluate_product_qty_for_arrange,
+)
+
+def test_effective_arranged_total_subtracts_old():
+    assert effective_arranged_total(arranged_total=100, old_plan_qty=30, new_plan_qty=40) == 110.0
+
+def test_framework_customer_demand_not_fixed():
+    assert contract_product_has_fixed_qty(
+        contract_type="framework_sale", quantity=0, quantity_mode="customer_demand"
+    ) is False
+
+def test_sale_positive_qty_is_fixed():
+    assert contract_product_has_fixed_qty(
+        contract_type="sale", quantity=100, quantity_mode="fixed"
+    ) is True
+
+def test_evaluate_overflow_violation():
+    r = evaluate_product_qty_for_arrange(
+        has_fixed_qty=True,
+        contract_qty=100,
+        arranged_total=80,
+        old_plan_qty=0,
+        new_plan_qty=30,
+    )
+    assert r["kind"] == "violation"
+    assert r["overflowQty"] == 10.0
+
+def test_evaluate_remaining_reminder():
+    r = evaluate_product_qty_for_arrange(
+        has_fixed_qty=True,
+        contract_qty=200,
+        arranged_total=150,
+        old_plan_qty=0,
+        new_plan_qty=40,
+    )
+    assert r["kind"] == "remaining"
+    assert r["remainingQty"] == 10.0
+
+def test_evaluate_framework_cumulative():
+    r = evaluate_product_qty_for_arrange(
+        has_fixed_qty=False,
+        contract_qty=None,
+        arranged_total=126.5,
+        old_plan_qty=0,
+        new_plan_qty=28,
+    )
+    assert r["kind"] == "framework_cumulative"
+    assert r["arrangedTotalQty"] == 154.5
+```
+
+- [ ] **Step 2: 跑测确认失败**
+
+```bash
+cd backend && source venv/bin/activate && pytest tests/test_pl_contract_qty_arrange_check.py -q
+```
+
+- [ ] **Step 3: 实现纯函数（同文件末尾或清晰分区）**
+
+- `effective_arranged_total(arranged_total, old_plan_qty, new_plan_qty) -> float`（round 3）
+- `contract_product_has_fixed_qty(contract_type, quantity, quantity_mode) -> bool`  
+  - 框架 + `customer_demand` → False  
+  - quantity 有效且 `> 0` → True  
+  - 否则 False
+- `evaluate_product_qty_for_arrange(...)` → `{kind, ...}`  
+  - fixed 且 effective > contractQty → violation  
+  - fixed 且不超 → remaining  
+  - 非 fixed 且 new_plan_qty > 0 → framework_cumulative  
+  - new_plan_qty == 0 → skip（返回 None 或 kind=skip）
+
+- [ ] **Step 4: 跑测通过并 commit**
+
+```bash
+pytest tests/test_pl_contract_qty_arrange_check.py -q
+git add backend/app/services/pl_contract_plan_qty.py backend/tests/test_pl_contract_qty_arrange_check.py
+git commit -m "feat(pl): add contract qty arrange check pure helpers"
+```
+
+---
+
+### Task 2: 编排 — 按计划收集 violations / reminders
+
+**Files:**
+- Modify: `backend/app/services/pl_contract_plan_qty.py`
+- Test: `backend/tests/test_pl_contract_qty_arrange_check.py`（可用 SimpleNamespace / mock 聚合）
+
+- [ ] **Step 1: 实现 `collect_plan_contract_qty_checks(db, tenant_no, plan_id)`**
+
+逻辑概要：
+
+1. 加载计划；查本计划全部 `PlPlanContract`（及遗留 `plan.contract_id`）
+2. 按 `contract_id` 分组；对每个合同：
+   - `aggregate_contract_product_plan_qtys(...)` 得 arrangedTotal
+   - 加载 CRM 合同 + 产品行（`CrmContract` / `CrmContractProduct`）
+   - 计算本计划在该合同 scope 下各 product 的 **old/new**（同为当前库内 plan_qty；校验时「旧」=「新」均为当前值即可——因为 arrangedTotal 在首次安排前不含本计划，调整时含本计划旧量；`effective = arranged - old + new` 且 old=new=当前库内量 → `effective = arranged`。  
+     **重要纠正：** 调整时用户可能已先「保存」改过 plan_qty，则库内已是新量，arrangedTotal 仍含**保存前**旧量吗？  
+     → 若先保存再 arrange：库内已是新量，但 arrangedTotal 按库内段产品汇总，**已含新量**，则 `old` 与 `new` 都应取当前库内量，`effective = arranged` 正确（扣新加新抵消）。  
+     → 若未先保存（仅列表安排、数量未改）：同样 old=new=库内量。  
+     → **结论：在「安排前计划已落库」前提下，`old_plan_qty = new_plan_qty = 本计划对该合同 scope 的当前 plan_qty 汇总`，`effective = arrangedTotal`。**  
+     Spec 中的「扣旧加新」在「保存并安排先落库」后数学上等价于直接用含本计划的 arrangedTotal；若本计划尚不在已安排桶（首次），arrangedTotal 不含本计划，则必须 `effective = arrangedTotal + new`（old=0 因不在桶内）。
+
+实现时用：
+
+```python
+in_bucket = _bucket_for_plan(plan...) is not None
+plan_qty_on_contract = sum_qtys_for_scope(... 仅本计划对该合同的 scope ...)
+if in_bucket:
+    effective = arranged_total  # 已含当前库内量
+    old = plan_qty_on_contract
+    new = plan_qty_on_contract
+else:
+    effective = arranged_total + plan_qty_on_contract
+    old = 0
+    new = plan_qty_on_contract
+```
+
+对每个合同产品 × 本计划在该合同上有量的 product_id（或合同所有产品中本计划涉及的）调用 `evaluate_product_qty_for_arrange`。
+
+3. 返回 `(violations: list[dict], reminders: list[dict])`
+
+- [ ] **Step 2: `assert_contract_qty_for_arrange`** — violations 非空则 `HTTPException(422, detail={code, message, items})`
+
+- [ ] **Step 3: 单测覆盖 in_bucket / not in_bucket / 多产品**
+
+- [ ] **Step 4: commit**
+
+```bash
+git commit -m "feat(pl): collect and assert contract qty on arrange"
+```
+
+---
+
+### Task 3: Schema + arrange API 接线
+
+**Files:**
+- Modify: `backend/app/schemas/pl/transport.py`
+- Modify: `backend/app/api/pl/transport.py`
+
+- [ ] **Step 1: 增加 Pydantic 模型**
+
+```python
+class ContractQtyReminderItem(BaseModel):
+    contract_id: str = Field(..., alias="contractId")
+    contract_no: Optional[str] = Field(None, alias="contractNo")
+    contract_type: Optional[str] = Field(None, alias="contractType")
+    stop_kind: str = Field(..., alias="stopKind")  # load | delivery
+    product_id: str = Field(..., alias="productId")
+    product_name: Optional[str] = Field(None, alias="productName")
+    mode: str  # remaining | framework_cumulative
+    contract_qty: Optional[float] = Field(None, alias="contractQty")
+    arranged_total_qty: float = Field(..., alias="arrangedTotalQty")
+    remaining_qty: Optional[float] = Field(None, alias="remainingQty")
+    this_plan_qty: float = Field(..., alias="thisPlanQty")
+    model_config = {"populate_by_name": True}
+
+class ContractQtyReminder(BaseModel):
+    items: list[ContractQtyReminderItem] = Field(default_factory=list)
+    model_config = {"populate_by_name": True}
+```
+
+`TransportArrangeResult` 增加 `contract_qty_reminder: Optional[ContractQtyReminder] = Field(None, alias="contractQtyReminder")`
+
+- [ ] **Step 2: `arrange_transport` 在危化证校验后、写运输前：**
+
+```python
+from app.services.pl_contract_plan_qty import (
+    assert_contract_qty_for_arrange,
+    build_contract_qty_reminder_for_plan,
+)
+assert_contract_qty_for_arrange(db, tenant_no=tenant_no, plan_id=plan_id)
+```
+
+成功 `_build_arrange_result` 后：
+
+```python
+result.contract_qty_reminder = build_contract_qty_reminder_for_plan(db, tenant_no=tenant_no, plan_id=plan_id)
+```
+
+注意：assert 时计划尚未变为 arranged；reminder 在 commit 后计划状态已更新为 arranged，应用**同一 evaluate 公式**基于「即将/已经计入」的有效合计，避免剩余量跳变。推荐：assert 与 reminder 在写运输**前**用同一 `collect` 结果——violations 则抛错；无 violations 则缓存 `reminders`，成功后原样写入 result（不要在 commit 后再算一遍导致口径漂移）。
+
+- [ ] **Step 3: 手动或集成测 arrange 超限返回 422**
+
+- [ ] **Step 4: commit**
+
+```bash
+git commit -m "feat(pl): enforce contract qty on transport arrange API"
+```
+
+---
+
+### Task 4: 前端类型与弹窗
+
+**Files:**
+- Modify: `web-admin/src/api/pl/index.ts`
+- Create: `web-admin/src/utils/pl-contract-qty-reminder.ts`
+- Modify: 安排成功调用链（`usePlPlanArrangeSubmit.ts` 的调用方：`pl-plan-form-dialog`、列表安排弹窗等）
+
+- [ ] **Step 1: 扩展 `TransportArrangeResult` 类型**
+
+- [ ] **Step 2: 工具函数**
+
+```ts
+export function formatContractQtyViolationMessage(detail: unknown): string
+export function formatContractQtyReminderMessage(reminder: ContractQtyReminder): string
+export async function showContractQtyReminderIfAny(reminder?: ContractQtyReminder | null)
+export async function showContractQtyViolationIfMatch(error: unknown): Promise<boolean>
+```
+
+- [ ] **Step 3: `submitPlanArrange` 成功后调用 `showContractQtyReminderIfAny`；catch 中优先 `showContractQtyViolationIfMatch`**
+
+找到所有 `transportApi.arrange` / `submitPlanArrange` 调用点一并接上。
+
+- [ ] **Step 4: 文案对齐 Demo（知道了 / 无法安排运输）**
+
+- [ ] **Step 5: commit**
+
+```bash
+git commit -m "feat(web-admin): show contract qty block and post-arrange reminders"
+```
+
+---
+
+### Task 5: 验证与图谱
+
+- [ ] `pytest tests/test_pl_contract_qty_arrange_check.py tests/test_pl_contract_plan_qty.py -q`
+- [ ] 前端相关文件 lint（若改 vue）
+- [ ] `./scripts/update-graph.sh backend`（或 frontend，视改动）
+- [ ] 对照 spec 测试清单手工过一遍关键路径
+
+---
+
+## 执行备注
+
+- 最小改动；不改合同详情三列语义
+- 草稿仅保存不调用 collect/assert
+- YAGNI：不做独立预检 API

web-admin/src/api/pl/index.ts

@@ -629,6 +629,8 @@ export interface PlanQuery {
   certUploadStatus?: string
   /** 0=未确认齐备 1=已确认齐备 */
   hazardCertReady?: number
+  /** CRM 合同 ID：仅返回关联该合同的计划 */
+  contractId?: string
   page?: number
   pageSize?: number
 }
@@ -1099,6 +1101,25 @@ export interface TransportArrangeResult extends Transport {
   siteRecords?: TransportSiteLink[]
   inventoryRecords?: TransportInventoryLink[]
   ioOrder?: TransportIoOrderBrief
+  contractQtyReminder?: ContractQtyReminder
+}
+
+export interface ContractQtyReminderItem {
+  contractId: string
+  contractNo?: string
+  contractType?: string
+  stopKind: string
+  productId: string
+  productName?: string
+  mode: 'remaining' | 'framework_cumulative' | string
+  contractQty?: number
+  arrangedTotalQty: number
+  remainingQty?: number
+  thisPlanQty: number
+}
+
+export interface ContractQtyReminder {
+  items: ContractQtyReminderItem[]
 }
 
 export interface PlIoOrderItem {

web-admin/src/components/business/pl-plan-form-dialog/index.vue

@@ -251,6 +251,7 @@
           <el-icon><Close /></el-icon>取消
         </el-button>
         <el-button
+          v-if="showSaveButton"
           :type="showSaveAndArrangeButton ? 'default' : 'primary'"
           :class="showSaveAndArrangeButton ? 'btn-ghost' : undefined"
           :loading="submitLoading"
@@ -481,6 +482,8 @@
   const loadedPlanSnapshot = ref<Plan | null>(null)
   const selectedDriverId = ref('')
 
+  /** 已安排：禁止仅「保存」，须走保存并安排/调整运输，避免计划与地磅等联动字段脱节 */
+  const showSaveButton = computed(() => loadedTransportStatus.value !== 'arranged')
   const showSaveAndArrangeButton = computed(() => {
     if (!isEdit.value) return true
     if (props.openForArrange && loadedTransportStatus.value === 'arranged') return false
@@ -1438,7 +1441,11 @@
       getUnitLabel
     })
 
-    await submitPlanArrange(planId, payload)
+    try {
+      await submitPlanArrange(planId, payload)
+    } catch {
+      return false
+    }
     ElMessage.success(isRearrange ? '运输调整成功' : '运输安排成功')
     innerVisible.value = false
     emit('success', { planId, arranged: true })

web-admin/src/composables/usePlPlanArrangeSubmit.ts

@@ -10,6 +10,10 @@ import {
 import { pickBxData } from '@/utils/http/bx'
 import { planRequiresSiteLink } from '@/utils/pl-internal-customer'
 import type { PlanDeliveryForm, PlanLoadForm } from '@/utils/pl-plan-stop-form'
+import {
+  showContractQtyReminderIfAny,
+  showContractQtyViolationIfMatch
+} from '@/utils/pl-contract-qty-reminder'
 
 export interface ArrangeHistoricalFlags {
   hasHistoricalDibang: boolean
@@ -202,10 +206,16 @@ export async function submitPlanArrange(
   planId: string,
   payload: TransportArrange
 ): Promise<TransportArrangeResult> {
-  const res = await transportApi.arrange(planId, payload)
-  const result = pickBxData<TransportArrangeResult>(res)
-  if (!result) {
-    throw new Error('安排失败：未返回结果')
+  try {
+    const res = await transportApi.arrange(planId, payload)
+    const result = pickBxData<TransportArrangeResult>(res)
+    if (!result) {
+      throw new Error('安排失败：未返回结果')
+    }
+    await showContractQtyReminderIfAny(result.contractQtyReminder)
+    return result
+  } catch (error) {
+    await showContractQtyViolationIfMatch(error)
+    throw error
   }
-  return result
 }

web-admin/src/utils/http/bx.ts

@@ -149,7 +149,13 @@ bxAxios.interceptors.response.use(
       await userStore.logOut()
     } else {
       const detail = err.response?.data?.detail || err.response?.data?.msg
-      ElMessage.error(detail || err.message || '网络异常')
+      const message =
+        typeof detail === 'string'
+          ? detail
+          : detail && typeof detail === 'object'
+            ? detail.message || detail.msg || JSON.stringify(detail)
+            : detail
+      ElMessage.error(message || err.message || '网络异常')
     }
     return Promise.reject(err)
   }

web-admin/src/utils/pl-contract-qty-reminder.ts

@@ -0,0 +1,108 @@
+import { ElMessageBox } from 'element-plus'
+import type { ContractQtyReminder, ContractQtyReminderItem } from '@/api/pl'
+
+export const CONTRACT_QTY_VIOLATION_CODE = 'CONTRACT_QTY_OVERFLOW'
+
+function formatQty(n: number | undefined | null): string {
+  if (n == null || Number.isNaN(Number(n))) return '—'
+  return Number(n).toFixed(3)
+}
+
+function stopKindLabel(kind?: string): string {
+  if (kind === 'delivery') return '送货'
+  if (kind === 'load') return '装货'
+  return kind || ''
+}
+
+export function extractContractQtyViolationDetail(error: unknown): {
+  message: string
+  items: Array<Record<string, unknown>>
+} | null {
+  const err = error as {
+    response?: { status?: number; data?: { detail?: unknown } }
+  }
+  const detail = err?.response?.data?.detail
+  if (!detail || typeof detail !== 'object') return null
+  const d = detail as { code?: string; message?: string; items?: unknown }
+  if (d.code !== CONTRACT_QTY_VIOLATION_CODE) return null
+  const items = Array.isArray(d.items) ? (d.items as Array<Record<string, unknown>>) : []
+  return {
+    message: d.message || '合同产品安排数量超过合同数量',
+    items
+  }
+}
+
+export function formatContractQtyViolationMessage(
+  items: Array<Record<string, unknown>>
+): string {
+  if (!items.length) return '合同产品安排数量超过合同数量，请调整后再试。'
+  const lines = items.map((it) => {
+    const name = String(it.productName || it.productId || '产品')
+    const contractNo = String(it.contractNo || it.contractId || '')
+    const side = stopKindLabel(String(it.stopKind || ''))
+    const contractQty = formatQty(Number(it.contractQty))
+    const effective = formatQty(Number(it.effectiveTotal ?? it.arrangedTotalQty))
+    const overflow = formatQty(Number(it.overflowQty))
+    const meta = [contractNo, side].filter(Boolean).join(' · ')
+    return `「${name}」安排后合计 ${effective}，超过合同数量 ${contractQty}（超 ${overflow}）${meta ? `；${meta}` : ''}`
+  })
+  return `以下产品「已安排合计 + 本次安排」超过合同数量，请调整后再试：\n\n${lines.join('\n')}`
+}
+
+export function formatContractQtyReminderMessage(reminder: ContractQtyReminder): string {
+  const items = reminder?.items || []
+  if (!items.length) return ''
+  const remaining = items.filter((i) => i.mode === 'remaining')
+  const framework = items.filter((i) => i.mode === 'framework_cumulative')
+  const blocks: string[] = []
+
+  if (remaining.length) {
+    const lines = remaining.map((it: ContractQtyReminderItem) => {
+      const name = it.productName || it.productId
+      const meta = [it.contractNo, stopKindLabel(it.stopKind)].filter(Boolean).join(' · ')
+      return `「${name}」已安排合计（含本次）${formatQty(it.arrangedTotalQty)}，剩余可安排 ${formatQty(it.remainingQty)}${meta ? `；${meta}` : ''}`
+    })
+    blocks.push(`合同数量校验通过。各产品剩余可安排量：\n${lines.join('\n')}`)
+  }
+
+  if (framework.length) {
+    const lines = framework.map((it: ContractQtyReminderItem) => {
+      const name = it.productName || it.productId
+      const meta = [it.contractNo, stopKindLabel(it.stopKind)].filter(Boolean).join(' · ')
+      return `「${name}」累计已安排 ${formatQty(it.arrangedTotalQty)}（本次 ${formatQty(it.thisPlanQty)}）${meta ? `；${meta}` : ''}`
+    })
+    blocks.push(
+      `关联框架合同无固定数量上限。本次安排后累计已安排：\n${lines.join('\n')}`
+    )
+  }
+
+  return blocks.join('\n\n')
+}
+
+/** @returns true 表示已按合同超量处理（调用方勿再抛出/重复提示） */
+export async function showContractQtyViolationIfMatch(error: unknown): Promise<boolean> {
+  const parsed = extractContractQtyViolationDetail(error)
+  if (!parsed) return false
+  await ElMessageBox.alert(formatContractQtyViolationMessage(parsed.items), '无法安排运输', {
+    type: 'error',
+    confirmButtonText: '知道了'
+  })
+  return true
+}
+
+export async function showContractQtyReminderIfAny(
+  reminder?: ContractQtyReminder | null
+): Promise<void> {
+  const items = reminder?.items || []
+  if (!items.length) return
+  const message = formatContractQtyReminderMessage(reminder!)
+  if (!message) return
+  const hasFramework = items.some((i) => i.mode === 'framework_cumulative')
+  const title = hasFramework && !items.some((i) => i.mode === 'remaining')
+    ? '安排成功 · 框架合同累计'
+    : '安排成功 · 合同剩余数量'
+  await ElMessageBox.alert(message, title, {
+    type: 'success',
+    confirmButtonText: '知道了'
+  })
+}