zkxw/backend/app/schemas/production_task.py

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


class ProductionTaskCreate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    work_order_id: str
    assignee_user_id: Optional[str] = Field(None, alias="assigneeUserId")
    task_type: Optional[str] = 'cell_storage'
    task_name: Optional[str] = None
    route_id: Optional[str] = None
    description: Optional[str] = None
    remark: Optional[str] = None


class ProductionTaskForwardBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    to_user_id: str = Field(..., alias="toUserId")
    remark: Optional[str] = None


class ProductionTaskAssigneeUpdate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    assignee_user_id: str = Field(..., alias="assigneeUserId")
    remark: Optional[str] = None


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


class ProductionTaskListByWorkOrdersBody(BaseModel):
    """按多个工单 ID 批量查询生产任务（POST body，避免 GET URL 过长）"""

    model_config = ConfigDict(populate_by_name=True)

    work_order_ids: List[str] = Field(..., alias="workOrderIds", min_length=1)
    include_step_summary: bool = Field(True, alias="includeStepSummary")
    page_size: int = Field(100, alias="pageSize", ge=1, le=500)


class ProductionTaskUpdate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    task_name: Optional[str] = None
    description: Optional[str] = None
    remark: Optional[str] = None
    attachments: Optional[str] = None
    equipment_ids: Optional[List[str]] = Field(None, alias="equipmentIds")


class ProductionTaskStatusUpdate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    task_status: str = Field(..., alias="taskStatus")
    conclusion: Optional[str] = Field(None, description="生产结论 cg=成功 sb=失败（完成时必填）")
    remark: Optional[str] = Field(None, description="备注（完成时可选）")


class TaskStepAdd(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    process_id: Optional[str] = None
    process_code: Optional[str] = None
    category_id: Optional[str] = None
    process_name: Optional[str] = None
    description: Optional[str] = None
    equipment_ids: Optional[List[str]] = Field(None, alias="equipmentIds")
    estimated_duration: Optional[str] = None
    sop_ids: Optional[str] = None
    sop_content: Optional[str] = Field(None, alias="sopContent")


class TaskStepUpdate(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    process_name: Optional[str] = None
    description: Optional[str] = None
    equipment_ids: Optional[List[str]] = Field(None, alias="equipmentIds")
    estimated_duration: Optional[str] = None
    actual_duration: Optional[str] = None
    sop_ids: Optional[str] = None
    sop_content: Optional[str] = Field(None, alias="sopContent")
    step_status: Optional[str] = None


class TaskStepMaterialItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    material_code: Optional[str] = None
    material_name: Optional[str] = None
    spec: Optional[str] = None
    spec_model: Optional[str] = None
    standard_usage: Optional[str] = None
    actual_usage: Optional[str] = None
    unit: Optional[str] = None
    requisition_usage_id: Optional[str] = Field(None, alias="requisitionUsageId")
    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
    remark: Optional[str] = None


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


class TaskInputItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    input_type: str = Field(..., alias="inputType")
    input_id: Optional[str] = Field(None, alias="inputId")
    input_code: Optional[str] = Field(None, alias="inputCode")
    input_name: Optional[str] = Field(None, alias="inputName")
    remark: Optional[str] = None


class TaskOutputItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    # 关联字段
    cell_product_id: Optional[str] = Field(None, alias="cellProductId")
    # 细胞产品完整字段
    product_code: Optional[str] = Field(None, alias="productCode")
    product_name: Optional[str] = Field(None, alias="productName")
    product_type_id: Optional[str] = Field(None, alias="productTypeId")
    spec_id: Optional[str] = Field(None, alias="specId")
    spec_name: Optional[str] = Field(None, alias="specName")
    passage: Optional[str] = None
    batch_no: Optional[str] = Field(None, alias="batchNo")
    storage_slot_id: Optional[str] = Field(None, alias="storageSlotId")
    expected_cell_count: Optional[float] = Field(None, alias="expectedCellCount")
    viability: Optional[float] = None
    freeze_date: Optional[str] = Field(None, alias="freezeDate")
    quantity: Optional[float] = None
    unit: Optional[str] = None
    expiry_date: Optional[str] = Field(None, alias="expiryDate")
    product_images: Optional[str] = Field(None, alias="productImages")
    product_status: Optional[str] = Field(None, alias="productStatus")
    source_type: Optional[str] = Field(None, alias="sourceType")
    # 兼容旧字段
    output_name: Optional[str] = Field(None, alias="outputName")
    output_barcode: Optional[str] = Field(None, alias="outputBarcode")
    remark: Optional[str] = None
    attachments: Optional[str] = None


class TaskStepReorderBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    task_id: str = Field(..., alias="taskId")
    step_ids: List[str] = Field(..., alias="stepIds")


class ProductionTaskStepDispatchItem(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    step_id: str = Field(..., alias="stepId")
    assignee_user_ids: List[str] = Field(default_factory=list, alias="assigneeUserIds")
    planned_start_time: Optional[str] = Field(None, alias="plannedStartTime")


class ProductionTaskStepDispatchBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    steps: List[ProductionTaskStepDispatchItem] = Field(..., min_length=1)


class SpawnTemporaryWorkOrderBody(BaseModel):
    model_config = ConfigDict(populate_by_name=True)
    remark: Optional[str] = None