#!/usr/bin/env python3 """ Slack Message Poster CLI Tool A command-line tool that posts messages to Slack channels using the Slack Web API. Supports both CLI mode and interactive chat mode with persistent configuration. """ import argparse import json import os import sys import threading import time from datetime import datetime from pathlib import Path from typing import Optional, Dict, Any, List import re try: from slack_sdk import WebClient from slack_sdk.errors import SlackApiError, SlackClientError except ImportError: print("Error: slack_sdk not installed. Run: pip install slack_sdk") sys.exit(1) try: import openai except ImportError: print("Warning: openai not installed. AI features will be disabled. Run: pip install openai") openai = None # Try to load environment variables from .env file try: from dotenv import load_dotenv load_dotenv() except ImportError: # dotenv not installed, continue without it pass class AIResponseHandler: """Handles AI-powered responses using Perplexity API.""" def __init__(self, api_key: str): """Initialize the AI response handler.""" if openai is None: raise ImportError("OpenAI library not installed") self.client = openai.OpenAI( api_key=api_key, base_url="https://api.perplexity.ai" ) def is_question(self, text: str) -> bool: """Check if the text is a question.""" # Simple question detection patterns question_patterns = [ r'\?$', # Ends with question mark r'^(what|where|when|why|how|who|which|can|could|would|should|is|are|was|were|do|does|did)\b', # Starts with question words r'\b(what|where|when|why|how|who|which|can|could|would|should|is|are|was|were|do|does|did)\b', # Contains question words ] text_lower = text.lower().strip() for pattern in question_patterns: if re.search(pattern, text_lower): return True return False def get_ai_response(self, question: str) -> str: """Get AI response for a question.""" try: response = self.client.chat.completions.create( model="sonar", messages=[ { "role": "system", "content": "You are a helpful assistant. Provide concise, accurate answers to questions. Keep responses brief and informative." }, { "role": "user", "content": question } ], max_tokens=200, temperature=0.7 ) return response.choices[0].message.content.strip() except Exception as e: return f"[AI Error] Failed to get response: {str(e)}" class ConfigManager: """Manages persistent configuration storage.""" def __init__(self, config_file: str = "slack_config.json"): """Initialize the configuration manager.""" self.config_file = Path(config_file) self.config = self._load_config() def _load_config(self) -> Dict[str, Any]: """Load configuration from file.""" if self.config_file.exists(): try: with open(self.config_file, 'r') as f: return json.load(f) except (json.JSONDecodeError, IOError): return {} return {} def save_config(self) -> bool: """Save configuration to file.""" try: with open(self.config_file, 'w') as f: json.dump(self.config, f, indent=2) return True except IOError: return False def get_token(self) -> Optional[str]: """Get stored Slack token.""" return self.config.get('token') def get_channel(self) -> Optional[str]: """Get stored channel ID.""" return self.config.get('channel') def get_perplexity_key(self) -> Optional[str]: """Get stored Perplexity API key.""" return self.config.get('perplexity_key') def set_token(self, token: str) -> None: """Set and save Slack token.""" self.config['token'] = token self.save_config() def set_channel(self, channel: str) -> None: """Set and save channel ID.""" self.config['channel'] = channel self.save_config() def set_perplexity_key(self, api_key: str) -> None: """Set and save Perplexity API key.""" self.config['perplexity_key'] = api_key self.save_config() def clear_config(self) -> None: """Clear all stored configuration.""" self.config = {} if self.config_file.exists(): self.config_file.unlink() class SlackPoster: """Main class for posting messages to Slack.""" def __init__(self, token: str): """Initialize the Slack client with the provided token.""" self.client = WebClient(token=token) def post_message(self, channel: str, message: str) -> dict: """ Post a message to the specified Slack channel. Args: channel: Slack channel ID (e.g., C04ABC123) message: Message text to post Returns: dict: Response from Slack API Raises: SlackApiError: If the API call fails SlackClientError: If there's a client error """ try: response = self.client.chat_postMessage( channel=channel, text=message ) return response except SlackApiError as e: raise e except SlackClientError as e: raise e def get_channel_history(self, channel: str, limit: int = 10) -> List[dict]: """ Get recent messages from a Slack channel. Args: channel: Slack channel ID (e.g., C04ABC123) limit: Number of messages to retrieve (max 1000) Returns: List[dict]: List of message objects Raises: SlackApiError: If the API call fails SlackClientError: If there's a client error """ try: response = self.client.conversations_history( channel=channel, limit=limit ) return response.get('messages', []) except SlackApiError as e: raise e except SlackClientError as e: raise e def get_user_info(self, user_id: str) -> dict: """ Get user information by user ID. Args: user_id: Slack user ID (e.g., U123ABC456) Returns: dict: User information Raises: SlackApiError: If the API call fails SlackClientError: If there's a client error """ try: response = self.client.users_info(user=user_id) return response.get('user', {}) except SlackApiError as e: raise e except SlackClientError as e: raise e class MessageReceiver: """Handles receiving and displaying messages from Slack channels.""" def __init__(self, slack_poster: SlackPoster, channel: str): """Initialize the message receiver.""" self.slack_poster = slack_poster self.channel = channel self.last_message_timestamp = None self.running = False self.thread = None def start_listening(self): """Start listening for new messages in a separate thread.""" if self.running: return self.running = True self.thread = threading.Thread(target=self._listen_loop, daemon=True) self.thread.start() def stop_listening(self): """Stop listening for messages.""" self.running = False if self.thread: self.thread.join(timeout=1) def _listen_loop(self): """Main listening loop that polls for new messages.""" while self.running: try: messages = self.slack_poster.get_channel_history(self.channel, limit=5) # Filter out messages we've already seen new_messages = [] for msg in reversed(messages): # Process oldest first msg_timestamp = msg.get('ts') if self.last_message_timestamp is None or float(msg_timestamp) > float(self.last_message_timestamp): new_messages.append(msg) # Display new messages for msg in new_messages: self._display_message(msg) if self.last_message_timestamp is None or float(msg.get('ts')) > float(self.last_message_timestamp): self.last_message_timestamp = msg.get('ts') time.sleep(2) # Poll every 2 seconds except Exception as e: print(f"[ERROR] Failed to receive messages: {e}") time.sleep(5) # Wait longer on error def _display_message(self, msg: dict): """Display a received message.""" try: user_id = msg.get('user', '') text = msg.get('text', '') timestamp = msg.get('ts', '') # Get user info user_name = 'Unknown' if user_id: try: user_info = self.slack_poster.get_user_info(user_id) user_name = user_info.get('real_name', user_info.get('name', user_id)) except: user_name = user_id # Format timestamp try: dt = datetime.fromtimestamp(float(timestamp)) formatted_time = dt.strftime('%H:%M:%S') except (ValueError, TypeError): formatted_time = timestamp # Display the message print(f"\n[{formatted_time}] {user_name}: {text}") print("> ", end="", flush=True) # Restore prompt except Exception as e: print(f"[ERROR] Failed to display message: {e}") def get_recent_messages(self, limit: int = 10): """Get and display recent messages from the channel.""" try: messages = self.slack_poster.get_channel_history(self.channel, limit=limit) print(f"\n[RECENT MESSAGES] Last {len(messages)} messages in #{self.channel}:") print("-" * 60) for msg in reversed(messages): # Show newest first self._display_message(msg) print("-" * 60) print("> ", end="", flush=True) except Exception as e: print(f"[ERROR] Failed to get recent messages: {e}") print("> ", end="", flush=True) def get_user_input(prompt: str, required: bool = True) -> str: """ Get user input with optional validation. Args: prompt: The prompt to display to the user required: Whether the input is required Returns: str: User input """ while True: value = input(prompt).strip() if value or not required: return value print("This field is required. Please try again.") def validate_token(token: str) -> bool: """ Basic validation for Slack token format. Args: token: The token to validate Returns: bool: True if token format looks valid """ if not token: return False # Basic format validation for Slack tokens valid_prefixes = ['xoxb-', 'xoxp-', 'xoxa-', 'xoxr-'] return any(token.startswith(prefix) for prefix in valid_prefixes) def validate_channel(channel: str) -> bool: """ Basic validation for Slack channel ID format. Args: channel: The channel ID to validate Returns: bool: True if channel format looks valid """ if not channel: return False # Slack channel IDs typically start with C, D, or G return channel.startswith(('C', 'D', 'G')) and len(channel) > 1 def print_help(): """Print help information for interactive mode.""" print("\n" + "="*70) print("SLACK AI-POWERED TWO-WAY CHAT MODE - Available Commands:") print("="*70) print(" /help, /h - Show this help message") print(" /quit, /q, /exit - Exit the chat") print(" /token - Set/update Slack token") print(" /channel - Set/update channel ID") print(" /perplexity - Set/update Perplexity API key") print(" /status - Show current configuration") print(" /clear - Clear saved configuration") print(" /test - Test current configuration") print(" /listen - Start listening for incoming messages") print(" /stop - Stop listening for messages") print(" /history [n] - Show recent message history (default: 10)") print(" /ai on/off - Enable/disable AI responses") print(" - Send message to current channel") print("="*70) print("AI FEATURES:") print(" - Ask questions and get AI-powered answers automatically") print(" - AI detects questions and responds with intelligent answers") print(" - Use /perplexity to set your Perplexity API key") print(" - Use /ai on to enable automatic AI responses") print("="*70) print("TWO-WAY FEATURES:") print(" - Messages sent by others will appear automatically when listening") print(" - Use /listen to start receiving messages in real-time") print(" - Use /history to see recent conversation") print("="*70) def interactive_chat_mode(config_manager: ConfigManager): """Run interactive chat mode.""" print("\n[CHAT] Welcome to Slack AI-Powered Two-Way Interactive Chat Mode!") print("Type your messages to send them to Slack.") print("Ask questions and get AI-powered answers automatically!") print("Use /help for available commands.\n") # Check if we have stored configuration (try environment variables first, then config file) token = os.getenv('SLACK_BOT_TOKEN') or config_manager.get_token() channel = os.getenv('SLACK_CHANNEL_ID') or config_manager.get_channel() perplexity_key = os.getenv('PERPLEXITY_API_KEY') or config_manager.get_perplexity_key() if not token: print("No Slack token found. Please set one using /token ") elif not channel: print("No channel ID found. Please set one using /channel ") else: print(f"[OK] Using configuration:") print(f" Channel: {channel}") print(f" Token: {token[:10]}...") if perplexity_key: print(f" Perplexity API: {perplexity_key[:10]}...") else: print(" Perplexity API: Not set (use /perplexity )") print("\nReady to chat! Type your message or use /help for commands.") print("Use /listen to start receiving messages from others.") print("Ask questions like 'What is the capital of India?' for AI answers!\n") poster = None message_receiver = None ai_handler = None ai_enabled = False while True: try: user_input = input("> ").strip() if not user_input: continue # Handle commands if user_input.startswith('/'): command = user_input.lower().split()[0] if command in ['/help', '/h']: print_help() continue elif command in ['/quit', '/q', '/exit']: print("\n[BYE] Goodbye!") break elif command == '/token': parts = user_input.split(' ', 1) if len(parts) < 2: print("[ERROR] Usage: /token ") continue new_token = parts[1].strip() if validate_token(new_token): config_manager.set_token(new_token) token = new_token poster = None # Reset poster to use new token print("[OK] Token updated successfully!") else: print("[ERROR] Invalid token format. Tokens should start with 'xoxb-', 'xoxp-', 'xoxa-', or 'xoxr-'") continue elif command == '/channel': parts = user_input.split(' ', 1) if len(parts) < 2: print("[ERROR] Usage: /channel ") continue new_channel = parts[1].strip() if validate_channel(new_channel): config_manager.set_channel(new_channel) channel = new_channel print(f"[OK] Channel updated to: {channel}") else: print("[ERROR] Invalid channel format. Channel IDs should start with 'C', 'D', or 'G'") continue elif command == '/status': print(f"\n[STATUS] Current Configuration:") print(f" Token: {token[:10] + '...' if token else 'Not set'}") print(f" Channel: {channel if channel else 'Not set'}") print(f" Perplexity API: {perplexity_key[:10] + '...' if perplexity_key else 'Not set'}") print(f" AI Responses: {'Enabled' if ai_enabled else 'Disabled'}") print(f" AI Handler: {'Ready' if ai_handler else 'Not initialized'}") continue elif command == '/clear': confirm = input("[WARNING] Are you sure you want to clear all saved configuration? (y/N): ").strip().lower() if confirm in ['y', 'yes']: config_manager.clear_config() token = None channel = None poster = None print("[OK] Configuration cleared!") else: print("[CANCELLED] Configuration not cleared.") continue elif command == '/test': if not token or not channel: print("[ERROR] Please set both token and channel before testing.") continue try: if not poster: poster = SlackPoster(token) test_message = "[TEST] Test message from Slack Chat CLI" response = poster.post_message(channel, test_message) timestamp = response.get('ts', 'unknown') # Convert timestamp to readable format try: dt = datetime.fromtimestamp(float(timestamp)) formatted_time = dt.strftime('%H:%M:%S') except (ValueError, TypeError): formatted_time = timestamp # Display the test message print(f"[{formatted_time}] You: {test_message}") print("[OK] Test message sent successfully!") except Exception as e: print(f"[ERROR] Test failed: {e}") continue elif command == '/listen': if not token or not channel: print("[ERROR] Please set both token and channel before listening.") continue try: if not poster: poster = SlackPoster(token) if message_receiver and message_receiver.running: print("[INFO] Already listening for messages.") else: message_receiver = MessageReceiver(poster, channel) message_receiver.start_listening() print("[OK] Started listening for incoming messages!") print(" Messages from others will appear automatically.") except Exception as e: print(f"[ERROR] Failed to start listening: {e}") continue elif command == '/stop': if message_receiver and message_receiver.running: message_receiver.stop_listening() print("[OK] Stopped listening for messages.") else: print("[INFO] Not currently listening for messages.") continue elif command == '/history': if not token or not channel: print("[ERROR] Please set both token and channel before viewing history.") continue try: if not poster: poster = SlackPoster(token) # Parse limit if provided parts = user_input.split() limit = 10 if len(parts) > 1: try: limit = int(parts[1]) limit = min(limit, 50) # Cap at 50 for performance except ValueError: pass if not message_receiver: message_receiver = MessageReceiver(poster, channel) message_receiver.get_recent_messages(limit) except Exception as e: print(f"[ERROR] Failed to get message history: {e}") continue elif command == '/perplexity': parts = user_input.split(' ', 1) if len(parts) < 2: print("[ERROR] Usage: /perplexity ") continue new_key = parts[1].strip() config_manager.set_perplexity_key(new_key) perplexity_key = new_key # Initialize AI handler try: ai_handler = AIResponseHandler(perplexity_key) print("[OK] Perplexity API key updated and AI handler initialized!") except ImportError: print("[ERROR] OpenAI library not installed. Run: pip install openai") except Exception as e: print(f"[ERROR] Failed to initialize AI handler: {e}") continue elif command == '/ai': parts = user_input.split() if len(parts) < 2: print(f"[INFO] AI responses are currently {'enabled' if ai_enabled else 'disabled'}") print("[USAGE] /ai on - Enable AI responses") print("[USAGE] /ai off - Disable AI responses") continue action = parts[1].lower() if action == 'on': if not perplexity_key: print("[ERROR] Please set Perplexity API key first using /perplexity ") continue if not ai_handler: try: ai_handler = AIResponseHandler(perplexity_key) except Exception as e: print(f"[ERROR] Failed to initialize AI handler: {e}") continue ai_enabled = True print("[OK] AI responses enabled! Ask questions to get AI answers.") elif action == 'off': ai_enabled = False print("[OK] AI responses disabled.") else: print("[ERROR] Usage: /ai on or /ai off") continue else: print(f"[ERROR] Unknown command: {command}. Use /help for available commands.") continue # Send message if not token or not channel: print("[ERROR] Please set both token and channel before sending messages.") print(" Use /token and /channel ") continue try: if not poster: poster = SlackPoster(token) response = poster.post_message(channel, user_input) timestamp = response.get('ts', 'unknown') # Convert timestamp to readable format try: dt = datetime.fromtimestamp(float(timestamp)) formatted_time = dt.strftime('%H:%M:%S') except (ValueError, TypeError): formatted_time = timestamp # Display the sent message in the terminal print(f"[{formatted_time}] You: {user_input}") print(f"[OK] Message sent to #{channel}") # Check if AI should respond to this message if ai_enabled and ai_handler and ai_handler.is_question(user_input): try: print("[AI] Thinking...") ai_response = ai_handler.get_ai_response(user_input) # Display AI response in terminal first ai_timestamp = datetime.now().strftime('%H:%M:%S') print(f"[{ai_timestamp}] AI Bot: {ai_response}") # Then send AI response to Slack poster.post_message(channel, f"🤖 AI: {ai_response}") print(f"[OK] AI response sent to #{channel}") except Exception as e: print(f"[AI ERROR] Failed to generate AI response: {e}") except SlackApiError as e: error_msg = str(e) if "invalid_auth" in error_msg.lower(): print("[ERROR] Invalid token. Please update your token using /token ") elif "channel_not_found" in error_msg.lower(): print(f"[ERROR] Channel '{channel}' not found. Please check the channel ID.") elif "not_in_channel" in error_msg.lower(): print(f"[ERROR] Bot is not a member of channel '{channel}'. Please add the bot to the channel.") elif "rate_limited" in error_msg.lower(): print("[ERROR] Rate limited. Please wait before trying again.") else: print(f"[ERROR] Slack API Error: {error_msg}") except SlackClientError as e: print(f"[ERROR] Slack Client Error: {e}") except Exception as e: print(f"[ERROR] Unexpected error: {e}") except KeyboardInterrupt: print("\n\n[BYE] Goodbye!") if message_receiver: message_receiver.stop_listening() break except EOFError: print("\n\n[BYE] Goodbye!") if message_receiver: message_receiver.stop_listening() break def main(): """Main function to handle CLI arguments and execute the message posting.""" parser = argparse.ArgumentParser( description="Post a message to a Slack channel or start interactive chat mode", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # CLI Mode - Send a single message python slack_poster.py -t xoxb-1234567890-1234567890-abcdefghijklmnopqrstuvwx -c C04ABC123 -m "Hello World!" python slack_poster.py --token xoxb-1234567890-1234567890-abcdefghijklmnopqrstuvwx --channel C04ABC123 --message "Hello World!" # Interactive Chat Mode python slack_poster.py --chat python slack_poster.py -i # CLI Mode with prompts for missing parameters python slack_poster.py """ ) parser.add_argument( '--token', '-t', type=str, help='Slack Bot/User OAuth token (e.g., xoxb-...)' ) parser.add_argument( '--channel', '-c', type=str, help='Slack channel ID (e.g., C04ABC123)' ) parser.add_argument( '--message', '-m', type=str, help='Message text to post' ) parser.add_argument( '--chat', '-i', '--interactive', action='store_true', help='Start interactive chat mode' ) args = parser.parse_args() # Initialize configuration manager config_manager = ConfigManager() # If interactive chat mode is requested if args.chat: interactive_chat_mode(config_manager) return # CLI Mode - Send a single message # Get token (try environment variable first, then command line, then config, then interactive) token = args.token or os.getenv('SLACK_BOT_TOKEN') or config_manager.get_token() if not token: print("Slack Token not provided. Please enter it interactively:") token = get_user_input("Enter your Slack token: ") # Save token for future use config_manager.set_token(token) # Validate token format if not validate_token(token): print("Error: Invalid token format. Slack tokens typically start with 'xoxb-', 'xoxp-', 'xoxa-', or 'xoxr-'") sys.exit(1) # Get channel (try environment variable first, then command line, then config, then interactive) channel = args.channel or os.getenv('SLACK_CHANNEL_ID') or config_manager.get_channel() if not channel: print("Channel ID not provided. Please enter it interactively:") channel = get_user_input("Enter Slack channel ID (e.g., C04ABC123): ") # Save channel for future use config_manager.set_channel(channel) # Validate channel format if not validate_channel(channel): print("Error: Invalid channel format. Channel IDs typically start with 'C', 'D', or 'G'") sys.exit(1) # Get message message = args.message if not message: print("Message not provided. Please enter it interactively:") message = get_user_input("Enter message text: ") # Validate message if not message.strip(): print("Error: Message cannot be empty") sys.exit(1) # Initialize Slack poster and send message try: poster = SlackPoster(token) response = poster.post_message(channel, message) # Extract timestamp from response timestamp = response.get('ts', 'unknown') channel_name = response.get('channel', channel) # Convert timestamp to readable format try: dt = datetime.fromtimestamp(float(timestamp)) formatted_time = dt.strftime('%Y-%m-%d %H:%M:%S UTC') except (ValueError, TypeError): formatted_time = timestamp print(f"[SUCCESS] Message posted to channel {channel_name}") print(f"[TIMESTAMP] {formatted_time}") print(f"[MESSAGE] {message}") except SlackApiError as e: error_msg = str(e) if "invalid_auth" in error_msg.lower(): print("[ERROR] Invalid token. Please check your Slack token.") elif "channel_not_found" in error_msg.lower(): print(f"[ERROR] Channel '{channel}' not found. Please check the channel ID.") elif "not_in_channel" in error_msg.lower(): print(f"[ERROR] Bot is not a member of channel '{channel}'. Please add the bot to the channel.") elif "rate_limited" in error_msg.lower(): print("[ERROR] Rate limited. Please wait before trying again.") else: print(f"[ERROR] Slack API Error: {error_msg}") sys.exit(1) except SlackClientError as e: print(f"[ERROR] Slack Client Error: {e}") sys.exit(1) except Exception as e: print(f"[ERROR] Unexpected error: {e}") sys.exit(1) if __name__ == "__main__": main()