Turn website visitors into outbound pipeline

"""Daily job: turn yesterday's high-confidence website visitors into enriched contacts in a campaign list."""
import datetime
import json
import os
import time

import requests

BASE = "https://api.tryfuse.ai/api/v1/business"
AUTH = (os.environ["FUSE_API_KEY"], "")
LIST_NAME = "Website visitors - decision makers"
CUSTOMERS_LIST_ID = os.environ.get("FUSE_CUSTOMERS_LIST_ID")  # optional: a company list of your customers
STATE_FILE = "processed_domains.json"


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)


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 bare(domain):
    domain = domain.lower().strip()
    for prefix in ("https://", "http://", "www."):
        domain = domain.removeprefix(prefix)
    return domain.rstrip("/")


def recent_visitors(feed, since):
    """Yield feed rows with a visit on or after `since` (YYYY-MM-DD), newest first."""
    page = 1
    while True:
        body = call("GET", "/website-visitors", params={
            "type": feed, "since": since, "sortBy": "lastVisitDate", "sortOrder": "desc",
            "pageSize": 100, "page": page,
        })
        yield from body["visitors"]
        if page >= body["pagination"]["totalPages"]:
            return
        page += 1


since = (datetime.date.today() - datetime.timedelta(days=1)).isoformat()
processed = set(json.load(open(STATE_FILE))) if os.path.exists(STATE_FILE) else set()

# 2. Companies that visited since yesterday, with high confidence
domains = {
    bare(row["companyWebsite"])
    for row in recent_visitors("companies", since)
    if row["confidenceInterval"] == "high" and row["companyWebsite"]
} - processed

# 3. Leave out customers
if CUSTOMERS_LIST_ID:
    domains -= {bare(domain) for domain in read_domains(CUSTOMERS_LIST_ID)}

# 4-5. Find and save the decision-makers at those companies
list_id = None
if domains:
    filters = {
        "job_company_website": sorted(domains)[:1000],
        "job_title_levels": [{"status": "include", "value": level} for level in ("cxo", "vp", "director")],
        "job_title_role": [{"status": "include", "value": "finance"}],
    }
    total = call("POST", "/prospects/search", json={"filters": filters, "size": 1})["total"]
    print(f"{len(domains)} new companies, {total} matching decision-makers")
    if total:
        saved = call("POST", "/prospects/search/save-to-list", json={
            "filters": filters, "listName": LIST_NAME, "limit": min(total, 500), "enrich": "email",
        })
        list_id = saved["listIds"][0]
        print("Enrichment", wait_for_enrichment(saved["jobId"])["status"])

# 6. Add identified people, resolving before uploading so repeat runs never duplicate them
if list_id:
    known, uploaded = [], []
    for row in recent_visitors("people", since):
        if row["confidenceInterval"] != "high" or not row["personLinkedin"]:
            continue
        try:
            known.append(call("POST", "/contacts/resolve", json={"linkedinUrl": row["personLinkedin"]})["contactId"])
        except FuseError as error:
            if error.code != "CONTACT_NOT_FOUND":
                raise
            contact = {"linkedinUrl": row["personLinkedin"]}
            if row["companyName"]:
                contact["companyName"] = row["companyName"]
            if row["companyWebsite"]:
                contact["companyDomain"] = bare(row["companyWebsite"])
            uploaded += call("POST", f"/lists/{list_id}/contacts", json={"contacts": [contact]})["contactIds"]
    if known:
        call("POST", f"/lists/{list_id}/members", json={"contactIds": known[:5000]})
    people = known + uploaded
    for start in range(0, len(people), 500):
        try:
            call("POST", f"/lists/{list_id}/enrich", json={"mode": "email", "contactIds": people[start:start + 500]})
        except FuseError as error:
            if error.code != "ALL_CONTACTS_ALREADY_ENRICHED":
                raise
    print(f"Added {len(known)} known and {len(uploaded)} new identified visitors")

json.dump(sorted(processed | domains), open(STATE_FILE, "w"))