生产任务集合 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: 两表（集合头 + 条目：任务名称/可选工艺路线/描述/备注）。CRUD 挂 production.py。派发 POST /production-tasks/from-set 在同一事务内循环调用现有 create_production_task。前端 P01 维护页 + DispatchTaskDialog 增加「任务集合」方式 + 工单「存为任务集合」。

Tech Stack: FastAPI + SQLAlchemy + MySQL 5.7；Vue 3 + Element Plus；权限 useMenuPerm / require_menu_code

Spec: docs/superpowers/specs/2026-08-18-production-task-set-design.md

File map

文件	职责
Create: `docs/sql/alter_biz_production_task_set.sql`	两表 DDL + 菜单 + 角色授权（幂等）
Modify: `待同步到生产的sql脚本`	追加上线文件名
Modify: `backend/app/models/models.py`	`BizProductionTaskSet`、`BizProductionTaskSetItem`
Create: `backend/app/crud/production_task_set.py`	纯函数 + CRUD + from-work-order + from-set
Create: `backend/tests/test_production_task_set_helpers.py`	条目规范化 / 空名跳过 / 缺失路线
Create: `backend/app/schemas/production_task_set.py`	Pydantic body
Modify: `backend/app/api/production.py`	路由注册（from-set 必须写在 `/{task_id}` 之前）
Create: `web-admin/src/api/production-task-set.ts`	集合 API
Modify: `web-admin/src/api/production-task.ts`	`fetchCreateProductionTasksFromSet`
Create: `web-admin/src/utils/production-task-set.ts`	工单任务 → 条目预览映射
Create: `web-admin/src/utils/production-task-set.test.ts`	映射单测
Create: `web-admin/src/views/production/production-task-set/index.vue`	P01 维护页
Modify: `web-admin/src/router/modules/sqxp.ts`	静态路由
Modify: `web-admin/src/views/production/production-work-order/DispatchTaskDialog.vue`	派发方式
Create: `web-admin/src/views/production/production-work-order/SaveProductionTaskSetDialog.vue`	存为集合弹窗
Modify: `ProductionWorkOrderCardList.vue`、`index.vue`、`tabs/ProductionWorkOrderTabPanel.vue`	菜单项与命令
`./scripts/update-graph.sh`	实现后增量更新图谱

常量（全任务统一）

常量	值
头表	`biz_production_task_set`
条目表	`biz_production_task_set_item`
C 菜单 id	`95400000000000000000000000000001`
F list	`95400000000000000000000000000002`
F add	`95400000000000000000000000000003`
F edit	`95400000000000000000000000000004`
F remove	`95400000000000000000000000000005`
parent_id	`PRB00000000000000000000000000001`
工单菜单 id	`90000000000000000000000000000011`
path	`production-task-set`
component	`/production/production-task-set/index`
C order_num	`12`
perms list	`production:task-set:list`
perms add	`production:task-set:add`
perms edit	`production:task-set:edit`
perms remove	`production:task-set:remove`
Hash	`/#/production-group/production-base/production-task-set`
API 集合	`/web/admin/production-task-sets`
API from-set	`POST /web/admin/production-tasks/from-set`

可用性约定（相对 spec 的落地）： SQL 将 C+F 授予「已有任意 PRB 子菜单」或「已有生产工单菜单」的角色，这样只开工单的人也能选集合 / 存为集合。GET .../active 使用 require_any_menu_code(list, production:work-order:list)。POST .../from-set 使用 production:work-order:list（与单条创建任务同权）。POST .../from-work-order 使用 production:task-set:add。

Task 1: SQL 建表 + 菜单授权

Files:

Create: docs/sql/alter_biz_production_task_set.sql
Modify: 待同步到生产的sql脚本
[ ] Step 1: 写幂等迁移脚本

创建 docs/sql/alter_biz_production_task_set.sql：

-- 生产任务集合：头表 + 条目 + 菜单授权（幂等）
SET @tenant_no = CONVERT('zkxw' USING utf8mb4) COLLATE utf8mb4_unicode_ci;
SET @prb_id = CONVERT('PRB00000000000000000000000000001' USING utf8mb4) COLLATE utf8mb4_unicode_ci;
SET @wo_menu_id = CONVERT('90000000000000000000000000000011' USING utf8mb4) COLLATE utf8mb4_unicode_ci;
SET @menu_c = CONVERT('95400000000000000000000000000001' USING utf8mb4) COLLATE utf8mb4_unicode_ci;
SET @menu_list = CONVERT('95400000000000000000000000000002' USING utf8mb4) COLLATE utf8mb4_unicode_ci;
SET @menu_add = CONVERT('95400000000000000000000000000003' USING utf8mb4) COLLATE utf8mb4_unicode_ci;
SET @menu_edit = CONVERT('95400000000000000000000000000004' USING utf8mb4) COLLATE utf8mb4_unicode_ci;
SET @menu_remove = CONVERT('95400000000000000000000000000005' USING utf8mb4) COLLATE utf8mb4_unicode_ci;

CREATE TABLE IF NOT EXISTS `biz_production_task_set` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `set_name` varchar(200) NOT NULL COMMENT '集合名称',
  `status` char(1) NOT NULL DEFAULT '1' COMMENT '1启用 0停用',
  `tenant_no` varchar(64) NOT NULL DEFAULT 'zkxw' COMMENT '租户编号',
  `order_num` int(11) DEFAULT NULL COMMENT '排序号',
  `remark` varchar(512) DEFAULT NULL COMMENT '备注',
  `create_by` varchar(64) DEFAULT NULL COMMENT '创建人id',
  `create_dept_id` varchar(64) NOT NULL DEFAULT '' COMMENT '创建人部门ID',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `idx_pts_tenant_status` (`tenant_no`,`del_flag`,`status`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='生产任务集合';

CREATE TABLE IF NOT EXISTS `biz_production_task_set_item` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `set_id` varchar(64) NOT NULL COMMENT '集合ID',
  `task_name` varchar(200) NOT NULL COMMENT '任务名称',
  `route_id` varchar(64) DEFAULT NULL COMMENT '工艺路线ID',
  `route_name` varchar(100) DEFAULT NULL COMMENT '路线名称快照',
  `description` varchar(500) DEFAULT NULL COMMENT '描述',
  `remark` varchar(512) DEFAULT NULL COMMENT '备注',
  `tenant_no` varchar(64) NOT NULL DEFAULT 'zkxw' COMMENT '租户编号',
  `order_num` int(11) DEFAULT NULL COMMENT '排序号',
  `create_by` varchar(64) DEFAULT NULL COMMENT '创建人id',
  `create_dept_id` varchar(64) NOT NULL DEFAULT '' COMMENT '创建人部门ID',
  `create_time` datetime NOT NULL COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记',
  PRIMARY KEY (`id`) USING BTREE,
  KEY `idx_ptsi_set` (`set_id`,`del_flag`,`order_num`) USING BTREE
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='生产任务集合条目';

菜单：对照 docs/sql/alter_sys_menu_add_production_waste_output.sql 的 INSERT/UPDATE 幂等写法。C 菜单：

menu_name='生产任务集合'，parent_id=@prb_id，menu_type='C'
path='production-task-set'，component='/production/production-task-set/index'
perms='production:task-set:list'，visible='1'，order_num=12

四个 F：父级 @menu_c，visible='0'，perms 分别为 list/add/edit/remove，名称「查询/新增/编辑/删除」。

角色授权：已有 PRB 下任一有效子菜单 或 工单菜单 @wo_menu_id 的角色，幂等授予 @prb_id + C + 四个 F。MySQL 5.7 禁止同句读改 sys_role_menu，用派生表，对照废弃物脚本 3a/3b。

Step 2: 待同步清单

在 待同步到生产的sql脚本 追加一行：docs/sql/alter_biz_production_task_set.sql

Step 3: Commit

git add docs/sql/alter_biz_production_task_set.sql 待同步到生产的sql脚本
git commit -m "$(cat <<'EOF'
feat: add production task set schema and menu SQL

EOF
)"

