import os import sys import time import requests from dotenv import load_dotenv # Load local environment settings load_dotenv() API_URL = os.getenv("API_URL", "http://localhost:3000") API_KEY = os.getenv("API_KEY", "") if not API_KEY: print("[Error] API_KEY is missing in .env! Cannot push data to backend.") sys.exit(1) headers = { "x-api-key": API_KEY, "Content-Type": "application/json" } def get_all_streams(): """Fetch all streams from the backend""" try: res = requests.get(f"{API_URL}/api/streams", headers=headers, timeout=10) if res.status_code == 200: data = res.json() # API returns { streams: [...] } or just [...] if isinstance(data, list): return data return data.get("streams", []) except Exception as e: print(f"[Error] Не удалось получить список стримов: {e}") return [] def find_vod_id_by_stream_id(stream_id): """Try to find a VOD ID for a given Twitch stream ID using GQL""" url = "https://gql.twitch.tv/gql" gql_headers = { "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko", "User-Agent": "Mozilla/5.0" } payload = { "query": f""" query {{ channel(name: "{os.getenv('TWITCH_CHANNEL', 'winx_prinx')}") {{ videos(first: 30, type: ARCHIVE) {{ edges {{ node {{ id broadcastType lengthSeconds publishedAt stream {{ id }} }} }} }} }} }} """ } try: res = requests.post(url, json=payload, headers=gql_headers, timeout=10) if res.status_code == 200: data = res.json() edges = data.get("data", {}).get("channel", {}).get("videos", {}).get("edges", []) for edge in edges: node = edge.get("node", {}) stream_info = node.get("stream") or {} if stream_info.get("id") == str(stream_id): return node.get("id") except Exception as e: pass return None def extract_vod_id(stream): """Extract VOD ID from stream record""" twitch_id = str(stream.get("twitch_stream_id", "")) if twitch_id.startswith("vod-"): return twitch_id[4:] # Live stream ID — try to find corresponding VOD print(f" -> Поиск VOD для live stream ID {twitch_id}...") vod_id = find_vod_id_by_stream_id(twitch_id) if vod_id: print(f" -> Найден VOD ID: {vod_id}") else: print(f" -> VOD не найден (стрим мог быть удалён или VOD недоступен)") return vod_id def download_vod_roles(vod_id): """Download chat log of a past VOD and extract only roles, with retry on failure""" roles_dict = {} url = "https://gql.twitch.tv/gql" gql_headers = { "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko", "User-Agent": "Mozilla/5.0" } current_offset = 0 total_messages = 0 MAX_RETRIES = 5 with requests.Session() as session: while True: payload = { "operationName": "VideoCommentsByOffsetOrCursor", "variables": { "videoID": str(vod_id), "contentOffsetSeconds": current_offset }, "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a" } } } success = False for attempt in range(MAX_RETRIES): try: res = session.post(url, json=payload, headers=gql_headers, timeout=15) if res.status_code == 429: wait = 2 ** attempt print(f"\n [!] Rate limited (429). Повтор через {wait}с...") time.sleep(wait) continue if res.status_code != 200: print(f"\n [!] Ошибка GQL статус {res.status_code}") break data = res.json() if isinstance(data, list): data = data[0] video = data.get("data", {}).get("video", {}) if not video: return list(roles_dict.values()) # VOD ended or not found edges = (video.get("comments") or {}).get("edges") or [] if not edges: return list(roles_dict.values()) # Reached end for edge in edges: if not edge: continue node = edge.get("node") if not node: continue commenter = node.get("commenter") if not commenter: continue user = commenter.get("login") if not user: continue display_name = commenter.get("displayName", user) message = node.get("message") if not message: continue timestamp = node.get("createdAt") user_badges = message.get("userBadges") or [] badges = [b.get("setID") for b in user_badges if b] # Debug: print first badge encounter to verify GQL structure if total_messages < 3 and user_badges: print(f"\n [DEBUG] userBadges raw: {user_badges}") print(f" [DEBUG] parsed badges: {badges}") is_mod = "moderator" in badges or "broadcaster" in badges is_sub = "subscriber" in badges or "founder" in badges is_vip = "vip" in badges # Only store users who have at least one badge if is_mod or is_sub or is_vip: roles_dict[user.lower()] = { "username": user, "displayName": display_name, "isMod": is_mod, "isSub": is_sub, "isVip": is_vip, "timestamp": timestamp } total_messages += 1 print(f" -> Обработано {total_messages} сообщений, найдено {len(roles_dict)} уникальных пользователей...", end="\r") last_offset = edges[-1].get("node", {}).get("contentOffsetSeconds") if last_offset is not None: next_offset = last_offset + 1 if next_offset <= current_offset: next_offset = current_offset + 30 current_offset = next_offset else: return list(roles_dict.values()) # Tiny sleep to avoid aggressive spamming time.sleep(0.1) success = True break # Success — go to next page except Exception as e: wait = 2 ** attempt # 1, 2, 4, 8, 16 seconds print(f"\n [!] Ошибка (попытка {attempt+1}/{MAX_RETRIES}): {e}") if attempt < MAX_RETRIES - 1: print(f" [!] Повтор через {wait}с...") time.sleep(wait) if not success: print(f"\n [!] Не удалось получить данные после {MAX_RETRIES} попыток. Останавливаемся на offset={current_offset}.") break return list(roles_dict.values()) def upload_roles(roles, vod_id): """Send roles to backend in batches""" batch_size = 500 for i in range(0, len(roles), batch_size): batch = roles[i:i+batch_size] try: res = requests.post(f"{API_URL}/api/log/roles", json={"roles": batch}, headers=headers, timeout=10) if res.status_code == 200: print(f" -> Обновлено пользователей: {i + len(batch)} / {len(roles)}", end="\r") else: print(f"\n [!] Ошибка отправки: {res.status_code}") except Exception as e: print(f"\n [!] Ошибка сети: {e}") print() if __name__ == "__main__": print("=========================================================") print(" Ultra-Fast Role Backfiller (AUTO MODE) ") print("=========================================================") print(f"Backend: {API_URL}") print() # Fetch all streams print("[1] Получение списка стримов с сервера...") streams = get_all_streams() if not streams: print("[Error] Список стримов пустой или не удалось получить!") sys.exit(1) # Filter only VODs (twitch_stream_id starts with "vod-") vod_streams = [s for s in streams if str(s.get("twitch_stream_id", "")).startswith("vod-")] print(f"[+] Найдено стримов всего: {len(streams)}") print() # Process each stream for idx, stream in enumerate(streams, 1): vod_id = extract_vod_id(stream) title = stream.get("title", "Без названия") print(f"[{idx}/{len(streams)}] Stream — «{title[:50]}»") if not vod_id: print(" -> Пропуск: нет VOD ID") continue roles = download_vod_roles(vod_id) print(f"\n -> Найдено пользователей с бейджами: {len(roles)}") if roles: upload_roles(roles, vod_id) print(f" -> ✅ Готово!") else: print(f" -> Чат пустой или VOD недоступен, пропуск.") print() print("=========================================================") print(" Восстановление ролей завершено! ") print("=========================================================")