入库生产工单 + 生产任务 + 细胞产品库实施计划

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: 后端 10 张新表，遵循全克隆模式（工艺路线步骤及子表完整克隆到任务步骤）。CRUD 层参考 qc_work_order.py（触发创建）和 process_route.py（主子表级联、克隆、软删除）。前端列表页遵循 unit/index.vue 模板，状态字段全部使用 <el-tag> 展示（参考 qc-work-order/index.vue 和 qc-task/index.vue）。

Tech Stack: FastAPI + SQLAlchemy + Pydantic v2 (后端) | Vue 3 + Element Plus + Tailwind CSS 4 (前端) | MySQL 5.7

设计规格文档: docs/superpowers/specs/2026-05-19-production-work-order-design.md

前端约定提醒:

所有状态/结论/类型字段在表格列和查看弹窗中使用 <el-tag> + 语义色（参考 views/quality-inspection/qc-work-order/index.vue 的 tagType 模式）
语义色规则: success=已完成/正常, warning=进行中/待确认, info=待配置/待执行/默认, danger=异常/已作废
弹窗表单 label-position="top", 不使用 backdrop-blur
遵循 Fresh Mint 色板 (--primary #10b981)

文件清单

操作	文件
新建	`docs/sql/production_work_order.sql`
追加	`backend/app/models/models.py`
追加	`backend/app/models/__init__.py`
新建	`backend/app/crud/production_work_order.py`
新建	`backend/app/crud/production_task.py`
新建	`backend/app/crud/cell_product.py`
新建	`backend/app/schemas/production_work_order.py`
新建	`backend/app/schemas/production_task.py`
新建	`backend/app/schemas/cell_product.py`
追加	`backend/app/api/production.py`
修改	`backend/app/crud/sample_reception.py` (触发点)
新建	`web-admin/src/api/production-work-order.ts`
新建	`web-admin/src/api/production-task.ts`
新建	`web-admin/src/api/cell-product.ts`
新建	`web-admin/src/views/production/cell-product/index.vue`
新建	`web-admin/src/views/production/production-work-order/index.vue`
新建	`web-admin/src/views/production/production-task/index.vue`
追加	`web-admin/src/router/modules/sqxp.ts`

Task 1: SQL DDL — 10 张新表 + 3 个菜单

Files:

Create: docs/sql/production_work_order.sql
[ ] Step 1: 创建 SQL DDL 文件

创建 docs/sql/production_work_order.sql，包含 10 张表的 CREATE TABLE 语句（IF NOT EXISTS）和 3 条菜单 INSERT。

完整的 SQL DDL 直接参考设计规格文档 docs/superpowers/specs/2026-05-19-production-work-order-design.md 第 3.3 节中的全部建表语句，包含：

biz_cell_product — 细胞产品库
biz_production_work_order — 入库生产工单
biz_production_task — 生产任务
biz_production_task_input — 任务输入子表
biz_production_task_step — 任务工艺步骤
biz_production_task_step_reagent — 步骤试剂（含 actual_usage）
biz_production_task_step_consumable — 步骤耗材（含 actual_usage）
biz_production_task_step_output — 步骤产出物（含 output_barcode）
biz_production_task_step_data_collection — 步骤数据采集（含 collection_value）
biz_production_task_output — 任务输出子表

菜单 INSERT（放在【生产管理】一级菜单下，参考现有菜单的 parent_id 和 order_num 排序）：

-- 菜单数据：假设生产管理一级菜单ID已知，需从 sys_menu 查询
-- 先查询生产管理菜单ID: SELECT id FROM sys_menu WHERE menu_name='生产管理' AND parent_id='1';
-- 以下用占位 {production_menu_id}，实际执行时替换

INSERT INTO `sys_menu` (`id`, `menu_name`, `parent_id`, `order_num`, `path`, `component`, `menu_type`, `visible`, `status`, `perms`, `icon`, `create_by`, `create_time`, `update_by`, `update_time`, `remark`)
VALUES
(UUID(), '入库生产工单', {production_menu_id}, 90, 'production-work-order', 'production/production-work-order/index', 'C', '1', '1', 'production:workOrder:list', NULL, '1', NOW(), '1', NOW(), NULL),
(UUID(), '生产任务', {production_menu_id}, 91, 'production-task', 'production/production-task/index', 'C', '1', '1', 'production:task:list', NULL, '1', NOW(), '1', NOW(), NULL),
(UUID(), '细胞产品库', {production_menu_id}, 92, 'cell-product', 'production/cell-product/index', 'C', '1', '1', 'production:cellProduct:list', NULL, '1', NOW(), '1', NOW(), NULL);

Step 2: 在数据库中执行 DDL

在开发环境执行 SQL 文件创建 10 张表和菜单数据。

Step 3: Commit

git add docs/sql/production_work_order.sql
git commit -m "feat(production): add DDL for production work order, task, and cell product tables"

Task 2: SQLAlchemy 模型 — 10 个新模型

Files:

Modify: backend/app/models/models.py
Modify: backend/app/models/__init__.py
[ ] Step 1: 在 models.py 末尾添加 10 个 SQLAlchemy 模型

参考现有模型模式（如 BizQcWorkOrder、BizProcessRouteStep 等），在 models.py 末尾追加以下 10 个模型类：

BizCellProduct — 对应 biz_cell_product
BizProductionWorkOrder — 对应 biz_production_work_order
BizProductionTask — 对应 biz_production_task
BizProductionTaskInput — 对应 biz_production_task_input
BizProductionTaskStep — 对应 biz_production_task_step
BizProductionTaskStepReagent — 对应 biz_production_task_step_reagent
BizProductionTaskStepConsumable — 对应 biz_production_task_step_consumable
BizProductionTaskStepOutput — 对应 biz_production_task_step_output
BizProductionTaskStepDataCollection — 对应 biz_production_task_step_data_collection
BizProductionTaskOutput — 对应 biz_production_task_output

每个模型遵循现有约定：

__tablename__ 对应表名
字段类型与 DDL 一致
to_dict() 方法（参考 BizQcWorkOrder.to_dict() 模式）
关系字段使用 relationship() 定义主子关系

关键模型示例（BizProductionWorkOrder）：

class BizProductionWorkOrder(Base):
    __tablename__ = 'biz_production_work_order'

    id = Column(String(64), primary_key=True)
    work_order_no = Column(String(32), nullable=False, comment='工单号(前缀SC)')
    reception_id = Column(String(64), nullable=False, comment='样本接收记录ID')
    protocol_sample_id = Column(String(64), nullable=False, comment='协议样本ID')
    collection_protocol_id = Column(String(64), nullable=False, comment='协议单ID')
    protocol_no = Column(String(64), comment='协议号快照')
    sample_code = Column(String(32), comment='样本编码快照')
    reception_source = Column(String(20), nullable=False, default='cell_storage')
    work_order_status = Column(String(16), nullable=False, default='dpz')
    # public fields...
    tenant_no = Column(String(64), nullable=False, default='zkxw')
    order_num = Column(Integer)
    remark = Column(String(512))
    create_by = Column(String(64))
    create_dept_id = Column(String(64), nullable=False, default='')
    create_time = Column(DateTime, nullable=False, default=datetime.now)
    update_by = Column(String(64))
    update_time = Column(DateTime, nullable=False, default=datetime.now, onupdate=datetime.now)
    del_flag = Column(SmallInteger, nullable=False, default=0)

    tasks = relationship('BizProductionTask', backref='work_order', foreign_keys='BizProductionTask.work_order_id')

    def to_dict(self):
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}

