python/hubspot_calls.py

"""
HubSpot Calls → Supabase (incremental since a millisecond cursor)

Usage from orchestrator:
    import load_hubspot_calls
    load_hubspot_calls.main(since_ms=<int milliseconds since epoch UTC>)

Direct CLI:
    # epoch ms
    python load_hubspot_calls.py 1754025600000
    # ISO-8601
    python load_hubspot_calls.py 2025-08-01T09:30:00Z
    # Back-compat date (floors to 00:00Z)
    python load_hubspot_calls.py 2025-08-01
"""

import os
import re
import time
import logging
import datetime
from typing import Dict, List, Optional, Tuple, Union

import httpx
import hubspot
from dotenv import load_dotenv
from supabase import create_client
from hubspot.crm.objects.calls import ApiException as CallsApiException

from hubspot_utils import (
    parse_ts, try_parse_int, clean_text, deduplicate_by_key,
)
from supabase_utils import (
    batched_insert, update_sync_metadata,
)

# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
logging.basicConfig(
    filename=f"logs/hubspot_calls_pipeline_{datetime.datetime.now().strftime('%Y-%m-%d')}.log",
    filemode="a",
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
)

# -----------------------------------------------------------------------------
# Environment
# -----------------------------------------------------------------------------
load_dotenv()
HUBSPOT_TOKEN = os.getenv("HUBSPOT_TOKEN")
SUPABASE_URL = os.getenv("SUPABASE_URL")
SUPABASE_SERVICE_ROLE_KEY = os.getenv("SUPABASE_SERVICE_ROLE_KEY")
# Optional bootstrap cursor if orchestrator doesn't provide one
BOOTSTRAP_SINCE_MS_ENV = os.getenv("HUBSPOT_CALLS_SINCE_MS")

if not HUBSPOT_TOKEN:
    raise RuntimeError("HUBSPOT_TOKEN is not set")
if not SUPABASE_URL or not SUPABASE_SERVICE_ROLE_KEY:
    raise RuntimeError("Supabase env vars are not set")

hubspot_client = hubspot.Client.create(access_token=HUBSPOT_TOKEN)
supabase_client = create_client(SUPABASE_URL, SUPABASE_SERVICE_ROLE_KEY)

# -----------------------------------------------------------------------------
# Config
# -----------------------------------------------------------------------------
CALL_PROPERTIES = [
    "hs_call_body",
    "hs_call_deal_stage_during_call",
    "hs_call_direction",
    "hs_call_disposition",
    "hs_call_duration",
    "hs_call_from_number",
    "hs_call_from_number_nickname",
    "hs_call_phone_number_source",
    "hs_call_primary_deal",
    "hs_call_source",
    "hs_call_status",
    "hs_call_title",
    "hs_call_to_number",
    "hs_call_to_number_nickname",
    "hs_created_by",
    "hs_created_by_user_id",
    "hs_createdate",
    "hs_engagement_source",
    "hs_engagements_last_contacted",
    "hs_lastmodifieddate",
    "hs_modified_by",
    "hs_object_source",
    "hs_timestamp",
    "hs_updated_by_user_id",
    "hubspot_owner_assigneddate",
    "hubspot_owner_id",
    "call_summary_details",
]

# -----------------------------------------------------------------------------
# Time helpers
# -----------------------------------------------------------------------------


def _ensure_utc(dt: datetime.datetime) -> datetime.datetime:
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=datetime.timezone.utc)
    return dt.astimezone(datetime.timezone.utc)


def floor_to_utc_midnight(dt: datetime.datetime) -> datetime.datetime:
    dt = _ensure_utc(dt)
    return dt.replace(hour=0, minute=0, second=0, microsecond=0)


def _parse_iso_like_to_dt(value: str) -> datetime.datetime:
    if value.endswith("Z"):
        value = value[:-1] + "+00:00"
    dt = datetime.datetime.fromisoformat(value)
    return _ensure_utc(dt)


