Deduplicating a CRM Export by Company Domain over REST with Python

crm.csv (id, company_name, domain, owner, updated_at)
    ↓
normalize existing domains (lowercase, strip scheme, www., path)
    ↓
distinct normalized names of records with no domain
    ↓ POST /routines/{Company Domain}/run  (1 credit per distinct name)
fill domains, group by normalized domain
    ↓
keep the best record per group -> crm_clean.csv, merged list

Implement a Python script that deduplicates a CRM company export using the Clay Public API.

Requirements:
- Base URL: https://api.clay.com/public/v0, auth header "clay-api-key".
- Input crm.csv has the columns id, company_name, domain, owner, updated_at. Some rows have no domain.
- Normalize an existing domain by lowercasing it, taking the hostname from a URL when one is given, and removing a leading "www.".
- Normalize a company name by lowercasing it, removing periods and commas, removing a trailing "inc", "llc", "ltd", "corp" or "co", and collapsing whitespace.
- For records with no domain, look up each distinct normalized name once with the "Company Domain" routine, function:t_0tjqm9rUjfGY4MunGDM. Its input is "Company Name" and its result is { "Domain": "<domain>" }. It costs 1 credit per row. Run it with POST /routines/{id}/run, poll GET /routines/run/{routine_run_id}/results until status is not "in_progress", and match rows by id, because results do not come back in input order.
- A record whose domain stays empty after the lookup is kept as its own group.
- Group by normalized domain. In each group keep the record with the most filled fields among company_name, domain and owner, and break ties with the latest updated_at.
- Write crm_clean.csv and print the number of input records, clean records, names looked up, and domains resolved, one line per merged group, and the credits spent.
- A lookup returns the domain Clay has for the company, which can differ from the domain a CRM stores. Do not assume two records for the same company share a domain.
- Run the verification step below before finishing.

mkdir clay-crm-dedupe && cd clay-crm-dedupe
python3 -m venv .venv && source .venv/bin/activate
pip install requests

cat > crm.csv <<'CSV'
id,company_name,domain,owner,updated_at
1,Notion,notion.so,dana,2026-06-01
2,"Notion Labs, Inc.",,,2026-02-11
3,NOTION,https://www.notion.so/,sam,2026-09-30
4,Figma,figma.com,,2026-03-04
5,Figma Inc,,jo,2026-08-19
6,Linear,linear.app,dana,2026-07-07
7,Stripe,,sam,2026-05-20
8,"Stripe, Inc.",,,2026-01-09
CSV

export CLAY_API_KEY=clay_scoped_...

import os
import time

import requests

BASE = "https://api.clay.com/public/v0"

session = requests.Session()
session.headers.update({"clay-api-key": os.environ["CLAY_API_KEY"], "content-type": "application/json"})


def clay(path, method="GET", body=None):
    res = session.request(method, f"{BASE}{path}", json=body, timeout=60)
    if not res.ok:
        raise RuntimeError(f"{res.status_code} {path}: {res.text}")
    return res.json()


def balance():
    return clay("/credits/balance")["balance"]


def all_results(run_id):
    """Fetch every row. The endpoint returns 20 rows per page unless limit is set, and adds a cursor when more remain."""
    page = clay(f"/routines/run/{run_id}/results?limit=100")
    rows = list(page.get("data", []))
    while page.get("cursor"):
        page = clay(f"/routines/run/{run_id}/results?limit=100&cursor={page['cursor']}")
        rows.extend(page.get("data", []))
    page["data"] = rows
    return page


def run_routine(routine_id, rows, timeout_s=600):
    started = clay(f"/routines/{routine_id}/run", "POST", {"items": rows})
    run_id = started["routine_run_id"]
    deadline = time.time() + timeout_s
    while time.time() < deadline:
        time.sleep(5)
        results = all_results(run_id)
        if results["status"] != "in_progress":
            return results
    raise RuntimeError(f"run {run_id} still in progress after {timeout_s} seconds")

import csv
import json
import re
import sys
import time
from collections import defaultdict
from urllib.parse import urlparse

