jyapayne/stream-bot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Token and controller commands work

Browse source
This commit is contained in:
Joey Yakimowich-Payne 2025-05-15 12:38:57 -06:00
commit 1a2df3e539
8 changed files with 1137 additions and 376 deletions
Download patch file Download diff file Expand all files Collapse all files

180
scripts/setup_twitch_auth.py
View file

@ -7,9 +7,7 @@ This script helps users generate a Twitch user access token with proper IRC scop
import os
import sys
import logging
import json
import argparse
import time
from dotenv import load_dotenv
# Load environment variables if available
@ -25,43 +23,7 @@ from src.core.auth import TwitchAuth
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger('twitch-auth-setup')
# Path to default config
DEFAULT_CONFIG_PATH = os.path.join("data", "default_config.json")
def load_default_config():
"""Load default configuration from JSON file"""
if os.path.exists(DEFAULT_CONFIG_PATH):
try:
with open(DEFAULT_CONFIG_PATH, 'r') as f:
return json.load(f)
except json.JSONDecodeError:
logger.warning(f"Invalid JSON in {DEFAULT_CONFIG_PATH}, ignoring")
return {}
def save_default_config(config_data):
"""Save configuration to default config JSON file"""
# Ensure directory exists
os.makedirs(os.path.dirname(DEFAULT_CONFIG_PATH), exist_ok=True)
# If file exists, load current data and update it
existing_data = {}
if os.path.exists(DEFAULT_CONFIG_PATH):
try:
with open(DEFAULT_CONFIG_PATH, 'r') as f:
existing_data = json.load(f)
except json.JSONDecodeError:
logger.warning(f"Invalid JSON in {DEFAULT_CONFIG_PATH}, overwriting")
# Update with new data
existing_data.update(config_data)
# Save back to file
with open(DEFAULT_CONFIG_PATH, 'w') as f:
json.dump(existing_data, f, indent=2)
logger.info(f"Configuration saved to {DEFAULT_CONFIG_PATH}")
def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mode=False, force_new_token=False, use_cached_token=False):
def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mode=False, force_new_token=False):
"""
Setup Twitch authentication by generating a user access token
@ -71,46 +33,21 @@ def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mo
scopes: Space-separated list of scopes
manual_mode: Whether to use manual mode (no browser)
force_new_token: Force generation of a new token even if a cached one exists
use_cached_token: Whether to use a cached token if available (default: False)
Returns:
bool: True if successful, False otherwise
"""
try:
# Load default config
default_config = load_default_config()
# Use provided values, then env vars
client_id = client_id or os.environ.get('TWITCH_CLIENT_ID')
client_secret = client_secret or os.environ.get('TWITCH_CLIENT_SECRET')
# Priority of values:
# 1. Function arguments
# 2. Environment variables
# 3. Default config
# Use provided values, then env vars, then default config
client_id = client_id or os.environ.get('TWITCH_CLIENT_ID') or default_config.get('TWITCH_CLIENT_ID')
client_secret = client_secret or os.environ.get('TWITCH_CLIENT_SECRET') or default_config.get('TWITCH_CLIENT_SECRET')
if not client_id:
client_id = input("Enter your Twitch Client ID: ")
if not client_secret:
client_secret = input("Enter your Twitch Client Secret: ")
if not client_id or not client_secret:
logger.error("Client ID and Client Secret are required")
return False
# Default scopes for IRC chat - prioritize env var, then default config
# Default scopes for IRC chat
default_scopes = "chat:read chat:edit"
scopes = scopes or os.environ.get('TWITCH_SCOPES') or default_config.get('TWITCH_SCOPES', default_scopes)
scopes = scopes or os.environ.get('TWITCH_SCOPES', default_scopes)
# Define redirect URI
redirect_uri = os.environ.get('TWITCH_REDIRECT_URI') or default_config.get('TWITCH_REDIRECT_URI', "https://localhost:3000")
print("\n===== Twitch Authentication Setup =====")
print(f"Client ID: {client_id}")
print(f"Redirect URI: {redirect_uri} (must match your Twitch app settings)")
print(f"Requested Scopes: {scopes}")
print("=====================================\n")
redirect_uri = os.environ.get('TWITCH_REDIRECT_URI', "https://localhost:3000")
# Make sure the cache directory exists
cache_dir = os.path.join("data", "cache")
@ -118,11 +55,6 @@ def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mo
# Create the token cache file path
token_cache_file = os.path.join(cache_dir, "token_cache.json")
# By default, clear any cached tokens unless explicitly told to use them
if (not use_cached_token or force_new_token) and os.path.exists(token_cache_file):
logger.info("Clearing cached token to generate fresh credentials")
os.remove(token_cache_file)
# Create auth handler
auth = TwitchAuth(
@ -133,97 +65,12 @@ def setup_twitch_auth(client_id=None, client_secret=None, scopes=None, manual_mo
scopes=scopes
)
# Check if a valid token is already available
has_valid_token = False
if use_cached_token and auth.access_token and auth.token_expiry and time.time() < (auth.token_expiry - 60):
has_valid_token = True
# Using cached token
if not force_new_token:
print("\nUsing cached token (valid and not expired).")
print("If you want to force a new token, run with --force-new-token")
try:
# Try to validate the token to make sure it's working
token_info = auth.validate_token()
print(f"Token belongs to user: {token_info.get('login', 'unknown')}")
expiry_hours = (auth.token_expiry - time.time()) / 3600
print(f"Token expires in {expiry_hours:.1f} hours")
# Skip browser flow and continue with cached token
oauth_token = f"oauth:{auth.access_token}"
except Exception as e:
logger.warning(f"Error validating cached token: {e}")
logger.warning("Will attempt to get a new token")
has_valid_token = False
# If no valid token, go through the auth flow
if not has_valid_token or force_new_token:
# Show security warning about certificates
if not manual_mode:
print("\n⚠️ IMPORTANT SECURITY NOTE ⚠️")
print("This application uses a self-signed certificate for HTTPS, which is required by Twitch.")
print("When your browser opens, you will see a security warning.")
print("This is expected for local development. Please proceed to the site anyway:")
print(" • In Chrome: Click 'Advanced' and then 'Proceed to localhost (unsafe)'")
print(" • In Firefox: Click 'Advanced', then 'Accept the Risk and Continue'")
print(" • In Edge: Click 'Details' and then 'Go on to the webpage'\n")
print("NOTE: If the browser doesn't open automatically, a URL will be provided for you to copy and paste.")
input("Press Enter to continue...")
# Start OAuth flow
logger.info("Starting OAuth authentication flow...")
oauth_token = auth.get_oauth_token(manual_auth=manual_mode)
# At this point, we should have a valid token
if auth.access_token:
# Get token info
try:
token_info = auth.validate_token()
print("\n===== Authentication Successful =====")
print(f"Access Token: {auth.access_token}")
if auth.refresh_token:
print(f"Refresh Token: {auth.refresh_token}")
print(f"Token Scopes: {token_info.get('scopes', [])}")
print(f"User Name: {token_info.get('login', 'unknown')}")
print("=====================================\n")
except Exception as e:
logger.error(f"Failed to validate token: {e}")
print("\n===== Authentication Status =====")
print(f"Access Token: {auth.access_token}")
if auth.refresh_token:
print(f"Refresh Token: {auth.refresh_token}")
print("Warning: Could not validate token")
print("============================\n")
# Store token info in default_config.json
if input("Would you like to save these credentials to default_config.json? (y/n): ").lower() == 'y':
config_data = {
"TWITCH_CLIENT_ID": client_id,
"TWITCH_CLIENT_SECRET": client_secret,
}
# Try to use the validated user info
if 'token_info' in locals():
username = token_info.get('login', '')
config_data["TWITCH_USERNAME"] = username
# Get channel name
default_channel = username if 'username' in locals() else os.environ.get('TWITCH_CHANNEL', '') or default_config.get('TWITCH_CHANNEL', '')
channel = input(f"Enter Twitch channel to join (default: {default_channel}): ") or default_channel
config_data["TWITCH_CHANNEL"] = channel
# Save access and refresh tokens
config_data["TWITCH_ACCESS_TOKEN"] = auth.access_token
if auth.refresh_token:
config_data["TWITCH_REFRESH_TOKEN"] = auth.refresh_token
# Save to default_config.json
save_default_config(config_data)
# Get the token, which will automatically trigger setup if needed
try:
oauth_token = auth.get_oauth_token(manual_auth=manual_mode, force_new_token=force_new_token)
return True
else:
logger.error("Failed to get OAuth token")
except ValueError as e:
logger.error(f"Authentication failed: {e}")
return False
except Exception as e:
@ -238,11 +85,10 @@ def main():
parser.add_argument("--scopes", help="Space-separated list of scopes")
parser.add_argument("--manual", action="store_true", help="Use manual mode (no browser)")
parser.add_argument("--force-new-token", action="store_true", help="Force generation of a new token even if a cached one exists")
parser.add_argument("--use-cached-token", action="store_true", help="Use cached token if available (default is to clear cached tokens)")
args = parser.parse_args()
if setup_twitch_auth(args.client_id, args.client_secret, args.scopes, args.manual, args.force_new_token, args.use_cached_token):
if setup_twitch_auth(args.client_id, args.client_secret, args.scopes, args.manual, args.force_new_token):
print("Setup completed successfully!")
sys.exit(0)
else: