feat: Track 打通已完成/已入库/已出库状态闭环与外部联动

- 完工转交入库同步 status=COMPLETED;MOM 入库/出库回调同步 status
- 新增 mom-outbound webhook(发货出库标记已出库),lookup 返回 material_id
- VALID_OVERALL_STATUS 新增已出库;update_overall_status 同步 status 字段
- WIP 矩阵区分已入库/待仓库收货;虚拟节点正确处理已出库/转入在库人

backend/app/services/product_service.py

@@ -154,7 +154,7 @@ async def get_product_by_serial(db: AsyncSession, serial_number: str) -> Product
                 t.created_by = prev[-1].assignee_id
                 assignee_ids.add(t.created_by)
 
-    # 🔧 在库设备若无「在库」任务，追加虚拟「在库」节点（展示谁转入在库）
+    # 🔧 在库设备若无「在库」任务，追加虚拟节点（区分"待收货"与"已实收"）
     def _has_warehouse_task(tasks):
         for t in tasks:
             if t.task_name and ("在库" in t.task_name or "入库" in t.task_name):
@@ -165,16 +165,83 @@ async def get_product_by_serial(db: AsyncSession, serial_number: str) -> Product
     has_warehouse_task = _has_warehouse_task(task_tree)
     if product.current_location_id == "virtual_warehouse" and not has_warehouse_task and all_mains:
         from app.schemas.task import TaskResponse
+        from app.models.task_log import TaskLog as _TL
+        # 🔧 反查 webhook 入库接收日志：判断 MOM 是否已扫码实收
+        inbound_log = (
+            await db.execute(
+                select(_TL)
+                .join(Task, _TL.task_id == Task.id)
+                .where(
+                    Task.product_id == product.id,
+                    _TL.action_type == "warehouse_inbound",
+                )
+                .order_by(_TL.created_at.desc())
+                .limit(1)
+            )
+        ).scalars().first()
+
         last_main = all_mains[-1]
+
+        # 🔧 最后操作"转入库"的人 = 该产品最后一次 complete 日志的操作人
+        # (完工转交入库会记一条 complete 日志，operator 为发起转入库操作的人，
+        #  可能是最后工序负责人本人，也可能是代操作的主管)
+        transfer_log = (
+            await db.execute(
+                select(_TL)
+                .join(Task, _TL.task_id == Task.id)
+                .where(
+                    Task.product_id == product.id,
+                    _TL.action_type == "complete",
+                )
+                .order_by(_TL.created_at.desc())
+                .limit(1)
+            )
+        ).scalars().first()
+        last_transfer_operator = transfer_log.operator_id if transfer_log else None
+
+        # 🔧 反查出库日志：产品是否已被 MOM 发货出库
+        outbound_log = (
+            await db.execute(
+                select(_TL)
+                .join(Task, _TL.task_id == Task.id)
+                .where(
+                    Task.product_id == product.id,
+                    _TL.action_type == "warehouse_outbound",
+                )
+                .order_by(_TL.created_at.desc())
+                .limit(1)
+            )
+        ).scalars().first()
+
+        if product.overall_status == "已出库":
+            # 情况 C：MOM 已发货出库 → 虚拟节点反映"已出库"，负责人为出库操作人
+            node_name = "已出库"
+            node_status = "OUTBOUND"
+            node_assignee = (outbound_log.operator_id if outbound_log else None) or last_transfer_operator or last_main.assignee_id
+            node_created_by = last_transfer_operator or last_main.assignee_id
+        elif inbound_log is not None:
+            # 情况 B：MOM 已扫码实收 → "在库" 已完成，负责人为仓库接收人
+            node_name = "在库"
+            node_status = "COMPLETED"
+            node_assignee = inbound_log.operator_id or "仓库"
+            # "转入在库"始终显示操作转入库的人（车间），而不是 MOM 接收人
+            node_created_by = last_transfer_operator or last_main.assignee_id
+        else:
+            # 情况 A：MOM 还没扫码 → "待收货" 进行中，负责人为占位"待仓库扫码"
+            node_name = "待收货"
+            node_status = "IN_PROGRESS"
+            node_assignee = "待仓库扫码"
+            node_created_by = last_transfer_operator or last_main.assignee_id
+
         virtual = TaskResponse(
             id=uuid.uuid4(),
             product_id=product.id,
             product_sn=product.serial_number,
             product_material=product.material_name or product.material_id or "",
             parent_task_id=None,
-            task_name="在库",
-            assignee_id="virtual_warehouse",
-            status="COMPLETED",
+            task_name=node_name,
+            assignee_id=node_assignee,
+            status=node_status,
             notify_parent_on_complete=False,
             is_rework=False,
             task_type=None,
@@ -185,9 +252,12 @@ async def get_product_by_serial(db: AsyncSession, serial_number: str) -> Product
             created_at=last_main.created_at,
             child_tasks=[],
             records=[],
-            created_by=last_main.assignee_id,  # 最后一道主工序负责人（近似转入人）
+            created_by=node_created_by,
         )
         task_tree.append(virtual)
