This code is a Python script that interacts with the...

September 2, 2025 at 03:52 PM

import os import time import json import datetime import requests SLACK_TOKEN = os.environ.get("SLACK_TOKEN") or os.environ.get("SLACK_BOT_TOKEN") SLACK_API_BASE = os.environ.get("SLACK_API_BASE", "https://slack.com/api") LOOKBACK_DAYS = int(os.environ.get("LOOKBACK_DAYS", "7")) SESSION = requests.Session() if SLACK_TOKEN: SESSION.headers.update({"Authorization": f"Bearer {SLACK_TOKEN}"}) SESSION.headers.update({"Content-Type": "application/x-www-form-urlencoded"}) EXCLUDE_SUBTYPES = { "bot_message", "channel_join", "channel_leave", "channel_topic", "channel_purpose", "channel_name", "tombstone", "message_deleted", "file_comment", } def api_call(method: str, params: dict | None = None) -> dict: url = f"{SLACK_API_BASE}/{method}" while True: resp = SESSION.post(url, data=params or {}) if resp.status_code == 429: time.sleep(int(resp.headers.get("Retry-After", "1")) + 1) continue data = resp.json() if not data.get("ok"): # Minimal failure signaling for automations raise RuntimeError(f"{method} failed: {data}") return data def paginate(method: str, params: dict, list_key: str): cursor = None while True: p = dict(params or {}) if cursor: p["cursor"] = cursor data = api_call(method, p) for item in data.get(list_key, []): yield item cursor = data.get("response_metadata", {}).get("next_cursor") if not cursor: break def get_licensed_users() -> dict: users = {} for u in paginate("users.list", {"limit": 200}, "members"): if u.get("deleted"): continue if u.get("is_bot") or u.get("is_app_user"): continue if u.get("is_restricted") or u.get("is_ultra_restricted"): # guests continue uid = u["id"] profile = u.get("profile", {}) or {} display = profile.get("display_name_normalized") or profile.get("display_name") \ or profile.get("real_name_normalized") or profile.get("real_name") \ or u.get("name") or uid real = profile.get("real_name") or display users[uid] = {"id": uid, "display_name": display, "real_name": real} return users def list_conversation_ids(types: str) -> list[str]: return [c["id"] for c in paginate("conversations.list", {"types": types, "limit": 1000}, "channels")] def iter_history_with_replies(channel_id: str, oldest_ts: float, latest_ts: float): for m in paginate( "conversations.history", {"channel": channel_id, "oldest": oldest_ts, "latest": latest_ts, "limit": 1000, "inclusive": True}, "messages", ): yield m if int(m.get("reply_count", 0)) > 0: parent_ts = m.get("ts") if parent_ts: for r in paginate( "conversations.replies", { "channel": channel_id, "ts": parent_ts, "oldest": oldest_ts, "latest": latest_ts, "limit": 200, "inclusive": True, }, "messages", ): if r.get("ts") == parent_ts: continue yield r def is_countable_user_message(msg: dict, allowed_user_ids: set[str]) -> str | None: if msg.get("type") != "message": return None subtype = msg.get("subtype") if subtype in EXCLUDE_SUBTYPES: return None uid = msg.get("user") if uid and uid in allowed_user_ids: # exclude true bot posts that sometimes include 'user' but also have bot_id if msg.get("bot_id"): return None return uid return None def main(): if not SLACK_TOKEN: print(json.dumps({"error": "Missing SLACK_TOKEN"})) return now_ts = time.time() oldest_ts = now_ts - LOOKBACK_DAYS * 86400 users = get_licensed_users() allowed = set(users.keys()) public_counts = {uid: 0 for uid in allowed} dm_counts = {uid: 0 for uid in allowed} # Public channels for cid in list_conversation_ids("public_channel"): for msg in iter_history_with_replies(cid, oldest_ts, now_ts): uid = is_countable_user_message(msg, allowed) if uid: public_counts[uid] += 1 # DMs (IMs and multi-person IMs) for cid in list_conversation_ids("im,mpim"): for msg in iter_history_with_replies(cid, oldest_ts, now_ts): uid = is_countable_user_message(msg, allowed) if uid: dm_counts[uid] += 1 start_iso = datetime.datetime.fromtimestamp(oldest_ts, tz=datetime.timezone.utc).isoformat() end_iso = datetime.datetime.fromtimestamp(now_ts, tz=datetime.timezone.utc).isoformat() results = [] for uid in sorted(allowed): dm = dm_counts.get(uid, 0) pub = public_counts.get(uid, 0) results.append( { "user_id": uid, "display_name": users[uid]["display_name"], "real_name": users[uid]["real_name"], "dm_messages": dm, "public_messages": pub, "total_messages": dm + pub, "start_utc": start_iso, "end_utc": end_iso, "lookback_days": LOOKBACK_DAYS, } ) print(json.dumps({"results": results}, indent=2)) if __name__ == "__main__": main()

