import os import sys # Setup dynamic CUDA paths before importing Whisper nvidia_subdirs = [ ("nvidia", "cublas", "lib"), ("nvidia", "cudnn", "lib"), ("nvidia", "cuda_nvrtc", "lib") ] added_paths = [] for p in sys.path: if "site-packages" in p: for subdir in nvidia_subdirs: full_path = os.path.join(p, *subdir) if os.path.exists(full_path) and full_path not in added_paths: added_paths.append(full_path) if added_paths: existing = os.environ.get("LD_LIBRARY_PATH", "") if existing: os.environ["LD_LIBRARY_PATH"] = ":".join(added_paths) + ":" + existing else: os.environ["LD_LIBRARY_PATH"] = ":".join(added_paths) import time import subprocess import requests import numpy as np from dotenv import load_dotenv from faster_whisper import WhisperModel # Load local environment settings load_dotenv() API_URL = os.getenv("API_URL", "http://localhost:3000") API_KEY = os.getenv("API_KEY", "") TWITCH_CHANNEL = os.getenv("TWITCH_CHANNEL", "winx_prinx").lower() WHISPER_MODEL_SIZE = os.getenv("WHISPER_MODEL", "base") WHISPER_DEVICE = os.getenv("WHISPER_DEVICE", "cpu") WHISPER_COMPUTE_TYPE = os.getenv("WHISPER_COMPUTE_TYPE", "int8") WHISPER_CPU_THREADS = int(os.getenv("WHISPER_CPU_THREADS", "2")) 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" } # ========================================================================= # PUBLIC TWITCH GQL API CLIENT # ========================================================================= def get_vod_details(vod_id): """Fetch VOD metadata using public Twitch GQL API""" url = "https://gql.twitch.tv/gql" gql_headers = { "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } payload = { "query": f""" query {{ video(id: "{vod_id}") {{ title createdAt lengthSeconds owner {{ login }} }} }} """ } try: res = requests.post(url, json=payload, headers=gql_headers, timeout=10) if res.status_code == 200: data = res.json() return data.get("data", {}).get("video") except Exception as e: print(f"[Error] Failed to fetch VOD details: {e}") return None def download_vod_chat(vod_id, start_offset=0, end_offset=None): """Download chat log of a past VOD using GQL offset-based pagination (bypasses integrity checks)""" comments = [] seen_ids = set() url = "https://gql.twitch.tv/gql" gql_headers = { "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko", "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" } current_offset = start_offset print(f"\n[Chat] Запуск скачивания чата для VOD {vod_id} с секунды {start_offset} до {'конца' if end_offset is None else str(end_offset) + 's'}...") MAX_RETRIES = 5 with requests.Session() as session: while True: if end_offset is not None and current_offset >= end_offset: print(f"\n[Chat] Достигнут конечный офсет {end_offset}s. Завершаем скачивание чата.") break 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[Chat] Rate limited (429). Повтор через {wait}с...") time.sleep(wait) continue if res.status_code != 200: print(f"\n[Chat] Ошибка GQL статус {res.status_code}") break data = res.json() if isinstance(data, list): data = data[0] video = data.get("data", {}).get("video", {}) if not video: break comments_edge = video.get("comments") or {} edges = comments_edge.get("edges") or [] if not edges: break for edge in edges: if not edge: continue node = edge.get("node") if not node: continue msg_id = node.get("id") if not msg_id or msg_id in seen_ids: 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 fragments = message.get("fragments") or [] message_text = "".join([f.get("text", "") for f in fragments if f]) timestamp = node.get("createdAt") # Badges parse user_badges = message.get("userBadges") or [] badges = [b.get("setID") for b in user_badges if b] is_mod = "moderator" in badges or "broadcaster" in badges is_sub = "subscriber" in badges or "founder" in badges is_vip = "vip" in badges is_streamer = (user.lower() == TWITCH_CHANNEL) seen_ids.add(msg_id) comments.append({ "id": msg_id, "username": user, "displayName": display_name, "message": message_text, "timestamp": timestamp, "isStreamer": is_streamer, "isMod": is_mod, "isSub": is_sub, "isVip": is_vip }) print(f"-> Загружено {len(comments)} комментариев...", end="\r") # Progress offset 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: break # Tiny sleep to avoid aggressive spamming time.sleep(0.1) success = True break except Exception as e: wait = 2 ** attempt print(f"\n[Chat] Ошибка при загрузке (попытка {attempt+1}/{MAX_RETRIES}): {e}") if attempt < MAX_RETRIES - 1: print(f"[Chat] Повтор через {wait}с...") time.sleep(wait) if not success: print(f"\n[Chat] Не удалось продолжить скачивание из-за ошибок на offset {current_offset}.") break print(f"\n[Chat] Загрузка завершена. Всего сообщений: {len(comments)}") return comments # ========================================================================= # AUDIO TRANSCRIBER (WHISPER) # ========================================================================= def transcribe_vod_audio(vod_id, start_time_iso, model, start_offset=0, end_offset=None): """Download VOD audio stream, seek to start_offset, transcribe with Whisper and upload""" print(f"\n[Audio] Запуск скачивания и распознавания речи для VOD {vod_id} с секунды {start_offset} до {'конца' if end_offset is None else str(end_offset) + 's'}...") # Check if streamlink is installed try: subprocess.run(["streamlink", "--version"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) except FileNotFoundError: print("[Error] 'streamlink' utility is not installed! Cannot fetch VOD audio.") return cmd_streamlink = ["streamlink", f"twitch.tv/videos/{vod_id}", "audio,worst", "-O"] if start_offset > 0: h = start_offset // 3600 m = (start_offset % 3600) // 60 s = start_offset % 60 offset_str = f"{h:02d}:{m:02d}:{s:02d}" cmd_streamlink.extend(["--hls-start-offset", offset_str]) print(f"[Audio] Запуск streamlink с поиском --hls-start-offset {offset_str}") cmd_ffmpeg = ["ffmpeg", "-i", "pipe:0", "-ac", "1", "-ar", "16000", "-f", "s16le", "-"] p_streamlink = None p_ffmpeg = None try: p_streamlink = subprocess.Popen(cmd_streamlink, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) p_ffmpeg = subprocess.Popen(cmd_ffmpeg, stdin=p_streamlink.stdout, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL) # 30-second audio chunks for batch processing (more efficient offline) # 16000Hz * 2 bytes * 30 seconds = 960,000 bytes chunk_seconds = 30 chunk_size = 16000 * 2 * chunk_seconds offset_seconds = start_offset try: from datetime import datetime clean_str = start_time_iso.replace('Z', '+00:00') base_start_time = datetime.fromisoformat(clean_str).timestamp() except Exception as parse_e: print(f"[Audio] Error parsing start_time_iso '{start_time_iso}': {parse_e}") # Fallback to seconds-only precision clean_t = start_time_iso.split(".")[0].replace("Z", "").replace("+00:00", "") base_start_time = time.mktime(time.strptime(clean_t, "%Y-%m-%dT%H:%M:%S")) while True: if end_offset is not None and offset_seconds >= end_offset: print(f"\n[Audio] Достигнут конечный офсет {end_offset}s. Завершаем транскрибацию аудио.") break pcm_data = p_ffmpeg.stdout.read(chunk_size) if not pcm_data: break # Convert raw 16-bit PCM bytes to float32 numpy array audio_np = np.frombuffer(pcm_data, dtype=np.int16).astype(np.float32) / 32768.0 # Transcribe segment segments, info = model.transcribe( audio_np, beam_size=5, language="ru", temperature=0.0, condition_on_previous_text=False, log_prob_threshold=-0.8, no_speech_threshold=0.6, vad_filter=True, vad_parameters=dict(threshold=0.5, min_silence_duration_ms=500) ) words_list = [] for segment in segments: text = segment.text.strip() if text: words_list.extend(text.split()) if words_list: # Calculate correct timestamp based on elapsed video time word_time_epoch = base_start_time + offset_seconds word_time_iso = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(word_time_epoch)) # Send to backend try: url = f"{API_URL}/api/log/voice" res = requests.post(url, json={"words": words_list, "timestamp": word_time_iso}, headers=headers, timeout=5) if res.status_code == 200: print(f"[{time.strftime('%H:%M:%S', time.gmtime(offset_seconds))}] Распознано и отправлено {len(words_list)} слов.") except Exception as e: print(f"[Sync Error] Не удалось отправить слова: {e}") offset_seconds += chunk_seconds except Exception as err: print(f"[Audio Error] Ошибка конвейера аудио: {err}") finally: for p in [p_ffmpeg, p_streamlink]: if p: try: p.terminate() p.wait(timeout=2) except: try: p.kill() except: pass print("[Audio] Распознавание аудиодорожки завершено.") # ========================================================================= # MAIN EXECUTION # ========================================================================= if __name__ == "__main__": print("=========================================================") print(" Twitch VOD Backfiller & Speech Sync (winx_prinx) ") print("=========================================================") print(f"Target Channel: {TWITCH_CHANNEL}") print(f"API Backend: {API_URL}") print("=========================================================") vod_id = input("Введите ID Twitch VOD (например, 2154382910): ").strip() if not vod_id: print("ID VOD не введен. Выход.") sys.exit(0) # 1. Fetch metadata print("[Sync] Получение метаданных VOD...") details = get_vod_details(vod_id) if not details: print("[Error] Не удалось получить информацию о VOD. Проверьте ID видео.") sys.exit(1) owner = details.get("owner", {}).get("login", "").lower() if owner != TWITCH_CHANNEL: print(f"[Warning] Владелец VOD ({owner}) не совпадает с целевым каналом ({TWITCH_CHANNEL})!") confirm = input("Продолжить импорт? (y/n): ").strip().lower() if confirm != 'y': sys.exit(0) title = details.get("title", "Архивный стрим") created_at_raw = details.get("createdAt") # e.g. "2026-06-09T17:15:30Z" length_seconds = details.get("lengthSeconds", 0) # timezone-safe calculation using datetime from datetime import datetime, timedelta created_at = created_at_raw.replace("+00:00", "").replace("Z", "") + "Z" dt_start = datetime.fromisoformat(created_at_raw.replace("Z", "+00:00")) dt_end = dt_start + timedelta(seconds=length_seconds) end_time = dt_end.strftime("%Y-%m-%dT%H:%M:%SZ") print(f"\nНазвание: {title}") print(f"Начало: {created_at}") print(f"Длительность: {length_seconds // 3600}ч {(length_seconds % 3600) // 60}м") # 2. Sync stream record in DB print("\n[Sync] Регистрация сессии стрима на бэкенде...") try: url = f"{API_URL}/api/streams/sync-vod-direct" payload = { "twitchStreamId": f"vod-{vod_id}", "title": title, "category": "Архив", "startTime": created_at, "endTime": end_time } res = requests.post(url, json=payload, headers=headers, timeout=5) if res.status_code != 200: print(f"[Error] Не удалось синхронизировать сессию стрима: {res.status_code} - {res.text}") sys.exit(1) print("-> Сессия стрима успешно зарегистрирована в базе!") except Exception as e: print(f"[Error] Ошибка сети при обращении к бэкенду: {e}") sys.exit(1) # 3. Check for existing backfill status (resume/append support) chat_start_offset = 0 voice_start_offset = 0 try: status_url = f"{API_URL}/api/streams/backfill-status?twitchStreamId=vod-{vod_id}" res = requests.get(status_url, headers=headers, timeout=5) if res.status_code == 200: status_data = res.json() # Chat check msg_count = status_data.get("messageCount", 0) latest_msg_ts = status_data.get("latestMessageTimestamp") if msg_count > 0 and latest_msg_ts: from datetime import datetime clean_ts = latest_msg_ts.replace('Z', '+00:00') latest_epoch = datetime.fromisoformat(clean_ts).timestamp() clean_created = created_at.replace('Z', '+00:00') created_epoch = datetime.fromisoformat(clean_created).timestamp() diff = int(latest_epoch - created_epoch) if diff > 0: chat_start_offset = diff print(f"[Sync] Найден существующий чат в базе ({msg_count} сообщений). Возобновляем загрузку с секунды {chat_start_offset}...") # Voice check voice_count = status_data.get("voiceCount", 0) latest_voice_ts = status_data.get("latestVoiceTimestamp") if voice_count > 0 and latest_voice_ts: from datetime import datetime clean_ts = latest_voice_ts.replace('Z', '+00:00') latest_epoch = datetime.fromisoformat(clean_ts).timestamp() clean_created = created_at.replace('Z', '+00:00') created_epoch = datetime.fromisoformat(clean_created).timestamp() diff = int(latest_epoch - created_epoch) if diff > 0: voice_start_offset = (diff // 30) * 30 print(f"[Sync] Найден распознанный голос в базе ({voice_count} слов). Возобновляем транскрибацию с секунды {voice_start_offset}...") except Exception as e: print(f"[Sync Warning] Не удалось получить статус дозаписи: {e}") # 4. Download and Upload Chat comments chat_comments = download_vod_chat(vod_id, start_offset=chat_start_offset) if chat_comments: print("\n[Sync] Отправка чата на сервер (пакетами по 100 сообщений)...") batch_size = 100 for i in range(0, len(chat_comments), batch_size): batch = chat_comments[i:i+batch_size] try: url = f"{API_URL}/api/log/messages" res = requests.post(url, json={"messages": batch}, headers=headers, timeout=5) if res.status_code == 200: print(f"-> Отправлено сообщений: {i + len(batch)} / {len(chat_comments)}", end="\r") else: print(f"\n-> Ошибка отправки пакета: {res.status_code}") except Exception as e: print(f"\n-> Ошибка сети при отправке пакета: {e}") print("\n-> Синхронизация чата успешно завершена!") # 5. Transcribe Audio if voice_start_offset >= length_seconds: print(f"\n[Sync] Голосовая дорожка уже полностью распознана (секунда {voice_start_offset} из {length_seconds}). Транскрибация не требуется.") else: prompt_str = f"\nХотите запустить скачивание и транскрибацию аудиодорожки (Whisper)? (y/n) [Продолжить с {voice_start_offset}s]: " if voice_start_offset > 0 else "\nХотите запустить скачивание и транскрибацию аудиодорожки (Whisper)? (y/n): " transcribe_confirm = input(prompt_str).strip().lower() if transcribe_confirm == 'y': # Init Whisper print(f"\n[Whisper] Загрузка модели {WHISPER_MODEL_SIZE}...") try: model = WhisperModel( WHISPER_MODEL_SIZE, device=WHISPER_DEVICE, compute_type=WHISPER_COMPUTE_TYPE, cpu_threads=WHISPER_CPU_THREADS ) transcribe_vod_audio(vod_id, created_at, model, start_offset=voice_start_offset) except Exception as e: print(f"[Whisper Error] Не удалось инициализировать Whisper: {e}") # 5. Mark Completed on Server try: url_mark = f"{API_URL}/api/streams/mark-backfilled" requests.post(url_mark, json={"twitchStreamId": f"vod-{vod_id}"}, headers=headers, timeout=5) print("-> Статус импорта успешно обновлен на сервере!") except Exception as e: print(f"[Warning] Не удалось отметить стрим как импортированный: {e}") print("\n=========================================================") print(" Импорт VOD завершен! ") print("=========================================================")