Post a daily Slack leaderboard of rep activity from HubSpot using code and cron

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

curl -s "https://api.hubapi.com/crm/v3/owners?limit=100" \
  -H "Authorization: Bearer $HUBSPOT_TOKEN" \
  | jq '.results[] | {id: .id, name: (.firstName + " " + .lastName)}'

def get_yesterday_range():
    today = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
    yesterday = today - timedelta(days=1)
    return str(int(yesterday.timestamp() * 1000)), str(int(today.timestamp() * 1000))
 
def search_engagements(object_type, start_ms, end_ms):
    """Search calls, emails, or meetings within a time range."""
    all_results = []
    after = "0"
    while True:
        resp = requests.post(
            f"https://api.hubapi.com/crm/v3/objects/{object_type}/search",
            headers=HEADERS,
            json={
                "filterGroups": [{
                    "filters": [
                        {"propertyName": "hs_timestamp", "operator": "GTE", "value": start_ms},
                        {"propertyName": "hs_timestamp", "operator": "LT", "value": end_ms},
                    ]
                }],
                "properties": ["hs_timestamp", "hubspot_owner_id"],
                "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
 
start_ms, end_ms = get_yesterday_range()
calls = search_engagements("calls", start_ms, end_ms)
emails = search_engagements("emails", start_ms, end_ms)
meetings = search_engagements("meetings", start_ms, end_ms)
 
print(f"Yesterday: {len(calls)} calls, {len(emails)} emails, {len(meetings)} meetings")

def build_leaderboard(calls, emails, meetings, owner_map):
    reps = {}
 
    def count(items, activity_type):
        for item in items:
            owner_id = item["properties"].get("hubspot_owner_id")
            if not owner_id:
                continue
            if owner_id not in reps:
                reps[owner_id] = {"calls": 0, "emails": 0, "meetings": 0, "total": 0}
            reps[owner_id][activity_type] += 1
            reps[owner_id]["total"] += 1
 
    count(calls, "calls")
    count(emails, "emails")
    count(meetings, "meetings")
 
    ranked = sorted(reps.items(), key=lambda x: x[1]["total"], reverse=True)
 
    medals = ["🥇", "🥈", "🥉"]
    lines = []
    for i, (owner_id, counts) in enumerate(ranked):
        medal = medals[i] if i < 3 else f"{i + 1}."
        name = owner_map.get(owner_id, f"Owner {owner_id}")
        lines.append(
            f"{medal} *{name}* — {counts['total']} activities "
            f"({counts['calls']}C {counts['emails']}E {counts['meetings']}M)"
        )
 
    return lines

from slack_sdk import WebClient
 
def post_leaderboard(lines):
    yesterday = datetime.now(timezone.utc) - timedelta(days=1)
    date_str = yesterday.strftime("%A, %b %d")
 
    blocks = [
        {"type": "header", "text": {"type": "plain_text", "text": "🏆 Rep Activity Leaderboard"}},
        {"type": "context", "elements": [
            {"type": "mrkdwn", "text": f"Activity for {date_str}"}
        ]},
        {"type": "divider"},
        {"type": "section", "text": {"type": "mrkdwn", "text": "\n".join(lines)}},
        {"type": "context", "elements": [
            {"type": "mrkdwn", "text": "C = Calls | E = Emails | M = Meetings"}
        ]},
    ]
 
    slack = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
    result = slack.chat_postMessage(
        channel=os.environ["SLACK_CHANNEL_ID"],
        text="Rep Activity Leaderboard",
        blocks=blocks,
        unfurl_links=False,
    )
    print(f"Posted leaderboard: {result['ts']}")
 
# --- Main ---
if __name__ == "__main__":
    owner_map = get_owners()
    start_ms, end_ms = get_yesterday_range()
    calls = search_engagements("calls", start_ms, end_ms)
    emails = search_engagements("emails", start_ms, end_ms)
    meetings = search_engagements("meetings", start_ms, end_ms)
    lines = build_leaderboard(calls, emails, meetings, owner_map)
    if lines:
        post_leaderboard(lines)
    else:
        print("No activity logged yesterday — skipping leaderboard.")