#!/usr/bin/env python3 """ User mapping system for better identification Maps Slack user IDs to readable names and metadata """ import json import csv from pathlib import Path from slack_sdk import WebClient from slack_sdk.errors import SlackApiError from config_multi_user import SLACK_BOT_TOKEN, setup_logging, BASE_DIR class UserMapper: """Manages user ID to name mappings and metadata""" def __init__(self): """Initialize the user mapper""" self.logger = setup_logging('user_mapper') self.client = WebClient(token=SLACK_BOT_TOKEN) self.mapping_file = BASE_DIR / "user_mappings.json" self.csv_mapping_file = BASE_DIR / "user_directory.csv" self.user_map = self.load_mappings() def load_mappings(self): """Load existing user mappings from file""" if self.mapping_file.exists(): try: with open(self.mapping_file, 'r') as f: mappings = json.load(f) self.logger.info(f"Loaded {len(mappings)} user mappings") return mappings except Exception as e: self.logger.error(f"Failed to load mappings: {e}") return {} return {} def save_mappings(self): """Save user mappings to file""" try: with open(self.mapping_file, 'w') as f: json.dump(self.user_map, f, indent=2, sort_keys=True) self.logger.info(f"Saved {len(self.user_map)} user mappings") # Also save as CSV for easy viewing/editing self.save_csv_directory() except Exception as e: self.logger.error(f"Failed to save mappings: {e}") def save_csv_directory(self): """Save user directory as CSV for easy viewing in Excel""" try: with open(self.csv_mapping_file, 'w', newline='') as f: fieldnames = ['user_id', 'full_name', 'display_name', 'email', 'department', 'title', 'team', 'manager', 'location', 'phone', 'slack_handle', 'sanitized_name', 'notes'] writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() for user_id, info in sorted(self.user_map.items(), key=lambda x: x[1].get('full_name', '')): row = { 'user_id': user_id, 'full_name': info.get('full_name', ''), 'display_name': info.get('display_name', ''), 'email': info.get('email', ''), 'department': info.get('department', ''), 'title': info.get('title', ''), 'team': info.get('team', ''), 'manager': info.get('manager', ''), 'location': info.get('location', ''), 'phone': info.get('phone', ''), 'slack_handle': info.get('slack_handle', ''), 'sanitized_name': info.get('sanitized_name', ''), 'notes': info.get('notes', '') } writer.writerow(row) self.logger.info(f"Saved user directory CSV: {self.csv_mapping_file}") except Exception as e: self.logger.error(f"Failed to save CSV directory: {e}") def fetch_user_from_slack(self, user_id): """Fetch user information from Slack API""" try: response = self.client.users_info(user=user_id) user = response['user'] profile = user.get('profile', {}) # Extract all useful information user_info = { 'user_id': user_id, 'full_name': profile.get('real_name', 'Unknown'), 'first_name': profile.get('first_name', ''), 'last_name': profile.get('last_name', ''), 'display_name': profile.get('display_name', ''), 'email': profile.get('email', ''), 'title': profile.get('title', ''), 'phone': profile.get('phone', ''), 'status_text': profile.get('status_text', ''), 'status_emoji': profile.get('status_emoji', ''), 'team': user.get('team_id', ''), 'slack_handle': user.get('name', ''), # The @handle 'is_bot': user.get('is_bot', False), 'is_admin': user.get('is_admin', False), 'is_owner': user.get('is_owner', False), 'timezone': user.get('tz', ''), 'locale': user.get('locale', ''), # Custom fields (if your Slack has them) 'department': profile.get('fields', {}).get('department', {}).get('value', ''), 'manager': profile.get('fields', {}).get('manager', {}).get('value', ''), 'location': profile.get('fields', {}).get('location', {}).get('value', ''), # Create sanitized filename version (no spaces or special chars) 'sanitized_name': self.sanitize_name(profile.get('real_name', user_id)) } return user_info except SlackApiError as e: self.logger.error(f"Failed to fetch user {user_id} from Slack: {e}") return None def sanitize_name(self, name): """Create a filename-safe version of the name""" # Replace spaces with underscores, remove special characters import re sanitized = re.sub(r'[^\w\s-]', '', name) sanitized = re.sub(r'[-\s]+', '_', sanitized) return sanitized.lower() def update_user(self, user_id, force_update=False): """Update or add a user to the mapping""" if user_id in self.user_map and not force_update: return self.user_map[user_id] user_info = self.fetch_user_from_slack(user_id) if user_info: self.user_map[user_id] = user_info self.save_mappings() return user_info return None def update_all_users(self, user_ids): """Update information for multiple users""" updated = 0 for user_id in user_ids: if self.update_user(user_id): updated += 1 self.logger.debug(f"Updated user {user_id}") self.logger.info(f"Updated {updated}/{len(user_ids)} users") return updated def get_user_info(self, user_id): """Get user information from cache or fetch if needed""" if user_id not in self.user_map: self.update_user(user_id) return self.user_map.get(user_id, {'full_name': 'Unknown', 'sanitized_name': user_id}) def get_user_name(self, user_id): """Get just the full name for a user""" info = self.get_user_info(user_id) return info.get('full_name', 'Unknown') def get_sanitized_name(self, user_id): """Get filename-safe name for a user""" info = self.get_user_info(user_id) return info.get('sanitized_name', user_id) def add_custom_info(self, user_id, custom_data): """Add custom information for a user (department, team, etc.)""" if user_id not in self.user_map: self.update_user(user_id) if user_id in self.user_map: self.user_map[user_id].update(custom_data) self.save_mappings() return True return False def bulk_import_from_csv(self, csv_file_path): """Import custom user data from a CSV file""" try: with open(csv_file_path, 'r') as f: reader = csv.DictReader(f) for row in reader: user_id = row.get('user_id') if user_id: # Remove user_id from the dict to avoid duplication custom_data = {k: v for k, v in row.items() if k != 'user_id' and v} self.add_custom_info(user_id, custom_data) self.logger.info(f"Imported custom data from {csv_file_path}") return True except Exception as e: self.logger.error(f"Failed to import CSV: {e}") return False def generate_readable_filename(self, user_id, date_str): """Generate a readable filename for user data files""" user_info = self.get_user_info(user_id) name = user_info.get('sanitized_name', user_id) return f"presence_{name}_{date_str}.csv" # Singleton instance _mapper_instance = None def get_user_mapper(): """Get or create singleton user mapper instance""" global _mapper_instance if _mapper_instance is None: _mapper_instance = UserMapper() return _mapper_instance # Utility script for managing users if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description='User mapping management') parser.add_argument('--update-all', action='store_true', help='Update all users from Slack') parser.add_argument('--show', action='store_true', help='Show current user mappings') parser.add_argument('--add-custom', type=str, nargs=3, metavar=('USER_ID', 'FIELD', 'VALUE'), help='Add custom field for a user') parser.add_argument('--import-csv', type=str, help='Import custom data from CSV') args = parser.parse_args() mapper = get_user_mapper() if args.update_all: from config_multi_user import SLACK_USER_IDS print(f"Updating {len(SLACK_USER_IDS)} users from Slack...") mapper.update_all_users(SLACK_USER_IDS) print("Update complete!") if args.show: print("\nCurrent User Mappings:") print("-" * 80) for user_id, info in sorted(mapper.user_map.items(), key=lambda x: x[1].get('full_name', '')): print(f"{user_id}: {info['full_name']:<30} ({info.get('email', 'no email')})") if info.get('department'): print(f" Department: {info['department']}") if info.get('title'): print(f" Title: {info['title']}") if args.add_custom: user_id, field, value = args.add_custom if mapper.add_custom_info(user_id, {field: value}): print(f"Added {field}={value} for user {user_id}") else: print(f"Failed to update user {user_id}") if args.import_csv: if mapper.bulk_import_from_csv(args.import_csv): print(f"Imported custom data from {args.import_csv}") else: print("Import failed")