Article content and detailed guides remain in English. The selected language applies to controls and quick instructions.

Back to articles

How to Clean and Deduplicate Email Lists in CSV Files with Python

On this page

When to Use Python for Email List Cleaning

Python is the right tool when you need to clean email data at scale, apply custom logic, or integrate cleaning into an automated pipeline:

Scenario Why Python Alternative
Lists over 100,000 rows Spreadsheets slow down; Python handles millions Database (SQL)
Custom validation rules Regex, domain checks, business rules No easy alternative
Recurring cleaning tasks Script runs on schedule; consistent results Manual process
Multiple file formats Read CSV, XLSX, JSON, TXT in one script Different tools per format
Integration with other tools Output to CRM, ESP, database, API Manual import/export
Complex deduplication Match across columns, fuzzy matching, cross-file Specialised deduplication tools

Basic CSV Email Cleaning with the csv Module

Read, deduplicate and write

import csv
import re

def clean_email(email):
    """Normalise an email address."""
    if not email:
        return None
    email = email.strip().lower()
    # Remove surrounding quotes or angle brackets
    email = email.strip('"\'<>')
    return email

def is_valid_email(email):
    """Basic email format validation."""
    if not email:
        return False
    pattern = r'^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

# Read, clean and deduplicate
seen = set()
clean_rows = []

with open('contacts.csv', 'r', encoding='utf-8') as f:
    reader = csv.DictReader(f)
    fieldnames = reader.fieldnames

    for row in reader:
        email = clean_email(row.get('email', ''))

        if not email or not is_valid_email(email):
            continue

        if email in seen:
            continue

        seen.add(email)
        row['email'] = email
        clean_rows.append(row)

# Write cleaned data
with open('contacts_clean.csv', 'w', encoding='utf-8',
          newline='') as f:
    writer = csv.DictWriter(f, fieldnames=fieldnames)
    writer.writeheader()
    writer.writerows(clean_rows)

print(f"Original: {len(seen) + (len(clean_rows))} rows")
print(f"After cleaning: {len(clean_rows)} rows")
print(f"Removed: {len(seen) - len(clean_rows)} duplicates")

Cleaning with pandas

Basic deduplication

import pandas as pd

# Read CSV
df = pd.read_csv('contacts.csv')

# Normalise email column
df['email'] = (df['email']
    .astype(str)
    .str.strip()
    .str.lower()
    .str.strip('"\'<>'))

# Remove rows with no email
df = df[df['email'].notna() & (df['email'] != '') &
        (df['email'] != 'nan')]

# Remove duplicates (keep first occurrence)
before = len(df)
df = df.drop_duplicates(subset='email', keep='first')
after = len(df)

print(f"Removed {before - after} duplicates")

# Save
df.to_csv('contacts_clean.csv', index=False)

Advanced cleaning pipeline

import pandas as pd
import re

def clean_email_series(series):
    """Clean a pandas Series of email addresses."""
    return (series
        .astype(str)
        .str.strip()
        .str.lower()
        .str.strip('"\'<>')
        .str.replace(r'\s+', '', regex=True))

def validate_email(email):
    """Validate email format."""
    if not isinstance(email, str) or email in ('', 'nan', 'none'):
        return False
    pattern = r'^[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}$'
    return bool(re.match(pattern, email))

def is_role_based(email):
    """Check if email is role-based."""
    role_prefixes = [
        'info', 'admin', 'support', 'sales', 'contact',
        'help', 'office', 'hello', 'mail', 'team',
        'webmaster', 'postmaster', 'noreply', 'no-reply',
        'billing', 'accounts', 'hr', 'jobs', 'careers',
        'marketing', 'press', 'media', 'feedback',
        'abuse', 'security', 'privacy', 'legal',
        'compliance', 'service', 'enquiries', 'inquiries'
    ]
    local_part = email.split('@')[0]
    return local_part in role_prefixes

def is_disposable_domain(email):
    """Check against common disposable email domains."""
    disposable_domains = {
        'mailinator.com', 'guerrillamail.com', 'tempmail.com',
        'throwaway.email', 'yopmail.com', 'sharklasers.com',
        'guerrillamailblock.com', 'grr.la', 'dispostable.com',
        'trashmail.com', 'fakeinbox.com', 'tempinbox.com',
        'maildrop.cc', 'getairmail.com', 'temp-mail.org'
    }
    domain = email.split('@')[1] if '@' in email else ''
    return domain in disposable_domains

def is_free_provider(email):
    """Check if email uses a free provider."""
    free_providers = {
        'gmail.com', 'yahoo.com', 'hotmail.com',
        'outlook.com', 'aol.com', 'icloud.com',
        'mail.com', 'protonmail.com', 'zoho.com',
        'yandex.com', 'gmx.com', 'live.com',
        'msn.com', 'me.com', 'qq.com', '163.com'
    }
    domain = email.split('@')[1] if '@' in email else ''
    return domain in free_providers

# Load data
df = pd.read_csv('contacts.csv')
original_count = len(df)

# Step 1: Clean email format
df['email'] = clean_email_series(df['email'])

# Step 2: Validate format
df['valid'] = df['email'].apply(validate_email)
df = df[df['valid']].drop(columns=['valid'])