Task 2: SQLAlchemy 模型

Files:

Modify: backend/app/models/models.py（在 BizProcessRouteStepDataCollection 之后、# 生产执行 注释之前插入）
[ ] Step 1: 插入两个模型类

class BizProductionTaskSet(Base):
    """生产任务集合 biz_production_task_set"""

    __tablename__ = "biz_production_task_set"

    id = Column(String(64), primary_key=True, comment="主键ID")
    set_name = Column(String(200), nullable=False, comment="集合名称")
    status = Column(CHAR(1), nullable=False, default="1", comment="1启用 0停用")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    order_num = Column(Integer, nullable=True, comment="排序号")
    remark = Column(String(512), nullable=True, comment="备注")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")


class BizProductionTaskSetItem(Base):
    """生产任务集合条目 biz_production_task_set_item"""

    __tablename__ = "biz_production_task_set_item"

    id = Column(String(64), primary_key=True, comment="主键ID")
    set_id = Column(String(64), nullable=False, comment="集合ID")
    task_name = Column(String(200), nullable=False, comment="任务名称")
    route_id = Column(String(64), nullable=True, comment="工艺路线ID")
    route_name = Column(String(100), nullable=True, comment="路线名称快照")
    description = Column(String(500), nullable=True, comment="描述")
    remark = Column(String(512), nullable=True, comment="备注")
    tenant_no = Column(String(64), nullable=False, default="zkxw", comment="租户编号")
    order_num = Column(Integer, nullable=True, comment="排序号")
    create_by = Column(String(64), nullable=True, comment="创建人id")
    create_dept_id = Column(String(64), nullable=False, default="", comment="创建人部门ID")
    create_time = Column(DateTime, nullable=False, default=datetime.now, comment="创建时间")
    update_by = Column(String(64), nullable=True, comment="更新人id")
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now, comment="更新时间")
    del_flag = Column(SmallInteger, nullable=False, default=0, comment="删除标记")

字段对齐 BizQcInspectionItemTemplate 写法（CHAR(1) 状态、公共字段顺序）。

Step 2: Commit

git add backend/app/models/models.py
git commit -m "$(cat <<'EOF'
feat: add production task set SQLAlchemy models

EOF
)"

Task 3: 纯函数单测（TDD）

Files:

Create: backend/app/crud/production_task_set.py（本任务只放纯函数）
Create: backend/tests/test_production_task_set_helpers.py
[ ] Step 1: Write the failing test

# backend/tests/test_production_task_set_helpers.py
import unittest

from app.crud.production_task_set import (
    collect_missing_route_ids,
    normalize_item_payload,
    skip_empty_named_tasks,
)


class TestNormalizeItemPayload(unittest.TestCase):
    def test_requires_task_name(self):
        self.assertIsNone(normalize_item_payload({"taskName": "  ", "routeId": "r1"}))

    def test_keeps_optional_fields(self):
        item = normalize_item_payload(
            {
                "taskName": "分离",
                "routeId": " r1 ",
                "description": " d ",
                "remark": " m ",
            }
        )
        self.assertEqual(
            item,
            {
                "task_name": "分离",
                "route_id": "r1",
                "description": "d",
                "remark": "m",
            },
        )

    def test_empty_route_becomes_none(self):
        item = normalize_item_payload({"taskName": "培养", "routeId": ""})
        self.assertEqual(item["route_id"], None)
        self.assertEqual(item["description"], None)
        self.assertEqual(item["remark"], None)


class TestSkipEmptyNamedTasks(unittest.TestCase):
    def test_skips_blank_names_and_counts(self):
        kept, skipped = skip_empty_named_tasks(
            [
                {"task_name": "A", "route_id": "r1", "description": None, "remark": "x"},
                {"task_name": "  ", "route_id": "r2"},
                {"task_name": None, "route_id": None},
            ]
        )
        self.assertEqual(skipped, 2)
        self.assertEqual(len(kept), 1)
        self.assertEqual(kept[0]["task_name"], "A")
        self.assertEqual(kept[0]["remark"], "x")


class TestCollectMissingRouteIds(unittest.TestCase):
    def test_empty_route_not_missing(self):
        self.assertEqual(
            collect_missing_route_ids(
                [{"route_id": None}, {"route_id": ""}],
                {"alive"},
            ),
            [],
        )

    def test_lists_missing_preserving_order_unique(self):
        self.assertEqual(
            collect_missing_route_ids(
                [{"route_id": "gone"}, {"route_id": "alive"}, {"route_id": "gone"}],
                {"alive"},
            ),
            ["gone"],
        )

Step 2: Run test to verify it fails

cd backend && source venv/bin/activate && python -m pytest tests/test_production_task_set_helpers.py -v

Expected: FAIL（模块或函数不存在）

Step 3: Write minimal implementation

在 backend/app/crud/production_task_set.py：

"""生产任务集合 CRUD"""
from __future__ import annotations

from typing import Any, Dict, List, Optional, Set, Tuple


def _strip_or_none(raw: Any) -> Optional[str]:
    text = str(raw or "").strip()
    return text or None