其他 9 个模型按相同模式编写，字段完全对齐 DDL。子表模型通过 task_id 或 step_id 外键关联父表。

Step 2: 在 init.py 中导出新模型

在 backend/app/models/__init__.py 的 import 列表中添加 10 个新模型的导入。

Step 3: 验证模型加载

cd backend && source venv/bin/activate && python -c "from app.models import BizProductionWorkOrder, BizProductionTask, BizCellProduct; print('Models loaded OK')"

Expected: Models loaded OK

Step 4: Commit

git add backend/app/models/models.py backend/app/models/__init__.py
git commit -m "feat(production): add SQLAlchemy models for work order, task, and cell product"

Task 3: Pydantic Schemas

Files:

Create: backend/app/schemas/production_work_order.py
Create: backend/app/schemas/production_task.py
Create: backend/app/schemas/cell_product.py
[ ] Step 1: 创建工单 schemas

创建 backend/app/schemas/production_work_order.py，参考 backend/app/schemas/process_route.py 的模式：

from pydantic import BaseModel
from typing import Optional
from datetime import datetime


class ProductionWorkOrderBase(BaseModel):
    remark: Optional[str] = None


class ProductionWorkOrderListQuery(BaseModel):
    page: int = 1
    page_size: int = 20
    work_order_no: Optional[str] = None
    work_order_status: Optional[str] = None
    protocol_no: Optional[str] = None
    sample_code: Optional[str] = None


class ProductionWorkOrderComplete(BaseModel):
    remark: Optional[str] = None

Step 2: 创建任务 schemas

创建 backend/app/schemas/production_task.py，包含任务 CRUD、步骤操作、输入/输出的所有 schema：

from pydantic import BaseModel
from typing import Optional, List
from datetime import datetime


class ProductionTaskCreate(BaseModel):
    work_order_id: str
    task_name: Optional[str] = None
    route_id: str  # 必选工艺路线
    description: Optional[str] = None
    remark: Optional[str] = None


