"""产品服务 — 业务逻辑层：扫码查询、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.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]: """使用 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() def _collect_ids(tasks): for t in tasks: if t.assignee_id: assignee_ids.add(t.assignee_id) if t.child_tasks: _collect_ids(t.child_tasks) for t in top_tasks: if t.assignee_id: assignee_ids.add(t.assignee_id) _collect_ids(task_tree) 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, 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, 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, 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(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} 的产品") # ── 权限校验（无 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"]), ), ).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 或当前操作该产品主线任务的人才能修改宏观状态", ) 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 映射为真实姓名（带 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.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() # 🔧 批量预计算 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) # 🔧 动态主干状态+位置：只从主干任务中获取最高优先级任务的 task_name + assignee_id overall_names: dict[uuid.UUID, str] = {} main_assignees: 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"]), ), ) 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") # JOIN 回 tasks 拿 task_name + assignee_id（同优先级取最新创建的） main_stmt = ( select(Task.product_id, Task.task_name, Task.assignee_id) .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: pid, tname, assignee = row[0], row[1], row[2] overall_names[pid] = tname if assignee: main_assignees[pid] = assignee # 🔧 动态主干的 assignee_id → 查中文姓名 dynamic_location_ids = list(main_assignees.values()) dynamic_name_map = _lookup_display_names(dynamic_location_ids) # 🔧 合并：静态位置姓名（兜底）+ 动态主干位置姓名（优先） static_location_ids = [p.current_location_id for p in products if p.current_location_id] merged_location_ids = list(set(static_location_ids + dynamic_location_ids)) merged_name_map = _lookup_display_names(merged_location_ids) # 🔧 批量查询每个产品活跃任务的最新记录 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, 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) .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) 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=( main_assignees.get(p.id) or p.current_location_id ), current_location_name=( "仓库" if (main_assignees.get(p.id) or p.current_location_id) == "virtual_warehouse" else dynamic_name_map.get(main_assignees.get(p.id, "")) or 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, overall_status=overall_names.get(p.id) or p.overall_status, status=p.status, created_at=p.created_at, latest_record_time=latest_record_map.get(p.id, (None, None, False))[0], latest_record_content=latest_record_map.get(p.id, (None, None, False))[1], latest_record_has_images=latest_record_map.get(p.id, (None, None, False))[2], ) 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()