+        # 真实 username 负责人（含仓库接收人）加入中文名映射；占位符无需映射
+        if virtual.assignee_id and virtual.assignee_id not in ("待仓库扫码", "virtual_warehouse"):
+            assignee_ids.add(virtual.assignee_id)
         if virtual.created_by:
             assignee_ids.add(virtual.created_by)
 
@@ -347,7 +417,7 @@ async def update_product(db: AsyncSession, product_id: uuid.UUID, data: ProductU
     )
 
 
-VALID_OVERALL_STATUS = {"备货", "生产", "测试", "维修", "在库"}
+VALID_OVERALL_STATUS = {"备货", "生产", "测试", "维修", "在库", "待仓库收货", "已入库", "已出库"}
 
 
 async def update_overall_status(
@@ -411,6 +481,14 @@ async def update_overall_status(
             )
 
     product.overall_status = status_value
+    # 同步 status 字段，保证与整体状态口径一致（修复"只改整体状态不改 status"的旧缺陷）
+    _OVERALL_TO_STATUS = {
+        "已入库": "ARCHIVED",
+        "在库": "ARCHIVED",
+        "已出库": "OUTBOUND",
+        "待仓库收货": "COMPLETED",
+    }
+    product.status = _OVERALL_TO_STATUS.get(status_value, "WIP")
     await db.commit()
     await db.refresh(product)
 
@@ -451,36 +529,74 @@ async def get_all_products(
             )
         ).distinct()
 
-    # 状态筛选 — 大小写不敏感，支持组合过滤
+    # 状态筛选 — 大小写不敏感
+    # 口径与 macro_status 完全一致（废弃 Product.status 的恒值判断），业务状态定义：
+    #   COMPLETED = 车间完工待实收 → overall_status == '待仓库收货'
+    #   ARCHIVED  = 仓库已实收   → overall_status == '已入库'（'在库' 为旧命名，等价）
     if status_filter and status_filter.strip():
-        from sqlalchemy import func
+        from sqlalchemy import func, and_, exists
         sf = status_filter.strip().upper()
+
+        # 最后一条任务（created_at 最新）状态为 COMPLETED 的产品子查询（兼容旧数据用）
+        ranked = (
+            select(
+                Task.product_id, Task.status,
+                func.row_number().over(
+                    partition_by=Task.product_id,
+                    order_by=Task.created_at.desc(),
+                ).label("rn"),
+            ).subquery("sf_latest_task")
+        )
+        latest_completed_ids = select(ranked.c.product_id).where(
+            ranked.c.rn == 1, ranked.c.status == "COMPLETED",
+        )
+
+        archived_cond = Product.overall_status.in_(["已入库", "在库"])
+        completed_cond = or_(
+            Product.overall_status == "待仓库收货",
+            # 兼容旧数据：无定位（current_location_id IS NULL）且最后一条任务已完成
+            and_(
+                Product.current_location_id.is_(None),
+                Product.id.in_(latest_completed_ids),
+            ),
+        )
+        # 未进入"完结"态（NULL 视为未完结，避免三值逻辑误过滤）
+        not_finished = or_(
+            Product.overall_status.is_(None),
+            ~Product.overall_status.in_(["待仓库收货", "已入库", "在库"]),
+        )
+
+        def _has_task_status(task_status: str):
+            """存在指定状态任务 且 未完结的 EXISTS 谓词"""
+            return exists(
+                select(Task.id).where(Task.product_id == Product.id, Task.status == task_status)
+            )
+
         if sf == "DONE":
-            # "已完成" 匹配 COMPLETED 或 ARCHIVED
-            stmt = stmt.where(
-                or_(
-                    func.upper(Product.status) == "COMPLETED",
-                    func.upper(Product.status) == "ARCHIVED",
-                )
-            )
+            # "已完成/已入库" = 待仓库收货(COMPLETED) 或 已入库(ARCHIVED)
+            stmt = stmt.where(or_(archived_cond, completed_cond))
+        elif sf == "ARCHIVED":
+            # 已入库 → ARCHIVED
+            stmt = stmt.where(archived_cond)
+        elif sf == "COMPLETED":
+            # 待仓库收货 → COMPLETED（含旧的无定位已完成数据）
+            stmt = stmt.where(completed_cond)
+        elif sf == "WIP":
+            stmt = stmt.where(not_finished, _has_task_status("WIP"))
         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()
