460 lines
17 KiB
Python
460 lines
17 KiB
Python
"""产品服务 — 业务逻辑层：扫码查询、CRUD"""
|
||
from __future__ import annotations
|
||
import uuid
|
||
from fastapi import HTTPException, status
|
||
from sqlalchemy import select, or_, cast, String
|
||
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,
|
||
task_type=task.task_type,
|
||
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, creator_username: str = "") -> 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,
|
||
current_location_id=creator_username or None, # 谁创建，初始位置就是谁
|
||
)
|
||
db.add(product)
|
||
await db.commit()
|
||
await db.refresh(product, ["order"])
|
||
|
||
# 查创建者的真实姓名
|
||
creator_display_name = ""
|
||
if creator_username:
|
||
name_map = _lookup_display_names([creator_username])
|
||
creator_display_name = name_map.get(creator_username, "")
|
||
|
||
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,
|
||
current_location_name=creator_display_name or None,
|
||
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)
|
||
|
||
|
||
def _lookup_display_names(location_ids: list[str]) -> dict[str, str]:
|
||
"""批量查询 MOM sys_user，将 username 映射为真实姓名"""
|
||
if not location_ids:
|
||
return {}
|
||
from app.core.mom_database import MomSessionLocal
|
||
from sqlalchemy import text
|
||
db = MomSessionLocal()
|
||
try:
|
||
# 过滤掉特殊值
|
||
real_ids = [uid for uid in location_ids if uid and uid != "virtual_warehouse"]
|
||
if not real_ids:
|
||
return {}
|
||
# 用 LIKE 模糊匹配批量查出
|
||
conditions = " OR ".join([f"username LIKE '%/{uid}'" for uid in real_ids])
|
||
result = db.execute(
|
||
text(f"SELECT username, SPLIT_PART(username, '/', 1) as display_name FROM sys_user WHERE {conditions}")
|
||
)
|
||
mapping = {}
|
||
for row in result:
|
||
full_username = row[0]
|
||
display_name = row[1]
|
||
# 从 full_username 末尾提取短用户名: "张三/zhangsan01" → "zhangsan01"
|
||
short = full_username.split("/")[-1] if "/" in full_username else full_username
|
||
mapping[short] = display_name
|
||
return mapping
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
async def get_all_products(
|
||
db: AsyncSession,
|
||
skip: int = 0,
|
||
limit: int = 50,
|
||
keyword: str | None = None,
|
||
status_filter: str | None = None,
|
||
) -> list[ProductResponse]:
|
||
"""
|
||
获取产品列表 — 支持多维 keyword 搜索 + 状态筛选
|
||
|
||
keyword: 同时模糊匹配 serial_number (产品身份证)、material_name/id (规格型号)、order_no (订单号)
|
||
status_filter: 按产品状态过滤 (如 PENDING / WIP / COMPLETED / ARCHIVED)
|
||
"""
|
||
stmt = select(Product).options(selectinload(Product.order))
|
||
|
||
# keyword 多字段 OR 模糊搜索
|
||
if keyword and keyword.strip():
|
||
kw = f"%{keyword.strip()}%"
|
||
stmt = stmt.outerjoin(ProductionOrder, Product.order_id == ProductionOrder.id).where(
|
||
or_(
|
||
Product.serial_number.ilike(kw),
|
||
Product.material_name.ilike(kw),
|
||
cast(Product.material_id, String).ilike(kw),
|
||
Product.spec_model.ilike(kw),
|
||
ProductionOrder.order_no.ilike(kw),
|
||
)
|
||
).distinct()
|
||
|
||
# 状态筛选 — 大小写不敏感，支持组合过滤
|
||
if status_filter and status_filter.strip():
|
||
from sqlalchemy import func
|
||
sf = status_filter.strip().upper()
|
||
if sf == "DONE":
|
||
# "已完成" 匹配 COMPLETED 或 ARCHIVED
|
||
stmt = stmt.where(
|
||
or_(
|
||
func.upper(Product.status) == "COMPLETED",
|
||
func.upper(Product.status) == "ARCHIVED",
|
||
)
|
||
)
|
||
elif sf == "PENDING":
|
||
# "待流转" — 产品状态 PENDING 且所有顶层任务均未分配人
|
||
stmt = (
|
||
stmt.outerjoin(Task, Task.product_id == Product.id)
|
||
.where(func.upper(Product.status) == "PENDING")
|
||
.where(Task.assignee_id.is_(None))
|
||
.distinct()
|
||
)
|
||
elif sf == "PENDING_ASSIGNED":
|
||
# "待接收" — 产品状态 PENDING 但已有任务被分配（等待工人扫码）
|
||
stmt = (
|
||
stmt.outerjoin(Task, Task.product_id == Product.id)
|
||
.where(func.upper(Product.status) == "PENDING")
|
||
.where(Task.assignee_id.isnot(None))
|
||
.distinct()
|
||
)
|
||
else:
|
||
stmt = stmt.where(func.upper(Product.status) == sf)
|
||
|
||
stmt = stmt.offset(skip).limit(limit).order_by(Product.created_at.desc())
|
||
|
||
result = await db.execute(stmt)
|
||
products = result.scalars().all()
|
||
|
||
# 批量查询当前位置对应的真实姓名
|
||
location_ids = [p.current_location_id for p in products if p.current_location_id]
|
||
name_map = _lookup_display_names(location_ids)
|
||
|
||
# 🔧 批量预计算 macro_status：一次性查出所有产品关联的任务状态
|
||
product_ids = [p.id for p in products]
|
||
macro_map: dict[uuid.UUID, str] = {}
|
||
if product_ids:
|
||
from sqlalchemy import case, func as sa_func
|
||
task_stmt = (
|
||
select(
|
||
Task.product_id,
|
||
sa_func.max(case(
|
||
(Task.status == "WIP", 3),
|
||
(Task.status == "PENDING", 2),
|
||
(Task.status == "COMPLETED", 1),
|
||
(Task.status == "ARCHIVED", 1),
|
||
else_=0,
|
||
)).label("prio"),
|
||
)
|
||
.where(Task.product_id.in_(product_ids))
|
||
.group_by(Task.product_id)
|
||
)
|
||
task_result = await db.execute(task_stmt)
|
||
prio_to_status = {3: "WIP", 2: "PENDING", 1: "COMPLETED", 0: None}
|
||
for row in task_result:
|
||
macro_map[row[0]] = prio_to_status.get(row[1], None)
|
||
|
||
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,
|
||
current_location_name=(
|
||
"仓库" if p.current_location_id == "virtual_warehouse"
|
||
else name_map.get(p.current_location_id) if p.current_location_id
|
||
else None
|
||
),
|
||
macro_status=macro_map.get(p.id) or p.status, # 优先任务树状态，兜底产品状态
|
||
overall_status=p.overall_status,
|
||
status=p.status,
|
||
created_at=p.created_at,
|
||
)
|
||
for p in products
|
||
]
|
||
|
||
|
||
async def delete_product(db: AsyncSession, product_id: uuid.UUID) -> None:
|
||
"""删除产品及其关联任务"""
|
||
product = await get_product(db, product_id)
|
||
|
||
# 删除关联任务记录
|
||
from app.models.task import TaskRecord
|
||
tasks_result = await db.execute(
|
||
select(Task).where(Task.product_id == product_id)
|
||
)
|
||
tasks = tasks_result.scalars().all()
|
||
for task in tasks:
|
||
await db.execute(
|
||
select(TaskRecord).where(TaskRecord.task_id == task.id)
|
||
)
|
||
# 级联删除已在模型中定义，直接删任务
|
||
# 删除产品（task 有外键 CASCADE？检查模型）
|
||
# 手动删关联任务确保完整
|
||
for task in tasks:
|
||
await db.delete(task)
|
||
await db.delete(product)
|
||
await db.commit()
|