# twitch_bot.py - Twitch бот import socket import re import time import threading import random import requests import json from datetime import datetime from config import * from commands import COMMANDS, MOD_COMMANDS, AUTO_MESSAGES from database import Database class TwitchBot: def __init__(self, database, telegram_bot=None): self.db = database self.telegram_bot = telegram_bot # Ссылка на Telegram бота self.sock = socket.socket() self.connected = False self.broadcaster = "" # Статус стрима self.is_stream_online = False # Отслеживание активности self.user_messages = {} self.stream_start_time = datetime.now() self.total_messages = 0 self.active_users = set() # Обычные розыгрыши self.giveaway_active = False self.giveaway_item = "" self.eligible_users = [] self.current_winner = None self.winner_deadline = 0 self.waiting_for_trade_link = False # Активный дроп self.active_drop_enabled = False self.active_drop_users = {} self.drop_winners = [] self.current_drop_winner = None self.drop_winner_deadline = 0 self.waiting_for_drop_trade_link = False # Игра с числами self.number_game_active = False self.secret_number = 0 self.number_game_max = 0 self.number_game_winner = None self.number_game_deadline = 0 self.waiting_for_number_game_link = False # ========== TWITCH API ========== def check_stream_status(self): """Проверить статус стрима через публичное API""" try: url = f"https://decapi.me/twitch/uptime/{self.broadcaster}" response = requests.get(url, timeout=5) if response.status_code == 200: text = response.text.strip() is_online = text != f"{self.broadcaster} is offline" if is_online != self.is_stream_online: print(f"{'🟢' if is_online else '🔴'} [TWITCH] Стрим {'ОНЛАЙН' if is_online else 'ОФЛАЙН'}") self.is_stream_online = is_online return is_online return self.is_stream_online except Exception as e: print(f"⚠️ [TWITCH] Ошибка проверки: {e}") return self.is_stream_online def stream_status_checker(self): """Постоянная проверка статуса стрима""" while self.connected: self.check_stream_status() time.sleep(CHECK_STREAM_INTERVAL) # ========== ПОДКЛЮЧЕНИЕ ========== def connect(self): """Подключение к Twitch IRC""" try: self.sock.connect((TWITCH_IRC, TWITCH_PORT)) self.sock.send(f"PASS {OAUTH_TOKEN}\r\n".encode()) self.sock.send(f"NICK {BOT_USERNAME}\r\n".encode()) self.sock.send(f"JOIN {CHANNEL}\r\n".encode()) self.sock.send("CAP REQ :twitch.tv/tags\r\n".encode()) self.sock.send("CAP REQ :twitch.tv/commands\r\n".encode()) self.connected = True self.broadcaster = CHANNEL[1:] print(f"✅ [TWITCH] Подключено к {CHANNEL}") except Exception as e: print(f"❌ [TWITCH] Ошибка подключения: {e}") def send_message(self, message): """Отправка сообщения в чат""" if not self.is_stream_online: print(f"⏸️ [TWITCH] Не отправлено (офлайн): {message[:50]}...") return try: self.sock.send(f"PRIVMSG {CHANNEL} :{message}\r\n".encode()) print(f"📤 [TWITCH] {message}") time.sleep(1) except Exception as e: print(f"❌ [TWITCH] Ошибка отправки: {e}") # ========== ПАРСИНГ ========== def parse_message(self, data): """Парсинг IRC сообщения""" try: if "PRIVMSG" not in data: return "", None, None if data.startswith("@"): tags_end = data.find(" :") tags = data[:tags_end] if tags_end > 0 else "" username_match = re.search(r":(\w+)!\w+@\w+\.tmi\.twitch\.tv", data) if not username_match: return "", None, None username = username_match.group(1) msg_match = re.search(r"PRIVMSG #\w+ :(.+)", data) if msg_match: message = msg_match.group(1).strip() return tags, username, message pattern = r":(\w+)!\w+@\w+\.tmi\.twitch\.tv PRIVMSG #\w+ :(.+)" match = re.search(pattern, data) if match: username = match.group(1) message = match.group(2).strip() return "", username, message except: pass return "", None, None def is_mod_or_broadcaster(self, tags, username): """Проверка прав модератора""" if username.lower() == self.broadcaster.lower(): return True if "badges=" in tags: badges_match = re.search(r"badges=([^;]+)", tags) if badges_match: badges = badges_match.group(1) if "broadcaster/" in badges or "moderator/" in badges: return True return False # ========== ОТСЛЕЖИВАНИЕ АКТИВНОСТИ ========== def track_user_activity(self, username): """Отслеживание активности пользователей""" if username.lower() != BOT_USERNAME.lower(): self.user_messages[username] = self.user_messages.get(username, 0) + 1 self.total_messages += 1 self.active_users.add(username) if self.active_drop_enabled: self.active_drop_users[username] = time.time() # ========== ОБРАБОТКА КОМАНД ========== def handle_regular_commands(self, message): """Обработка обычных команд""" message_lower = message.lower().strip() if message_lower in COMMANDS: self.send_message(COMMANDS[message_lower]) return True return False def handle_mod_basic_commands(self, tags, username, message): """Обработка базовых команд модераторов""" if not self.is_mod_or_broadcaster(tags, username): return False message_lower = message.lower().strip() if message_lower in MOD_COMMANDS: self.send_message(MOD_COMMANDS[message_lower]) return True # !stats if message_lower == "!stats": duration = datetime.now() - self.stream_start_time hours = int(duration.total_seconds() // 3600) minutes = int((duration.total_seconds() % 3600) // 60) self.send_message( f"📊 Стрим: {hours}ч {minutes}м | Сообщений: {self.total_messages} | Юзеров: {len(self.active_users)}") return True # !savestats if message_lower == "!savestats": if self.db.save_stream_stats(self.stream_start_time, self.total_messages, self.active_users): self.send_message("✅ Статистика сохранена!") self.stream_start_time = datetime.now() self.total_messages = 0 self.active_users.clear() else: self.send_message("❌ Ошибка сохранения") return True return False # ========== АКТИВНЫЙ ДРОП ========== def get_active_drop_users(self): """Получить активных пользователей за последние 3 минуты""" current_time = time.time() active_users = [] for username, last_time in list(self.active_drop_users.items()): if current_time - last_time <= ACTIVE_DROP_ACTIVITY_WINDOW: active_users.append(username) else: del self.active_drop_users[username] return active_users def active_drop_announcer(self): """Автоматические объявления о дропе""" while self.active_drop_enabled: time.sleep(ACTIVE_DROP_ANNOUNCE_INTERVAL) if self.active_drop_enabled and self.is_stream_online: active_count = len(self.get_active_drop_users()) self.send_message( f"🎁 АКТИВНЫЙ ДРОП НА СТРИМЕ! За просмотр турнира и активность в чате вы участвуете в дропе 🔥 💬 Пишите в чат, оставайтесь на стриме — в любой момент можно получить скин! (Активных: {active_count})") def start_active_drop(self): """Запуск активного дропа""" if self.active_drop_enabled: self.send_message("⚠️ Активный дроп уже запущен!") return self.active_drop_enabled = True self.active_drop_users.clear() self.drop_winners.clear() self.send_message("🎁 АКТИВНЫЙ ДРОП ЗАПУЩЕН! 🔥 Пишите в чат и оставайтесь на стриме!") threading.Thread(target=self.active_drop_announcer, daemon=True).start() print("✅ [TWITCH] Активный дроп запущен") def execute_drop(self): """Выбрать победителя дропа""" if not self.active_drop_enabled: self.send_message("⚠️ Активный дроп не запущен!") return active_users = self.get_active_drop_users() if not active_users: self.send_message("😢 Нет активных участников!") return # Считаем количество участников user_count = len(active_users) self.current_drop_winner = random.choice(active_users) self.drop_winner_deadline = time.time() + DROP_RESPONSE_TIME self.waiting_for_drop_trade_link = True self.send_message(f"🎉 ДРОП! Победитель: @{self.current_drop_winner}! ⏰ У тебя 3 минуты! Отправь трейд-ссылку") # ОБНОВЛЕННЫЙ БЛОК ЗАПИСИ data = { "winner": self.current_drop_winner, "active": True, "user_count": user_count # Добавляем количество участников } with open('winner_data.json', 'w', encoding='utf-8') as f: json.dump(data, f) print(f"DEBUG: Победитель {self.current_drop_winner} отправлен в оверлей") # 2. Запускаем поток для очистки (через 10-15 секунд) threading.Thread(target=self.delayed_clear).start() def timeout_check(): time.sleep(DROP_RESPONSE_TIME) if self.waiting_for_drop_trade_link: self.send_message(f"⏰ @{self.current_drop_winner} не успел!") self.waiting_for_drop_trade_link = False self.current_drop_winner = None threading.Thread(target=timeout_check, daemon=True).start() def handle_active_drop_commands(self, tags, username, message): """Обработка команд активного дропа""" if not self.is_mod_or_broadcaster(tags, username): return False message_lower = message.lower().strip() if message_lower == "!activedrop": self.start_active_drop() return True if message_lower == "!stopactivedrop": self.active_drop_enabled = False self.send_message(f"🛑 Дроп завершён! Победителей: {len(self.drop_winners)}") return True if message_lower == "!drop": self.execute_drop() return True if message_lower == "!dropstats": active = len(self.get_active_drop_users()) self.send_message(f"📊 Активных: {active} | Победителей: {len(self.drop_winners)}") return True return False # ========== ИГРА С ЧИСЛАМИ ========== def start_number_game(self, max_number): """Запуск игры с угадыванием числа""" if self.number_game_active: self.send_message("⚠️ Игра уже идёт!") return self.number_game_active = True self.secret_number = random.randint(1, max_number) self.number_game_max = max_number self.send_message(f"🎲 ИГРА НАЧАЛАСЬ! Угадайте число от 1 до {max_number}!") self.send_message(f"💬 Пишите свои варианты в чат! Первый кто угадает - получит приз!") print(f"🎲 [TWITCH] ===== ИГРА С ЧИСЛАМИ ЗАПУЩЕНА =====") print(f"🎲 [TWITCH] Диапазон: 1 - {max_number}") print(f"🎲 [TWITCH] Загаданное число: {self.secret_number}") print(f"🎲 [TWITCH] ===========================================") def check_number_guess(self, username, message): """Проверка попытки угадать число""" if not self.number_game_active: return # Пробуем извлечь число из сообщения try: # Ищем числа в сообщении numbers = re.findall(r'\b\d+\b', message) if not numbers: return guess = int(numbers[0]) # Берём первое число # Проверяем диапазон if guess < 1 or guess > self.number_game_max: return # Проверяем угадал ли if guess == self.secret_number: self.number_game_active = False self.number_game_winner = username self.number_game_deadline = time.time() + DROP_RESPONSE_TIME self.waiting_for_number_game_link = True self.send_message(f"🎉 ПОБЕДА! @{username} угадал число {self.secret_number}!") self.send_message(f"⏰ У тебя 3 минуты! Отправь трейд-ссылку в чат!") print(f"🎉 [TWITCH] ===== ПОБЕДИТЕЛЬ НАЙДЕН =====") print(f"🎉 [TWITCH] Победитель: {username}") print(f"🎉 [TWITCH] Угаданное число: {self.secret_number}") print(f"🎉 [TWITCH] ================================") # Запускаем таймер def timeout_check(): time.sleep(DROP_RESPONSE_TIME) if self.waiting_for_number_game_link: self.send_message(f"⏰ @{self.number_game_winner} не успел!") self.waiting_for_number_game_link = False self.number_game_winner = None print(f"⏰ [TWITCH] {username} не отправил трейд-ссылку вовремя") threading.Thread(target=timeout_check, daemon=True).start() else: # Подсказка if guess < self.secret_number: hint = "больше" else: hint = "меньше" print(f"🎲 [TWITCH] {username}: {guess} (число {hint})") except ValueError: pass def handle_number_game_commands(self, tags, username, message): """Обработка команд игры с числами""" if not self.is_mod_or_broadcaster(tags, username): return False message_lower = message.lower().strip() print(f"🔍 [DEBUG] Проверяю команду игры: '{message_lower}'") # !randomnum <максимум> if message_lower.startswith("!randomnum"): parts = message.split() if len(parts) >= 2 and parts[1].isdigit(): max_num = int(parts[1]) if max_num < 10: self.send_message("❌ Число должно быть минимум 10!") elif max_num > 100000: self.send_message("❌ Число не должно превышать 100000!") else: self.start_number_game(max_num) else: self.send_message("❌ Используй: !randomnum <число>") return True # !stopnumber - остановить игру if message_lower == "!stopnumber": print(f"🔍 [DEBUG] Команда !stopnumber получена") print(f"🔍 [DEBUG] Игра активна: {self.number_game_active}") if self.number_game_active: self.number_game_active = False self.send_message(f"🛑 Игра остановлена! Загаданное число было: {self.secret_number}") print(f"🛑 [TWITCH] Игра остановлена модератором ({username})") else: self.send_message("❌ Игра не запущена!") return True return False # ========== ПРОВЕРКА ТРЕЙД-ССЫЛОК ========== def check_trade_link(self, username, message): """Проверка трейд-ссылки от победителя""" # Для обычного розыгрыша if self.waiting_for_trade_link: if username.lower() == self.current_winner.lower(): match = re.search(TRADE_LINK_PATTERN, message) if match: trade_link = match.group(0) if time.time() <= self.winner_deadline: self.send_message(f"✅ @{username} прошёл проверку!") if self.db.save_winner(username, trade_link): self.send_message(f"💾 Данные сохранены!") self.giveaway_active = False self.waiting_for_trade_link = False self.current_winner = None # Для активного дропа if self.waiting_for_drop_trade_link: if username.lower() == self.current_drop_winner.lower(): match = re.search(TRADE_LINK_PATTERN, message) if match: trade_link = match.group(0) if time.time() <= self.drop_winner_deadline: self.send_message(f"✅ @{username} дроп получен!") if self.db.save_winner(username, trade_link): self.send_message(f"💾 Победитель сохранён!") self.drop_winners.append(username) self.waiting_for_drop_trade_link = False self.current_drop_winner = None # Задержка 5 секунд перед очисткой оверлея def delayed_clear(): time.sleep(5) self.clear_overlay() threading.Thread(target=delayed_clear, daemon=True).start() # Для игры с числами if self.waiting_for_number_game_link: if username.lower() == self.number_game_winner.lower(): match = re.search(TRADE_LINK_PATTERN, message) if match: trade_link = match.group(0) if time.time() <= self.number_game_deadline: self.send_message(f"✅ @{username} приз получен!") if self.db.save_winner(username, trade_link): self.send_message(f"💾 Победитель сохранён!") print(f"💾 [TWITCH] Трейд-ссылка от {username} сохранена в БД") self.waiting_for_number_game_link = False self.number_game_winner = None # ========== АВТОСООБЩЕНИЯ ========== def auto_message_sender(self, message, interval): """Автоматическая отправка сообщений""" while self.connected: time.sleep(interval * 60) if self.connected and self.is_stream_online: self.send_message(message) def start_auto_messages(self): """Запуск автосообщений""" for auto_msg in AUTO_MESSAGES: threading.Thread( target=self.auto_message_sender, args=(auto_msg["message"], auto_msg["interval"]), daemon=True ).start() # ========== ОСНОВНОЙ ЦИКЛ ========== def run(self): """Запуск бота""" if not self.connected: self.connect() # Проверка статуса стрима через DecAPI (без регистрации) print("🔍 [TWITCH] Запуск мониторинга статуса стрима...") self.check_stream_status() threading.Thread(target=self.stream_status_checker, daemon=True).start() self.start_auto_messages() print("✅ [TWITCH] Бот запущен!") while True: try: data = self.sock.recv(2048).decode('utf-8', errors='ignore') if data.startswith("PING"): self.sock.send("PONG :tmi.twitch.tv\r\n".encode()) continue tags, username, message = self.parse_message(data) if username and message: self.track_user_activity(username) self.check_trade_link(username, message) # Проверяем игру с числами (до обработки команд) self.check_number_guess(username, message) if self.handle_regular_commands(message): continue if self.handle_mod_basic_commands(tags, username, message): continue if self.handle_active_drop_commands(tags, username, message): continue if self.handle_number_game_commands(tags, username, message): continue except KeyboardInterrupt: print("\n👋 [TWITCH] Остановка...") self.db.save_stream_stats(self.stream_start_time, self.total_messages, self.active_users) break except Exception as e: print(f"⚠️ [TWITCH] Ошибка: {e}") time.sleep(5) self.sock.close()