Clean and enrich your CRM data, then sync it back

"""Import a CRM export, merge duplicates, verify and enrich contacts, then push the list back."""
import csv
import os
import time

import requests

BASE = "https://api.tryfuse.ai/api/v1/business"
AUTH = (os.environ["FUSE_API_KEY"], "")
EXPORT_FILE = "hubspot-contacts.csv"
LIST_NAME = "HubSpot contacts - Oct 2026"
CRM = "hubspot"
EMAIL_HEADER, OWNER_HEADER = "Email", "Contact owner"


class FuseError(Exception):
    def __init__(self, status, code, message):
        super().__init__(f"{status} {code}: {message}")
        self.code = code


def call(method, path, **kwargs):
    """Send one request, wait out 429s, and raise FuseError on any other failure."""
    while True:
        response = requests.request(method, BASE + path, auth=AUTH, timeout=120, **kwargs)
        if response.status_code == 429:
            time.sleep(int(response.headers.get("Retry-After", 5)))
            continue
        if response.status_code >= 400:
            try:
                error = response.json().get("error", {})
            except ValueError:
                error = {}
            raise FuseError(response.status_code, error.get("code"), error.get("message"))
        return response.json() if response.content else None


def wait_until(read, done, delay=10, max_delay=60):
    """Poll read() until done(result) is true, waiting before each poll and backing off."""
    while True:
        time.sleep(delay)
        result = read()
        if done(result):
            return result
        delay = min(delay * 1.5, max_delay)


def wait_for_list(list_id):
    return wait_until(lambda: call("GET", f"/lists/{list_id}")["list"],
                      lambda current: current["completionStatus"] == "complete")


def wait_for_enrichment(job_id):
    """Poll an enrichment job until it ends. A stopped job waits for a top-up and a resume in the app."""
    return wait_until(lambda: call("GET", f"/jobs/{job_id}")["job"],
                      lambda job: job["status"] in ("completed", "cancelled", "failed", "stopped"))


def wait_for_dedupe(job_id):
    return wait_until(lambda: call("GET", f"/contacts/dedupe-jobs/{job_id}")["dedupeJob"],
                      lambda current: current["status"] in ("completed", "failed"))


# 1. Upload the export (files over 10 MB go through POST /imports/presign first)
with open(EXPORT_FILE, "rb") as handle:
    parsed = call("POST", "/imports", files={"file": (EXPORT_FILE, handle, "text/csv")})["import"]
print(f"{parsed['rowCount']} rows. Suggested mapping: {parsed['suggestedMapping']}")

# 2. Commit into a list without enrichment
committed = call("POST", f"/imports/{parsed['importId']}/commit", json={
    "listName": LIST_NAME,
    "mapping": {"Website": "Company Domain"},
    "onDuplicate": "merge",
    "enrich": "none",
})["import"]
list_id = committed["listId"]
print("Ignored headers:", committed["ignoredHeaders"])
fuse_list = wait_for_list(list_id)

# 3. Merge duplicates that involve this list
scan_id = call("POST", "/contacts/dedupe-scan", json={"matchFields": ["email"]})["jobId"]
if wait_for_dedupe(scan_id)["status"] == "completed":
    group_ids, page = [], 1
    while True:
        result = call("GET", f"/contacts/dedupe-jobs/{scan_id}",
                      params={"pageNum": page, "limit": 200})["dedupeJob"]["scanResult"]
        group_ids += [group["groupId"] for group in result["groups"]
                      if any(member_list["id"] == list_id for member_list in group["lists"])]
        if page >= result["pagination"]["totalPages"]:
            break
        page += 1
    if group_ids and input(f"Merge {len(group_ids)} duplicate groups? This cannot be undone. [y/N] ").lower() == "y":
        merge_id = call("POST", "/contacts/dedupe-merge",
                        json={"scanJobId": scan_id, "groupIds": group_ids})["jobId"]
        print(wait_for_dedupe(merge_id)["mergeResult"])

# 4. Re-verify existing emails, then find missing emails and phones
total = call("GET", f"/lists/{list_id}")["list"]["totalRecords"]
call("POST", f"/lists/{list_id}/enrich", json={"mode": "validate_emails", "limit": min(total, 5000)})
wait_for_list(list_id)
try:
    started = call("POST", f"/lists/{list_id}/enrich", json={"mode": "phone_and_email", "limit": min(total, 10000)})
    if started.get("jobId"):
        print("Enrichment", wait_for_enrichment(started["jobId"])["status"])
except FuseError as error:
    if error.code != "ALL_CONTACTS_ALREADY_ENRICHED":
        raise
print("Metrics:", call("GET", f"/lists/{list_id}")["list"]["metrics"])

# 5. Bring the CRM owner back as a custom column
with open(EXPORT_FILE, newline="") as handle:
    owners = {row[EMAIL_HEADER].strip().lower(): row[OWNER_HEADER]
              for row in csv.DictReader(handle) if row.get(EMAIL_HEADER) and row.get(OWNER_HEADER)}
try:
    column_id = call("POST", f"/lists/{list_id}/columns", json={"name": "CRM owner", "type": "string"})["column"]["id"]
except FuseError as error:
    if error.code != "COLUMN_NAME_TAKEN":
        raise
    column_id = next(column["id"] for column in call("GET", f"/lists/{list_id}/columns")["columns"]
                     if column["name"].lower() == "crm owner")
values, page = [], 1
while True:
    rows = call("GET", f"/lists/{list_id}/rows", params={"limit": 1000, "pageNum": page})
    for row in rows["data"]:
        owner = owners.get((row.get("primaryEmail") or "").lower())
        if owner:
            values.append({"contactId": row["id"], "value": owner})
    if page >= rows["pagination"]["totalPages"]:
        break
    page += 1
for start in range(0, len(values), 500):
    call("PATCH", f"/lists/{list_id}/columns/{column_id}/values", json={"values": values[start:start + 500]})
print(f"Wrote the CRM owner on {len(values)} rows")

# 6. Push the whole list, which fills empty fields and never overwrites your team's edits
if CRM not in call("GET", "/integrations")["integrations"]["connected"]:
    raise SystemExit(f"Connect {CRM} in the Fuse app first.")
print(call("POST", f"/lists/{list_id}/crm-push", json={"crm": CRM}))