xiaji
cd5fc2ca11
添加两个测试脚本： 1. test_email.py - 验证SMTP邮件发送功能 2. test_celery_email.py - 测试Celery异步邮件任务 同时更新.gitignore以包含新的测试文件
550 lines
19 KiB
Python
550 lines
19 KiB
Python
#!/usr/bin/env python
|
||
"""
|
||
邮件发送测试脚本
|
||
用于验证生产环境邮件配置是否正确
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
import time
|
||
from pathlib import Path
|
||
from loguru import logger
|
||
|
||
|
||
def test_email_config():
|
||
"""测试邮件配置是否正确"""
|
||
logger.info("开始测试邮件配置...")
|
||
try:
|
||
# 初始化Django环境
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
|
||
import django
|
||
django.setup()
|
||
|
||
from django.conf import settings
|
||
from django.core.mail import get_connection
|
||
from email.mime.text import MIMEText
|
||
from email.mime.multipart import MIMEMultipart
|
||
|
||
# 检查邮件配置
|
||
email_config = {
|
||
'EMAIL_BACKEND': getattr(settings, 'EMAIL_BACKEND', None),
|
||
'EMAIL_HOST': getattr(settings, 'EMAIL_HOST', None),
|
||
'EMAIL_PORT': getattr(settings, 'EMAIL_PORT', None),
|
||
'EMAIL_USE_TLS': getattr(settings, 'EMAIL_USE_TLS', None),
|
||
'EMAIL_USE_SSL': getattr(settings, 'EMAIL_USE_SSL', None),
|
||
'EMAIL_HOST_USER': getattr(settings, 'EMAIL_HOST_USER', None),
|
||
'EMAIL_HOST_PASSWORD': '***' if getattr(settings, 'EMAIL_HOST_PASSWORD', None) else None,
|
||
}
|
||
|
||
logger.info("邮件配置信息:")
|
||
for key, value in email_config.items():
|
||
if value is None:
|
||
logger.warning(f" {key}: 未配置")
|
||
else:
|
||
logger.info(f" {key}: {value}")
|
||
|
||
# 验证必要配置
|
||
required_configs = ['EMAIL_HOST', 'EMAIL_PORT', 'EMAIL_HOST_USER']
|
||
missing_configs = [cfg for cfg in required_configs if not getattr(settings, cfg, None)]
|
||
|
||
if missing_configs:
|
||
logger.error(f"缺少必要的邮件配置: {', '.join(missing_configs)}")
|
||
logger.info("请在系统配置页面或settings.py中配置以下参数:")
|
||
logger.info(" - EMAIL_HOST: SMTP服务器地址")
|
||
logger.info(" - EMAIL_PORT: SMTP端口（通常是587或465）")
|
||
logger.info(" - EMAIL_HOST_USER: 发件邮箱")
|
||
logger.info(" - EMAIL_HOST_PASSWORD: 发件邮箱密码")
|
||
return False
|
||
|
||
logger.success("邮件配置测试通过！")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"邮件配置测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_smtp_connection():
|
||
"""测试SMTP连接"""
|
||
logger.info("开始测试SMTP连接...")
|
||
try:
|
||
# 初始化Django环境
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
|
||
import django
|
||
django.setup()
|
||
|
||
from django.conf import settings
|
||
from django.core.mail import get_connection
|
||
from django.core.mail.backends.smtp import EmailBackend
|
||
|
||
# 获取SMTP配置
|
||
host = getattr(settings, 'EMAIL_HOST', 'localhost')
|
||
port = getattr(settings, 'EMAIL_PORT', 587)
|
||
username = getattr(settings, 'EMAIL_HOST_USER', '')
|
||
password = getattr(settings, 'EMAIL_HOST_PASSWORD', '')
|
||
use_tls = getattr(settings, 'EMAIL_USE_TLS', True)
|
||
use_ssl = getattr(settings, 'EMAIL_USE_SSL', False)
|
||
timeout = 10 # 连接超时时间
|
||
|
||
logger.info(f"连接SMTP服务器: {host}:{port}")
|
||
logger.info(f"使用TLS: {use_tls}, 使用SSL: {use_ssl}")
|
||
|
||
# 创建邮件后端
|
||
backend = EmailBackend(
|
||
host=host,
|
||
port=port,
|
||
username=username,
|
||
password=password,
|
||
use_tls=use_tls,
|
||
use_ssl=use_ssl,
|
||
timeout=timeout,
|
||
fail_silently=False
|
||
)
|
||
|
||
# 测试连接
|
||
logger.info("正在建立SMTP连接...")
|
||
connection = backend.open()
|
||
|
||
if connection:
|
||
logger.success(f"SMTP连接成功！服务器: {host}:{port}")
|
||
logger.info(f"连接状态: {connection}")
|
||
|
||
# 获取服务器信息
|
||
try:
|
||
# 某些SMTP服务器支持EHLO/HELO命令获取信息
|
||
logger.info("SMTP连接已建立，准备发送测试邮件")
|
||
except Exception as e:
|
||
logger.warning(f"无法获取服务器详细信息: {e}")
|
||
|
||
# 关闭连接
|
||
backend.close()
|
||
return True
|
||
else:
|
||
logger.error("SMTP连接失败，未能建立连接")
|
||
return False
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
logger.error(f"SMTP连接测试失败: {error_msg}")
|
||
|
||
# 提供详细的错误诊断
|
||
if "connection refused" in error_msg.lower():
|
||
logger.error("🔌 连接被拒绝，可能的原因:")
|
||
logger.error(" 1. SMTP服务器地址错误或不可用")
|
||
logger.error(" 2. SMTP端口被防火墙阻止")
|
||
logger.error(" 3. SMTP服务器未运行")
|
||
logger.error("")
|
||
logger.error("💡 解决方案:")
|
||
logger.error(" 1. 检查SMTP服务器地址是否正确")
|
||
logger.error(" 2. 确认端口号（常用端口: 587 for TLS, 465 for SSL）")
|
||
logger.error(" 3. 检查防火墙设置: sudo ufw status")
|
||
logger.error(" 4. 测试网络连通性: telnet smtp.example.com 587")
|
||
|
||
elif "authentication failed" in error_msg.lower() or "535" in error_msg:
|
||
logger.error("🔐 认证失败，可能的原因:")
|
||
logger.error(" 1. 用户名或密码错误")
|
||
logger.error(" 2. 邮箱账号被禁用或锁定")
|
||
logger.error(" 3. 需要使用应用专用密码（如 Gmail）")
|
||
logger.error("")
|
||
logger.error("💡 解决方案:")
|
||
logger.error(" 1. 检查用户名和密码是否正确")
|
||
logger.error(" 2. 如果是Gmail，检查是否开启了两步验证")
|
||
logger.error(" 3. 如果是Gmail，使用应用专用密码而不是登录密码")
|
||
logger.error(" 4. 检查邮箱是否允许SMTP访问")
|
||
|
||
elif "timeout" in error_msg.lower():
|
||
logger.error("⏱️ 连接超时，可能的原因:")
|
||
logger.error(" 1. SMTP服务器地址不可达")
|
||
logger.error(" 2. 网络连接问题")
|
||
logger.error(" 3. SMTP服务器响应过慢")
|
||
logger.error("")
|
||
logger.error("💡 解决方案:")
|
||
logger.error(" 1. 检查网络连接: ping smtp.example.com")
|
||
logger.error(" 2. 检查DNS解析: nslookup smtp.example.com")
|
||
logger.error(" 3. 尝试使用IP地址直接连接")
|
||
logger.error(" 4. 联系网络管理员检查网络设置")
|
||
|
||
else:
|
||
logger.error("请检查:")
|
||
logger.error(" 1. SMTP服务器配置是否正确")
|
||
logger.error(" 2. 邮箱账号和密码是否正确")
|
||
logger.error(" 3. 网络连接是否正常")
|
||
logger.error(" 4. 防火墙是否允许出站连接")
|
||
|
||
return False
|
||
|
||
|
||
def test_send_simple_email():
|
||
"""测试发送简单邮件"""
|
||
logger.info("开始测试发送简单邮件...")
|
||
try:
|
||
# 初始化Django环境
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
|
||
import django
|
||
django.setup()
|
||
|
||
from django.conf import settings
|
||
from django.core.mail import EmailMessage
|
||
from django.utils import timezone
|
||
|
||
# 获取配置
|
||
from_email = getattr(settings, 'EMAIL_HOST_USER', None)
|
||
if not from_email:
|
||
logger.error("未配置发件邮箱 (EMAIL_HOST_USER)")
|
||
return False
|
||
|
||
# 获取收件人（如果没有配置，使用发件人自己）
|
||
to_email = getattr(settings, 'EMAIL_HOST_USER', from_email)
|
||
if isinstance(to_email, list):
|
||
recipient_list = to_email
|
||
else:
|
||
recipient_list = [to_email]
|
||
|
||
# 获取SMTP配置
|
||
host = getattr(settings, 'EMAIL_HOST', 'localhost')
|
||
port = getattr(settings, 'EMAIL_PORT', 587)
|
||
username = getattr(settings, 'EMAIL_HOST_USER', '')
|
||
password = getattr(settings, 'EMAIL_HOST_PASSWORD', '')
|
||
use_tls = getattr(settings, 'EMAIL_USE_TLS', True)
|
||
|
||
# 创建测试邮件
|
||
subject = f"家庭日报系统测试邮件 - {timezone.now().strftime('%Y-%m-%d %H:%M:%S')}"
|
||
body = f"""
|
||
这是一封测试邮件，用于验证家庭日报系统的邮件发送功能。
|
||
|
||
测试时间: {timezone.now().strftime('%Y-%m-%d %H:%M:%S')}
|
||
发件服务器: {host}:{port}
|
||
使用TLS: {use_tls}
|
||
|
||
如果收到此邮件，说明邮件系统配置正确！
|
||
|
||
---
|
||
家庭日报系统
|
||
自动发送
|
||
"""
|
||
|
||
logger.info(f"准备发送邮件:")
|
||
logger.info(f" 发件人: {from_email}")
|
||
logger.info(f" 收件人: {recipient_list}")
|
||
logger.info(f" 主题: {subject}")
|
||
|
||
# 创建邮件
|
||
email = EmailMessage(
|
||
subject=subject,
|
||
body=body,
|
||
from_email=from_email,
|
||
to=recipient_list,
|
||
)
|
||
email.content_subtype = 'plain'
|
||
email.encoding = 'utf-8'
|
||
|
||
# 发送邮件
|
||
logger.info("正在发送邮件...")
|
||
sent_count = email.send(fail_silently=False)
|
||
|
||
if sent_count > 0:
|
||
logger.success(f"邮件发送成功！发送给 {len(recipient_list)} 个收件人")
|
||
logger.info(f"邮件ID: {email.message().get('Message-ID', '未知')}")
|
||
return True
|
||
else:
|
||
logger.error("邮件发送失败，未能发送任何邮件")
|
||
return False
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
logger.error(f"发送邮件失败: {error_msg}")
|
||
|
||
# 特定错误处理
|
||
if "SMTP AUTH required" in error_msg or "535" in error_msg:
|
||
logger.error("需要SMTP认证，请检查用户名和密码配置")
|
||
elif "SMTP server rejected" in error_msg or "550" in error_msg:
|
||
logger.error("邮件被服务器拒绝，可能的原因:")
|
||
logger.error(" 1. 收件人地址不存在")
|
||
logger.error(" 2. 发件人邮箱被列入黑名单")
|
||
logger.error(" 3. 邮件内容被判定为垃圾邮件")
|
||
elif "Connection refused" in error_msg:
|
||
logger.error("无法连接到SMTP服务器，请检查服务器地址和端口")
|
||
else:
|
||
logger.error("发送邮件时发生未知错误")
|
||
|
||
return False
|
||
|
||
|
||
def test_send_html_email_with_attachment():
|
||
"""测试发送HTML邮件（带附件）"""
|
||
logger.info("开始测试发送HTML邮件（带附件）...")
|
||
try:
|
||
# 初始化Django环境
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
|
||
import django
|
||
django.setup()
|
||
|
||
from django.conf import settings
|
||
from django.core.mail import EmailMessage
|
||
from django.template.loader import render_to_string
|
||
from django.utils import timezone
|
||
|
||
# 获取配置
|
||
from_email = getattr(settings, 'EMAIL_HOST_USER', None)
|
||
to_email = getattr(settings, 'EMAIL_HOST_USER', from_email)
|
||
if isinstance(to_email, list):
|
||
recipient_list = to_email
|
||
else:
|
||
recipient_list = [to_email]
|
||
|
||
# 创建HTML内容
|
||
html_content = f"""
|
||
<!DOCTYPE html>
|
||
<html>
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<style>
|
||
body {{ font-family: Arial, sans-serif; line-height: 1.6; color: #333; }}
|
||
.container {{ max-width: 600px; margin: 0 auto; padding: 20px; }}
|
||
.header {{ background-color: #4CAF50; color: white; padding: 20px; text-align: center; }}
|
||
.content {{ padding: 20px; background-color: #f9f9f9; }}
|
||
.footer {{ text-align: center; padding: 20px; color: #666; font-size: 12px; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<div class="container">
|
||
<div class="header">
|
||
<h1>📋 家庭日报系统</h1>
|
||
<p>HTML邮件测试</p>
|
||
</div>
|
||
<div class="content">
|
||
<h2>测试成功！</h2>
|
||
<p>这是一封HTML格式的测试邮件。</p>
|
||
<p><strong>测试时间:</strong> {timezone.now().strftime('%Y-%m-%d %H:%M:%S')}</p>
|
||
<p><strong>状态:</strong> ✅ 邮件发送功能正常</p>
|
||
</div>
|
||
<div class="footer">
|
||
<p>这是自动发送的测试邮件，请勿回复。</p>
|
||
</div>
|
||
</div>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
subject = f"家庭日报系统 HTML测试邮件 - {timezone.now().strftime('%Y-%m-%d')}"
|
||
|
||
# 创建邮件
|
||
email = EmailMessage(
|
||
subject=subject,
|
||
body=html_content,
|
||
from_email=from_email,
|
||
to=recipient_list,
|
||
)
|
||
email.content_subtype = 'html'
|
||
email.encoding = 'utf-8'
|
||
|
||
# 添加测试附件
|
||
attachment_content = f"家庭日报系统测试附件\n测试时间: {timezone.now().isoformat()}\n"
|
||
email.attach('test_attachment.txt', attachment_content, 'text/plain')
|
||
|
||
logger.info(f"准备发送HTML邮件给 {len(recipient_list)} 个收件人")
|
||
|
||
# 发送邮件
|
||
sent_count = email.send(fail_silently=False)
|
||
|
||
if sent_count > 0:
|
||
logger.success(f"HTML邮件发送成功！包含 1 个文本附件")
|
||
return True
|
||
else:
|
||
logger.error("HTML邮件发送失败")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"发送HTML邮件失败: {e}")
|
||
logger.warning("HTML邮件测试失败，但基础邮件功能可能正常")
|
||
return False
|
||
|
||
|
||
def test_email_performance():
|
||
"""测试邮件发送性能"""
|
||
logger.info("开始测试邮件发送性能...")
|
||
try:
|
||
# 初始化Django环境
|
||
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
|
||
import django
|
||
django.setup()
|
||
|
||
from django.conf import settings
|
||
from django.core.mail import EmailMessage
|
||
from django.utils import timezone
|
||
import time
|
||
|
||
# 获取配置
|
||
from_email = getattr(settings, 'EMAIL_HOST_USER', None)
|
||
to_email = getattr(settings, 'EMAIL_HOST_USER', from_email)
|
||
if isinstance(to_email, list):
|
||
recipient_list = to_email
|
||
else:
|
||
recipient_list = [to_email]
|
||
|
||
# 性能测试
|
||
test_count = 3
|
||
times = []
|
||
|
||
for i in range(test_count):
|
||
subject = f"性能测试邮件 #{i+1} - {timezone.now().strftime('%H:%M:%S')}"
|
||
body = f"性能测试邮件 #{i+1}\n测试时间: {timezone.now().isoformat()}"
|
||
|
||
email = EmailMessage(
|
||
subject=subject,
|
||
body=body,
|
||
from_email=from_email,
|
||
to=recipient_list,
|
||
)
|
||
|
||
start_time = time.time()
|
||
sent_count = email.send(fail_silently=False)
|
||
elapsed_time = time.time() - start_time
|
||
|
||
times.append(elapsed_time)
|
||
|
||
if sent_count > 0:
|
||
logger.info(f" 邮件 #{i+1}: 发送成功，耗时 {elapsed_time:.3f}秒")
|
||
else:
|
||
logger.error(f" 邮件 #{i+1}: 发送失败")
|
||
return False
|
||
|
||
# 计算平均时间
|
||
avg_time = sum(times) / len(times)
|
||
min_time = min(times)
|
||
max_time = max(times)
|
||
|
||
logger.info(f"\n性能测试结果:")
|
||
logger.info(f" 发送邮件数: {test_count}")
|
||
logger.info(f" 平均耗时: {avg_time:.3f}秒")
|
||
logger.info(f" 最快耗时: {min_time:.3f}秒")
|
||
logger.info(f" 最慢耗时: {max_time:.3f}秒")
|
||
|
||
# 性能评估
|
||
if avg_time < 2:
|
||
logger.success("邮件发送性能优秀！")
|
||
elif avg_time < 5:
|
||
logger.info("邮件发送性能良好")
|
||
elif avg_time < 10:
|
||
logger.warning("邮件发送性能一般，建议优化")
|
||
else:
|
||
logger.error("邮件发送性能较差，请检查SMTP服务器")
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"性能测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
logger.info("=" * 60)
|
||
logger.info("=== 邮件发送功能测试开始 ===")
|
||
logger.info("测试环境: Ubuntu + Django Email + SMTP")
|
||
logger.info("=" * 60)
|
||
|
||
tests_passed = 0
|
||
total_tests = 5
|
||
|
||
# 测试1: 邮件配置
|
||
logger.info("\n[测试1] 邮件配置测试")
|
||
if test_email_config():
|
||
tests_passed += 1
|
||
|
||
# 测试2: SMTP连接
|
||
logger.info("\n[测试2] SMTP连接测试")
|
||
if test_smtp_connection():
|
||
tests_passed += 1
|
||
|
||
# 测试3: 发送简单邮件
|
||
logger.info("\n[测试3] 发送简单文本邮件")
|
||
if test_send_simple_email():
|
||
tests_passed += 1
|
||
|
||
# 测试4: 发送HTML邮件
|
||
logger.info("\n[测试4] 发送HTML邮件（带附件）")
|
||
if test_send_html_email_with_attachment():
|
||
tests_passed += 1
|
||
|
||
# 测试5: 性能测试
|
||
logger.info("\n[测试5] 邮件发送性能测试")
|
||
if test_email_performance():
|
||
tests_passed += 1
|
||
|
||
# 测试总结
|
||
logger.info("\n" + "=" * 60)
|
||
logger.info("测试总结:")
|
||
logger.info(f"通过测试: {tests_passed}/{total_tests}")
|
||
logger.info("=" * 60)
|
||
|
||
if tests_passed == total_tests:
|
||
logger.success("所有测试通过！邮件系统配置正确。")
|
||
logger.info("\n✅ 邮件系统可以正常工作")
|
||
logger.info("✅ SMTP连接正常")
|
||
logger.info("✅ 邮件发送功能正常")
|
||
logger.info("✅ HTML邮件格式正常")
|
||
logger.info("✅ 附件功能正常")
|
||
logger.info("\n建议:")
|
||
logger.info("1. 定期检查SMTP服务器状态")
|
||
logger.info("2. 监控邮件发送成功率")
|
||
logger.info("3. 注意邮箱的发送限制（每日发送上限）")
|
||
return 0
|
||
elif tests_passed >= 3:
|
||
logger.warning("部分测试通过，邮件系统基本可用。")
|
||
logger.info("\n需要检查:")
|
||
if tests_passed < 5:
|
||
logger.info("1. 检查失败的测试项")
|
||
logger.info("2. 参考错误信息进行排查")
|
||
logger.info("3. 确认收件箱是否收到测试邮件")
|
||
return 1
|
||
else:
|
||
logger.error("多数测试失败，邮件系统无法正常工作。")
|
||
logger.info("\n紧急处理:")
|
||
logger.info("1. ❌ 检查SMTP服务器配置")
|
||
logger.info("2. ❌ 验证邮箱账号和密码")
|
||
logger.info("3. ❌ 检查网络连接")
|
||
logger.info("4. 参考README中的邮件配置章节")
|
||
logger.info("5. 查看详细错误日志")
|
||
return 1
|
||
|
||
|
||
if __name__ == "__main__":
|
||
# 配置日志
|
||
logger.remove()
|
||
|
||
# 创建日志目录
|
||
log_dir = Path("/var/log/celery")
|
||
log_dir.mkdir(parents=True, exist_ok=True)
|
||
log_dir.chmod(0o755)
|
||
|
||
# 添加控制台输出
|
||
logger.add(
|
||
sys.stdout,
|
||
format="<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
|
||
level="INFO"
|
||
)
|
||
|
||
# 添加日志文件输出
|
||
log_file = log_dir / "test_email.log"
|
||
logger.add(
|
||
log_file,
|
||
format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{function}:{line} - {message}",
|
||
level="INFO",
|
||
rotation="1 day",
|
||
retention="7 days",
|
||
encoding="utf-8"
|
||
)
|
||
|
||
logger.info(f"邮件测试日志将同时输出到控制台和 {log_file}")
|
||
logger.info("=" * 60)
|
||
|
||
try:
|
||
exit_code = main()
|
||
sys.exit(exit_code)
|
||
except KeyboardInterrupt:
|
||
logger.warning("测试被用户中断")
|
||
sys.exit(1)
|
||
except Exception as e:
|
||
logger.error(f"测试过程中发生未预期错误: {e}")
|
||
sys.exit(1) |