feat(services): 重写核心业务逻辑 — 接收/驳回返工/裂变转交
task_service 新增 3 个核心业务函数: - receive_task(): PENDING→WIP, 记录 received_at, 仅 PENDING 状态可接收 - reject_task(): →REJECTED, 记录 reject_reason+completed_at, 防呆闭环: 自动查找父任务负责人, 创建 is_rework=True 的返工任务 - transfer_task(): →COMPLETED, 记录 completed_at, 裂变转交: 遍历 next_assignees 批量创建 PENDING 任务, 多路裂变(>1)时子任务挂载形成树状分支, virtual_warehouse 入库逻辑更新 Product.current_location_id - complete_task(): 保留兼容, 同步使用新状态常量 TASK_STATUS_* product_service 重构: - get_product_by_serial(): 新增 task_tree 返回完整递归任务树 - _load_task_tree(): 递归加载产品关联的所有任务嵌套关系
This commit is contained in:
@ -0,0 +1,40 @@
|
||||
"""业务逻辑层"""
|
||||
from app.services.product_service import (
|
||||
get_product_by_serial,
|
||||
get_product,
|
||||
create_product,
|
||||
update_product,
|
||||
get_all_products,
|
||||
)
|
||||
from app.services.task_service import (
|
||||
get_task,
|
||||
get_top_level_tasks,
|
||||
create_task,
|
||||
update_task,
|
||||
complete_task,
|
||||
receive_task,
|
||||
reject_task,
|
||||
transfer_task,
|
||||
create_subtask,
|
||||
get_all_tasks,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
# Product
|
||||
"get_product_by_serial",
|
||||
"get_product",
|
||||
"create_product",
|
||||
"update_product",
|
||||
"get_all_products",
|
||||
# Task
|
||||
"get_task",
|
||||
"get_top_level_tasks",
|
||||
"create_task",
|
||||
"update_task",
|
||||
"complete_task",
|
||||
"receive_task",
|
||||
"reject_task",
|
||||
"transfer_task",
|
||||
"create_subtask",
|
||||
"get_all_tasks",
|
||||
]
|
||||
|
||||
161
backend/app/services/product_service.py
Normal file
161
backend/app/services/product_service.py
Normal file
@ -0,0 +1,161 @@
|
||||
"""产品服务 — 业务逻辑层：扫码查询、CRUD"""
|
||||
from __future__ import annotations
|
||||
import uuid
|
||||
from fastapi import HTTPException, status
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.models.product import Product
|
||||
from app.models.production_order import ProductionOrder
|
||||
from app.models.task import Task
|
||||
from app.schemas.product import ProductCreate, ProductUpdate, ProductResponse, ProductScanResponse
|
||||
from app.schemas.task import TaskSummaryResponse, TaskResponse
|
||||
|
||||
|
||||
def _task_to_response(task: Task) -> TaskResponse:
|
||||
"""将 Task ORM 对象递归转为 TaskResponse（含子任务树）"""
|
||||
return TaskResponse(
|
||||
id=task.id,
|
||||
product_id=task.product_id,
|
||||
parent_task_id=task.parent_task_id,
|
||||
task_name=task.task_name,
|
||||
assignee_id=task.assignee_id,
|
||||
status=task.status,
|
||||
notify_parent_on_complete=task.notify_parent_on_complete,
|
||||
is_rework=task.is_rework,
|
||||
reject_reason=task.reject_reason,
|
||||
received_at=task.received_at,
|
||||
completed_at=task.completed_at,
|
||||
created_at=task.created_at,
|
||||
child_tasks=[_task_to_response(c) for c in task.child_tasks],
|
||||
)
|
||||
|
||||
|
||||
async def _load_task_tree(db: AsyncSession, product_id: uuid.UUID) -> list[TaskResponse]:
|
||||
"""递归加载产品下的完整任务树"""
|
||||
# 先取顶层任务
|
||||
result = await db.execute(
|
||||
select(Task)
|
||||
.options(selectinload(Task.child_tasks))
|
||||
.where(
|
||||
Task.product_id == product_id,
|
||||
Task.parent_task_id.is_(None),
|
||||
)
|
||||
.order_by(Task.created_at)
|
||||
)
|
||||
top_tasks = result.scalars().all()
|
||||
|
||||
# 递归加载每层子任务
|
||||
async def _load_children(t: Task):
|
||||
for child in t.child_tasks:
|
||||
child_result = await db.execute(
|
||||
select(Task)
|
||||
.options(selectinload(Task.child_tasks))
|
||||
.where(Task.id == child.id)
|
||||
)
|
||||
refreshed = child_result.scalar_one()
|
||||
t.child_tasks[t.child_tasks.index(child)] = refreshed
|
||||
await _load_children(refreshed)
|
||||
|
||||
for task in top_tasks:
|
||||
await _load_children(task)
|
||||
|
||||
return [_task_to_response(t) for t in top_tasks]
|
||||
|
||||
|
||||
async def get_product_by_serial(db: AsyncSession, serial_number: str) -> ProductScanResponse:
|
||||
"""扫码查询：根据 16 位序列号查出产品 + 所属订单 + 完整任务树"""
|
||||
result = await db.execute(
|
||||
select(Product)
|
||||
.options(
|
||||
selectinload(Product.order),
|
||||
selectinload(Product.parent_product),
|
||||
)
|
||||
.where(Product.serial_number == serial_number)
|
||||
)
|
||||
product = result.scalar_one_or_none()
|
||||
if not product:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"未找到序列号为 {serial_number} 的产品",
|
||||
)
|
||||
|
||||
# 获取顶层任务摘要（兼容旧接口）
|
||||
top_tasks_result = await db.execute(
|
||||
select(Task)
|
||||
.where(
|
||||
Task.product_id == product.id,
|
||||
Task.parent_task_id.is_(None),
|
||||
)
|
||||
.order_by(Task.created_at)
|
||||
)
|
||||
top_tasks = top_tasks_result.scalars().all()
|
||||
|
||||
# 获取完整任务树（递归嵌套，供前端渲染十字矩阵树状图）
|
||||
task_tree = await _load_task_tree(db, product.id)
|
||||
|
||||
return ProductScanResponse(
|
||||
id=product.id,
|
||||
serial_number=product.serial_number,
|
||||
order_id=product.order_id,
|
||||
order_no=product.order.order_no if product.order else "",
|
||||
material_id=product.material_id,
|
||||
parent_product_id=product.parent_product_id,
|
||||
current_location_id=product.current_location_id,
|
||||
status=product.status,
|
||||
created_at=product.created_at,
|
||||
top_level_tasks=[
|
||||
TaskSummaryResponse.model_validate(t) for t in top_tasks
|
||||
],
|
||||
task_tree=task_tree,
|
||||
)
|
||||
|
||||
|
||||
async def get_product(db: AsyncSession, product_id: uuid.UUID) -> Product:
|
||||
"""获取产品，不存在则 404"""
|
||||
result = await db.execute(
|
||||
select(Product)
|
||||
.options(selectinload(Product.order))
|
||||
.where(Product.id == product_id)
|
||||
)
|
||||
product = result.scalar_one_or_none()
|
||||
if not product:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"产品不存在: {product_id}",
|
||||
)
|
||||
return product
|
||||
|
||||
|
||||
async def create_product(db: AsyncSession, data: ProductCreate) -> ProductResponse:
|
||||
"""创建产品"""
|
||||
product = Product(**data.model_dump())
|
||||
db.add(product)
|
||||
await db.commit()
|
||||
await db.refresh(product)
|
||||
return ProductResponse.model_validate(product)
|
||||
|
||||
|
||||
async def update_product(db: AsyncSession, product_id: uuid.UUID, data: ProductUpdate) -> ProductResponse:
|
||||
"""更新产品"""
|
||||
product = await get_product(db, product_id)
|
||||
update_data = data.model_dump(exclude_unset=True)
|
||||
for field, value in update_data.items():
|
||||
setattr(product, field, value)
|
||||
await db.commit()
|
||||
await db.refresh(product)
|
||||
return ProductResponse.model_validate(product)
|
||||
|
||||
|
||||
async def get_all_products(db: AsyncSession, skip: int = 0, limit: int = 50) -> list[ProductResponse]:
|
||||
"""获取产品列表"""
|
||||
result = await db.execute(
|
||||
select(Product)
|
||||
.options(selectinload(Product.order))
|
||||
.offset(skip)
|
||||
.limit(limit)
|
||||
.order_by(Product.created_at.desc())
|
||||
)
|
||||
products = result.scalars().all()
|
||||
return [ProductResponse.model_validate(p) for p in products]
|
||||
593
backend/app/services/task_service.py
Normal file
593
backend/app/services/task_service.py
Normal file
@ -0,0 +1,593 @@
|
||||
"""任务服务 — 核心业务逻辑：接收、驳回返工、裂变转交、无限嵌套子任务"""
|
||||
from __future__ import annotations
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from fastapi import HTTPException, status
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
from app.models.task import Task, TASK_STATUS_PENDING, TASK_STATUS_WIP, TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED
|
||||
from app.models.product import Product
|
||||
from app.models.task_log import TaskLog
|
||||
from app.schemas.task import (
|
||||
TaskCreate,
|
||||
TaskUpdate,
|
||||
TaskCompleteRequest,
|
||||
TaskRejectRequest,
|
||||
TaskTransferRequest,
|
||||
SubtaskCreate,
|
||||
TaskResponse,
|
||||
TaskCompleteResponse,
|
||||
TaskTransferResponse,
|
||||
TaskSummaryResponse,
|
||||
TaskListResponse,
|
||||
)
|
||||
|
||||
# 特殊位置常量
|
||||
VIRTUAL_WAREHOUSE = "virtual_warehouse"
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 内部辅助函数
|
||||
# ============================================================
|
||||
|
||||
async def _get_task_or_404(db: AsyncSession, task_id: uuid.UUID) -> Task:
|
||||
"""获取任务，不存在则 404"""
|
||||
result = await db.execute(
|
||||
select(Task)
|
||||
.options(
|
||||
selectinload(Task.child_tasks),
|
||||
selectinload(Task.parent_task),
|
||||
selectinload(Task.product),
|
||||
)
|
||||
.where(Task.id == task_id)
|
||||
)
|
||||
task = result.scalar_one_or_none()
|
||||
if not task:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"任务不存在: {task_id}",
|
||||
)
|
||||
return task
|
||||
|
||||
|
||||
async def _get_task_with_children_recursive(db: AsyncSession, task_id: uuid.UUID) -> Task:
|
||||
"""递归加载任务及其所有子孙任务"""
|
||||
result = await db.execute(
|
||||
select(Task)
|
||||
.options(selectinload(Task.child_tasks))
|
||||
.where(Task.id == task_id)
|
||||
)
|
||||
task = result.scalar_one_or_none()
|
||||
if not task:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"任务不存在: {task_id}",
|
||||
)
|
||||
|
||||
# 递归加载每一层子任务
|
||||
async def _load_children(t: Task):
|
||||
for child in t.child_tasks:
|
||||
child_result = await db.execute(
|
||||
select(Task)
|
||||
.options(selectinload(Task.child_tasks))
|
||||
.where(Task.id == child.id)
|
||||
)
|
||||
refreshed_child = child_result.scalar_one()
|
||||
t.child_tasks[t.child_tasks.index(child)] = refreshed_child
|
||||
await _load_children(refreshed_child)
|
||||
|
||||
await _load_children(task)
|
||||
return task
|
||||
|
||||
|
||||
def _to_response(task: Task) -> TaskResponse:
|
||||
"""将 Task ORM 对象转为递归 TaskResponse"""
|
||||
return TaskResponse(
|
||||
id=task.id,
|
||||
product_id=task.product_id,
|
||||
parent_task_id=task.parent_task_id,
|
||||
task_name=task.task_name,
|
||||
assignee_id=task.assignee_id,
|
||||
status=task.status,
|
||||
notify_parent_on_complete=task.notify_parent_on_complete,
|
||||
is_rework=task.is_rework,
|
||||
reject_reason=task.reject_reason,
|
||||
received_at=task.received_at,
|
||||
completed_at=task.completed_at,
|
||||
created_at=task.created_at,
|
||||
child_tasks=[_to_response(c) for c in task.child_tasks],
|
||||
)
|
||||
|
||||
|
||||
async def _check_all_critical_children_completed(
|
||||
db: AsyncSession, task_id: uuid.UUID
|
||||
) -> tuple[bool, list[str]]:
|
||||
"""
|
||||
检查指定任务下所有 notify_parent_on_complete=True 的子任务是否都已完成。
|
||||
返回 (是否全部完成, 未完成的子任务名称列表)。
|
||||
"""
|
||||
result = await db.execute(
|
||||
select(Task).where(
|
||||
Task.parent_task_id == task_id,
|
||||
Task.notify_parent_on_complete.is_(True),
|
||||
)
|
||||
)
|
||||
critical_children = result.scalars().all()
|
||||
|
||||
incomplete = [
|
||||
child.task_name
|
||||
for child in critical_children
|
||||
if child.status != TASK_STATUS_COMPLETED
|
||||
]
|
||||
return len(incomplete) == 0, incomplete
|
||||
|
||||
|
||||
async def _create_task_log(
|
||||
db: AsyncSession,
|
||||
task_id: uuid.UUID,
|
||||
action_type: str,
|
||||
operator_id: str | None = None,
|
||||
remark: str | None = None,
|
||||
) -> TaskLog:
|
||||
"""创建任务操作日志"""
|
||||
log = TaskLog(
|
||||
task_id=task_id,
|
||||
operator_id=operator_id,
|
||||
action_type=action_type,
|
||||
remark=remark,
|
||||
)
|
||||
db.add(log)
|
||||
return log
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 公开 API — 查询
|
||||
# ============================================================
|
||||
|
||||
async def get_task(db: AsyncSession, task_id: uuid.UUID) -> TaskResponse:
|
||||
"""获取任务详情 — 递归包含所有子任务"""
|
||||
task = await _get_task_with_children_recursive(db, task_id)
|
||||
return _to_response(task)
|
||||
|
||||
|
||||
async def get_top_level_tasks(db: AsyncSession, product_id: uuid.UUID) -> list[TaskSummaryResponse]:
|
||||
"""获取产品的顶层任务列表"""
|
||||
result = await db.execute(
|
||||
select(Task)
|
||||
.where(
|
||||
Task.product_id == product_id,
|
||||
Task.parent_task_id.is_(None),
|
||||
)
|
||||
.order_by(Task.created_at)
|
||||
)
|
||||
tasks = result.scalars().all()
|
||||
return [TaskSummaryResponse.model_validate(t) for t in tasks]
|
||||
|
||||
|
||||
async def create_task(db: AsyncSession, data: TaskCreate) -> TaskResponse:
|
||||
"""创建任务"""
|
||||
task = Task(**data.model_dump())
|
||||
db.add(task)
|
||||
await db.commit()
|
||||
await db.refresh(task)
|
||||
return _to_response(task)
|
||||
|
||||
|
||||
async def update_task(db: AsyncSession, task_id: uuid.UUID, data: TaskUpdate) -> TaskResponse:
|
||||
"""更新任务"""
|
||||
task = await _get_task_or_404(db, task_id)
|
||||
update_data = data.model_dump(exclude_unset=True)
|
||||
for field, value in update_data.items():
|
||||
setattr(task, field, value)
|
||||
await db.commit()
|
||||
await db.refresh(task)
|
||||
return _to_response(task)
|
||||
|
||||
|
||||
async def get_all_tasks(
|
||||
db: AsyncSession, product_id: uuid.UUID | None = None, skip: int = 0, limit: int = 50
|
||||
) -> TaskListResponse:
|
||||
"""获取任务列表，可按产品筛选"""
|
||||
stmt = select(Task).options(selectinload(Task.child_tasks))
|
||||
if product_id:
|
||||
stmt = stmt.where(Task.product_id == product_id)
|
||||
stmt = stmt.offset(skip).limit(limit).order_by(Task.created_at.desc())
|
||||
|
||||
result = await db.execute(stmt)
|
||||
tasks = result.scalars().all()
|
||||
|
||||
# 构造嵌套响应时只返回顶层任务
|
||||
all_tasks = [_to_response(t) for t in tasks if t.parent_task_id is None]
|
||||
return TaskListResponse(tasks=all_tasks, total=len(all_tasks))
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 核心业务 1：确认接收 (PENDING → WIP)
|
||||
# ============================================================
|
||||
|
||||
async def receive_task(
|
||||
db: AsyncSession, task_id: uuid.UUID, operator_id: str | None = None
|
||||
) -> TaskResponse:
|
||||
"""
|
||||
操作员确认接收任务。
|
||||
|
||||
校验：只有状态为 PENDING 的任务可接收。
|
||||
动作：状态改为 WIP，记录 received_at 为当前时间。
|
||||
"""
|
||||
task = await _get_task_or_404(db, task_id)
|
||||
|
||||
# 校验：只有 PENDING 状态可接收
|
||||
if task.status != TASK_STATUS_PENDING:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"只有待接收(PENDING)状态的任务可接收，当前状态: {task.status}",
|
||||
)
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
task.status = TASK_STATUS_WIP
|
||||
task.received_at = now
|
||||
|
||||
await _create_task_log(
|
||||
db, task_id,
|
||||
action_type="receive",
|
||||
operator_id=operator_id,
|
||||
remark=f"操作员确认接收任务「{task.task_name}」",
|
||||
)
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(task)
|
||||
|
||||
return _to_response(task)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 核心业务 2：品质驳回 (→ REJECTED + 返工闭环)
|
||||
# ============================================================
|
||||
|
||||
async def reject_task(
|
||||
db: AsyncSession, task_id: uuid.UUID, request: TaskRejectRequest, operator_id: str | None = None
|
||||
) -> TaskResponse:
|
||||
"""
|
||||
品质驳回：将当前任务标记为 REJECTED，并自动创建返工任务给上一道工序负责人。
|
||||
|
||||
防呆闭环逻辑：
|
||||
1. 将当前任务状态改为 REJECTED，记录 reject_reason 和 completed_at。
|
||||
2. 查找当前任务的父任务 (parent_task) 的负责人 (assignee_id)。
|
||||
- 若有父任务：返工任务分配给父任务的 assignee_id。
|
||||
- 若无父任务（顶层任务）：返工任务分配给当前任务自己的 assignee_id。
|
||||
3. 为该负责人新建一个完全一样的任务，但 is_rework=True，status=PENDING。
|
||||
4. 新返工任务挂在同一个 parent_task_id 下（与原任务同级）。
|
||||
"""
|
||||
task = await _get_task_or_404(db, task_id)
|
||||
|
||||
# 校验：不能重复驳回已完成/已驳回的任务
|
||||
if task.status in (TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"任务状态为 {task.status}，无法驳回",
|
||||
)
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# --- 1. 标记当前任务为已驳回 ---
|
||||
task.status = TASK_STATUS_REJECTED
|
||||
task.reject_reason = request.reason
|
||||
task.completed_at = now
|
||||
|
||||
await _create_task_log(
|
||||
db, task_id,
|
||||
action_type="reject",
|
||||
operator_id=operator_id,
|
||||
remark=f"品质驳回: {request.reason}",
|
||||
)
|
||||
|
||||
# --- 2. 确定返工任务的负责人（上一道工序的负责人） ---
|
||||
rework_assignee_id: str | None = None
|
||||
if task.parent_task_id:
|
||||
# 有父任务：查父任务的 assignee
|
||||
parent_result = await db.execute(
|
||||
select(Task).where(Task.id == task.parent_task_id)
|
||||
)
|
||||
parent_task = parent_result.scalar_one_or_none()
|
||||
if parent_task:
|
||||
rework_assignee_id = parent_task.assignee_id
|
||||
else:
|
||||
# 无父任务（顶层）：返工给自己
|
||||
rework_assignee_id = task.assignee_id
|
||||
|
||||
# --- 3. 创建返工任务 ---
|
||||
rework_task = Task(
|
||||
product_id=task.product_id,
|
||||
parent_task_id=task.parent_task_id, # 与原任务同级
|
||||
task_name=task.task_name,
|
||||
assignee_id=rework_assignee_id,
|
||||
status=TASK_STATUS_PENDING,
|
||||
notify_parent_on_complete=task.notify_parent_on_complete,
|
||||
is_rework=True,
|
||||
)
|
||||
db.add(rework_task)
|
||||
await db.flush()
|
||||
|
||||
await _create_task_log(
|
||||
db, rework_task.id,
|
||||
action_type="create",
|
||||
operator_id=operator_id,
|
||||
remark=f"返工任务（驳回自「{task.task_name}」，原因: {request.reason}），分配给 {rework_assignee_id}",
|
||||
)
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(task)
|
||||
|
||||
return _to_response(task)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 核心业务 3：完工并裂变转交 (→ COMPLETED + 裂变创建下家任务)
|
||||
# ============================================================
|
||||
|
||||
async def transfer_task(
|
||||
db: AsyncSession, task_id: uuid.UUID, request: TaskTransferRequest, operator_id: str | None = None
|
||||
) -> TaskTransferResponse:
|
||||
"""
|
||||
完工并裂变转交：
|
||||
|
||||
动作 1（闭环当前节点）：
|
||||
- 将当前任务状态改为 COMPLETED，记录 completed_at。
|
||||
|
||||
动作 2（解析下家）：
|
||||
- 遍历 next_assignees 列表。
|
||||
- 如果包含 'virtual_warehouse'，则将 Product 的 current_location_id 设为 'virtual_warehouse'。
|
||||
- 为每一个 assignee_id（非 virtual_warehouse）新建一条 Task 记录（状态 PENDING）。
|
||||
|
||||
裂变逻辑：
|
||||
- 如果 len(next_assignees) > 1：多路裂变 → 所有新任务挂到当前任务下（parent_task_id = 当前任务ID）。
|
||||
- 如果当前任务本身就是子任务（有 parent_task_id）：单路转交也挂到同一父任务下。
|
||||
- 否则（顶层单路转交）：新任务与当前任务同级（parent_task_id = None）。
|
||||
- 更新 Product 的 current_location_id 为对应的人员（非仓库）。
|
||||
"""
|
||||
task = await _get_task_or_404(db, task_id)
|
||||
|
||||
# 校验：不能重复完成
|
||||
if task.status == TASK_STATUS_COMPLETED:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail=f"任务「{task.task_name}」已经完成，无需重复操作",
|
||||
)
|
||||
|
||||
# 校验：卡点逻辑 — 检查关键子任务
|
||||
all_done, incomplete_names = await _check_all_critical_children_completed(db, task_id)
|
||||
if not all_done:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"请等待相关子任务完成：{', '.join(incomplete_names)}",
|
||||
)
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# --- 动作 1：闭环当前节点 ---
|
||||
task.status = TASK_STATUS_COMPLETED
|
||||
task.completed_at = now
|
||||
|
||||
await _create_task_log(
|
||||
db, task_id,
|
||||
action_type="complete",
|
||||
operator_id=operator_id,
|
||||
remark=request.note or f"完成任务「{task.task_name}」，转交至下一道工序",
|
||||
)
|
||||
|
||||
# --- 动作 2：解析下家 & 裂变 ---
|
||||
has_warehouse = VIRTUAL_WAREHOUSE in request.next_assignees
|
||||
real_assignees = [a for a in request.next_assignees if a != VIRTUAL_WAREHOUSE]
|
||||
|
||||
# 判断是否需要裂变（树状结构）
|
||||
is_fission = len(real_assignees) > 1
|
||||
is_child_task = task.parent_task_id is not None
|
||||
|
||||
created_tasks: list[Task] = []
|
||||
|
||||
for assignee_id in real_assignees:
|
||||
# 确定新任务的 parent_task_id（裂变逻辑）
|
||||
if is_fission:
|
||||
# 多路裂变：所有新任务挂在当前任务下，形成树状分支
|
||||
new_parent_task_id = task.id
|
||||
elif is_child_task:
|
||||
# 当前任务是子任务，单路转交也保持在同一父任务下
|
||||
new_parent_task_id = task.parent_task_id
|
||||
else:
|
||||
# 顶层单路转交：同级
|
||||
new_parent_task_id = None
|
||||
|
||||
new_task = Task(
|
||||
product_id=task.product_id,
|
||||
parent_task_id=new_parent_task_id,
|
||||
task_name=request.next_task_name,
|
||||
assignee_id=assignee_id,
|
||||
status=TASK_STATUS_PENDING,
|
||||
notify_parent_on_complete=False,
|
||||
is_rework=False,
|
||||
)
|
||||
db.add(new_task)
|
||||
created_tasks.append(new_task)
|
||||
|
||||
# 批量 flush 以生成 ID
|
||||
await db.flush()
|
||||
|
||||
for nt in created_tasks:
|
||||
await _create_task_log(
|
||||
db, nt.id,
|
||||
action_type="create",
|
||||
operator_id=operator_id,
|
||||
remark=f"由任务「{task.task_name}」裂变转交创建，分配给 {nt.assignee_id}",
|
||||
)
|
||||
|
||||
# --- 更新 Product 的 current_location_id ---
|
||||
product_result = await db.execute(
|
||||
select(Product).where(Product.id == task.product_id)
|
||||
)
|
||||
product = product_result.scalar_one_or_none()
|
||||
if product:
|
||||
if has_warehouse and not real_assignees:
|
||||
# 只有仓库，没有实际人员 → 入库
|
||||
product.current_location_id = VIRTUAL_WAREHOUSE
|
||||
elif real_assignees:
|
||||
# 有实际人员 → 指向第一个人员（或可考虑指向多人中的主负责人）
|
||||
product.current_location_id = real_assignees[0]
|
||||
|
||||
await db.commit()
|
||||
|
||||
# --- 构建响应 ---
|
||||
refreshed_task = await _get_task_with_children_recursive(db, task_id)
|
||||
created_task_responses = []
|
||||
for nt in created_tasks:
|
||||
await db.refresh(nt)
|
||||
created_task_responses.append(_to_response(nt))
|
||||
|
||||
assignee_list = ", ".join(real_assignees)
|
||||
location_info = ""
|
||||
if has_warehouse:
|
||||
location_info = "，产品已入库(virtual_warehouse)"
|
||||
|
||||
return TaskTransferResponse(
|
||||
completed_task=_to_response(refreshed_task),
|
||||
created_tasks=created_task_responses,
|
||||
message=(
|
||||
f"任务「{task.task_name}」已完成，"
|
||||
f"已创建 {len(created_tasks)} 个下一道工序任务「{request.next_task_name}」"
|
||||
f"（接收人: {assignee_list}）{location_info}"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 保留兼容：旧版 complete_task（单步完成/转交）
|
||||
# ============================================================
|
||||
|
||||
async def complete_task(
|
||||
db: AsyncSession, task_id: uuid.UUID, request: TaskCompleteRequest
|
||||
) -> TaskCompleteResponse:
|
||||
"""
|
||||
核心业务：完成任务 + 可选创建下一步任务。
|
||||
|
||||
逻辑：
|
||||
1. 检查当前任务是否已完成（幂等）
|
||||
2. 检查所有 notify_parent_on_complete=True 的子任务是否都已完成
|
||||
→ 如果存在未完成的关键子任务，返回 400 错误
|
||||
3. 将当前任务状态改为 completed，记录日志
|
||||
4. 如果提供了 next_task_name 和 next_assignee_id，创建下一步任务
|
||||
5. 返回完成结果
|
||||
"""
|
||||
task = await _get_task_or_404(db, task_id)
|
||||
|
||||
# 注入子任务数据到当前对象以便后续检查
|
||||
children_result = await db.execute(
|
||||
select(Task).where(Task.parent_task_id == task_id)
|
||||
)
|
||||
task.child_tasks = children_result.scalars().all()
|
||||
|
||||
# --- 1. 幂等检查 ---
|
||||
if task.status == TASK_STATUS_COMPLETED:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_409_CONFLICT,
|
||||
detail=f"任务「{task.task_name}」已经完成，无需重复操作",
|
||||
)
|
||||
|
||||
# --- 2. 卡点逻辑：检查关键子任务 ---
|
||||
all_done, incomplete_names = await _check_all_critical_children_completed(db, task_id)
|
||||
if not all_done:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"请等待相关子任务完成：{', '.join(incomplete_names)}",
|
||||
)
|
||||
|
||||
now = datetime.now(timezone.utc)
|
||||
|
||||
# --- 3. 标记当前任务为已完成 ---
|
||||
task.status = TASK_STATUS_COMPLETED
|
||||
task.completed_at = now
|
||||
await _create_task_log(
|
||||
db, task_id,
|
||||
action_type="complete",
|
||||
operator_id=request.operator_id,
|
||||
remark=request.remark or f"完成任务: {task.task_name}",
|
||||
)
|
||||
|
||||
# --- 4. 可选：创建下一步任务（转交） ---
|
||||
next_task = None
|
||||
if request.next_task_name and request.next_assignee_id:
|
||||
next_task = Task(
|
||||
product_id=task.product_id,
|
||||
parent_task_id=task.parent_task_id, # 与已完成任务同级
|
||||
task_name=request.next_task_name,
|
||||
assignee_id=request.next_assignee_id,
|
||||
status=TASK_STATUS_PENDING,
|
||||
notify_parent_on_complete=False,
|
||||
)
|
||||
db.add(next_task)
|
||||
await db.flush()
|
||||
await _create_task_log(
|
||||
db, next_task.id,
|
||||
action_type="create",
|
||||
operator_id=request.operator_id,
|
||||
remark=f"由任务「{task.task_name}」完成后转交创建",
|
||||
)
|
||||
|
||||
await db.commit()
|
||||
|
||||
# --- 5. 构建响应 ---
|
||||
refreshed_task = await _get_task_with_children_recursive(db, task_id)
|
||||
next_task_response = None
|
||||
if next_task:
|
||||
await db.refresh(next_task)
|
||||
next_task_response = _to_response(next_task)
|
||||
|
||||
return TaskCompleteResponse(
|
||||
completed_task=_to_response(refreshed_task),
|
||||
next_task=next_task_response,
|
||||
message=f"任务「{task.task_name}」已完成"
|
||||
+ (f"，已创建下一步任务「{request.next_task_name}」" if next_task else ""),
|
||||
)
|
||||
|
||||
|
||||
# ============================================================
|
||||
# 子任务
|
||||
# ============================================================
|
||||
|
||||
async def create_subtask(
|
||||
db: AsyncSession, parent_task_id: uuid.UUID, data: SubtaskCreate
|
||||
) -> TaskResponse:
|
||||
"""
|
||||
在现有任务下创建子任务 — 支持无限层级嵌套。
|
||||
新子任务继承父任务的 product_id。
|
||||
"""
|
||||
parent = await _get_task_or_404(db, parent_task_id)
|
||||
|
||||
# 禁止在已完成/已驳回的任务下创建子任务
|
||||
if parent.status in (TASK_STATUS_COMPLETED, TASK_STATUS_REJECTED):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail=f"无法在状态为 {parent.status} 的任务「{parent.task_name}」下创建子任务",
|
||||
)
|
||||
|
||||
subtask = Task(
|
||||
product_id=parent.product_id,
|
||||
parent_task_id=parent_task_id,
|
||||
task_name=data.task_name,
|
||||
assignee_id=data.assignee_id,
|
||||
status=TASK_STATUS_PENDING,
|
||||
notify_parent_on_complete=data.notify_parent_on_complete,
|
||||
)
|
||||
db.add(subtask)
|
||||
await db.commit()
|
||||
await db.refresh(subtask)
|
||||
|
||||
await _create_task_log(
|
||||
db, subtask.id,
|
||||
action_type="create",
|
||||
operator_id=data.assignee_id,
|
||||
remark=f"创建子任务「{data.task_name}」，父任务: 「{parent.task_name}」",
|
||||
)
|
||||
await db.commit()
|
||||
|
||||
return _to_response(subtask)
|
||||
Reference in New Issue
Block a user