class ProductionTaskListQuery(BaseModel):
    page: int = 1
    page_size: int = 20
    work_order_id: Optional[str] = None
    task_no: Optional[str] = None
    task_status: Optional[str] = None


class ProductionTaskUpdate(BaseModel):
    task_name: Optional[str] = None
    description: Optional[str] = None
    remark: Optional[str] = None
    attachments: Optional[str] = None


class ProductionTaskStatusUpdate(BaseModel):
    task_status: str


class TaskStepAdd(BaseModel):
    process_code: Optional[str] = None
    category_id: Optional[str] = None
    process_name: Optional[str] = None
    description: Optional[str] = None
    equipment_name: Optional[str] = None
    estimated_duration: Optional[str] = None
    sop_ids: Optional[str] = None


class TaskStepUpdate(BaseModel):
    process_name: Optional[str] = None
    description: Optional[str] = None
    equipment_name: Optional[str] = None
    estimated_duration: Optional[str] = None
    actual_duration: Optional[str] = None
    sop_ids: Optional[str] = None
    step_status: Optional[str] = None


class TaskStepMaterialItem(BaseModel):
    # 试剂或耗材通用结构
    material_code: Optional[str] = None
    material_name: Optional[str] = None
    spec: Optional[str] = None  # 试剂用spec, 耗材用spec_model
    spec_model: Optional[str] = None
    standard_usage: Optional[str] = None
    actual_usage: Optional[str] = None
    unit: Optional[str] = None
    is_key_material: Optional[str] = '0'
    is_required: Optional[str] = '1'
    remark: Optional[str] = None


class TaskStepOutputItem(BaseModel):
    output_type_id: Optional[str] = None
    output_name: Optional[str] = None
    output_barcode: Optional[str] = None


class TaskStepDataCollectionItem(BaseModel):
    collection_name: Optional[str] = None
    unit: Optional[str] = None
    collection_value: Optional[str] = None


class TaskInputItem(BaseModel):
    input_type: str  # sample / cell_product
    input_id: Optional[str] = None
    input_code: Optional[str] = None
    input_name: Optional[str] = None


class TaskOutputItem(BaseModel):
    cell_product_id: Optional[str] = None
    storage_slot_id: Optional[str] = None
    output_name: Optional[str] = None
    output_barcode: Optional[str] = None
    quantity: Optional[float] = None
    unit: Optional[str] = None
    remark: Optional[str] = None
    attachments: Optional[str] = None

Step 3: 创建细胞产品 schemas

创建 backend/app/schemas/cell_product.py：

from pydantic import BaseModel
from typing import Optional
from datetime import datetime


class CellProductCreate(BaseModel):
    product_code: str
    product_name: str
    product_type_id: Optional[str] = None
    spec_id: Optional[str] = None
    batch_no: Optional[str] = None
    storage_slot_id: Optional[str] = None
    product_status: Optional[str] = 'zk'
    source_type: Optional[str] = 'manual'
    quantity: Optional[float] = None
    unit: Optional[str] = None
    expiry_date: Optional[datetime] = None
    product_images: Optional[str] = None
    remark: Optional[str] = None


class CellProductUpdate(BaseModel):
    product_name: Optional[str] = None
    product_type_id: Optional[str] = None
    spec_id: Optional[str] = None
    batch_no: Optional[str] = None
    storage_slot_id: Optional[str] = None
    product_status: Optional[str] = None
    quantity: Optional[float] = None
    unit: Optional[str] = None
    expiry_date: Optional[datetime] = None
    product_images: Optional[str] = None
    remark: Optional[str] = None


class CellProductListQuery(BaseModel):
    page: int = 1
    page_size: int = 20
    product_code: Optional[str] = None
    product_name: Optional[str] = None
    product_type_id: Optional[str] = None
    product_status: Optional[str] = None

Step 4: Commit

git add backend/app/schemas/production_work_order.py backend/app/schemas/production_task.py backend/app/schemas/cell_product.py
git commit -m "feat(production): add Pydantic schemas for work order, task, and cell product"

Task 4: CRUD — 入库生产工单

Files:

Create: backend/app/crud/production_work_order.py
[ ] Step 1: 创建工单 CRUD

创建 backend/app/crud/production_work_order.py，参考 backend/app/crud/qc_work_order.py 中 create_work_order_from_reception() 的模式：

核心函数：

create_production_work_order_from_reception(db, reception, tenant_no, user_id, create_dept_id) — 样本接收触发创建
list_production_work_orders(db, tenant_no, query, ...) — 分页列表
get_production_work_order(db, id, tenant_no) — 详情
complete_production_work_order(db, id, tenant_no, user_id, ...) — 完成工单

工单号生成逻辑（前缀 SC）：

