Qualify leads with AI before you enrich them

"""Qualify a raw prospect list with an AI smart column, then enrich only the leads that fit."""
import json
import os
import time

import requests

BASE = "https://api.tryfuse.ai/api/v1/business"
AUTH = (os.environ["FUSE_API_KEY"], "")
FINISHED_RUNS = ("completed", "partially_completed", "failed", "cancelled")


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=60, **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_run(list_id, job_id):
    job = wait_until(
        lambda: call("GET", f"/lists/{list_id}/smart-columns/{job_id}/status")["job"],
        lambda current: current["status"] in FINISHED_RUNS,
    )
    print(f"Run {job['status']}: {job['successfulRows']} answered, "
          f"{job['failedRows']} failed, {job['creditsUsed']} credits")
    return job


# 1. Save a broad list without enrichment
raw_id = call("POST", "/prospects/search/save-to-list", json={
    "filters": {
        "job_title_role": [{"status": "include", "value": "operations"}],
        "job_title_levels": [{"status": "include", "value": "director"},
                             {"status": "include", "value": "vp"}],
        "industry": [{"status": "include", "value": "logistics and supply chain"}],
        "location_country": [{"status": "include", "value": "united states"}],
    },
    "listName": "Logistics ops leaders - raw",
    "limit": 2000,
    "enrich": "none",
})["listIds"][0]
raw = wait_for_list(raw_id)

# 2-3. Test the ICP question on the first 100 rows
created = call("POST", f"/lists/{raw_id}/smart-columns", json={
    "name": "ICP fit",
    "prompt": "Does this person's company run its own fleet of delivery vehicles? Check the company "
              "website and recent news. Answer with exactly one word: yes or no.",
    "engine": "standard",
    "runOn": "first_100",
})
column_id = created["column"]["id"]
wait_for_run(raw_id, created["jobId"])
for row in call("GET", f"/lists/{raw_id}/rows", params={"limit": 10})["data"]:
    print(row.get("companyName"), "->", row["columnValues"].get(column_id))
if input("Do these answers look right? Answer the rest of the list [y/N] ").strip().lower() != "y":
    raise SystemExit(f"Edit the prompt with PATCH /lists/{raw_id}/smart-columns/{column_id}, then continue.")

# 4. Answer every row the sample skipped
price = call("GET", f"/lists/{raw_id}/smart-columns/{column_id}")["column"]["creditsPerRow"]
print(f"About {price * (raw['totalRecords'] - 100)} credits to answer the remaining rows")
rerun = call("POST", f"/lists/{raw_id}/smart-columns/{column_id}/rerun", json={"mode": "new_only"})
wait_for_run(raw_id, rerun["jobId"])

# 5. Count the rows that qualify
qualified = {"customColumns": {column_id: {"dataType": "string", "include": ["yes"]}}}
qualified_count = call("GET", f"/lists/{raw_id}/rows", params={
    "filters": json.dumps(qualified),
    "limit": 1,
})["pagination"]["totalRecords"]
print(f"{qualified_count} of {raw['totalRecords']} rows qualify")
if qualified_count == 0:
    raise SystemExit("Nothing qualified. Loosen the question or the search.")

# 6. Copy the qualified rows into their own list
qualified_id = call("POST", "/lists/save-to-list", json={
    "sourceListId": raw_id,
    "listName": "Logistics ops leaders - qualified",
    "filters": qualified,
    "limit": qualified_count,
})["listIds"][0]
wait_for_list(qualified_id)

# 7. Enrich only the qualified contacts
started = call("POST", f"/lists/{qualified_id}/enrich", json={"mode": "email", "limit": qualified_count})
if started.get("jobId"):
    print("Enrichment", wait_for_enrichment(started["jobId"])["status"])
enriched = call("GET", f"/lists/{qualified_id}")["list"]
print("Qualified contacts with a work email:", enriched["metrics"]["totalWithProfessionalEmail"])