shanxia2.0/docs/superpowers/plans/2026-10-09-pl-plan-copy-seq.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: 为 pl_plan 增加租户内固定流水 copy_seq_no：历史按创建时间回填；新计划首次「复制发群」时发号；小程序复制文案在「计划编号」上方输出 序号：n。

Architecture: 落库字段 + 专用 POST .../ensure-copy-seq（行锁 + 租户 MAX+1，含软删）。小程序复制前先 ensure，失败则中止复制。不在创建计划路径发号，不在列表/详情 UI 展示。

Tech Stack: MySQL 5.7、FastAPI、SQLAlchemy、uni-app Vue 3、pytest（对齐 allocate_reimburse_title_seq / test_bx_reimburse_title_seq.py）

Spec: docs/superpowers/specs/2026-10-09-pl-plan-copy-seq-design.md

---

File map

文件	职责
docs/sql/migrate_pl_plan_copy_seq_no_20261009.sql	ALTER + 按租户回填
docs/sql/shanxia2.0.sql	基线 pl_plan 定义同步 copy_seq_no（若仓库要求基线一致）
backend/app/models/pl_models.py	PlPlan.copy_seq_no
backend/app/services/pl_driver_plan_service.py	allocate_or_get_plan_copy_seq + ensure_copy_seq_for_driver
backend/app/api/pl/driver_plan.py	POST /{plan_id}/ensure-copy-seq
backend/tests/test_pl_plan_copy_seq.py	SQLite 发号 / 幂等 / 含软删 MAX
miniprogram/api/pl.js	driverPlanApi.ensureCopySeq
miniprogram/utils/plDriverPlanCopy.js	文案首行 序号：
miniprogram/pages/pl/driver-plan/list.vue	handleCopy：get → ensure → copy
miniprogram/pages/pl/driver-plan/detail.vue	保存成功复制：ensure → copy

---

Task 1: Schema 迁移 + ORM

Files:

• Create: docs/sql/migrate_pl_plan_copy_seq_no_20261009.sql
• Modify: backend/app/models/pl_models.py（PlPlan 公共字段前）

• Modify: docs/sql/shanxia2.0.sql（pl_plan CREATE 中增加同列；搜索 CREATE TABLE `pl_plan`）

• [ ] Step 1: 写迁移 SQL

-- 运输计划复制发群固定序号（租户内流水）
-- MySQL 5.7
-- 2026-10-09
-- 若列已存在，跳过 ALTER（重复执行会报 Duplicate column）。

ALTER TABLE `pl_plan`
  ADD COLUMN `copy_seq_no` int(11) DEFAULT NULL COMMENT '复制发群序号（租户内固定流水，首次复制时分配）' AFTER `plan_no`,
  ADD KEY `idx_pl_plan_tenant_copy_seq` (`tenant_no`, `copy_seq_no`);

-- 历史回填：按租户 create_time ASC, id ASC，含软删；已有值不覆盖
-- MySQL 5.7 用户变量按租户编号（会话内执行）
SET @prev_tenant := NULL;
SET @rn := 0;

UPDATE `pl_plan` p
INNER JOIN (
  SELECT
    t.id,
    t.seq_no
  FROM (
    SELECT
      id,
      tenant_no,
      @rn := IF(@prev_tenant = tenant_no, @rn + 1, 1) AS seq_no,
      @prev_tenant := tenant_no AS _tenant
    FROM (
      SELECT id, tenant_no, create_time
      FROM `pl_plan`
      ORDER BY tenant_no ASC, create_time ASC, id ASC
    ) ordered
  ) t
) ranked ON ranked.id = p.id
SET p.`copy_seq_no` = ranked.seq_no
WHERE p.`copy_seq_no` IS NULL;

在目标库手动执行一次（或按运维流程执行）。重复执行：第二次 ADD COLUMN 会失败，属预期；回填 WHERE IS NULL 可单独重跑（需先去掉 ADD，或整文件只跑一次）。

• Step 2: ORM 增加字段

在 PlPlan 中 plan_no 后增加：

copy_seq_no = Column(Integer, nullable=True, comment="复制发群序号（租户内固定流水）")

（文件顶部已有 Integer import 则复用。）

• Step 3: 同步 docs/sql/shanxia2.0.sql 中 pl_plan 表结构

在 `plan_no` varchar(32) ... 下一行增加：

  `copy_seq_no` int(11) DEFAULT NULL COMMENT '复制发群序号（租户内固定流水，首次复制时分配）',

并在 KEY 区增加 KEY `idx_pl_plan_tenant_copy_seq` (`tenant_no`, `copy_seq_no`)（若该 dump 含索引段）。

• Step 4: Commit

