Files
track/backend/app/services/product_service.py

310 lines
11 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, TaskRecordResponse
def _task_to_response(task: Task) -> TaskResponse:
"""将 Task ORM 对象递归转为 TaskResponse含子任务树"""
product_sn = ""
product_material = ""
try:
if task.product:
product_sn = task.product.serial_number or ""
product_material = (task.product.material_name or task.product.material_id or "")
except Exception:
pass
return TaskResponse(
id=task.id,
product_id=task.product_id,
product_sn=product_sn,
product_material=product_material,
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,
remark=task.remark,
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],
records=[TaskRecordResponse.model_validate(r) for r in (task.records or [])],
)
async def _load_task_tree(db: AsyncSession, product_id: uuid.UUID) -> list[TaskResponse]:
"""递归加载产品下的完整任务树"""
# 先取顶层任务
result = await db.execute(
select(Task)
.options(selectinload(Task.child_tasks), selectinload(Task.records))
.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), selectinload(Task.records))
.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,
external_serial=product.external_serial,
order_id=product.order_id,
order_no=product.order.order_no if product.order else "",
material_id=product.material_id,
material_name=product.material_name,
spec_model=product.spec_model,
category=product.category,
material_type=product.material_type,
parent_product_id=product.parent_product_id,
current_location_id=product.current_location_id,
overall_status=product.overall_status,
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:
"""创建产品 — 自动生成 16 位 HEX 序列号"""
from app.services.counter_service import ensure_sequence, next_hex_id
from app.models.production_order import ProductionOrder
await ensure_sequence(db)
hex_id = await next_hex_id(db)
# 处理订单: 如果传了 order_no 但没传 order_id查找或创建
order_id = data.order_id
if not order_id and data.order_no:
result = await db.execute(
select(ProductionOrder).where(ProductionOrder.order_no == data.order_no.strip())
)
existing = result.scalar_one_or_none()
if existing:
order_id = existing.id
else:
new_order = ProductionOrder(order_no=data.order_no.strip())
db.add(new_order)
await db.flush()
order_id = new_order.id
product = Product(
serial_number=hex_id,
order_id=order_id,
material_id=data.material_id,
material_name=data.material_name or None,
spec_model=data.spec_model or None,
category=data.category or None,
material_type=data.material_type or None,
external_serial=data.external_serial,
parent_product_id=data.parent_product_id,
)
db.add(product)
await db.commit()
await db.refresh(product, ["order"])
return ProductResponse(
id=product.id,
serial_number=product.serial_number,
external_serial=product.external_serial,
order_id=product.order_id,
order_no=product.order.order_no if product.order else (data.order_no or ""),
material_id=product.material_id,
material_name=product.material_name,
spec_model=product.spec_model,
category=product.category,
material_type=product.material_type,
parent_product_id=product.parent_product_id,
current_location_id=product.current_location_id,
overall_status=product.overall_status,
status=product.status,
created_at=product.created_at,
)
async def update_product(db: AsyncSession, product_id: uuid.UUID, data: ProductUpdate) -> ProductResponse:
"""更新产品"""
from app.models.production_order import ProductionOrder
product = await get_product(db, product_id)
update_data = data.model_dump(exclude_unset=True)
# 处理 order_no → order_id 映射
if "order_no" in update_data:
order_no_val = update_data.pop("order_no")
if order_no_val and order_no_val.strip():
result = await db.execute(
select(ProductionOrder).where(ProductionOrder.order_no == order_no_val.strip())
)
existing = result.scalar_one_or_none()
if existing:
product.order_id = existing.id
else:
new_order = ProductionOrder(order_no=order_no_val.strip())
db.add(new_order)
await db.flush()
product.order_id = new_order.id
else:
product.order_id = None
for field, value in update_data.items():
setattr(product, field, value)
await db.commit()
await db.refresh(product, ["order"])
return ProductResponse(
id=product.id,
serial_number=product.serial_number,
external_serial=product.external_serial,
order_id=product.order_id,
order_no=product.order.order_no if product.order else "",
material_id=product.material_id,
material_name=product.material_name,
spec_model=product.spec_model,
category=product.category,
material_type=product.material_type,
parent_product_id=product.parent_product_id,
current_location_id=product.current_location_id,
overall_status=product.overall_status,
status=product.status,
created_at=product.created_at,
)
VALID_OVERALL_STATUS = {"备货", "生产", "测试", "维修", "在库"}
async def update_overall_status(db: AsyncSession, serial_number: str, status_value: str) -> ProductScanResponse:
"""更新产品宏观状态"""
if status_value not in VALID_OVERALL_STATUS:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"无效状态: {status_value},合法值: {', '.join(sorted(VALID_OVERALL_STATUS))}",
)
result = await db.execute(
select(Product)
.options(selectinload(Product.order))
.where(Product.serial_number == serial_number)
)
product = result.scalar_one_or_none()
if not product:
raise HTTPException(status_code=404, detail=f"未找到序列号 {serial_number} 的产品")
product.overall_status = status_value
await db.commit()
await db.refresh(product)
return await get_product_by_serial(db, serial_number)
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(
id=p.id,
serial_number=p.serial_number,
external_serial=p.external_serial,
order_id=p.order_id,
order_no=p.order.order_no if p.order else "",
material_id=p.material_id,
material_name=p.material_name,
spec_model=p.spec_model,
category=p.category,
material_type=p.material_type,
parent_product_id=p.parent_product_id,
current_location_id=p.current_location_id,
overall_status=p.overall_status,
status=p.status,
created_at=p.created_at,
)
for p in products
]