如何通过 TwitchIO 版本 2.10.0 登录 twitch 机器人？

我有一个可以运行的机器人，大概已经使用了一周，它会更新令牌，添加新的令牌。当我在寻找如何使这一切工作时，我花了很多时间，所以我在这里分发它，也许有人会找到它，这样可以节省时间

先决条件

• TwitchIO 2.10.0
• Python 3.10
• 你的机器人的 Twitch 帐户
• 机器人工作频道的 Twitch 帐户

步骤 1：在 Twitch 上注册你的应用

1. 前往Twitch 开发者控制台
2. 点击“注册您的申请”
3. 填写必填字段：
   • 名称：你的机器人的名称
   • OAuth 重定向 URL：（http://localhost:3000用于本地开发）
   • 类别：选择“聊天机器人”
4. 客户类型：机密
5. 点击“创建”
6. 创建后，单击新应用程序上的“管理”。
7. 记下您的客户ID
8. 生成并记下您的客户端密钥

第 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：创建 token 验证和刷新模块

创建一个名为的文件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. 按照说明进行操作：
   • 输入您的客户端 ID 和客户端密钥
   • 输入机器人用户名和频道名称
   • 使用机器人账户授权应用程序

步骤 8：启动机器人

python3 main.py

如果一切设置正确，您的机器人应该连接到指定的 Twitch 频道并响应命令!пинг。

常见问题及其解决方案

令牌验证失败

如果 token 验证失败，可能是因为：

• 令牌过期（应自动更新）
• 客户端 ID 或密钥无效
• 权限（范围）不正确

解决方案：重新启动脚本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.