def _generate_work_order_no(db, tenant_no):
    today_str = datetime.now().strftime('%Y%m%d')
    prefix = f'SC{today_str}'
    # 查询今天已有的最大编号
    last = db.query(BizProductionWorkOrder).filter(
        BizProductionWorkOrder.tenant_no == tenant_no,
        BizProductionWorkOrder.work_order_no.like(f'{prefix}%'),
        BizProductionWorkOrder.del_flag == 0
    ).order_by(BizProductionWorkOrder.work_order_no.desc()).first()
    seq = 1
    if last:
        seq = int(last.work_order_no[-4:]) + 1
    return f'{prefix}{seq:04d}'

触发创建函数核心逻辑：

def create_production_work_order_from_reception(db, reception, tenant_no, user_id, create_dept_id):
    # 从 reception 获取协议号、样本编码等快照
    wo = BizProductionWorkOrder(
        id=str(uuid.uuid4()).replace('-', '').upper(),
        work_order_no=_generate_work_order_no(db, tenant_no),
        reception_id=reception.id,
        protocol_sample_id=reception.protocol_sample_id,
        collection_protocol_id=reception.collection_protocol_id,
        protocol_no=_get_protocol_no(db, reception),  # 查协议号快照
        sample_code=_get_sample_code(db, reception),   # 查样本编码快照
        reception_source=reception.reception_source,
        work_order_status='dpz',
        tenant_no=tenant_no,
        create_by=user_id,
        create_dept_id=create_dept_id,
    )
    db.add(wo)
    db.flush()
    return wo

注意：遵循租户隔离，查询带 tenant_no 过滤，数据权限按角色合并。

Step 2: Commit

git add backend/app/crud/production_work_order.py
git commit -m "feat(production): add CRUD for production work order"

Task 5: CRUD — 生产任务（核心克隆逻辑）

Files:

Create: backend/app/crud/production_task.py
[ ] Step 1: 创建任务 CRUD

创建 backend/app/crud/production_task.py，参考 backend/app/crud/process_route.py 的克隆和级联模式。

核心函数：

create_production_task(db, tenant_no, user_id, create_dept_id, data) — 创建任务并克隆工艺路线
list_production_tasks(db, tenant_no, query, ...) — 分页列表
get_production_task_detail(db, task_id, tenant_no) — 详情（含步骤+子表）
update_production_task(db, task_id, tenant_no, user_id, data) — 更新基本信息
update_task_status(db, task_id, tenant_no, user_id, status) — 更新状态
add_task_step(db, task_id, tenant_no, user_id, data) — 新增步骤
update_task_step(db, step_id, tenant_no, user_id, data) — 更新步骤
delete_task_step(db, step_id, tenant_no, user_id) — 删除步骤（软删除+级联子表）
replace_step_reagents(db, step_id, tenant_no, user_id, items) — 替换步骤试剂
replace_step_consumables(db, step_id, tenant_no, user_id, items) — 替换步骤耗材
replace_step_outputs(db, step_id, tenant_no, user_id, items) — 替换步骤产出物
replace_step_data_collections(db, step_id, tenant_no, user_id, items) — 替换数据采集项
set_task_inputs(db, task_id, tenant_no, user_id, items) — 设置任务输入
set_task_outputs(db, task_id, tenant_no, user_id, items) — 设置任务输出

克隆工艺路线的核心逻辑（参考 process_route.py 中从工艺库克隆到路线步骤的模式）：

def _clone_route_to_task(db, task_id, route_id, tenant_no, user_id, create_dept_id):
    """将工艺路线的所有步骤及子表克隆到生产任务"""
    steps = db.query(BizProcessRouteStep).filter(
        BizProcessRouteStep.route_id == route_id,
        BizProcessRouteStep.del_flag == 0
    ).order_by(BizProcessRouteStep.step_order).all()

    for route_step in steps:
        step_id = str(uuid.uuid4()).replace('-', '').upper()
        task_step = BizProductionTaskStep(
            id=step_id,
            task_id=task_id,
            step_order=route_step.step_order,
            process_code=route_step.process_code,
            category_id=route_step.category_id,
            process_name=route_step.process_name,
            description=route_step.description,
            equipment_name=route_step.equipment_name,
            estimated_duration=route_step.estimated_duration,
            sop_ids=route_step.sop_ids,
            step_status='dzx',
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id=create_dept_id,
        )
        db.add(task_step)

        # 克隆试剂
        _clone_step_reagents(db, step_id, route_step.id, tenant_no, user_id, create_dept_id)
        # 克隆耗材
        _clone_step_consumables(db, step_id, route_step.id, tenant_no, user_id, create_dept_id)
        # 克隆产出物
        _clone_step_outputs(db, step_id, route_step.id, tenant_no, user_id, create_dept_id)
        # 克隆数据采集项（collection_value 留空）
        _clone_step_data_collections(db, step_id, route_step.id, tenant_no, user_id, create_dept_id)

