This Python code defines a `SlackMessageMetrics` class and a `main`...

September 2, 2025 at 02:57 PM

import os
import time
import logging
from typing import Dict, List, Optional, Tuple
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

# Set up logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class SlackMessageMetrics:
    def __init__(self, token: str):
        self.client = WebClient(token=token)
        self.metrics = {
            'public_channel_messages': 0,
            'direct_messages': 0,
            'total_messages': 0
        }
    
    def get_all_conversations(self) -> List[Dict]:
        conversations = []
        cursor = None
        
        while True:
            try:
                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
                    
            except SlackApiError as e:
                logger.error(f"Error fetching conversations: {e}")
                raise e
        
        return conversations
    
    def get_message_count(self, conversation_id: str, conversation_type: str) -> int:
        message_count = 0
        cursor = None
        
        while True:
            try:
                result = self.client.conversations_history(
                    channel=conversation_id,
                    limit=1000,
                    cursor=cursor
                )
                
                # Filter out bot messages and system messages
                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')
                
                # Add delay to respect rate limits
                time.sleep(1)
                
                if not cursor:
                    break
                    
            except SlackApiError as e:
                if e.response.get('error') == 'not_in_channel':
                    logger.info(f"Skipping {conversation_id} - bot not in channel")
                    break
                logger.error(f"Error fetching messages for {conversation_id}: {e.response.get('error')}")
                break
        
        return message_count
    
    def categorize_conversations(self, conversations: List[Dict]) -> Dict[str, List[Dict]]:
        categorized = {
            'public_channels': [],
            'private_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_channel') and conv.get('is_private'):
                categorized['private_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: List[Dict], conv_type: str) -> int:
        total_count = 0
        
        logger.info(f"Processing {len(conversations)} {conv_type}...")
        
        for i, conv in enumerate(conversations, 1):
            conv_name = conv.get('name', conv['id'])
            logger.info(f"Processing {conv_type} {i}/{len(conversations)}: {conv_name}")
            
            message_count = self.get_message_count(conv['id'], conv_type)
            total_count += message_count
            
            logger.info(f"  Found {message_count} messages")
        
        return total_count
    
    def gather_metrics(self, options: Optional[Dict] = None) -> Dict[str, int]:
        if options is None:
            options = {}
        
        include_private_channels = options.get('include_private_channels', False)
        include_group_dms = options.get('include_group_dms', True)
        
        try:
            logger.info("Fetching all conversations...")
            all_conversations = self.get_all_conversations()
            
            logger.info(f"Found {len(all_conversations)} total conversations")
            
            categorized = self.categorize_conversations(all_conversations)
            
            logger.info("Conversation breakdown:")
            logger.info(f"  Public channels: {len(categorized['public_channels'])}")
            logger.info(f"  Private channels: {len(categorized['private_channels'])}")
            logger.info(f"  Direct messages: {len(categorized['direct_messages'])}")
            logger.info(f"  Group DMs: {len(categorized['group_direct_messages'])}")
            
            # Count public channel messages
            self.metrics['public_channel_messages'] = self.process_conversations(
                categorized['public_channels'], 
                'public channels'
            )
            
            # Count direct messages (1:1 DMs)
            dm_count = self.process_conversations(
                categorized['direct_messages'], 
                'direct messages'
            )
            
            # Optionally include group DMs in DM count
            group_dm_count = 0
            if include_group_dms:
                group_dm_count = self.process_conversations(
                    categorized['group_direct_messages'], 
                    'group direct messages'
                )
            
            self.metrics['direct_messages'] = dm_count + group_dm_count
            
            # Optionally include private channels
            if include_private_channels:
                private_channel_count = self.process_conversations(
                    categorized['private_channels'], 
                    'private channels'
                )
                self.metrics['public_channel_messages'] += private_channel_count
            
            self.metrics['total_messages'] = (
                self.metrics['public_channel_messages'] + 
                self.metrics['direct_messages']
            )
            
            return self.metrics
            
        except Exception as e:
            logger.error(f"Error gathering metrics: {e}")
            raise e
    
    def generate_report(self) -> str:
        total = self.metrics['total_messages']
        public_percent = (
            (self.metrics['public_channel_messages'] / total) * 100 
            if total > 0 else 0
        )
        dm_percent = (
            (self.metrics['direct_messages'] / total) * 100 
            if total > 0 else 0
        )
        
        ratio = (
            self.metrics['public_channel_messages'] / 
            max(self.metrics['direct_messages'], 1)
        )
        
        report = f"""
        
        return report


def main():
    """
    Main function to run the Slack metrics collection.
    
    Returns:
        Dictionary with success status and data/error
    """
    # Get Slack token from environment variable
    slack_token = os.getenv('SLACK_BOT_TOKEN')
    
    if not slack_token:
        error_msg = "Please set your Slack bot token in the SLACK_BOT_TOKEN environment variable"
        logger.error(error_msg)
        return {
            'success': False,
            'error': error_msg
        }
    
    metrics_collector = SlackMessageMetrics(slack_token)
    
    try:
        logger.info("Starting Slack message metrics collection...")
        
        # Configure options
        options = {
            'include_private_channels': False,  # Set to True to include private channels in public count
            'include_group_dms': True,         # Include group DMs in direct message count
        }
        
        metrics = metrics_collector.gather_metrics(options)
        
        print(metrics_collector.generate_report())
        
        # Return structured data
        return {
            'success': True,
            'data': metrics
        }
        
    except Exception as e:
        error_msg = f"Failed to collect metrics: {str(e)}"
        logger.error(error_msg)
        return {
            'success': False,
            'error': error_msg
        }


if __name__ == "__main__":
    import sys
    
    result = main()
    sys.exit(0 if result['success'] else 1)


# Example usage as a module:
"""
from slack_metrics import SlackMessageMetrics

# Initialize with your token
collector = SlackMessageMetrics('xoxb-your-bot-token')

# Gather metrics with custom options
options = {
    'include_private_channels': True,
    'include_group_dms': True
}

metrics = collector.gather_metrics(options)
print(collector.generate_report())

# Access raw data
print(f"Public messages: {metrics['public_channel_messages']}")
print(f"Direct messages: {metrics['direct_messages']}")
"""

This Python code defines a SlackMessageMetrics class and a main function to gather and report message metrics from Slack using the Slack API. Here's an explanation of what the code does:

Functionality:

Slack API Integration:
- The script uses the slack_sdk library to interact with Slack's Web API.
- It requires a bot token (SLACK_BOT_TOKEN) to authenticate requests.
Metrics Gathering:
- The script collects message counts from various types of Slack conversations:
  - Public channels
  - Private channels (optional)
  - Direct messages (DMs)
  - Group DMs (multi-person DMs or MPIMs)
- It distinguishes between conversation types and aggregates message counts while filtering out bot and system messages.
Message Categorization:
- Slack conversations (channels) are categorized into:
  - Public channels (is_channel and not is_private)
  - Private channels (is_channel and is_private)
  - Direct messages (is_im)
  - Group DMs (is_mpim)
Message Count Retrieval:
- The script fetches and counts Slack messages for each conversation using conversations_history, respecting rate limits by introducing delays (time.sleep(1)).
- It handles exceptions like not_in_channel where the bot isn't a member of the channel.
Metrics Aggregation:
- It calculates:
  - Total messages in public channels
  - Total direct messages, including optional group DMs
- Optionally includes private channel messages in public message totals.
- Computes the total number of messages across all categories.
Reporting:
- Generates a textual report with:
  - Total messages
  - Percentage breakdown of public vs. DM messages
  - Ratio of public to direct messages
Logging:
- Logs progress and errors using the logging module for debugging and traceability.
CLI Integration:
- The script has a main function intended for command-line execution. It retrieves the Slack token from an environment variable (SLACK_BOT_TOKEN), processes metrics, and reports results.
Error Handling:
- Errors are logged (e.g., Slack API errors, missing environment variables), and execution gracefully terminates if issues occur.

Classes and Functions:

SlackMessageMetrics:
- __init__: Initializes the Slack WebClient and prepares a dictionary to hold message counts.
- get_all_conversations: Retrieves all Slack conversations (public, private channels, DMs, MPIMs).
- get_message_count: Counts non-bot messages in a specified channel or conversation.
- categorize_conversations: Categorizes conversations into public channels, private channels, DMs, and MPIMs.
- process_conversations: Counts messages for all conversations of a specific type.
- gather_metrics: Orchestrates message collection and aggregation, with optional settings to include private channels or MPIMs.
- generate_report: Formats and generates a summary report of the message metrics.
main Function:
- Fetches the Slack bot token (SLACK_BOT_TOKEN) from an environment variable.
- Initializes a SlackMessageMetrics instance and gathers metrics with configurable options.
- Prints the generated report and handles errors gracefully.
Script Execution:
- Runs the main function when executed as a standalone script, returning 0 for success or 1 for failure.

Example Workflow:

The script fetches Slack bot token from the environment and establishes a connection with Slack.
It retrieves the list of all Slack conversations and categorizes them.
For each category (e.g., public channels, DMs), it counts user messages.
Depending on configured options, it includes or excludes private channels or group DMs.
Aggregates totals and generates a comprehensive report of message statistics.
Outputs the report and logs relevant information.

Use Case:

This script is useful for analyzing Slack message activity, such as:
- Tracking engagement in public channels vs. DMs.
- Reporting metrics for team activity in an organization.
- Providing insights into team communication patterns.

This script is ready for running as a CLI tool or as a Python module for integration into broader systems.

Generate your own explanations

Download our vscode extension

Read other generated explanations

Built by @thebuilderjr