local_worker/vod_backfiller.py

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("=========================================================")