Inicio Explorar Axuda
Rexistro Iniciar sesión
wuzilong
/
zkxw
1
0
Fork 0
Ficheiros Incidencias 0 Pull Requests 0 Wiki
Árbore: 56cee052d4
Ramas Etiquetas
zkxw
/
docs
/
superpowers
/
plans
/
2026-05-19-process-route-management.md

2026-05-19-process-route-management.md 84 KB
Histórico Raw

工艺路线管理 实施计划

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: 后端 6 张新表（路线主表 + 步骤 + 4 张步骤子表），CRUD 层遵循现有 process.py 模式（主子表级联、软删除、row_to_dict 序列化）。前端列表页遵循 unit/index.vue 模板，路线图配置弹窗使用三栏 flex 布局 + vue-draggable-plus 排序 + HTML5 Drag API 跨栏拖入。

Tech Stack: FastAPI + SQLAlchemy + Pydantic v2 (后端) | Vue 3 + Element Plus + vue-draggable-plus (前端) | MySQL 5.7

设计规格文档: docs/superpowers/specs/2026-05-19-process-route-management-design.md

文件清单

操作 文件
新建 docs/sql/production_process_route.sql
追加 backend/app/models/models.py
追加 backend/app/models/__init__.py
新建 backend/app/crud/process_route.py
新建 backend/app/schemas/process_route.py
追加 backend/app/api/production.py
追加 web-admin/src/api/production.ts
新建 web-admin/src/views/production/process-route/index.vue
追加 web-admin/src/router/modules/sqxp.ts

Task 1: SQL DDL — 6 张表 + 菜单数据

Files:

  • Create: docs/sql/production_process_route.sql

  • [ ] Step 1: 创建 SQL 文件

-- ============================================================================
-- 工艺路线管理 DDL（6张业务表 + 菜单 INSERT + 菜单 order_num 调整）
-- ============================================================================

