# bot_twitch.py import os import requests import asyncio from twitchio.ext import commands import importlib import cmd_twitch import modules import modules.utility from modules.db import log_message, lookup_user, log_bot_event class TwitchBot(commands.Bot): def __init__(self, config, log_func): self.client_id = os.getenv("TWITCH_CLIENT_ID") self.client_secret = os.getenv("TWITCH_CLIENT_SECRET") self.token = os.getenv("TWITCH_BOT_TOKEN") self.refresh_token = os.getenv("TWITCH_REFRESH_TOKEN") self.log = log_func # Use the logging function from bots.py self.config = config self.db_conn = None # We'll set this self.help_data = None # We'll set this later # 1) Initialize the parent Bot FIRST super().__init__( token=self.token, prefix="!", initial_channels=config["twitch_channels"] ) self.log("Twitch bot initiated") # 2) Then load commands self.load_commands() def set_db_connection(self, db_conn): """ Store the DB connection so that commands can use it. """ self.db_conn = db_conn async def event_message(self, message): """ Called every time a Twitch message is received (chat message in a channel). We'll use this to track the user in our 'users' table. """ # If it's the bot's own message, ignore if message.echo: return # Log the command if it's a command if message.content.startswith("!"): _cmd = message.content[1:] # Remove the leading "!" _cmd_args = _cmd.split(" ")[1:] _cmd = _cmd.split(" ", 1)[0] self.log(f"Command '{_cmd}' (Twitch) initiated by {message.author.name} in #{message.channel.name}", "DEBUG") if len(_cmd_args) > 1: self.log(f"!{_cmd} arguments: {_cmd_args}", "DEBUG") try: # Typically message.author is not None for normal chat messages author = message.author if not author: # just in case return is_bot = False # TODO Implement automatic bot account check user_id = str(author.id) user_name = author.name display_name = author.display_name or user_name self.log(f"Message detected, attempting UUI lookup on {user_name} ...", "DEBUG") modules.utility.track_user_activity( db_conn=self.db_conn, log_func=self.log, platform="twitch", user_id=user_id, username=user_name, display_name=display_name, user_is_bot=is_bot ) self.log("... UUI lookup complete.", "DEBUG") user_data = lookup_user(db_conn=self.db_conn, log_func=self.log, identifier=str(message.author.id), identifier_type="twitch_user_id") user_uuid = user_data["UUID"] if user_data else "UNKNOWN" from modules.db import log_message log_message( db_conn=self.db_conn, log_func=self.log, user_uuid=user_uuid, message_content=message.content or "", platform="twitch", channel=message.channel.name, attachments="" ) except Exception as e: self.log(f"... UUI lookup failed: {e}", "ERROR") # Pass message contents to commands processing await self.handle_commands(message) async def event_ready(self): self.log(f"Twitch bot is online as {self.nick}") log_bot_event(self.db_conn, self.log, "TWITCH_RECONNECTED", "Twitch bot logged in.") async def event_disconnected(self): self.log("Twitch bot has lost connection!", "WARNING") log_bot_event(self.db_conn, self.log, "TWITCH_DISCONNECTED", "Twitch bot lost connection.") async def refresh_access_token(self): """ Refreshes the Twitch access token using the stored refresh token. Retries up to 3 times before logging a fatal error. """ self.log("Attempting to refresh Twitch token...") url = "https://id.twitch.tv/oauth2/token" params = { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "grant_type": "refresh_token" } for attempt in range(3): # Attempt up to 3 times try: response = requests.post(url, params=params) data = response.json() if "access_token" in data: self.token = data["access_token"] self.refresh_token = data.get("refresh_token", self.refresh_token) os.environ["TWITCH_BOT_TOKEN"] = self.token os.environ["TWITCH_REFRESH_TOKEN"] = self.refresh_token self.update_env_file() self.log("Twitch token refreshed successfully. Restarting bot...") # Restart the TwitchIO connection await self.close() # Close the old connection await self.start() # Restart with the new token return # Exit function after successful refresh else: self.log(f"Twitch token refresh failed (Attempt {attempt+1}/3): {data}", "WARNING") except Exception as e: self.log(f"Twitch token refresh error (Attempt {attempt+1}/3): {e}", "ERROR") await asyncio.sleep(10) # Wait before retrying # If all attempts fail, log error self.log("Twitch token refresh failed after 3 attempts.", "FATAL") def update_env_file(self): """ Updates the .env file with the new Twitch token. """ try: with open(".env", "r") as file: lines = file.readlines() with open(".env", "w") as file: for line in lines: if line.startswith("TWITCH_BOT_TOKEN="): file.write(f"TWITCH_BOT_TOKEN={self.token}\n") elif line.startswith("TWITCH_REFRESH_TOKEN="): file.write(f"TWITCH_REFRESH_TOKEN={self.refresh_token}\n") else: file.write(line) self.log("Updated .env file with new Twitch token.") except Exception as e: self.log(f"Failed to update .env file: {e}", "ERROR") def load_commands(self): """ Load all commands from cmd_twitch.py """ try: cmd_twitch.setup(self) self.log("Twitch commands loaded successfully.") # Now load the help info from dictionary/help_twitch.json help_json_path = "dictionary/help_twitch.json" modules.utility.initialize_help_data( bot=self, help_json_path=help_json_path, is_discord=False, # Twitch log_func=self.log ) except Exception as e: self.log(f"Error loading Twitch commands: {e}", "ERROR") async def run(self): """ Run the Twitch bot, refreshing tokens if needed. """ retries = 0 while True: if retries > 3: self.log(f"Twitch bot failed to connect after {retries} attempts.", "CIRITCAL") break # Break loop if repeatedly failing to connect to Twitch try: await self.start() #while True: # await self.refresh_access_token() # await asyncio.sleep(10800) # Refresh every 3 hours except Exception as e: retries += 1 self.log(f"Twitch bot failed to start: {e}", "CRITICAL") if "Invalid or unauthorized Access Token passed." in str(e): try: await self.refresh_access_token() self.log("Retrying bot connection after token refresh...", "INFO") await self.start() # Restart connection with new token return # Exit retry loop except Exception as e: self.log(f"Unable to refresh Twitch token! Twitch bot will be offline!", "CRITICAL") if self._keeper: self._keeper.cancel() if "'NoneType' object has no attribute 'cancel'" in str(e): self.log(f"The Twitch bot experienced an initialization glitch. Try starting again", "FATAL") await asyncio.sleep(5) # Wait before retrying to authenticate