def to_epoch_ms(dt_or_str: Union[str, datetime.datetime]) -> int:
    if isinstance(dt_or_str, str):
        dt = _parse_iso_like_to_dt(dt_or_str)
    elif isinstance(dt_or_str, datetime.datetime):
        dt = _ensure_utc(dt_or_str)
    else:
        raise TypeError(f"Unsupported type for to_epoch_ms: {type(dt_or_str)}")
    return int(dt.timestamp() * 1000)


def parse_any_ts_ms(value: Optional[Union[str, int, float]]) -> Optional[int]:
    """
    Accepts:
      - ms-epoch as str/int/float
      - seconds-epoch as str/int/float (auto *1000)
      - ISO-8601 string
    Returns ms since epoch or None.
    """
    if value is None:
        return None
    try:
        v = int(str(value))
        if v < 10_000_000_000:  # seconds → ms
            v *= 1000
        return v
    except ValueError:
        pass
    try:
        return to_epoch_ms(str(value))
    except Exception:
        logging.warning("Could not parse timestamp value=%r", value)
        return None

# -----------------------------------------------------------------------------
# Search IDs (ts > since_ms)
# -----------------------------------------------------------------------------


def _search_call_ids_from(since_ms: int, prop: str) -> List[str]:
    """
    Search calls where {prop} > since_ms.
    Sort ascending so we can advance cursor monotonically.
    """
    url = "https://api.hubapi.com/crm/v3/objects/calls/search"
    headers = {
        "Authorization": f"Bearer {HUBSPOT_TOKEN}",
        "Content-Type": "application/json",
        "Accept": "application/json",
    }
    payload = {
        "filterGroups": [{
            "filters": [
                {"propertyName": prop, "operator": "GT", "value": str(since_ms)},
            ]
        }],
        "limit": 100,
        "sorts": [{"propertyName": prop, "direction": "ASCENDING"}],
    }

    ids: List[str] = []
    after: Optional[str] = None
    with httpx.Client(timeout=30.0) as client:
        while True:
            body = dict(payload)
            if after:
                body["after"] = after

            resp = client.post(url, headers=headers, json=body)
            if resp.status_code >= 400:
                try:
                    logging.error(
                        "Call search error for prop '%s': %s", prop, resp.json())
                except Exception:
                    logging.error(
                        "Call search error for prop '%s': %s", prop, resp.text)
                resp.raise_for_status()

            data = resp.json()
            ids.extend([obj["id"] for obj in data.get("results", []) or []])

            after = (data.get("paging") or {}).get("next", {}).get("after")
            if not after:
                break
            time.sleep(0.1)

    return ids


def search_call_ids_after_ms(since_ms: int) -> Tuple[List[str], str]:
    """
    Search call IDs where {prop} > since_ms (strictly greater),
    sorted ASC so the max timestamp at the end is monotonic.
    Returns a tuple of (call_ids, prop_used).
    """
    props_to_try = ["hs_createdate"]
    last_err = None

    for prop in props_to_try:
        try:
            ids = _search_call_ids_from(since_ms, prop)
            logging.info(
                "Call search with '%s' returned %d IDs.", prop, len(ids))
            return ids, prop
        except httpx.HTTPStatusError as e:
            last_err = e
            continue

    if last_err:
        raise last_err
    return [], "hs_createdate"

# -----------------------------------------------------------------------------
# Read-by-ID (with associations)
# -----------------------------------------------------------------------------


