How to Automate Slack Channel Reports using the API

1import os, time, logging, requests
2from collections import defaultdict
3from datetime import datetime, timedelta, timezone
4
5logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
6
7BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
8HEADERS = {"Authorization": f"Bearer {BOT_TOKEN}"}
9REPORT_CHANNEL = os.environ["SLACK_REPORT_CHANNEL_ID"]
10CHANNELS = [("C01234567", "team-general"), ("C02345678", "engineering")]
11
12def fetch_messages(channel_id, days_back=7):
13    oldest = (datetime.now(timezone.utc) - timedelta(days=days_back)).timestamp()
14    messages, cursor = [], None
15    while True:
16        payload = {"channel": channel_id, "limit": 200, "oldest": str(oldest)}
17        if cursor:
18            payload["cursor"] = cursor
19        resp = requests.post("https://slack.com/api/conversations.history",
20                             headers=HEADERS, json=payload)
21        if resp.status_code == 429:
22            wait = int(resp.headers.get("Retry-After", 60)) + 1
23            logging.warning(f"Rate limited — sleeping {wait}s (Tier 1 limit)")
24            time.sleep(wait)
25            continue
26        data = resp.json()
27        if not data.get("ok"):
28            logging.error(f"conversations.history error: {data.get('error')}")
29            break
30        messages.extend(data.get("messages", []))
31        if not data.get("has_more"):
32            break
33        cursor = data["response_metadata"]["next_cursor"]
34        time.sleep(1.2)
35    return messages
36
37def build_stats(messages, channel_name):
38    user_counts, messages_per_day = defaultdict(int), defaultdict(int)
39    threads = reactions = 0
40    for m in messages:
41        user_counts[m.get("user", "bot")] += 1
42        if m.get("reply_count", 0) > 0:
43            threads += 1
44        for r in m.get("reactions", []):
45            reactions += r.get("count", 0)
46        day = datetime.fromtimestamp(float(m.get("ts", 0)), tz=timezone.utc).strftime("%Y-%m-%d")
47        messages_per_day[day] += 1
48    top = sorted(user_counts.items(), key=lambda x: x[1], reverse=True)[:3]
49    return {"name": channel_name, "messages": len(messages),
50            "users": len(user_counts), "threads": threads,
51            "reactions": reactions, "top": top,
52            "avg": round(len(messages) / max(len(messages_per_day), 1), 1)}
53
54def post_report(stats, period):
55    top_text = "\n".join(f"<@{u}>: {c}" for u, c in stats["top"]) or "None"
56    blocks = [
57        {"type": "header", "text": {"type": "plain_text", "text": f"Weekly: #{stats['name']}"}},
58        {"type": "section", "fields": [
59            {"type": "mrkdwn", "text": f"*Messages:*\n{stats['messages']}"},
60            {"type": "mrkdwn", "text": f"*Users:*\n{stats['users']}"},
61            {"type": "mrkdwn", "text": f"*Threads:*\n{stats['threads']}"},
62            {"type": "mrkdwn", "text": f"*Reactions:*\n{stats['reactions']}"}
63        ]},
64        {"type": "section", "text": {"type": "mrkdwn", "text": f"*Top:*\n{top_text}"}},
65        {"type": "context", "elements": [{"type": "mrkdwn",
66             "text": f"Period: {period} | Avg {stats['avg']}/day"}]}
67    ]
68    resp = requests.post("https://slack.com/api/chat.postMessage",
69                         headers=HEADERS,
70                         json={"channel": REPORT_CHANNEL,
71                               "text": f"Report: #{stats['name']}", "blocks": blocks})
72    return resp.json()
73
74if __name__ == "__main__":
75    period = f"Last 7 days ending {datetime.now(timezone.utc).strftime('%Y-%m-%d')}"
76    for cid, cname in CHANNELS:
77        msgs = fetch_messages(cid, 7)
78        stats = build_stats(msgs, cname)
79        result = post_report(stats, period)
80        logging.info(f"Report for #{cname}: {result.get('ok')} ts={result.get('ts')}")
81        time.sleep(1.5)
82