Как авторизоваться на твич бота через TwitchIO версии 2.10.0?

У меня есть рабочий бот, пользуюсь им уже, наверное, неделю, токены обновляет, новые добавляет Когда сам искал как всё это заставить работать потратил кучу времени, поэтому распространяю тут, авось найдётся кому это сократит время

Предварительные требования

• TwitchIO 2.10.0
• Python 3.10
• Аккаунт Twitch для вашего бота
• Аккаунт Twitch для канала, где будет работать бот

Шаг 1: Зарегистрируйте ваше приложение на Twitch

1. Перейдите в Консоль разработчика Twitch
2. Нажмите "Register Your Application" (Зарегистрировать приложение)
3. Заполните необходимые поля:
   • Name: Имя вашего бота
   • OAuth Redirect URL: http://localhost:3000 (для локальной разработки)
   • Category: Выберите "Chat Bot" (Чат-бот)
4. Тип клиента: Конфиденциально
5. Нажмите "Create" (Создать)
6. После создания нажмите "Manage" (Управление) на вашем новом приложении
7. Запишите ваш Client ID (Идентификатор клиента)
8. Сгенерируйте и запишите ваш Client Secret (Секретный код клиента)

Шаг 2: Установите необходимые пакеты

pip3 install twitchio python-dotenv requests

Шаг 3: Создайте скрипт настройки аутентификации

Создайте файл с именем twitch_oauth_setup.py:

#!/usr/bin/env python3
import os
import sys
import webbrowser
import requests
from urllib.parse import urlparse, parse_qs

print("Настройка Twitch OAuth для бота")
print("==============================")

# Функция для получения ввода пользователя
def get_input(prompt):
    return input(f"{prompt}: ").strip()

# Проверка наличия файла .env
env_data = {}
if os.path.exists('.env'):
    print("Найден существующий файл .env. Чтение данных...")
    with open('.env', 'r') as file:
        for line in file:
            if '=' in line:
                key, value = line.strip().split('=', 1)
                env_data[key] = value
    print("Данные из файла .env загружены.")

# Получение необходимых данных от пользователя или из .env
client_id = env_data.get('CLIENT_ID') or get_input("Введите ваш Client ID")
client_secret = env_data.get('CLIENT_SECRET') or get_input("Введите ваш Client Secret")
bot_username = env_data.get('BOT_USERNAME') or get_input("Введите имя аккаунта бота (НЕ ваш основной аккаунт)")
channel_name = env_data.get('CHANNEL_NAME') or get_input("Введите имя канала, где будет работать бот")

# Сохраняем во временные данные .env
temp_env_data = {
    'BOT_USERNAME': bot_username,
    'CHANNEL_NAME': channel_name,
    'CLIENT_ID': client_id,
    'CLIENT_SECRET': client_secret,
}

# Генерация URL авторизации
# Эти разрешения позволяют боту читать/отправлять сообщения и выполнять действия модерации
scopes = 'chat:read+chat:edit+channel:moderate+moderator:manage:chat_messages+moderator:manage:banned_users+whispers:edit'
auth_url = f"https://id.twitch.tv/oauth2/authorize?response_type=code&client_id={client_id}&redirect_uri=http://localhost:3000&scope={scopes}&force_verify=true"

print("\n============ ВАЖНО ============")
print("1. Перед продолжением ВЫЙДИТЕ из своего основного аккаунта Twitch")
print("2. Войдите в аккаунт БОТА ({})".format(bot_username))
print("3. Убедитесь, что вы назначили бота модератором вашего канала")
print("   Для этого напишите в чате вашего канала: /mod {}".format(bot_username))
print("============================\n")

input("После выполнения этих шагов нажмите Enter для продолжения...")

# Открываем URL авторизации в браузере
print(f"\nОткрывается URL для авторизации. Войдите в аккаунт БОТА ({bot_username}) и разрешите доступ.")
print(f"URL авторизации: {auth_url}")
webbrowser.open(auth_url)

# Ожидаем ввода кода авторизации
print("\nПосле авторизации вы будете перенаправлены на localhost с ошибкой 'Этот сайт недоступен'.")
print("Скопируйте весь URL из адресной строки браузера (он содержит код авторизации).")

redirect_url = get_input("Вставьте полный URL, на который вы были перенаправлены")

# Извлечение кода авторизации из URL
parsed_url = urlparse(redirect_url)
query_params = parse_qs(parsed_url.query)
auth_code = query_params.get('code', [None])[0]

if not auth_code:
    print("Ошибка: Не удалось извлечь код авторизации из URL.")
    sys.exit(1)

