231 lines
8.8 KiB
Python
231 lines
8.8 KiB
Python
# bot_twitch.py
|
|
import os
|
|
import requests
|
|
import asyncio
|
|
from twitchio.ext import commands
|
|
import importlib
|
|
import cmd_twitch
|
|
|
|
import globals
|
|
|
|
import modules
|
|
import modules.utility
|
|
from modules.db import log_message, lookup_user, log_bot_event
|
|
|
|
class TwitchBot(commands.Bot):
|
|
def __init__(self, config):
|
|
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 = globals.log # 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"]
|
|
)
|
|
|
|
globals.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]
|
|
globals.log(f"Command '{_cmd}' (Twitch) initiated by {message.author.name} in #{message.channel.name}", "DEBUG")
|
|
if len(_cmd_args) > 1: globals.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
|
|
|
|
globals.log(f"Message detected, attempting UUI lookup on {user_name} ...", "DEBUG")
|
|
|
|
modules.utility.track_user_activity(
|
|
db_conn=self.db_conn,
|
|
platform="twitch",
|
|
user_id=user_id,
|
|
username=user_name,
|
|
display_name=display_name,
|
|
user_is_bot=is_bot
|
|
)
|
|
|
|
globals.log("... UUI lookup complete.", "DEBUG")
|
|
|
|
user_data = lookup_user(db_conn=self.db_conn, 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,
|
|
user_uuid=user_uuid,
|
|
message_content=message.content or "",
|
|
platform="twitch",
|
|
channel=message.channel.name,
|
|
attachments=""
|
|
)
|
|
|
|
except Exception as e:
|
|
globals.log(f"... UUI lookup failed: {e}", "ERROR")
|
|
|
|
# Pass message contents to commands processing
|
|
await self.handle_commands(message)
|
|
|
|
|
|
async def event_ready(self):
|
|
globals.log(f"Twitch bot is online as {self.nick}")
|
|
log_bot_event(self.db_conn, "TWITCH_RECONNECTED", "Twitch bot logged in.")
|
|
|
|
async def event_disconnected(self):
|
|
globals.log("Twitch bot has lost connection!", "WARNING")
|
|
log_bot_event(self.db_conn, "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.
|
|
"""
|
|
globals.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()
|
|
|
|
globals.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:
|
|
globals.log(f"Twitch token refresh failed (Attempt {attempt+1}/3): {data}", "WARNING")
|
|
|
|
except Exception as e:
|
|
globals.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
|
|
globals.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)
|
|
|
|
globals.log("Updated .env file with new Twitch token.")
|
|
|
|
except Exception as e:
|
|
globals.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)
|
|
globals.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
|
|
)
|
|
|
|
except Exception as e:
|
|
globals.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:
|
|
globals.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
|
|
globals.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()
|
|
globals.log("Retrying bot connection after token refresh...", "INFO")
|
|
await self.start() # Restart connection with new token
|
|
return # Exit retry loop
|
|
except Exception as e:
|
|
globals.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):
|
|
globals.log(f"The Twitch bot experienced an initialization glitch. Try starting again", "FATAL")
|
|
await asyncio.sleep(5) # Wait before retrying to authenticate
|
|
|