How to export 1 million B2B contacts to CSV (bulk export API)

curl -s https://api.leadocean.io/v1/exports/columns | python3 -c \
  "import json,sys; print(json.load(sys.stdin)['data']['ceilings'])"

curl -s https://api.leadocean.io/v1/usage -H "x-api-key: $LEADOCEAN_API_KEY"

curl -s -X POST "https://api.leadocean.io/v1/people/search?count=true" \
  -H "x-api-key: $LEADOCEAN_API_KEY" -H "content-type: application/json" \
  -d '{
    "country": ["US"],
    "jobLevel": ["Director"],
    "emailStatus": ["verified", "catch_all_valid", "catch_all"],
    "limit": 1
  }'

"""plan.py: build non-overlapping slices worth 1,000,000 rows"""
import json, os, requests

BASE = "https://api.leadocean.io"
H = {"x-api-key": os.environ["LEADOCEAN_API_KEY"], "content-type": "application/json"}
TARGET, CEILING = 1_000_000, 50_000
STATUSES = ["verified", "catch_all_valid", "catch_all"]
FIXED = {"country": ["US"], "jobLevel": ["Director"]}
FUNCTIONS = ["Advertising & Marketing", "Engineering", "Finance & Accounting",
             "General Business & Management", "Human Resources", "Information Technology",
             "Operations", "Sales & Business Development"]   # add the other 14 as needed
RANGES = ["1-10", "11-50", "51-200", "201-500", "501-1000", "1001-5000", "5001-10000", "10001+"]

def count(filters):
    body = {**filters, "limit": 1}
    r = requests.post(f"{BASE}/v1/people/search", headers=H, params={"count": "true"}, json=body)
    r.raise_for_status()
    meta = r.json()["meta"]
    return meta.get("total") or 0

slices = []
for fn in FUNCTIONS:
    for rng in RANGES:
        cell = {**FIXED, "jobFunction": [fn], "employeeRange": [rng]}
        cell_total = count({**cell, "emailStatus": STATUSES})
        if cell_total == 0:
            continue
        parts = [{**cell, "emailStatus": STATUSES}] if cell_total <= CEILING else \
                [{**cell, "emailStatus": [s]} for s in STATUSES]
        for f in parts:
            n = count(f)
            if n == 0:
                continue
            if n > CEILING:
                print("still over 50,000, split further:", f)
                continue
            slices.append({"filters": f, "rows": n})

total = 0
for s in slices:
    s["rows"] = min(s["rows"], TARGET - total)
    total += s["rows"]
    if total >= TARGET:
        slices = slices[: slices.index(s) + 1]
        break

json.dump(slices, open("slices.json", "w"), indent=1)
print(len(slices), "slices,", total, "rows")

"""run.py: one export per slice, download each file"""
import json, os, time, requests
from datetime import datetime, timezone

BASE = "https://api.leadocean.io"
H = {"x-api-key": os.environ["LEADOCEAN_API_KEY"], "content-type": "application/json"}

def start(i, s):
    body = {"name": f"slice {i:03d}", "filters": s["filters"],
            "limit": s["rows"], "preset": "full"}
    while True:
        r = requests.post(f"{BASE}/v1/exports", headers=H, json=body)
        if r.status_code == 202:
            return r.json()["data"]
        err = r.json().get("error", {})
        if r.status_code == 429:
            when = (err.get("details") or {}).get("resetsAt")
            wait = 60
            if when:
                t = datetime.fromisoformat(when.replace("Z", "+00:00"))
                wait = max(60, (t - datetime.now(timezone.utc)).total_seconds())
            print(f"slice {i}: {err.get('message')} Sleeping {int(wait)}s")
            time.sleep(wait)
            continue
        raise SystemExit(f"slice {i}: {r.status_code} {err}")

def wait_done(export_id):
    while True:
        d = requests.get(f"{BASE}/v1/exports/{export_id}", headers=H).json()["data"]
        if d["status"] in ("done", "failed", "cancelled"):
            return d
        time.sleep(d.get("pollAfter") or 10)

for i, s in enumerate(json.load(open("slices.json"))):
    path = f"slice_{i:03d}.csv"
    if os.path.exists(path):
        continue                       # already downloaded on an earlier run
    job = start(i, s)
    print(f"slice {i}: {job['id']} perRow={job['perRow']} remaining={job['remaining']}")
    d = wait_done(job["id"])
    if d["status"] != "done":
        print(f"slice {i}: {d['status']} {d.get('error')} rows={d['rows']} billed={d['billed']}")
        continue
    with requests.get(f"{BASE}/v1/exports/{job['id']}/download", headers=H, stream=True) as r:
        r.raise_for_status()
        with open(path, "wb") as f:
            for chunk in r.iter_content(1 << 20):
                f.write(chunk)
    print(f"slice {i}: {d['rows']} rows, billed {d['billed']}")

"""merge.py: merge the slice files and dedupe on person_id"""
import csv, glob

seen, writer = set(), None
with open("contacts_1m.csv", "w", newline="") as out:
    for path in sorted(glob.glob("slice_*.csv")):
        with open(path, newline="") as f:
            reader = csv.DictReader(f)
            if writer is None:
                writer = csv.DictWriter(out, fieldnames=reader.fieldnames)
                writer.writeheader()
            for row in reader:
                if row["person_id"] in seen:
                    continue
                seen.add(row["person_id"])
                writer.writerow(row)
print(len(seen), "unique people")

{
  "success": true,
  "data": {
    "id": "<export id>",
    "name": "slice 000",
    "status": "queued",
    "requested": 50000,
    "estimatedTotal": 50000,
    "reveal": "email_phone",
    "perRow": 1,
    "remaining": null,
    "pollAfter": 10,
    "file": null
  }
}