Personalize every email with AI research

"""Write a researched opening line for every prospect and merge it into a manual campaign."""
import os
import time

import requests

BASE = "https://api.tryfuse.ai/api/v1/business"
AUTH = (os.environ["FUSE_API_KEY"], "")
LIST_ID = os.environ["FUSE_LIST_ID"]
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_for_run(job_id, delay=10, max_delay=60):
    while True:
        time.sleep(delay)
        job = call("GET", f"/lists/{LIST_ID}/smart-columns/{job_id}/status")["job"]
        if job["status"] in FINISHED_RUNS:
            return job
        delay = min(delay * 1.5, max_delay)


def all_rows():
    page = 1
    while True:
        body = call("GET", f"/lists/{LIST_ID}/rows", params={"limit": 1000, "pageNum": page})
        yield from body["data"]
        if page >= body["pagination"]["totalPages"]:
            return
        page += 1


# 1. Test the opener prompt on ten rows
created = call("POST", f"/lists/{LIST_ID}/smart-columns", json={
    "name": "Opener",
    "prompt": "Write one sentence, under 25 words, that opens a cold email to this person by referring to "
              "something specific their company announced, launched or published in the last six months. "
              "No greeting, no mention of our product, no invented facts. If you find nothing specific, "
              "write one sentence about a challenge common to their role instead.",
    "engine": "deep_research",
    "runOn": "first_10",
})
column_id = created["column"]["id"]
wait_for_run(created["jobId"])

# 2. Show each sample line with its sources
for row in call("GET", f"/lists/{LIST_ID}/rows", params={"limit": 10})["data"]:
    line = row["columnValues"].get(column_id)
    if line:
        sources = call("GET", f"/lists/{LIST_ID}/rows/{row['id']}/columns/{column_id}/provenance")
        print(f"{row['firstName']} at {row['companyName']}: {line}\n  {sources['provenance']['sourceUrls']}")
if input("Use these lines? [y/N] ").strip().lower() != "y":
    raise SystemExit(f"Edit the prompt with PATCH /lists/{LIST_ID}/smart-columns/{column_id}, then continue.")

# 3. Write lines for the rest of the list
rerun = call("POST", f"/lists/{LIST_ID}/smart-columns/{column_id}/rerun", json={"mode": "new_only"})

# 4. Create a manual campaign and find the column's token while the run works
campaign_id = call("POST", "/campaigns", json={
    "campaignName": "Fintech finance leaders - researched openers",
    "campaignType": "manual",
    "listId": LIST_ID,
    "channels": ["email", "email"],
    "timezone": "Europe/London",
})["campaignId"]
token = next(v["token"] for v in call("GET", f"/campaigns/{campaign_id}/variables")["variables"]
             if v["name"] == "Opener")

# 5. Write both steps around the researched line
first, follow_up = sorted(call("GET", f"/campaigns/{campaign_id}/steps")["steps"],
                          key=lambda step: step["sequenceNumber"])
call("PATCH", f"/campaigns/{campaign_id}/steps/{first['id']}", json={
    "subject": "Quick question, {Contact First Name}",
    "body": "Hi {Contact First Name},\n\n" + token + "\n\nWe help finance teams at companies like "
            "{Contact Company} close the books in three days instead of ten. "
            "Worth a 20-minute call next week?\n\n{Sender First Name}",
    "daysBetween": 0,
})
call("PATCH", f"/campaigns/{campaign_id}/steps/{follow_up['id']}", json={
    "subject": "Re: Quick question, {Contact First Name}",
    "body": "Hi {Contact First Name}, worth a look? Happy to send a two-minute video instead of a call.\n\n"
            "{Sender First Name}",
    "daysBetween": 3,
})

# 6. Wait for every line, then take contacts without one off the list
if wait_for_run(rerun["jobId"])["status"] not in ("completed", "partially_completed"):
    raise SystemExit("The run did not finish; check its status before approving.")
gaps = [row["id"] for row in all_rows() if not row["columnValues"].get(column_id)]
if gaps:
    print(f"{len(gaps)} contacts have no opener and will be removed from the list")
    for start in range(0, len(gaps), 5000):
        call("DELETE", f"/lists/{LIST_ID}/members", json={"contactIds": gaps[start:start + 5000]})

# 7. Preview, then approve
print(call("POST", f"/campaigns/{campaign_id}/steps/{first['id']}/generate")["preview"]["body"])
if input("Approve and start sending? [y/N] ").strip().lower() == "y":
    call("POST", f"/campaigns/{campaign_id}/approve")