track/backend/app/services/product_service.py

"""产品服务 — 业务逻辑层：扫码查询、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()