email-validator.py

from flask import Flask, request, jsonify
from email_validator import validate_email, EmailNotValidError
import dns.resolver
import smtplib
import time

app = Flask(__name__)

def validate_email_address(email):
    """Validates the email syntax and checks the domain for MX records."""
    try:
        # Validate email syntax
        v = validate_email(email)
        email = v["email"]  # Normalized email
        domain = email.split('@')[1]

        # Check if the domain has MX records
        mx_records = dns.resolver.resolve(domain, 'MX')
        if not mx_records:
            return {"valid": False, "reason": "No MX records found for the domain."}

        return {"valid": True, "email": email}
    except EmailNotValidError as e:
        return {"valid": False, "reason": str(e)}
    except dns.resolver.NXDOMAIN:
        return {"valid": False, "reason": "Domain does not exist."}
    except Exception as e:
        return {"valid": False, "reason": f"Unexpected error: {str(e)}"}

def smtp_check(email):
    """Performs SMTP validation to check if the recipient email exists."""
    try:
        domain = email.split('@')[1]
        mx_records = dns.resolver.resolve(domain, 'MX')
        mx_record = str(mx_records[0].exchange)

        # Connect to the mail server
        server = smtplib.SMTP(mx_record)
        server.set_debuglevel(0)
        server.helo()
        server.mail('test@example.com')  # Sender email
        code, message = server.rcpt(email)  # Recipient email
        server.quit()

        if code == 250:
            return {"valid": True}
        else:
            return {"valid": False, "reason": message.decode()}
    except Exception as e:
        return {"valid": False, "reason": f"SMTP validation error: {str(e)}"}

@app.route('/validate', methods=['POST'])
def validate():
    """API endpoint to validate an email address."""
    data = request.json
    email = data.get("email")
    if not email:
        return jsonify({"error": "Email is required"}), 400

    # Step 1: Syntax and domain validation
    syntax_result = validate_email_address(email)
    if not syntax_result['valid']:
        return jsonify(syntax_result)

    # Step 2: Optional SMTP validation
    smtp_result = smtp_check(email)
    if smtp_result['valid']:
        return jsonify({"valid": True, "email": email})
    else:
        return jsonify(smtp_result)

@app.route('/validate_bulk', methods=['POST'])
def validate_bulk():
    """API endpoint to validate multiple email addresses in bulk."""
    data = request.json
    emails = data.get("emails")
    if not emails or not isinstance(emails, list):
        return jsonify({"error": "A list of emails is required"}), 400

    results = []
    for email in emails:
        # Add a small delay to avoid triggering rate limits or being flagged
        time.sleep(0.1)  # 100ms delay

        # Step 1: Syntax and domain validation
        syntax_result = validate_email_address(email)
        if not syntax_result['valid']:
            results.append({"email": email, "valid": False, "reason": syntax_result['reason']})
            continue

        # Step 2: Optional SMTP validation
        smtp_result = smtp_check(email)
        if smtp_result['valid']:
            results.append({"email": email, "valid": True})
        else:
            results.append({"email": email, "valid": False, "reason": smtp_result['reason']})

    return jsonify(results)

if __name__ == "__main__":
    app.run(debug=True)