每个 _clone_step_* 辅助函数读取 biz_process_route_step_* 中 step_id == route_step.id 的记录，复制字段到对应的 biz_production_task_step_* 表中，将 FK 从 step_id 换为新的 task step id。试剂的 actual_usage、耗材的 actual_usage、采集项的 collection_value 均留空。

子表替换模式（参考 process_route.py）：

def replace_step_reagents(db, step_id, tenant_no, user_id, items):
    # 软删除旧数据
    db.query(BizProductionTaskStepReagent).filter(
        BizProductionTaskStepReagent.step_id == step_id,
        BizProductionTaskStepReagent.del_flag == 0
    ).update({'del_flag': 1})
    # 插入新数据
    for i, item in enumerate(items):
        db.add(BizProductionTaskStepReagent(
            id=str(uuid.uuid4()).replace('-', '').upper(),
            step_id=step_id,
            reagent_code=item.material_code,
            reagent_name=item.material_name,
            spec=item.spec,
            standard_usage=item.standard_usage,
            actual_usage=item.actual_usage,
            unit=item.unit,
            is_key_material=item.is_key_material,
            order_num=i,
            tenant_no=tenant_no,
            create_by=user_id,
            create_dept_id='',
        ))

Step 2: Commit

git add backend/app/crud/production_task.py
git commit -m "feat(production): add CRUD for production task with route cloning"

Task 6: CRUD — 细胞产品库

Files:

Create: backend/app/crud/cell_product.py
[ ] Step 1: 创建细胞产品 CRUD

创建 backend/app/crud/cell_product.py，标准 CRUD 模式（参考 backend/app/crud/consumable.py）：

核心函数：

list_cell_products(db, tenant_no, query, ...) — 分页列表
get_cell_product(db, id, tenant_no) — 详情
create_cell_product(db, tenant_no, user_id, create_dept_id, data) — 新增
update_cell_product(db, id, tenant_no, user_id, data) — 更新
delete_cell_product(db, id, tenant_no, user_id) — 软删除

产品编码唯一性校验（租户内）：

existing = db.query(BizCellProduct).filter(
    BizCellProduct.tenant_no == tenant_no,
    BizCellProduct.product_code == data.product_code,
    BizCellProduct.del_flag == 0
).first()
if existing:
    raise ValueError(f'产品编码 {data.product_code} 已存在')

Step 2: Commit

git add backend/app/crud/cell_product.py
git commit -m "feat(production): add CRUD for cell product"

Task 7: 触发点 — 样本接收后创建生产工单

Files:

Modify: backend/app/crud/sample_reception.py
[ ] Step 1: 修改 sample_reception.py 触发逻辑

在 backend/app/crud/sample_reception.py 中，找到第 757-766 行的 if conclusion == "js": 块，在现有质检工单创建之后添加生产工单创建：

# 现有代码（约第 757 行起）
if conclusion == "js":
    from app.crud import qc_work_order as qc_wo_svc
    qc_wo_svc.create_work_order_from_reception(
        db, reception=obj, tenant_no=tenant_no,
        user_id=user_id, create_dept_id=create_dept_id or "",
    )

    # ===== 新增：创建入库生产工单 =====
    from app.crud import production_work_order as prod_wo_svc
    prod_wo_svc.create_production_work_order_from_reception(
        db, reception=obj, tenant_no=tenant_no,
        user_id=user_id, create_dept_id=create_dept_id or "",
    )

注意：两个工单在同一个 db.commit() 事务中，保证原子性。

Step 2: 验证无语法错误

cd backend && source venv/bin/activate && python -c "from app.crud.sample_reception import *; print('Import OK')"

Expected: Import OK

Step 3: Commit

git add backend/app/crud/sample_reception.py
git commit -m "feat(production): trigger production work order creation on sample reception"

Task 8: API 路由

Files:

Modify: backend/app/api/production.py
[ ] Step 1: 在 production.py 中追加 3 组路由

在 backend/app/api/production.py 末尾追加路由。参考现有路由的认证模式（get_current_user、require_permissions）。

入库生产工单路由：

# ===================== 入库生产工单 =====================

@router.get('/production-work-orders')
async def api_list_production_work_orders(...):
    """入库生产工单列表"""
    ...

@router.get('/production-work-orders/{wo_id}')
async def api_get_production_work_order(wo_id: str, ...):
    """工单详情"""
    ...

@router.put('/production-work-orders/{wo_id}/complete')
async def api_complete_production_work_order(wo_id: str, ...):
    """完成工单"""
    ...

生产任务路由（含步骤 CRUD）：

# ===================== 生产任务 =====================

@router.post('/production-tasks')
async def api_create_production_task(...):
    """创建任务（选工艺路线，克隆步骤）"""
    ...