print(f"Код авторизации получен: {auth_code}")

# Обмен кода авторизации на токены
print("\nОбмен кода авторизации на токены доступа...")

token_url = "https://id.twitch.tv/oauth2/token"
payload = {
    'client_id': client_id,
    'client_secret': client_secret,
    'code': auth_code,
    'grant_type': 'authorization_code',
    'redirect_uri': 'http://localhost:3000'
}

try:
    response = requests.post(token_url, data=payload)
    response.raise_for_status()
    token_data = response.json()
    
    # Проверяем, что в ответе есть необходимые токены
    if 'access_token' not in token_data or 'refresh_token' not in token_data:
        print("Ошибка: Не удалось получить токены доступа. Ответ сервера:")
        print(token_data)
        sys.exit(1)
    
    access_token = token_data['access_token']
    refresh_token = token_data['refresh_token']
    scopes = token_data.get('scope', [])
    
    print("Токены успешно получены!")
    print(f"Access Token: {access_token[:10]}...{access_token[-5:]} (срок действия: {token_data['expires_in']} секунд)")
    print(f"Refresh Token: {refresh_token[:10]}...{refresh_token[-5:]}")
    print(f"Полученные разрешения: {', '.join(scopes)}")
    
    # Проверяем валидность токена
    print("\nПроверка валидности токена...")
    headers = {
        'Authorization': f'Bearer {access_token}'
    }
    validate_response = requests.get('https://id.twitch.tv/oauth2/validate', headers=headers)
    
    if validate_response.status_code == 200:
        validate_data = validate_response.json()
        print(f"Токен действителен! Авторизован как: {validate_data.get('login')}")
        print(f"ID пользователя: {validate_data.get('user_id')}")
        
        # Проверяем, что токен принадлежит аккаунту бота
        if validate_data.get('login') != bot_username.lower():
            print(f"\nВНИМАНИЕ! Токен получен для аккаунта {validate_data.get('login')}, а не для бота {bot_username}!")
            print("Это означает, что вы авторизовались под неправильным аккаунтом.")
            print("Рекомендуется очистить куки браузера и повторить процесс, войдя под аккаунтом бота.")
            proceed = input("Хотите продолжить несмотря на это? (да/нет): ").lower()
            if proceed != 'да':
                print("Операция прервана пользователем.")
                sys.exit(1)
    else:
        print("Ошибка при валидации токена:")
        print(validate_response.text)
    
    # Сохраняем все данные в файл .env
    print("\nСохранение данных в файл .env...")
    temp_env_data['ACCESS_TOKEN'] = access_token
    temp_env_data['REFRESH_TOKEN'] = refresh_token
    
    with open('.env', 'w') as env_file:
        for key, value in temp_env_data.items():
            env_file.write(f"{key}={value}\n")
    
    print("\nНастройка успешно завершена! Файл .env создан/обновлен.")
    
except requests.exceptions.RequestException as e:
    print(f"Ошибка при запросе токенов: {e}")
    sys.exit(1)

Шаг 4: Создайте модуль валидации и обновления токенов

Создайте файл с именем twitch_auth.py:

import os
import logging
import requests
from dotenv import load_dotenv, set_key

