Send a weekly Slack report on HubSpot sequence performance using code and cron

# Verify your token can access sequences
curl -s "https://api.hubapi.com/crm/v3/objects/sequences?limit=1" \
  -H "Authorization: Bearer $HUBSPOT_TOKEN" | head -c 300

import os
import time
import requests
from datetime import datetime, timezone, timedelta
 
HUBSPOT_TOKEN = os.environ["HUBSPOT_TOKEN"]
HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
 
def get_sequences():
    """Fetch all sequences from HubSpot."""
    sequences = []
    after = "0"
    while True:
        resp = requests.get(
            "https://api.hubapi.com/crm/v3/objects/sequences",
            headers=HEADERS,
            params={"limit": 100, "after": after, "properties": "hs_sequence_name"},
        )
        resp.raise_for_status()
        data = resp.json()
        sequences.extend(data["results"])
        if data.get("paging", {}).get("next", {}).get("after"):
            after = data["paging"]["next"]["after"]
        else:
            break
    return sequences

def get_enrollments(sequence_id, start_ms):
    """Get enrollments for a specific sequence in the given time range."""
    all_results = []
    after = "0"
    while True:
        resp = requests.post(
            "https://api.hubapi.com/crm/v3/objects/sequence_enrollments/search",
            headers=HEADERS,
            json={
                "filterGroups": [{
                    "filters": [
                        {"propertyName": "hs_sequence_id", "operator": "EQ", "value": str(sequence_id)},
                        {"propertyName": "hs_enrollment_start_date", "operator": "GTE", "value": start_ms},
                    ]
                }],
                "properties": [
                    "hs_sequence_id", "hs_enrollment_state",
                    "hs_was_email_opened", "hs_was_email_replied",
                    "hs_was_meeting_booked",
                ],
                "limit": 100,
                "after": after,
            },
        )
        resp.raise_for_status()
        data = resp.json()
        all_results.extend(data["results"])
        if data.get("paging", {}).get("next", {}).get("after"):
            after = data["paging"]["next"]["after"]
        else:
            break
    return all_results

def calculate_metrics(sequences, start_ms):
    report = []
    for seq in sequences:
        seq_name = seq["properties"].get("hs_sequence_name", f"Sequence {seq['id']}")
        enrollments = get_enrollments(seq["id"], start_ms)
        time.sleep(0.2)  # Rate limit: 5 req/sec
 
        if not enrollments:
            continue
 
        enrolled = len(enrollments)
        opened = sum(1 for e in enrollments if e["properties"].get("hs_was_email_opened") == "true")
        replied = sum(1 for e in enrollments if e["properties"].get("hs_was_email_replied") == "true")
        meetings = sum(1 for e in enrollments if e["properties"].get("hs_was_meeting_booked") == "true")
 
        report.append({
            "name": seq_name,
            "enrolled": enrolled,
            "opened": opened,
            "replied": replied,
            "meetings": meetings,
            "open_rate": round(opened / enrolled * 100, 1),
            "reply_rate": round(replied / enrolled * 100, 1),
            "meeting_rate": round(meetings / enrolled * 100, 1),
        })
 
    report.sort(key=lambda r: r["enrolled"], reverse=True)
    return report

from slack_sdk import WebClient
 
def post_report(report):
    total_enrolled = sum(r["enrolled"] for r in report)
    total_replied = sum(r["replied"] for r in report)
    total_meetings = sum(r["meetings"] for r in report)
 
    seq_lines = []
    for r in report:
        seq_lines.append(
            f"*{r['name']}* ({r['enrolled']} enrolled)\n"
            f"    Open: {r['open_rate']}% | Reply: {r['reply_rate']}% | Meeting: {r['meeting_rate']}%"
        )
 
    blocks = [
        {"type": "header", "text": {"type": "plain_text", "text": "📧 Weekly Sequence Performance Report"}},
        {"type": "section", "fields": [
            {"type": "mrkdwn", "text": f"*Total Enrolled*\n{total_enrolled}"},
            {"type": "mrkdwn", "text": f"*Total Replies*\n{total_replied}"},
            {"type": "mrkdwn", "text": f"*Meetings Booked*\n{total_meetings}"},
        ]},
        {"type": "divider"},
        {"type": "section", "text": {
            "type": "mrkdwn",
            "text": "*Per-Sequence Breakdown*\n\n" + "\n\n".join(seq_lines),
        }},
        {"type": "context", "elements": [{
            "type": "mrkdwn",
            "text": f"Last 7 days | Generated {datetime.now().strftime('%A, %B %d, %Y')}",
        }]},
    ]
 
    slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
    result = slack.chat_postMessage(
        channel=os.environ["SLACK_CHANNEL_ID"],
        text="Weekly Sequence Performance Report",
        blocks=blocks,
        unfurl_links=False,
    )
    print(f"Posted report: {result['ts']}")
 
# --- Main ---
if __name__ == "__main__":
    now = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
    start = now - timedelta(days=7)
    start_ms = str(int(start.timestamp() * 1000))
 
    sequences = get_sequences()
    print(f"Found {len(sequences)} sequences")
 
    report = calculate_metrics(sequences, start_ms)
    if report:
        post_report(report)
    else:
        print("No enrollment data for the last 7 days — skipping report.")