354 lines
12 KiB
Python
354 lines
12 KiB
Python
#!/usr/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
审批人解析器
|
||
负责根据节点配置解析实际的审批人列表
|
||
"""
|
||
import logging
|
||
from typing import List, Dict, Any
|
||
|
||
from sqlalchemy import select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AssigneeResolver:
|
||
"""
|
||
审批人解析器
|
||
|
||
支持的审批人类型:
|
||
- user: 指定用户
|
||
- role: 指定角色（获取角色下所有用户）
|
||
- department: 指定部门（获取部门下所有用户）
|
||
- superior: 上级主管（根据发起人的部门层级）
|
||
- manager: 直属经理（发起人的直属上级）
|
||
- initiator: 发起人自己
|
||
- form_field: 表单字段（从表单数据中获取用户）
|
||
"""
|
||
|
||
async def resolve(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""
|
||
解析审批人
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
node_config: 节点配置（包含 assigneeType, assignees 等）
|
||
instance: 流程实例
|
||
form_data: 表单数据
|
||
|
||
Returns:
|
||
List[str]: 用户ID列表
|
||
"""
|
||
assignee_type = node_config.get('assigneeType', 'user')
|
||
|
||
resolvers = {
|
||
'user': self._resolve_users,
|
||
'role': self._resolve_by_role,
|
||
'department': self._resolve_by_department,
|
||
'superior': self._resolve_superior,
|
||
'manager': self._resolve_manager,
|
||
'initiator': self._resolve_initiator,
|
||
'form_field': self._resolve_from_form,
|
||
}
|
||
|
||
resolver = resolvers.get(assignee_type)
|
||
if not resolver:
|
||
logger.warning(f"未知的审批人类型: {assignee_type}")
|
||
return []
|
||
|
||
try:
|
||
return await resolver(db, node_config, instance, form_data)
|
||
except Exception as e:
|
||
logger.exception(f"解析审批人失败: {e}")
|
||
return []
|
||
|
||
async def _resolve_users(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""指定用户"""
|
||
assignees = node_config.get('assignees', [])
|
||
logger.info(f"[_resolve_users] assignees from config: {assignees}")
|
||
result = [str(uid) for uid in assignees] if assignees else []
|
||
logger.info(f"[_resolve_users] result: {result}")
|
||
return result
|
||
|
||
async def _resolve_by_role(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""根据角色获取用户（通过用户角色关联表）"""
|
||
from core.user.model import User
|
||
from core.user.user_role_model import UserRole
|
||
|
||
role_ids = node_config.get('assignees', [])
|
||
if not role_ids:
|
||
return []
|
||
|
||
# 通过 UserRole 关联表查询拥有指定角色的用户
|
||
stmt = select(User.id).join(
|
||
UserRole,
|
||
User.id == UserRole.user_id
|
||
).where(
|
||
UserRole.role_id.in_(role_ids),
|
||
User.is_deleted == False,
|
||
User.is_active == True,
|
||
).distinct()
|
||
|
||
result = await db.execute(stmt)
|
||
users = result.scalars().all()
|
||
|
||
logger.info(f"[_resolve_by_role] role_ids: {role_ids}, found users: {users}")
|
||
return [str(uid) for uid in users]
|
||
|
||
async def _resolve_by_department(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""根据部门获取用户"""
|
||
from core.user.model import User
|
||
|
||
dept_ids = node_config.get('assignees', [])
|
||
if not dept_ids:
|
||
return []
|
||
|
||
stmt = select(User.id).where(
|
||
User.dept_id.in_(dept_ids),
|
||
User.is_deleted == False,
|
||
User.is_active == True,
|
||
).distinct()
|
||
|
||
result = await db.execute(stmt)
|
||
users = result.scalars().all()
|
||
|
||
return [str(uid) for uid in users]
|
||
|
||
async def _resolve_superior(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""
|
||
获取上级主管
|
||
|
||
assigneeLevel 含义:
|
||
- 1 = 直接上级（当前部门的负责人）
|
||
- 2 = 上级的上级（父部门的负责人）
|
||
- 以此类推
|
||
"""
|
||
from core.user.model import User
|
||
from core.dept.model import Dept
|
||
|
||
level = node_config.get('assigneeLevel', 1)
|
||
initiator_id = instance.initiator_id
|
||
|
||
# 获取发起人信息
|
||
stmt = select(User).where(User.id == initiator_id)
|
||
result = await db.execute(stmt)
|
||
initiator = result.scalar_one_or_none()
|
||
|
||
if not initiator or not initiator.dept_id:
|
||
logger.warning(f"发起人 {initiator_id} 没有部门信息")
|
||
return []
|
||
|
||
# 获取部门信息
|
||
current_dept_id = initiator.dept_id
|
||
for _ in range(level - 1):
|
||
stmt = select(Dept).where(Dept.id == current_dept_id)
|
||
result = await db.execute(stmt)
|
||
dept = result.scalar_one_or_none()
|
||
if dept and dept.parent_id:
|
||
current_dept_id = dept.parent_id
|
||
else:
|
||
break
|
||
|
||
# 获取部门负责人
|
||
stmt = select(Dept).where(Dept.id == current_dept_id)
|
||
result = await db.execute(stmt)
|
||
current_dept = result.scalar_one_or_none()
|
||
|
||
if current_dept and current_dept.lead_id:
|
||
# 如果负责人是发起人自己，尝试向上找
|
||
if str(current_dept.lead_id) == str(initiator_id):
|
||
logger.info(f"部门负责人是发起人自己，尝试向上查找")
|
||
if current_dept.parent_id:
|
||
stmt = select(Dept).where(Dept.id == current_dept.parent_id)
|
||
result = await db.execute(stmt)
|
||
parent_dept = result.scalar_one_or_none()
|
||
if parent_dept and parent_dept.lead_id:
|
||
return [str(parent_dept.lead_id)]
|
||
return [str(current_dept.lead_id)]
|
||
|
||
logger.warning(f"部门没有负责人")
|
||
return []
|
||
|
||
async def _resolve_manager(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""
|
||
获取直属经理
|
||
|
||
assigneeLevel 含义:
|
||
- 1 = 直属经理（发起人的 manager）
|
||
- 2 = 经理的经理（发起人的 manager 的 manager）
|
||
- 以此类推
|
||
"""
|
||
from core.user.model import User
|
||
|
||
level = node_config.get('assigneeLevel', 1)
|
||
initiator_id = instance.initiator_id
|
||
|
||
current_user_id = initiator_id
|
||
for _ in range(level):
|
||
stmt = select(User).where(User.id == current_user_id)
|
||
result = await db.execute(stmt)
|
||
user = result.scalar_one_or_none()
|
||
|
||
if user and user.manager_id:
|
||
current_user_id = user.manager_id
|
||
else:
|
||
logger.warning(f"用户没有设置直属经理")
|
||
return []
|
||
|
||
return [str(current_user_id)]
|
||
|
||
async def _resolve_initiator(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""发起人自己"""
|
||
if instance.initiator_id:
|
||
return [str(instance.initiator_id)]
|
||
return []
|
||
|
||
async def _resolve_from_form(
|
||
self,
|
||
db: AsyncSession,
|
||
node_config: Dict,
|
||
instance: Any,
|
||
form_data: Dict,
|
||
) -> List[str]:
|
||
"""从表单字段获取用户，支持多字段收集，支持 form-selector 间接查询"""
|
||
field_names = node_config.get('assigneeFields') or node_config.get('recipientFields')
|
||
if not field_names:
|
||
single = node_config.get('assigneeField') or node_config.get('recipientField', '')
|
||
field_names = [single] if single else []
|
||
|
||
if not field_names:
|
||
return []
|
||
|
||
# 构建 form-selector 字段映射表: fieldName -> {formCode, userField}
|
||
field_mappings = {}
|
||
for mapping in (node_config.get('assigneeFieldMappings') or node_config.get('recipientFieldMappings') or []):
|
||
field_mappings[mapping['fieldName']] = mapping
|
||
|
||
result = []
|
||
seen = set()
|
||
for field_name in field_names:
|
||
value = form_data.get(field_name)
|
||
if not value:
|
||
continue
|
||
|
||
mapping = field_mappings.get(field_name)
|
||
if mapping:
|
||
# form-selector 字段: value 是引用表单记录的 ID，需要间接查询用户字段
|
||
user_ids = await self._resolve_form_selector_field(
|
||
db, value, mapping.get('formCode', ''), mapping.get('userField', '')
|
||
)
|
||
for uid_str in user_ids:
|
||
if uid_str and uid_str not in seen:
|
||
seen.add(uid_str)
|
||
result.append(uid_str)
|
||
else:
|
||
# user-selector 字段: value 直接就是用户 ID
|
||
values = value if isinstance(value, list) else [value]
|
||
for uid in values:
|
||
uid_str = str(uid)
|
||
if uid_str and uid_str not in seen:
|
||
seen.add(uid_str)
|
||
result.append(uid_str)
|
||
|
||
return result
|
||
|
||
async def _resolve_form_selector_field(
|
||
self,
|
||
db: AsyncSession,
|
||
record_ids: Any,
|
||
form_code: str,
|
||
user_field: str,
|
||
) -> List[str]:
|
||
"""
|
||
从 form-selector 引用的表单记录中提取用户字段值
|
||
|
||
Args:
|
||
db: 数据库会话
|
||
record_ids: 引用的表单记录 ID（单个或列表）
|
||
form_code: 引用的表单编码
|
||
user_field: 引用表单中的用户字段名
|
||
"""
|
||
if not form_code or not user_field:
|
||
logger.warning(f"form-selector 映射配置不完整: form_code={form_code}, user_field={user_field}")
|
||
return []
|
||
|
||
ids = record_ids if isinstance(record_ids, list) else [record_ids]
|
||
if not ids:
|
||
return []
|
||
|
||
result = []
|
||
try:
|
||
from online_dev.form_data_manager.service import FormDataService
|
||
|
||
service = await FormDataService.create_service(db, form_code)
|
||
|
||
for record_id in ids:
|
||
try:
|
||
record = await service.get(db, str(record_id))
|
||
if not record:
|
||
logger.warning(f"form-selector 引用记录不存在: form_code={form_code}, id={record_id}")
|
||
continue
|
||
|
||
user_value = record.get(user_field)
|
||
if not user_value:
|
||
continue
|
||
|
||
user_values = user_value if isinstance(user_value, list) else [user_value]
|
||
for uid in user_values:
|
||
result.append(str(uid))
|
||
except Exception as e:
|
||
logger.warning(f"查询 form-selector 引用记录失败: form_code={form_code}, id={record_id}, error={e}")
|
||
continue
|
||
|
||
except Exception as e:
|
||
logger.exception(f"form-selector 字段解析失败: {e}")
|
||
|
||
return result
|
||
|
||
|
||
# 全局实例
|
||
assignee_resolver = AssigneeResolver()
|