feat: add backfill resume support and session/retry fixes for VODs

local_worker/role_backfiller.py

@@ -102,107 +102,116 @@ def download_vod_roles(vod_id):
     total_messages = 0
     MAX_RETRIES = 5
 
-    while True:
-        payload = {
-            "operationName": "VideoCommentsByOffsetOrCursor",
-            "variables": {
-                "videoID": str(vod_id),
-                "contentOffsetSeconds": current_offset
-            },
-            "extensions": {
-                "persistedQuery": {
-                    "version": 1,
-                    "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a"
+    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 = requests.post(url, json=payload, headers=gql_headers, timeout=15)
-                if res.status_code != 200:
-                    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:
+            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
-                    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())
-
-                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
+                    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())
 

local_worker/vod_backfiller.py

@@ -95,108 +95,130 @@ def download_vod_chat(vod_id, start_offset=0, end_offset=None):
     current_offset = start_offset
     print(f"\n[Chat] Запуск скачивания чата для VOD {vod_id} с секунды {start_offset} до {'конца' if end_offset is None else str(end_offset) + 's'}...")
     
-    while True:
-        if end_offset is not None and current_offset >= end_offset:
-            print(f"\n[Chat] Достигнут конечный офсет {end_offset}s. Завершаем скачивание чата.")
-            break
+    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"
+            payload = {
+                "operationName": "VideoCommentsByOffsetOrCursor",
+                "variables": {
+                    "videoID": str(vod_id),
+                    "contentOffsetSeconds": current_offset
+                },
+                "extensions": {
+                    "persistedQuery": {
+                        "version": 1,
+                        "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a"
+                    }
                 }
             }
-        }
-            
-        try:
-            res = requests.post(url, json=payload, headers=gql_headers, timeout=10)
-            if res.status_code != 200:
-                print(f"[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
+            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
                     
-                commenter = node.get("commenter")
-                if not commenter:
-                    continue
-                
-                user = commenter.get("login")
-                if not user:
-                    continue
+                    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")
                     
-                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)
             
-            # 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:
+            if not success:
+                print(f"\n[Chat] Не удалось продолжить скачивание из-за ошибок на offset {current_offset}.")
                 break
-        except Exception as e:
-            print(f"\n[Chat] Ошибка при загрузке: {e}")
-            break
-            
+                
     print(f"\n[Chat] Загрузка завершена. Всего сообщений: {len(comments)}")
     return comments
 
@@ -376,8 +398,51 @@ if __name__ == "__main__":
         print(f"[Error] Ошибка сети при обращении к бэкенду: {e}")
         sys.exit(1)
         
-    # 3. Download and Upload Chat comments
-    chat_comments = download_vod_chat(vod_id)
+    # 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
@@ -394,7 +459,7 @@ if __name__ == "__main__":
                 print(f"\n-> Ошибка сети при отправке пакета: {e}")
         print("\n-> Синхронизация чата успешно завершена!")
         
-    # 4. Transcribe Audio
+    # 5. Transcribe Audio
     transcribe_confirm = input("\nХотите запустить скачивание и транскрибацию аудиодорожки (Whisper)? (y/n): ").strip().lower()
     if transcribe_confirm == 'y':
         # Init Whisper
@@ -406,7 +471,7 @@ if __name__ == "__main__":
                 compute_type=WHISPER_COMPUTE_TYPE,
                 cpu_threads=WHISPER_CPU_THREADS
             )
-            transcribe_vod_audio(vod_id, created_at, model)
+            transcribe_vod_audio(vod_id, created_at, model, start_offset=voice_start_offset)
         except Exception as e:
             print(f"[Whisper Error] Не удалось инициализировать Whisper: {e}")
             

local_worker/worker.py

@@ -722,6 +722,22 @@ def run_stream_capture(model):
         
         print(f"[Capture] Stream is LIVE! Starting audio pipeline...")
         
+        # Notify backend that stream has started and get stream id for coverage
+        notify_stream_start()
+        
+        # Start coverage window for the active stream
+        try:
+            import requests as _req
+            _headers = {"x-api-key": API_KEY}
+            _stream_res = _req.get(f"{API_URL}/api/streams/active", headers=_headers, timeout=5)
+            if _stream_res.status_code == 200:
+                _active = _stream_res.json()
+                _sid = _active.get("stream", {}).get("id") or _active.get("id")
+                if _sid:
+                    start_coverage(_sid, source='live')
+        except Exception as _e:
+            print(f"[Coverage] Could not get active stream for coverage: {_e}")
+        
         # Start streamlink piping audio to ffmpeg, converting it to raw 16kHz mono 16-bit PCM
         streamlink_cmd = ["streamlink", f"twitch.tv/{TWITCH_CHANNEL}", "audio,worst", "-O"]
         ffmpeg_cmd = ["ffmpeg", "-i", "pipe:0", "-ac", "1", "-ar", "16000", "-f", "s16le", "-"]
@@ -853,22 +869,6 @@ if __name__ == "__main__":
     t_sender.start()
     t_heartbeat.start()
     
