Build an AI outbound campaign from a prompt

"""Turn a sentence about your buyer into an approved AI email campaign."""
import os
import time

import requests

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


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)


# 1. Describe the buyer and preview five matches
preview = call("POST", "/prospects/search", json={
    "query": "Heads of finance at UK fintech companies with 50 to 500 employees",
    "size": 5,
})
print(f"{preview['total']} matches. Unresolved: {preview.get('unresolvedCriteria')}")

# 2-3. Check the balance, then save 200 matches with work emails
print("Credit balance:", call("GET", "/credits")["balance"])
list_id = call("POST", "/prospects/search/save-to-list", json={
    "filters": preview["filters"],
    "listName": "UK fintech finance leaders",
    "limit": 200,
    "enrich": "email",
})["listIds"][0]

# 4. Wait for the list to fill
fuse_list = wait_until(
    lambda: call("GET", f"/lists/{list_id}")["list"],
    lambda current: current["completionStatus"] == "complete",
)
print("Contacts with a work email:", fuse_list["metrics"]["totalWithProfessionalEmail"])

# 5. Choose what grounds the copy
hubs = [hub for hub in call("GET", "/knowledge-hubs")["knowledgeHubs"] if hub["status"] == "complete"]
style = call("GET", "/writing-styles")["writingStyles"][0]  # the default style comes first
if not call("GET", "/accounts", params={"channel": "email"})["accounts"]:
    raise SystemExit("Connect a mailbox in the Fuse app before you approve.")

# 6. Create the AI campaign, retrying while the list's reachability settles
campaign_body = {
    "campaignName": "UK fintech finance leaders - Q4",
    "campaignType": "ai-generated",
    "listId": list_id,
    "channels": ["email", "email", "email"],
    "userPrompt": "Book intro calls with finance leaders who still run month-end close by hand. "
                  "Friendly, concise, no jargon.",
    "writingStyleId": style["id"],
    "ctaLink": "https://example.com/demo",
    "timezone": "Europe/London",
    "excludeContacted": "last_3_months",
}
if hubs:
    campaign_body["knowledgeHubId"] = hubs[0]["id"]
for attempt in range(5):
    try:
        campaign_id = call("POST", "/campaigns", json=campaign_body)["campaignId"]
        break
    except FuseError as error:
        if error.code != "LIST_INCOMPATIBLE_WITH_CHANNELS" or attempt == 4:
            raise
        time.sleep(90)

# 7. Wait for the AI to plan the sequence
campaign = wait_until(
    lambda: call("GET", f"/campaigns/{campaign_id}")["campaign"],
    lambda current: current["initializationStatus"] in ("completed", "failed"),
)
if campaign["initializationStatus"] == "failed":
    raise SystemExit(campaign["initializationError"])

# 8. Preview every step for a sample contact
for step in call("GET", f"/campaigns/{campaign_id}/steps")["steps"]:
    rendered = call("POST", f"/campaigns/{campaign_id}/steps/{step['id']}/generate")["preview"]
    print(f"\nStep {step['sequenceNumber']}: {step['topic']}\n{rendered['subject']}\n{rendered['body']}")

# 9. Approving sends real email, so ask first
if input("\nApprove and start sending? [y/N] ").strip().lower() == "y":
    print(call("POST", f"/campaigns/{campaign_id}/approve"))