Cleaning an email list from a CSV with a short script

To clean an email list from a CSV, normalize and dedupe the addresses, look up each unique domain once, and split the rows into keep, review and drop files based on the domain category and spam verdict. The Python script below does that with only the standard library, stays under a 60 requests/minute API limit, and can be stopped and resumed without repeating lookups.

It's aimed at the common case: a list exported from a form tool, event platform or old CRM, which you want to import somewhere without dragging in throwaway inboxes and garbage rows.

What "clean" means here

Be clear about what a script like this can and can't remove.

Dedupe by jane.doe@gmail.com vs janedoe@gmail.com is a separate decision with real trade-offs. The Gmail dots and plus addressing guide covers it; this script only dedupes exact matches.

The script

Save as clean_emails.py. Python 3.9+ and no packages to install.

#!/usr/bin/env python3
"""Classify and split a CSV of emails using the pingutil lookup API."""
import argparse, csv, json, os, re, sys, time
import urllib.request, urllib.error

API_URL = "https://pingutil.com/v1/lookup"
MIN_INTERVAL = 1.1  # seconds between requests, ~55/min under the 60/min key limit
SYNTAX = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+quot;)


def lookup(key, value, check_dns):
    body = json.dumps({"input": value, "skip_enrichment": True, "check_dns": check_dns}).encode()
    for attempt in range(5):
        req = urllib.request.Request(API_URL, data=body, method="POST", headers={
            "Authorization": f"Bearer {key}",
            "Content-Type": "application/json",
        })
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.load(resp)
        except urllib.error.HTTPError as e:
            err = json.loads(e.read() or b"{}")
            if e.code == 400:
                return {"category": "unknown", "invalid": True}
            if e.code == 401:
                sys.exit("401 unauthorized: check PINGUTIL_KEY")
            if e.code == 429:
                wait = err.get("retry_after_seconds", 60)
                print(f"  rate limited, sleeping {wait}s", file=sys.stderr)
                time.sleep(wait)
                continue
            time.sleep(2 ** attempt)  # 5xx: exponential backoff
        except urllib.error.URLError:
            time.sleep(2 ** attempt)  # network blip
    return None  # give up on this one; it stays out of the cache and is retried next run


def decide(r):
    if r.get("invalid"):
        return "drop", "invalid"
    v = r["verdict"]
    if v["is_spam"]:
        return "drop", v["reason"]
    ds = r.get("domain_signals")
    if ds and not ds["can_receive_email"]:
        return "drop", "no_mail_server"
    if v["low_trust"]:
        return "review", v["reason"]
    if r["category"] in ("isp", "unknown"):
        return "review", r["category"]
    return "keep", ""


