257 lines
8.5 KiB
Python
257 lines
8.5 KiB
Python
"""
|
||
审批相关的工具函数
|
||
"""
|
||
import json
|
||
from datetime import datetime
|
||
from .models import OperationLog, User, Team, Approval
|
||
|
||
def is_department_id(value):
|
||
"""
|
||
判断personincharge字段的值是部门ID还是审批员用户名
|
||
|
||
统一规则：
|
||
- 如果是纯数字字符串（如 "1", "2", "123"），表示部门ID
|
||
- 如果包含非数字字符（如 "张三", "李四"），表示审批员用户名
|
||
|
||
Args:
|
||
value: personincharge字段的值（字符串）
|
||
|
||
Returns:
|
||
bool: True表示是部门ID，False表示是审批员用户名
|
||
|
||
示例:
|
||
>>> is_department_id("1")
|
||
True
|
||
>>> is_department_id("123")
|
||
True
|
||
>>> is_department_id("张三")
|
||
False
|
||
>>> is_department_id("dept:1")
|
||
False
|
||
"""
|
||
if not value:
|
||
return False
|
||
# 判断是否为纯数字字符串（去除首尾空格）
|
||
return str(value).strip().isdigit()
|
||
|
||
|
||
def format_personincharge(value, is_department=False):
|
||
"""
|
||
格式化personincharge字段的值
|
||
|
||
统一规则：
|
||
- 如果是部门ID，确保是纯数字字符串
|
||
- 如果是审批员用户名，保持原样
|
||
|
||
Args:
|
||
value: 部门ID（整数或字符串）或审批员用户名（字符串）
|
||
is_department: 是否为部门ID，默认False（审批员用户名）
|
||
|
||
Returns:
|
||
str: 格式化后的personincharge值
|
||
"""
|
||
if not value:
|
||
return ''
|
||
|
||
if is_department:
|
||
# 部门ID：转换为字符串，确保是纯数字
|
||
try:
|
||
return str(int(value))
|
||
except (ValueError, TypeError):
|
||
raise ValueError(f"部门ID必须是数字，当前值: {value}")
|
||
else:
|
||
# 审批员用户名：保持原样，但确保是字符串
|
||
return str(value).strip()
|
||
|
||
|
||
def log_operation(request, operation_type, module, action, target_type, target_id=None,
|
||
target_name=None, old_data=None, new_data=None, remark=None):
|
||
"""
|
||
记录操作日志
|
||
|
||
Args:
|
||
request: Django request对象
|
||
operation_type: 操作类型（DELETE, CREATE, UPDATE, APPROVE等）
|
||
module: 模块名称（User, Business, Finance等）
|
||
action: 操作描述（如"删除用户"、"创建立项"等）
|
||
target_type: 目标类型（如User, ProjectRegistration等）
|
||
target_id: 目标ID
|
||
target_name: 目标名称（如用户名、项目名等）
|
||
old_data: 操作前的数据（字典，会自动转换为JSON）
|
||
new_data: 操作后的数据（字典，会自动转换为JSON）
|
||
remark: 备注信息
|
||
|
||
Returns:
|
||
OperationLog对象
|
||
"""
|
||
try:
|
||
# 获取操作人信息
|
||
token = request.META.get('token') or request.META.get('HTTP_AUTHORIZATION', '').replace('Bearer ', '')
|
||
operator = '未知用户'
|
||
operator_id = None
|
||
|
||
if token:
|
||
try:
|
||
user = User.objects.get(token=token, is_deleted=False)
|
||
operator = user.username
|
||
operator_id = user.id
|
||
except User.DoesNotExist:
|
||
pass
|
||
|
||
# 获取IP地址
|
||
ip_address = request.META.get('HTTP_X_FORWARDED_FOR', '').split(',')[0].strip()
|
||
if not ip_address:
|
||
ip_address = request.META.get('REMOTE_ADDR', '')
|
||
|
||
# 获取用户代理
|
||
user_agent = request.META.get('HTTP_USER_AGENT', '')
|
||
|
||
# 转换数据为JSON字符串
|
||
old_data_str = json.dumps(old_data, ensure_ascii=False) if old_data else None
|
||
new_data_str = json.dumps(new_data, ensure_ascii=False) if new_data else None
|
||
|
||
# 创建日志记录
|
||
log = OperationLog.objects.create(
|
||
operator=operator,
|
||
operator_id=operator_id,
|
||
operation_type=operation_type,
|
||
module=module,
|
||
action=action,
|
||
target_type=target_type,
|
||
target_id=str(target_id) if target_id else None,
|
||
target_name=target_name,
|
||
old_data=old_data_str,
|
||
new_data=new_data_str,
|
||
ip_address=ip_address,
|
||
user_agent=user_agent,
|
||
request_path=request.path,
|
||
remark=remark
|
||
)
|
||
return log
|
||
except Exception as e:
|
||
# 日志记录失败不应该影响主业务流程
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
logger.error(f"记录操作日志失败: {str(e)}")
|
||
return None
|
||
|
||
|
||
def create_team_approval(team_name, title, content, approval_type, user_id, request=None):
|
||
"""
|
||
根据团队类型创建审批流程
|
||
|
||
规则：
|
||
- 个人团队（personal）：不需要审核人，直接抄送财务（personincharge设为"财务"）
|
||
- 团队（team）：需要指定审核人（可以多个），审核通过后才抄送财务
|
||
|
||
Args:
|
||
team_name: 团队名称
|
||
title: 审批标题
|
||
content: 审批内容
|
||
approval_type: 审批类型（如"立项登记"、"付款申请"等）
|
||
user_id: 关联的业务ID（字符串）
|
||
request: Django request对象（可选，用于记录日志）
|
||
|
||
Returns:
|
||
Approval对象或None
|
||
"""
|
||
try:
|
||
# 查找团队
|
||
team = Team.objects.prefetch_related('approvers').filter(name=team_name, is_deleted=False).first()
|
||
|
||
if not team:
|
||
# 如果找不到团队，使用默认流程（需要审核人）
|
||
return None
|
||
|
||
today = datetime.now().strftime("%Y-%m-%d")
|
||
|
||
if team.team_type == 'personal':
|
||
# 个人团队：直接抄送财务，不需要审核人
|
||
approval = Approval.objects.create(
|
||
title=title,
|
||
content=content,
|
||
times=today,
|
||
personincharge='财务', # 直接抄送财务
|
||
state='已抄送财务', # 直接标记为已抄送财务
|
||
type=approval_type,
|
||
user_id=str(user_id)
|
||
)
|
||
return approval
|
||
|
||
elif team.team_type == 'team':
|
||
# 团队：需要审核人审核
|
||
approvers = team.approvers.filter(is_deleted=False)
|
||
if not approvers.exists():
|
||
# 如果没有审核人，返回None（应该在前端验证）
|
||
return None
|
||
|
||
# 将审核人用户名用逗号连接（多个审核人）
|
||
approver_names = ','.join([approver.username for approver in approvers])
|
||
|
||
approval = Approval.objects.create(
|
||
title=title,
|
||
content=content,
|
||
times=today,
|
||
personincharge=approver_names, # 多个审核人用逗号分隔
|
||
state='审核中',
|
||
type=approval_type,
|
||
user_id=str(user_id)
|
||
)
|
||
return approval
|
||
|
||
return None
|
||
|
||
except Exception as e:
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
logger.error(f"创建团队审批失败: {str(e)}")
|
||
return None
|
||
|
||
|
||
def get_team_approval_info(team_name):
|
||
"""
|
||
获取团队的审批信息
|
||
|
||
Args:
|
||
team_name: 团队名称
|
||
|
||
Returns:
|
||
dict: {
|
||
'team_type': 'personal' 或 'team',
|
||
'needs_approval': True/False, # 是否需要审核
|
||
'approvers': [{'id': 1, 'username': '张三'}, ...], # 审核人列表
|
||
'direct_to_finance': True/False # 是否直接抄送财务
|
||
}
|
||
"""
|
||
try:
|
||
team = Team.objects.prefetch_related('approvers').filter(name=team_name, is_deleted=False).first()
|
||
|
||
if not team:
|
||
return {
|
||
'team_type': None,
|
||
'needs_approval': True, # 默认需要审核
|
||
'approvers': [],
|
||
'direct_to_finance': False
|
||
}
|
||
|
||
approvers = list(team.approvers.filter(is_deleted=False).values('id', 'username'))
|
||
|
||
return {
|
||
'team_type': team.team_type,
|
||
'needs_approval': team.team_type == 'team',
|
||
'approvers': approvers,
|
||
'direct_to_finance': team.team_type == 'personal'
|
||
}
|
||
|
||
except Exception as e:
|
||
import logging
|
||
logger = logging.getLogger(__name__)
|
||
logger.error(f"获取团队审批信息失败: {str(e)}")
|
||
return {
|
||
'team_type': None,
|
||
'needs_approval': True,
|
||
'approvers': [],
|
||
'direct_to_finance': False
|
||
}
|
||
|