This Python code defines a `SlackUserMessageMetrics` class designed to interact...

September 2, 2025 at 03:02 PM

import os
import time
import json
from datetime import datetime, timedelta
from slack_sdk import WebClient

class SlackUserMessageMetrics:
    def __init__(self, token):
        self.client = WebClient(token=token)
        
        # Calculate timestamp for 7 days ago
        seven_days_ago = datetime.now() - timedelta(days=7)
        self.oldest_timestamp = str(int(seven_days_ago.timestamp()))
    
    def get_licensed_users(self):
        users = []
        cursor = None
        
        while True:
            result = self.client.users_list(
                limit=1000,
                cursor=cursor
            )
            
            # Filter for non-bot, non-deleted licensed users
            licensed_users = [
                user for user in result['members']
                if not user.get('deleted') and 
                   not user.get('is_bot') and
                   user.get('id') != 'USLACKBOT'
            ]
            
            users.extend(licensed_users)
            cursor = result.get('response_metadata', {}).get('next_cursor')
            
            if not cursor:
                break
        
        return users
    
    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_user_messages_in_conversation(self, conversation_id, user_id):
        public_messages = 0
        direct_messages = 0
        cursor = None
        
        while True:
            try:
                result = self.client.conversations_history(
                    channel=conversation_id,
                    limit=1000,
                    cursor=cursor,
                    oldest=self.oldest_timestamp
                )
                
                # Filter messages by user
                user_messages = [
                    msg for msg in result['messages']
                    if msg.get('user') == user_id and
                       msg.get('type') == 'message' and 
                       not msg.get('subtype')
                ]
                
                # Get conversation info to determine type
                conv_info = self.client.conversations_info(channel=conversation_id)
                conv = conv_info['channel']
                
                if conv.get('is_channel') and not conv.get('is_private'):
                    public_messages += len(user_messages)
                elif conv.get('is_im') or conv.get('is_mpim'):
                    direct_messages += len(user_messages)
                
                cursor = result.get('response_metadata', {}).get('next_cursor')
                
                time.sleep(0.5)
                
                if not cursor:
                    break
            except:
                break
        
        return public_messages, direct_messages
    
    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 get_user_metrics(self, user_id, conversations):
        public_total = 0
        direct_total = 0
        
        # Process public channels
        for conv in conversations['public_channels']:
            try:
                result = self.client.conversations_history(
                    channel=conv['id'],
                    limit=1000,
                    oldest=self.oldest_timestamp
                )
                
                user_messages = [
                    msg for msg in result['messages']
                    if msg.get('user') == user_id and
                       msg.get('type') == 'message' and 
                       not msg.get('subtype')
                ]
                
                public_total += len(user_messages)
                time.sleep(0.5)
            except:
                continue
        
        # Process direct messages and group DMs
        for conv_list in [conversations['direct_messages'], conversations['group_direct_messages']]:
            for conv in conv_list:
                try:
                    result = self.client.conversations_history(
                        channel=conv['id'],
                        limit=1000,
                        oldest=self.oldest_timestamp
                    )
                    
                    user_messages = [
                        msg for msg in result['messages']
                        if msg.get('user') == user_id and
                           msg.get('type') == 'message' and 
                           not msg.get('subtype')
                    ]
                    
                    direct_total += len(user_messages)
                    time.sleep(0.5)
                except:
                    continue
        
        return {
            'public_channel_messages': public_total,
            'direct_messages': direct_total,
            'total_messages': public_total + direct_total
        }
    
    def gather_all_user_metrics(self):
        users = self.get_licensed_users()
        conversations = self.get_all_conversations()
        categorized_conversations = self.categorize_conversations(conversations)
        
        user_metrics = {}
        
        for user in users:
            user_id = user['id']
            user_name = user.get('real_name') or user.get('name', user_id)
            user_email = user.get('profile', {}).get('email', '')
            
            metrics = self.get_user_metrics(user_id, categorized_conversations)
            
            user_metrics[user_id] = {
                'user_name': user_name,
                'user_email': user_email,
                'public_channel_messages': metrics['public_channel_messages'],
                'direct_messages': metrics['direct_messages'],
                'total_messages': metrics['total_messages']
            }
        
        return user_metrics


