Post a daily Slack leaderboard of rep activity from HubSpot using an agent skill

mkdir -p .claude/skills/rep-leaderboard/scripts

---
name: rep-leaderboard
description: Generates a daily rep activity leaderboard from HubSpot (calls, emails, meetings) and posts it to Slack with emoji medals for the top 3 reps.
disable-model-invocation: true
allowed-tools: Bash(python *)
---
 
Generate a daily rep activity leaderboard by running the bundled script:
 
1. Run: `python $SKILL_DIR/scripts/leaderboard.py`
2. Review the output for any errors
3. Confirm the leaderboard was posted to Slack

#!/usr/bin/env python3
"""
Daily Rep Activity Leaderboard: HubSpot -> Slack
Fetches yesterday's calls, emails, and meetings, ranks reps, posts to Slack.
"""
import os
import sys
from datetime import datetime, timezone, timedelta
 
try:
    import requests
except ImportError:
    os.system("pip install requests slack_sdk -q")
    import requests
 
from slack_sdk import WebClient
 
# --- Config ---
HUBSPOT_TOKEN = os.environ.get("HUBSPOT_TOKEN")
SLACK_TOKEN = os.environ.get("SLACK_BOT_TOKEN")
SLACK_CHANNEL = os.environ.get("SLACK_CHANNEL_ID")
 
if not all([HUBSPOT_TOKEN, SLACK_TOKEN, SLACK_CHANNEL]):
    print("ERROR: Missing required env vars: HUBSPOT_TOKEN, SLACK_BOT_TOKEN, SLACK_CHANNEL_ID")
    sys.exit(1)
 
HEADERS = {"Authorization": f"Bearer {HUBSPOT_TOKEN}", "Content-Type": "application/json"}
 
# --- Date range: yesterday midnight to today midnight (UTC) ---
today = datetime.now(timezone.utc).replace(hour=0, minute=0, second=0, microsecond=0)
yesterday = today - timedelta(days=1)
start_ms = str(int(yesterday.timestamp() * 1000))
end_ms = str(int(today.timestamp() * 1000))
 
# --- Fetch owners ---
print("Fetching owners...")
owners_resp = requests.get("https://api.hubapi.com/crm/v3/owners?limit=100", headers=HEADERS)
owners_resp.raise_for_status()
owner_map = {}
for o in owners_resp.json()["results"]:
    owner_map[o["id"]] = f"{o.get('firstName', '')} {o.get('lastName', '')}".strip() or o["email"]
 
# --- Search engagements ---
def search(object_type):
    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()
        results.extend(data["results"])
        if data.get("paging", {}).get("next", {}).get("after"):
            after = data["paging"]["next"]["after"]
        else:
            break
    return results
 
print("Fetching yesterday's activities...")
calls = search("calls")
emails = search("emails")
meetings = search("meetings")
print(f"Found: {len(calls)} calls, {len(emails)} emails, {len(meetings)} meetings")
 
# --- Rank reps ---
reps = {}
for items, activity_type in [(calls, "calls"), (emails, "emails"), (meetings, "meetings")]:
    for item in items:
        oid = item["properties"].get("hubspot_owner_id")
        if not oid:
            continue
        if oid not in reps:
            reps[oid] = {"calls": 0, "emails": 0, "meetings": 0, "total": 0}
        reps[oid][activity_type] += 1
        reps[oid]["total"] += 1
 
ranked = sorted(reps.items(), key=lambda x: x[1]["total"], reverse=True)
medals = ["\U0001F947", "\U0001F948", "\U0001F949"]
 
lines = []
for i, (oid, c) in enumerate(ranked):
    medal = medals[i] if i < 3 else f"{i + 1}."
    name = owner_map.get(oid, f"Owner {oid}")
    lines.append(f"{medal} *{name}* — {c['total']} activities ({c['calls']}C {c['emails']}E {c['meetings']}M)")
 
if not lines:
    print("No activity logged yesterday — skipping.")
    sys.exit(0)
 
# --- Post to Slack ---
date_str = yesterday.strftime("%A, %b %d")
blocks = [
    {"type": "header", "text": {"type": "plain_text", "text": "\U0001F3C6 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"}]},
]
 
print("Posting to Slack...")
slack = WebClient(token=SLACK_TOKEN)
result = slack.chat_postMessage(channel=SLACK_CHANNEL, text="Rep Activity Leaderboard", blocks=blocks, unfurl_links=False)
print(f"Posted leaderboard: {result['ts']}")

# Claude Code
/rep-leaderboard
 
# Or run directly
python .claude/skills/rep-leaderboard/scripts/leaderboard.py

# crontab -e
0 8 * * 1-5 cd /path/to/project && python .claude/skills/rep-leaderboard/scripts/leaderboard.py

name: Daily Rep Leaderboard
on:
  schedule:
    - cron: '0 13 * * 1-5'
  workflow_dispatch: {}
jobs:
  leaderboard:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with:
          python-version: '3.12'
      - run: pip install requests slack_sdk
      - run: python .claude/skills/rep-leaderboard/scripts/leaderboard.py
        env:
          HUBSPOT_TOKEN: ${{ secrets.HUBSPOT_TOKEN }}
          SLACK_BOT_TOKEN: ${{ secrets.SLACK_BOT_TOKEN }}
          SLACK_CHANNEL_ID: ${{ secrets.SLACK_CHANNEL_ID }}