✅ 邮件发送测试
这是一封通过 Celery异步任务 发送的HTML格式邮件。
任务ID: {task_id}
发送时间: {send_time}
状态: 发送成功
📈 统计信息
阅读记录: {reading_count} 篇
感悟记录: {insights_count} 条
家庭事项: {tasks_count} 项
📋 说明
此邮件由Celery异步任务自动发送。
如需取消订阅，请联系系统管理员。
'''.format(
today=report_data['today'],
task_id=task_id,
send_time=timezone.now().strftime('%Y-%m-%d %H:%M:%S'),
reading_count=report_data['reading']['count'],
insights_count=report_data['insights']['count'],
tasks_count=len(report_data['tasks'])
)
# 创建邮件后端
backend = EmailBackend(
host=host,
port=port,
username=username,
password=password,
use_tls=use_tls,
fail_silently=False
)
# 创建邮件
subject = f"[Celery] 家庭日报 {report_data['today']} - 测试报告"
email = EmailMultiAlternatives(
subject=subject,
body=html_content,
from_email=from_email,
to=recipient_list,
connection=backend
)
email.attach_alternative(html_content, "text/html")
email.encoding = 'utf-8'
# 添加附件（如果需要）
if include_attachment:
attachment_content = '''家庭日报测试报告
发送时间: {send_time}
任务ID: {task_id}
发送方式: Celery异步任务
这是一份自动生成的测试报告
'''.format(
send_time=timezone.now().isoformat(),
task_id=task_id
)
email.attach('daily_report_test.txt', attachment_content, 'text/plain')
logger.info(f"[任务 {task_id}] 已添加测试附件")
# 发送邮件
logger.info(f"[任务 {task_id}] 正在发送HTML报告邮件...")
sent_count = email.send(fail_silently=False)
if sent_count > 0:
logger.success(f"[任务 {task_id}] HTML报告邮件发送成功！")
return {
'status': 'success',
'task_id': task_id,
'type': 'html_report',
'sent_to': len(recipient_list),
'has_attachment': include_attachment,
'timestamp': timezone.now().isoformat()
}
else:
raise Exception("邮件发送返回0")
except Exception as e:
error_msg = str(e)
error_traceback = traceback.format_exc()
logger.error(f"[任务 {task_id}] HTML报告邮件发送失败: {error_msg}")
logger.error(f"[任务 {task_id}] 错误详情:\n{error_traceback}")
return {
'status': 'failed',
'task_id': task_id,
'type': 'html_report',
'error': error_msg,
'timestamp': timezone.now().isoformat()
}
def test_celery_email_task():
"""测试Celery异步邮件任务"""
logger.info("开始测试Celery异步邮件任务...")
try:
# 初始化Django环境
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'diary_family.settings')
import django
django.setup()
from django.utils import timezone
# 发送测试任务
logger.info("发送Celery邮件任务到队列...")
# 发送简单文本邮件任务
task1 = celery_send_test_email.delay(test_mode=True)
logger.info(f"简单邮件任务已发送: {task1.id}")
# 发送HTML报告邮件任务
task2 = celery_send_html_report_email.delay(include_attachment=False)
logger.info(f"HTML报告邮件任务已发送: {task2.id}")
# 等待任务完成
logger.info("等待任务执行（最多30秒）...")
results = []
tasks = [
('简单邮件', task1),
('HTML报告邮件', task2)
]
for name, task in tasks:
try:
# 等待任务完成
result = task.get(timeout=30)
logger.info(f"{name}任务结果: {result}")
results.append((name, 'success', result))
except Exception as e:
logger.warning(f"{name}任务获取结果失败（可能是超时）: {e}")
# 检查任务状态
try:
state = task.state
logger.info(f"{name}任务状态: {state}")
results.append((name, state, None))
except Exception as state_e:
logger.error(f"获取任务状态失败: {state_e}")
results.append((name, 'unknown', None))
# 统计结果
success_count = sum(1 for _, status, _ in results if status == 'success')
logger.info(f"\n任务执行结果: {success_count}/{len(tasks)} 成功")
return success_count == len(tasks)
except Exception as e:
logger.error(f"Celery邮件任务测试失败: {e}")
logger.info("可能的原因:")
logger.info("1. Celery worker未运行")
logger.info("2. Redis连接问题")
logger.info("3. 任务执行超时")
return False
def main():
"""主测试函数"""
logger.info("=" * 60)
logger.info("=== Celery邮件发送功能测试开始 ===")
logger.info("测试环境: Ubuntu + Celery + Redis + SMTP")
logger.info("=" * 60)
tests_passed = 0
total_tests = 4
# 测试1: Celery邮件配置
logger.info("\n[测试1] Celery邮件配置测试")
if test_celery_email_config():
tests_passed += 1
# 测试2: Worker状态
logger.info("\n[测试2] Celery Worker状态检查")
if test_celery_worker_status():
tests_passed += 1
# 测试3: Celery邮件任务
logger.info("\n[测试3] Celery异步邮件任务测试")
if test_celery_email_task():
tests_passed += 1
# 测试4: 发送单个测试邮件（同步）
logger.info("\n[测试4] 同步发送测试邮件")
try:
from django.conf import settings
from django.core.mail import EmailMessage
from django.utils import timezone
from core.models import SystemConfig
from django.core.mail.backends.smtp import EmailBackend
# 从数据库获取邮件配置
config = SystemConfig.get_config()
# 优先使用sender_email，其次使用smtp_username作为发件人
from_email = config.sender_email or config.smtp_username
if not from_email:
raise ValueError("未配置发件邮箱")
to_email = config.recipient_email or from_email
recipient_list = [to_email] if isinstance(to_email, str) else to_email
# 获取SMTP配置
host = config.smtp_server or 'localhost'
port = config.smtp_port or 587
username = config.smtp_username or ''
password = config.smtp_password or ''
use_tls = True # 默认使用TLS
# 创建邮件后端
backend = EmailBackend(
host=host,
port=port,
username=username,
password=password,
use_tls=use_tls,
fail_silently=False
)
subject = "[直接测试] Celery邮件测试 - {}".format(timezone.now().strftime('%H:%M:%S'))
body = '''
这是直接发送的测试邮件，用于验证SMTP配置
发送时间: {send_time}
发送方式: 直接发送（非Celery）
如果Celery异步任务测试失败，但此测试成功
说明SMTP配置正确，问题可能在Celery或Redis配置
---
家庭日报系统
'''.format(
send_time=timezone.now().isoformat()
)
email = EmailMessage(
subject=subject,
body=body,
from_email=from_email,
to=recipient_list,
connection=backend
)
sent = email.send(fail_silently=False)
if sent > 0:
logger.success("同步邮件发送成功！")
tests_passed += 1
else:
logger.error("同步邮件发送失败")
except Exception as e:
logger.error(f"同步邮件测试失败: {e}")
# 测试总结
logger.info("\n" + "=" * 60)
logger.info("测试总结:")
logger.info(f"通过测试: {tests_passed}/{total_tests}")
logger.info("=" * 60)
if tests_passed == total_tests:
logger.success("所有测试通过！Celery邮件功能正常。")
logger.info("\n✅ Celery配置正确")
logger.info("✅ Redis连接正常")
logger.info("✅ SMTP配置正常")
logger.info("✅ 异步邮件任务正常")
logger.info("\n建议:")
logger.info("1. 定期检查Celery worker状态")
logger.info("2. 监控任务队列长度")
logger.info("3. 设置任务超时和重试机制")
logger.info("4. 配置任务监控（如Flower）")
return 0
elif tests_passed >= 2:
logger.warning("部分测试通过，Celery邮件功能基本可用。")
logger.info("\n需要检查:")
logger.info("1. Celery worker是否运行")
logger.info("2. Redis服务是否正常")
logger.info("3. SMTP配置是否正确")
return 1
else:
logger.error("多数测试失败，Celery邮件功能异常。")
logger.info("\n紧急处理:")
logger.info("1. ❌ 检查Celery worker: sudo supervisorctl status celery_worker")
logger.info("2. ❌ 检查Redis: sudo systemctl status redis-server")
logger.info("3. ❌ 检查SMTP配置")
logger.info("4. ❌ 查看日志: tail -f /var/log/celery/worker.log")
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="