-            )
+            stmt = stmt.where(not_finished, _has_task_status("PENDING"))
         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()
-            )
+            # 存在已分配(等待扫码)的待接收任务
+            stmt = stmt.where(not_finished, exists(
+                select(Task.id).where(
+                    Task.product_id == Product.id,
+                    Task.status == "PENDING",
+                    Task.assignee_id.isnot(None),
+                )
+            ))
         else:
-            stmt = stmt.where(func.upper(Product.status) == sf)
+            # 兜底：其他状态码按"存在该状态任务"匹配（未完结）
+            stmt = stmt.where(not_finished, _has_task_status(sf))
 
     stmt = stmt.offset(skip).limit(limit).order_by(Product.created_at.desc())
 
@@ -512,6 +628,27 @@ async def get_all_products(
         for row in task_result:
             macro_map[row[0]] = prio_to_status.get(row[1], None)
 
+    # 🔧 每个产品最后一条任务（created_at 最新）的状态 → 用于"已完成"判定
+    last_task_status_map: dict[uuid.UUID, str] = {}
+    if product_ids:
+        from sqlalchemy import func as sa_func
+        ranked = (
+            select(
+                Task.product_id, Task.status,
+                sa_func.row_number().over(
+                    partition_by=Task.product_id,
+                    order_by=Task.created_at.desc(),
+                ).label("rn"),
+            )
+            .where(Task.product_id.in_(product_ids))
+            .subquery("last_task")
+        )
+        last_result = await db.execute(
+            select(ranked.c.product_id, ranked.c.status).where(ranked.c.rn == 1)
+        )
+        for row in last_result:
+            last_task_status_map[row[0]] = row[1]
+
     # 🔧 动态主干状态名：只从主干任务中获取最高优先级任务的 task_name（宏观状态名）
     overall_names: dict[uuid.UUID, str] = {}
     if product_ids:
@@ -630,11 +767,11 @@ async def get_all_products(
 
     # 🔧 生产总天数（自然天 + 工作日）：自创建至今
     import math
-    from app.core.time_utils import to_beijing as _tb, working_duration_hours as _wdh
+    from app.core.time_utils import get_beijing_time as _gbt, to_beijing as _tb, working_duration_hours as _wdh
     from app.models.holiday import Holiday as _Holiday
     hres2 = await db.execute(select(_Holiday.day))
     holidays2 = {r[0] for r in hres2}
-    now2 = get_beijing_time()
+    now2 = _gbt()
 
     def _prod_days(created_at):
         created = _tb(created_at)
@@ -649,6 +786,23 @@ async def get_all_products(
     for _p in products:
         production_days_map[_p.id] = _prod_days(_p.created_at)
 
+    def _resolve_macro_status(p: Product) -> str:
+        """宏观状态 — 以 overall_status 为核心的状态定义（用户确认）：
+        - ARCHIVED(已入库): overall_status == '已入库'（'在库' 为旧命名，等价）
+        - COMPLETED(已完成): overall_status == '待仓库收货'；
+            兼容旧数据：无定位（current_location_id IS NULL）且最后一条任务 COMPLETED
+        - 其余: 沿用原 WIP/PENDING 任务优先级；无任何任务的产品 → PENDING
+        """
+        if p.overall_status in ("已入库", "在库"):
+            return "ARCHIVED"
+        if p.overall_status == "已出库":
+            return "OUTBOUND"
+        if p.overall_status == "待仓库收货":
+            return "COMPLETED"
+        if last_task_status_map.get(p.id) == "COMPLETED" and p.current_location_id is None:
+            return "COMPLETED"
+        return macro_map.get(p.id) or "PENDING"
+
     return [
         ProductResponse(
             id=p.id,
@@ -671,7 +825,7 @@ async def get_all_products(
                 else ("仓库" if p.current_location_id == "virtual_warehouse"
                       else merged_name_map.get(p.current_location_id) if p.current_location_id else None)
             ),
-            macro_status=macro_map.get(p.id) or p.status,
+            macro_status=_resolve_macro_status(p),
             overall_status=overall_names.get(p.id) or p.overall_status,
             status=p.status,
             created_at=p.created_at,