This code is a Python script that interacts with the Slack API to analyze and generate a report of user messages in public Slack channels and direct messages (DM/IM), over a configurable lookback period.

Key Steps in the Code:

  1. Environment Configuration:

    • Fetches configuration values from environment variables:
      • SLACK_TOKEN or SLACK_BOT_TOKEN: The API token used to authenticate against the Slack API.
      • SLACK_API_BASE: The base Slack API URL (defaults to https://slack.com/api).
      • LOOKBACK_DAYS: Defines how many days back to look for message histories (defaults to 7).

  2. Slack API Request Helpers:

    • api_call(): A function to send API requests to Slack endpoints. It handles retry logic for rate-limited errors (HTTP 429).
    • paginate(): Handles pagination for Slack API calls where responses return partial data. It iteratively retrieves all pages of a result set using Slack's next_cursor value.

  3. User and Channel Data Collection:

    • get_licensed_users(): Retrieves a list of all active, non-bot, non-guest Slack users from the users.list API endpoint.
    • list_conversation_ids(): Retrieves a list of channel IDs for either public channels or direct/multi-person messaging channels.

  4. Message Analysis:

    • iter_history_with_replies(): Iterates through the message history of a channel (both parent messages and replies) using conversations.history and conversations.replies API endpoints.
    • is_countable_user_message(): Determines whether a message is considered a "countable" user message, filtering out certain system/bot messages and irrelevant subtypes.

  5. Processing Message Data:

    • Loops through public channels, direct messages (im), and multi-person DMs (mpim), counting messages for each user that meet the criteria for "countable" messages.
    • Counts are separated into public (public_counts) and direct message (dm_counts) categories.

  6. Results and Reporting:

    • Compiles the message counts for each user into a list of result dictionaries, including:
      • User ID, display name, real name.
      • Counts of public and DM messages, and the total message count.
      • The time range covered by the analysis, in UTC ISO format.
      • The configured lookback period in days.
    • Outputs the final results as a formatted JSON object, printed to standard output.

Example Output:

The output is a JSON object, for example:

{
  "results": [
    {
      "user_id": "U12345",
      "display_name": "JohnDoe",
      "real_name": "John Doe",
      "dm_messages": 10,
      "public_messages": 25,
      "total_messages": 35,
      "start_utc": "2023-10-01T00:00:00+00:00",
      "end_utc": "2023-10-08T00:00:00+00:00",
      "lookback_days": 7
    },
    ...
  ]
}

Purpose and Use Case:

This script is likely intended for analyzing Slack user activity, such as:

  • Monitoring employee engagement by counting messages in public channels and private messages.
  • Generating activity summaries for reporting or audits.
  • Providing insights into Slack usage over a defined timeframe.

Preconditions:

  • A valid Slack API token with sufficient permissions is required.
  • The environment variables (SLACK_TOKEN, LOOKBACK_DAYS, etc.) need to be set before running the script.

Notes:

  • The code filters out certain message "subtypes" (e.g., system-generated messages) and bot messages to focus only on relevant user-generated content.
  • Rate-limited API calls (HTTP 429) are handled by retrying after the Retry-After duration specified by Slack.
Generate your own explanations
Download our vscode extension
Read other generated explanations

Built by @thebuilderjr
Sponsored by beam analytics
Read our terms and privacy policy
Forked from openai-quickstart-node