-    # Notify backend that stream has started and get stream id for coverage
-    notify_stream_start()
-    
-    # Start coverage window for the active stream
-    try:
-        import requests as _req
-        _headers = {"x-api-key": API_KEY}
-        _stream_res = _req.get(f"{API_URL}/api/streams/active", headers=_headers, timeout=5)
-        if _stream_res.status_code == 200:
-            _active = _stream_res.json()
-            _sid = _active.get("stream", {}).get("id") or _active.get("id")
-            if _sid:
-                start_coverage(_sid, source='live')
-    except Exception as _e:
-        print(f"[Coverage] Could not get active stream for coverage: {_e}")
-    
     # 3. Start Audio Capture (Blocks main thread)
     try:
         if CAPTURE_METHOD == "stream":

server/db.js

@@ -2022,3 +2022,85 @@ export async function logViewerJoin(username) {
     }
   }
 }
+
+/**
+ * Get backfill status for a specific stream by its twitch_stream_id
+ */
+export async function getStreamBackfillStatus(twitchStreamId) {
+  let stream = null;
+  if (dbMode === 'sqlite') {
+    stream = sqliteDb.prepare('SELECT * FROM streams WHERE twitch_stream_id = ?').get(twitchStreamId);
+  } else {
+    const { data, error } = await supabase
+      .from('streams')
+      .select('*')
+      .eq('twitch_stream_id', twitchStreamId)
+      .limit(1);
+    if (data && data.length > 0) {
+      stream = data[0];
+    }
+  }
+
+  if (!stream) return null;
+
+  let latestMessageTimestamp = null;
+  let latestVoiceTimestamp = null;
+  let messageCount = 0;
+  let voiceCount = 0;
+
+  if (dbMode === 'sqlite') {
+    const msgStats = sqliteDb.prepare('SELECT COUNT(*) as count, MAX(timestamp) as latest FROM messages WHERE stream_id = ?').get(stream.id);
+    messageCount = msgStats.count;
+    latestMessageTimestamp = msgStats.latest;
+
+    const voiceStats = sqliteDb.prepare('SELECT COUNT(*) as count, MAX(timestamp) as latest FROM voice_words WHERE stream_id = ?').get(stream.id);
+    voiceCount = voiceStats.count;
+    latestVoiceTimestamp = voiceStats.latest;
+  } else {
+    // Supabase
+    const { count: msgCount } = await supabase
+      .from('messages')
+      .select('*', { count: 'exact', head: true })
+      .eq('stream_id', stream.id);
+    messageCount = msgCount || 0;
+
+    if (messageCount > 0) {
+      const { data: latestMsg } = await supabase
+        .from('messages')
+        .select('timestamp')
+        .eq('stream_id', stream.id)
+        .order('timestamp', { ascending: false })
+        .limit(1);
+      if (latestMsg && latestMsg.length > 0) {
+        latestMessageTimestamp = latestMsg[0].timestamp;
+      }
+    }
+
+    const { count: vCount } = await supabase
+      .from('voice_words')
+      .select('*', { count: 'exact', head: true })
+      .eq('stream_id', stream.id);
+    voiceCount = vCount || 0;
+
+    if (voiceCount > 0) {
+      const { data: latestVoice } = await supabase
+        .from('voice_words')
+        .select('timestamp')
+        .eq('stream_id', stream.id)
+        .order('timestamp', { ascending: false })
+        .limit(1);
+      if (latestVoice && latestVoice.length > 0) {
+        latestVoiceTimestamp = latestVoice[0].timestamp;
+      }
+    }
+  }
+
+  return {
+    streamId: stream.id,
+    twitchStreamId: stream.twitch_stream_id,
+    messageCount,
+    latestMessageTimestamp,
+    voiceCount,
+    latestVoiceTimestamp
+  };
+}

server/server.js

@@ -39,7 +39,8 @@ import {
   cleanupGhostStreams,
   startCoverageWindow,
   updateCoverageHeartbeat,
-  endCoverageWindow
+  endCoverageWindow,
+  getStreamBackfillStatus
 } from './db.js';
 // import { initializeEventSub } from './eventsub.js';
 import { cache, cacheMiddleware } from './cache.js';
@@ -283,6 +284,25 @@ app.get('/api/streams/pending-backfill', authenticateWorker, async (req, res) =>
   }
 });
 
+// GET backfill status for a specific stream (chat messages count & latest timestamp, voice words count & latest timestamp)
+app.get('/api/streams/backfill-status', authenticateWorker, async (req, res) => {
+  const { twitchStreamId } = req.query;
+  if (!twitchStreamId) {
+    return res.status(400).json({ error: 'Missing twitchStreamId query parameter' });
+  }
+
+  try {
+    const status = await getStreamBackfillStatus(twitchStreamId);
+    if (!status) {
+      return res.status(404).json({ error: 'Stream not found' });
+    }
+    res.json(status);
+  } catch (err) {
+    console.error('[API Error] /api/streams/backfill-status:', err);
+    res.status(500).json({ error: 'Internal Server Error' });
+  }
+});
+
 // GET list of streams missing twitch_vod_id
 app.get('/api/streams/missing-vod', authenticateWorker, async (req, res) => {
   try {