Fix region mapping and add line items

python/hubspot_companies.py

@@ -353,8 +353,7 @@ def map_company_data_for_db(companies: List[Dict]) -> List[Dict]:
             "industry": c.get("industry"),
             "macro_industry_grouping": c.get("macro_industry_grouping"),
             "closest_review_date": parse_ts(c.get("closest_review_date___clone")),
-            "region": c.get(c.get("region")),
-            "closest_review_date": parse_ts(c.get("closest_review_date___clone")),
+            "region": c.get("region"),
             "source_1": c.get("source_1"),
             "source_2": c.get("source_2"),
             "record_source": c.get("hs_object_source_label"),
@@ -497,10 +496,12 @@ if __name__ == "__main__":
     import sys
     if len(sys.argv) > 1:
         try:
-            main(since_ms=_parse_cli_arg_to_ms(sys.argv[1]))
+            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}")
+                f"Invalid timestamp. Provide epoch ms, ISO-8601, or YYYY-MM-DD. Error: {e}"
+            )
             sys.exit(1)
+        main(since_ms=since)
     else:
         main()

python/hubspot_lineitems.py

@@ -0,0 +1,394 @@
+"""
+HubSpot Line Items → Supabase (incremental since a millisecond cursor)
+
+Usage from orchestrator:
+    import hubspot_lineitems
+    hubspot_lineitems.main(since_ms=<int milliseconds since epoch UTC>)
+
+Direct CLI:
+    # epoch ms
+    python hubspot_lineitems.py 1754025600000
+    # ISO-8601
+    python hubspot_lineitems.py 2025-08-01T09:30:00Z
+    # Back-compat date (floors to 00:00Z)
+    python hubspot_lineitems.py 2025-08-01
+"""
+import os
+import re
+import time
+import logging
+import datetime
+from typing import List, Dict, Optional, Tuple, Union
+
+import httpx
+import hubspot
+from dotenv import load_dotenv
+from supabase import create_client
+from hubspot.crm.line_items import ApiException as LineItemsApiException
+
+from hubspot_utils import (
+    parse_ts, try_parse_int, try_parse_float, deduplicate_by_key
+)
+from supabase_utils import (
+    insert_into_supabase_table, update_sync_metadata
+)
+
+# -----------------------------------------------------------------------------
+# Constants
+# -----------------------------------------------------------------------------
+LINEITEM_PROPERTIES = [
+    "hs_product_id",
+    "hs_object_id",
+    "name",
+    "description",
+    "quantity",
+    "hs_term_in_months",
+    "recurringbillingfrequency",
+    "price",
+    "hs_cost_of_goods_sold",
+    "discount",
+    "hs_margin",
+    "hs_margin_tcv",
+    "amount",
+    "network_carrier",
+    "supplier",
+    "pipeline",
+    "hs_product_type",
+    "deal_type",
+    "cnx_type",
+    "createdate",
+    "hs_lastmodifieddate",
+]
+
+# -----------------------------------------------------------------------------
+# Logging
+# -----------------------------------------------------------------------------
+logging.basicConfig(
+    filename=f"logs/hubspot_lineitems_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_LINEITEMS_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)
+
+# -----------------------------------------------------------------------------
+# 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 isinstance(value, str) and 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 / sec-epoch / ISO-8601; returns ms since epoch or None.
+    """
+    if value is None:
+        return None
+    try:
+        v = int(str(value))
+        if v < 10_000_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) with property fallback
+# -----------------------------------------------------------------------------
+def _search_lineitem_ids_from(since_ms: int, prop: str) -> List[str]:
+    """
+    Search line ite,s where {prop} > since_ms (epoch-ms).
+    Sort ascending so we can advance the cursor monotonically.
+    """
+    url = "https://api.hubapi.com/crm/v3/objects/line_items/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("Line item search error for prop '%s': %s", prop, resp.json())
+                except Exception:
+                    logging.error("Line item 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_lineitem_ids_after_ms(since_ms: int) -> Tuple[List[str], str]:
+    """
+    Try these properties in order; return (ids, prop_used) for the first successful search:
+        1) hs_lastmodifieddate
+        2) lastmodifieddate
+        3) createdate
+        4) hs_createdate
+    """
+    props_to_try = ["createdate", "hs_createdate"]
+    last_err = None
+
+    for prop in props_to_try:
+        try:
+            ids = _search_lineitem_ids_from(since_ms, prop)
+            logging.info("Line itemsearch 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_lastmodifieddate"
+
+def read_deals_by_ids(
+    lineitem_ids: List[str],
+    cursor_prop: str,
+) -> Tuple[List[Dict], List[Dict], Optional[int]]:
+    """
+    Read line items by ID including deal associations.
+    Returns: (all_lineitems, deal_lineitem_links, max_ts_ms_for_cursor_prop)
+    """
+    if not lineitem_ids:
+        return [], [], None
+
+    all_lineitems: List[Dict] = []
+    deal_lineitem_links: List[Dict] = []
+
+    assoc_types = ["deals"]
+
+    max_ts_ms: Optional[int] = None
+
+    for i, did in enumerate(lineitem_ids, start=1):
+        try:
+            record = hubspot_client.crm.line_items.basic_api.get_by_id(
+                line_item_id=did, properties=LINEITEM_PROPERTIES, associations=assoc_types, archived=False
+            )
+            p = record.properties or {}
+
+            # Track max timestamp for the cursor property we used in the search
+            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
+
+            # Deals
+            if getattr(record, "associations", None) and record.associations.get("deals"):
+                bucket = record.associations["deals"]
+                if getattr(bucket, "results", None):
+                    for a in bucket.results:
+                        if a.id and a.id.isdigit():
+                            deal_lineitem_links.append({
+                                "deal_id": try_parse_int(record.id),
+                                "lineitem_id": try_parse_int(a.id),
+                            })
+
+            # Created date: accept either createdate or hs_createdate
+            created_iso = p.get("createdate") or p.get("hs_createdate")
+
+            all_lineitems.append({
+                "product_id": try_parse_int(p.get("hs_product_id")),
+                "lineitem_id": try_parse_int(record.id),
+                "name": p.get("name"),
+                "description": p.get("description"),
+                "quantity": try_parse_int(p.get("quantity")),
+                "term_in_months": try_parse_int(p.get("hs_term_in_months")),
+                "billing_frequency": p.get("recurringbillingfrequency"),
+                "price": try_parse_float(p.get("price")),
+                "unit_cost": try_parse_float(p.get("hs_cost_of_goods_sold")),
+                "discount": try_parse_float(p.get("discount")),
+                "margin": try_parse_float(p.get("hs_margin")),
+                "total_contract_value_margin": try_parse_float(
+                    p.get("hs_margin_tcv")
+                ),
+                "net_price": try_parse_float(p.get("amount")),
+                "network_carrier": p.get("network_carrier"),
+                "supplier": p.get("supplier"),
+                "pipeline": p.get("pipeline"),
+                "product_type": p.get("hs_product_type"),
+                "deal_type": p.get("deal_type"),
+                "cnx_type": p.get("cnx_type"),
+                "hubspot_create_date": parse_ts(created_iso),
+                "hubspot_modified_date": parse_ts(
+                    p.get("hs_lastmodifieddate")
+                ),
+            })
+
+            if i % 200 == 0:
+                logging.info("Read %d deals...", i)
+            time.sleep(0.05)
+
+        except httpx.HTTPStatusError as e:
+            logging.error("HTTP error reading line item %s: %s", did, e)
+        except (LineItemsApiException, httpx.HTTPError) as e:
+            logging.error("Error reading line item %s: %s", did, e)
+
+    return all_lineitems, deal_lineitem_links, max_ts_ms
+
+# -----------------------------------------------------------------------------
+# Upsert
+# -----------------------------------------------------------------------------
+def upsert_lineitems(lineitems: List[Dict], deal_lineitem_links: List[Dict]) -> None:
+    if lineitems:
+        insert_into_supabase_table(supabase_client, "hubspot_lineitems", lineitems)
+        print(f"Upserted {len(lineitems)} line items.")
+
+    if deal_lineitem_links:
+        deal_lineitem_links = deduplicate_by_key(deal_lineitem_links, key=("deal_id", "lineitem_id"))
+        insert_into_supabase_table(
+            supabase_client,
+            "hubspot_deal_lineitems",
+            deal_lineitem_links,
+            on_conflict=["deal_id", "lineitem_id"],
+        )
+        print(f"Upserted {len(deal_lineitem_links)} deal-lineitem associations.")
+
+# -----------------------------------------------------------------------------
+# Main (timestamp cursor)
+# -----------------------------------------------------------------------------
+def main(since_ms: Optional[int] = None):
+    """
+    Orchestrates:
+      1) Search line item IDs with <cursor_prop> > since_ms (property fallback)
+      2) Read full line items with associations (track max timestamp for <cursor_prop>)
+      3) Upsert into Supabase
+      4) Update sync metadata with { last_sync_metadata, last_sync_time, cursor_prop }
+    """
+    # 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_LINEITEMS_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 line items with timestamp > {since_ms} ...")
+    ids, cursor_prop = search_lineitem_ids_after_ms(since_ms)
+    print(f"Search property: {cursor_prop}. Found {len(ids)} line item IDs.")
+
+    now_iso = datetime.datetime.now(datetime.timezone.utc).isoformat()
+
+    if not ids:
+        print("No line items beyond the cursor. Updating sync metadata and exiting.")
+        update_sync_metadata(supabase_client, "hubspot_lineitems", now_iso)
+        return
+
+    print("Reading line items (with associations)...")
+    deals, deal_lineitem_links, max_ts_ms = read_deals_by_ids(ids, cursor_prop)
+
+    print("Upserting into Supabase...")
+    upsert_lineitems(deals, deal_lineitem_links)
+
+    # Advance cursor to max timestamp we actually ingested for the chosen property
+    new_cursor_ms = max_ts_ms if max_ts_ms is not None else since_ms
+
+    update_sync_metadata(supabase_client, "hubspot_lineitems", now_iso)
+
+    print(f"Line items 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)
+    """
+    # epoch ms or seconds
+    if re.fullmatch(r"\d{10,13}", arg):
+        v = int(arg)
+        if v < 10_000_000_000_000:  # seconds -> ms
+            v *= 1000
+        return v
+
+    # YYYY-MM-DD
+    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))
+
+    # ISO-8601
+    return to_epoch_ms(arg)
+
+if __name__ == "__main__":
+    import sys
+    if len(sys.argv) > 1:
+        try:
+            main(since_ms=_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)
+    else:
+        main()