# Настройка логирования
logging.basicConfig(level=logging.INFO, 
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('twitch_auth')

# Загрузка переменных окружения из файла .env
load_dotenv()

# Настройки аутентификации
BOT_USERNAME = os.getenv('BOT_USERNAME')
CHANNEL_NAME = os.getenv('CHANNEL_NAME')
CLIENT_ID = os.getenv('CLIENT_ID')
CLIENT_SECRET = os.getenv('CLIENT_SECRET')
ACCESS_TOKEN = os.getenv('ACCESS_TOKEN')
REFRESH_TOKEN = os.getenv('REFRESH_TOKEN')

def refresh_oauth_token_sync():
    """Синхронная версия обновления OAuth токена для использования перед инициализацией бота"""
    global ACCESS_TOKEN, REFRESH_TOKEN
    logger.info("Попытка обновить OAuth токен синхронно...")
    
    try:
        # Подготовка запроса для обновления токена
        refresh_url = "https://id.twitch.tv/oauth2/token"
        payload = {
            'client_id': CLIENT_ID,
            'client_secret': CLIENT_SECRET,
            'refresh_token': REFRESH_TOKEN,
            'grant_type': 'refresh_token'
        }
        
        # Выполнение запроса
        response = requests.post(refresh_url, data=payload)
        response.raise_for_status()
        token_data = response.json()
        
        # Обновление токенов
        ACCESS_TOKEN = token_data['access_token']
        REFRESH_TOKEN = token_data['refresh_token']
        
        # Обновление файла .env
        dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
        set_key(dotenv_path, 'ACCESS_TOKEN', ACCESS_TOKEN)
        set_key(dotenv_path, 'REFRESH_TOKEN', REFRESH_TOKEN)
        
        logger.info("OAuth токен успешно обновлен")
        return True
    except Exception as e:
        logger.error(f"Не удалось обновить OAuth токен: {str(e)}")
        return False

def validate_token_sync():
    """Синхронная версия валидации токена для использования перед инициализацией бота"""
    global ACCESS_TOKEN
    try:
        headers = {
            'Authorization': f'Bearer {ACCESS_TOKEN}',
            'Client-Id': CLIENT_ID
        }
        response = requests.get('https://id.twitch.tv/oauth2/validate', headers=headers)
        
        if response.status_code == 200:
            data = response.json()
            logger.info(f"Токен успешно валидирован. Пользователь: {data.get('login')}, истекает через: {data.get('expires_in')} секунд")
            # Если токен близок к истечению (менее 1 часа), обновляем его
            if data.get('expires_in', 0) < 3600:
                logger.info("Токен близок к истечению, обновляем...")
                refresh_oauth_token_sync()
            return True
        else:
            logger.warning(f"Валидация токена не удалась: {response.status_code}")
            return refresh_oauth_token_sync()
    except Exception as e:
        logger.error(f"Ошибка при валидации токена: {str(e)}")
        return refresh_oauth_token_sync()

def get_auth_credentials():
    """Возвращает текущие учетные данные аутентификации для использования ботом"""
    return {
        'bot_username': BOT_USERNAME,
        'channel_name': CHANNEL_NAME,
        'client_id': CLIENT_ID,
        'client_secret': CLIENT_SECRET,
        'access_token': ACCESS_TOKEN,
        'refresh_token': REFRESH_TOKEN
    }

Шаг 5: Создайте класс бота

Создайте файл с именем bot.py:

import os
import logging
import requests
from twitchio.ext import commands
from dotenv import set_key, load_dotenv

# Настройка логирования
logging.basicConfig(level=logging.INFO, 
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('twitch_bot')

load_dotenv()

class TwitchBot(commands.Bot):
    def __init__(self, auth_creds):
        # Получение учетных данных аутентификации
        self.bot_username = auth_creds['bot_username']
        self.channel_name = auth_creds['channel_name']
        self.client_id = auth_creds['client_id']
        self.client_secret = auth_creds['client_secret']
        self._access_token = auth_creds['access_token']
        self._refresh_token = auth_creds['refresh_token']
        
        # Инициализация бота с нашим токеном доступа, префиксом и списком каналов для подключения при запуске
        logger.info(f"Инициализация бота с именем пользователя: {self.bot_username}")
        logger.info(f"Подключение к каналу: {self.channel_name}")
        
        # Вызов родительского конструктора для инициализации бота
        super().__init__(
            token=self._access_token,
            prefix='!',
            initial_channels=[self.channel_name]
        )
        
        # Валидация токена при запуске
        self.loop.create_task(self.validate_token())
        
    async def validate_token(self):
        """Валидация текущего токена и обновление при необходимости"""
        try:
            headers = {
                'Authorization': f'Bearer {self._access_token}',
                'Client-Id': self.client_id
            }
            response = requests.get('https://id.twitch.tv/oauth2/validate', headers=headers)
            
            if response.status_code == 200:
                data = response.json()
                logger.info(f"Токен успешно валидирован. Пользователь: {data.get('login')}, истекает через: {data.get('expires_in')} секунд")
                # Если токен близок к истечению (менее 1 часа), обновляем его
                if data.get('expires_in', 0) < 3600:
                    logger.info("Токен близок к истечению, обновляем...")
                    await self.refresh_oauth_token()
            else:
                logger.warning(f"Валидация токена не удалась: {response.status_code}")
                await self.refresh_oauth_token()
        except Exception as e:
            logger.error(f"Ошибка при валидации токена: {str(e)}")
            await self.refresh_oauth_token()

    async def event_ready(self):
        """Вызывается один раз, когда бот подключается."""
        logger.info(f"Вход выполнен как | {self.nick}")
        logger.info(f"ID пользователя | {self.user_id}")
        
        # Валидация токена при запуске
        await self.validate_token()

    async def event_message(self, message):
        """Вызывается при получении сообщения в канале."""
        # Логирование деталей сообщения
        logger.info(f"Получено сообщение: {message.content}")
        logger.info(f"Автор сообщения: {message.author.name if message.author else 'None'}")
        
        # Игнорирование сообщений от самого бота
        if message.author is None or message.author.name.lower() == self.bot_username.lower():
            return
            
        # Обработка команд
        await self.handle_commands(message)
        
        # Пример: ответ на конкретное сообщение
        if message.content.lower() == 'привет бот':
            await message.channel.send(f'Привет, @{message.author.name}!')

    async def refresh_oauth_token(self):
        """Обновление OAuth токена с использованием refresh токена"""
        try:
            refresh_url = "https://id.twitch.tv/oauth2/token"
            payload = {
                'client_id': self.client_id,
                'client_secret': self.client_secret,
                'refresh_token': self._refresh_token,
                'grant_type': 'refresh_token'
            }
            
            response = requests.post(refresh_url, data=payload)
            response.raise_for_status()
            token_data = response.json()
            
            # Обновление токенов
            self._access_token = token_data['access_token']
            self._refresh_token = token_data['refresh_token']
            
            # Обновление файла .env
            dotenv_path = os.path.join(os.path.dirname(__file__), '.env')
            set_key(dotenv_path, 'ACCESS_TOKEN', self._access_token)
            set_key(dotenv_path, 'REFRESH_TOKEN', self._refresh_token)
            
            # Обновление токена в боте
            self._connection.token = self._access_token
            
            logger.info("OAuth токен успешно обновлен")
            return True
        except Exception as e:
            logger.error(f"Не удалось обновить OAuth токен: {str(e)}")
            return False

    @commands.command(name='пинг')
    async def ping_command(self, ctx):
        """Простая команда, которая отвечает 'Понг!'"""
        await ctx.send(f'Понг! @{ctx.author.name}')

Шаг 6: Создайте основное приложение

Создайте файл с именем main.py:

import os
import logging
from dotenv import load_dotenv
from bot import TwitchBot
from twitch_auth import validate_token_sync, get_auth_credentials

# Настройка логирования
logging.basicConfig(level=logging.INFO, 
                    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
logger = logging.getLogger('twitch_bot')

def main():
    # Валидация и обновление токена перед запуском бота
    logger.info("Валидация Twitch OAuth токена...")
    if validate_token_sync():
        logger.info("Токен действителен или был успешно обновлен")
        # Получение обновленных учетных данных после потенциального обновления
        auth_creds = get_auth_credentials()
    else:
        logger.error("Не удалось валидировать или обновить токен. Пожалуйста, запустите twitch_oauth_setup.py для получения новых токенов.")
        return
    
    # Инициализация бота с валидированными учетными данными
    bot = TwitchBot(auth_creds)
    
    # Запуск бота
    logger.info("Запуск бота...")
    bot.run()

if __name__ == "__main__":
    main()

Шаг 7: Запустите скрипт настройки

1. Запустите скрипт настройки OAuth для получения токенов:

   python3 twitch_oauth_setup.py
   

2. Следуйте инструкциям для:
   • Ввода вашего Client ID и Client Secret
   • Ввода имени пользователя бота и имени канала
   • Авторизации приложения с аккаунтом бота

Шаг 8: Запустите бота

python3 main.py

Если всё настроено правильно, ваш бот должен подключиться к указанному каналу Twitch и отвечать на команду !пинг.

Распространенные проблемы и их решения

Валидация токена не удается

Если валидация токена не удается, это может быть из-за:

• Истекших токенов (должны автоматически обновляться)
• Недействительного Client ID или Secret
• Неправильных разрешений (scopes)

Решение: Перезапустите скрипт twitch_oauth_setup.py для получения новых токенов.

Бот не отвечает на команды

Проверьте, что:

• Бот подключен к правильному каналу
• Вы используете правильный префикс команд (по умолчанию !)
• Бот имеет необходимые разрешения в канале

Бот не может отправлять сообщения

Убедитесь, что:

• Аккаунт бота не заблокирован и не имеет тайм-аута в канале
• Бот имеет разрешение chat:edit
• В канале не включен режим медленного чата или режим "только для подписчиков"

Расширение возможностей вашего бота

Чтобы добавить больше команд, просто добавьте методы в класс вашего бота с декоратором @commands.command():

@commands.command(name='привет')
async def hello_command(self, ctx):
    await ctx.send(f'Привет, @{ctx.author.name}!')

Для более продвинутых функций ознакомьтесь с документацией TwitchIO.