122 lines
4.5 KiB
Python
122 lines
4.5 KiB
Python
# bot_twitch.py
|
|
import os
|
|
import requests
|
|
import asyncio
|
|
from twitchio.ext import commands
|
|
import importlib
|
|
import cmd_twitch
|
|
|
|
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
|
|
|
|
# 1) Initialize the parent Bot FIRST
|
|
super().__init__(
|
|
token=self.token,
|
|
prefix="!",
|
|
initial_channels=config["twitch_channels"]
|
|
)
|
|
|
|
self.log("Twitch bot initiated", "INFO")
|
|
|
|
# 2) Then load commands
|
|
self.load_commands()
|
|
|
|
async def refresh_access_token(self):
|
|
"""
|
|
Refreshes the Twitch access token using the stored refresh token.
|
|
Retries up to 2 times before logging a fatal error.
|
|
"""
|
|
self.log("Attempting to refresh Twitch token...", "INFO")
|
|
|
|
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.", "INFO")
|
|
return # Success, exit function
|
|
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.", "INFO")
|
|
|
|
except Exception as e:
|
|
self.log(f"Failed to update .env file: {e}", "ERROR")
|
|
|
|
def load_commands(self):
|
|
"""
|
|
Load all commands dynamically from cmd_twitch.py.
|
|
"""
|
|
try:
|
|
importlib.reload(cmd_twitch)
|
|
cmd_twitch.setup(self)
|
|
self.log("Twitch commands loaded successfully.", "INFO")
|
|
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.
|
|
"""
|
|
try:
|
|
self.log(f"Twitch bot connecting...", "INFO")
|
|
self.log(f"...Consider online if no further messages", "INFO")
|
|
await self.start()
|
|
#while True:
|
|
# await self.refresh_access_token()
|
|
# await asyncio.sleep(10800) # Refresh every 3 hours
|
|
except Exception as e:
|
|
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()
|
|
except Exception as e:
|
|
self.log(f"Unable to refresh Twitch token! Twitch bot will be offline!", "CRITICAL")
|