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.