Django

pip install emailverify

# settings.py
EMAILVERIFY_API_KEY = os.environ.get('EMAILVERIFY_API_KEY')

# 可选设置
EMAILVERIFY_TIMEOUT = 10  # 秒
EMAILVERIFY_BLOCK_DISPOSABLE = True
EMAILVERIFY_BLOCK_ROLE_BASED = False
EMAILVERIFY_CACHE_DURATION = 3600  # 1 小时

# validators.py
from django.core.exceptions import ValidationError
from django.conf import settings
from emailverify import Client
import logging

logger = logging.getLogger(__name__)

def validate_email_deliverable(email):
    """验证电子邮件地址是否可递送。"""
    client = Client(api_key=settings.EMAILVERIFY_API_KEY)

    try:
        result = client.verify(email)

        if result.status == 'invalid':
            raise ValidationError(
                '请输入有效的电子邮件地址。',
                code='invalid_email'
            )

        if getattr(settings, 'EMAILVERIFY_BLOCK_DISPOSABLE', True):
            if result.result.disposable:
                raise ValidationError(
                    '不允许使用一次性电子邮件地址。',
                    code='disposable_email'
                )

        if getattr(settings, 'EMAILVERIFY_BLOCK_ROLE_BASED', False):
            if result.result.role:
                raise ValidationError(
                    '不允许使用基于角色的电子邮件地址。',
                    code='role_email'
                )

    except Exception as e:
        # 记录错误但不阻止提交
        logger.warning(f'电子邮件验证失败 {email}：{e}')


class EmailDeliverableValidator:
    """具有可定制选项的基于类的验证器。"""

    def __init__(self, block_disposable=True, block_role_based=False):
        self.block_disposable = block_disposable
        self.block_role_based = block_role_based
        self.client = Client(api_key=settings.EMAILVERIFY_API_KEY)

    def __call__(self, email):
        try:
            result = self.client.verify(email)

            if result.status == 'invalid':
                raise ValidationError(
                    '请输入有效的电子邮件地址。',
                    code='invalid_email'
                )

            if self.block_disposable and result.result.disposable:
                raise ValidationError(
                    '不允许使用一次性电子邮件地址。',
                    code='disposable_email'
                )

            if self.block_role_based and result.result.role:
                raise ValidationError(
                    '不允许使用基于角色的电子邮件地址。',
                    code='role_email'
                )

        except ValidationError:
            raise
        except Exception as e:
            logger.warning(f'电子邮件验证失败：{e}')

    def __eq__(self, other):
        return (
            isinstance(other, EmailDeliverableValidator) and
            self.block_disposable == other.block_disposable and
            self.block_role_based == other.block_role_based
        )

# models.py
from django.db import models
from .validators import validate_email_deliverable

class User(models.Model):
    email = models.EmailField(
        unique=True,
        validators=[validate_email_deliverable]
    )
    name = models.CharField(max_length=255)
    email_verification_status = models.CharField(
        max_length=20,
        blank=True,
        null=True
    )
    email_verification_score = models.FloatField(
        blank=True,
        null=True
    )
    email_verified_at = models.DateTimeField(
        blank=True,
        null=True
    )

    class Meta:
        db_table = 'users'

# forms.py
from django import forms
from django.core.validators import EmailValidator
from .validators import EmailDeliverableValidator

class RegistrationForm(forms.Form):
    name = forms.CharField(max_length=255)
    email = forms.EmailField(
        validators=[
            EmailValidator(),
            EmailDeliverableValidator(
                block_disposable=True,
                block_role_based=False
            )
        ]
    )
    password = forms.CharField(widget=forms.PasswordInput)
    password_confirm = forms.CharField(widget=forms.PasswordInput)

    def clean(self):
        cleaned_data = super().clean()
        password = cleaned_data.get('password')
        password_confirm = cleaned_data.get('password_confirm')

        if password and password_confirm and password != password_confirm:
            raise forms.ValidationError('密码不匹配。')

        return cleaned_data

# forms.py
from django import forms
from .models import User
from .validators import validate_email_deliverable

class UserForm(forms.ModelForm):
    class Meta:
        model = User
        fields = ['name', 'email']

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        # 向电子邮件字段添加验证器
        self.fields['email'].validators.append(validate_email_deliverable)

# views.py
from django.shortcuts import render, redirect
from django.contrib import messages
from django.views import View
from emailverify import Client
from django.conf import settings
from .forms import RegistrationForm
from .models import User

