feat: add email notice and settings page but still have some issue
This commit is contained in:
@@ -0,0 +1,2 @@
|
||||
|
||||
|
||||
@@ -0,0 +1,22 @@
|
||||
from django.contrib import admin
|
||||
from .models import NotificationEmail, NotificationSettings
|
||||
|
||||
@admin.register(NotificationEmail)
|
||||
class NotificationEmailAdmin(admin.ModelAdmin):
|
||||
list_display = ('email', 'is_enabled', 'description', 'created_at')
|
||||
list_filter = ('is_enabled', 'created_at')
|
||||
search_fields = ('email', 'description')
|
||||
list_editable = ('is_enabled',)
|
||||
ordering = ('-created_at',)
|
||||
|
||||
@admin.register(NotificationSettings)
|
||||
class NotificationSettingsAdmin(admin.ModelAdmin):
|
||||
list_display = ('email_notification_enabled', 'updated_at')
|
||||
|
||||
def has_add_permission(self, request):
|
||||
# 只允许存在一个设置实例
|
||||
return not NotificationSettings.objects.exists()
|
||||
|
||||
def has_delete_permission(self, request, obj=None):
|
||||
# 不允许删除设置
|
||||
return False
|
||||
@@ -0,0 +1,6 @@
|
||||
from django.apps import AppConfig
|
||||
|
||||
|
||||
class EmailConfig(AppConfig):
|
||||
default_auto_field = 'django.db.models.BigAutoField'
|
||||
name = 'email_notice'
|
||||
@@ -0,0 +1,297 @@
|
||||
from django.utils.deprecation import MiddlewareMixin
|
||||
from django.http import JsonResponse
|
||||
from .services import EmailNotificationService
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from django.core.exceptions import ObjectDoesNotExist
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class EmailNotificationMiddleware(MiddlewareMixin):
|
||||
"""邮件通知中间件"""
|
||||
|
||||
def __init__(self, get_response):
|
||||
self.get_response = get_response
|
||||
super().__init__(get_response)
|
||||
# 用于存储删除前的数据
|
||||
self._delete_data_cache = {}
|
||||
|
||||
def process_request(self, request):
|
||||
"""在处理请求前，如果是删除操作，先获取要删除的数据"""
|
||||
try:
|
||||
# 只处理DELETE操作
|
||||
if request.method != 'DELETE':
|
||||
return None
|
||||
|
||||
# 只处理API请求
|
||||
if not request.path.startswith('/api/'):
|
||||
return None
|
||||
|
||||
# 从URL路径中提取ID
|
||||
path_parts = request.path.split('/')
|
||||
record_id = None
|
||||
for part in path_parts:
|
||||
if part.isdigit():
|
||||
record_id = part
|
||||
break
|
||||
|
||||
if not record_id:
|
||||
return None
|
||||
|
||||
# 根据路径类型获取对应的数据
|
||||
if '/api/items/' in request.path:
|
||||
data = self._get_item_data_before_delete(record_id)
|
||||
if data:
|
||||
self._delete_data_cache[f"item_{record_id}"] = data
|
||||
elif '/api/records/' in request.path or '/api/finance/' in request.path:
|
||||
data = self._get_finance_data_before_delete(record_id)
|
||||
if data:
|
||||
self._delete_data_cache[f"finance_{record_id}"] = data
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"获取删除前数据失败: {e}")
|
||||
|
||||
return None
|
||||
|
||||
def _get_item_data_before_delete(self, item_id):
|
||||
"""获取物品删除前的数据"""
|
||||
try:
|
||||
from items.models import Item
|
||||
item = Item.objects.get(id=item_id)
|
||||
return {
|
||||
'id': item.id,
|
||||
'name': item.name,
|
||||
'serial_number': item.serial_number,
|
||||
'status': item.status,
|
||||
'category': item.category,
|
||||
'owner': item.owner,
|
||||
'purchase_date': str(item.purchase_date) if item.purchase_date else '',
|
||||
'purchase_price': str(item.purchase_price) if item.purchase_price else '',
|
||||
'description': item.description,
|
||||
'updated_at': str(item.updated_at) if hasattr(item, 'updated_at') else '',
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"获取物品数据失败: {e}")
|
||||
return None
|
||||
|
||||
def _get_finance_data_before_delete(self, record_id):
|
||||
"""获取财务记录删除前的数据"""
|
||||
try:
|
||||
from finance.models import FinanceRecord
|
||||
record = FinanceRecord.objects.get(id=record_id)
|
||||
return {
|
||||
'id': record.id,
|
||||
'amount': str(record.amount),
|
||||
'transaction_type': record.transaction_type,
|
||||
'description': record.description,
|
||||
'department': record.department.name if record.department else '',
|
||||
'category': record.category.name if record.category else '',
|
||||
'approver': record.approver,
|
||||
'transaction_date': str(record.transaction_date),
|
||||
'proof_images': [str(img.image) for img in record.proof_images.all()] if hasattr(record, 'proof_images') else []
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"获取财务记录数据失败: {e}")
|
||||
return None
|
||||
|
||||
def process_response(self, request, response):
|
||||
"""处理响应，在数据修改操作成功后异步发送邮件通知"""
|
||||
try:
|
||||
# 只处理API请求
|
||||
if not request.path.startswith('/api/'):
|
||||
return response
|
||||
|
||||
# 只处理数据修改操作
|
||||
if request.method not in ['POST', 'PUT', 'PATCH', 'DELETE']:
|
||||
return response
|
||||
|
||||
# 只处理成功的响应
|
||||
if not (200 <= response.status_code < 300):
|
||||
return response
|
||||
|
||||
# 获取用户信息
|
||||
user_info = self._get_user_info(request)
|
||||
|
||||
# 异步处理邮件通知，避免阻塞API响应
|
||||
if '/api/items/' in request.path:
|
||||
self._handle_item_operation_async(request, response, user_info)
|
||||
elif '/api/records/' in request.path or '/api/finance/' in request.path:
|
||||
self._handle_finance_operation_async(request, response, user_info)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"邮件通知中间件处理失败: {e}")
|
||||
|
||||
return response
|
||||
|
||||
def _get_user_info(self, request):
|
||||
"""获取用户信息"""
|
||||
try:
|
||||
if hasattr(request, 'user') and request.user.is_authenticated:
|
||||
return f"{request.user.username} ({request.user.email})"
|
||||
else:
|
||||
return f"匿名用户 (IP: {self._get_client_ip(request)})"
|
||||
except:
|
||||
return "未知用户"
|
||||
|
||||
def _get_client_ip(self, request):
|
||||
"""获取客户端IP"""
|
||||
x_forwarded_for = request.META.get('HTTP_X_FORWARDED_FOR')
|
||||
if x_forwarded_for:
|
||||
ip = x_forwarded_for.split(',')[0]
|
||||
else:
|
||||
ip = request.META.get('REMOTE_ADDR')
|
||||
return ip
|
||||
|
||||
def _handle_item_operation_async(self, request, response, user_info):
|
||||
"""异步处理物品操作"""
|
||||
def send_notification():
|
||||
try:
|
||||
operation_type = self._get_operation_type(request.method, request.path)
|
||||
|
||||
# 对于DELETE操作，使用预先获取的数据
|
||||
if request.method == 'DELETE':
|
||||
# 从URL路径中提取ID
|
||||
path_parts = request.path.split('/')
|
||||
item_id = None
|
||||
for part in path_parts:
|
||||
if part.isdigit():
|
||||
item_id = part
|
||||
break
|
||||
|
||||
# 尝试从缓存中获取删除前的数据
|
||||
cached_data = self._delete_data_cache.get(f"item_{item_id}")
|
||||
if cached_data:
|
||||
notification_data = {
|
||||
'id': cached_data.get('id', item_id),
|
||||
'name': f"[已删除] {cached_data.get('name', '未知物品')}",
|
||||
'serial_number': cached_data.get('serial_number', ''),
|
||||
'status': f"[删除前状态: {cached_data.get('status', '未知')}]",
|
||||
'category': cached_data.get('category', ''),
|
||||
'owner': cached_data.get('owner', ''),
|
||||
'purchase_date': cached_data.get('purchase_date', ''),
|
||||
'purchase_price': cached_data.get('purchase_price', ''),
|
||||
'description': f"删除前描述: {cached_data.get('description', '')}",
|
||||
'timestamp': cached_data.get('updated_at', ''),
|
||||
'operation_path': request.path,
|
||||
'operation_method': request.method
|
||||
}
|
||||
else:
|
||||
# 尝试从响应中获取数据
|
||||
if hasattr(response, 'data'):
|
||||
item_data = response.data
|
||||
else:
|
||||
try:
|
||||
content = response.content.decode('utf-8')
|
||||
item_data = json.loads(content) if content else {}
|
||||
except:
|
||||
item_data = {}
|
||||
|
||||
# 确保item_data不为None
|
||||
if item_data is None:
|
||||
item_data = {}
|
||||
|
||||
# 构建通知数据
|
||||
notification_data = {
|
||||
'id': item_data.get('id', ''),
|
||||
'name': item_data.get('name', ''),
|
||||
'serial_number': item_data.get('serial_number', ''),
|
||||
'status': item_data.get('status', ''),
|
||||
'category': item_data.get('category', ''),
|
||||
'owner': item_data.get('owner', ''),
|
||||
'timestamp': item_data.get('updated_at', ''),
|
||||
'operation_path': request.path,
|
||||
'operation_method': request.method
|
||||
}
|
||||
|
||||
EmailNotificationService.send_operation_notification(
|
||||
operation_type, '物品', notification_data, user_info
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"异步处理物品操作通知失败: {e} 如果需要查看删除前的数据，请检查以前的邮件。")
|
||||
|
||||
# 创建并启动后台线程
|
||||
thread = threading.Thread(target=send_notification)
|
||||
thread.daemon = True # 设置为守护线程
|
||||
thread.start()
|
||||
|
||||
def _handle_finance_operation_async(self, request, response, user_info):
|
||||
"""异步处理财务操作"""
|
||||
def send_notification():
|
||||
try:
|
||||
operation_type = self._get_operation_type(request.method, request.path)
|
||||
|
||||
# 对于DELETE操作，响应通常不包含数据，需要从URL路径中提取ID
|
||||
if request.method == 'DELETE':
|
||||
# 从URL路径中提取ID
|
||||
path_parts = request.path.split('/')
|
||||
record_id = None
|
||||
for part in path_parts:
|
||||
if part.isdigit():
|
||||
record_id = part
|
||||
break
|
||||
|
||||
notification_data = {
|
||||
'id': record_id or '未知',
|
||||
'amount': '已删除记录',
|
||||
'transaction_type': '已删除',
|
||||
'description': '财务记录已被删除',
|
||||
'department': '未知',
|
||||
'category': '未知',
|
||||
'approver': '未知',
|
||||
'timestamp': '删除时间未知',
|
||||
'operation_path': request.path,
|
||||
'operation_method': request.method
|
||||
}
|
||||
else:
|
||||
# 尝试从响应中获取数据
|
||||
if hasattr(response, 'data'):
|
||||
finance_data = response.data
|
||||
else:
|
||||
try:
|
||||
content = response.content.decode('utf-8')
|
||||
finance_data = json.loads(content) if content else {}
|
||||
except:
|
||||
finance_data = {}
|
||||
|
||||
# 确保finance_data不为None
|
||||
if finance_data is None:
|
||||
finance_data = {}
|
||||
|
||||
# 构建通知数据
|
||||
notification_data = {
|
||||
'id': finance_data.get('id', ''),
|
||||
'amount': finance_data.get('amount', ''),
|
||||
'transaction_type': finance_data.get('transaction_type', ''),
|
||||
'description': finance_data.get('description', ''),
|
||||
'department': finance_data.get('department', ''),
|
||||
'category': finance_data.get('category', ''),
|
||||
'approver': finance_data.get('approver', ''),
|
||||
'timestamp': finance_data.get('transaction_date', ''),
|
||||
'operation_path': request.path,
|
||||
'operation_method': request.method
|
||||
}
|
||||
|
||||
EmailNotificationService.send_operation_notification(
|
||||
operation_type, '财务记录', notification_data, user_info
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"异步处理财务操作通知失败: {e} 如果需要查看删除前的数据，请检查以前的邮件。")
|
||||
|
||||
# 创建并启动后台线程
|
||||
thread = threading.Thread(target=send_notification)
|
||||
thread.daemon = True # 设置为守护线程
|
||||
thread.start()
|
||||
|
||||
def _get_operation_type(self, method, path):
|
||||
"""根据HTTP方法和路径判断操作类型"""
|
||||
if method == 'POST':
|
||||
return 'CREATE'
|
||||
elif method in ['PUT', 'PATCH']:
|
||||
return 'UPDATE'
|
||||
elif method == 'DELETE':
|
||||
return 'DELETE'
|
||||
else:
|
||||
return 'UNKNOWN'
|
||||
@@ -0,0 +1,66 @@
|
||||
from django.db import models
|
||||
from django.core.validators import EmailValidator
|
||||
|
||||
class NotificationEmail(models.Model):
|
||||
"""通知邮箱模型"""
|
||||
email = models.EmailField(
|
||||
unique=True,
|
||||
validators=[EmailValidator()],
|
||||
verbose_name='邮箱地址'
|
||||
)
|
||||
is_enabled = models.BooleanField(
|
||||
default=True,
|
||||
verbose_name='是否启用'
|
||||
)
|
||||
description = models.CharField(
|
||||
max_length=200,
|
||||
blank=True,
|
||||
null=True,
|
||||
verbose_name='描述'
|
||||
)
|
||||
created_at = models.DateTimeField(
|
||||
auto_now_add=True,
|
||||
verbose_name='创建时间'
|
||||
)
|
||||
updated_at = models.DateTimeField(
|
||||
auto_now=True,
|
||||
verbose_name='更新时间'
|
||||
)
|
||||
|
||||
class Meta:
|
||||
verbose_name = '通知邮箱'
|
||||
verbose_name_plural = '通知邮箱'
|
||||
ordering = ['-created_at']
|
||||
|
||||
def __str__(self):
|
||||
status = "启用" if self.is_enabled else "禁用"
|
||||
return f"{self.email} ({status})"
|
||||
|
||||
class NotificationSettings(models.Model):
|
||||
"""通知设置模型"""
|
||||
email_notification_enabled = models.BooleanField(
|
||||
default=True,
|
||||
verbose_name='邮件通知总开关'
|
||||
)
|
||||
created_at = models.DateTimeField(
|
||||
auto_now_add=True,
|
||||
verbose_name='创建时间'
|
||||
)
|
||||
updated_at = models.DateTimeField(
|
||||
auto_now=True,
|
||||
verbose_name='更新时间'
|
||||
)
|
||||
|
||||
class Meta:
|
||||
verbose_name = '通知设置'
|
||||
verbose_name_plural = '通知设置'
|
||||
|
||||
def __str__(self):
|
||||
status = "开启" if self.email_notification_enabled else "关闭"
|
||||
return f"邮件通知: {status}"
|
||||
|
||||
@classmethod
|
||||
def get_settings(cls):
|
||||
"""获取通知设置（单例模式）"""
|
||||
settings, created = cls.objects.get_or_create(pk=1)
|
||||
return settings
|
||||
@@ -0,0 +1,232 @@
|
||||
from django.core.mail import send_mail
|
||||
from django.conf import settings
|
||||
import json
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
class EmailNotificationService:
|
||||
"""邮件通知服务"""
|
||||
|
||||
@staticmethod
|
||||
def get_notification_settings():
|
||||
"""获取通知设置"""
|
||||
try:
|
||||
from .models import NotificationEmail, NotificationSettings
|
||||
|
||||
# 获取全局设置
|
||||
global_settings = NotificationSettings.get_settings()
|
||||
|
||||
# 获取启用的邮箱列表
|
||||
enabled_emails = NotificationEmail.objects.filter(is_enabled=True).values_list('email', flat=True)
|
||||
|
||||
return {
|
||||
'email_enabled': global_settings.email_notification_enabled,
|
||||
'notification_emails': list(enabled_emails)
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"获取通知设置失败: {e}")
|
||||
return {'email_enabled': False, 'notification_emails': []}
|
||||
|
||||
@staticmethod
|
||||
def send_operation_notification(operation_type, model_name, instance_data, user_info=None):
|
||||
"""
|
||||
发送操作通知邮件到多个邮箱
|
||||
|
||||
Args:
|
||||
operation_type: 操作类型 ('CREATE', 'UPDATE', 'DELETE')
|
||||
model_name: 模型名称 (如 'Item', 'FinanceRecord')
|
||||
instance_data: 实例数据
|
||||
user_info: 操作用户信息
|
||||
"""
|
||||
try:
|
||||
settings_data = EmailNotificationService.get_notification_settings()
|
||||
|
||||
if not settings_data.get('email_enabled') or not settings_data.get('notification_emails'):
|
||||
return
|
||||
|
||||
notification_emails = settings_data.get('notification_emails', [])
|
||||
if not notification_emails:
|
||||
logger.warning("没有配置启用的通知邮箱")
|
||||
return
|
||||
|
||||
# 构建邮件内容
|
||||
subject = f"[爱特工作室管理系统] {model_name}数据变更通知"
|
||||
|
||||
operation_map = {
|
||||
'CREATE': '创建',
|
||||
'UPDATE': '更新',
|
||||
'DELETE': '删除'
|
||||
}
|
||||
|
||||
operation_text = operation_map.get(operation_type, operation_type)
|
||||
|
||||
message = f"""
|
||||
系统数据变更通知
|
||||
|
||||
操作类型: {operation_text}
|
||||
数据类型: {model_name}
|
||||
操作时间: {instance_data.get('timestamp', '未知')}
|
||||
操作用户: {user_info or '系统'}
|
||||
|
||||
变更详情:
|
||||
{json.dumps(instance_data, ensure_ascii=False, indent=2)}
|
||||
|
||||
---
|
||||
此邮件由爱特工作室物品管理及财务管理系统自动发送
|
||||
"""
|
||||
|
||||
# 发送邮件到所有启用的通知邮箱
|
||||
for email in notification_emails:
|
||||
if email.strip(): # 确保邮箱不为空
|
||||
try:
|
||||
send_mail(
|
||||
subject=subject,
|
||||
message=message,
|
||||
from_email=settings.EMAIL_HOST_USER,
|
||||
recipient_list=[email.strip()],
|
||||
fail_silently=False,
|
||||
)
|
||||
logger.info(f"邮件通知发送成功到 {email}: {operation_type} {model_name}")
|
||||
except Exception as e:
|
||||
logger.error(f"发送邮件到 {email} 失败: {e}")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"发送邮件通知失败: {e}")
|
||||
|
||||
@staticmethod
|
||||
def send_item_operation_notification(operation_type, item_instance, user_info=None):
|
||||
"""发送物品操作通知"""
|
||||
try:
|
||||
instance_data = {
|
||||
'id': getattr(item_instance, 'id', None),
|
||||
'name': getattr(item_instance, 'name', ''),
|
||||
'serial_number': getattr(item_instance, 'serial_number', ''),
|
||||
'status': getattr(item_instance, 'status', ''),
|
||||
'category': getattr(item_instance, 'category', ''),
|
||||
'owner': getattr(item_instance, 'owner', ''),
|
||||
'timestamp': str(getattr(item_instance, 'updated_at', '')),
|
||||
}
|
||||
|
||||
EmailNotificationService.send_operation_notification(
|
||||
operation_type, '物品', instance_data, user_info
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"发送物品操作通知失败: {e}")
|
||||
|
||||
@staticmethod
|
||||
def send_finance_operation_notification(operation_type, finance_instance, user_info=None):
|
||||
"""发送财务记录操作通知"""
|
||||
try:
|
||||
instance_data = {
|
||||
'id': getattr(finance_instance, 'id', None),
|
||||
'amount': str(getattr(finance_instance, 'amount', '')),
|
||||
'transaction_type': getattr(finance_instance, 'transaction_type', ''),
|
||||
'description': getattr(finance_instance, 'description', ''),
|
||||
'department': str(getattr(finance_instance, 'department', '')),
|
||||
'category': str(getattr(finance_instance, 'category', '')),
|
||||
'approver': getattr(finance_instance, 'approver', ''),
|
||||
'timestamp': str(getattr(finance_instance, 'transaction_date', '')),
|
||||
}
|
||||
|
||||
EmailNotificationService.send_operation_notification(
|
||||
operation_type, '财务记录', instance_data, user_info
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"发送财务记录操作通知失败: {e}")
|
||||
|
||||
@staticmethod
|
||||
def update_notification_emails(emails_data):
|
||||
"""
|
||||
更新通知邮箱列表（使用数据库存储）
|
||||
|
||||
Args:
|
||||
emails_data: 邮箱数据列表，格式：[{'email': 'xxx@xxx.com', 'is_enabled': True, 'description': '描述'}]
|
||||
"""
|
||||
try:
|
||||
from .models import NotificationEmail
|
||||
|
||||
# 清空现有邮箱
|
||||
NotificationEmail.objects.all().delete()
|
||||
|
||||
# 添加新的邮箱
|
||||
for email_info in emails_data:
|
||||
if isinstance(email_info, str):
|
||||
# 兼容旧格式（纯邮箱字符串）
|
||||
NotificationEmail.objects.create(
|
||||
email=email_info.strip(),
|
||||
is_enabled=True
|
||||
)
|
||||
elif isinstance(email_info, dict):
|
||||
# 新格式（包含详细信息）
|
||||
NotificationEmail.objects.create(
|
||||
email=email_info.get('email', '').strip(),
|
||||
is_enabled=email_info.get('is_enabled', True),
|
||||
description=email_info.get('description', '')
|
||||
)
|
||||
|
||||
logger.info(f"通知邮箱配置已更新: {len(emails_data)} 个邮箱")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"更新通知邮箱配置失败: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def update_global_notification_setting(enabled):
|
||||
"""
|
||||
更新全局邮件通知开关
|
||||
|
||||
Args:
|
||||
enabled: 是否启用邮件通知
|
||||
"""
|
||||
try:
|
||||
from .models import NotificationSettings
|
||||
|
||||
settings = NotificationSettings.get_settings()
|
||||
settings.email_notification_enabled = enabled
|
||||
settings.save()
|
||||
|
||||
logger.info(f"全局邮件通知设置已更新: {enabled}")
|
||||
return True
|
||||
except Exception as e:
|
||||
logger.error(f"更新全局邮件通知设置失败: {e}")
|
||||
return False
|
||||
|
||||
@staticmethod
|
||||
def get_all_notification_emails():
|
||||
"""获取所有通知邮箱（包括禁用的）"""
|
||||
try:
|
||||
from .models import NotificationEmail
|
||||
|
||||
emails = NotificationEmail.objects.all().values(
|
||||
'id', 'email', 'is_enabled', 'description', 'created_at', 'updated_at'
|
||||
)
|
||||
return list(emails)
|
||||
except Exception as e:
|
||||
logger.error(f"获取通知邮箱列表失败: {e}")
|
||||
return []
|
||||
|
||||
@staticmethod
|
||||
def toggle_email_status(email_id, is_enabled):
|
||||
"""
|
||||
切换指定邮箱的启用状态
|
||||
|
||||
Args:
|
||||
email_id: 邮箱ID
|
||||
is_enabled: 是否启用
|
||||
"""
|
||||
try:
|
||||
from .models import NotificationEmail
|
||||
|
||||
email_obj = NotificationEmail.objects.get(id=email_id)
|
||||
email_obj.is_enabled = is_enabled
|
||||
email_obj.save()
|
||||
|
||||
logger.info(f"邮箱 {email_obj.email} 状态已更新: {is_enabled}")
|
||||
return True
|
||||
except NotificationEmail.DoesNotExist:
|
||||
logger.error(f"邮箱ID {email_id} 不存在")
|
||||
return False
|
||||
except Exception as e:
|
||||
logger.error(f"更新邮箱状态失败: {e}")
|
||||
return False
|
||||
@@ -0,0 +1,47 @@
|
||||
from django.test import TestCase
|
||||
from django.core import mail
|
||||
from .models import NotificationEmail, NotificationSettings
|
||||
from .services import EmailNotificationService
|
||||
|
||||
class EmailNotificationTestCase(TestCase):
|
||||
def setUp(self):
|
||||
"""设置测试数据"""
|
||||
self.email1 = NotificationEmail.objects.create(
|
||||
email="test1@example.com",
|
||||
is_enabled=True,
|
||||
description="测试邮箱1"
|
||||
)
|
||||
self.email2 = NotificationEmail.objects.create(
|
||||
email="test2@example.com",
|
||||
is_enabled=False,
|
||||
description="测试邮箱2"
|
||||
)
|
||||
|
||||
def test_get_notification_settings(self):
|
||||
"""测试获取通知设置"""
|
||||
settings = EmailNotificationService.get_notification_settings()
|
||||
self.assertTrue(settings['email_enabled'])
|
||||
self.assertIn("test1@example.com", settings['notification_emails'])
|
||||
self.assertNotIn("test2@example.com", settings['notification_emails'])
|
||||
|
||||
def test_toggle_email_status(self):
|
||||
"""测试切换邮箱状态"""
|
||||
result = EmailNotificationService.toggle_email_status(self.email2.id, True)
|
||||
self.assertTrue(result)
|
||||
|
||||
self.email2.refresh_from_db()
|
||||
self.assertTrue(self.email2.is_enabled)
|
||||
|
||||
def test_update_notification_emails(self):
|
||||
"""测试更新通知邮箱"""
|
||||
emails_data = [
|
||||
{'email': 'new1@example.com', 'is_enabled': True, 'description': '新邮箱1'},
|
||||
{'email': 'new2@example.com', 'is_enabled': False, 'description': '新邮箱2'}
|
||||
]
|
||||
|
||||
result = EmailNotificationService.update_notification_emails(emails_data)
|
||||
self.assertTrue(result)
|
||||
|
||||
# 检查数据库中的邮箱数量
|
||||
self.assertEqual(NotificationEmail.objects.count(), 2)
|
||||
self.assertTrue(NotificationEmail.objects.filter(email='new1@example.com').exists())
|
||||
@@ -0,0 +1,9 @@
|
||||
from django.urls import path
|
||||
from . import views
|
||||
|
||||
app_name = 'email_notice'
|
||||
|
||||
urlpatterns = [
|
||||
path('api/notification-settings/', views.notification_settings, name='notification_settings'),
|
||||
path('api/toggle-email-status/', views.toggle_email_status, name='toggle_email_status'),
|
||||
]
|
||||
@@ -0,0 +1,118 @@
|
||||
from django.http import JsonResponse
|
||||
from django.views.decorators.csrf import csrf_exempt
|
||||
from django.views.decorators.http import require_http_methods
|
||||
from .services import EmailNotificationService
|
||||
import json
|
||||
import re
|
||||
|
||||
@csrf_exempt
|
||||
@require_http_methods(["POST", "GET"])
|
||||
def notification_settings(request):
|
||||
"""通知设置API"""
|
||||
|
||||
if request.method == 'GET':
|
||||
try:
|
||||
settings_data = EmailNotificationService.get_notification_settings()
|
||||
all_emails = EmailNotificationService.get_all_notification_emails()
|
||||
return JsonResponse({
|
||||
'success': True,
|
||||
'data': {
|
||||
'email_enabled': settings_data.get('email_enabled', False),
|
||||
'notification_emails': settings_data.get('notification_emails', []),
|
||||
'all_emails': all_emails
|
||||
}
|
||||
})
|
||||
except Exception as e:
|
||||
return JsonResponse({'success': False, 'error': str(e)}, status=500)
|
||||
|
||||
elif request.method == 'POST':
|
||||
try:
|
||||
data = json.loads(request.body)
|
||||
emails_to_update = data.get('notification_emails')
|
||||
email_enabled_status = data.get('email_enabled')
|
||||
|
||||
# 标志位，用于判断是否执行了任何更新操作
|
||||
was_updated = False
|
||||
|
||||
# 1. 如果请求中包含 'notification_emails' 键，则处理邮箱列表
|
||||
if emails_to_update is not None: # 允许 emails_to_update 为空列表 []
|
||||
email_regex = re.compile(r'^[^\s@]+@[^\s@]+\.[^\s@]+$')
|
||||
valid_emails_data = []
|
||||
|
||||
for email_info in emails_to_update:
|
||||
# 兼容前端可能发送的字符串或对象格式
|
||||
if isinstance(email_info, str):
|
||||
if email_info.strip() and email_regex.match(email_info.strip()):
|
||||
valid_emails_data.append({
|
||||
'email': email_info.strip(),
|
||||
'is_enabled': True,
|
||||
'description': ''
|
||||
})
|
||||
elif isinstance(email_info, dict):
|
||||
email = email_info.get('email', '').strip()
|
||||
if email and email_regex.match(email):
|
||||
valid_emails_data.append({
|
||||
'email': email,
|
||||
'is_enabled': email_info.get('is_enabled', True),
|
||||
'description': email_info.get('description', '')
|
||||
})
|
||||
|
||||
# 调用服务层更新邮箱，服务层需要能处理空列表
|
||||
EmailNotificationService.update_notification_emails(valid_emails_data)
|
||||
was_updated = True
|
||||
|
||||
if email_enabled_status is not None:
|
||||
EmailNotificationService.update_global_notification_setting(email_enabled_status)
|
||||
was_updated = True
|
||||
|
||||
# 执行更新成功，返回成功
|
||||
if was_updated:
|
||||
return JsonResponse({
|
||||
'success': True,
|
||||
'message': '通知设置已更新',
|
||||
# 返回最新的数据状态
|
||||
'data': {
|
||||
'email_enabled': EmailNotificationService.get_notification_settings().get('email_enabled', False),
|
||||
'all_emails': EmailNotificationService.get_all_notification_emails()
|
||||
}
|
||||
})
|
||||
else:
|
||||
# 如果请求体为空或不包含任何有效键，则返回错误
|
||||
return JsonResponse({'success': False, 'error': '未提供任何有效的更新数据'}, status=400)
|
||||
|
||||
except json.JSONDecodeError:
|
||||
return JsonResponse({'success': False, 'error': '无效的JSON数据'}, status=400)
|
||||
except Exception as e:
|
||||
return JsonResponse({'success': False, 'error': str(e)}, status=500)
|
||||
|
||||
return JsonResponse({'success': False, 'error': '不支持的请求方法'}, status=405)
|
||||
|
||||
@csrf_exempt
|
||||
@require_http_methods(["POST"])
|
||||
def toggle_email_status(request):
|
||||
"""切换邮箱启用状态API"""
|
||||
try:
|
||||
data = json.loads(request.body)
|
||||
email_id = data.get('email_id')
|
||||
is_enabled = data.get('is_enabled', True)
|
||||
|
||||
if not email_id:
|
||||
return JsonResponse({'success': False, 'error': '缺少邮箱ID'}, status=400)
|
||||
|
||||
success = EmailNotificationService.toggle_email_status(email_id, is_enabled)
|
||||
|
||||
if success:
|
||||
return JsonResponse({
|
||||
'success': True,
|
||||
'message': f'邮箱状态已更新为{"启用" if is_enabled else "禁用"}',
|
||||
'data': {
|
||||
'all_emails': EmailNotificationService.get_all_notification_emails()
|
||||
}
|
||||
})
|
||||
else:
|
||||
return JsonResponse({'success': False, 'error': '更新邮箱状态失败'}, status=500)
|
||||
|
||||
except json.JSONDecodeError:
|
||||
return JsonResponse({'success': False, 'error': '无效的JSON数据'}, status=400)
|
||||
except Exception as e:
|
||||
return JsonResponse({'success': False, 'error': str(e)}, status=500)
|
||||
Reference in New Issue
Block a user