diff --git a/backend/app/api/v1/endpoints/products.py b/backend/app/api/v1/endpoints/products.py index 6ae6fb2..b75e2e4 100644 --- a/backend/app/api/v1/endpoints/products.py +++ b/backend/app/api/v1/endpoints/products.py @@ -14,7 +14,7 @@ from app.schemas.product import ( ProductResponse, ProductScanResponse, ) -from app.services import product_service +from app.services import product_service, product_finalize_service from app.services.auth_service import get_current_user from app.services.qrcode_service import generate_qrcode_png @@ -147,6 +147,42 @@ async def update_product_overall_status( ) +# ============================================================ +# 产品收口 — 管理员将产品修正为「已入库」/「已出库」（可反向纠错） +# ============================================================ + +class ProductFinalizeRequest(BaseModel): + """管理员收口入参""" + status: str = Field(..., min_length=1, max_length=20, description="收口目标: 已入库 | 已出库") + note: str | None = Field(None, max_length=200, description="备注(选填)") + + +@router.post("/scan/{serial_number}/finalize", response_model=ProductScanResponse) +async def finalize_product_status_endpoint( + serial_number: str, + data: ProductFinalizeRequest, + db: AsyncSession = Depends(get_db), + current_user: dict = Depends(get_current_user), +): + """管理员将产品整体收口为「已入库」或「已出库」（支持 入库<->出库 反向互切纠错）。 + + 与 MOM 出入库回调落库语义一致：同步 product.overall_status/status、写 + warehouse_inbound/outbound 日志、幂等追加「扫码入库/扫码出库」主线收尾节点。 + 权限：仅 SUPER_ADMIN / SUPERVISOR。 + """ + from fastapi import HTTPException, status + from app.services.task_service import ADMIN_ROLES + + if (current_user or {}).get("role") not in ADMIN_ROLES: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="仅超级管理员或主管可执行入库/出库收口", + ) + return await product_finalize_service.finalize_product_status( + db, serial_number, data.status, current_user, data.note, + ) + + # ============================================================ # 协同留言板 # ============================================================ diff --git a/backend/app/services/product_finalize_service.py b/backend/app/services/product_finalize_service.py new file mode 100644 index 0000000..d3344b4 --- /dev/null +++ b/backend/app/services/product_finalize_service.py @@ -0,0 +1,146 @@ +"""产品收口服务 — 管理员一键把产品修正为「已入库」或「已出库」（可反向互切纠错）。 + +落库语义与 MOM 仓储回调（webhooks.mom_inbound / mom_outbound）保持一致： + 1) 改 product.overall_status + product.status（映射为 ARCHIVED / OUTBOUND）； + 2) 写一条 warehouse_inbound / warehouse_outbound 的 task_log； + 3) 幂等追加「扫码入库 / 扫码出库」WAREHOUSE 主线节点 + TaskRecord。 + +与 update_overall_status（仅改状态、面向主线负责人）不同：本服务面向管理员做收口， +会补全流转树收尾节点，使任务全景/详情/矩阵能正确显示入库/出库收口。 +""" +from __future__ import annotations + +from fastapi import HTTPException +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.time_utils import get_beijing_time +from app.models.product import Product +from app.models.task import Task, TaskRecord +from app.models.task_log import TaskLog +from app.services.product_service import get_product_by_serial + +# target -> (product.status 映射码, 追加节点名, task_log.action_type) +_FINALIZE_MAP = { + "已入库": ("ARCHIVED", "扫码入库", "warehouse_inbound"), + "已出库": ("OUTBOUND", "扫码出库", "warehouse_outbound"), +} + + +async def _append_warehouse_node(db: AsyncSession, product: Product, task_name: str) -> None: + """在流转树主干末尾幂等追加仓储收口主线节点 + TaskRecord（对齐 webhooks._append_warehouse_task）""" + existing = ( + await db.execute( + select(Task.id).where( + Task.product_id == product.id, + Task.task_name == task_name, + ).limit(1) + ) + ).scalar_one_or_none() + if existing: + return + + last_main = ( + await db.execute( + select(Task) + .where( + Task.product_id == product.id, + (Task.parent_task_id.is_(None)) + | (Task.task_type.in_(["TRANSFER", "RECOVERY"])), + ) + .order_by(Task.created_at.desc()) + .limit(1) + ) + ).scalar_one_or_none() + + node = Task( + product_id=product.id, + parent_task_id=last_main.id if last_main else None, + task_name=task_name, + assignee_id=None, # ★ 不能填非 UUID，否则前端头像解析报错导致节点跳过渲染 + status="COMPLETED", + task_type="WAREHOUSE", + completed_at=get_beijing_time(), + remark=f"管理员收口：追加{task_name}收尾节点", + ) + db.add(node) + await db.flush() + db.add(TaskRecord(task_id=node.id, remark=f"管理员收口：追加{task_name}收尾节点")) + + +async def finalize_product_status( + db: AsyncSession, + serial_number: str, + target: str, + current_user: dict | None, + note: str | None = None, +): + """把产品整体收口到 target（已入库 / 已出库）。已处于目标态则幂等直接返回。""" + target = (target or "").strip() + if target not in _FINALIZE_MAP: + raise HTTPException(status_code=400, detail="收口目标状态仅支持：已入库 / 已出库") + + product = ( + await db.execute(select(Product).where(Product.serial_number == serial_number)) + ).scalar_one_or_none() + if product is None: + raise HTTPException(status_code=404, detail=f"未找到序列号为 {serial_number} 的产品") + + # 幂等：已在目标状态则不动（避免重复写日志/节点） + if product.overall_status == target: + return await get_product_by_serial(db, serial_number) + + # 管理员强收口：仍有在产/待接收任务的，一并结束为完工收口，避免宏观口径分裂 + res = await db.execute( + update(Task) + .where(Task.product_id == product.id, Task.status.in_(["WIP", "PENDING"])) + .values( + status="COMPLETED", + received_at=get_beijing_time(), + completed_at=get_beijing_time(), + ) + ) + ended_active = res.rowcount or 0 + + status_code, node_name, action_type = _FINALIZE_MAP[target] + product.overall_status = target + product.status = status_code + product.current_location_id = "virtual_warehouse" + + # task_log 绑目标任务：优先「在库」任务，其次该产品最新任务（对齐 webhook） + log_task = ( + await db.execute( + select(Task) + .where(Task.product_id == product.id, Task.task_name.ilike("%在库%")) + .order_by(Task.created_at.desc()) + .limit(1) + ) + ).scalar_one_or_none() + if log_task is None: + log_task = ( + await db.execute( + select(Task).where(Task.product_id == product.id) + .order_by(Task.created_at.desc()) + .limit(1) + ) + ).scalar_one_or_none() + + operator = (current_user or {}).get("username") or "virtual_warehouse" + remark = f"管理员[{operator}]收口为{target}" + if ended_active: + remark += f"；一并结束{ended_active}个进行中/待接收任务" + if note and note.strip(): + remark += f"；备注：{note.strip()}" + if log_task is not None: + db.add( + TaskLog( + task_id=log_task.id, + operator_id=str(operator)[:64], + action_type=action_type, + remark=remark, + ) + ) + + await _append_warehouse_node(db, product, node_name) + await db.commit() + return await get_product_by_serial(db, serial_number)