class RegistrationView(View):
    template_name = 'registration/register.html'

    def get(self, request):
        form = RegistrationForm()
        return render(request, self.template_name, {'form': form})

    def post(self, request):
        form = RegistrationForm(request.POST)

        if form.is_valid():
            # 使用完整结果进行额外验证
            client = Client(api_key=settings.EMAILVERIFY_API_KEY)
            email = form.cleaned_data['email']

            try:
                result = client.verify(email)

                # 创建带有验证数据的用户
                user = User.objects.create(
                    name=form.cleaned_data['name'],
                    email=email,
                    email_verification_status=result.status,
                    email_verification_score=result.score,
                )
                user.set_password(form.cleaned_data['password'])
                user.save()

                messages.success(request, '注册成功！')
                return redirect('login')

            except Exception as e:
                messages.error(request, '注册失败。请重试。')

        return render(request, self.template_name, {'form': form})

# middleware.py
from django.http import JsonResponse
from django.conf import settings
from emailverify import Client
import logging
import json

logger = logging.getLogger(__name__)

class EmailVerificationMiddleware:
    """在 POST 请求中验证电子邮件地址的中间件。"""

    def __init__(self, get_response):
        self.get_response = get_response
        self.client = Client(api_key=settings.EMAILVERIFY_API_KEY)
        self.protected_paths = getattr(
            settings,
            'EMAILVERIFY_PROTECTED_PATHS',
            ['/api/register/', '/api/subscribe/']
        )

    def __call__(self, request):
        if request.method == 'POST' and self._should_verify(request.path):
            email = self._extract_email(request)

            if email:
                try:
                    result = self.client.verify(email)

                    if result.status == 'invalid':
                        return JsonResponse(
                            {'error': '无效的电子邮件地址'},
                            status=400
                        )

                    # 将验证结果附加到请求
                    request.email_verification = result

                except Exception as e:
                    logger.warning(f'电子邮件验证失败：{e}')

        return self.get_response(request)

    def _should_verify(self, path):
        return any(path.startswith(p) for p in self.protected_paths)

    def _extract_email(self, request):
        # 尝试 JSON 正文
        if request.content_type == 'application/json':
            try:
                data = json.loads(request.body)
                return data.get('email')
            except json.JSONDecodeError:
                pass

        # 尝试 POST 数据
        return request.POST.get('email')

# settings.py
MIDDLEWARE = [
    # ... 其他中间件
    'myapp.middleware.EmailVerificationMiddleware',
]

EMAILVERIFY_PROTECTED_PATHS = [
    '/api/register/',
    '/api/subscribe/',
    '/api/contact/',
]

# services.py
from django.core.cache import cache
from django.conf import settings
from emailverify import Client
import hashlib

class EmailVerificationService:
    def __init__(self):
        self.client = Client(api_key=settings.EMAILVERIFY_API_KEY)
        self.cache_duration = getattr(
            settings,
            'EMAILVERIFY_CACHE_DURATION',
            3600
        )

    def verify(self, email):
        cache_key = self._get_cache_key(email)
        cached = cache.get(cache_key)

        if cached:
            return cached

        result = self.client.verify(email)
        cache.set(cache_key, result, self.cache_duration)

        return result

    def is_valid(self, email):
        result = self.verify(email)
        return result.status == 'valid'

    def is_disposable(self, email):
        result = self.verify(email)
        return result.result.disposable

    def clear_cache(self, email):
        cache.delete(self._get_cache_key(email))

    def _get_cache_key(self, email):
        email_hash = hashlib.md5(email.lower().encode()).hexdigest()
        return f'email_verification:{email_hash}'


# 使用示例
verification_service = EmailVerificationService()

if verification_service.is_valid(email):
    # 处理有效的电子邮件
    pass

# serializers.py
from rest_framework import serializers
from django.conf import settings
from emailverify import Client

class RegistrationSerializer(serializers.Serializer):
    name = serializers.CharField(max_length=255)
    email = serializers.EmailField()
    password = serializers.CharField(write_only=True)

    def validate_email(self, value):
        client = Client(api_key=settings.EMAILVERIFY_API_KEY)

        try:
            result = client.verify(value)

            if result.status == 'invalid':
                raise serializers.ValidationError(
                    '请输入有效的电子邮件地址。'
                )

            if result.result.disposable:
                raise serializers.ValidationError(
                    '不允许使用一次性电子邮件。'
                )

            # 为以后使用存储结果
            self.context['email_verification'] = result

        except serializers.ValidationError:
            raise
        except Exception as e:
            # 记录但不阻止
            pass

        return value