@router.get('/production-tasks')
async def api_list_production_tasks(...):
    """任务列表"""
    ...

@router.get('/production-tasks/{task_id}')
async def api_get_production_task(task_id: str, ...):
    """任务详情"""
    ...

@router.put('/production-tasks/{task_id}')
async def api_update_production_task(task_id: str, ...):
    """更新任务基本信息"""
    ...

@router.put('/production-tasks/{task_id}/status')
async def api_update_task_status(task_id: str, ...):
    """更新任务状态"""
    ...

@router.post('/production-tasks/{task_id}/steps')
async def api_add_task_step(task_id: str, ...):
    """新增步骤"""
    ...

@router.put('/production-tasks/{task_id}/steps/{step_id}')
async def api_update_task_step(task_id: str, step_id: str, ...):
    """更新步骤"""
    ...

@router.delete('/production-tasks/{task_id}/steps/{step_id}')
async def api_delete_task_step(task_id: str, step_id: str, ...):
    """删除步骤"""
    ...

@router.put('/production-tasks/{task_id}/steps/{step_id}/reagents')
async def api_replace_step_reagents(task_id: str, step_id: str, ...):
    """替换步骤试剂"""
    ...

@router.put('/production-tasks/{task_id}/steps/{step_id}/consumables')
async def api_replace_step_consumables(task_id: str, step_id: str, ...):
    """替换步骤耗材"""
    ...

@router.put('/production-tasks/{task_id}/steps/{step_id}/data-collection')
async def api_replace_step_data_collection(task_id: str, step_id: str, ...):
    """替换步骤数据采集"""
    ...

@router.post('/production-tasks/{task_id}/inputs')
async def api_set_task_inputs(task_id: str, ...):
    """设置任务输入"""
    ...

@router.post('/production-tasks/{task_id}/outputs')
async def api_set_task_outputs(task_id: str, ...):
    """设置任务输出"""
    ...

细胞产品库路由：

# ===================== 细胞产品库 =====================

@router.get('/cell-products')
async def api_list_cell_products(...):
    """产品列表"""
    ...

@router.get('/cell-products/{product_id}')
async def api_get_cell_product(product_id: str, ...):
    """产品详情"""
    ...

@router.post('/cell-products')
async def api_create_cell_product(...):
    """新增产品"""
    ...

@router.put('/cell-products/{product_id}')
async def api_update_cell_product(product_id: str, ...):
    """更新产品"""
    ...

@router.delete('/cell-products/{product_id}')
async def api_delete_cell_product(product_id: str, ...):
    """删除产品（软删除）"""
    ...

Step 2: 验证路由注册

cd backend && source venv/bin/activate && python -c "from app.api.production import router; print(f'Routes: {len(router.routes)}')"

Expected: 输出路由数量，比原来多 20+ 条

Step 3: Commit

git add backend/app/api/production.py
git commit -m "feat(production): add API routes for work order, task, and cell product"

Task 9: 前端 API 模块

Files:

Create: web-admin/src/api/production-work-order.ts
Create: web-admin/src/api/production-task.ts
Create: web-admin/src/api/cell-product.ts
[ ] Step 1: 创建工单 API

创建 web-admin/src/api/production-work-order.ts，参考 web-admin/src/api/qc-work-order.ts 模式：

import request from '@/utils/request'

// 工单列表
export function getProductionWorkOrderList(params: any) {
  return request({ url: '/web/admin/production-work-orders', method: 'get', params })
}

// 工单详情
export function getProductionWorkOrderDetail(id: string) {
  return request({ url: `/web/admin/production-work-orders/${id}`, method: 'get' })
}

// 完成工单
export function completeProductionWorkOrder(id: string, data: any) {
  return request({ url: `/web/admin/production-work-orders/${id}/complete`, method: 'put', data })
}

Step 2: 创建任务 API

创建 web-admin/src/api/production-task.ts：

import request from '@/utils/request'

// 创建任务
export function createProductionTask(data: any) {
  return request({ url: '/web/admin/production-tasks', method: 'post', data })
}

// 任务列表
export function getProductionTaskList(params: any) {
  return request({ url: '/web/admin/production-tasks', method: 'get', params })
}

// 任务详情
export function getProductionTaskDetail(id: string) {
  return request({ url: `/web/admin/production-tasks/${id}`, method: 'get' })
}

// 更新任务
export function updateProductionTask(id: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${id}`, method: 'put', data })
}

// 更新任务状态
export function updateProductionTaskStatus(id: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${id}/status`, method: 'put', data })
}

// 新增步骤
export function addTaskStep(taskId: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${taskId}/steps`, method: 'post', data })
}

// 更新步骤
export function updateTaskStep(taskId: string, stepId: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${taskId}/steps/${stepId}`, method: 'put', data })
}

