Trust Center Analytics in Hex: Which Customers Read Which Documents

import time
from datetime import datetime, timedelta, timezone

import pandas as pd
import requests

# WOLFIA_API_KEY is provided by Hex from the project secret of the same name.
BASE_URL = "https://api.wolfia.com/v1"
HEADERS = {"X-API-Key": WOLFIA_API_KEY}
REQUEST_PAUSE_SECONDS = 2
MAXIMUM_EVENTS_PER_REQUEST = 100

# Anonymous visits are real engagement, so they are labelled rather than dropped.
UNATTRIBUTED_DOMAIN = "unattributed"
UNATTRIBUTED_LABEL = "Unidentified visitor"


def to_microseconds(moment: datetime) -> int:
    """Wolfia timestamps are microseconds since the Unix epoch."""
    return int(moment.timestamp() * 1_000_000)


def month_windows(first_month: datetime, last_month: datetime):
    """Yield (start, end) pairs covering one calendar month each."""
    cursor = first_month
    while cursor < last_month:
        following_month = (cursor + timedelta(days=32)).replace(day=1)
        yield cursor, min(following_month, last_month)
        cursor = following_month


def list_documents() -> list[dict]:
    """Every document in the trust center, including ones since removed."""
    documents, page = [], 1
    while True:
        response = requests.get(
            f"{BASE_URL}/trustportal/analytics/documents",
            headers=HEADERS,
            params={"page": page, "page_size": 100},
            timeout=30,
        )
        response.raise_for_status()
        body = response.json()
        documents.extend(body["documents"])
        if page >= body["total_pages"]:
            return documents
        page += 1
        time.sleep(REQUEST_PAUSE_SECONDS)


def fetch_interactions(document_id: str, window_start: datetime, window_end: datetime) -> list[dict]:
    response = requests.get(
        f"{BASE_URL}/trustportal/analytics/documents/{document_id}",
        headers=HEADERS,
        params={
            "start_date": to_microseconds(window_start),
            "end_date": to_microseconds(window_end),
            "activity_limit": MAXIMUM_EVENTS_PER_REQUEST,
        },
        timeout=30,
    )
    response.raise_for_status()
    detail = response.json()
    return [
        {
            "event_id": event["event_id"],
            "account_domain": event["account_domain"] or UNATTRIBUTED_DOMAIN,
            "account_name": event["account_name"] or UNATTRIBUTED_LABEL,
            "user_email": event.get("user_email") or "",
            "user_name": event.get("user_name") or "",
            "document_title": detail["document_title"],
            "event_type": event["event_type"],
            "occurred_at": datetime.fromtimestamp(
                event["occurred_at"] / 1_000_000, tz=timezone.utc
            ),
            "has_known_account": bool(event["account_domain"]),
            "has_known_person": bool(event.get("user_email")),
        }
        for event in detail["recent_activity"]
    ]


def build_interactions_frame(first_month: datetime, last_month: datetime) -> pd.DataFrame:
    documents = list_documents()
    rows = []
    for window_start, window_end in month_windows(first_month, last_month):
        for document in documents:
            rows.extend(fetch_interactions(document["document_id"], window_start, window_end))
            time.sleep(REQUEST_PAUSE_SECONDS)
    return pd.DataFrame(rows).drop_duplicates(subset="event_id")


trust_center_interactions = build_interactions_frame(
    first_month=datetime(2026, 1, 1, tzinfo=timezone.utc),
    last_month=datetime(2026, 8, 1, tzinfo=timezone.utc),
)

by_account = trust_center_interactions[trust_center_interactions["has_known_account"]]
by_person = trust_center_interactions[trust_center_interactions["has_known_person"]]