zkxw/backend/app/crud/production_task_step_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 BizProductionTaskStep, BizProductionTaskStepAssignee, SysRole, SysUser, SysUserRole

PRODUCTION_STEP_ASSIGNEE_ROLE = "实验专员"


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 _active_users_with_role_in_tenant(
    db: Session, user_ids: List[str], tenant_no: str, role_name: str
) -> Set[str]:
    ids = [x.strip() for x in user_ids if (x or "").strip()]
    rn = (role_name or "").strip()
    if not ids or not rn:
        return set()
    rows = (
        db.query(SysUser.id)
        .join(SysUserRole, SysUserRole.user_id == SysUser.id)
        .join(SysRole, SysRole.id == SysUserRole.role_id)
        .filter(
            SysUser.id.in_(ids),
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
            SysUserRole.del_flag == 0,
            SysUserRole.tenant_no == tenant_no,
            SysRole.del_flag == 0,
            SysRole.status == "1",
            SysRole.tenant_no == tenant_no,
            SysRole.role_name == rn,
        )
        .distinct()
        .all()
    )
    return {r[0] for r in rows}


def validate_assignee_user_ids(
    db: Session,
    user_ids: List[str],
    tenant_no: str,
    *,
    required_role_name: Optional[str] = PRODUCTION_STEP_ASSIGNEE_ROLE,
) -> 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("存在无效或已停用的负责人用户")
    rn = (required_role_name or "").strip()
    if rn:
        role_valid = _active_users_with_role_in_tenant(db, ordered, tenant_no, rn)
        role_invalid = [u for u in ordered if u not in role_valid]
        if role_invalid:
            raise ValueError(f"负责人须为「{rn}」角色用户")
    return ordered


def sync_assignees_for_step(
    db: Session,
    step_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(BizProductionTaskStepAssignee)
        .filter(
            BizProductionTaskStepAssignee.step_id == step_id,
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.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(
            BizProductionTaskStepAssignee(
                id=generate_uuid(),
                step_id=step_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 enrich_step_assignees(
    db: Session, steps: List[Dict], *, tenant_no: str
) -> None:
    """批量填充 assigneeUserIds / assigneeNames / assignees。"""
    step_ids = [s["id"] for s in steps if s.get("id")]
    if not step_ids:
        return
    rows = (
        db.query(BizProductionTaskStepAssignee, SysUser)
        .outerjoin(SysUser, eq_unicode(SysUser.id, BizProductionTaskStepAssignee.user_id))
        .filter(
            BizProductionTaskStepAssignee.step_id.in_(step_ids),
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.del_flag == 0,
        )
        .order_by(
            BizProductionTaskStepAssignee.order_num.asc(),
            BizProductionTaskStepAssignee.create_time.asc(),
        )
        .all()
    )
    by_step: Dict[str, List] = {}
    for assignee, user in rows:
        by_step.setdefault(assignee.step_id, []).append((assignee, user))
    for step in steps:
        pairs = by_step.get(step["id"], [])
        step["assigneeUserIds"] = [a.user_id for a, _ in pairs]
        step["assigneeNames"] = [
            f"{(u.nick_name or u.user_name or '')}（{u.user_no or ''}）".strip()
            for _, u in pairs
            if u
        ]
        step["assignees"] = [
            {
                "userId": a.user_id,
                "nickName": (u.nick_name or u.user_name or "") if u else "",
                "userNo": (u.user_no or "") if u else "",
            }
            for a, u in pairs
        ]


def enrich_step_completed_by_names(db: Session, steps: List[Dict]) -> None:
    uids = list({(s.get("completedBy") or "").strip() for s in steps if (s.get("completedBy") or "").strip()})
    if not uids:
        return
    users = {
        u.id: u
        for u in db.query(SysUser).filter(SysUser.id.in_(uids), SysUser.del_flag == 0).all()
    }
    for step in steps:
        uid = (step.get("completedBy") or "").strip()
        if not uid:
            step["completedByName"] = ""
            continue
        u = users.get(uid)
        step["completedByName"] = (
            f"{(u.nick_name or u.user_name or '')}（{u.user_no or ''}）".strip() if u else uid
        )


def user_is_step_assignee(
    db: Session, step_id: str, user_id: str, tenant_no: str
) -> bool:
    uid = (user_id or "").strip()
    if not uid:
        return False
    row = (
        db.query(BizProductionTaskStepAssignee.id)
        .filter(
            BizProductionTaskStepAssignee.step_id == step_id,
            BizProductionTaskStepAssignee.user_id == uid,
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.del_flag == 0,
        )
        .first()
    )
    return row is not None


def user_is_any_step_assignee_of_task(
    db: Session, task_id: str, user_id: str, tenant_no: str
) -> bool:
    uid = (user_id or "").strip()
    if not uid:
        return False
    row = (
        db.query(BizProductionTaskStepAssignee.id)
        .join(
            BizProductionTaskStep,
            BizProductionTaskStep.id == BizProductionTaskStepAssignee.step_id,
        )
        .filter(
            BizProductionTaskStep.task_id == task_id,
            BizProductionTaskStep.del_flag == 0,
            BizProductionTaskStepAssignee.user_id == uid,
            BizProductionTaskStepAssignee.tenant_no == tenant_no,
            BizProductionTaskStepAssignee.del_flag == 0,
        )
        .first()
    )
    return row is not None