def read_calls_by_ids(
    call_ids: List[str],
    cursor_prop: str,
) -> Tuple[List[Dict], List[Dict], List[Dict], List[Dict], Optional[int]]:
    """
    Read calls by ID with properties and associations (contacts/companies/deals).
    Returns: calls, call_contact_links, call_company_links, call_deal_links, max_ts_ms
    """
    if not call_ids:
        return [], [], [], [], None

    calls: List[Dict] = []
    call_contact_links: List[Dict] = []
    call_company_links: List[Dict] = []
    call_deal_links: List[Dict] = []

    assoc_types = ["contacts", "companies", "deals"]
    max_ts_ms: Optional[int] = None

    for i, cid in enumerate(call_ids, start=1):
        try:
            record = hubspot_client.crm.objects.calls.basic_api.get_by_id(
                cid,
                properties=CALL_PROPERTIES,
                associations=assoc_types,
                archived=False,
            )
            p = record.properties or {}

            # Track max timestamp based on the cursor property we searched on
            cursor_val = p.get(cursor_prop)
            ts_ms = parse_any_ts_ms(cursor_val)
            if ts_ms is not None and (max_ts_ms is None or ts_ms > max_ts_ms):
                max_ts_ms = ts_ms

            calls.append({
                "call_id": try_parse_int(record.id),
                "hubspot_owner_id": try_parse_int(p.get("hubspot_owner_id")),
                "call_title": p.get("hs_call_title"),
                "call_body": clean_text(p.get("hs_call_body") or "") or None,
                "deal_stage": p.get("hs_call_deal_stage_during_call"),
                "call_direction": p.get("hs_call_direction"),
                "call_outcome": p.get("hs_call_disposition"),
                "call_duration": try_parse_int(p.get("hs_call_duration")),
                "from_number": p.get("hs_call_from_number"),
                "from_number_name": p.get("hs_call_from_number_nickname"),
                "to_number": p.get("hs_call_to_number"),
                "to_number_name": p.get("hs_call_to_number_nickname"),
                "phone_number_source": p.get("hs_call_phone_number_source"),
                "primary_deal_id": try_parse_int(p.get("hs_call_primary_deal")),
                "recording_url": p.get("hs_call_recording_url"),
                "call_source": p.get("hs_call_source"),
                "call_status": p.get("hs_call_status"),
                "call_summary_details": p.get("call_summary_details"),
                "engagement_source": p.get("hs_engagement_source"),
                "engagements_last_contacted": parse_ts(p.get("hs_engagements_last_contacted")),
                "activity_date": parse_ts(p.get("hs_timestamp")),
                "owner_assigned_at": parse_ts(p.get("hubspot_owner_assigneddate")),
                "hubspot_created_at": parse_ts(p.get("hs_createdate")),
                "hubspot_updated_at": parse_ts(p.get("hs_lastmodifieddate")),
                "hubspot_created_by": p.get("hs_created_by_user_id"),
                "hubspot_updated_by": try_parse_int(p.get("hs_updated_by_user_id")),
            })

            assoc = record.associations or {}

            if assoc.get("contacts") and getattr(assoc["contacts"], "results", None):
                for a in assoc["contacts"].results:
                    if a.id and a.id.isdigit():
                        call_contact_links.append({
                            "call_id": try_parse_int(record.id),
                            "contact_id": try_parse_int(a.id),
                        })

            if assoc.get("companies") and getattr(assoc["companies"], "results", None):
                for a in assoc["companies"].results:
                    if a.id and a.id.isdigit():
                        call_company_links.append({
                            "call_id": try_parse_int(record.id),
                            "company_id": try_parse_int(a.id),
                        })

            if assoc.get("deals") and getattr(assoc["deals"], "results", None):
                for a in assoc["deals"].results:
                    if a.id and a.id.isdigit():
                        call_deal_links.append({
                            "call_id": try_parse_int(record.id),
                            "deal_id": try_parse_int(a.id),
                        })

            if i % 200 == 0:
                logging.info("Read %d calls...", i)

            time.sleep(0.05)

        except httpx.HTTPStatusError as e:
            logging.error("HTTP error reading call %s: %s", cid, e)
        except (CallsApiException, httpx.HTTPError) as e:
            logging.error("Error reading call %s: %s", cid, e)

    return calls, call_contact_links, call_company_links, call_deal_links, max_ts_ms

# -----------------------------------------------------------------------------
# Upsert
# -----------------------------------------------------------------------------


