This code is a Python application using the `slack_sdk` to...

September 2, 2025 at 02:58 PM

import os
import time
import json
from slack_sdk import WebClient

class SlackMessageMetrics:
    def __init__(self, token):
        self.client = WebClient(token=token)
    
    def get_all_conversations(self):
        conversations = []
        cursor = None
        
        while True:
            result = self.client.conversations_list(
                limit=1000,
                cursor=cursor,
                types='public_channel,private_channel,mpim,im'
            )
            
            conversations.extend(result['channels'])
            cursor = result.get('response_metadata', {}).get('next_cursor')
            
            if not cursor:
                break
        
        return conversations
    
    def get_message_count(self, conversation_id):
        message_count = 0
        cursor = None
        
        while True:
            try:
                result = self.client.conversations_history(
                    channel=conversation_id,
                    limit=1000,
                    cursor=cursor
                )
                
                user_messages = [
                    msg for msg in result['messages']
                    if not msg.get('bot_id') and 
                       msg.get('type') == 'message' and 
                       not msg.get('subtype')
                ]
                
                message_count += len(user_messages)
                cursor = result.get('response_metadata', {}).get('next_cursor')
                
                time.sleep(1)
                
                if not cursor:
                    break
            except:
                break
        
        return message_count
    
    def categorize_conversations(self, conversations):
        categorized = {
            'public_channels': [],
            'direct_messages': [],
            'group_direct_messages': []
        }
        
        for conv in conversations:
            if conv.get('is_channel') and not conv.get('is_private'):
                categorized['public_channels'].append(conv)
            elif conv.get('is_im'):
                categorized['direct_messages'].append(conv)
            elif conv.get('is_mpim'):
                categorized['group_direct_messages'].append(conv)
        
        return categorized
    
    def process_conversations(self, conversations):
        total_count = 0
        
        for conv in conversations:
            message_count = self.get_message_count(conv['id'])
            total_count += message_count
        
        return total_count
    
    def gather_metrics(self, include_group_dms=True):
        all_conversations = self.get_all_conversations()
        categorized = self.categorize_conversations(all_conversations)
        
        public_channel_messages = self.process_conversations(categorized['public_channels'])
        direct_messages = self.process_conversations(categorized['direct_messages'])
        
        if include_group_dms:
            group_dm_messages = self.process_conversations(categorized['group_direct_messages'])
            direct_messages += group_dm_messages
        
        total_messages = public_channel_messages + direct_messages
        
        return {
            'public_channel_messages': public_channel_messages,
            'direct_messages': direct_messages,
            'total_messages': total_messages
        }


def main():
    slack_token = os.getenv('SLACK_BOT_TOKEN')
    
    if not slack_token:
        return {'error': 'SLACK_BOT_TOKEN environment variable not set'}
    
    collector = SlackMessageMetrics(slack_token)
    metrics = collector.gather_metrics()
    
    return metrics


if __name__ == "__main__":
    result = main()
    print(json.dumps(result, indent=2))

This code is a Python application using the slack_sdk to collect metrics on Slack conversations and messages. Below is a breakdown of its functionality:

1. Initialization (`SlackMessageMetrics` class):

The class initializes a Slack WebClient instance using a token, allowing for API requests to interact with Slack.

2. Fetching Conversations:

get_all_conversations(): Retrieves all Slack conversations (channels, direct messages, group DMs) using conversations.list from the Slack API. It supports pagination using cursor to handle large datasets.
- It specifically looks for public/private channels, direct messages (DMs), and multi-party instant messages (group DMs).
- The function collects the entire list of conversations.

3. Counting Messages in a Conversation:

get_message_count(conversation_id): Counts user messages within a specific conversation (channel, direct message, or group DM):
- Uses the conversations.history API to retrieve all messages from a conversation.
- Filters out bot-generated messages and system messages (type != "message" or subtype set).
- Applies pagination via next_cursor for conversations with many messages.
- Introduces a 1-second delay (time.sleep(1)) between requests to avoid hitting Slack rate limits.
- Returns a message count for the conversation.

4. Categorizing Conversations:

categorize_conversations(conversations): Organizes conversations into three groups:
- public_channels: Channels that are public.
- direct_messages: One-to-one DMs.
- group_direct_messages: Multi-participant DMs.

5. Processing Conversations:

process_conversations(conversations): Loops through a list of conversations, calls get_message_count for each, and accumulates the total message count.
Allows processing of either public channels, DMs, or group DMs.

6. Gathering Metrics:

gather_metrics(include_group_dms=True):
- Fetches and categorizes all conversations using get_all_conversations and categorize_conversations.
- Counts messages across public channels, direct messages, and optionally group DMs.
- Totals the message counts from all groups.
- Returns a dictionary summarizing:
  - Messages in public channels.
  - Messages in DMs (including group chats if enabled).
  - Total number of messages.

7. Main Entry Point:

main() Function:
- Retrieves the SLACK_BOT_TOKEN from the environment, which is needed for the Slack API.
- If the token is not available, returns an error message.
- Creates an instance of SlackMessageMetrics and gathers metrics via gather_metrics.
- Outputs the result (formatted as JSON).

8. Program Execution:

When the script is run directly (__name__ == "__main__"):
- The main() function is executed, and the metrics are printed as a JSON object.

Summary of Output:

The program outputs a JSON object with metrics reflecting the distribution of Slack messages, e.g.:

{
  "public_channel_messages": 12345,
  "direct_messages": 6789,
  "total_messages": 19134
}

Use Cases:

Analyze how active different Slack channels or conversations are.
Track overall usage and activity within an organization's Slack workspace.
Generate high-level reports about communication patterns.

Generate your own explanations

Download our vscode extension

Read other generated explanations

Built by @thebuilderjr