zkxw/backend/app/crud/qc_task_assignee.py

"""质检任务负责人 CRUD"""
from __future__ import annotations

from datetime import datetime
from typing import Dict, List, Optional, Set

from sqlalchemy.orm import Session

from app.core.db_collation import eq_unicode
from app.core.utils import generate_uuid
from app.models.models import BizQcTaskAssignee, SysUser


def _active_users_in_tenant(
    db: Session, user_ids: List[str], tenant_no: str
) -> Set[str]:
    ids = [x.strip() for x in user_ids if (x or "").strip()]
    if not ids:
        return set()
    rows = (
        db.query(SysUser.id)
        .filter(
            SysUser.id.in_(ids),
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
        )
        .all()
    )
    return {r[0] for r in rows}


def validate_assignee_user_ids(
    db: Session, user_ids: List[str], tenant_no: str
) -> List[str]:
    """去重并校验用户有效，返回有序列表。"""
    seen: Set[str] = set()
    ordered: List[str] = []
    for uid in user_ids:
        u = (uid or "").strip()
        if not u or u in seen:
            continue
        seen.add(u)
        ordered.append(u)
    if not ordered:
        raise ValueError("请至少指定一名负责人")
    valid = _active_users_in_tenant(db, ordered, tenant_no)
    invalid = [u for u in ordered if u not in valid]
    if invalid:
        raise ValueError("存在无效或已停用的负责人用户")
    return ordered


def sync_assignees_for_task(
    db: Session,
    task_id: str,
    user_ids: List[str],
    *,
    tenant_no: str,
    operator_id: Optional[str] = None,
    create_dept_id: str = "",
) -> None:
    """全量替换任务负责人（软删旧记录 + 插入新记录）。"""
    normalized = validate_assignee_user_ids(db, user_ids, tenant_no)
    now = datetime.now()
    existing = (
        db.query(BizQcTaskAssignee)
        .filter(
            BizQcTaskAssignee.task_id == task_id,
            BizQcTaskAssignee.tenant_no == tenant_no,
            BizQcTaskAssignee.del_flag == 0,
        )
        .all()
    )
    keep_ids = set(normalized)
    for row in existing:
        if row.user_id not in keep_ids:
            row.del_flag = 1
            row.update_by = operator_id
            row.update_time = now
            db.add(row)
    existing_map = {r.user_id: r for r in existing if r.del_flag == 0}
    for idx, uid in enumerate(normalized):
        if uid in existing_map:
            row = existing_map[uid]
            row.order_num = idx + 1
            row.update_by = operator_id
            row.update_time = now
            db.add(row)
            continue
        db.add(
            BizQcTaskAssignee(
                id=generate_uuid(),
                task_id=task_id,
                user_id=uid,
                tenant_no=tenant_no,
                order_num=idx + 1,
                create_by=operator_id,
                create_dept_id=create_dept_id or "",
                create_time=now,
                update_by=operator_id,
                update_time=now,
                del_flag=0,
            )
        )


def soft_delete_assignees_by_task_ids(
    db: Session, task_ids: List[str], tenant_no: str
) -> None:
    if not task_ids:
        return
    now = datetime.now()
    rows = (
        db.query(BizQcTaskAssignee)
        .filter(
            BizQcTaskAssignee.task_id.in_(task_ids),
            BizQcTaskAssignee.tenant_no == tenant_no,
            BizQcTaskAssignee.del_flag == 0,
        )
        .all()
    )
    for row in rows:
        row.del_flag = 1
        row.update_time = now
        db.add(row)


def list_assignees_by_task_ids(
    db: Session, task_ids: List[str], tenant_no: str
) -> Dict[str, List[Dict[str, str]]]:
    if not task_ids:
        return {}
    rows = (
        db.query(BizQcTaskAssignee, SysUser)
        .outerjoin(SysUser, eq_unicode(SysUser.id, BizQcTaskAssignee.user_id))
        .filter(
            BizQcTaskAssignee.task_id.in_(task_ids),
            BizQcTaskAssignee.tenant_no == tenant_no,
            BizQcTaskAssignee.del_flag == 0,
        )
        .order_by(BizQcTaskAssignee.order_num.asc(), BizQcTaskAssignee.create_time.asc())
        .all()
    )
    out: Dict[str, List[Dict[str, str]]] = {}
    for assignee, user in rows:
        out.setdefault(assignee.task_id, []).append(
            {
                "userId": assignee.user_id,
                "userName": (user.nick_name or user.user_no or assignee.user_id)
                if user
                else assignee.user_id,
            }
        )
    return out


def is_user_assignee(db: Session, task_id: str, user_id: str, tenant_no: str) -> bool:
    if not task_id or not user_id:
        return False
    row = (
        db.query(BizQcTaskAssignee.id)
        .filter(
            BizQcTaskAssignee.task_id == task_id,
            BizQcTaskAssignee.user_id == user_id,
            BizQcTaskAssignee.tenant_no == tenant_no,
            BizQcTaskAssignee.del_flag == 0,
        )
        .first()
    )
    return row is not None


def enrich_tasks_with_assignees(
    db: Session, items: List[dict], tenant_no: str
) -> List[dict]:
    if not items:
        return items
    task_ids = [t.get("id") for t in items if t.get("id")]
    mapping = list_assignees_by_task_ids(db, task_ids, tenant_no)
    for item in items:
        rows = mapping.get(item.get("id") or "", [])
        item["assigneeUserIds"] = [r["userId"] for r in rows]
        item["assigneeUserNames"] = "、".join(r["userName"] for r in rows if r.get("userName"))
    return items