def upsert_calls(
    calls: List[Dict],
    call_contact_links: List[Dict],
    call_company_links: List[Dict],
    call_deal_links: List[Dict],
) -> None:
    if calls:
        calls = deduplicate_by_key(calls, key="call_id")
        batched_insert(
            supabase_client, "hubspot_calls", calls,
            batch_size=1000, on_conflict=["call_id"]
        )
        print(f"Upserted {len(calls)} calls.")

    if call_contact_links:
        call_contact_links = deduplicate_by_key(
            call_contact_links, key=("call_id", "contact_id"))
        batched_insert(
            supabase_client, "hubspot_call_contacts", call_contact_links,
            batch_size=1000, on_conflict=["call_id", "contact_id"]
        )
        print(f"Upserted {len(call_contact_links)} call-contact associations.")

    if call_company_links:
        call_company_links = deduplicate_by_key(
            call_company_links, key=("call_id", "company_id"))
        batched_insert(
            supabase_client, "hubspot_call_companies", call_company_links,
            batch_size=1000, on_conflict=["call_id", "company_id"]
        )
        print(f"Upserted {len(call_company_links)} call-company associations.")

    if call_deal_links:
        call_deal_links = deduplicate_by_key(
            call_deal_links, key=("call_id", "deal_id"))
        batched_insert(
            supabase_client, "hubspot_call_deals", call_deal_links,
            batch_size=1000, on_conflict=["call_id", "deal_id"]
        )
        print(f"Upserted {len(call_deal_links)} call-deal associations.")

# -----------------------------------------------------------------------------
# Main (timestamp cursor)
# -----------------------------------------------------------------------------


def main(since_ms: Optional[int] = None) -> None:
    """
    Orchestrates:
      1) Search call IDs with hs_createdate > since_ms
      2) Read full calls with associations (track max timestamp for cursor prop)
      3) Upsert into Supabase
      4) Update sync metadata with last_sync_time
    """
    # Resolve since_ms
    if since_ms is None and BOOTSTRAP_SINCE_MS_ENV:
        try:
            since_ms = int(BOOTSTRAP_SINCE_MS_ENV)
        except ValueError:
            raise RuntimeError(
                "HUBSPOT_CALLS_SINCE_MS must be an integer (ms) if set.")

    if since_ms is None:
        # Default: today@00:00:00Z for first run
        today0 = floor_to_utc_midnight(
            datetime.datetime.now(datetime.timezone.utc))
        since_ms = to_epoch_ms(today0)

    print(f"Searching calls with timestamp > {since_ms} ...")
    ids, cursor_prop = search_call_ids_after_ms(since_ms)
    print(f"Search property: {cursor_prop}. Found {len(ids)} call IDs.")

    now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()

    if not ids:
        print("No calls beyond the cursor. Updating sync metadata and exiting.")
        update_sync_metadata(supabase_client, "hubspot_calls", now_iso)
        return

    print("Reading calls (with associations)...")
    calls, call_contact_links, call_company_links, call_deal_links, max_ts_ms = read_calls_by_ids(
        ids, cursor_prop)

    print("Upserting into Supabase...")
    upsert_calls(calls, call_contact_links, call_company_links, call_deal_links)

    new_cursor_ms = max_ts_ms if max_ts_ms is not None else since_ms

    update_sync_metadata(supabase_client, "hubspot_calls", now_iso)

    print(
        f"Calls sync complete. Advanced cursor to {new_cursor_ms} using prop '{cursor_prop}'.")

# -----------------------------------------------------------------------------
# CLI
# -----------------------------------------------------------------------------


def _parse_cli_arg_to_ms(arg: str) -> int:
    """
    Accept:
      - integer epoch ms
      - ISO-8601 (Z or offset)
      - YYYY-MM-DD (floors to 00:00Z)
    """
    if re.fullmatch(r"\d{10,13}", arg):
        v = int(arg)
        if v < 10_000_000_000_000:  # seconds -> ms
            v *= 1000
        return v

    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", arg):
        d = datetime.datetime.strptime(
            arg, "%Y-%m-%d").replace(tzinfo=datetime.timezone.utc)
        return to_epoch_ms(floor_to_utc_midnight(d))

    return to_epoch_ms(arg)


if __name__ == "__main__":
    import sys
    if len(sys.argv) > 1:
        try:
            since = _parse_cli_arg_to_ms(sys.argv[1])
        except Exception as e:
            print(
                f"Invalid timestamp. Provide epoch ms, ISO-8601, or YYYY-MM-DD. Error: {e}"
            )
            sys.exit(1)
        main(since_ms=since)
    else:
        main()