from clay import balance, run_routine

COMPANY_DOMAIN = "function:t_0tjqm9rUjfGY4MunGDM"


def normalize_domain(value):
    value = value.strip().lower()
    if not value:
        return ""
    host = urlparse(value if "//" in value else "//" + value).hostname or ""
    return host.removeprefix("www.")


def normalize_name(value):
    value = re.sub(r"[.,]", "", value.lower())
    value = re.sub(r"\s+(inc|llc|ltd|corp|co)$", "", value)
    return re.sub(r"\s+", " ", value).strip()


def main():
    with open("crm.csv", newline="") as f:
        records = list(csv.DictReader(f))
    for r in records:
        r["domain_norm"] = normalize_domain(r["domain"])

    # Resolve a domain only for records that lack one, and only once per distinct normalized name.
    missing = sorted({normalize_name(r["company_name"]) for r in records if not r["domain_norm"]})
    before = balance()
    resolved = {}
    if missing:
        results = run_routine(COMPANY_DOMAIN, [{"id": str(i), "inputs": {"Company Name": n}} for i, n in enumerate(missing)])
        for item in results.get("data", []):
            domain = (item.get("result") or {}).get("Domain", "")
            resolved[missing[int(item["id"])]] = normalize_domain(domain)
    for r in records:
        if not r["domain_norm"]:
            r["domain_norm"] = resolved.get(normalize_name(r["company_name"]), "")
            r["domain_source"] = "clay"
        else:
            r["domain_source"] = "crm"

    groups = defaultdict(list)
    for r in records:
        groups[r["domain_norm"] or f"unresolved:{r['id']}"].append(r)

    # Keep the record with the most filled fields, then the most recently updated.
    def completeness(r):
        return (sum(1 for k in ("company_name", "domain", "owner") if r[k].strip()), r["updated_at"])

    kept, merged = [], []
    for domain, group in groups.items():
        group.sort(key=completeness, reverse=True)
        kept.append(group[0])
        if len(group) > 1:
            merged.append({"domain": domain, "kept": group[0]["id"], "merged_away": [g["id"] for g in group[1:]]})

    with open("crm_clean.csv", "w", newline="") as f:
        writer = csv.writer(f)
        writer.writerow(["id", "company_name", "domain", "owner", "updated_at"])
        for r in sorted(kept, key=lambda r: r["id"]):
            writer.writerow([r["id"], r["company_name"], r["domain_norm"], r["owner"], r["updated_at"]])

    time.sleep(5)
    print(json.dumps({"input_records": len(records), "clean_records": len(kept), "domains_resolved": sum(1 for v in resolved.values() if v), "names_looked_up": len(missing)}))
    for m in merged:
        print(json.dumps(m))
    print(json.dumps({"credits_spent": before - balance()}))


if __name__ == "__main__":
    try:
        main()
    except RuntimeError as err:
        print(err, file=sys.stderr)
        sys.exit(1)

python3 crm_dedupe.py
cat crm_clean.csv

{"input_records": 8, "clean_records": 5, "domains_resolved": 3, "names_looked_up": 3}
{"domain": "notion.so", "kept": "3", "merged_away": ["1"]}
{"domain": "figma.com", "kept": "5", "merged_away": ["4"]}
{"domain": "stripe.com", "kept": "7", "merged_away": ["8"]}
{"credits_spent": 3.0}
id,company_name,domain,owner,updated_at
2,"Notion Labs, Inc.",firstround.com,,2026-02-11
3,NOTION,notion.so,sam,2026-09-30
5,Figma Inc,figma.com,jo,2026-08-19
6,Linear,linear.app,dana,2026-07-07
7,Stripe,stripe.com,sam,2026-05-20

verification:
  status: verified
  tested_at: "2026-10-12"
  product_version: "public/v0"
  command: "python3 crm_dedupe.py"
  expected_result: "Records without a domain are looked up once per distinct normalized name, records are grouped by normalized domain, merged groups are listed, and credits_spent equals names_looked_up."