bytelyst-devops-tools/Slack Message/slack_poster.py
reethu2703 fd58016586 Add AI-powered question answering with Perplexity API integration
- Integrate Perplexity AI API for intelligent question answering
- Add AIResponseHandler class with question detection and response generation
- Implement automatic question detection using regex patterns
- Add AI commands: /perplexity, /ai on/off for configuration
- Display AI responses in terminal before sending to Slack
- Update configuration management to store Perplexity API key
- Add comprehensive AI features documentation to README
- Update requirements.txt with OpenAI library dependency
- Fix Perplexity model name to use valid 'sonar' model
- Enhance interactive chat mode with AI-powered capabilities
- Maintain backward compatibility with existing CLI functionality
2025-10-05 15:21:06 +05:30

870 lines
33 KiB
Python

#!/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 <token> - Set/update Slack token")
print(" /channel <id> - Set/update channel ID")
print(" /perplexity <key> - 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(" <message> - 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 <key> 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 <your-token>")
elif not channel:
print("No channel ID found. Please set one using /channel <channel-id>")
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 <key>)")
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 <your-slack-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 <channel-id>")
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 <your-perplexity-api-key>")
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 <key>")
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 <your-token> and /channel <channel-id>")
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 <new-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()