// 删除步骤
export function deleteTaskStep(taskId: string, stepId: string) {
  return request({ url: `/web/admin/production-tasks/${taskId}/steps/${stepId}`, method: 'delete' })
}

// 替换步骤试剂
export function replaceStepReagents(taskId: string, stepId: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${taskId}/steps/${stepId}/reagents`, method: 'put', data })
}

// 替换步骤耗材
export function replaceStepConsumables(taskId: string, stepId: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${taskId}/steps/${stepId}/consumables`, method: 'put', data })
}

// 替换步骤数据采集
export function replaceStepDataCollection(taskId: string, stepId: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${taskId}/steps/${stepId}/data-collection`, method: 'put', data })
}

// 设置任务输入
export function setTaskInputs(taskId: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${taskId}/inputs`, method: 'post', data })
}

// 设置任务输出
export function setTaskOutputs(taskId: string, data: any) {
  return request({ url: `/web/admin/production-tasks/${taskId}/outputs`, method: 'post', data })
}

Step 3: 创建细胞产品 API

创建 web-admin/src/api/cell-product.ts：

import request from '@/utils/request'

export function getCellProductList(params: any) {
  return request({ url: '/web/admin/cell-products', method: 'get', params })
}

export function getCellProductDetail(id: string) {
  return request({ url: `/web/admin/cell-products/${id}`, method: 'get' })
}

export function createCellProduct(data: any) {
  return request({ url: '/web/admin/cell-products', method: 'post', data })
}

export function updateCellProduct(id: string, data: any) {
  return request({ url: `/web/admin/cell-products/${id}`, method: 'put', data })
}

export function deleteCellProduct(id: string) {
  return request({ url: `/web/admin/cell-products/${id}`, method: 'delete' })
}

Step 4: Commit

git add web-admin/src/api/production-work-order.ts web-admin/src/api/production-task.ts web-admin/src/api/cell-product.ts
git commit -m "feat(production): add frontend API modules for work order, task, and cell product"

Task 10: 前端路由注册

Files:

Modify: web-admin/src/router/modules/sqxp.ts
[ ] Step 1: 在 sqxp.ts 中添加 3 个路由

参考现有生产管理路由（如 process-route）的注册方式，添加：

{
  path: '/production/cell-product',
  name: 'ProductionCellProduct',
  component: () => import('@/views/production/cell-product/index.vue'),
  meta: { title: '细胞产品库', icon: '...', },
},
{
  path: '/production/production-work-order',
  name: 'ProductionWorkOrder',
  component: () => import('@/views/production/production-work-order/index.vue'),
  meta: { title: '入库生产工单', icon: '...', },
},
{
  path: '/production/production-task',
  name: 'ProductionTask',
  component: () => import('@/views/production/production-task/index.vue'),
  meta: { title: '生产任务', icon: '...', },
},

Step 2: Commit

git add web-admin/src/router/modules/sqxp.ts
git commit -m "feat(production): add frontend routes for work order, task, and cell product"

Task 11: 前端页面 — 细胞产品库

Files:

Create: web-admin/src/views/production/cell-product/index.vue
[ ] Step 1: 创建细胞产品库页面

参考模板页面 web-admin/src/views/system/unit/index.vue（标准 list 页面），创建 web-admin/src/views/production/cell-product/index.vue。

页面结构：

列表：产品编码、产品名称、产品类型（Tag）、规格、批次号、储物格位置、状态（Tag）、有效期
新增弹窗：产品编码、产品名称、产品类型（下拉关联 biz_cell_product_type）、规格（下拉关联 biz_cell_product_spec）、批次号、储物格（下拉关联库位管理叶子节点）、数量、单位、有效期、产品图片（附件上传）、备注
编辑弹窗：同新增
查看弹窗：只读展示
删除：软删除

Tag 色规则（状态字段 product_status）：

const productStatusTag = (status) => {
  const map = {
    zk: { label: '在库', type: 'success' },
    yc: { label: '已出库', type: 'warning' },
    ysy: { label: '已使用', type: 'info' },
  }
  return map[status] || { label: status, type: 'info' }
}

在表格列模板中：

<el-tag :type="productStatusTag(row.productStatus).type" size="small">
  {{ productStatusTag(row.productStatus).label }}
</el-tag>

Step 2: Commit

git add web-admin/src/views/production/cell-product/index.vue
git commit -m "feat(production): add cell product management page"

Task 12: 前端页面 — 入库生产工单

Files:

Create: web-admin/src/views/production/production-work-order/index.vue
[ ] Step 1: 创建入库生产工单页面

参考 web-admin/src/views/quality-inspection/qc-work-order/index.vue 的模式，创建 web-admin/src/views/production/production-work-order/index.vue。

页面结构：

