Files
track/backend/app/services/product_service.py
duxingchen 8028be58c4 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(): 递归加载产品关联的所有任务嵌套关系
2026-08-04 17:03:37 +08:00

162 lines
5.4 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""产品服务 — 业务逻辑层扫码查询、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]