-- ---------------------------------------------------------------------------
-- 1. 工艺路线主表 biz_process_route
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS `biz_process_route` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `route_code` varchar(64) NOT NULL COMMENT '路线编码',
  `route_name` varchar(100) NOT NULL COMMENT '路线名称',
  `product_type_id` varchar(64) DEFAULT NULL COMMENT '细胞产品类型ID',
  `version` varchar(50) DEFAULT NULL COMMENT '版本号(如V1.0)',
  `is_current` char(1) DEFAULT '0' COMMENT '是否当前版本(0-否 1-是)',
  `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 DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记：0-正常 1-已删除',
  PRIMARY KEY (`id`),
  KEY `idx_route_tenant` (`tenant_no`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='工艺路线';

-- ---------------------------------------------------------------------------
-- 2. 路线工序步骤 biz_process_route_step
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS `biz_process_route_step` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `route_id` varchar(64) NOT NULL COMMENT '所属路线ID',
  `process_id` varchar(64) DEFAULT NULL COMMENT '来源工艺库工艺ID',
  `step_order` int(11) DEFAULT NULL COMMENT '步骤顺序',
  `process_code` varchar(64) DEFAULT NULL COMMENT '工序代码(克隆)',
  `category_id` varchar(64) DEFAULT NULL COMMENT '工序分类ID(可编辑)',
  `process_name` varchar(100) DEFAULT NULL COMMENT '工序名称(可编辑)',
  `description` varchar(500) DEFAULT NULL COMMENT '描述(可编辑)',
  `equipment_name` varchar(200) DEFAULT NULL COMMENT '预计使用设备(可编辑)',
  `estimated_duration` varchar(100) DEFAULT NULL COMMENT '预估耗时(可编辑)',
  `sop_ids` varchar(1000) DEFAULT NULL COMMENT '关联SOP文档ID(逗号分隔)',
  `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 DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记：0-正常 1-已删除',
  PRIMARY KEY (`id`),
  KEY `idx_prs_route` (`route_id`, `del_flag`),
  KEY `idx_prs_tenant` (`tenant_no`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='路线工序步骤';

-- ---------------------------------------------------------------------------
-- 3. 步骤试剂 biz_process_route_step_reagent
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS `biz_process_route_step_reagent` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `step_id` varchar(64) NOT NULL COMMENT '所属步骤ID',
  `reagent_code` varchar(64) DEFAULT NULL COMMENT '试剂编码',
  `reagent_name` varchar(100) DEFAULT NULL COMMENT '试剂名称',
  `spec` varchar(200) DEFAULT NULL COMMENT '规格',
  `standard_usage` varchar(100) DEFAULT NULL COMMENT '标准用量',
  `unit` varchar(64) DEFAULT NULL COMMENT '单位',
  `is_key_material` char(1) DEFAULT '0' COMMENT '是否关键物料(0-否 1-是)',
  `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 DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记：0-正常 1-已删除',
  PRIMARY KEY (`id`),
  KEY `idx_prsr_step` (`step_id`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='步骤试剂';

-- ---------------------------------------------------------------------------
-- 4. 步骤耗材 biz_process_route_step_consumable
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS `biz_process_route_step_consumable` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `step_id` varchar(64) NOT NULL COMMENT '所属步骤ID',
  `consumable_code` varchar(64) DEFAULT NULL COMMENT '耗材编码',
  `consumable_name` varchar(100) DEFAULT NULL COMMENT '耗材名称',
  `spec_model` varchar(200) DEFAULT NULL COMMENT '规格型号',
  `standard_usage` varchar(100) DEFAULT NULL COMMENT '标准用量',
  `unit` varchar(64) DEFAULT NULL COMMENT '单位',
  `is_required` char(1) DEFAULT '1' COMMENT '是否必需(1-必须 0-可选)',
  `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 DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记：0-正常 1-已删除',
  PRIMARY KEY (`id`),
  KEY `idx_prsc_step` (`step_id`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='步骤耗材';

-- ---------------------------------------------------------------------------
-- 5. 步骤产出物 biz_process_route_step_output
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS `biz_process_route_step_output` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `step_id` varchar(64) NOT NULL COMMENT '所属步骤ID',
  `output_type_id` varchar(64) DEFAULT NULL COMMENT '产出类型ID',
  `output_name` varchar(200) DEFAULT NULL COMMENT '产出物名称',
  `output_barcode` varchar(200) DEFAULT NULL COMMENT '产出物条码',
  `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 DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记：0-正常 1-已删除',
  PRIMARY KEY (`id`),
  KEY `idx_prso_step` (`step_id`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='步骤产出物';

-- ---------------------------------------------------------------------------
-- 6. 步骤数据采集 biz_process_route_step_data_collection
-- ---------------------------------------------------------------------------
CREATE TABLE IF NOT EXISTS `biz_process_route_step_data_collection` (
  `id` varchar(64) NOT NULL COMMENT '主键ID',
  `step_id` varchar(64) NOT NULL COMMENT '所属步骤ID',
  `collection_name` varchar(100) DEFAULT NULL COMMENT '采集信息名称',
  `unit` varchar(64) DEFAULT NULL COMMENT '单位',
  `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 DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
  `update_by` varchar(64) DEFAULT NULL COMMENT '更新人id',
  `update_time` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
  `del_flag` tinyint(4) NOT NULL DEFAULT '0' COMMENT '删除标记：0-正常 1-已删除',
  PRIMARY KEY (`id`),
  KEY `idx_prsdc_step` (`step_id`, `del_flag`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci COMMENT='步骤数据采集';

-- ============================================================================
-- 菜单数据
-- ============================================================================
INSERT INTO `sys_menu` (`id`, `menu_name`, `parent_id`, `menu_type`, `path`, `component`, `perms`, `is_frame`, `visible`, `status`, `tenant_no`, `order_num`, `remark`, `create_by`, `create_dept_id`, `create_time`, `update_by`, `update_time`, `del_flag`)
VALUES ('90000000000000000000000000000010', '工艺路线管理', '90000000000000000000000000000001', 'C', 'process-route', '/production/process-route/index', 'production:process-route:list', 1, '1', '1', 'zkxw', 7, NULL, NULL, '', '2026-05-19 00:00:00', NULL, '2026-05-19 00:00:00', 0);

-- SOP文档 order_num 从 7 改为 8
UPDATE `sys_menu` SET `order_num` = 8 WHERE `id` = '90000000000000000000000000000004';
  • Step 2: 在 MySQL 中执行 SQL 验证建表成功

Run: 在 MySQL 客户端执行 docs/sql/production_process_route.sql Expected: 6 张表创建成功，菜单插入成功，SOP order_num 更新成功

  • Step 3: 提交
git add docs/sql/production_process_route.sql
git commit -m "feat: add process route management SQL DDL (6 tables + menu)"

Task 2: SQLAlchemy Model — 6 个新模型

Files:

  • Modify: backend/app/models/models.py (追加到文件末尾 from app.models.qc_param 之前)
  • Modify: backend/app/models/__init__.py (追加 6 个 import + 6 个 all 条目)

参考: 现有 BizProcess (line 1854)、BizProcessReagent (line 1878)、BizProcessConsumable (line 1902)、BizProcessOutput (line 1926) 的写法。

  • Step 1: 在 models.py 末尾追加 6 个 Model

在 from app.models.qc_param import ... 这行之前，追加以下内容：

class BizProcessRoute(Base):
    """工艺路线 biz_process_route"""
    __tablename__ = "biz_process_route"

    id = Column(String(64), primary_key=True, comment="主键ID")
    route_code = Column(String(64), nullable=False, comment="路线编码")
    route_name = Column(String(100), nullable=False, comment="路线名称")
    product_type_id = Column(String(64), nullable=True, comment="细胞产品类型ID")
    version = Column(String(50), nullable=True, comment="版本号")
    is_current = Column(CHAR(1), nullable=True, default="0", comment="是否当前版本(0-否 1-是)")
    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 BizProcessRouteStep(Base):
    """路线工序步骤 biz_process_route_step"""
    __tablename__ = "biz_process_route_step"

    id = Column(String(64), primary_key=True, comment="主键ID")
    route_id = Column(String(64), nullable=False, comment="所属路线ID")
    process_id = Column(String(64), nullable=True, comment="来源工艺库工艺ID")
    step_order = Column(Integer, nullable=True, comment="步骤顺序")
    process_code = Column(String(64), nullable=True, comment="工序代码(克隆)")
    category_id = Column(String(64), nullable=True, comment="工序分类ID")
    process_name = Column(String(100), nullable=True, comment="工序名称")
    description = Column(String(500), nullable=True, comment="描述")
    equipment_name = Column(String(200), nullable=True, comment="预计使用设备")
    estimated_duration = Column(String(100), nullable=True, comment="预估耗时")
    sop_ids = Column(String(1000), nullable=True, comment="关联SOP文档ID(逗号分隔)")
    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 BizProcessRouteStepReagent(Base):
    """步骤试剂 biz_process_route_step_reagent"""
    __tablename__ = "biz_process_route_step_reagent"

    id = Column(String(64), primary_key=True, comment="主键ID")
    step_id = Column(String(64), nullable=False, comment="所属步骤ID")
    reagent_code = Column(String(64), nullable=True, comment="试剂编码")
    reagent_name = Column(String(100), nullable=True, comment="试剂名称")
    spec = Column(String(200), nullable=True, comment="规格")
    standard_usage = Column(String(100), nullable=True, comment="标准用量")
    unit = Column(String(64), nullable=True, comment="单位")
    is_key_material = Column(CHAR(1), nullable=True, default="0", comment="是否关键物料(0-否 1-是)")
    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 BizProcessRouteStepConsumable(Base):
    """步骤耗材 biz_process_route_step_consumable"""
    __tablename__ = "biz_process_route_step_consumable"

    id = Column(String(64), primary_key=True, comment="主键ID")
    step_id = Column(String(64), nullable=False, comment="所属步骤ID")
    consumable_code = Column(String(64), nullable=True, comment="耗材编码")
    consumable_name = Column(String(100), nullable=True, comment="耗材名称")
    spec_model = Column(String(200), nullable=True, comment="规格型号")
    standard_usage = Column(String(100), nullable=True, comment="标准用量")
    unit = Column(String(64), nullable=True, comment="单位")
    is_required = Column(CHAR(1), nullable=True, default="1", comment="是否必需(1-必须 0-可选)")
    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="删除标记")


class BizProcessRouteStepOutput(Base):
    """步骤产出物 biz_process_route_step_output"""
    __tablename__ = "biz_process_route_step_output"

    id = Column(String(64), primary_key=True, comment="主键ID")
    step_id = Column(String(64), nullable=False, comment="所属步骤ID")
    output_type_id = Column(String(64), nullable=True, comment="产出类型ID")
    output_name = Column(String(200), nullable=True, comment="产出物名称")
    output_barcode = Column(String(200), nullable=True, comment="产出物条码")
    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 BizProcessRouteStepDataCollection(Base):
    """步骤数据采集 biz_process_route_step_data_collection"""
    __tablename__ = "biz_process_route_step_data_collection"

    id = Column(String(64), primary_key=True, comment="主键ID")
    step_id = Column(String(64), nullable=False, comment="所属步骤ID")
    collection_name = Column(String(100), nullable=True, comment="采集信息名称")
    unit = Column(String(64), nullable=True, comment="单位")
    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="删除标记")
  • Step 2: 在 __init__.py 追加 import 和 all

在 import 块追加（在 BizProcessDataCollection, 之后）：

    BizProcessRoute,
    BizProcessRouteStep,
    BizProcessRouteStepReagent,
    BizProcessRouteStepConsumable,
    BizProcessRouteStepOutput,
    BizProcessRouteStepDataCollection,

在 __all__ 列表追加（在 "BizProcessDataCollection", 之后）：

    "BizProcessRoute",
    "BizProcessRouteStep",
    "BizProcessRouteStepReagent",
    "BizProcessRouteStepConsumable",
    "BizProcessRouteStepOutput",
    "BizProcessRouteStepDataCollection",
  • Step 3: 验证后端能正常启动

Run: cd backend && source venv/bin/activate && python -c "from app.models import BizProcessRoute, BizProcessRouteStep; print('Models OK')" Expected: Models OK

  • Step 4: 提交
git add backend/app/models/models.py backend/app/models/__init__.py
git commit -m "feat: add process route SQLAlchemy models (6 classes)"

Task 3: Pydantic Schemas

Files:

  • Create: backend/app/schemas/process_route.py

参考: 后端现有模式是在 API 文件中 inline 定义 Pydantic schema。但本模块 schema 数量多（~15个），独立文件更清晰。

  • Step 1: 创建 schema 文件
"""工艺路线管理 Pydantic Schemas"""
from pydantic import BaseModel, ConfigDict, Field
from typing import Optional, List


# ---------------------------------------------------------------------------
# 路线主表
# ---------------------------------------------------------------------------

class ProcessRouteCreateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    route_code: str = Field(..., alias="routeCode")
    route_name: str = Field(..., alias="routeName")
    product_type_id: Optional[str] = Field(None, alias="productTypeId")
    version: Optional[str] = Field(None, alias="version")
    remark: Optional[str] = None


class ProcessRouteUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    route_code: Optional[str] = Field(None, alias="routeCode")
    route_name: Optional[str] = Field(None, alias="routeName")
    product_type_id: Optional[str] = Field(None, alias="productTypeId")
    version: Optional[str] = Field(None, alias="version")
    remark: Optional[str] = None


class SetCurrentBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    route_id: str = Field(..., alias="routeId")


# ---------------------------------------------------------------------------
# 步骤基础信息
# ---------------------------------------------------------------------------

class StepUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    process_code: Optional[str] = Field(None, alias="processCode")
    category_id: Optional[str] = Field(None, alias="categoryId")
    process_name: Optional[str] = Field(None, alias="processName")
    description: Optional[str] = None
    equipment_name: Optional[str] = Field(None, alias="equipmentName")
    estimated_duration: Optional[str] = Field(None, alias="estimatedDuration")
    sop_ids: Optional[str] = Field(None, alias="sopIds")


class StepAddBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    route_id: str = Field(..., alias="routeId")
    process_id: str = Field(..., alias="processId")
    step_order: Optional[int] = Field(None, alias="stepOrder")


class StepReorderBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    route_id: str = Field(..., alias="routeId")
    step_ids: List[str] = Field(..., alias="stepIds")


# ---------------------------------------------------------------------------
# 步骤子表 items
# ---------------------------------------------------------------------------

class StepReagentItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    reagent_code: Optional[str] = Field(None, alias="reagentCode")
    reagent_name: Optional[str] = Field(None, alias="reagentName")
    spec: Optional[str] = None
    standard_usage: Optional[str] = Field(None, alias="standardUsage")
    unit: Optional[str] = None
    is_key_material: Optional[str] = Field("0", alias="isKeyMaterial")
    remark: Optional[str] = None


class StepConsumableItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    consumable_code: Optional[str] = Field(None, alias="consumableCode")
    consumable_name: Optional[str] = Field(None, alias="consumableName")
    spec_model: Optional[str] = Field(None, alias="specModel")
    standard_usage: Optional[str] = Field(None, alias="standardUsage")
    unit: Optional[str] = None
    is_required: Optional[str] = Field("1", alias="isRequired")
    remark: Optional[str] = None


class StepOutputItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    output_type_id: Optional[str] = Field(None, alias="outputTypeId")
    output_name: Optional[str] = Field(None, alias="outputName")
    output_barcode: Optional[str] = Field(None, alias="outputBarcode")
    remark: Optional[str] = None


class StepDataCollectionItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    collection_name: Optional[str] = Field(None, alias="collectionName")
    unit: Optional[str] = None
    order_num: Optional[int] = Field(0, alias="orderNum")
    remark: Optional[str] = None


# ---------------------------------------------------------------------------
# 子表更新 body（step_id + items[]）
# ---------------------------------------------------------------------------

class StepSubTableUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    items: List = Field(...)


class StepReagentUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    items: List[StepReagentItem] = Field(...)


class StepConsumableUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    items: List[StepConsumableItem] = Field(...)


class StepOutputUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    items: List[StepOutputItem] = Field(...)


class StepDataCollectionUpdateBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    items: List[StepDataCollectionItem] = Field(...)
  • Step 2: 验证 import 正常

Run: cd backend && source venv/bin/activate && python -c "from app.schemas.process_route import ProcessRouteCreateBody; print('Schemas OK')" Expected: Schemas OK

  • Step 3: 提交
git add backend/app/schemas/process_route.py
git commit -m "feat: add process route Pydantic schemas"

Task 4: CRUD 层 — process_route.py

Files:

  • Create: backend/app/crud/process_route.py

参考: backend/app/crud/process.py 的完整模式：_row_to_dict 序列化、_create_sub_* 辅助函数、_soft_delete_sub_records 级联软删除、完整替换子表策略。

本文件包含：

  1. 6 个 _row_to_dict 序列化函数
  2. 路线主表 CRUD：list_routes、get_route、get_route_detail、create_route、update_route、soft_delete_route、set_current_route
  3. 步骤 CRUD：add_step（克隆工艺+子表）、update_step、reorder_steps、soft_delete_step
  4. 子表更新：update_step_reagents、update_step_consumables、update_step_outputs、update_step_data_collections
  • Step 1: 创建 CRUD 文件

文件内容遵循 process.py 模式，关键函数签名：

"""工艺路线 CRUD（biz_process_route + biz_process_route_step + 4张步骤子表）"""
from __future__ import annotations

import math
from datetime import datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import and_, func, or_
from sqlalchemy.orm import Session

from app.core.utils import generate_uuid
from app.models.models import (
    BizProcess,
    BizProcessReagent,
    BizProcessConsumable,
    BizProcessOutput,
    BizProcessDataCollection,
    BizProcessRoute,
    BizProcessRouteStep,
    BizProcessRouteStepReagent,
    BizProcessRouteStepConsumable,
    BizProcessRouteStepOutput,
    BizProcessRouteStepDataCollection,
)


# ---------------------------------------------------------------------------
# 序列化函数
# ---------------------------------------------------------------------------

def _route_row_to_dict(r: BizProcessRoute) -> Dict[str, Any]:
    return {
        "id": r.id,
        "routeCode": r.route_code,
        "routeName": r.route_name,
        "productTypeId": r.product_type_id or "",
        "version": r.version or "",
        "isCurrent": r.is_current if r.is_current is not None else "0",
        "tenantNo": r.tenant_no,
        "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 "",
    }


def _step_row_to_dict(s: BizProcessRouteStep) -> Dict[str, Any]:
    return {
        "id": s.id,
        "routeId": s.route_id,
        "processId": s.process_id or "",
        "stepOrder": s.step_order if s.step_order is not None else 0,
        "processCode": s.process_code or "",
        "categoryId": s.category_id or "",
        "processName": s.process_name or "",
        "description": s.description or "",
        "equipmentName": s.equipment_name or "",
        "estimatedDuration": s.estimated_duration or "",
        "sopIds": s.sop_ids or "",
        "remark": s.remark or "",
    }


def _step_reagent_row_to_dict(r: BizProcessRouteStepReagent) -> Dict[str, Any]:
    return {
        "id": r.id,
        "stepId": r.step_id,
        "reagentCode": r.reagent_code or "",
        "reagentName": r.reagent_name or "",
        "spec": r.spec or "",
        "standardUsage": r.standard_usage or "",
        "unit": r.unit or "",
        "isKeyMaterial": r.is_key_material if r.is_key_material is not None else "0",
        "remark": r.remark or "",
    }


def _step_consumable_row_to_dict(c: BizProcessRouteStepConsumable) -> Dict[str, Any]:
    return {
        "id": c.id,
        "stepId": c.step_id,
        "consumableCode": c.consumable_code or "",
        "consumableName": c.consumable_name or "",
        "specModel": c.spec_model or "",
        "standardUsage": c.standard_usage or "",
        "unit": c.unit or "",
        "isRequired": c.is_required if c.is_required is not None else "1",
        "remark": c.remark or "",
    }


def _step_output_row_to_dict(o: BizProcessRouteStepOutput) -> Dict[str, Any]:
    return {
        "id": o.id,
        "stepId": o.step_id,
        "outputTypeId": o.output_type_id or "",
        "outputName": o.output_name or "",
        "outputBarcode": o.output_barcode or "",
        "remark": o.remark or "",
    }


def _step_data_collection_row_to_dict(d: BizProcessRouteStepDataCollection) -> Dict[str, Any]:
    return {
        "id": d.id,
        "stepId": d.step_id,
        "collectionName": d.collection_name or "",
        "unit": d.unit or "",
        "orderNum": d.order_num if d.order_num is not None else 0,
        "remark": d.remark or "",
    }


# ---------------------------------------------------------------------------
# 路线主表 CRUD
# ---------------------------------------------------------------------------

def list_routes(db: Session, *, page: int = 1, page_size: int = 10,
                route_name: Optional[str] = None, route_code: Optional[str] = None,
                product_type_id: Optional[str] = None,
                tenant_no: Optional[str] = None) -> Tuple[List[Dict], int]:
    """路线分页列表，支持 route_name 模糊、route_code 精确、product_type_id 筛选"""
    def _filtered():
        q = db.query(BizProcessRoute).filter(BizProcessRoute.del_flag == 0)
        if tenant_no:
            q = q.filter(BizProcessRoute.tenant_no == tenant_no)
        if route_name and route_name.strip():
            q = q.filter(BizProcessRoute.route_name.like(f"%{route_name.strip()}%"))
        if route_code and route_code.strip():
            q = q.filter(BizProcessRoute.route_code == route_code.strip())
        if product_type_id and product_type_id.strip():
            q = q.filter(BizProcessRoute.product_type_id == product_type_id.strip())
        return q

    total = _filtered().order_by(None).with_entities(func.count(BizProcessRoute.id)).scalar() or 0
    page = max(1, page)
    page_size = min(max(1, page_size), 100)
    rows = (
        _filtered()
        .order_by(BizProcessRoute.order_num.asc().nullslast(), BizProcessRoute.create_time.desc())
        .offset((page - 1) * page_size)
        .limit(page_size)
        .all()
    )
    return [_route_row_to_dict(x) for x in rows], int(total)


def get_route(db: Session, route_id: str) -> Optional[BizProcessRoute]:
    return db.query(BizProcessRoute).filter(
        and_(BizProcessRoute.id == route_id, BizProcessRoute.del_flag == 0)
    ).first()


def get_route_detail(db: Session, route_id: str) -> Optional[Dict[str, Any]]:
    """路线详情 = 主表 + 步骤列表(含每个步骤的4张子表)"""
    route = get_route(db, route_id)
    if not route:
        return None
    result = _route_row_to_dict(route)

    steps = (
        db.query(BizProcessRouteStep)
        .filter(and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0))
        .order_by(BizProcessRouteStep.step_order.asc(), BizProcessRouteStep.create_time.asc())
        .all()
    )
    step_list = []
    for s in steps:
        sd = _step_row_to_dict(s)
        # 加载4张子表
        reagents = db.query(BizProcessRouteStepReagent).filter(
            and_(BizProcessRouteStepReagent.step_id == s.id, BizProcessRouteStepReagent.del_flag == 0)
        ).order_by(BizProcessRouteStepReagent.create_time.asc()).all()
        consumables = db.query(BizProcessRouteStepConsumable).filter(
            and_(BizProcessRouteStepConsumable.step_id == s.id, BizProcessRouteStepConsumable.del_flag == 0)
        ).order_by(BizProcessRouteStepConsumable.create_time.asc()).all()
        outputs = db.query(BizProcessRouteStepOutput).filter(
            and_(BizProcessRouteStepOutput.step_id == s.id, BizProcessRouteStepOutput.del_flag == 0)
        ).order_by(BizProcessRouteStepOutput.create_time.asc()).all()
        data_collections = db.query(BizProcessRouteStepDataCollection).filter(
            and_(BizProcessRouteStepDataCollection.step_id == s.id, BizProcessRouteStepDataCollection.del_flag == 0)
        ).order_by(BizProcessRouteStepDataCollection.order_num.asc()).all()

        sd["reagents"] = [_step_reagent_row_to_dict(r) for r in reagents]
        sd["consumables"] = [_step_consumable_row_to_dict(c) for c in consumables]
        sd["outputs"] = [_step_output_row_to_dict(o) for o in outputs]
        sd["dataCollections"] = [_step_data_collection_row_to_dict(d) for d in data_collections]
        step_list.append(sd)

    result["steps"] = step_list
    return result


