356 lines
16 KiB
Python
356 lines
16 KiB
Python
import os
|
|
import time
|
|
import random
|
|
import argparse
|
|
import json
|
|
from typing import List, Optional, Dict, Any, Tuple, Union, Literal
|
|
from src.core.twitch import TwitchBot, CommandCallback
|
|
from dotenv import load_dotenv
|
|
|
|
load_dotenv()
|
|
|
|
# Default config file path
|
|
DEFAULT_CONFIG_PATH = "default_config.json"
|
|
|
|
# Example command handlers
|
|
def cmd_hello(username: str, args: List[str], bot: TwitchBot) -> None:
|
|
"""Simple hello command that greets the user"""
|
|
bot.send_message(f"Hello, {username}!")
|
|
|
|
def cmd_dice(username: str, args: List[str], bot: TwitchBot) -> None:
|
|
"""Roll a dice with specified number of sides"""
|
|
sides: int = 6 # Default to 6-sided dice
|
|
|
|
if args and args[0].isdigit():
|
|
sides = int(args[0])
|
|
|
|
result: int = random.randint(1, sides)
|
|
bot.send_message(f"@{username} rolled a {result} (d{sides})")
|
|
|
|
def cmd_echo(username: str, args: List[str], bot: TwitchBot) -> None:
|
|
"""Echo back the user's message"""
|
|
if args:
|
|
message: str = " ".join(args)
|
|
bot.send_message(f"Echo: {message}")
|
|
else:
|
|
bot.send_message(f"@{username}, you didn't provide a message to echo!")
|
|
|
|
def cmd_magic8ball(username: str, args: List[str], bot: TwitchBot) -> None:
|
|
"""Magic 8-ball that gives random answers"""
|
|
responses: List[str] = [
|
|
"It is certain.",
|
|
"It is decidedly so.",
|
|
"Without a doubt.",
|
|
"Yes, definitely.",
|
|
"You may rely on it.",
|
|
"As I see it, yes.",
|
|
"Most likely.",
|
|
"Outlook good.",
|
|
"Yes.",
|
|
"Signs point to yes.",
|
|
"Reply hazy, try again.",
|
|
"Ask again later.",
|
|
"Better not tell you now.",
|
|
"Cannot predict now.",
|
|
"Concentrate and ask again.",
|
|
"Don't count on it.",
|
|
"My reply is no.",
|
|
"My sources say no.",
|
|
"Outlook not so good.",
|
|
"Very doubtful."
|
|
]
|
|
bot.send_message(f"@{username}, {random.choice(responses)}")
|
|
|
|
def cmd_queue_stats(username: str, args: List[str], bot: TwitchBot) -> None:
|
|
"""Show queue statistics"""
|
|
stats = bot.get_queue_stats()
|
|
bot.send_message(f"Queue stats: {stats['total_processed']}/{stats['total_enqueued']} processed, {stats['pending']} pending")
|
|
|
|
def cmd_token_status(username: str, args: List[str], bot: TwitchBot) -> None:
|
|
"""Show status of the Twitch API token (streamer only)"""
|
|
# Check if user is an admin
|
|
if not bot.is_admin(username):
|
|
bot.send_message(f"@{username}, only the channel owner or admins can check token status.")
|
|
return
|
|
|
|
if bot.access_token and bot.token_expiry:
|
|
# Calculate time remaining
|
|
time_remaining = bot.token_expiry - time.time()
|
|
if time_remaining > 0:
|
|
hours = int(time_remaining // 3600)
|
|
minutes = int((time_remaining % 3600) // 60)
|
|
bot.send_message(f"@{username}, token is valid for {hours}h {minutes}m.")
|
|
else:
|
|
bot.send_message(f"@{username}, token is expired and will be refreshed on next use.")
|
|
else:
|
|
bot.send_message(f"@{username}, no token available. It will be generated when needed.")
|
|
|
|
def cmd_stop(username: str, args: List[str], bot: TwitchBot) -> None:
|
|
"""Stop the bot (streamer only)"""
|
|
# Check if user is an admin
|
|
if not bot.is_admin(username):
|
|
bot.send_message(f"@{username}, only the channel owner or admins can stop the bot.")
|
|
return
|
|
|
|
bot.send_message(f"Bot is shutting down by request of @{username}.")
|
|
# Stop the bot
|
|
bot.stop()
|
|
|
|
def main() -> None:
|
|
# Parse command line arguments
|
|
parser: argparse.ArgumentParser = argparse.ArgumentParser(description='Twitch Chat Bot')
|
|
parser.add_argument('--advanced', action='store_true', help='Enable advanced features')
|
|
parser.add_argument('--game-control', action='store_true', help='Enable game control features')
|
|
parser.add_argument('--game-mode', choices=['direct', 'vote'], default='direct',
|
|
help='Game control mode: direct (immediate commands) or vote (timed voting)')
|
|
parser.add_argument('--cooldown', type=float, default=2.0,
|
|
help='Cooldown time between commands in seconds (for direct mode)')
|
|
parser.add_argument('--input-type', choices=['keyboard', 'controller'], default='keyboard',
|
|
help='Input type: keyboard/mouse or Xbox controller emulation')
|
|
parser.add_argument('--use-queue', action='store_true', help='Use Huey task queue for processing commands')
|
|
parser.add_argument('--start-consumer', action='store_true', help='Start Huey consumer within the bot process')
|
|
parser.add_argument('--token-cache', type=str, default="data/cache/token_cache.json",
|
|
help='Path to token cache file (default: data/cache/token_cache.json)')
|
|
parser.add_argument('--admin-users', type=str, help='Comma-separated list of additional admin users who have the same permissions as the channel owner')
|
|
parser.add_argument('--config', type=str, default=DEFAULT_CONFIG_PATH,
|
|
help=f'Path to configuration file (default: {DEFAULT_CONFIG_PATH})')
|
|
parser.add_argument('--force-new-token', action='store_true', help='Force generation of a new token even if a cached one exists')
|
|
args: argparse.Namespace = parser.parse_args()
|
|
|
|
# Load configuration from file if it exists
|
|
config: Dict[str, Any] = {}
|
|
if os.path.exists(args.config):
|
|
try:
|
|
with open(args.config, 'r') as f:
|
|
config = json.load(f)
|
|
print(f"Loaded configuration from {args.config}")
|
|
except json.JSONDecodeError:
|
|
print(f"Error: {args.config} is not a valid JSON file")
|
|
except Exception as e:
|
|
print(f"Error loading config file: {e}")
|
|
|
|
# Get Twitch credentials from environment variables
|
|
username: Optional[str] = os.environ.get("TWITCH_USERNAME")
|
|
client_id: Optional[str] = os.environ.get("TWITCH_CLIENT_ID")
|
|
client_secret: Optional[str] = os.environ.get("TWITCH_CLIENT_SECRET")
|
|
channel: Optional[str] = os.environ.get("TWITCH_CHANNEL")
|
|
|
|
# Check for required credentials
|
|
if not all([username, client_id, client_secret, channel]):
|
|
print("Please set the following environment variables:")
|
|
print(" TWITCH_USERNAME - The bot's username")
|
|
print(" TWITCH_CLIENT_ID - Your app's registered client ID")
|
|
print(" TWITCH_CLIENT_SECRET - Your app's registered client secret")
|
|
print(" TWITCH_CHANNEL - The channel to join")
|
|
print("\nTo get client ID and secret:")
|
|
print("1. Go to https://dev.twitch.tv/console/apps")
|
|
print("2. Create a new application or use an existing one")
|
|
print("3. Get the client ID and generate a client secret")
|
|
|
|
# For testing, you can uncomment and set these values
|
|
# username = "your_bot_username"
|
|
# client_id = "your_client_id"
|
|
# client_secret = "your_client_secret"
|
|
# channel = "your_channel"
|
|
|
|
if not all([username, client_id, client_secret, channel]):
|
|
return
|
|
|
|
# Type assertion to satisfy type checker (we checked these are not None above)
|
|
username = username if username is not None else ""
|
|
client_id = client_id if client_id is not None else ""
|
|
client_secret = client_secret if client_secret is not None else ""
|
|
channel = channel if channel is not None else ""
|
|
|
|
# Parse admin users with priority: command line > env var > config file
|
|
admin_users: Optional[List[str]] = None
|
|
|
|
# 1. From command line
|
|
if args.admin_users:
|
|
admin_users = [user.strip() for user in args.admin_users.split(",")]
|
|
print(f"Additional admin users from command line: {', '.join(admin_users)}")
|
|
|
|
# 2. From environment variable
|
|
elif os.environ.get("TWITCH_ADMIN_USERS"):
|
|
env_admins = os.environ.get("TWITCH_ADMIN_USERS", "")
|
|
admin_users = [user.strip() for user in env_admins.split(",")]
|
|
print(f"Additional admin users from environment: {', '.join(admin_users)}")
|
|
|
|
# 3. From config file
|
|
elif "admin_users" in config and isinstance(config["admin_users"], list):
|
|
admin_users = config["admin_users"]
|
|
print(f"Additional admin users from config file: {', '.join(admin_users)}")
|
|
|
|
# Make sure the cache directory exists
|
|
cache_dir = os.path.dirname(args.token_cache)
|
|
os.makedirs(cache_dir, exist_ok=True)
|
|
|
|
print(f"Using token cache file: {args.token_cache}")
|
|
if args.force_new_token:
|
|
print("Forcing new token generation (any cached token will be ignored)")
|
|
# Remove cached token if forcing new one
|
|
if os.path.exists(args.token_cache):
|
|
try:
|
|
os.remove(args.token_cache)
|
|
print("Removed existing token cache file")
|
|
except Exception as e:
|
|
print(f"Warning: Could not remove existing token cache: {e}")
|
|
else:
|
|
print("Using cached token if available")
|
|
|
|
# Create bot instance - explicitly pass None for access_token to avoid using env vars
|
|
bot: TwitchBot = TwitchBot(
|
|
username=username,
|
|
client_id=client_id,
|
|
client_secret=client_secret,
|
|
channel=channel,
|
|
use_queue=args.use_queue,
|
|
token_cache_file=args.token_cache,
|
|
admin_users=admin_users,
|
|
force_new_token=args.force_new_token,
|
|
access_token=None, # Explicitly set to None to avoid using env vars
|
|
refresh_token=None # Explicitly set to None to avoid using env vars
|
|
)
|
|
|
|
# Register basic commands
|
|
bot.register_command("hello", cmd_hello)
|
|
bot.register_command("dice", cmd_dice)
|
|
bot.register_command("echo", cmd_echo)
|
|
bot.register_command("8ball", cmd_magic8ball)
|
|
bot.register_command("token", cmd_token_status)
|
|
bot.register_command("stop", cmd_stop)
|
|
|
|
# Add queue-related commands if using the queue
|
|
if args.use_queue:
|
|
bot.register_command("qstats", cmd_queue_stats)
|
|
|
|
# Start the Huey consumer if requested
|
|
if args.start_consumer:
|
|
try:
|
|
from src.queue.server import start_consumer, set_input_mode, get_input_mode
|
|
import threading
|
|
|
|
# Make sure input mode is set before starting consumer
|
|
if args.game_control and args.input_type:
|
|
print(f"Consumer will use input mode: {get_input_mode()}")
|
|
|
|
consumer_thread = threading.Thread(target=start_consumer)
|
|
consumer_thread.daemon = True
|
|
consumer_thread.start()
|
|
print("Started Huey consumer thread")
|
|
print("Note: For production, it's better to use: python -m huey.bin.huey_consumer src.queue.server.huey")
|
|
except ImportError as e:
|
|
print(f"Failed to start consumer: {e}")
|
|
|
|
# Enable advanced features if requested
|
|
if args.advanced:
|
|
try:
|
|
from src.features.examples import AdvancedExamples
|
|
print("Enabling advanced features...")
|
|
advanced: AdvancedExamples = AdvancedExamples(bot)
|
|
print("Advanced features enabled: polls, voting, timers, points system")
|
|
if os.name == 'nt':
|
|
print("Sound effects enabled (Windows only)")
|
|
# Create sounds directory if it doesn't exist
|
|
if not os.path.exists("sounds"):
|
|
os.makedirs("sounds")
|
|
print("Created 'sounds' directory. Add .wav files to use with !sound command")
|
|
except ImportError as e:
|
|
print(f"Failed to import advanced features: {e}")
|
|
|
|
# Enable game control if requested
|
|
if args.game_control:
|
|
try:
|
|
from src.game.controller import GameController, PYNPUT_AVAILABLE
|
|
|
|
# Check if controller input is requested
|
|
if args.input_type == 'controller':
|
|
try:
|
|
from src.game.input.gamepad import VirtualController, VGAMEPAD_AVAILABLE
|
|
if not VGAMEPAD_AVAILABLE:
|
|
print("WARNING: vgamepad library not found. Install it with: pip install vgamepad")
|
|
print("Falling back to keyboard/mouse input.")
|
|
args.input_type = 'keyboard'
|
|
except ImportError:
|
|
print("WARNING: controller_support.py not found or error importing it.")
|
|
print("Falling back to keyboard/mouse input.")
|
|
args.input_type = 'keyboard'
|
|
|
|
# Persist the input type to make it available to the queue consumer process
|
|
try:
|
|
from src.queue.server import set_input_mode, INPUT_MODE_KEYBOARD, INPUT_MODE_CONTROLLER
|
|
if args.input_type == 'keyboard':
|
|
set_input_mode(INPUT_MODE_KEYBOARD)
|
|
print(f"Set persistent input mode to keyboard (this will be used by queue consumer)")
|
|
else:
|
|
set_input_mode(INPUT_MODE_CONTROLLER)
|
|
print(f"Set persistent input mode to controller (this will be used by queue consumer)")
|
|
except ImportError as e:
|
|
print(f"Warning: Could not set persistent input mode: {e}")
|
|
|
|
print(f"Enabling game control in {args.game_mode} mode with {args.input_type} input...")
|
|
|
|
if args.input_type == 'keyboard' and not PYNPUT_AVAILABLE:
|
|
print("WARNING: pynput library not found. Install it with: pip install pynput")
|
|
print("Keyboard/mouse control will be simulated but not actually perform any actions.")
|
|
print("Run 'pip install pynput' to enable actual keyboard/mouse control.")
|
|
|
|
# Only initialize the GameController when NOT using queue mode
|
|
# When in queue mode, the controller is initialized in the queue consumer process
|
|
if not args.use_queue:
|
|
game_controller: GameController = GameController(
|
|
bot,
|
|
mode=args.game_mode,
|
|
cooldown=args.cooldown,
|
|
input_type=args.input_type
|
|
)
|
|
|
|
if args.game_mode == "direct":
|
|
print(f"Commands will execute immediately with a {args.cooldown}s cooldown per user")
|
|
else:
|
|
print("Commands will be collected through voting")
|
|
print("Streamer can start a vote with !startvote and end with !endvote")
|
|
print("Users vote with !vote [command]")
|
|
|
|
print("Use !gamehelp to see available game commands")
|
|
print("Check user stats with !gamestats [username]")
|
|
else:
|
|
print("Game control will be handled by the queue consumer process")
|
|
print("Use !gamehelp to see available game commands")
|
|
|
|
except ImportError as e:
|
|
print(f"Failed to import game control features: {e}")
|
|
print("Make sure you have the required dependencies:")
|
|
print(" pip install pynput")
|
|
if args.input_type == 'controller':
|
|
print(" pip install vgamepad # For controller support")
|
|
|
|
# Start the bot
|
|
print(f"Starting bot for channel #{channel}")
|
|
print("Basic commands: !hello, !dice [sides], !echo [message], !8ball")
|
|
print("Admin commands: !token (check token status)")
|
|
|
|
if args.use_queue:
|
|
print("Queue mode enabled, commands will be processed by the queue system")
|
|
print("Use !qstats to see queue statistics")
|
|
|
|
if args.force_new_token:
|
|
print("Note: Forcing generation of a new authentication token")
|
|
|
|
try:
|
|
# Make sure requests library is installed
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
print("ERROR: The requests library is required for the new authentication method.")
|
|
print("Install it with: pip install requests")
|
|
return
|
|
|
|
# Now start the bot
|
|
bot.start(use_queue=args.use_queue)
|
|
except KeyboardInterrupt:
|
|
print("Bot stopped by user")
|
|
bot.stop()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|