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

运输计划安排 · 合同数量硬拦与成功后提醒 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: 写失败单测

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: 跑测确认失败

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

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 因不在桶内）。

实现时用：

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。

1. 返回 (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

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 模型

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 在危化证校验后、写运输前：

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 后：

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

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: 工具函数

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

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