git add docs/sql/migrate_pl_plan_copy_seq_no_20261009.sql docs/sql/shanxia2.0.sql backend/app/models/pl_models.py
git commit -m "$(cat <<'EOF'
feat(pl): add pl_plan.copy_seq_no for copy-to-group sequence

Persist a tenant-scoped fixed serial for driver-plan clipboard text;
include migration backfill by create_time.
EOF
)"

---

Task 2: Service 发号逻辑 + 单测（TDD）

Files:

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

• Create: backend/tests/test_pl_plan_copy_seq.py

• [ ] Step 1: 写失败测试

创建 backend/tests/test_pl_plan_copy_seq.py：

"""pl_plan.copy_seq_no：租户内 MAX+1，含软删；已有号幂等返回。"""
from datetime import date, datetime

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.pl_models import PlPlan
from app.services.pl_driver_plan_service import allocate_or_get_plan_copy_seq

TENANT = "t-copy-seq"


def _session():
    engine = create_engine("sqlite://")
    PlPlan.__table__.create(engine)
    return sessionmaker(bind=engine)()


def _plan(
    *,
    plan_id: str,
    plan_no: str,
    create_time: datetime,
    copy_seq_no=None,
    del_flag: int = 0,
    tenant_no: str = TENANT,
) -> PlPlan:
    return PlPlan(
        id=plan_id,
        plan_no=plan_no,
        copy_seq_no=copy_seq_no,
        create_date=create_time.date() if hasattr(create_time, "date") else date(2026, 1, 1),
        belongs_company="shanxia",
        plan_type="direct",
        follow_status="pending_arrange",
        transport_status="pending",
        archive_status="unarchived",
        complete_status="no",
        has_freight=0,
        total_amount=0,
        freight_amount=0,
        hazard_cert_ready=0,
        erp_entered=0,
        tenant_no=tenant_no,
        create_dept_id="d1",
        create_time=create_time,
        update_time=create_time,
        del_flag=del_flag,
    )


def test_allocate_first_seq_when_tenant_empty():
    db = _session()
    db.add(_plan(plan_id="p1", plan_no="PO1", create_time=datetime(2026, 10, 1, 10, 0, 0)))
    db.commit()

    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="p1") == 1
    db.commit()
    assert db.query(PlPlan).get("p1").copy_seq_no == 1


def test_allocate_skips_soft_deleted_max():
    """软删占用的号计入 MAX，不回收。"""
    db = _session()
    db.add(
        _plan(
            plan_id="old",
            plan_no="PO0",
            create_time=datetime(2026, 9, 1, 10, 0, 0),
            copy_seq_no=3,
            del_flag=1,
        )
    )
    db.add(
        _plan(
            plan_id="new",
            plan_no="PO1",
            create_time=datetime(2026, 10, 1, 10, 0, 0),
            copy_seq_no=None,
        )
    )
    db.commit()

    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="new") == 4


def test_allocate_idempotent_when_already_set():
    db = _session()
    db.add(
        _plan(
            plan_id="p1",
            plan_no="PO1",
            create_time=datetime(2026, 10, 1, 10, 0, 0),
            copy_seq_no=7,
        )
    )
    db.commit()

    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="p1") == 7
    assert db.query(PlPlan).get("p1").copy_seq_no == 7


def test_allocate_missing_plan_returns_none():
    db = _session()
    assert allocate_or_get_plan_copy_seq(db, tenant_no=TENANT, plan_id="missing") is None

• Step 2: 跑测试确认失败

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

Expected: FAIL（allocate_or_get_plan_copy_seq 未定义或 ImportError）

• Step 3: 实现发号函数

在 pl_driver_plan_service.py 顶部确保有：

from sqlalchemy import and_, func, or_

（已有 func 则不重复。）增加：

def allocate_or_get_plan_copy_seq(
    db: Session, *, tenant_no: str, plan_id: str
) -> Optional[int]:
    """
    返回计划的 copy_seq_no；若为空则按租户 MAX(含软删)+1 写入后返回。
    调用方负责 commit。计划不存在返回 None。
    """
    plan = (
        db.query(PlPlan)
        .filter(PlPlan.id == plan_id, PlPlan.tenant_no == tenant_no)
        .with_for_update()
        .first()
    )
    if not plan:
        return None
    if plan.copy_seq_no is not None:
        return int(plan.copy_seq_no)

    max_seq = (
        db.query(func.max(PlPlan.copy_seq_no))
        .filter(PlPlan.tenant_no == tenant_no)
        .with_for_update()
        .scalar()
    )
    next_seq = int(max_seq or 0) + 1
    plan.copy_seq_no = next_seq
    db.flush()
    return next_seq


