Reach buying committees at newly funded companies

"""Daily job: find the buying committee at newly funded companies and add them to a campaign list."""
import json
import os
import time

import requests

BASE = "https://api.tryfuse.ai/api/v1/business"
AUTH = (os.environ["FUSE_API_KEY"], "")
AGENT_ID = os.environ["FUSE_FUNDING_AGENT_ID"]  # created once in step 2
PEOPLE_LIST_NAME = "Funded - buying committee"
STATE_FILE = "searched_domains.json"
FILTERS = {
    "job_title_role": [{"status": "include", "value": role} for role in ("finance", "operations")],
    "job_title_levels": [{"status": "include", "value": level} for level in ("cxo", "vp", "director")],
}


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 find_agent(agent_id):
    """GET /agents is the only read that returns an agent's results listId."""
    page = 1
    while True:
        body = call("GET", "/agents", params={"limit": 100, "pageNum": page})
        for agent in body["agents"]:
            if agent["id"] == agent_id:
                return agent
        if page >= body["pagination"]["totalPages"]:
            raise SystemExit(f"Agent {agent_id} not found")
        page += 1


def read_domains(list_id):
    """filter-values reads up to 1,000 rows at a time, so page through a longer list with start."""
    domains, start = set(), 1
    while True:
        page = call("GET", f"/lists/{list_id}/filter-values",
                    params={"field": "domain", "start": start, "amount": 1000})
        domains.update(page["values"])
        start += 1000
        if start > page["total"]:
            return domains


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)


searched = set(json.load(open(STATE_FILE))) if os.path.exists(STATE_FILE) else set()

# 3-4. The funded companies the agent has delivered that we have not searched yet
agent = find_agent(AGENT_ID)
if agent["status"] in ("paused", "expired", "failed"):
    print(f"Warning: the agent is {agent['status']} and has stopped delivering.")
new_domains = sorted(read_domains(agent["listId"]) - searched)[:1000]
if not new_domains:
    raise SystemExit("No newly funded companies since the last run.")

# 5. Find the buying committee at those companies and save them with work emails
filters = {**FILTERS, "job_company_website": new_domains}
total = call("POST", "/prospects/search", json={"filters": filters, "size": 1})["total"]
print(f"{len(new_domains)} new companies, {total} matching people")
if total:
    saved = call("POST", "/prospects/search/save-to-list", json={
        "filters": filters, "listName": PEOPLE_LIST_NAME, "limit": min(total, 1000), "enrich": "email",
    })
    print("Enrichment", wait_for_enrichment(saved["jobId"])["status"])
    people = call("GET", f"/lists/{saved['listIds'][0]}")["list"]
    print("Contacts with a work email:", people["metrics"]["totalWithProfessionalEmail"])

json.dump(sorted(searched | set(new_domains)), open(STATE_FILE, "w"))