def create_route(db: Session, *, route_code: str, route_name: str,
                 product_type_id: Optional[str] = None, version: Optional[str] = None,
                 remark: Optional[str] = None, tenant_no: str = "zkxw",
                 user_id: Optional[str] = None) -> BizProcessRoute:
    rc = (route_code or "").strip()
    if not rc:
        raise ValueError("路线编码不能为空")
    now = datetime.now()
    obj = BizProcessRoute(
        id=generate_uuid(),
        route_code=rc,
        route_name=(route_name or "").strip() or rc,
        product_type_id=(product_type_id or "").strip() or None,
        version=(version or "").strip() or None,
        is_current="0",
        tenant_no=tenant_no,
        order_num=0,
        remark=(remark or "").strip() or None,
        create_by=user_id,
        create_dept_id="",
        create_time=now,
        update_by=user_id,
        update_time=now,
        del_flag=0,
    )
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def update_route(db: Session, route_id: str, *, user_id: Optional[str] = None, **kwargs) -> BizProcessRoute:
    obj = get_route(db, route_id)
    if not obj:
        raise ValueError("路线不存在")
    for field in ["route_code", "route_name", "product_type_id", "version", "remark"]:
        val = kwargs.get(field)
        if val is not None:
            setattr(obj, field, (val or "").strip() or None if field != "route_code" else (val or "").strip())
    obj.update_by = user_id
    obj.update_time = datetime.now()
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


