shanxia2.0/docs/superpowers/plans/2026-07-11-pl-dibang-plan-quantity-accumulation.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: 在 pl_transport_db_link 记录 contributed_qty；revoke_regeneratable_dibang_for_transport 统一回减；_sync_transport_inventory_links 统一累加。取消计划与调整运输复用现有 revoke 入口，无需新增 API。

Tech Stack: FastAPI + SQLAlchemy + pytest；MySQL 5.7

OpenSpec: openspec/changes/pl-dibang-plan-quantity-accumulation/
Design: docs/superpowers/specs/2026-07-11-pl-dibang-plan-quantity-accumulation-design.md

---

文件结构

操作	路径	职责
Create	docs/sql/20260711_pl_transport_db_link_contributed_qty.sql	增量 DDL
Modify	docs/sql/pl_tables.sql	基线同步 contributed_qty
Modify	backend/app/models/pl_models.py	ORM contributed_qty
Modify	backend/app/services/pl_dibang_bridge_service.py	累加/回减逻辑
Create	backend/tests/test_pl_dibang_plan_quantity.py	单元测试

---

Task 1: 数据库与 ORM

Files:

• Create: docs/sql/20260711_pl_transport_db_link_contributed_qty.sql

• Modify: docs/sql/pl_tables.sql

• Modify: backend/app/models/pl_models.py

• Step 1: 编写增量 SQL

-- 运输联动贡献量（吨），用于月计划量累加/回减
ALTER TABLE pl_transport_db_link
  ADD COLUMN contributed_qty DECIMAL(15,3) NULL COMMENT '本次联动贡献量（吨）' AFTER db_inventory_id;

• Step 2: 更新 pl_tables.sql

在 pl_transport_db_link 的 db_inventory_id 后增加：

  `contributed_qty` decimal(15,3) DEFAULT NULL COMMENT '本次联动贡献量（吨）',

• Step 3: ORM 字段

在 PlTransportDbLink 增加：

contributed_qty = Column(DECIMAL(15, 3), nullable=True, comment="本次联动贡献量（吨）")

• Step 4: 本地执行 DDL（开发库）

mysql -u ... -p ... < docs/sql/20260711_pl_transport_db_link_contributed_qty.sql

---

Task 2: 桥接服务 — 辅助函数

Files:

• Modify: backend/app/services/pl_dibang_bridge_service.py

• Step 1: 新增 _apply_plan_quantity_delta

def _apply_plan_quantity_delta(
    db_plan: DbPurchasePlan,
    delta: float,
    *,
    user_id: Optional[str],
) -> None:
    current = float(db_plan.planned_quantity or 0)
    db_plan.planned_quantity = max(0.0, current + float(delta))
    db_plan.update_by = user_id

• Step 2: 新增 _resolve_link_contributed_qty

def _resolve_link_contributed_qty(
    link: PlTransportDbLink,
    inventory: Optional[DbInboundOutbound],
) -> float:
    if link.contributed_qty is not None:
        return float(link.contributed_qty)
    if inventory and inventory.quantity is not None:
        return float(inventory.quantity)
    return 0.0

• Step 3: 新增 _subtract_plan_contribution

在作废每条 link 前调用：

def _subtract_plan_contribution(
    db: Session,
    *,
    tenant_no: str,
    link: PlTransportDbLink,
    inventory: Optional[DbInboundOutbound],
    user_id: Optional[str],
) -> None:
    if not link.db_purchase_plan_id:
        return
    db_plan = (
        db.query(DbPurchasePlan)
        .filter(
            DbPurchasePlan.id == link.db_purchase_plan_id,
            DbPurchasePlan.tenant_no == tenant_no,
            DbPurchasePlan.del_flag == 0,
        )
        .first()
    )
    if not db_plan:
        return
    qty = _resolve_link_contributed_qty(link, inventory)
    if qty:
        _apply_plan_quantity_delta(db_plan, -qty, user_id=user_id)

---

Task 3: 桥接服务 — 累加（sync）

Files:

• Modify: backend/app/services/pl_dibang_bridge_service.py

• Step 1: 在 _sync_transport_inventory_links 产品循环内，取得 contrib

contrib = _product_quantity(product)

• Step 2: 复用月计划分支增加累加

在 if not db_plan: ... _create_db_plan(...) 的 else 分支（或 db_plan 已存在时）：

_apply_plan_quantity_delta(db_plan, contrib, user_id=user_id)

注意：_create_db_plan 新建时 planned_quantity 已设为 contrib，仅复用时执行 +=。

• Step 3: 创建 link 时写入 contributed_qty

link = PlTransportDbLink(
    ...
    contributed_qty=Decimal(str(contrib)),
    ...
)

---

Task 4: 桥接服务 — 回减（revoke）

Files:

• Modify: backend/app/services/pl_dibang_bridge_service.py

• Step 1: 在 revoke_regeneratable_dibang_for_transport 第二段循环（实际作废）中，软删前调用回减

在 if inventory: inventory.del_flag = 1 之前：

_subtract_plan_contribution(
    db,
    tenant_no=tenant_no,
    link=link,
    inventory=inventory,
    user_id=user_id,
)

• Step 2: 确认 cancel_plan 与 arrange_transport 调整分支无需改入口（已调用 revoke）

---

Task 5: 单元测试

Files:

• Create: backend/tests/test_pl_dibang_plan_quantity.py

• Step 1: 编写测试（使用项目现有 pytest + DB session 夹具模式；若无夹具则用 mock Session 测纯函数）

至少覆盖：

def test_apply_plan_quantity_delta_never_below_zero():
    ...

def test_sync_creates_plan_with_contributed_qty():
    # 无月计划 → planned_quantity = 30, link.contributed_qty = 30
    ...

def test_sync_reuses_plan_and_accumulates():
    # 已有 30，再 +31 → 61
    ...

def test_revoke_subtracts_on_cancel():
    # 61 - 30 → 31
    ...

def test_revoke_then_sync_on_rearrange():
    # 30 回减后 +25 → 25
    ...

def test_revoke_blocks_when_weighing_exists():
    # 月计划量不变
    ...

• Step 2: 运行测试

cd backend && source venv/bin/activate && pytest tests/test_pl_dibang_plan_quantity.py -v

Expected: 全部 PASS

---

Task 6: 图谱与验收

• Step 1: 更新知识图谱

./scripts/update-graph.sh backend

• Step 2: 手工验收清单

步骤	操作	预期月计划量
1	计划 A 安排，B+C，30t，勾地磅	30
2	计划 B 安排，B+C，31t，勾地磅	61
3	取消计划 A	31
4	计划 B 调整 31→25，勾地磅	25
5	对已完成/已关联磅单单据尝试取消	失败，量不变

---

Spec 覆盖自检

OpenSpec Scenario	Task
首次联动创建月计划	Task 3
后续联动累加月计划	Task 3
调整运输变更数量	Task 4 + Task 3
调整运输取消勾选地磅	Task 4
调整运输被阻断	Task 4（现有门禁不变）
取消计划回减月计划量	Task 4
取消计划被阻断	Task 4
取消后月计划量下限	Task 2 _apply_plan_quantity_delta

---

执行方式

Plan 已保存至 docs/superpowers/plans/2026-07-11-pl-dibang-plan-quantity-accumulation.md。

1. Subagent-Driven（推荐） — 每 Task 派发子代理，逐 Task 评审
2. Inline Execution — 本会话按 Task 顺序直接改代码

需要我开始实现吗？