def main():
    slack_token = os.getenv('SLACK_BOT_TOKEN')
    
    if not slack_token:
        return {'error': 'SLACK_BOT_TOKEN environment variable not set'}
    
    collector = SlackUserMessageMetrics(slack_token)
    metrics = collector.gather_all_user_metrics()
    
    # Add summary metadata
    result = {
        'period': 'Last 7 days',
        'timestamp': datetime.now().isoformat(),
        'user_count': len(metrics),
        'users': metrics
    }
    
    return result


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

This Python code defines a SlackUserMessageMetrics class designed to interact with the Slack API and compute message metrics for Slack users. Here is what the code does step-by-step:

Purpose:

The script collects and processes metrics related to user messages in a Slack workspace over the last 7 days. It fetches message counts across different conversation types (e.g., public channels, direct messages, and group direct messages) for each user.

Code Breakdown:

Imports Required Libraries:
Libraries such as os, time, json, datetime, and slack_sdk are imported for accessing environment variables, time delays, JSON manipulation, date handling, and Slack API functionality.
Class Initialization:
The SlackUserMessageMetrics class is initialized with a Slack token to authorize API requests. It also calculates the Unix timestamp for 7 days prior (oldest_timestamp) to filter messages from that point onward.
Fetch Licensed Users:
- The get_licensed_users method retrieves all users in the Slack workspace, excluding bots, deleted users, and the system-provided USLACKBOT.
- Results are paginated using cursor.
Fetch All Conversations:
- The get_all_conversations method retrieves all workspace conversations (public channels, private channels, direct messages, and group direct messages) using the conversations_list API.
- Conversations are paginated using cursor.
Get User Messages for a Conversation:
- The get_user_messages_in_conversation method computes the number of public messages and direct messages a specific user sent in a specific conversation within the last 7 days.
- It determines whether the conversation is a public channel, direct message, or group DM to categorize the message type.
Categorize Conversations:
- The categorize_conversations method organizes all retrieved Slack conversations into three categories:
  - public_channels: Public channels viewable by everyone.
  - direct_messages: 1-on-1 direct messages.
  - group_direct_messages: Multi-person direct messages.
Compute a User's Message Metrics:
- The get_user_metrics method calculates the total messages a user has sent across all public channels, direct messages, and group direct messages.
- For each conversation type, it calls the Slack API to fetch message history and filters messages authored by the given user.
Compute Metrics for All Users:
- The gather_all_user_metrics method combines the above processes to gather message statistics for all licensed users:
  1. Fetches all users.
  2. Fetches and categorizes all conversations.
  3. Computes metrics for each user (total messages, public channel messages, and direct messages).
- User data such as real_name, email, and metrics is consolidated into a dictionary.
Main Function:
- The main function checks for the SLACK_BOT_TOKEN environment variable to authenticate the script with the Slack workspace.
- If valid, it creates an instance of SlackUserMessageMetrics, gathers user metrics (gather_all_user_metrics), and packages the results in a final dictionary:
  - Period: Last 7 days.
  - Current timestamp.
  - Total user count.
  - Metrics for each user.
- The result is printed as a formatted JSON string.

Output:

The script outputs a JSON object containing:

The analysis time range (Last 7 days).
Total number of licensed (non-bot, active) users analyzed.
Detailed message metrics (public, direct, total) for each user, along with their name and email.

Key Functionality:

Uses Slack API methods:
- users_list: Fetch workspace users.
- conversations_list: Fetch workspace conversations.
- conversations_info: Retrieve metadata about a conversation.
- conversations_history: Fetch message history for a conversation.
Filters Messages:
- Filters by user, Slack message type, and time period (7 days).
Aggregates User Metrics:
- Categorizes and aggregates messages sent across conversation types.

Practical Use:

The code is useful for Slack workspace admins or analysts looking to monitor user activity, message trends, or engagement within their organization.

Generate your own explanations

Download our vscode extension

Read other generated explanations

Built by @thebuilderjr