"""产品服务 — 业务逻辑层：扫码查询、CRUD""" from __future__ import annotations import uuid from fastapi import HTTPException, status from sqlalchemy import select, or_, cast, String, delete, update from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.core.lifecycle import ( ALL_OVERALL_STEPS, allowed_steps, is_step_allowed, phase_label, resolve_phase_for_step, ) 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 [])], created_by=getattr(task, "created_by", None), ) async def _load_task_tree(db: AsyncSession, product_id: uuid.UUID) -> list[TaskResponse]: """使用 PostgreSQL Recursive CTE 一次性加载产品下完整任务树（消除 N+1）""" from app.services.task_tree_loader import load_task_trees_by_product tasks = await load_task_trees_by_product(db, product_id) return [_task_to_response(t) for t in 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) # 🔧 收集任务树中所有 assignee_id → 查中文姓名映射 assignee_ids: set[str] = set() all_task_ids: list[uuid.UUID] = [] def _collect_ids(tasks): for t in tasks: all_task_ids.append(t.id) if t.assignee_id: assignee_ids.add(t.assignee_id) if t.child_tasks: _collect_ids(t.child_tasks) for t in top_tasks: all_task_ids.append(t.id) if t.assignee_id: assignee_ids.add(t.assignee_id) _collect_ids(task_tree) # 🔧 批量查 task_logs: 谁创建了每个任务 creator_map: dict[uuid.UUID, str] = {} if all_task_ids: from app.models.task_log import TaskLog from sqlalchemy import func # 每个 task 取最早的 create 日志的 operator_id sub = ( select( TaskLog.task_id, TaskLog.operator_id, func.row_number().over( partition_by=TaskLog.task_id, order_by=TaskLog.created_at.asc() ).label("rn") ) .where( TaskLog.task_id.in_(all_task_ids), TaskLog.action_type == "create" ) ).subquery() log_result = await db.execute( select(sub.c.task_id, sub.c.operator_id).where(sub.c.rn == 1) ) for row in log_result: if row[1]: creator_map[row[0]] = row[1] # 注入 created_by 到 task_tree 和 top_tasks（并收集 created_by 到中文名映射） def _inject_creator(tasks): for t in tasks: t.created_by = creator_map.get(t.id) if t.created_by: assignee_ids.add(t.created_by) if t.child_tasks: _inject_creator(t.child_tasks) _inject_creator(task_tree) for t in top_tasks: t.created_by = creator_map.get(t.id) if t.created_by: assignee_ids.add(t.created_by) # 🔧 兜底：在库/入库任务无创建日志时，用该产品在库前最近一道主工序的负责人作为「转入人」 all_mains: list = [] def _collect_main(tasks): for t in tasks: if not t.parent_task_id or t.task_type in ("TRANSFER", "RECOVERY", "WAREHOUSE"): all_mains.append(t) if t.child_tasks: _collect_main(t.child_tasks) _collect_main(task_tree) all_mains.sort(key=lambda t: t.created_at) for t in all_mains: is_w = t.task_name and ("在库" in t.task_name or "入库" in t.task_name) if is_w and not t.created_by: prev = [m for m in all_mains if m.created_at < t.created_at and m.assignee_id] if prev: 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): return True if t.child_tasks and _has_warehouse_task(t.child_tasks): return True return False 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 = "ARCHIVED" 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 = "COMPLETED" 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=node_name, assignee_id=node_assignee, status=node_status, notify_parent_on_complete=False, is_rework=False, task_type=None, remark=None, reject_reason=None, received_at=last_main.received_at or last_main.created_at, completed_at=last_main.completed_at, created_at=last_main.created_at, child_tasks=[], records=[], 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) # 🔧 中文名映射（负责人 + 创建人，供前端显示"谁转入在库"等） assignee_names = _lookup_display_names(list(assignee_ids)) 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, lifecycle_phase=product.lifecycle_phase, created_at=product.created_at, top_level_tasks=[ TaskSummaryResponse.model_validate(t) for t in top_tasks ], task_tree=task_tree, assignee_names=assignee_names, # 🔧 username→中文姓名 ) 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, lifecycle_phase=product.lifecycle_phase, 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, lifecycle_phase=product.lifecycle_phase, created_at=product.created_at, ) # 全部合法宏观状态（两阶段并集）— 仅作"完全非法取值"的第一道粗筛； # 阶段内的细分校验（售后回流设备禁止改回「备货 / 生产」）见下方 is_step_allowed VALID_OVERALL_STATUS = ALL_OVERALL_STEPS async def update_overall_status( db: AsyncSession, serial_number: str, status_value: str, current_user: dict | None = None, ) -> ProductScanResponse: """更新产品宏观状态 权限校验： - SUPER_ADMIN 角色：直接放行 - 当前操作该产品主线任务（WIP/PENDING 状态主干任务）的人：放行 - 其他：403 """ if status_value not in VALID_OVERALL_STATUS: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"无效状态: {status_value}，合法值: {'、'.join(sorted(ALL_OVERALL_STEPS))}", ) 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} 的产品") # ── 权限校验（无 current_user 一律拒绝，杜绝空 dict 绕过）── if not current_user: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="请先登录", ) user_role = current_user.get("role", "") user_username = current_user.get("username", "") # SUPER_ADMIN 直接放行 if user_role != "SUPER_ADMIN": # 检查当前用户是否是该产品主线任务的负责人 main_task_result = await db.execute( select(Task).where( Task.product_id == product.id, Task.status.in_(["WIP", "PENDING"]), or_( Task.parent_task_id.is_(None), Task.task_type.in_(["TRANSFER", "RECOVERY", "WAREHOUSE"]), ), ).order_by(Task.created_at.desc()).limit(1) ) main_task = main_task_result.scalar_one_or_none() has_permission = ( main_task is not None and main_task.assignee_id == user_username ) if not has_permission: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="只有 SUPER_ADMIN 或当前操作该产品主线任务的人才能修改宏观状态", ) # ── 选项隔离：阶段感知校验 ── # 售后回流设备禁止被改回「备货 / 生产」；选定售后专属工序（发货测试 / 售后维修） # 则设备随即进入售后生命周期（无历史记录的老设备由此进入售后阶段）。 phase = resolve_phase_for_step(product.lifecycle_phase, status_value) if not is_step_allowed(phase, status_value): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=( f"产品当前处于{phase_label(product.lifecycle_phase)}，不允许改为「{status_value}」。" f"该阶段可选：{'、'.join(allowed_steps(product.lifecycle_phase))}" ), ) product.lifecycle_phase = phase 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) return await get_product_by_serial(db, serial_number) def _lookup_display_names(location_ids: list[str]) -> dict[str, str]: """批量查询 MOM sys_user，将 username 映射为真实姓名（带 2h TTL 缓存）""" from app.services.mom_cache import get_display_names return get_display_names(location_ids) 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.external_serial.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() # 状态筛选 — 大小写不敏感 # 口径与 macro_status 完全一致（废弃 Product.status 的恒值判断），业务状态定义： # COMPLETED = 车间完工待实收 → overall_status == '待仓库收货' # ARCHIVED = 仓库已实收 → overall_status == '已入库'（'在库' 为旧命名，等价） if status_filter and status_filter.strip(): 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_(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": stmt = stmt.where(not_finished, _has_task_status("PENDING")) elif sf == "PENDING_ASSIGNED": # 存在已分配(等待扫码)的待接收任务 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(not_finished, _has_task_status(sf)) stmt = stmt.offset(skip).limit(limit).order_by(Product.created_at.desc()) result = await db.execute(stmt) products = result.scalars().all() # 🔧 批量预计算 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 == "REJECTED", 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) # 🔧 每个产品最后一条任务（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: from sqlalchemy import and_, func as sa_func, case as sa_case main_where = and_( Task.product_id.in_(product_ids), or_( Task.parent_task_id.is_(None), Task.task_type.in_(["TRANSFER", "RECOVERY", "WAREHOUSE"]), ), ) prio_expr = sa_case( (Task.status == "WIP", 3), (Task.status == "PENDING", 2), (Task.status == "COMPLETED", 1), else_=0, ) max_prio = ( select(Task.product_id, sa_func.max(prio_expr).label("prio")) .where(main_where) .group_by(Task.product_id) ).subquery("mp") main_stmt = ( select(Task.product_id, Task.task_name) .join(max_prio, and_( Task.product_id == max_prio.c.product_id, prio_expr == max_prio.c.prio, )) .where(main_where) .order_by(Task.product_id, Task.created_at.desc()) .distinct(Task.product_id) ) main_result = await db.execute(main_stmt) for row in main_result: overall_names[row[0]] = row[1] # 🔧 当前位置：汇总所有活跃任务（WIP/PENDING，不分主线/分支）的负责人，去重保序 active_assignees_map: dict[uuid.UUID, list[str]] = {} if product_ids: active_stmt = ( select(Task.product_id, Task.assignee_id) .where( Task.product_id.in_(product_ids), Task.status.in_(["WIP", "PENDING"]), Task.assignee_id.isnot(None), ) .order_by(Task.product_id, Task.created_at) ) active_result = await db.execute(active_stmt) for row in active_result: pid, assignee = row[0], row[1] lst = active_assignees_map.setdefault(pid, []) if assignee not in lst: lst.append(assignee) # 🔧 活跃负责人 + 静态位置（兜底）的 username → 中文姓名（一次批量查） all_active_ids = [uid for ids in active_assignees_map.values() for uid in ids] static_location_ids = [p.current_location_id for p in products if p.current_location_id] merged_location_ids = list(set(all_active_ids + static_location_ids)) merged_name_map = _lookup_display_names(merged_location_ids) # 🔧 批量查询每个产品活跃任务的最新记录（含操作人 assignee_id） latest_record_map: dict[uuid.UUID, tuple] = {} if product_ids: from app.models.task import TaskRecord as TR wip_pending_ids = select(Task.id).where( and_( Task.product_id.in_(product_ids), Task.status.in_(["WIP", "PENDING"]), ) ).subquery() ranked = ( select(TR.task_id, TR.remark, TR.images, TR.created_at, Task.product_id, Task.assignee_id, sa_func.row_number().over( partition_by=Task.product_id, order_by=TR.created_at.desc() ).label("rn")) .join(Task, TR.task_id == Task.id) .where(Task.id.in_(select(wip_pending_ids.c.id))) ).subquery() rec_result = await db.execute( select(ranked.c.product_id, ranked.c.created_at, ranked.c.remark, ranked.c.images, ranked.c.assignee_id) .where(ranked.c.rn == 1) ) for row in rec_result: has_img = bool(row[3] and row[3] != "[]" and row[3] != "null") latest_record_map[row[0]] = (row[1], row[2], has_img, row[4]) # 🔧 当前人滞留时长：每个产品活跃任务(WIP/PENDING)最早接手时间 → 小时（排除非工作日） active_duration_map: dict[uuid.UUID, float] = {} if product_ids: from sqlalchemy import func as sa_func from app.core.time_utils import get_beijing_time, to_beijing, working_duration_hours from app.models.holiday import Holiday hres = await db.execute(select(Holiday.day)) holidays = {r[0] for r in hres} start_stmt = ( select( Task.product_id, sa_func.min(sa_func.coalesce(Task.received_at, Task.created_at)), ) .where( Task.product_id.in_(product_ids), Task.status.in_(["WIP", "PENDING"]), ) .group_by(Task.product_id) ) start_result = await db.execute(start_stmt) now = get_beijing_time() for row in start_result: start = row[1] if start is None: continue start_bj = to_beijing(start) # 🚀 naive 按 UTC 转北京时间（修复多算8小时） active_duration_map[row[0]] = working_duration_hours(start_bj, now, holidays) # 🔧 生产总天数（自然天 + 工作日）：自创建至今 import math 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 = _gbt() def _prod_days(created_at): created = _tb(created_at) if not created: return 1, 1 natural = max(1, math.ceil((now2 - created).total_seconds() / 86400)) work_hours = _wdh(created, now2, holidays2) workdays = max(1, math.ceil(work_hours / 24)) return natural, workdays production_days_map: dict = {} 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, 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=( ",".join(active_assignees_map.get(p.id, [])) or p.current_location_id ), current_location_name=( ", ".join(merged_name_map.get(uid, uid) for uid in active_assignees_map.get(p.id, [])) if active_assignees_map.get(p.id) 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=_resolve_macro_status(p), overall_status=overall_names.get(p.id) or p.overall_status, status=p.status, lifecycle_phase=p.lifecycle_phase, created_at=p.created_at, latest_record_time=latest_record_map.get(p.id, (None, None, False, None))[0], latest_record_content=latest_record_map.get(p.id, (None, None, False, None))[1], latest_record_has_images=latest_record_map.get(p.id, (None, None, False, None))[2], latest_record_assignee_id=latest_record_map.get(p.id, (None, None, False, None))[3], latest_record_assignee_name=( merged_name_map.get(latest_record_map.get(p.id, (None, None, False, None))[3]) if latest_record_map.get(p.id, (None, None, False, None))[3] else None ), active_duration_hours=active_duration_map.get(p.id), production_days=production_days_map.get(p.id, (1, 1))[0], production_days_workdays=production_days_map.get(p.id, (1, 1))[1], ) 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 from app.models.task_log import TaskLog # 🚀 1. 切断产品自引用：子产品的 parent_product_id 置空 await db.execute( update(Product).where(Product.parent_product_id == product_id).values(parent_product_id=None) ) # 2. 查询所有关联任务 tasks_result = await db.execute( select(Task).where(Task.product_id == product_id) ) tasks = tasks_result.scalars().all() # 🚀 3. 切断任务自引用：子任务的 parent_task_id 置空 for task in tasks: await db.execute( update(Task).where(Task.parent_task_id == task.id).values(parent_task_id=None) ) # 4. 删除任务记录、日志、任务本身 for task in tasks: await db.execute(delete(TaskRecord).where(TaskRecord.task_id == task.id)) await db.execute(delete(TaskLog).where(TaskLog.task_id == task.id)) await db.delete(task) # 5. 删除产品（product_messages 有 ON DELETE CASCADE 自动级联） await db.delete(product) await db.commit()