#!/usr/bin/env python3 """ Fixed Multi-user Slack API client for presence tracking Uses individual users.getPresence calls since batch API is deprecated """ from slack_sdk import WebClient from slack_sdk.errors import SlackApiError from datetime import datetime import time from config_multi_user import ( SLACK_BOT_TOKEN, TIMEZONE, setup_logging, TRACK_ALL_USERS, TRACK_DOMAINS, EXCLUDED_USER_IDS, SLACK_USER_IDS ) class MultiUserSlackClient: """Wrapper for Slack API presence operations for multiple users""" def __init__(self): """Initialize the Slack client""" self.logger = setup_logging('slack_client_multi') self.client = WebClient(token=SLACK_BOT_TOKEN) self._validate_connection() self.user_cache = {} # Cache user info to reduce API calls self.users_to_track = self._determine_users_to_track() def _validate_connection(self): """Validate the Slack connection on initialization""" try: auth_response = self.client.auth_test() self.logger.info(f"Connected to Slack workspace: {auth_response['team']}") self.workspace_id = auth_response['team_id'] except SlackApiError as e: self.logger.error(f"Failed to connect to Slack: {e.response['error']}") raise def _determine_users_to_track(self): """Determine which users to track based on configuration""" if SLACK_USER_IDS: # Specific user IDs configured self.logger.info(f"Tracking specific users: {SLACK_USER_IDS}") return SLACK_USER_IDS elif TRACK_ALL_USERS or TRACK_DOMAINS: # Will determine dynamically based on workspace users self.logger.info("Will determine users dynamically") return self._get_workspace_users() else: return [] def _get_workspace_users(self): """Get list of users to track from workspace""" try: users_to_track = [] # Get all users (without presence - that's deprecated) response = self.client.users_list(limit=200) users = response['members'] # Handle pagination while response.get('response_metadata', {}).get('next_cursor'): cursor = response['response_metadata']['next_cursor'] response = self.client.users_list(cursor=cursor, limit=200) users.extend(response['members']) for user in users: # Skip bots, deleted users, and excluded users if user.get('is_bot') or user.get('deleted') or user['id'] in EXCLUDED_USER_IDS: continue # Skip Slackbot if user['id'] == 'USLACKBOT': continue # Apply filtering based on configuration should_track = False if TRACK_ALL_USERS: should_track = True elif TRACK_DOMAINS: email = user.get('profile', {}).get('email', '') if any(email.endswith(f"@{domain}") for domain in TRACK_DOMAINS): should_track = True if should_track: users_to_track.append(user['id']) # Cache user info self.user_cache[user['id']] = { 'name': user.get('real_name', user.get('name', 'Unknown')), 'email': user.get('profile', {}).get('email', ''), 'display_name': user.get('profile', {}).get('display_name', '') } self.logger.info(f"Found {len(users_to_track)} users to track") return users_to_track except SlackApiError as e: self.logger.error(f"Failed to get workspace users: {e.response['error']}") return [] def get_all_users_presence(self): """ Get presence for all configured users using individual API calls Since batch API (users.list with presence) is deprecated Returns: list: List of presence data dictionaries for each user """ check_time = datetime.now(TIMEZONE) results = [] if not self.users_to_track: self.logger.warning("No users to track") return results self.logger.info(f"Checking presence for {len(self.users_to_track)} users") # Track API call timing start_time = time.time() successful_checks = 0 failed_checks = 0 for i, user_id in enumerate(self.users_to_track): try: # Get presence for this user presence_response = self.client.users_getPresence(user=user_id) # Get user info if not cached if user_id not in self.user_cache: try: user_info = self.client.users_info(user=user_id) self.user_cache[user_id] = { 'name': user_info['user'].get('real_name', 'Unknown'), 'email': user_info['user'].get('profile', {}).get('email', ''), 'display_name': user_info['user'].get('profile', {}).get('display_name', '') } except: self.user_cache[user_id] = {'name': 'Unknown', 'email': '', 'display_name': ''} # Build presence data presence_data = { 'timestamp': check_time.isoformat(), 'user_id': user_id, 'user_name': self.user_cache[user_id]['name'], 'presence': presence_response.get('presence', 'away'), 'auto_away': presence_response.get('auto_away', False), 'manual_away': presence_response.get('manual_away', False), 'last_activity': presence_response.get('last_activity', None) } results.append(presence_data) successful_checks += 1 # Log progress every 5 users if (i + 1) % 5 == 0: self.logger.debug(f"Checked {i + 1}/{len(self.users_to_track)} users") # Rate limiting - Slack allows ~50 requests per minute # With 10 users, we can afford 0.1 second delay # With 50 users, we need to be more careful if len(self.users_to_track) > 30: time.sleep(1.5) # Slower for large teams elif len(self.users_to_track) > 10: time.sleep(0.5) # Medium delay else: time.sleep(0.1) # Fast for small teams except SlackApiError as e: error_code = e.response.get('error', 'unknown') # Handle specific errors if error_code == 'user_not_found': self.logger.warning(f"User {user_id} not found - may have been deactivated") failed_checks += 1 elif error_code == 'rate_limited': # Rate limited - wait and retry retry_after = int(e.response.headers.get('Retry-After', 10)) self.logger.warning(f"Rate limited. Waiting {retry_after} seconds...") time.sleep(retry_after) # Retry this user try: presence_response = self.client.users_getPresence(user=user_id) presence_data = { 'timestamp': check_time.isoformat(), 'user_id': user_id, 'user_name': self.user_cache.get(user_id, {}).get('name', 'Unknown'), 'presence': presence_response.get('presence', 'away'), 'auto_away': presence_response.get('auto_away', False), 'manual_away': presence_response.get('manual_away', False), 'last_activity': presence_response.get('last_activity', None) } results.append(presence_data) successful_checks += 1 except: failed_checks += 1 else: self.logger.error(f"Failed to get presence for {user_id}: {error_code}") failed_checks += 1 except Exception as e: self.logger.error(f"Unexpected error for {user_id}: {e}") failed_checks += 1 # Log summary elapsed_time = time.time() - start_time self.logger.info( f"Presence check complete in {elapsed_time:.1f}s - " f"Success: {successful_checks}, Failed: {failed_checks}" ) # Count active users active_count = sum(1 for r in results if r['presence'] == 'active') self.logger.info(f"Active users: {active_count}/{len(results)}") return results def get_user_info(self, user_id): """Get cached user info or fetch if needed""" if user_id not in self.user_cache: try: response = self.client.users_info(user=user_id) self.user_cache[user_id] = { 'name': response['user'].get('real_name', 'Unknown'), 'email': response['user'].get('profile', {}).get('email', ''), 'display_name': response['user'].get('profile', {}).get('display_name', '') } except SlackApiError: self.user_cache[user_id] = {'name': 'Unknown', 'email': '', 'display_name': ''} return self.user_cache[user_id] # Singleton instance _client_instance = None def get_slack_client(): """Get or create a singleton Slack client instance""" global _client_instance if _client_instance is None: _client_instance = MultiUserSlackClient() return _client_instance