def set_current_route(db: Session, route_id: str, tenant_no: str) -> BizProcessRoute:
    """设置指定路线为当前版本：同 route_code 下先全部置0，再置目标为1"""
    obj = get_route(db, route_id)
    if not obj:
        raise ValueError("路线不存在")
    now = datetime.now()
    # 同编码全部置0
    siblings = db.query(BizProcessRoute).filter(
        and_(BizProcessRoute.route_code == obj.route_code,
             BizProcessRoute.tenant_no == tenant_no,
             BizProcessRoute.del_flag == 0)
    ).all()
    for s in siblings:
        s.is_current = "0"
        s.update_time = now
        db.add(s)
    # 目标置1
    obj.is_current = "1"
    obj.update_time = now
    db.add(obj)
    db.commit()
    db.refresh(obj)
    return obj


# ---------------------------------------------------------------------------
# 步骤子表辅助
# ---------------------------------------------------------------------------

def _soft_delete_sub_records(db: Session, model_class, step_id: str, now: datetime) -> None:
    rows = db.query(model_class).filter(
        and_(model_class.step_id == step_id, model_class.del_flag == 0)
    ).all()
    for r in rows:
        r.del_flag = 1
        r.update_time = now
        db.add(r)


def _create_step_reagent(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepReagent(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        reagent_code=(data.get("reagent_code") or "").strip() or None,
        reagent_name=(data.get("reagent_name") or "").strip() or None,
        spec=(data.get("spec") or "").strip() or None,
        standard_usage=(data.get("standard_usage") or "").strip() or None,
        unit=(data.get("unit") or "").strip() or None,
        is_key_material=data.get("is_key_material", "0"),
        remark=(data.get("remark") or "").strip() or None,
        order_num=0, create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


def _create_step_consumable(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepConsumable(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        consumable_code=(data.get("consumable_code") or "").strip() or None,
        consumable_name=(data.get("consumable_name") or "").strip() or None,
        spec_model=(data.get("spec_model") or "").strip() or None,
        standard_usage=(data.get("standard_usage") or "").strip() or None,
        unit=(data.get("unit") or "").strip() or None,
        is_required=data.get("is_required", "1"),
        remark=(data.get("remark") or "").strip() or None,
        order_num=0, create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


def _create_step_output(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepOutput(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        output_type_id=(data.get("output_type_id") or "").strip() or None,
        output_name=(data.get("output_name") or "").strip() or None,
        output_barcode=(data.get("output_barcode") or "").strip() or None,
        remark=(data.get("remark") or "").strip() or None,
        order_num=0, create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


def _create_step_data_collection(db, step_id, tenant_no, user_id, now, data):
    db.add(BizProcessRouteStepDataCollection(
        id=generate_uuid(), step_id=step_id, tenant_no=tenant_no,
        collection_name=(data.get("collection_name") or "").strip() or None,
        unit=(data.get("unit") or "").strip() or None,
        order_num=data.get("order_num", 0) or 0,
        remark=(data.get("remark") or "").strip() or None,
        create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    ))


# ---------------------------------------------------------------------------
# 步骤 CRUD
# ---------------------------------------------------------------------------

def add_step(db: Session, *, route_id: str, process_id: str, step_order: Optional[int] = None,
             tenant_no: str = "zkxw", user_id: Optional[str] = None) -> Dict[str, Any]:
    """从工艺库克隆创建步骤"""
    route = get_route(db, route_id)
    if not route:
        raise ValueError("路线不存在")

    # 读取源工艺
    proc = db.query(BizProcess).filter(
        and_(BizProcess.id == process_id, BizProcess.del_flag == 0)
    ).first()
    if not proc:
        raise ValueError("工艺不存在")

    now = datetime.now()
    # 计算 step_order
    if step_order is None:
        max_order = db.query(func.max(BizProcessRouteStep.step_order)).filter(
            and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0)
        ).scalar()
        step_order = (max_order or 0) + 1

    step = BizProcessRouteStep(
        id=generate_uuid(), route_id=route_id, process_id=process_id,
        step_order=step_order,
        process_code=proc.process_code, category_id=proc.category_id,
        process_name=proc.process_name, description=proc.description,
        equipment_name=proc.equipment_name, estimated_duration=proc.estimated_duration,
        sop_ids=proc.sop_ids,
        tenant_no=tenant_no, order_num=0,
        remark=None, create_by=user_id, create_dept_id="",
        create_time=now, update_by=user_id, update_time=now, del_flag=0,
    )
    db.add(step)
    db.flush()

    # 克隆4张子表
    src_reagents = db.query(BizProcessReagent).filter(
        and_(BizProcessReagent.process_id == process_id, BizProcessReagent.del_flag == 0)
    ).all()
    for r in src_reagents:
        _create_step_reagent(db, step.id, tenant_no, user_id, now, {
            "reagent_code": r.reagent_code, "reagent_name": r.reagent_name,
            "spec": r.spec, "standard_usage": r.standard_usage,
            "unit": r.unit, "is_key_material": r.is_key_material,
        })

    src_consumables = db.query(BizProcessConsumable).filter(
        and_(BizProcessConsumable.process_id == process_id, BizProcessConsumable.del_flag == 0)
    ).all()
    for c in src_consumables:
        _create_step_consumable(db, step.id, tenant_no, user_id, now, {
            "consumable_code": c.consumable_code, "consumable_name": c.consumable_name,
            "spec_model": c.spec_model, "standard_usage": c.standard_usage,
            "unit": c.unit, "is_required": c.is_required, "remark": c.remark,
        })

    src_outputs = db.query(BizProcessOutput).filter(
        and_(BizProcessOutput.process_id == process_id, BizProcessOutput.del_flag == 0)
    ).all()
    for o in src_outputs:
        _create_step_output(db, step.id, tenant_no, user_id, now, {
            "output_type_id": o.output_type_id, "output_name": o.output_name,
            "output_barcode": o.output_barcode,
        })

    src_collections = db.query(BizProcessDataCollection).filter(
        and_(BizProcessDataCollection.process_id == process_id, BizProcessDataCollection.del_flag == 0)
    ).all()
    for d in src_collections:
        _create_step_data_collection(db, step.id, tenant_no, user_id, now, {
            "collection_name": d.collection_name, "unit": d.unit, "order_num": d.order_num,
        })

    db.commit()

    # 返回完整 step（含子表）
    return get_route_detail(db, route_id)["steps"][-1] if get_route_detail(db, route_id)["steps"] else _step_row_to_dict(step)


def update_step(db: Session, step_id: str, *, user_id: Optional[str] = None, **kwargs) -> BizProcessRouteStep:
    step = db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
    ).first()
    if not step:
        raise ValueError("步骤不存在")
    for field in ["process_code", "category_id", "process_name", "description",
                   "equipment_name", "estimated_duration", "sop_ids"]:
        val = kwargs.get(field)
        if val is not None:
            setattr(step, field, (val or "").strip() or None)
    step.update_by = user_id
    step.update_time = datetime.now()
    db.add(step)
    db.commit()
    db.refresh(step)
    return step


def reorder_steps(db: Session, route_id: str, step_ids: List[str]) -> None:
    """按 step_ids 数组顺序更新 step_order"""
    now = datetime.now()
    for idx, sid in enumerate(step_ids):
        step = db.query(BizProcessRouteStep).filter(
            and_(BizProcessRouteStep.id == sid, BizProcessRouteStep.route_id == route_id,
                 BizProcessRouteStep.del_flag == 0)
        ).first()
        if step:
            step.step_order = idx + 1
            step.update_time = now
            db.add(step)
    db.commit()


def soft_delete_step(db: Session, step_id: str) -> None:
    step = db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.id == step_id, BizProcessRouteStep.del_flag == 0)
    ).first()
    if not step:
        raise ValueError("步骤不存在")
    now = datetime.now()
    step.del_flag = 1
    step.update_time = now
    db.add(step)
    _soft_delete_sub_records(db, BizProcessRouteStepReagent, step_id, now)
    _soft_delete_sub_records(db, BizProcessRouteStepConsumable, step_id, now)
    _soft_delete_sub_records(db, BizProcessRouteStepOutput, step_id, now)
    _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, step_id, now)
    db.commit()


def soft_delete_route(db: Session, route_id: str) -> None:
    """级联软删除路线 + 所有步骤 + 所有步骤子表"""
    route = get_route(db, route_id)
    if not route:
        raise ValueError("路线不存在")
    now = datetime.now()
    route.del_flag = 1
    route.update_time = now
    db.add(route)

    steps = db.query(BizProcessRouteStep).filter(
        and_(BizProcessRouteStep.route_id == route_id, BizProcessRouteStep.del_flag == 0)
    ).all()
    for s in steps:
        s.del_flag = 1
        s.update_time = now
        db.add(s)
        _soft_delete_sub_records(db, BizProcessRouteStepReagent, s.id, now)
        _soft_delete_sub_records(db, BizProcessRouteStepConsumable, s.id, now)
        _soft_delete_sub_records(db, BizProcessRouteStepOutput, s.id, now)
        _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, s.id, now)
    db.commit()


