zkxw/backend/app/services/oa_assignee.py

"""OA 审批人解析（不读 sys_role）"""
from __future__ import annotations

from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy.orm import Session

from app.crud import oa_approval as oa_crud
from app.models.models import SysDept, SysUser


class AssigneeResolveError(ValueError):
    """审批人解析失败"""


def _active_user_id(db: Session, user_id: Optional[str], tenant_no: str) -> Optional[str]:
    uid = (user_id or "").strip()
    if not uid:
        return None
    u = (
        db.query(SysUser)
        .filter(
            SysUser.id == uid,
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
            SysUser.status == "1",
        )
        .first()
    )
    return u.id if u else None


def resolve_dept_leader(
    db: Session, *, tenant_no: str, initiator_dept_id: Optional[str]
) -> Tuple[List[str], Dict[str, Any]]:
    dept_id = (initiator_dept_id or "").strip()
    if not dept_id:
        raise AssigneeResolveError("发起人部门为空，无法解析部门负责人")
    dept = (
        db.query(SysDept)
        .filter(SysDept.id == dept_id, SysDept.tenant_no == tenant_no, SysDept.del_flag == 0)
        .first()
    )
    if not dept:
        raise AssigneeResolveError("发起人部门不存在")
    leader_uid = _active_user_id(db, getattr(dept, "leader_user_id", None), tenant_no)
    if not leader_uid:
        raise AssigneeResolveError("部门负责人未配置或用户无效，请在组织管理中维护负责人")
    return [leader_uid], {"resolve_source": "dept_leader"}


def resolve_direct_supervisor(
    db: Session,
    *,
    tenant_no: str,
    initiator_user_id: str,
    initiator_dept_id: Optional[str],
    fallback_to_dept_leader: bool = True,
) -> Tuple[List[str], Dict[str, Any]]:
    initiator = (
        db.query(SysUser)
        .filter(
            SysUser.id == initiator_user_id,
            SysUser.tenant_no == tenant_no,
            SysUser.del_flag == 0,
        )
        .first()
    )
    if not initiator:
        raise AssigneeResolveError("发起人不存在")
    supervisor_uid = _active_user_id(
        db, getattr(initiator, "supervisor_user_id", None), tenant_no
    )
    if supervisor_uid:
        if supervisor_uid == initiator_user_id:
            raise AssigneeResolveError("直属主管不能是发起人本人，请修正用户主管配置")
        return [supervisor_uid], {"resolve_source": "direct_supervisor"}
    if fallback_to_dept_leader:
        users, meta = resolve_dept_leader(
            db, tenant_no=tenant_no, initiator_dept_id=initiator_dept_id or initiator.dept_id
        )
        meta = {**meta, "resolve_source": "supervisor_fallback_dept_leader"}
        return users, meta
    raise AssigneeResolveError("未配置直属主管，且未开启回退部门负责人")


def resolve_assignees(
    db: Session,
    *,
    tenant_no: str,
    assignee_config: Dict[str, Any],
    initiator_user_id: str,
    initiator_dept_id: Optional[str] = None,
    advanced: Optional[Dict[str, Any]] = None,
    initiator_select_user_ids: Optional[List[str]] = None,
) -> Tuple[List[str], Dict[str, Any]]:
    """
    按节点 approverConfig 解析处理人。
    assigneeType: users | oa_roles | dept_leader | direct_supervisor | initiator_select
    """
    cfg = assignee_config or {}
    assignee_type = (cfg.get("assigneeType") or cfg.get("assignee_type") or "").strip()
    advanced = advanced or {}
    fallback = advanced.get("supervisorFallbackToDeptLeader", True)
    if isinstance(fallback, str):
        fallback = fallback.lower() in ("1", "true", "yes")

    if assignee_type == "users":
        ids = cfg.get("userIds") or cfg.get("user_ids") or []
        out: List[str] = []
        seen = set()
        for uid in ids:
            real = _active_user_id(db, uid, tenant_no)
            if real and real not in seen:
                seen.add(real)
                out.append(real)
        if not out:
            raise AssigneeResolveError("指定成员为空或均无效")
        return out, {"resolve_source": "users"}

    if assignee_type == "oa_roles":
        role_ids = cfg.get("oaRoleIds") or cfg.get("oa_role_ids") or []
        # 也支持按编码
        codes = cfg.get("oaRoleCodes") or cfg.get("oa_role_codes") or []
        if codes:
            from app.models.models import OaRole

            extra = (
                db.query(OaRole.id)
                .filter(
                    OaRole.tenant_no == tenant_no,
                    OaRole.role_code.in_([c.upper() for c in codes]),
                    OaRole.del_flag == 0,
                    OaRole.status == "1",
                )
                .all()
            )
            role_ids = list(role_ids) + [r.id for r in extra]
        users = oa_crud.list_oa_role_user_ids(db, list(role_ids), tenant_no)
        # 过滤停用用户
        out = []
        seen = set()
        for uid in users:
            real = _active_user_id(db, uid, tenant_no)
            if real and real not in seen:
                seen.add(real)
                out.append(real)
        if not out:
            raise AssigneeResolveError("工作流角色成员为空或均无效")
        return out, {"resolve_source": "oa_roles"}

    if assignee_type == "dept_leader":
        return resolve_dept_leader(
            db, tenant_no=tenant_no, initiator_dept_id=initiator_dept_id
        )

    if assignee_type == "direct_supervisor":
        return resolve_direct_supervisor(
            db,
            tenant_no=tenant_no,
            initiator_user_id=initiator_user_id,
            initiator_dept_id=initiator_dept_id,
            fallback_to_dept_leader=bool(fallback),
        )

    if assignee_type == "initiator_select":
        ids = initiator_select_user_ids or cfg.get("userIds") or []
        out = []
        seen = set()
        for uid in ids:
            real = _active_user_id(db, uid, tenant_no)
            if real and real not in seen:
                seen.add(real)
                out.append(real)
        if not out:
            raise AssigneeResolveError("发起人未选择审批人")
        return out, {"resolve_source": "initiator_select"}

    raise AssigneeResolveError(f"不支持的审批人类型: {assignee_type or '(空)'}")