Verify a 50,000-row CSV with Python, asyncio and httpx

"""Verify a large CSV with the Cold Leads bulk API (Python 3.9+, httpx).

  pip install httpx
  export COLDLEADS_API_KEY=sk_...
  python verify_csv.py leads.csv leads_verified.csv --column email

Addresses are normalised and de-duplicated, sent in jobs of up to 5,000 with at most
5 jobs open at a time, and every job is polled until it is done. The output is the
input CSV plus verify_status, verify_score and verify_reason. Job ids are saved next
to the output file, so a second run resumes the same jobs instead of paying twice.
"""
from __future__ import annotations

import argparse
import asyncio
import csv
import hashlib
import json
import os
import re
import sys
import time
from pathlib import Path

import httpx

API_BASE = os.environ.get("COLDLEADS_API_BASE", "https://coldleads.app")
JOB_SIZE = 5000    # addresses per job (API maximum)
MAX_JOBS = 5       # queued or running jobs per account (API maximum)
POLL_PAUSE = 3.0   # seconds between two polls of the same job
MIN_GAP = 0.6      # seconds between any two requests: at most 100 a minute (limit: 120 per key)
# The pattern the API uses to pick addresses out of a list. Cells that do not match are
# skipped here and cost nothing.
EMAIL_RE = re.compile(r"[a-z0-9!#$%&'*+/=?^_`{|}~.-]+@[a-z0-9.-]+\.[a-z]{2,}")


class ApiError(Exception):
    pass


class Api:
    """HTTP calls paced across all tasks, with the documented 429 and 5xx handling."""

    def __init__(self, client: httpx.AsyncClient):
        self.client = client
        self.lock = asyncio.Lock()
        self.next_at = 0.0

    async def call(self, method: str, path: str, **kwargs) -> tuple[int, dict]:
        while True:
            async with self.lock:
                delay = self.next_at - time.monotonic()
                if delay > 0:
                    await asyncio.sleep(delay)
                self.next_at = time.monotonic() + MIN_GAP
            try:
                r = await self.client.request(method, path, **kwargs)
            except httpx.TransportError as e:
                if method != "GET":
                    # a POST may have reached the server: retrying could open a second job
                    raise ApiError(f"network error on {method} {path}: {e!r}") from e
                print(f"network error on {path}, retrying in 10 s", file=sys.stderr)
                await asyncio.sleep(10)
                continue
            try:
                data = r.json()
            except ValueError:
                data = {}
            if r.status_code == 429 and data.get("error") == "rate_limited":
                wait = int(r.headers.get("retry-after") or data.get("retry_after_seconds") or 60)
                print(f"rate limited, pausing {wait} s", file=sys.stderr)
                self.next_at = time.monotonic() + wait  # every task waits out the window
                continue
            if r.status_code >= 500 and method == "GET":
                await asyncio.sleep(10)
                continue
            return r.status_code, data


async def run_job(api, slots, stop, idx, emails, state, save_state, results):
    async with slots:
        job_id = state["jobs"].get(str(idx))
        while not job_id:
            if stop.is_set():
                return
            status, data = await api.call("POST", "/api/v1/verify/bulk", json={"emails": emails, "name": f"{state['name']} #{idx + 1}"})
            if status == 202:
                job_id = data["id"]
                state["jobs"][str(idx)] = job_id
                save_state()
            elif status == 429 and data.get("error") == "too_many_jobs":
                await asyncio.sleep(30)  # other jobs of this account are still open
            elif status == 402:
                print(f"job {idx + 1}: not enough credits ({data.get('credits')}), stopping", file=sys.stderr)
                stop.set()
            else:
                raise ApiError(f"job {idx + 1}: HTTP {status} {data}")
        shown = -1
        while True:
            status, job = await api.call("GET", f"/api/v1/verify/jobs/{job_id}")
            if status != 200:
                raise ApiError(f"job {job_id}: HTTP {status} {job}")
            if job["status"] in ("done", "failed"):
                break
            if job["done"] // 500 != shown // 500:
                shown = job["done"]
                print(f"job {idx + 1}: {job['done']}/{job['total']} {job['counts']}")
            await asyncio.sleep(POLL_PAUSE)  # each poll also processes the next addresses
        status, job = await api.call("GET", f"/api/v1/verify/jobs/{job_id}", params={"results": "1"})
        if status != 200:
            raise ApiError(f"job {job_id}: HTTP {status} {job}")
        for r in job.get("results", []):
            results[r["email"]] = r
        if job["status"] == "failed":
            print(f"job {idx + 1} stopped early (usually: credits ran out); kept {len(job.get('results', []))} results", file=sys.stderr)


