116 lines
4.1 KiB
Python
116 lines
4.1 KiB
Python
import os
|
|
import requests
|
|
import logging
|
|
import asyncio
|
|
from twitchio.ext import commands
|
|
import importlib
|
|
import cmd_twitch
|
|
|
|
class TwitchBot(commands.Bot):
|
|
def __init__(self, config, logger):
|
|
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.logger = logger
|
|
self.config = config
|
|
|
|
# 1) Initialize the parent Bot FIRST
|
|
super().__init__(
|
|
token=self.token,
|
|
prefix="!",
|
|
initial_channels=config["twitch_channels"]
|
|
)
|
|
|
|
# 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.logger.info("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.logger.info("Twitch token refreshed successfully.")
|
|
return # Success, exit function
|
|
else:
|
|
self.logger.warning(f"Twitch token refresh failed (Attempt {attempt+1}/3): {data}")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Twitch token refresh error (Attempt {attempt+1}/3): {e}")
|
|
|
|
await asyncio.sleep(10) # Wait before retrying
|
|
|
|
# If all attempts fail, log error
|
|
from bots import log_error
|
|
log_error("Twitch token refresh failed after 3 attempts.")
|
|
|
|
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.logger.info("Updated .env file with new Twitch token.")
|
|
|
|
except Exception as e:
|
|
self.logger.error(f"Failed to update .env file: {e}")
|
|
|
|
def load_commands(self):
|
|
"""
|
|
Load all commands dynamically from cmd_twitch.py.
|
|
"""
|
|
try:
|
|
importlib.reload(cmd_twitch)
|
|
cmd_twitch.setup(self)
|
|
self.logger.info("Twitch commands loaded successfully.")
|
|
except Exception as e:
|
|
self.logger.error(f"Error loading Twitch commands: {e}")
|
|
|
|
async def run(self):
|
|
"""
|
|
Run the Twitch bot, refreshing tokens if needed.
|
|
"""
|
|
try:
|
|
await self.connect() # Connect to Twitch
|
|
await self.start() # Start the bot event loop
|
|
while True:
|
|
await self.refresh_access_token()
|
|
await asyncio.sleep(10800) # 3 hours
|
|
except Exception as e:
|
|
from bots import log_error
|
|
log_error(f"Twitch bot failed to start: {e}")
|