a971d9366c
- Add interactive chat mode with persistent configuration storage - Implement real-time message receiving with polling mechanism - Add threading support for simultaneous send/receive operations - Create MessageReceiver class for handling incoming messages - Add commands: /listen, /stop, /history for two-way communication - Display sent messages in terminal with timestamps - Support for environment variables and .env file loading - Enhanced error handling and user-friendly prompts - Update README with comprehensive two-way communication documentation - Maintain backward compatibility with original CLI functionality
715 lines
26 KiB
Python
715 lines
26 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
|
|
|
|
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 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 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 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 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 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(" /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(" <message> - Send message to current channel")
|
|
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 Two-Way Interactive Chat Mode!")
|
|
print("Type your messages to send them to Slack.")
|
|
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()
|
|
|
|
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]}...")
|
|
|
|
print("\nReady to chat! Type your message or use /help for commands.")
|
|
print("Use /listen to start receiving messages from others.\n")
|
|
|
|
poster = None
|
|
message_receiver = None
|
|
|
|
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'}")
|
|
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
|
|
|
|
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}")
|
|
|
|
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()
|