async def main(argv=None) -> int:
    p = argparse.ArgumentParser(description="Verify a CSV column of e-mail addresses with Cold Leads")
    p.add_argument("input")
    p.add_argument("output")
    p.add_argument("--column", default="email")
    args = p.parse_args(argv)
    key = os.environ.get("COLDLEADS_API_KEY", "").strip()
    if not key.startswith("sk_"):
        print("Set COLDLEADS_API_KEY to a secret key (sk_...) from Settings -> API keys", file=sys.stderr)
        return 2

    with open(args.input, newline="", encoding="utf-8-sig") as f:
        reader = csv.DictReader(f)
        rows = list(reader)
        fields = list(reader.fieldnames or [])
    col = next((c for c in fields if c.strip().lower() == args.column.lower()), None)
    if col is None:
        print(f"column {args.column!r} not found in {fields}", file=sys.stderr)
        return 2
    norm = [(row.get(col) or "").strip().lower().strip("<>") for row in rows]
    unique = list(dict.fromkeys(e for e in norm if EMAIL_RE.fullmatch(e)))
    chunks = [unique[i:i + JOB_SIZE] for i in range(0, len(unique), JOB_SIZE)]

    state_path = Path(args.output + ".jobs.json")
    fingerprint = hashlib.sha256("\n".join(unique).encode()).hexdigest()
    state = json.loads(state_path.read_text()) if state_path.exists() else {"fingerprint": fingerprint, "name": Path(args.input).stem[:60], "jobs": {}}
    if state["fingerprint"] != fingerprint:
        print(f"{state_path} belongs to a different list; delete it to start over", file=sys.stderr)
        return 2

    def save_state():
        state_path.write_text(json.dumps(state, indent=2))

    headers = {"x-api-key": key, "User-Agent": "coldleads-csv-example/1.0"}
    timeout = httpx.Timeout(120.0, connect=10.0)  # a poll verifies a batch before it answers
    results: dict[str, dict] = {}
    async with httpx.AsyncClient(base_url=API_BASE, headers=headers, timeout=timeout) as client:
        api = Api(client)
        status, credits = await api.call("GET", "/api/v1/credits")
        if status != 200:
            print(f"API error {status}: {credits}", file=sys.stderr)
            return 1
        needed = sum(len(c) for i, c in enumerate(chunks) if str(i) not in state["jobs"])
        print(f"{len(rows)} rows, {len(unique)} unique addresses in {len(chunks)} jobs; {needed} credits needed, {credits['left']} left")
        if credits["left"] < needed:
            print(f"Not enough credits: short by {needed - credits['left']}. Add packs in Settings -> Billing.", file=sys.stderr)
            return 2
        slots, stop = asyncio.Semaphore(MAX_JOBS), asyncio.Event()
        try:
            await asyncio.gather(*(run_job(api, slots, stop, i, c, state, save_state, results) for i, c in enumerate(chunks)))
        except ApiError as e:
            print(f"{e}\nJob ids are saved in {state_path}: run the same command again to resume.", file=sys.stderr)
            return 1

    out_fields = fields + [c for c in ("verify_status", "verify_score", "verify_reason") if c not in fields]
    counts: dict[str, int] = {}
    with open(args.output, "w", newline="", encoding="utf-8") as f:
        w = csv.DictWriter(f, fieldnames=out_fields, extrasaction="ignore")
        w.writeheader()
        for row, email in zip(rows, norm):
            r = results.get(email)
            if r:
                row.update(verify_status=r["status"], verify_score=r["score"], verify_reason=r["reason"])
            elif not EMAIL_RE.fullmatch(email):
                row.update(verify_status="skipped", verify_score="", verify_reason="not_an_email")
            else:
                row.update(verify_status="", verify_score="", verify_reason="not_checked")
            counts[row["verify_status"] or "not_checked"] = counts.get(row["verify_status"] or "not_checked", 0) + 1
            w.writerow(row)
    print(f"wrote {args.output}: {counts}")
    return 0


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))

pip install httpx
export COLDLEADS_API_KEY=sk_your_secret_key
python verify_csv.py leads.csv leads_verified.csv --column email