# Step 3: Flag characteristics
df['is_role_based'] = df['email'].apply(is_role_based)
df['is_disposable'] = df['email'].apply(is_disposable_domain)
df['is_free_provider'] = df['email'].apply(is_free_provider)

# Step 4: Remove disposable emails
df = df[~df['is_disposable']]

# Step 5: Deduplicate
df = df.drop_duplicates(subset='email', keep='first')

# Step 6: Report
print(f"Original rows: {original_count}")
print(f"After cleaning: {len(df)}")
print(f"Role-based: {df['is_role_based'].sum()}")
print(f"Free provider: {df['is_free_provider'].sum()}")
print(f"Business email: "
      f"{(~df['is_free_provider'] & ~df['is_role_based']).sum()}")

# Step 7: Save (with or without flag columns)
output = df.drop(
    columns=['is_role_based', 'is_disposable', 'is_free_provider'])
output.to_csv('contacts_clean.csv', index=False)

# Optional: save segmented files
df[df['is_role_based']].to_csv(
    'contacts_role_based.csv', index=False)
df[~df['is_free_provider']].to_csv(
    'contacts_business.csv', index=False)

Common Email Cleaning Tasks

Fix common typos in email domains

DOMAIN_TYPOS = {
    'gmial.com': 'gmail.com',
    'gmal.com': 'gmail.com',
    'gmaill.com': 'gmail.com',
    'gnail.com': 'gmail.com',
    'gmail.co': 'gmail.com',
    'gamil.com': 'gmail.com',
    'yahooo.com': 'yahoo.com',
    'yaho.com': 'yahoo.com',
    'yahoo.co': 'yahoo.com',
    'hotmal.com': 'hotmail.com',
    'hotmial.com': 'hotmail.com',
    'hotmail.co': 'hotmail.com',
    'outlok.com': 'outlook.com',
    'outloo.com': 'outlook.com',
}

def fix_domain_typo(email):
    """Fix common domain typos."""
    if '@' not in email:
        return email
    local, domain = email.rsplit('@', 1)
    corrected = DOMAIN_TYPOS.get(domain, domain)
    return f"{local}@{corrected}"

df['email'] = df['email'].apply(fix_domain_typo)

Merge and deduplicate multiple CSV files

import pandas as pd
import glob

# Read all CSV files
files = glob.glob('contacts_*.csv')
frames = []

for f in files:
    df = pd.read_csv(f)
    df['source_file'] = f
    frames.append(df)

# Combine
combined = pd.concat(frames, ignore_index=True)
print(f"Total rows across all files: {len(combined)}")

# Normalise emails
combined['email'] = (combined['email']
    .astype(str)
    .str.strip()
    .str.lower())

# Deduplicate (keep first occurrence)
deduped = combined.drop_duplicates(subset='email', keep='first')
print(f"After deduplication: {len(deduped)}")
print(f"Duplicates removed: {len(combined) - len(deduped)}")

deduped.to_csv('contacts_merged_clean.csv', index=False)

Extract emails from a column with mixed content

import re

def extract_emails_from_text(text):
    """Extract all email addresses from a text string."""
    if not isinstance(text, str):
        return []
    pattern = r'[a-zA-Z0-9._%+\-]+@[a-zA-Z0-9.\-]+\.[a-zA-Z]{2,}'
    return re.findall(pattern, text)

# Apply to a column that contains mixed text
df['extracted_emails'] = df['notes'].apply(
    extract_emails_from_text)

# Explode into separate rows
df_exploded = df.explode('extracted_emails')
df_exploded = df_exploded[
    df_exploded['extracted_emails'].notna()]

print(f"Extracted {len(df_exploded)} email addresses")

Split a large file into batches

import pandas as pd
import math

df = pd.read_csv('contacts_clean.csv')
batch_size = 1000
num_batches = math.ceil(len(df) / batch_size)

for i in range(num_batches):
    start = i * batch_size
    end = start + batch_size
    batch = df.iloc[start:end]
    batch.to_csv(f'batch_{i+1:03d}.csv', index=False)
    print(f"Batch {i+1}: {len(batch)} rows")

Performance Tips

Tip Why How
Use pandas for large files Vectorised operations are faster than row-by-row loops Use .str accessor methods instead of .apply() when possible
Read only needed columns Reduces memory usage pd.read_csv('file.csv', usecols=['email', 'name'])
Process in chunks Handle files larger than RAM pd.read_csv('file.csv', chunksize=10000)
Use sets for deduplication O(1) lookup vs O(n) list search seen = set() instead of checking a list
Pre-compile regex Avoid recompiling the pattern for every row pattern = re.compile(r'...') then pattern.match(email)

When to Use a Dedicated Tool

Python scripts give you full control and work well for custom cleaning logic and pipeline integration. For quick, one-off cleaning jobs across multiple file formats (PDF, DOCX, XLSX, HTML, EML and others beyond CSV), upload files to Email Extractor instead of writing a custom parser for each format. It handles 19 file types, deduplicates automatically, and requires no code.

Extract emails

Explore tools

Verify emails

Check address validity before using your list.

ZeroBounce

Email Verification

Verifies email lists and provides tools for monitoring deliverability.

Useful when list cleaning and sender health belong in one workflow.

Explore ZeroBounce (opens in a new tab)