Python实用工具：深入解析 python-email-validator 库的邮件验证实践

pip install python-email-validator

from email_validator import validate_email, EmailNotValidError

def validate_single_email(email):
    try:
        # 调用 validate_email 函数进行验证，默认开启 DNS 验证（check_mx=True）
        valid = validate_email(email)
        # 验证通过后，返回的 valid 是一个 EmailValidationResult 对象
        print(f"邮箱 {email} 验证通过！")
        print(f"邮箱格式：{valid.email}")
        print(f"域名：{valid.domain}")
        print(f"MX 记录：{valid.mx_record}")
    except EmailNotValidError as e:
        # 捕获验证失败的异常，输出具体错误信息
        print(f"邮箱 {email} 验证失败：{str(e)}")

# 测试案例
validate_single_email("[email protected]")       # 有效邮箱，验证通过
validate_single_email("user@example")           # 缺少顶级域名，格式错误
validate_single_email("[email protected]")      # 连续点号，格式错误
validate_single_email("[email protected]") # 域名不存在，MX 记录验证失败

def validate_format_only(email):
    try:
        # 设置 check_mx=False 仅验证格式，不进行 DNS 查询
        valid = validate_email(email, check_mx=False)
        print(f"仅格式验证通过：{valid.email}")
    except EmailNotValidError as e:
        print(f"格式验证失败：{str(e)}")

validate_format_only("[email protected]")   # 格式正确，验证通过
validate_format_only("user@example")       # 格式错误，验证失败

def validate_with_mode(email, mode="strict"):
    try:
        # 通过参数 schema 设置验证模式，可选值为 "strict" 或 "loose"
        valid = validate_email(email, schema=mode)
        print(f"{mode} 模式下验证通过：{valid.email}")
    except EmailNotValidError as e:
        print(f"{mode} 模式下验证失败：{str(e)}")

# 测试案例：宽松模式允许带加号的邮箱
validate_with_mode("[email protected]", "loose")  # 宽松模式下通过
validate_with_mode("[email protected]", "strict")  # 严格模式下可能失败（取决于库的具体实现）

def validate_domain_whitelist(email, allowed_domains):
    try:
        valid = validate_email(email)
        # 检查域名是否在允许的白名单中
        if valid.domain not in allowed_domains:
            raise EmailNotValidError(f"域名 {valid.domain} 不在允许的列表中")
        print(f"邮箱 {email} 验证通过，属于允许的域名")
    except EmailNotValidError as e:
        print(f"验证失败：{str(e)}")

# 允许 example.com 和 gmail.com 域名
validate_domain_whitelist("[email protected]", ["example.com", "gmail.com"])  # 通过
validate_domain_whitelist("[email protected]", ["example.com", "gmail.com"])  # 失败，域名不在列表中

def validate_international_email(email):
    try:
        # 设置 allow_smtputf8=True 支持国际邮箱（如包含中文的邮箱）
        valid = validate_email(email, allow_smtputf8=True)
        print(f"国际邮箱验证通过：{valid.email}")
    except EmailNotValidError as e:
        print(f"验证失败：{str(e)}")

# 测试案例：中文邮箱（实际使用中需确保邮箱服务商支持）
validate_international_email("用户@示例.com")  # 需服务商支持 Punycode 编码，验证可能通过

from email_validator import validate_email, EmailNotValidError, DNSNotFoundError, TimeoutError

def register_user(email, password):
    # 第一步：基本格式与 MX 记录验证
    try:
        valid = validate_email(
            email,
            check_mx=True,        # 验证 MX 记录
            allow_smtputf8=True,  # 支持国际邮箱
            timeout=10            # 设置 DNS 查询超时时间为 10 秒
        )
        cleaned_email = valid.email  # 获取标准化后的邮箱地址（如自动转为小写）
    except EmailNotValidError as e:
        # 处理不同类型的验证错误
        error_message = str(e)
        if "is not a valid" in error_message:
            return "邮箱格式错误，请检查输入"
        elif "No MX record" in error_message:
            return "邮箱所属域名无法接收邮件，请更换其他邮箱"
        elif "Timeout" in error_message:
            return "验证超时，请重试"
        else:
            return "邮箱验证失败，请联系管理员"
    except Exception as e:
        return f"系统错误：{str(e)}"

    # 第二步：禁止临时邮箱（示例：禁止以 example.com 结尾的邮箱）
    if cleaned_email.endswith("@example.com"):
        return "禁止使用测试邮箱，请更换为真实邮箱"

    # 第三步：模拟用户注册逻辑（此处可连接数据库执行插入操作）
    # 假设注册成功
    return "注册成功！请查收验证邮件"

# 测试用例
print(register_user("[email protected]", "password"))       # 禁止使用的测试邮箱，返回禁止提示
print(register_user("[email protected]", "password")) # 域名不存在，返回 MX 记录错误
print(register_user("[email protected]", "password"))         # 有效邮箱，注册成功

import concurrent.futures

def validate_emails_in_bulk(emails):
    results = []
    with concurrent.futures.ThreadPoolExecutor() as executor:
        # 使用线程池并发验证邮箱，提升批量处理效率
        future_to_email = {executor.submit(validate_email, email): email for email in emails}
        for future in concurrent.futures.as_completed(future_to_email):
            email = future_to_email[future]
            try:
                valid = future.result()
                results.append((email, "valid", valid.domain))
            except EmailNotValidError as e:
                results.append((email, "invalid", str(e)))
            except Exception as e:
                results.append((email, "error", str(e)))
    return results

# 测试案例：验证多个邮箱
email_list = [
    "[email protected]",
    "[email protected]",
    "[email protected]",
    "user4@example",
    "user5@临时邮箱.com"  # 假设该域名无 MX 记录
]
bulk_results = validate_emails_in_bulk(email_list)

for email, status, info in bulk_results:
    print(f"邮箱：{email}，状态：{status}，详情：{info}")

def validate_disposable_email(email, disposable_domains):
    try:
        valid = validate_email(email)
        domain = valid.domain.lower()
        if domain in disposable_domains:
            raise EmailNotValidError("禁止使用临时邮箱")
        print(f"邮箱 {email} 验证通过")
    except EmailNotValidError as e:
        print(f"验证失败：{str(e)}")

# 临时邮箱域名列表（可从公开数据库获取最新列表）
disposable_domains = {
    "mailinator.com",
    "guerrillamail.com",
    "yopmail.com",
    "example.com"
}

validate_disposable_email("[email protected]", disposable_domains)  # 验证失败，提示禁止使用