def ensure_copy_seq_for_driver(
    db: Session, *, tenant_no: str, plan_id: str, personnel_id: str
) -> Tuple[Optional[int], Optional[str]]:
    """
    司机可见性与 GET 详情一致；通过后分配/返回 copy_seq_no。
    成功返回 (seq, None)；失败返回 (None, err)。
    """
    plan, transport = resolve_transport_for_driver_plan(
        db, tenant_no=tenant_no, plan_id=plan_id, personnel_id=personnel_id
    )
    if not plan:
        return None, "计划不存在"
    if not transport:
        return None, "无权限查看该运输计划或未安排运输"
    seq = allocate_or_get_plan_copy_seq(db, tenant_no=tenant_no, plan_id=plan.id)
    if seq is None:
        return None, "计划不存在"
    return seq, None

注意：allocate_or_get 不要过滤 del_flag=0（回填含软删；MAX 含软删）。可见性仍由 resolve_transport_for_driver_plan 限制（正常司机只能复制未删计划）。

• Step 4: 跑测试确认通过

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

Expected: PASS（若 SQLite 对 with_for_update 报错，可在 allocate 内对 SQLite dialect 跳过 with_for_update，或改用 .with_for_update(read=True)；优先保持与 bx_codegen.allocate_reimburse_title_seq 一致，该项目 SQLite 已能跑通。）

• Step 5: Commit

git add backend/app/services/pl_driver_plan_service.py backend/tests/test_pl_plan_copy_seq.py
git commit -m "$(cat <<'EOF'
feat(pl): allocate copy_seq_no on first ensure for driver plan

Add tenant MAX+1 allocation with row lock; cover soft-delete and
idempotent reuse in unit tests.
EOF
)"

---

Task 3: API 路由

Files:

• Modify: backend/app/api/pl/driver_plan.py

• [ ] Step 1: 增加路由

在 get_driver_plan_detail 之后、feedback 之前插入：

@router.post("/{plan_id}/ensure-copy-seq")
def ensure_driver_plan_copy_seq(
    plan_id: str,
    db: Session = Depends(get_db),
    user: DbUserContext = Depends(get_db_current_user),
):
    tenant_no = get_tenant_no(db, user.to_tenant_user_dict())
    personnel = _resolve_personnel_or_empty(db, user, tenant_no)
    if not personnel:
        raise HTTPException(status_code=404, detail="未关联驾押人员，无法查看")
    seq, err = drv_svc.ensure_copy_seq_for_driver(
        db,
        tenant_no=tenant_no,
        plan_id=plan_id,
        personnel_id=personnel.id,
    )
    if err or seq is None:
        raise HTTPException(status_code=404, detail=err or "计划不存在")
    db.commit()
    return success_response({"copySeqNo": seq})

• Step 2: 冒烟（可选，有本地库时）

用已绑定驾押人员的 token 调用：

# 示例：替换 TOKEN / PLAN_ID
curl -s -X POST "http://127.0.0.1:8000/api/transport/driver-plan/{PLAN_ID}/ensure-copy-seq" \
  -H "Authorization: Bearer {TOKEN}"

Expected JSON：code 成功且 data.copySeqNo 为整数；再调一次号不变。

• Step 3: Commit

git add backend/app/api/pl/driver_plan.py
git commit -m "$(cat <<'EOF'
feat(pl): expose POST driver-plan ensure-copy-seq

Drivers can allocate or fetch the fixed copy-to-group serial before
clipboard copy.
EOF
)"

---

Task 4: 小程序 API + 复制文案

Files:

• Modify: miniprogram/api/pl.js

• Modify: miniprogram/utils/plDriverPlanCopy.js

• [ ] Step 1: API 方法

在 driverPlanApi 中 get 旁增加：

  ensureCopySeq: async (planId) =>
    unwrapData(await db.post(`/transport/driver-plan/${planId}/ensure-copy-seq`, {})),

• Step 2: 文案首行

改 buildDriverPlanCopyText，在 lines 最前插入序号（支持 copySeqNo / copy_seq_no）：

export function buildDriverPlanCopyText(page = {}) {
  const stops = Array.isArray(page?.stops) ? page.stops : []
  const loadStop = stops.find((s) => stopTypeOf(s) === 'load')
  const deliveryStops = stops.filter((s) => stopTypeOf(s) === 'delivery')

  const rawSeq = page?.copySeqNo ?? page?.copy_seq_no
  const seqText =
    rawSeq != null && rawSeq !== '' ? String(rawSeq) : DASH

  const lines = [
    `序号：${seqText}`,
    `计划编号：${String(page?.planNo || page?.plan_no || '').trim() || DASH}`,
    `装货方：${loadStop ? customerNameOf(loadStop) : DASH}`
  ]
  // ... 其余逻辑保持不变

• Step 3: Commit

git add miniprogram/api/pl.js miniprogram/utils/plDriverPlanCopy.js
git commit -m "$(cat <<'EOF'
feat(mini): include copySeqNo in driver plan clipboard text

Add ensureCopySeq API helper and prefix 序号 line above plan number.
EOF
)"