# ---------------------------------------------------------------------------
# 子表完整替换更新
# ---------------------------------------------------------------------------

def update_step_reagents(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepReagent, step_id, now)
    for it in items:
        _create_step_reagent(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def update_step_consumables(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepConsumable, step_id, now)
    for it in items:
        _create_step_consumable(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def update_step_outputs(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepOutput, step_id, now)
    for it in items:
        _create_step_output(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def update_step_data_collections(db: Session, step_id: str, items: List[Dict], tenant_no: str, user_id: Optional[str] = None):
    now = datetime.now()
    _soft_delete_sub_records(db, BizProcessRouteStepDataCollection, step_id, now)
    for it in items:
        _create_step_data_collection(db, step_id, tenant_no, user_id, now, it)
    db.commit()


def total_pages(total: int, page_size: int) -> int:
    if total <= 0:
        return 0
    return int(math.ceil(total / float(page_size)))
  • Step 2: 验证 import 正常

Run: cd backend && source venv/bin/activate && python -c "from app.crud.process_route import list_routes, add_step; print('CRUD OK')" Expected: CRUD OK

  • Step 3: 提交
git add backend/app/crud/process_route.py
git commit -m "feat: add process route CRUD layer with clone and cascade delete"

Task 5: API 路由 — 追加到 production.py

Files:

  • Modify: backend/app/api/production.py (文件末尾追加 ~200 行)

参考: 现有 production.py 中的工艺库 API（line 735-943）模式：inline Pydantic 已改为从 schemas/process_route.py 导入，路由前缀为 /web/admin。

注意: 本模块的路由路径不按设计规格文档中的 /api/admin/production/route/，而是沿用现有模式 /web/admin/process-routes/*，保持与项目其他路由风格一致。

  • Step 1: 在 production.py 文件头部追加 import

在 from app.crud import process as process_svc 之后追加：

from app.crud import process_route as route_svc
from app.schemas.process_route import (
    ProcessRouteCreateBody,
    ProcessRouteUpdateBody,
    SetCurrentBody,
    StepAddBody,
    StepUpdateBody,
    StepReorderBody,
    StepReagentUpdateBody,
    StepConsumableUpdateBody,
    StepOutputUpdateBody,
    StepDataCollectionUpdateBody,
)
  • Step 2: 在文件末尾追加 API 路由

在 delete_process 函数之后追加以下全部路由：


# ---------------------------------------------------------------------------
# 8. 工艺路线管理
# ---------------------------------------------------------------------------


@router.get("/process-routes")
async def list_process_routes(
    page: int = Query(1, ge=1),
    pageSize: int = Query(10, ge=1, le=100, alias="pageSize"),
    routeName: Optional[str] = None,
    routeCode: Optional[str] = None,
    productTypeId: Optional[str] = None,
    user: dict = Depends(require_menu_code("production:process-route:list")),
    db: Session = Depends(get_db),
):
    tenant_no = _get_tenant_no(db, user)
    items, total = route_svc.list_routes(
        db, page=page, page_size=pageSize,
        route_name=routeName, route_code=routeCode,
        product_type_id=productTypeId, tenant_no=tenant_no,
    )
    tp = route_svc.total_pages(total, pageSize)
    return success_response({"list": items, "total": total, "page": page, "pageSize": pageSize, "totalPages": tp})


@router.get("/process-routes/{route_id}")
async def get_process_route_detail(
    route_id: str,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    detail = route_svc.get_route_detail(db, route_id)
    if not detail:
        return error_response(404, "路线不存在")
    return success_response(detail)


@router.post("/process-routes")
async def create_process_route(
    body: ProcessRouteCreateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = route_svc.create_route(
            db,
            route_code=body.route_code,
            route_name=body.route_name,
            product_type_id=body.product_type_id,
            version=body.version,
            remark=body.remark,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        return success_response(route_svc._route_row_to_dict(obj), message="创建成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/process-routes/{route_id}")
async def update_process_route(
    route_id: str,
    body: ProcessRouteUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        payload = body.model_dump(exclude_unset=True, by_alias=False)
        obj = route_svc.update_route(
            db, route_id,
            user_id=user.get("user_id"),
            route_code=payload.get("route_code"),
            route_name=payload.get("route_name"),
            product_type_id=payload.get("product_type_id"),
            version=payload.get("version"),
            remark=payload.get("remark"),
        )
        return success_response(route_svc._route_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/process-routes/set-current")
async def set_current_process_route(
    body: SetCurrentBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        obj = route_svc.set_current_route(db, body.route_id, tenant_no)
        return success_response(route_svc._route_row_to_dict(obj), message="设置成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/process-routes/{route_id}")
async def delete_process_route(
    route_id: str,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        route_svc.soft_delete_route(db, route_id)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ----- 步骤操作 -----

@router.post("/process-routes/step/add")
async def add_route_step(
    body: StepAddBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        step_data = route_svc.add_step(
            db,
            route_id=body.route_id,
            process_id=body.process_id,
            step_order=body.step_order,
            tenant_no=tenant_no,
            user_id=user.get("user_id"),
        )
        return success_response(step_data, message="添加成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/process-routes/step/update")
async def update_route_step(
    body: StepUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        payload = body.model_dump(exclude_unset=True, by_alias=False)
        payload.pop("step_id", None)
        obj = route_svc.update_step(db, body.step_id, user_id=user.get("user_id"), **payload)
        return success_response(route_svc._step_row_to_dict(obj), message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/process-routes/step/reorder")
async def reorder_route_steps(
    body: StepReorderBody,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        route_svc.reorder_steps(db, body.route_id, body.step_ids)
        return success_response(None, message="排序成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.delete("/process-routes/step/{step_id}")
async def delete_route_step(
    step_id: str,
    _user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        route_svc.soft_delete_step(db, step_id)
        return success_response(None, message="删除成功")
    except ValueError as e:
        return error_response(400, str(e))


# ----- 步骤子表 -----

@router.put("/process-routes/step/reagent/update")
async def update_step_reagents(
    body: StepReagentUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items_data = [it.model_dump(by_alias=False) for it in body.items]
        route_svc.update_step_reagents(db, body.step_id, items_data, tenant_no, user.get("user_id"))
        return success_response(None, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/process-routes/step/consumable/update")
async def update_step_consumables(
    body: StepConsumableUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items_data = [it.model_dump(by_alias=False) for it in body.items]
        route_svc.update_step_consumables(db, body.step_id, items_data, tenant_no, user.get("user_id"))
        return success_response(None, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/process-routes/step/output/update")
async def update_step_outputs(
    body: StepOutputUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items_data = [it.model_dump(by_alias=False) for it in body.items]
        route_svc.update_step_outputs(db, body.step_id, items_data, tenant_no, user.get("user_id"))
        return success_response(None, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))


@router.put("/process-routes/step/data-collection/update")
async def update_step_data_collections(
    body: StepDataCollectionUpdateBody,
    user: dict = Depends(require_admin_or_super_admin),
    db: Session = Depends(get_db),
):
    try:
        tenant_no = _get_tenant_no(db, user)
        items_data = [it.model_dump(by_alias=False) for it in body.items]
        route_svc.update_step_data_collections(db, body.step_id, items_data, tenant_no, user.get("user_id"))
        return success_response(None, message="更新成功")
    except ValueError as e:
        return error_response(400, str(e))
  • Step 2: 验证后端启动正常

Run: cd backend && source venv/bin/activate && python -c "from app.api.production import router; print(f'Routes: {len(router.routes)}')" Expected: 输出路由数量（应比原来增加约 15 个）

  • Step 3: 启动后端验证 Swagger

Run: cd backend && source venv/bin/activate && uvicorn app.main:app --reload --port 8000 打开 http://127.0.0.1:8000/docs，搜索 process-routes，确认所有端点可见。

  • Step 4: 提交
git add backend/app/api/production.py
git commit -m "feat: add process route API endpoints (14 routes)"

Task 6: 前端 API 函数 — 追加到 production.ts

Files:

  • Modify: web-admin/src/api/production.ts (文件末尾追加)

  • [ ] Step 1: 在 production.ts 文件末尾追加工艺路线 API 函数


// ---------------------------------------------------------------------------
// 工艺路线管理
// ---------------------------------------------------------------------------

export function fetchProcessRouteList(params?: {
  page?: number
  pageSize?: number
  routeName?: string
  routeCode?: string
  productTypeId?: string
}) {
  return request.get<{
    list: any[]
    total: number
    page: number
    pageSize: number
    totalPages: number
  }>({ url: '/web/admin/process-routes', params })
}

export function fetchProcessRouteDetail(id: string) {
  return request.get<any>({ url: `/web/admin/process-routes/${id}` })
}

export function fetchCreateProcessRoute(data: Record<string, any>) {
  return request.post({ url: '/web/admin/process-routes', params: data })
}

export function fetchUpdateProcessRoute(id: string, data: Record<string, any>) {
  return request.put({ url: `/web/admin/process-routes/${id}`, params: data })
}

export function fetchSetCurrentProcessRoute(routeId: string) {
  return request.put({ url: '/web/admin/process-routes/set-current', params: { routeId } })
}

export function fetchDeleteProcessRoute(id: string) {
  return request.del({ url: `/web/admin/process-routes/${id}` })
}

// 步骤操作
export function fetchAddRouteStep(data: { routeId: string; processId: string; stepOrder?: number }) {
  return request.post({ url: '/web/admin/process-routes/step/add', params: data })
}

export function fetchUpdateRouteStep(data: Record<string, any>) {
  return request.put({ url: '/web/admin/process-routes/step/update', params: data })
}

export function fetchReorderRouteSteps(data: { routeId: string; stepIds: string[] }) {
  return request.put({ url: '/web/admin/process-routes/step/reorder', params: data })
}

export function fetchDeleteRouteStep(stepId: string) {
  return request.del({ url: `/web/admin/process-routes/step/${stepId}` })
}

// 步骤子表
export function fetchUpdateStepReagents(data: { stepId: string; items: any[] }) {
  return request.put({ url: '/web/admin/process-routes/step/reagent/update', params: data })
}

export function fetchUpdateStepConsumables(data: { stepId: string; items: any[] }) {
  return request.put({ url: '/web/admin/process-routes/step/consumable/update', params: data })
}

export function fetchUpdateStepOutputs(data: { stepId: string; items: any[] }) {
  return request.put({ url: '/web/admin/process-routes/step/output/update', params: data })
}

export function fetchUpdateStepDataCollections(data: { stepId: string; items: any[] }) {
  return request.put({ url: '/web/admin/process-routes/step/data-collection/update', params: data })
}
  • Step 2: 验证无 TypeScript 报错

Run: cd web-admin && pnpm build:typecheck Expected: 无新增错误

  • Step 3: 提交
git add web-admin/src/api/production.ts
git commit -m "feat: add process route frontend API functions"

Task 7: 前端路由 — 追加到 sqxp.ts

Files:

  • Modify: web-admin/src/router/modules/sqxp.ts

  • [ ] Step 1: 在 ProductionManagement 的 children 数组中追加路由条目

在 ProductionSop 路由对象之后（path: 'sop' 那一项），追加：

      {
        id: '90000000000000000000000000000010',
        name: 'ProductionProcessRoute',
        path: 'process-route',
        component: '/production/process-route/index',
        meta: { title: '工艺路线管理', keepAlive: true, menuCode: 'production:process-route:list' }
      }
  • Step 2: 提交
git add web-admin/src/router/modules/sqxp.ts
git commit -m "feat: add process route frontend route"

Task 8: 前端页面 — 列表页 + 路线图配置弹窗

Files:

  • Create: web-admin/src/views/production/process-route/index.vue

参考:

  • 列表页结构: src/views/system/unit/index.vue（标准列表模板）
  • 三栏拖拽弹窗: 设计规格文档「前端设计 > 路线图配置弹窗」章节
  • vue-draggable-plus 用法: src/views/system/product/index.vue（import { VueDraggable as draggable } from 'vue-draggable-plus'）
  • 工艺库详情编辑: src/views/production/process/index.vue（子表 inline 表格模式）

本页面约 1800-2200 行，包含：

  1. 列表页（搜索栏 + 工具栏 + 表格 + 分页）
  2. 新增/编辑路线弹窗
  3. 路线图配置弹窗（三栏布局）
    • 左栏：el-collapse 工艺库卡片（HTML5 draggable）
    • 中栏：vue-draggable-plus 步骤卡片排序
    • 右栏：步骤详情编辑（基础信息 + 4 个 tab 子表）

因页面代码量大，此处给出完整骨架和关键交互逻辑，实施时按骨架填充。

  • Step 1: 创建页面文件骨架

创建 web-admin/src/views/production/process-route/index.vue，核心结构：

<!-- 工艺路线管理 -->
<template>
  <div class="page-container">
    <header class="page-head">
      <h1>工艺路线管理</h1>
      <p class="page-desc">管理工艺路线基础信息及路线图配置。</p>
    </header>

    <section class="panel">
      <div class="panel-top">
        <h2 class="title-bar-heading">路线列表</h2>
      </div>

      <!-- 搜索栏 -->
      <div class="filter-block">
        <el-form :model="searchForm" class="filter-form" @submit.prevent>
          <div class="filter-grid">
            <el-form-item label="路线名称">
              <el-input v-model="searchForm.routeName" placeholder="路线名称" clearable @keyup.enter="handleSearch" />
            </el-form-item>
            <el-form-item label="路线编码">
              <el-input v-model="searchForm.routeCode" placeholder="路线编码" clearable @keyup.enter="handleSearch" />
            </el-form-item>
            <el-form-item label="产品类型">
              <el-select v-model="searchForm.productTypeId" placeholder="全部" clearable filterable style="width:100%">
                <el-option v-for="t in productTypeOptions" :key="t.id" :label="t.productTypeName" :value="t.id" />
              </el-select>
            </el-form-item>
          </div>
          <div class="filter-actions-row">
            <span class="filter-hint">按路线名称、编码、产品类型筛选。</span>
            <div class="filter-actions-end">
              <el-button class="btn-ghost" @click="handleReset">重置</el-button>
              <el-button class="btn-main" type="primary" @click="handleSearch">搜索</el-button>
            </div>
          </div>
        </el-form>
      </div>

      <!-- 工具栏 -->
      <div class="list-toolbar">
        <div class="list-toolbar-row">
          <el-button class="btn-main" type="primary" @click="handleAdd"><el-icon><Plus /></el-icon>新增路线</el-button>
        </div>
      </div>

      <!-- 表格 -->
      <div class="list-body list-wrap">
        <el-table class="list-table" :data="tableData" v-loading="loading" stripe row-key="id">
          <el-table-column type="index" label="序号" width="68" align="center" header-align="center"
            :index="(i) => (pagination.page - 1) * pagination.pageSize + i + 1" />
          <el-table-column prop="routeName" label="路线名称" min-width="140" align="center" header-align="center" />
          <el-table-column prop="routeCode" label="路线编码" min-width="130" align="center" header-align="center" />
          <el-table-column label="产品类型" min-width="130" align="center" header-align="center">
            <template #default="{ row }">{{ getProductTypeName(row.productTypeId) }}</template>
          </el-table-column>
          <el-table-column prop="version" label="版本号" min-width="100" align="center" header-align="center" />
          <el-table-column label="当前版本" width="100" align="center" header-align="center">
            <template #default="{ row }">
              <el-tag v-if="row.isCurrent === '1'" type="success" size="small">当前</el-tag>
            </template>
          </el-table-column>
          <el-table-column prop="remark" label="备注" min-width="120" align="center" header-align="center" show-overflow-tooltip />
          <el-table-column prop="createTime" label="创建时间" min-width="180" align="center" header-align="center" />
          <el-table-column label="操作" width="320" fixed="right" align="center" header-align="center">
            <template #default="{ row }">
              <el-button class="btn-op-item primary" @click="handleRouteConfig(row)">路线图配置</el-button>
              <el-button class="btn-op-item primary" @click="handleEdit(row)">编辑</el-button>
              <el-button v-if="row.isCurrent !== '1'" class="btn-op-item" @click="handleSetCurrent(row)">设为当前</el-button>
              <el-button class="btn-op-item" @click="handleDelete(row)">删除</el-button>
            </template>
          </el-table-column>
        </el-table>
        <div class="pagination list-pager-bar">
          <el-pagination :current-page="pagination.page" :page-size="pagination.pageSize" :total="pagination.total"
            :page-sizes="[10, 20, 50]" layout="total, sizes, prev, pager, next"
            @size-change="(val) => ((pagination.page = 1), (pagination.pageSize = val), loadData())"
            @current-change="(val) => ((pagination.page = val), loadData())" />
        </div>
      </div>

      <!-- 新增/编辑弹窗 -->
      <el-dialog v-model="dialogVisible" :title="dialogTitle" width="560px" class="dialog-form-layout"
        align-center destroy-on-close :close-on-click-modal="false">
        <el-form ref="formRef" :model="formData" :rules="formRules" label-position="top" scroll-to-error>
          <div class="form-grid">
            <el-form-item label="路线编码" prop="routeCode">
              <el-input v-model="formData.routeCode" placeholder="请输入路线编码" :disabled="!!editingId" />
            </el-form-item>
            <el-form-item label="路线名称" prop="routeName">
              <el-input v-model="formData.routeName" placeholder="请输入路线名称" />
            </el-form-item>
            <el-form-item label="产品类型">
              <el-select v-model="formData.productTypeId" placeholder="请选择产品类型" clearable filterable style="width:100%">
                <el-option v-for="t in productTypeOptions" :key="t.id" :label="t.productTypeName" :value="t.id" />
              </el-select>
            </el-form-item>
            <el-form-item label="版本号">
              <el-input v-model="formData.version" placeholder="如 V1.0" />
            </el-form-item>
            <el-form-item label="备注" class="form-col-span-2">
              <el-input v-model="formData.remark" placeholder="可选" />
            </el-form-item>
          </div>
        </el-form>
        <template #footer>
          <el-button @click="dialogVisible = false">取消</el-button>
          <el-button type="primary" @click="handleSubmit">确定</el-button>
        </template>
      </el-dialog>

      <!-- ★ 路线图配置弹窗（三栏布局） -->
      <el-dialog v-model="configDialogVisible" :title="`路线图配置 - ${configRouteName} ${configRouteVersion}`"
        width="90vw" class="dialog-form-layout route-config-dialog" align-center destroy-on-close
        :close-on-click-modal="false">
        <div class="route-config-layout">
          <!-- 左栏：工艺库面板 -->
          <div class="route-config-left">
            <h3>工艺库</h3>
            <el-collapse v-model="expandedCategories" class="process-category-collapse">
              <el-collapse-item v-for="group in processLibrary" :key="group.categoryId || '__none__'"
                :title="group.categoryName" :name="group.categoryId || '__none__'">
                <div v-for="proc in group.processes" :key="proc.id" class="process-source-card"
                  draggable="true" @dragstart="onDragStart($event, proc.id)">
                  <div class="process-source-card-name">{{ proc.processName }}</div>
                  <div class="process-source-card-code">{{ proc.processCode }}</div>
                </div>
                <el-empty v-if="group.processes.length === 0" description="暂无工艺" :image-size="32" />
              </el-collapse-item>
            </el-collapse>
          </div>

          <!-- 中栏：流程布局区 -->
          <div class="route-config-center"
            @dragover.prevent="onDragOverCenter($event)"
            @drop="onDropToCenter($event)">
            <h3>工艺流程布局</h3>
            <draggable v-model="stepList" class="step-card-list" handle=".step-card-handle"
              :animation="180" :force-fallback="true" @end="onStepDragEnd">
              <template #item="{ element, index }">
                <div class="step-card" :class="{ active: selectedStepId === element.id }"
                  @click="selectStep(element)">
                  <div class="step-card-header">
                    <el-icon class="step-card-handle"><Rank /></el-icon>
                    <span class="step-card-order">{{ index + 1 }}</span>
                    <span class="step-card-name">{{ element.processName || '未命名' }}</span>
                    <el-button link type="danger" size="small" @click.stop="handleDeleteStep(element)">
                      <el-icon><Delete /></el-icon>
                    </el-button>
                  </div>
                </div>
              </template>
            </draggable>
            <div v-if="stepList.length === 0" class="step-empty-hint">
              <el-empty description="从左侧拖入工艺卡片，或点击已添加的步骤查看详情" :image-size="64" />
            </div>
          </div>

          <!-- 右栏：工艺详情编辑 -->
          <div class="route-config-right">
            <template v-if="selectedStep">
              <h3>工艺详情</h3>
              <el-form :model="selectedStep" label-position="top" class="step-detail-form">
                <el-form-item label="工序名称">
                  <el-input v-model="selectedStep.processName" @change="onStepFieldChange" />
                </el-form-item>
                <el-form-item label="工序分类">
                  <el-select v-model="selectedStep.categoryId" clearable filterable style="width:100%" @change="onStepFieldChange">
                    <el-option v-for="cat in categoryOptions" :key="cat.id" :label="cat.categoryName || cat.name" :value="cat.id" />
                  </el-select>
                </el-form-item>
                <el-form-item label="描述">
                  <el-input v-model="selectedStep.description" type="textarea" :rows="2" @change="onStepFieldChange" />
                </el-form-item>
                <el-form-item label="设备名称">
                  <el-input v-model="selectedStep.equipmentName" @change="onStepFieldChange" />
                </el-form-item>
                <el-form-item>
                  <template #label>预估耗时<span class="form-label-tip">（min）</span></template>
                  <el-input v-model="selectedStep.estimatedDuration" @change="onStepFieldChange" />
                </el-form-item>
                <el-form-item label="关联SOP">
                  <el-select v-model="selectedStep.sopIdsArr" multiple clearable filterable style="width:100%" @change="onSopChange">
                    <el-option v-for="sop in sopOptions" :key="sop.id" :label="sop.sopName || sop.name" :value="sop.id" />
                  </el-select>
                </el-form-item>
              </el-form>

              <!-- 子表 tabs -->
              <el-tabs v-model="activeSubTab" class="step-sub-tabs">
                <el-tab-pane label="试剂" name="reagent">
                  <!-- 试剂表格：列(编码/名称/规格/标准用量/单位/关键物料)，底部添加按钮，行删除 -->
                  <!-- 修改后调用 fetchUpdateStepReagents -->
                </el-tab-pane>
                <el-tab-pane label="耗材" name="consumable">
                  <!-- 耗材表格：列(编码/名称/规格型号/标准用量/单位/必需/备注)，同上 -->
                </el-tab-pane>
                <el-tab-pane label="产出" name="output">
                  <!-- 产出表格：列(产出类型[下拉]/名称/条码)，同上 -->
                </el-tab-pane>
                <el-tab-pane label="数据采集" name="dataCollection">
                  <!-- 采集表格：列(采集名称/单位)，同上 -->
                </el-tab-pane>
              </el-tabs>
            </template>
            <div v-else class="step-empty-hint">
              <el-empty description="请从左侧工艺库拖入工艺，或点击已添加的步骤查看详情" :image-size="64" />
            </div>
          </div>
        </div>
      </el-dialog>
    </section>
  </div>
</template>

<script setup lang="ts">
import { ref, reactive, computed, onMounted, nextTick } from 'vue'
import { ElMessage, ElMessageBox } from 'element-plus'
import { VueDraggable as draggable } from 'vue-draggable-plus'
import {
  fetchProcessRouteList, fetchProcessRouteDetail,
  fetchCreateProcessRoute, fetchUpdateProcessRoute,
  fetchSetCurrentProcessRoute, fetchDeleteProcessRoute,
  fetchAddRouteStep, fetchUpdateRouteStep, fetchReorderRouteSteps,
  fetchDeleteRouteStep, fetchUpdateStepReagents, fetchUpdateStepConsumables,
  fetchUpdateStepOutputs, fetchUpdateStepDataCollections,
} from '@/api/production'
import { fetchAllProcesses } from '@/api/production'
import { fetchAllProcessCategories } from '@/api/production'
import { fetchAllSopDocuments } from '@/api/production'
import { fetchCellProductTypeList } from '@/api/admin'

// ===== 列表页状态 =====
const loading = ref(false)
const tableData = ref<any[]>([])
const pagination = reactive({ page: 1, pageSize: 10, total: 0 })
const searchForm = reactive({ routeName: '', routeCode: '', productTypeId: '' })
const productTypeOptions = ref<any[]>([])

// ===== 新增/编辑弹窗 =====
const dialogVisible = ref(false)
const dialogTitle = ref('新增路线')
const editingId = ref('')
const formData = reactive({ routeCode: '', routeName: '', productTypeId: '', version: '', remark: '' })
const formRules = { routeCode: [{ required: true, message: '请输入路线编码', trigger: 'blur' }], routeName: [{ required: true, message: '请输入路线名称', trigger: 'blur' }] }

// ===== 路线图配置弹窗 =====
const configDialogVisible = ref(false)
const configRouteId = ref('')
const configRouteName = ref('')
const configRouteVersion = ref('')
const stepList = ref<any[]>([])
const selectedStepId = ref('')
const selectedStep = computed(() => stepList.value.find(s => s.id === selectedStepId.value) || null)
const activeSubTab = ref('reagent')

// 工艺库数据（按分类分组）
const processLibrary = ref<any[]>([])
const expandedCategories = ref<string[]>([])
const categoryOptions = ref<any[]>([])
const sopOptions = ref<any[]>([])

// ===== 列表操作 =====
async function loadData() { /* 调用 fetchProcessRouteList */ }
function handleSearch() { pagination.page = 1; loadData() }
function handleReset() { searchForm.routeName = ''; searchForm.routeCode = ''; searchForm.productTypeId = ''; handleSearch() }
function handleAdd() { /* 打开新增弹窗 */ }
function handleEdit(row: any) { /* 打开编辑弹窗 */ }
async function handleSubmit() { /* 新增或编辑提交 */ }
async function handleSetCurrent(row: any) { /* 调用 fetchSetCurrentProcessRoute */ }
async function handleDelete(row: any) { /* 确认后调用 fetchDeleteProcessRoute */ }
function getProductTypeName(id: string) { return productTypeOptions.value.find(t => t.id === id)?.productTypeName || '-' }

// ===== 路线图配置 =====
async function handleRouteConfig(row: any) {
  /* 1. 设置 configRouteId/Name/Version
     2. 调用 fetchAllProcesses + fetchAllProcessCategories → 组装 processLibrary
     3. 调用 fetchProcessRouteDetail → 填充 stepList
     4. 打开 configDialogVisible */
}

// 左栏拖拽
function onDragStart(e: DragEvent, processId: string) {
  e.dataTransfer?.setData('processId', processId)
}
function onDragOverCenter(e: DragEvent) { /* 视觉提示 */ }
async function onDropToCenter(e: DragEvent) {
  const processId = e.dataTransfer?.getData('processId')
  if (!processId) return
  const res = await fetchAddRouteStep({ routeId: configRouteId.value, processId, stepOrder: stepList.value.length + 1 })
  stepList.value.push(res)
}

// 中栏排序
async function onStepDragEnd() {
  const stepIds = stepList.value.map(s => s.id)
  await fetchReorderRouteSteps({ routeId: configRouteId.value, stepIds })
  stepList.value.forEach((s, i) => { s.stepOrder = i + 1 })
}

// 选中步骤
function selectStep(step: any) { selectedStepId.value = step.id }

// 删除步骤
async function handleDeleteStep(step: any) {
  await ElMessageBox.confirm('确认删除该步骤？', '提示', { type: 'warning' })
  await fetchDeleteRouteStep(step.id)
  stepList.value = stepList.value.filter(s => s.id !== step.id)
  if (selectedStepId.value === step.id) selectedStepId.value = ''
}

// 右栏字段实时保存
async function onStepFieldChange() {
  if (!selectedStep.value) return
  await fetchUpdateRouteStep({ stepId: selectedStep.value.id, ...selectedStep.value })
}
function onSopChange(val: string[]) {
  if (selectedStep.value) {
    selectedStep.value.sopIds = val.join(',')
    onStepFieldChange()
  }
}

onMounted(() => {
  loadData()
  fetchCellProductTypeList({ pageSize: 999 }).then(res => { productTypeOptions.value = res.list || [] })
  fetchAllProcessCategories().then(res => { categoryOptions.value = res || [] })
  fetchAllSopDocuments().then(res => { sopOptions.value = res || [] })
})
</script>

关键 CSS（追加在 <style lang="scss" scoped> 中）：

.route-config-dialog {
  :deep(.el-dialog__body) {
    padding: 0;
    max-height: calc(100vh - 100px);
    overflow: hidden;
  }
}
.route-config-layout {
  display: flex;
  height: calc(100vh - 120px);
  gap: 0;
}
.route-config-left {
  width: 240px;
  border-right: 1px solid #e5e7eb;
  padding: 16px;
  overflow-y: auto;
  flex-shrink: 0;
}
.route-config-center {
  flex: 1;
  padding: 16px;
  overflow-y: auto;
  min-width: 200px;
}
.route-config-right {
  width: 360px;
  border-left: 1px solid #e5e7eb;
  padding: 16px;
  overflow-y: auto;
  flex-shrink: 0;
}

.process-source-card {
  padding: 8px 12px;
  border: 1px solid #d1d5db;
  border-radius: 6px;
  margin-bottom: 8px;
  cursor: grab;
  transition: box-shadow 0.2s;
  &:hover { box-shadow: 0 2px 8px rgba(0,0,0,0.1); }
}
.process-source-card-name { font-size: 14px; font-weight: 500; }
.process-source-card-code { font-size: 12px; color: #9ca3af; }

.step-card {
  padding: 12px 16px;
  border: 2px solid #e5e7eb;
  border-radius: 8px;
  margin-bottom: 8px;
  cursor: pointer;
  transition: border-color 0.2s;
  &:hover { border-color: #10b981; }
  &.active { border-color: #10b981; background: #ecfdf5; }
}
.step-card-header { display: flex; align-items: center; gap: 8px; }
.step-card-handle { cursor: grab; color: #9ca3af; }
.step-card-order {
  background: #10b981;
  color: white;
  border-radius: 50%;
  width: 24px;
  height: 24px;
  display: flex;
  align-items: center;
  justify-content: center;
  font-size: 12px;
  font-weight: 600;
  flex-shrink: 0;
}
.step-card-name { flex: 1; font-size: 14px; font-weight: 500; }

.step-empty-hint { padding: 40px 0; }
.step-detail-form { margin-bottom: 16px; }
  • Step 2: 填充子表 tab 内容

参照 production/process/index.vue 中试剂/耗材/产出物/数据采集的 inline 表格模式，为 4 个 el-tab-pane 填充完整的表格 UI：

  • 每个 tab 内使用 el-table 展示数据
  • 底部"添加"按钮追加空行
  • 每行有删除按钮

  • 修改后调用对应的 fetchUpdateStep* API 保存

  • [ ] Step 3: 验证页面正常渲染

Run: cd web-admin && pnpm dev 打开 http://localhost:端口/#/production/process-route，确认页面正常显示。

  • Step 4: 提交
git add web-admin/src/views/production/process-route/
git commit -m "feat: add process route management page with drag-and-drop route config"

Task 9: 联调验证

  • Step 1: 后端启动验证

Run: cd backend && source venv/bin/activate && uvicorn app.main:app --reload --port 8000 打开 http://127.0.0.1:8000/docs，确认所有 process-routes 端点可访问。

  • Step 2: 前端启动验证

Run: cd web-admin && pnpm dev 打开工艺路线管理页面，验证：

  1. 列表页渲染正常
  2. 新增路线弹窗可提交
  3. 编辑路线弹窗可修改
  4. 设为当前版本可点击
  5. 路线图配置弹窗可打开
  6. 左栏工艺库按分类展示
  7. 拖拽工艺到中栏可创建步骤
  8. 中栏步骤可排序
  9. 点击步骤右栏可编辑
  10. 删除步骤可确认删除
  • Step 3: 最终提交

如有修复，提交所有修复。