Keep HubSpot company records current

"""Keep HubSpot company records current with HappyFetch: size, industry, headquarters, LinkedIn page.

Usage: python hubspot_enrich.py
Needs HUBSPOT_TOKEN, a private app token with the scopes crm.objects.companies.read,
crm.objects.companies.write and crm.schemas.companies.write, and HAPPYFETCH_API_KEY.
Run it weekly: companies it enriched in the last FRESH_DAYS days are skipped.
"""
import datetime
import os
import time
import uuid
from concurrent.futures import ThreadPoolExecutor

import requests

HAPPYFETCH = "https://api.happyfetch.net"
HUBSPOT = "https://api.hubapi.com"
HF = {"Authorization": f"Bearer {os.environ['HAPPYFETCH_API_KEY']}"}
HS = {"Authorization": f"Bearer {os.environ['HUBSPOT_TOKEN']}"}
FRESH_DAYS = 7

# HubSpot's own industry property only takes values from its fixed list, so the industry and the size
# band go into text properties of their own, created on the first run.
OWN_PROPERTIES = [
    {"name": "happyfetch_industry", "label": "Industry (HappyFetch)", "type": "string", "fieldType": "text"},
    {"name": "happyfetch_size_band", "label": "Size band (HappyFetch)", "type": "string", "fieldType": "text"},
    {"name": "happyfetch_enriched_at", "label": "Enriched by HappyFetch", "type": "datetime", "fieldType": "date"},
]


def ensure_properties() -> None:
    for prop in OWN_PROPERTIES:
        res = requests.post(f"{HUBSPOT}/crm/v3/properties/companies", headers=HS,
                            json={**prop, "groupName": "companyinformation"}, timeout=30)
        if res.status_code not in (201, 409):  # 409: it exists from an earlier run
            res.raise_for_status()


def companies():
    """Every company in HubSpot, 100 at a time, with its domain and when we last enriched it."""
    params = {"limit": 100, "properties": "domain,happyfetch_enriched_at"}
    while True:
        res = requests.get(f"{HUBSPOT}/crm/v3/objects/companies", headers=HS, params=params, timeout=30)
        res.raise_for_status()
        page = res.json()
        yield from page["results"]
        after = page.get("paging", {}).get("next", {}).get("after")
        if not after:
            return
        params["after"] = after


def due(company: dict, now: datetime.datetime) -> bool:
    props = company["properties"]
    if not props.get("domain"):
        return False
    last = props.get("happyfetch_enriched_at")
    return not last or now - datetime.datetime.fromisoformat(last.replace("Z", "+00:00")) > datetime.timedelta(days=FRESH_DAYS)


def lookup(domain: str):
    key = str(uuid.uuid4())  # one key per domain, reused on every retry: a retry never pays twice
    for attempt in range(6):
        try:
            res = requests.post(f"{HAPPYFETCH}/v1/domain", headers={**HF, "Idempotency-Key": key},
                                json={"domain": domain}, timeout=90)
            answer = res.json()
        except (requests.RequestException, ValueError):
            time.sleep(2**attempt)  # a timeout or a dropped connection
            continue
        if res.ok:
            return answer["data"], None
        error = answer["error"]
        if not error["retryable"]:
            return None, error["code"]
        time.sleep(error.get("retry_after_seconds") or 2**attempt)
    return None, "gave_up"


def properties(company: dict, stamp: str) -> dict:
    hq = company["headquarter"] or {}
    size = company["employeeCountRange"] or {}
    founded = (company["foundedOn"] or {}).get("year")
    values = {
        "numberofemployees": company["employeeCount"],
        "city": hq.get("city"),
        "country": hq.get("country"),
        "linkedin_company_page": company["linkedInUrl"],
        "founded_year": str(founded) if founded else None,
        "happyfetch_industry": company["industry"],
        "happyfetch_size_band": f"{size['start']}-{size['end'] or ''}" if size else None,
        "happyfetch_enriched_at": stamp,
    }
    return {k: v for k, v in values.items() if v is not None}  # never blank out what HubSpot already has


def main() -> None:
    ensure_properties()
    now = datetime.datetime.now(datetime.timezone.utc)
    stamp = now.isoformat(timespec="seconds").replace("+00:00", "Z")
    todo = [c for c in companies() if due(c, now)]
    print(f"{len(todo):,} companies to enrich")
    updates, found, missing, invalid, failed = [], 0, 0, 0, 0
    with ThreadPoolExecutor(max_workers=8) as pool:
        for company, (data, error) in zip(todo, pool.map(lambda c: lookup(c["properties"]["domain"]), todo)):
            if data:
                updates.append({"id": company["id"], "properties": properties(data, stamp)})
                found += 1
            elif error == "not_found":  # no company at that domain: don't ask again this week
                updates.append({"id": company["id"], "properties": {"happyfetch_enriched_at": stamp}})
                missing += 1
            elif error == "invalid_domain":  # not a website: never charged, worth fixing in HubSpot
                invalid += 1
            else:
                failed += 1  # left as it was, so the next run tries again
    for i in range(0, len(updates), 100):  # HubSpot takes 100 updates per call
        res = requests.post(f"{HUBSPOT}/crm/v3/objects/companies/batch/update", headers=HS,
                            json={"inputs": updates[i:i + 100]}, timeout=60)
        res.raise_for_status()
    print(f"Updated {found:,} companies. {missing:,} domains have no company, {invalid:,} aren't websites, "
          f"and {failed:,} failed and wait for the next run.")


if __name__ == "__main__":
    main()