# permissions.py
from rest_framework.permissions import BasePermission
from django.conf import settings
from emailverify import Client

class ValidEmailPermission(BasePermission):
    """仅允许具有有效电子邮件地址的请求。"""

    message = '需要有效的电子邮件地址。'

    def has_permission(self, request, view):
        email = request.data.get('email')

        if not email:
            return True  # 让其他验证器处理缺少的电子邮件

        client = Client(api_key=settings.EMAILVERIFY_API_KEY)

        try:
            result = client.verify(email)
            return result.status != 'invalid'
        except Exception:
            return True  # 在 API 错误时允许

# management/commands/verify_emails.py
from django.core.management.base import BaseCommand
from django.conf import settings
from emailverify import Client
from myapp.models import User
import time

class Command(BaseCommand):
    help = '验证所有用户电子邮件地址'

    def add_arguments(self, parser):
        parser.add_argument(
            '--batch-size',
            type=int,
            default=100,
            help='每个批次的电子邮件数'
        )

    def handle(self, *args, **options):
        client = Client(api_key=settings.EMAILVERIFY_API_KEY)
        batch_size = options['batch_size']

        users = User.objects.filter(
            email_verification_status__isnull=True
        )

        total = users.count()
        self.stdout.write(f'正在验证 {total} 个电子邮件...')

        # 收集用于批量验证的电子邮件
        emails = list(users.values_list('email', flat=True)[:10000])

        # 提交批量作业
        job = client.verify_bulk(emails)
        self.stdout.write(f'作业 ID：{job.job_id}')

        # 等待完成
        while True:
            status = client.get_bulk_job_status(job.job_id)
            self.stdout.write(
                f'进度：{status.progress_percent}%',
                ending='\r'
            )

            if status.status == 'completed':
                break

            time.sleep(5)

        self.stdout.write('')

        # 获取并处理结果
        results = client.get_bulk_job_results(job.job_id)

        for result in results.results:
            User.objects.filter(email=result.email).update(
                email_verification_status=result.status,
                email_verification_score=result.score,
            )

        self.stdout.write(self.style.SUCCESS('验证完成！'))

        # 打印摘要
        summary = {}
        for result in results.results:
            summary[result.status] = summary.get(result.status, 0) + 1

        for status, count in summary.items():
            self.stdout.write(f'  {status}: {count}')

# signals.py
from django.db.models.signals import pre_save
from django.dispatch import receiver
from django.conf import settings
from emailverify import Client
from .models import User
import logging

logger = logging.getLogger(__name__)

@receiver(pre_save, sender=User)
def verify_user_email(sender, instance, **kwargs):
    """在保存用户之前验证电子邮件。"""
    # 如果电子邮件未更改，则跳过
    if instance.pk:
        try:
            old_instance = User.objects.get(pk=instance.pk)
            if old_instance.email == instance.email:
                return
        except User.DoesNotExist:
            pass

    client = Client(api_key=settings.EMAILVERIFY_API_KEY)

    try:
        result = client.verify(instance.email)
        instance.email_verification_status = result.status
        instance.email_verification_score = result.score

    except Exception as e:
        logger.warning(f'电子邮件验证失败：{e}')

# tests.py
from django.test import TestCase
from unittest.mock import patch, MagicMock
from .validators import validate_email_deliverable
from django.core.exceptions import ValidationError

class EmailValidationTest(TestCase):
    @patch('validators.Client')
    def test_valid_email_passes(self, mock_client):
        mock_result = MagicMock()
        mock_result.status = 'valid'
        mock_result.result.disposable = False
        mock_result.result.role = False

        mock_client.return_value.verify.return_value = mock_result

        # 不应该抛出异常
        validate_email_deliverable('valid@example.com')

    @patch('validators.Client')
    def test_invalid_email_fails(self, mock_client):
        mock_result = MagicMock()
        mock_result.status = 'invalid'

        mock_client.return_value.verify.return_value = mock_result

        with self.assertRaises(ValidationError):
            validate_email_deliverable('invalid@example.com')

    @patch('validators.Client')
    def test_disposable_email_blocked(self, mock_client):
        mock_result = MagicMock()
        mock_result.status = 'valid'
        mock_result.result.disposable = True

        mock_client.return_value.verify.return_value = mock_result

        with self.assertRaises(ValidationError):
            validate_email_deliverable('temp@mailinator.com')