Turn a Deep Research run into a targeted campaign

"""Run Deep Research, keep the report as hub knowledge, and campaign the people it found."""
import os
import time

import requests

BASE = "https://api.tryfuse.ai/api/v1/business"
AUTH = (os.environ["FUSE_API_KEY"], "")
HUB_ID = os.environ["FUSE_KNOWLEDGE_HUB_ID"]  # your team's knowledge hub, already built
PROMPT = (
    "Find independent dental groups in Texas with five or more clinics that opened a new location in "
    "2026. For each, give the website, the number of clinics, the city of the newest clinic, and the "
    "name and LinkedIn profile of the CEO or COO."
)
FINISHED = ("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_enrichment(job_id, delay=10, max_delay=60):
    """Poll an enrichment job until it ends. A stopped job waits for a top-up and a resume in the app."""
    while True:
        time.sleep(delay)
        job = call("GET", f"/jobs/{job_id}")["job"]
        if job["status"] in ("completed", "cancelled", "failed", "stopped"):
            return job
        delay = min(delay * 1.5, max_delay)


# 2. Start the run; the idempotency key makes a retried start return the same run
run_id = call("POST", "/research/runs",
              json={"prompt": PROMPT, "effort": "medium"},
              headers={"X-Idempotency-Key": "texas-dental-groups-2026-10"})["run"]["runId"]

# 3. Follow the events until the run finishes
since = 0
while True:
    time.sleep(20)
    page = call("GET", f"/research/runs/{run_id}/events", params={"sinceSeq": since})
    for event in page["events"]:
        since = event["seq"]
        if event["type"] in ("run.status_changed", "progress"):
            print(event["type"], event["payload"])
    if page["status"] in FINISHED and page["nextSinceSeq"] is None:
        break

# 4. Read the report and the results list
run = call("GET", f"/research/runs/{run_id}")["run"]
if run["status"] != "completed":
    raise SystemExit(f"Run {run['status']}: {run['errorCode']}")
print(f"Spent {run['spend']['settledCredits']} of {run['spend']['capCredits']} credits")

# 5. Keep the report as knowledge the campaign can draw on
call("POST", f"/knowledge-hubs/{HUB_ID}/pages", json={
    "title": "Research: Texas dental groups opening new clinics",
    "category": "icp-use-cases",
    "content": run["reportMarkdown"][:60000],
})

# 6. Make the results reachable (a company list needs a people search instead)
if not run["resultListId"]:
    raise SystemExit("The run found no named people or companies; the report is on the hub.")
results = call("GET", f"/lists/{run['resultListId']}")["list"]
if results["entityType"] != "contactList":
    raise SystemExit("Company results: search for people at their domains, as in Funding round outreach.")
try:
    started = call("POST", f"/lists/{results['id']}/enrich",
                   json={"mode": "email", "limit": results["totalRecords"]})
    if started.get("jobId"):
        print("Enrichment", wait_for_enrichment(started["jobId"])["status"])
except FuseError as error:
    if error.code != "ALL_CONTACTS_ALREADY_ENRICHED":
        raise

# 7. Create the AI campaign, grounded on the team's hub (preview and approve it as in the first workflow)
campaign_body = {
    "campaignName": "Texas dental groups - new clinics",
    "campaignType": "ai-generated",
    "listId": results["id"],
    "channels": ["email", "email"],
    "ctaLink": "https://example.com/demo",
    "userPrompt": "These groups just opened a new clinic. Offer help standardizing scheduling and "
                  "billing across locations. Reference their expansion, briefly.",
    "excludeContacted": "last_3_months",
}
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)  # reachability stats catch up a little after enrichment
print("Draft campaign:", campaign_id)