def normalize_item_payload(raw: Optional[Dict[str, Any]]) -> Optional[Dict[str, Any]]:
    data = raw or {}
    task_name = str(data.get("taskName") or data.get("task_name") or "").strip()
    if not task_name:
        return None
    return {
        "task_name": task_name,
        "route_id": _strip_or_none(data.get("routeId") if "routeId" in data else data.get("route_id")),
        "description": _strip_or_none(data.get("description")),
        "remark": _strip_or_none(data.get("remark")),
    }


def skip_empty_named_tasks(tasks: List[Dict[str, Any]]) -> Tuple[List[Dict[str, Any]], int]:
    kept: List[Dict[str, Any]] = []
    skipped = 0
    for row in tasks or []:
        name = str(row.get("task_name") or row.get("taskName") or "").strip()
        if not name:
            skipped += 1
            continue
        kept.append(
            {
                "task_name": name,
                "route_id": _strip_or_none(row.get("route_id") if "route_id" in row else row.get("routeId")),
                "route_name": _strip_or_none(row.get("route_name") if "route_name" in row else row.get("routeName")),
                "description": _strip_or_none(row.get("description")),
                "remark": _strip_or_none(row.get("remark")),
            }
        )
    return kept, skipped


def collect_missing_route_ids(items: List[Dict[str, Any]], existing_ids: Set[str]) -> List[str]:
    missing: List[str] = []
    seen: Set[str] = set()
    alive = existing_ids or set()
    for item in items or []:
        rid = _strip_or_none(item.get("route_id"))
        if not rid or rid in alive or rid in seen:
            continue
        seen.add(rid)
        missing.append(rid)
    return missing

Step 4: Run tests to verify they pass

cd backend && source venv/bin/activate && python -m pytest tests/test_production_task_set_helpers.py -v

Expected: PASS

Step 5: Commit

git add backend/app/crud/production_task_set.py backend/tests/test_production_task_set_helpers.py
git commit -m "$(cat <<'EOF'
test: add production task set item helper coverage

EOF
)"

Task 4: CRUD 实现

Files:

Modify: backend/app/crud/production_task_set.py

对照 backend/app/crud/qc_inspection_item_template.py 的租户过滤、名称唯一、_row_to_dict。CRUD 函数只 flush，不 commit（与 create_production_task 一致，便于 from-set 整单回滚）。

Step 1: 实现序列化、唯一名、条目写入、CRUD

在同一文件继续追加（保留 Task 3 纯函数）：

from datetime import datetime

from sqlalchemy import func
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import (
    BizProcessRoute,
    BizProductionTask,
    BizProductionTaskSet,
    BizProductionTaskSetItem,
    BizProductionWorkOrder,
)


def _item_to_dict(r: BizProductionTaskSetItem) -> Dict[str, Any]:
    return {
        "id": r.id,
        "setId": r.set_id,
        "taskName": r.task_name or "",
        "routeId": r.route_id or "",
        "routeName": r.route_name or "",
        "description": r.description or "",
        "remark": r.remark or "",
        "orderNum": r.order_num if r.order_num is not None else 0,
    }


def _set_to_dict(r: BizProductionTaskSet, item_count: int = 0, items: Optional[List[Dict[str, Any]]] = None) -> Dict[str, Any]:
    data: Dict[str, Any] = {
        "id": r.id,
        "setName": r.set_name or "",
        "status": r.status or "1",
        "itemCount": item_count,
        "orderNum": r.order_num if r.order_num is not None else 0,
        "remark": r.remark or "",
        "createTime": r.create_time.strftime("%Y-%m-%d %H:%M:%S") if r.create_time else "",
        "updateTime": r.update_time.strftime("%Y-%m-%d %H:%M:%S") if r.update_time else "",
    }
    if items is not None:
        data["items"] = items
    return data


def set_name_exists(db: Session, name: str, tenant_no: str, exclude_id: Optional[str] = None) -> bool:
    n = (name or "").strip()
    if not n:
        return False
    q = db.query(BizProductionTaskSet).filter(
        BizProductionTaskSet.set_name == n,
        BizProductionTaskSet.tenant_no == tenant_no,
        BizProductionTaskSet.del_flag == 0,
    )
    if exclude_id:
        q = q.filter(BizProductionTaskSet.id != exclude_id)
    return q.first() is not None


def _require_status(status: Optional[str]) -> str:
    st = (status or "1").strip()
    if st not in ("0", "1"):
        raise ValueError("状态无效")
    return st


def _load_route_name_map(db: Session, tenant_no: str, route_ids: List[str]) -> Dict[str, str]:
    ids = [rid for rid in route_ids if rid]
    if not ids:
        return {}
    rows = (
        db.query(BizProcessRoute)
        .filter(
            BizProcessRoute.tenant_no == tenant_no,
            BizProcessRoute.del_flag == 0,
            BizProcessRoute.id.in_(ids),
        )
        .all()
    )
    return {r.id: (r.route_name or "") for r in rows}