---

Task 5: 列表 / 详情复制交互

Files:

• Modify: miniprogram/pages/pl/driver-plan/list.vue

• Modify: miniprogram/pages/pl/driver-plan/detail.vue

• [ ] Step 1: 列表 handleCopy

替换为（ensure 失败则中止，不写剪贴板）：

async function handleCopy(row) {
  const id = row?.planId || row?.plan_id
  if (!id) {
    uni.showToast({ title: '计划无效', icon: 'none' })
    return
  }
  uni.showLoading({ title: '加载中...', mask: true })
  try {
    const page = (await driverPlanApi.get(id)) || {}
    const ensured = await driverPlanApi.ensureCopySeq(id)
    const copySeqNo = ensured?.copySeqNo ?? ensured?.copy_seq_no
    if (copySeqNo == null || copySeqNo === '') {
      throw new Error('no seq')
    }
    uni.hideLoading()
    await copyDriverPlanToClipboard({ ...page, copySeqNo })
  } catch (e) {
    uni.hideLoading()
    uni.showToast({ title: '复制失败', icon: 'none' })
  }
}

• Step 2: 详情保存成功复制

在 promptCopyAfterSave 的 res.confirm 分支中，先 ensure 再复制：

        if (res.confirm) {
          try {
            const ensured = await driverPlanApi.ensureCopySeq(planId.value)
            const copySeqNo = ensured?.copySeqNo ?? ensured?.copy_seq_no
            if (copySeqNo == null || copySeqNo === '') {
              uni.showToast({ title: '复制失败', icon: 'none' })
              resolve()
              return
            }
            const ok = await copyDriverPlanToClipboard({
              ...(page.value || {}),
              copySeqNo
            })
            if (ok) {
              setTimeout(() => goList(), 400)
            }
          } catch (e) {
            uni.showToast({ title: '复制失败', icon: 'none' })
          }
        }

确认 detail.vue 已 import { driverPlanApi } from '@/api/pl'（或现有等价导入）。

• Step 3: 手工验收清单

1. 执行迁移后，任选两条历史计划首次复制：序号应符合创建时间先后，且再次复制不变。
2. 新建计划（不复制）：库中 copy_seq_no 为 NULL。
3. 对该新计划点「复制发群」：文案首行 序号：{MAX+1}，库中已写入。
4. 列表与回执弹窗两条路径均成功；UI 上无序号展示。
5. ensure 失败（断网/无权限）toast「复制失败」，剪贴板不更新为半成品（或保持原内容）。

• Step 4: Commit

git add miniprogram/pages/pl/driver-plan/list.vue miniprogram/pages/pl/driver-plan/detail.vue
git commit -m "$(cat <<'EOF'
feat(mini): ensure copy_seq before driver plan share-to-group

Call ensure-copy-seq on list and post-save copy; abort clipboard on
failure so drivers always get a fixed serial when copy succeeds.
EOF
)"

---

Task 6: 图谱更新 + 设计状态

Files:

• Modify: docs/superpowers/specs/2026-10-09-pl-plan-copy-seq-design.md（状态改为「已确认」）

• Run: ./scripts/update-graph.sh backend（若脚本支持）或仓库约定的 graphify 更新

• [ ] Step 1: 将设计文档头部 状态: 待用户确认 改为 已确认。

• [ ] Step 2: 代码改动后更新 backend 图谱：

./scripts/update-graph.sh backend

若脚本不存在，则在仓库根目录：graphify update backend（以 AGENTS.md / graphify.mdc 为准）。

• Step 3: Commit

git add docs/superpowers/specs/2026-10-09-pl-plan-copy-seq-design.md
# 若有 graphify 产出一并 add
git commit -m "$(cat <<'EOF'
docs(pl): mark copy-seq design confirmed and refresh graph

EOF
)"

---

Spec coverage checklist

Spec 项	Task
D1 租户累加	Task 2 MAX+1
D2 不回收 / 含软删	Task 1 回填 + Task 2 测试
D3 仅复制文案	Task 4–5（无 UI）
D4 首次复制发号	Task 3–5
D5 历史回填	Task 1 SQL
D6 / D7 文案格式	Task 4
D8 不用 GET 发号	Task 3 专用 POST
ensure 失败中止复制	Task 5
加密排除	无需改（前缀已有）

Placeholder / consistency self-review

• 函数名统一：allocate_or_get_plan_copy_seq、ensure_copy_seq_for_driver、响应键 copySeqNo。
• 迁移文件名与仓库惯例一致：docs/sql/migrate_*_YYYYMMDD.sql（非 migrations/ 子目录）。
• 无 TBD；SQLite with_for_update 风险已在 Task 2 Step 4 写明处理方式。