diary-family/test_email.py

#!/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 core.models import SystemConfig

        # 从数据库获取系统配置
        config = SystemConfig.get_config()

        # 检查邮件配置
        email_config = {
            'smtp_server': config.smtp_server,
            'smtp_port': config.smtp_port,
            'smtp_username': config.smtp_username,
            'smtp_password': '***' if config.smtp_password else None,
            'recipient_email': config.recipient_email,
            'send_time': config.send_time,
        }

        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 = ['smtp_server', 'smtp_port', 'smtp_username', 'smtp_password']
        missing_configs = []
        for cfg in required_configs:
            if not getattr(config, cfg, None):
                missing_configs.append(cfg)

        if missing_configs:
            logger.error(f"缺少必要的邮件配置: {', '.join(missing_configs)}")
            logger.info("请在系统配置页面配置以下参数:")
            logger.info("  - SMTP服务器: SMTP服务器地址")
            logger.info("  - SMTP端口: SMTP端口（通常是587或465）")
            logger.info("  - SMTP用户名: 发件邮箱")
            logger.info("  - SMTP密码: 发件邮箱密码")
            return False

        # 验证邮箱格式
        import re
        email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        
        # 验证发件人邮箱格式（优先使用sender_email，其次使用smtp_username）
        sender_email = config.sender_email or config.smtp_username
        if sender_email and not re.match(email_pattern, sender_email):
            logger.error(f"发件人邮箱格式不正确: {sender_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址")
            return False
        
        # 验证收件人邮箱格式（如果已配置）
        if config.recipient_email and not re.match(email_pattern, config.recipient_email):
            logger.error(f"收件人邮箱格式不正确: {config.recipient_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址")
            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.core.mail.backends.smtp import EmailBackend
        from core.models import SystemConfig

        # 从数据库获取SMTP配置
        config = SystemConfig.get_config()
        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
        use_ssl = False  # 默认不使用SSL
        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.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()
        from_email = config.sender_email or config.smtp_username or None
        if not from_email:
            logger.error("未配置发件邮箱")
            return False

        # 验证发件人邮箱格式
        import re
        email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_pattern, from_email):
            logger.error(f"发件人邮箱格式不正确: {from_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址")
            return False

        # 获取收件人（如果没有配置，使用发件人自己）
        to_email = config.recipient_email or from_email
        # 验证收件人邮箱格式
        if not re.match(email_pattern, to_email):
            logger.error(f"收件人邮箱格式不正确: {to_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址")
            return False
        recipient_list = [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

        # 创建测试邮件
        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}")

        # 创建邮件后端
        backend = EmailBackend(
            host=host,
            port=port,
            username=username,
            password=password,
            use_tls=use_tls,
            fail_silently=False
        )

        # 创建邮件
        email = EmailMessage(
            subject=subject,
            body=body,
            from_email=from_email,
            to=recipient_list,
            connection=backend
        )
        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.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()
        from_email = config.smtp_username or None
        if not from_email:
            logger.error("未配置发件邮箱")
            return False

        # 验证发件人邮箱格式
        import re
        email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_pattern, from_email):
            logger.error(f"发件人邮箱格式不正确: {from_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址，而不是手机号或其他格式")
            return False

        # 获取收件人（如果没有配置，使用发件人自己）
        to_email = config.recipient_email or from_email
        # 验证收件人邮箱格式
        if not re.match(email_pattern, to_email):
            logger.error(f"收件人邮箱格式不正确: {to_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址")
            return False
        recipient_list = [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

        # 创建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')}"

        # 创建邮件后端
        backend = EmailBackend(
            host=host,
            port=port,
            username=username,
            password=password,
            use_tls=use_tls,
            fail_silently=False
        )

        # 创建邮件
        email = EmailMessage(
            subject=subject,
            body=html_content,
            from_email=from_email,
            to=recipient_list,
            connection=backend
        )
        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.core.mail import EmailMessage
        from django.utils import timezone
        from core.models import SystemConfig
        from django.core.mail.backends.smtp import EmailBackend
        import time

        # 从数据库获取配置
        config = SystemConfig.get_config()
        from_email = config.smtp_username or None
        if not from_email:
            logger.error("未配置发件邮箱")
            return False

        # 验证发件人邮箱格式
        import re
        email_pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(email_pattern, from_email):
            logger.error(f"发件人邮箱格式不正确: {from_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址，而不是手机号或其他格式")
            return False

        # 获取收件人（如果没有配置，使用发件人自己）
        to_email = config.recipient_email or from_email
        # 验证收件人邮箱格式
        if not re.match(email_pattern, to_email):
            logger.error(f"收件人邮箱格式不正确: {to_email}")
            logger.error("请在系统配置页面输入有效的邮箱地址")
            return False
        recipient_list = [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
        )

        # 性能测试
        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,
                connection=backend
            )

            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)