def _normalize_items_or_raise(raw_items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    out: List[Dict[str, Any]] = []
    for raw in raw_items or []:
        item = normalize_item_payload(raw)
        if item:
            out.append(item)
    if not out:
        raise ValueError("请至少添加一条任务名称非空的条目")
    return out


def _assert_routes_exist(db: Session, tenant_no: str, items: List[Dict[str, Any]]) -> Dict[str, str]:
    wanted = [i["route_id"] for i in items if i.get("route_id")]
    name_map = _load_route_name_map(db, tenant_no, wanted)
    missing = collect_missing_route_ids(items, set(name_map.keys()))
    if missing:
        raise ValueError("存在无效或已删除的工艺路线")
    return name_map


def _replace_items(
    db: Session,
    *,
    set_id: str,
    tenant_no: str,
    items: List[Dict[str, Any]],
    route_names: Dict[str, str],
    user_id: Optional[str],
    create_dept_id: str,
) -> None:
    now = datetime.now()
    existing = (
        db.query(BizProductionTaskSetItem)
        .filter(
            BizProductionTaskSetItem.set_id == set_id,
            BizProductionTaskSetItem.tenant_no == tenant_no,
            BizProductionTaskSetItem.del_flag == 0,
        )
        .all()
    )
    for row in existing:
        row.del_flag = 1
        row.update_by = user_id
        row.update_time = now
        db.add(row)
    for idx, item in enumerate(items, start=1):
        rid = item.get("route_id")
        db.add(
            BizProductionTaskSetItem(
                id=generate_uuid(),
                set_id=set_id,
                task_name=item["task_name"],
                route_id=rid,
                route_name=route_names.get(rid or "") if rid else None,
                description=item.get("description"),
                remark=item.get("remark"),
                tenant_no=tenant_no,
                order_num=idx,
                create_by=user_id,
                create_dept_id=create_dept_id or "",
                create_time=now,
                update_by=user_id,
                update_time=now,
                del_flag=0,
            )
        )


def _item_count_map(db: Session, tenant_no: str, set_ids: List[str]) -> Dict[str, int]:
    if not set_ids:
        return {}
    rows = (
        db.query(BizProductionTaskSetItem.set_id, func.count(BizProductionTaskSetItem.id))
        .filter(
            BizProductionTaskSetItem.tenant_no == tenant_no,
            BizProductionTaskSetItem.del_flag == 0,
            BizProductionTaskSetItem.set_id.in_(set_ids),
        )
        .group_by(BizProductionTaskSetItem.set_id)
        .all()
    )
    return {sid: int(cnt) for sid, cnt in rows}


def list_sets(
    db: Session,
    *,
    tenant_no: str,
    page: int = 1,
    page_size: int = 10,
    keyword: Optional[str] = None,
    status: Optional[str] = None,
) -> Tuple[List[Dict[str, Any]], int]:
    q = db.query(BizProductionTaskSet).filter(
        BizProductionTaskSet.tenant_no == tenant_no,
        BizProductionTaskSet.del_flag == 0,
    )
    kw = (keyword or "").strip()
    if kw:
        q = q.filter(BizProductionTaskSet.set_name.like(f"%{kw}%"))
    st = (status or "").strip()
    if st in ("0", "1"):
        q = q.filter(BizProductionTaskSet.status == st)
    total = q.count()
    rows = (
        q.order_by(BizProductionTaskSet.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    counts = _item_count_map(db, tenant_no, [r.id for r in rows])
    return [_set_to_dict(r, item_count=counts.get(r.id, 0)) for r in rows], int(total)


def list_active_sets(db: Session, *, tenant_no: str) -> List[Dict[str, Any]]:
    rows = (
        db.query(BizProductionTaskSet)
        .filter(
            BizProductionTaskSet.tenant_no == tenant_no,
            BizProductionTaskSet.del_flag == 0,
            BizProductionTaskSet.status == "1",
        )
        .order_by(BizProductionTaskSet.create_time.desc())
        .all()
    )
    counts = _item_count_map(db, tenant_no, [r.id for r in rows])
    return [_set_to_dict(r, item_count=counts.get(r.id, 0)) for r in rows]


def get_set(db: Session, set_id: str, tenant_no: str) -> Optional[Dict[str, Any]]:
    row = (
        db.query(BizProductionTaskSet)
        .filter(
            BizProductionTaskSet.id == set_id,
            BizProductionTaskSet.tenant_no == tenant_no,
            BizProductionTaskSet.del_flag == 0,
        )
        .first()
    )
    if not row:
        return None
    items = (
        db.query(BizProductionTaskSetItem)
        .filter(
            BizProductionTaskSetItem.set_id == set_id,
            BizProductionTaskSetItem.tenant_no == tenant_no,
            BizProductionTaskSetItem.del_flag == 0,
        )
        .order_by(BizProductionTaskSetItem.order_num.asc(), BizProductionTaskSetItem.create_time.asc())
        .all()
    )
    return _set_to_dict(row, item_count=len(items), items=[_item_to_dict(i) for i in items])


def create_set(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    set_name: str,
    status: str = "1",
    remark: Optional[str] = None,
    items: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    name = (set_name or "").strip()
    if not name:
        raise ValueError("集合名称不能为空")
    if set_name_exists(db, name, tenant_no):
        raise ValueError("集合名称已存在")
    normalized = _normalize_items_or_raise(items or [])
    route_names = _assert_routes_exist(db, tenant_no, normalized)
    now = datetime.now()
    obj = BizProductionTaskSet(
        id=generate_uuid(),
        set_name=name,
        status=_require_status(status),
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id=create_dept_id or "",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.flush()
    _replace_items(
        db,
        set_id=obj.id,
        tenant_no=tenant_no,
        items=normalized,
        route_names=route_names,
        user_id=user_id,
        create_dept_id=create_dept_id,
    )
    db.flush()
    detail = get_set(db, obj.id, tenant_no)
    if not detail:
        raise ValueError("集合创建失败")
    return detail


def update_set(
    db: Session,
    set_id: str,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    set_name: Optional[str] = None,
    status: Optional[str] = None,
    remark: Optional[str] = None,
    items: Optional[List[Dict[str, Any]]] = None,
) -> Dict[str, Any]:
    obj = (
        db.query(BizProductionTaskSet)
        .filter(
            BizProductionTaskSet.id == set_id,
            BizProductionTaskSet.tenant_no == tenant_no,
            BizProductionTaskSet.del_flag == 0,
        )
        .first()
    )
    if not obj:
        raise ValueError("集合不存在")
    if set_name is not None:
        name = set_name.strip()
        if not name:
            raise ValueError("集合名称不能为空")
        if set_name_exists(db, name, tenant_no, exclude_id=set_id):
            raise ValueError("集合名称已存在")
        obj.set_name = name
    if status is not None:
        obj.status = _require_status(status)
    if remark is not None:
        obj.remark = remark.strip() or None
    if items is not None:
        normalized = _normalize_items_or_raise(items)
        route_names = _assert_routes_exist(db, tenant_no, normalized)
        _replace_items(
            db,
            set_id=obj.id,
            tenant_no=tenant_no,
            items=normalized,
            route_names=route_names,
            user_id=user_id,
            create_dept_id=obj.create_dept_id or "",
        )
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.flush()
    detail = get_set(db, obj.id, tenant_no)
    if not detail:
        raise ValueError("集合不存在")
    return detail


def delete_set(db: Session, set_id: str, *, tenant_no: str, user_id: Optional[str] = None) -> None:
    obj = (
        db.query(BizProductionTaskSet)
        .filter(
            BizProductionTaskSet.id == set_id,
            BizProductionTaskSet.tenant_no == tenant_no,
            BizProductionTaskSet.del_flag == 0,
        )
        .first()
    )
    if not obj:
        raise ValueError("集合不存在")
    now = datetime.now()
    obj.del_flag = 1
    obj.update_by = user_id
    obj.update_time = now
    db.add(obj)
    items = (
        db.query(BizProductionTaskSetItem)
        .filter(
            BizProductionTaskSetItem.set_id == set_id,
            BizProductionTaskSetItem.del_flag == 0,
        )
        .all()
    )
    for row in items:
        row.del_flag = 1
        row.update_by = user_id
        row.update_time = now
        db.add(row)
    db.flush()


def create_set_from_work_order(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    work_order_id: str,
    set_name: str,
    remark: Optional[str] = None,
) -> Dict[str, Any]:
    wo_id = (work_order_id or "").strip()
    wo = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == wo_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if not wo:
        raise ValueError("工单不存在")
    tasks = (
        db.query(BizProductionTask)
        .filter(
            BizProductionTask.work_order_id == wo_id,
            BizProductionTask.tenant_no == tenant_no,
            BizProductionTask.del_flag == 0,
        )
        .order_by(BizProductionTask.create_time.asc(), BizProductionTask.id.asc())
        .all()
    )
    raw = [
        {
            "task_name": t.task_name,
            "route_id": t.route_id,
            "description": t.description,
            "remark": t.remark,
        }
        for t in tasks
    ]
    kept, skipped = skip_empty_named_tasks(raw)
    if not kept:
        raise ValueError("无可保存的生产任务（任务名称均为空）")
    route_ids = [i["route_id"] for i in kept if i.get("route_id")]
    name_map = _load_route_name_map(db, tenant_no, route_ids)
    payload_items = []
    for i in kept:
        rid = i.get("route_id")
        if rid and rid not in name_map:
            rid = None
        payload_items.append(
            {
                "taskName": i["task_name"],
                "routeId": rid,
                "description": i.get("description"),
                "remark": i.get("remark"),
            }
        )
    detail = create_set(
        db,
        tenant_no=tenant_no,
        user_id=user_id,
        create_dept_id=create_dept_id,
        set_name=set_name,
        status="1",
        remark=remark,
        items=payload_items,
    )
    return {**detail, "skippedEmptyNameCount": skipped}

create_tasks_from_set：

def create_tasks_from_set(
    db: Session,
    *,
    tenant_no: str,
    user_id: Optional[str] = None,
    create_dept_id: str = "",
    work_order_id: str,
    set_id: str,
) -> Dict[str, Any]:
    from app.crud import production_task as production_task_svc

    wo_id = (work_order_id or "").strip()
    wo = (
        db.query(BizProductionWorkOrder)
        .filter(
            BizProductionWorkOrder.id == wo_id,
            BizProductionWorkOrder.tenant_no == tenant_no,
            BizProductionWorkOrder.del_flag == 0,
        )
        .first()
    )
    if not wo:
        raise ValueError("工单不存在")
    status = getattr(wo, "work_order_status", None) or ""
    if status not in ("dpz", "jxz"):
        raise ValueError("仅待配置或进行中的工单可派发任务")
    header = (
        db.query(BizProductionTaskSet)
        .filter(
            BizProductionTaskSet.id == (set_id or "").strip(),
            BizProductionTaskSet.tenant_no == tenant_no,
            BizProductionTaskSet.del_flag == 0,
        )
        .first()
    )
    if not header:
        raise ValueError("集合不存在")
    if (header.status or "1") != "1":
        raise ValueError("集合已停用")
    items = (
        db.query(BizProductionTaskSetItem)
        .filter(
            BizProductionTaskSetItem.set_id == header.id,
            BizProductionTaskSetItem.tenant_no == tenant_no,
            BizProductionTaskSetItem.del_flag == 0,
        )
        .order_by(BizProductionTaskSetItem.order_num.asc(), BizProductionTaskSetItem.create_time.asc())
        .all()
    )
    if not items:
        raise ValueError("集合没有可派发的任务")
    payloads = [
        {
            "route_id": row.route_id,
            "task_name": row.task_name,
            "description": row.description,
            "remark": row.remark,
        }
        for row in items
    ]
    name_map = _load_route_name_map(db, tenant_no, [p["route_id"] for p in payloads if p.get("route_id")])
    missing = collect_missing_route_ids(payloads, set(name_map.keys()))
    if missing:
        raise ValueError("工艺路线已删除，无法派发：" + "、".join(missing))
    created = []
    for row in items:
        task = production_task_svc.create_production_task(
            db,
            tenant_no=tenant_no,
            user_id=user_id,
            create_dept_id=create_dept_id,
            data={
                "work_order_id": wo_id,
                "task_name": row.task_name,
                "route_id": row.route_id,
                "description": row.description,
                "remark": row.remark,
            },
        )
        created.append(production_task_svc._task_to_dict(task))
    return {"createdCount": len(created), "list": created}

增加辅助：

def total_pages(total: int, page_size: int) -> int:
    if page_size <= 0:
        return 0
    return (int(total) + int(page_size) - 1) // int(page_size)

Step 2: 再跑 Task 3 单测，确认纯函数未破坏

cd backend && source venv/bin/activate && python -m pytest tests/test_production_task_set_helpers.py -v

Expected: PASS

Step 3: Commit

git add backend/app/crud/production_task_set.py
git commit -m "$(cat <<'EOF'
feat: implement production task set CRUD and batch dispatch

EOF
)"

Task 5: Schema + API 路由

Files:

Create: backend/app/schemas/production_task_set.py
Modify: backend/app/api/production.py
[ ] Step 1: Pydantic schemas

# backend/app/schemas/production_task_set.py
from typing import List, Optional

from pydantic import BaseModel, ConfigDict, Field


class ProductionTaskSetItemBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    task_name: str = Field(..., alias="taskName")
    route_id: Optional[str] = Field(None, alias="routeId")
    description: Optional[str] = None
    remark: Optional[str] = None


class ProductionTaskSetCreateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    set_name: str = Field(..., alias="setName")
    status: Optional[str] = "1"
    remark: Optional[str] = None
    items: List[ProductionTaskSetItemBody] = []


class ProductionTaskSetUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    set_name: Optional[str] = Field(None, alias="setName")
    status: Optional[str] = None
    remark: Optional[str] = None
    items: Optional[List[ProductionTaskSetItemBody]] = None


class ProductionTaskSetFromWorkOrderBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    work_order_id: str = Field(..., alias="workOrderId")
    set_name: str = Field(..., alias="setName")
    remark: Optional[str] = None


class ProductionTasksFromSetBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    work_order_id: str = Field(..., alias="workOrderId")
    set_id: str = Field(..., alias="setId")

Step 2: 注册路由

在 production.py 增加：

from app.crud import production_task_set as production_task_set_svc
from app.schemas.production_task_set import (
    ProductionTaskSetCreateBody,
    ProductionTaskSetFromWorkOrderBody,
    ProductionTaskSetUpdateBody,
    ProductionTasksFromSetBody,
)

POST /production-tasks/from-set 必须插在现有 @router.post("/production-tasks") 之后、任何 /production-tasks/{task_id} 之前。

@router.post("/production-tasks/from-set")
async def create_production_tasks_from_set(
    body: ProductionTasksFromSetBody,
    user: dict = Depends(require_menu_code("production:work-order:list")),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        result = production_task_set_svc.create_tasks_from_set(
            db,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
            create_dept_id="",
            work_order_id=body.work_order_id,
            set_id=body.set_id,
        )
        db.commit()
        return success_response(result, message="创建成功")
    except ValueError as e:
        db.rollback()
        return error_response(400, str(e))

集合路由（/active 与 /from-work-order 在 /{set_id} 之前）：

@router.get("/production-task-sets")
async def list_production_task_sets(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    keyword: Optional[str] = None,
    status: Optional[str] = None,
    user: dict = Depends(require_menu_code("production:task-set:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    items, total = production_task_set_svc.list_sets(
        db, tenant_no=tenant_no, page=page, page_size=pageSize, keyword=keyword, status=status,
    )
    tp = production_task_set_svc.total_pages(total, pageSize)
    return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})


@router.get("/production-task-sets/active")
async def list_active_production_task_sets(
    user: dict = Depends(require_any_menu_code("production:task-set:list", "production:work-order:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    items = production_task_set_svc.list_active_sets(db, tenant_no=tenant_no)
    return success_response({"list": items})


@router.post("/production-task-sets/from-work-order")
async def create_production_task_set_from_work_order(
    body: ProductionTaskSetFromWorkOrderBody,
    user: dict = Depends(require_menu_code("production:task-set:add")),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        detail = production_task_set_svc.create_set_from_work_order(
            db,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
            work_order_id=body.work_order_id,
            set_name=body.set_name,
            remark=body.remark,
        )
        db.commit()
        return success_response(detail, message="保存成功")
    except ValueError as e:
        db.rollback()
        return error_response(400, str(e))


@router.get("/production-task-sets/{set_id}")
async def get_production_task_set(
    set_id: str,
    user: dict = Depends(require_any_menu_code("production:task-set:list", "production:work-order:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    detail = production_task_set_svc.get_set(db, set_id, tenant_no)
    if not detail:
        return error_response(404, "集合不存在")
    return success_response(detail)


@router.post("/production-task-sets")
async def create_production_task_set(
    body: ProductionTaskSetCreateBody,
    user: dict = Depends(require_menu_code("production:task-set:add")),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        detail = production_task_set_svc.create_set(
            db,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
            set_name=body.set_name,
            status=body.status or "1",
            remark=body.remark,
            items=[x.model_dump(by_alias=False) for x in body.items],
        )
        db.commit()
        return success_response(detail, message="创建成功")
    except ValueError as e:
        db.rollback()
        return error_response(400, str(e))


@router.put("/production-task-sets/{set_id}")
async def update_production_task_set(
    set_id: str,
    body: ProductionTaskSetUpdateBody,
    user: dict = Depends(require_menu_code("production:task-set:edit")),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        data = body.model_dump(by_alias=False, exclude_unset=True)
        items = None
        if "items" in data:
            items = [x.model_dump(by_alias=False) for x in (body.items or [])]
        detail = production_task_set_svc.update_set(
            db,
            set_id,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
            set_name=data.get("set_name"),
            status=data.get("status"),
            remark=data.get("remark"),
            items=items,
        )
        db.commit()
        return success_response(detail, message="更新成功")
    except ValueError as e:
        db.rollback()
        return error_response(400, str(e))


@router.delete("/production-task-sets/{set_id}")
async def delete_production_task_set(
    set_id: str,
    user: dict = Depends(require_menu_code("production:task-set:remove")),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        production_task_set_svc.delete_set(db, set_id, tenant_no=tenant_no, user_id=user.get("user_id"))
        db.commit()
        return success_response(None, message="删除成功")
    except ValueError as e:
        db.rollback()
        return error_response(400, str(e))

normalize_item_payload 同时识别 task_name / taskName。

Step 3: Commit

git add backend/app/schemas/production_task_set.py backend/app/api/production.py
git commit -m "$(cat <<'EOF'
feat: expose production task set REST APIs

EOF
)"

Task 6: 前端 API 与映射工具（TDD）

Files:

Create: web-admin/src/api/production-task-set.ts
Modify: web-admin/src/api/production-task.ts
Create: web-admin/src/utils/production-task-set.ts
Create: web-admin/src/utils/production-task-set.test.ts
[ ] Step 1: Write the failing frontend test

// web-admin/src/utils/production-task-set.test.ts
import { describe, expect, it } from 'vitest'
import { mapWorkOrderTasksToSetItems } from './production-task-set'

describe('mapWorkOrderTasksToSetItems', () => {
  it('skips empty names and keeps route/description/remark', () => {
    const { items, skippedEmptyNameCount } = mapWorkOrderTasksToSetItems([
      { taskName: '分离', routeId: 'r1', routeName: '路线A', description: 'd', remark: 'm' },
      { taskName: '  ', routeId: 'r2' },
      { taskName: '培养' }
    ])
    expect(skippedEmptyNameCount).toBe(1)
    expect(items).toEqual([
      {
        taskName: '分离',
        routeId: 'r1',
        routeName: '路线A',
        description: 'd',
        remark: 'm'
      },
      {
        taskName: '培养',
        routeId: '',
        routeName: '',
        description: '',
        remark: ''
      }
    ])
  })
})

Step 2: Run test to verify it fails

cd web-admin && pnpm exec vitest run src/utils/production-task-set.test.ts

Expected: FAIL（模块不存在）

Step 3: Implement util + API modules

// web-admin/src/utils/production-task-set.ts
export type ProductionTaskSetItemPreview = {
  taskName: string
  routeId: string
  routeName: string
  description: string
  remark: string
}

export function mapWorkOrderTasksToSetItems(tasks: Record<string, any>[] | undefined): {
  items: ProductionTaskSetItemPreview[]
  skippedEmptyNameCount: number
} {
  const items: ProductionTaskSetItemPreview[] = []
  let skippedEmptyNameCount = 0
  for (const row of tasks || []) {
    const taskName = String(row.taskName || '').trim()
    if (!taskName) {
      skippedEmptyNameCount += 1
      continue
    }
    items.push({
      taskName,
      routeId: String(row.routeId || '').trim(),
      routeName: String(row.routeName || '').trim(),
      description: String(row.description || '').trim(),
      remark: String(row.remark || '').trim()
    })
  }
  return { items, skippedEmptyNameCount }
}

// web-admin/src/api/production-task-set.ts
import request from '@/utils/http'

type PageList<T> = {
  list: T[]
  total: number
  page: number
  pageSize: number
  totalPages: number
}

export type ProductionTaskSetItem = {
  id?: string
  taskName: string
  routeId?: string
  routeName?: string
  description?: string
  remark?: string
  orderNum?: number
}

export type ProductionTaskSet = {
  id: string
  setName: string
  status: string
  itemCount?: number
  items?: ProductionTaskSetItem[]
  remark?: string
  createTime?: string
  skippedEmptyNameCount?: number
}

export type ProductionTaskSetPayload = {
  setName: string
  status?: string
  remark?: string
  items: Array<{
    taskName: string
    routeId?: string
    description?: string
    remark?: string
  }>
}

export function fetchProductionTaskSetList(params?: {
  page?: number
  pageSize?: number
  keyword?: string
  status?: string
}) {
  return request.get<PageList<ProductionTaskSet>>({
    url: '/web/admin/production-task-sets',
    params
  })
}

export function fetchProductionTaskSetsActive() {
  return request.get<{ list: ProductionTaskSet[] }>({
    url: '/web/admin/production-task-sets/active'
  })
}

export function fetchProductionTaskSetDetail(id: string) {
  return request.get<ProductionTaskSet>({
    url: `/web/admin/production-task-sets/${id}`
  })
}

export function fetchCreateProductionTaskSet(data: ProductionTaskSetPayload) {
  return request.post<ProductionTaskSet>({
    url: '/web/admin/production-task-sets',
    params: data
  })
}

export function fetchUpdateProductionTaskSet(id: string, data: ProductionTaskSetPayload) {
  return request.put<ProductionTaskSet>({
    url: `/web/admin/production-task-sets/${id}`,
    params: data
  })
}

export function fetchDeleteProductionTaskSet(id: string) {
  return request.del({ url: `/web/admin/production-task-sets/${id}` })
}

export function fetchCreateProductionTaskSetFromWorkOrder(data: {
  workOrderId: string
  setName: string
  remark?: string
}) {
  return request.post<ProductionTaskSet>({
    url: '/web/admin/production-task-sets/from-work-order',
    params: data
  })
}

在 web-admin/src/api/production-task.ts 增加：

export function fetchCreateProductionTasksFromSet(data: { workOrderId: string; setId: string }) {
  return request.post<{ createdCount: number; list: any[] }>({
    url: '/web/admin/production-tasks/from-set',
    params: data
  })
}

Step 4: Run frontend test

cd web-admin && pnpm exec vitest run src/utils/production-task-set.test.ts

Expected: PASS

Step 5: Commit

git add web-admin/src/api/production-task-set.ts web-admin/src/api/production-task.ts \
  web-admin/src/utils/production-task-set.ts web-admin/src/utils/production-task-set.test.ts
git commit -m "$(cat <<'EOF'
feat: add production task set API client and item mapping

EOF
)"

Task 7: 维护页 + 静态路由

Files:

Create: web-admin/src/views/production/production-task-set/index.vue
Modify: web-admin/src/router/modules/sqxp.ts
[ ] Step 1: 静态路由

在 sqxp.ts 的 production-base children 中，ProductionProcessRoute 之后插入：

          {
            id: '95400000000000000000000000000001',
            name: 'ProductionTaskSet',
            path: 'production-task-set',
            component: '/production/production-task-set/index',
            meta: {
              title: '生产任务集合',
              keepAlive: true,
              menuCode: 'production:task-set:list'
            }
          }

Step 2: 列表页

对照 web-admin/src/views/quality-inspection/inspection-item-template/index.vue 的 P01 骨架（panel-top + title-bar-heading「生产任务集合列表」、filter、toolbar 仅「新增集合」、list-table + 分页）。

列：集合名称、任务条数 itemCount、状态 Tag（启用=success / 停用=info）、备注、创建时间 formatDisplayOrDash、操作 fixed="right" min-width ≥ 280：查看、编辑、启停、删除。权限：useMenuPerm 的 production:task-set:add|edit|remove。

新增/编辑弹窗：class="dialog-form-layout"，width="860px"，label-position="top"，align-center，modal-class="protocol-dialog-mask"，append-to-body。

表单字段：

集合名称（必填）
状态 el-select 启用/停用
备注
全宽条目：el-table 行内编辑。列：任务名称、工艺路线（el-select filterable remote，复用 fetchProcessRouteList，与 DispatchTaskDialog 相同 remote-method）、描述、备注、操作（上移/下移/删除）。表头工具「添加条目」。至少一行且任务名称非空才能提交。

查看弹窗：只读同样字段，无提交。

启停：fetchUpdateProductionTaskSet(id, { setName, status: next, remark, items }) 需带当前 items（先 fetchProductionTaskSetDetail）或 update 接口允许只改 status（Task 4 的 update_set 在 items is None 时不替换条目——启停只传 status）。

列表默认不展示 orderNum。

底栏按钮：Close 取消 / Check 确定，图标文字间距 6px。

Step 3: Commit

git add web-admin/src/views/production/production-task-set/index.vue web-admin/src/router/modules/sqxp.ts
git commit -m "$(cat <<'EOF'
feat: add production task set maintenance page

EOF
)"

Task 8: 派发弹窗支持任务集合

Files:

Modify: web-admin/src/views/production/production-work-order/DispatchTaskDialog.vue
[ ] Step 1: 阶段一增加派发方式

仅当 !props.editTaskId（新建）时展示。dispatchMode: 'single' | 'set'，默认 'single'。

有 production:task-set:list 或（因 SQL 已授权）直接展示两种方式；无 list 且无工单权限时只保留单条。实现：const canUseSet = computed(() => hasMenuPerm('production:task-set:list') || hasMenuPerm('production:work-order:list'))。canUseSet 为 false 时不渲染单选，行为与现状完全一致。

模板（插在现有表单网格之前）：

    <el-form-item v-if="!editTaskId && canUseSet" label="派发方式" class="form-col-span-2">
      <el-radio-group v-model="dispatchMode" @change="onDispatchModeChange">
        <el-radio-button value="single">单条任务</el-radio-button>
        <el-radio-button value="set">任务集合</el-radio-button>
      </el-radio-group>
    </el-form-item>

dispatchMode === 'single'：现有名称/路线/描述/备注不变；主按钮仍走 handleCreateTask → 进入 edit 阶段。

dispatchMode === 'set'：隐藏单条四字段；展示集合 el-select（fetchProductionTaskSetsActive）；@change 调 fetchProductionTaskSetDetail 填 setPreviewItems。只读表格列：任务名称、工艺路线（routeName 或「—」）、描述、备注。

主按钮文案仍为「生成任务」。handleCreateTask 开头：

    if (!props.editTaskId && dispatchMode.value === 'set') {
      await handleCreateFromSet()
      return
    }

  async function handleCreateFromSet() {
    const setId = String(selectedSetId.value || '').trim()
    const workOrderId = String(props.workOrderRow?.id || '').trim()
    if (!setId) {
      ElMessage.warning('请选择任务集合')
      return
    }
    if (!setPreviewItems.value.length) {
      ElMessage.warning('集合没有可派发的任务')
      return
    }
    submitting.value = true
    try {
      const result = await fetchCreateProductionTasksFromSet({ workOrderId, setId })
      ElMessage.success(`已生成 ${result?.createdCount || setPreviewItems.value.length} 条生产任务`)
      pendingListRefresh.value = true
      emit('success')
      handleClose()
    } catch {
      /* global error handling */
    } finally {
      submitting.value = false
    }
  }

禁止在集合成功后调用 enterEditPhase / needStepDispatch。

打开弹窗时（watch visible，非 editTaskId）：dispatchMode = 'single'，清空 selectedSetId 与预览。

Step 2: Commit

git add web-admin/src/views/production/production-work-order/DispatchTaskDialog.vue
git commit -m "$(cat <<'EOF'
feat: dispatch production tasks from a reusable set

EOF
)"

Task 9: 工单「存为任务集合」

Files:

Create: web-admin/src/views/production/production-work-order/SaveProductionTaskSetDialog.vue
Modify: web-admin/src/views/production/production-work-order/ProductionWorkOrderCardList.vue
Modify: web-admin/src/views/production/production-work-order/tabs/ProductionWorkOrderTabPanel.vue
[ ] Step 1: 存为弹窗组件

SaveProductionTaskSetDialog.vue：dialog-form-layout，宽约 720px。props：visible、workOrderId、tasks。字段：集合名称必填、备注可选、只读预览表（mapWorkOrderTasksToSetItems(tasks)）。无有效条目时确定按钮 disabled，提示「没有可保存的任务名称」。提交 fetchCreateProductionTaskSetFromWorkOrder。若 skippedEmptyNameCount > 0，成功消息带「已跳过 N 条空名称任务」。

Step 2: 卡片菜单

ProductionWorkOrderCardList.vue 的 … 下拉里，在编辑前增加：

                      <el-dropdown-item
                        v-if="canSaveTaskSet(row)"
                        command="save-set"
                      >
                        <el-icon><FolderAdd /></el-icon>存为任务集合
                      </el-dropdown-item>

  const { hasMenuPerm } = useMenuPerm()

  function canSaveTaskSet(row: ProductionWorkOrderItem) {
    if (isMineMode.value) return false
    if (!hasMenuPerm('production:task-set:add')) return false
    const tasks = resolvePresetTasks(row.id)
    return Array.isArray(tasks) && tasks.some((t) => String(t?.taskName || '').trim())
  }

在 ProductionWorkOrderTabPanel.vue 的 handleCommand 拦截（此处已有 presetTasksByWorkOrderId，不要把 save-set 再 emit 到 index.vue）：

    if (command === 'save-set') {
      saveSetRow.value = row
      saveSetVisible.value = true
      return
    }

tasks 用 presetTasksByWorkOrderId[row.id] || []。成功后不必整表重载工单，仅 ElMessage.success。

【我的生产任务】pageMode='mine'：canSaveTaskSet 为 false，无菜单项。

Step 3: Commit

git add web-admin/src/views/production/production-work-order/SaveProductionTaskSetDialog.vue \
  web-admin/src/views/production/production-work-order/ProductionWorkOrderCardList.vue \
  web-admin/src/views/production/production-work-order/tabs/ProductionWorkOrderTabPanel.vue
git commit -m "$(cat <<'EOF'
feat: save work-order tasks as a production task set

EOF
)"

Task 10: 图谱、自检与手工验收

Files: 按改动范围跑图谱

Step 1: 更新代码知识图谱

./scripts/update-graph.sh all

Step 2: 后端单测再跑一遍

cd backend && source venv/bin/activate && python -m pytest tests/test_production_task_set_helpers.py -v

Expected: PASS

Step 3: 前端单测

cd web-admin && pnpm exec vitest run src/utils/production-task-set.test.ts

Expected: PASS

Step 4: 手工对照 spec §9

执行 SQL 后登录：生产基础信息出现「生产任务集合」；可增改删启停；条目四字段；停用后派发下拉不可见。
生产工单「派发任务」单条：仍生成 1 条并进入编辑工艺。
选含 4 条的集合：一次生成 4 条，弹窗关闭，卡片任务区可见 4 条，不进三栏编辑。
集合内路线改为已删 ID（或先删路线）：生成失败，一条都不建。
「存为任务集合」保存名称/路线/描述/备注；空名称跳过。
【我的生产任务】无派发任务、无存为集合。

Step 5: Commit 图谱（若有 diff）

git add backend/graphify-out web-admin/graphify-out
git commit -m "$(cat <<'EOF'
chore: update graphify output after production task set

EOF
)"

Spec coverage

Spec 节	任务
§3 两表	Task 1–2
§4.1 CRUD / active	Task 4–5
§4.2 from-work-order	Task 4–5、9
§4.3 from-set 事务与路线失败	Task 4–5、8
§5.1 维护页 P01	Task 7
§5.2 派发弹窗二选一、生成后关闭	Task 8
§5.3 存为集合、mine 无入口	Task 9
§5.4 API 模块	Task 6
§6 边界	Task 3 单测 + Task 4 校验 + Task 10 手工
§7 菜单权限	Task 1、5、7–9
§8 非目标	不实现 Tab 绑定、步骤快照、mine 派发
§9 验收	Task 10

Type names（全局一致）

表/模型：BizProductionTaskSet / BizProductionTaskSetItem
函数：normalize_item_payload、skip_empty_named_tasks、collect_missing_route_ids、create_set_from_work_order、create_tasks_from_set
前端：fetchCreateProductionTasksFromSet、mapWorkOrderTasksToSetItems、dispatchMode: 'single' | 'set'

2026-08-18-production-task-set.md 56 KB Постоянная ссылка История Исходник

生产任务集合 Implementation Plan

File map

Task 1: SQL 建表 + 菜单授权

Task 2: SQLAlchemy 模型

Task 3: 纯函数单测（TDD）

Task 4: CRUD 实现

Task 5: Schema + API 路由

Task 6: 前端 API 与映射工具（TDD）

Task 7: 维护页 + 静态路由

Task 8: 派发弹窗支持任务集合

Task 9: 工单「存为任务集合」

Task 10: 图谱、自检与手工验收

Spec coverage

Type names（全局一致）

2026-08-18-production-task-set.md 56 KB

Постоянная ссылка История Исходник