def main():
    p = argparse.ArgumentParser()
    p.add_argument("csv_in")
    p.add_argument("--column", default="email")
    p.add_argument("--per-address", action="store_true",
                   help="look up every address (adds role/randomized checks, many more requests)")
    p.add_argument("--check-dns", action="store_true", help="drop domains that cannot receive mail")
    p.add_argument("--cache", default=".pingutil_cache.json")
    args = p.parse_args()

    key = os.environ.get("PINGUTIL_KEY") or sys.exit("set PINGUTIL_KEY")
    cache = json.load(open(args.cache)) if os.path.exists(args.cache) else {}

    with open(args.csv_in, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        fields = reader.fieldnames + ["category", "action", "reason"]
        rows, seen = [], set()
        for row in reader:
            email = (row.get(args.column) or "").strip().lower()
            if email in seen:
                continue  # exact duplicate
            seen.add(email)
            row[args.column] = email
            rows.append(row)

    def key_for(email):
        return email if args.per_address else email.rsplit("@", 1)[-1]

    todo = sorted({key_for(r[args.column]) for r in rows
                   if SYNTAX.match(r[args.column]) and key_for(r[args.column]) not in cache})
    print(f"{len(rows)} unique rows, {len(todo)} lookups needed", file=sys.stderr)

    last = 0.0
    for i, k in enumerate(todo, 1):
        time.sleep(max(0, MIN_INTERVAL - (time.monotonic() - last)))
        last = time.monotonic()
        result = lookup(key, k, args.check_dns)
        if result is not None:
            cache[k] = result
            if i % 25 == 0 or i == len(todo):
                json.dump(cache, open(args.cache, "w"))
                print(f"  {i}/{len(todo)}", file=sys.stderr)
    json.dump(cache, open(args.cache, "w"))

    stem = os.path.splitext(args.csv_in)[0]
    outs = {a: open(f"{stem}.{a}.csv", "w", newline="", encoding="utf-8") for a in ("keep", "review", "drop")}
    writers = {a: csv.DictWriter(f, fieldnames=fields) for a, f in outs.items()}
    for w in writers.values():
        w.writeheader()

    counts = {"keep": 0, "review": 0, "drop": 0}
    for row in rows:
        email = row[args.column]
        if not SYNTAX.match(email):
            r = {"invalid": True, "category": "unknown"}
        else:
            r = cache.get(key_for(email))
            if r is None:  # lookup failed after retries; don't guess
                r = {"category": "unknown", "verdict": {"is_spam": False, "low_trust": False, "reason": None}}
        action, reason = decide(r)
        row.update(category=r.get("category", "unknown"), action=action, reason=reason or "")
        writers[action].writerow(row)
        counts[action] += 1
    for f in outs.values():
        f.close()
    print(counts, file=sys.stderr)


if __name__ == "__main__":
    main()

Run it:

export PINGUTIL_KEY=pk_live_your_key
python3 clean_emails.py leads.csv --column email --check-dns
# writes leads.keep.csv, leads.review.csv, leads.drop.csv

Each output file keeps every original column and adds category, action and reason, so you can check the drop file before throwing anything away.

Design choices worth understanding

Dedupe by domain first

pingutil keys its results on the registrable domain (eTLD+1), and the two things that matter most for list cleaning, category and the disposable verdict, depend only on the domain. So the script sends acme.com, not jane@acme.com, and a list of 10,000 rows with 1,500 unique domains costs 1,500 requests. At about 55 per minute that's under half an hour instead of about three hours.

The trade-off: per-address signals like randomized local parts (ccjxx90sze@) and role addresses (info@, noreply@) are only computed when you send the full address. --per-address switches to that mode. It's more thorough and much slower, and on lists collected through a form you control it's usually worth it. On a scraped or purchased list, use it, because that's where generated addresses show up. (Also, think hard before emailing a purchased list at all.)

The local dedupe uses the host as written, while the API collapses subdomains (eu.mail.acme.com resolves to acme.com; see eTLD+1 explained). Worst case you spend an extra request or two per company.

Throttle on your side, then honour 429

The key limit is 60 requests per minute. There's no batch endpoint, so the script paces itself at one request every 1.1 seconds. If something else shares the key and you still get 429 rate_limited, the error body includes retry_after_seconds; the script sleeps for that and retries the same lookup. A 400 means the API rejected the input (an IP literal, a reserved TLD like .test, and so on), so it's marked invalid instead of retried.

Cache to disk so you can resume

Every 25 lookups the cache is written to .pingutil_cache.json. Kill the script, rerun it, and only the missing domains are fetched. Failed lookups are not cached, so a rerun retries them. Delete the cache file when you want fresh results; disposable lists are re-synced weekly, so a cache older than that will miss newly listed domains.

Skip enrichment

skip_enrichment: true skips the LLM company profile and returns category, email_signals and verdict in milliseconds. For cleaning, you don't need company data. If you want to enrich the keepers afterwards, run a second pass over just leads.keep.csv rows with category == "company"; the lead enrichment guide walks through which fields are worth keeping.

Tune the rules to your use

decide() is deliberately short so you can edit it. Some common changes:

• B2B-only import: move free and isp to drop. Read should your B2B signup require a work email first; plenty of real buyers use Gmail.
• Newsletter list: keep free and isp, but send role_based to drop, since shared inboxes rarely engage and are more likely to report spam.
• Students or researchers: educational passes through as keep already. Being on an academic domain doesn't prove someone is a current student, so don't treat it as verification.

--check-dns adds domain_signals, resolved over DNS-over-HTTPS. A domain with no MX and no A record can't receive mail, so the script drops it. That's a cheap way to catch typo domains that don't exist. It won't catch typos that happen to be registered.

Before you import

1. Open leads.drop.csv and skim the reason column. If a domain you recognize is in there, check it in the live demo.
2. Spot-check leads.review.csv. That's where judgment calls live: role addresses, ISP mail, unknowns.
3. Keep the cache file if you'll clean another export soon; it saves requests.
4. Remember that blocklists lag behind new disposable domains. This removes known junk, not all of it. The disposable domain list on GitHub is the community source pingutil syncs from, if you want to see what's covered.

If your team works in spreadsheets rather than a terminal, the same approach as an Apps Script lives in classifying a list of emails in Google Sheets.