列表：工单号、协议号、样本编码、来源类型（Tag）、工单状态（Tag）、创建时间
工具栏：搜索（工单号、状态）
行操作：查看详情、完成工单
详情弹窗/页面：展示工单基本信息 + 关联的生产任务列表

Tag 色规则：

工单状态 (work_order_status)：

const woStatusTag = (status) => {
  const map = {
    dpz: { label: '待配置', type: 'info' },
    ywc: { label: '已完成', type: 'success' },
  }
  return map[status] || { label: status, type: 'info' }
}

来源类型 (reception_source)：

const sourceTag = (source) => {
  const map = {
    cell_storage: { label: '细胞存储', type: 'success' },
    research_donation: { label: '科研捐赠', type: 'primary' },
  }
  return map[source] || { label: source, type: 'info' }
}

在表格和查看弹窗中均使用 <el-tag> 展示。

Step 2: Commit

git add web-admin/src/views/production/production-work-order/index.vue
git commit -m "feat(production): add production work order management page"

Task 13: 前端页面 — 生产任务

Files:

Create: web-admin/src/views/production/production-task/index.vue
[ ] Step 1: 创建生产任务页面

这是最复杂的前端页面。参考 web-admin/src/views/quality-inspection/qc-task/index.vue 和 web-admin/src/views/production/process-route/index.vue 的模式。

页面分两个视图：

视图一：任务列表

列表：任务号、所属工单号、任务名称、工艺路线、任务状态（Tag）、开始时间、完成时间
工具栏：搜索（任务号、状态、工单号筛选）
行操作：进入执行、查看详情

视图二：任务执行详情页（点击"进入执行"后展示）

顶部：任务基本信息（任务号、状态 Tag、所属工单）
第一区：任务输入（可编辑表格，选择样本/细胞产品）
第二区：工艺步骤列表
- 每个步骤卡片展示：步骤名、状态 Tag、设备、耗时
- 展开可编辑：试剂列表（标准用量 + 实际用量填写）、耗材列表（标准用量 + 实际用量填写）、产出物、数据采集（采集项名 + 单位 + 填写值）
- 步骤操作：开始执行（状态→jxz）、完成（状态→ywc）
- 增删工艺步骤按钮
第三区：任务输出（可编辑表格，选择细胞产品、储物格）
保存按钮

Tag 色规则：

任务状态 (task_status)：

const taskStatusTag = (status) => {
  const map = {
    dpz: { label: '待配置', type: 'info' },
    jxz: { label: '进行中', type: 'warning' },
    ywc: { label: '已完成', type: 'success' },
  }
  return map[status] || { label: status, type: 'info' }
}

步骤状态 (step_status)：

const stepStatusTag = (status) => {
  const map = {
    dzx: { label: '待执行', type: 'info' },
    jxz: { label: '进行中', type: 'warning' },
    ywc: { label: '已完成', type: 'success' },
  }
  return map[status] || { label: status, type: 'info' }
}

Step 2: Commit

git add web-admin/src/views/production/production-task/index.vue
git commit -m "feat(production): add production task execution page"

Task 14: 集成验证

Step 1: 后端启动验证

cd backend && source venv/bin/activate && python -c "from app.main import app; print('App started OK')"

Expected: App started OK

Step 2: 前端编译验证

cd web-admin && pnpm build

Expected: 构建成功，无 TS 错误

Step 3: 端到端手动验证

创建一个样本接收记录，accept_conclusion 选择 "js"（接收）
验证系统同时创建了入库质检工单和入库生产工单
在生产工单中创建生产任务，选择一个工艺路线
验证工艺步骤被完整克隆
在任务中增删步骤、录入实际用量和采集值
创建细胞产品并关联储物格
完成任务输出

Step 4: Final Commit

git add -A
git commit -m "feat(production): complete production work order and task system"

2026-05-19-production-work-order.md 39 KB Permalink Verlauf Originalformat

入库生产工单 + 生产任务 + 细胞产品库 实施计划

文件清单

Task 1: SQL DDL — 10 张新表 + 3 个菜单

Task 2: SQLAlchemy 模型 — 10 个新模型

Task 3: Pydantic Schemas

Task 4: CRUD — 入库生产工单

Task 5: CRUD — 生产任务（核心克隆逻辑）

Task 6: CRUD — 细胞产品库

Task 7: 触发点 — 样本接收后创建生产工单

Task 8: API 路由

Task 9: 前端 API 模块

Task 10: 前端路由注册

Task 11: 前端页面 — 细胞产品库

Task 12: 前端页面 — 入库生产工单

Task 13: 前端页面 — 生产任务

Task 14: 集成验证

2026-05-19-production-work-order.md 39 KB

Permalink Verlauf Originalformat

入库生产工单 + 生产任务 + 细胞产品库实施计划