local_worker/worker.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)

from datetime import datetime, timedelta, timezone
import time
last_resolve_time = 0
import socket
import select
import threading
import queue
import subprocess
import random
import requests
import numpy as np
from dotenv import load_dotenv
from faster_whisper import WhisperModel
from vod_backfiller import download_vod_chat, transcribe_vod_audio

# Load configurations
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()
CAPTURE_METHOD = os.getenv("CAPTURE_METHOD", "stream").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"))

# Verify essential secrets
if not API_KEY:
    print("[Error] API_KEY is missing in .env! Local worker cannot push data.")
    sys.exit(1)

# Thread-safe message queue for Twitch chat messages
chat_queue = queue.Queue()
# Flag to signal thread termination
stop_flag = threading.Event()

# =========================================================================
# TWITCH CHAT LOGGER (IRC CLIENT)
# =========================================================================

def parse_irc_tags(tags_str):
    """Parse IRC v3 tags into a dictionary"""
    tags = {}
    if not tags_str:
        return tags
    
    parts = tags_str.split(";")
    for part in parts:
        if "=" in part:
            k, v = part.split("=", 1)
            tags[k] = v
    return tags

def twitch_chat_listener():
    """Background thread to connect to Twitch IRC and read chat messages"""
    server = "irc.chat.twitch.tv"
    port = 6667
    
    # Generate a random anonymous nickname
    anon_nick = f"justinfan{random.randint(10000, 99999)}"
    
    print(f"[Twitch IRC] Connecting anonymously as {anon_nick}...")
    
    while not stop_flag.is_set():
        try:
            irc_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            irc_sock.settimeout(10.0)
            irc_sock.connect((server, port))
            
            # Authenticate anonymously
            irc_sock.send(f"PASS oauth:anonymous\r\n".encode("utf-8"))
            irc_sock.send(f"NICK {anon_nick}\r\n".encode("utf-8"))
            
            # Request tags and commands capability to see badges/sub status
            irc_sock.send("CAP REQ :twitch.tv/tags twitch.tv/commands twitch.tv/membership\r\n".encode("utf-8"))
            
            # Join target channel
            irc_sock.send(f"JOIN #{TWITCH_CHANNEL}\r\n".encode("utf-8"))
            print(f"[Twitch IRC] Joined channel #{TWITCH_CHANNEL}")
            
            buffer = ""
            irc_sock.setblocking(False)
            last_msg_time = time.time()
            
            while not stop_flag.is_set():
                # Check for network timeout (no messages for 5 minutes)
                if time.time() - last_msg_time > 300:
                    print("[Twitch IRC] Network timeout (no messages for 5 minutes). Reconnecting...")
                    break

                # Use select for non-blocking read with timeout to allow exit checks
                ready = select.select([irc_sock], [], [], 1.0)
                if not ready[0]:
                    continue
                
                try:
                    data = irc_sock.recv(4096).decode("utf-8", errors="ignore")
                except socket.timeout:
                    continue
                
                if not data:
                    print("[Twitch IRC] Connection closed by remote host.")
                    break
                
                last_msg_time = time.time()
                buffer += data
                while "\r\n" in buffer:
                    line, buffer = buffer.split("\r\n", 1)
                    
                    # 1. Handle Ping-Pong
                    if line.startswith("PING"):
                        irc_sock.send("PONG :tmi.twitch.tv\r\n".encode("utf-8"))
                        continue
                    
                    # 2. Parse PRIVMSG (chat message)
                    # Twitch format: @tags :user!user@user.tmi.twitch.tv PRIVMSG #channel :message
                    if "PRIVMSG" in line:
                        tags = {}
                        tags_str = ""
                        
                        # Extract tags if they exist
                        if line.startswith("@"):
                            tags_str, remainder = line[1:].split(" ", 1)
                            tags = parse_irc_tags(tags_str)
                            line = remainder
                        
                        parts = line.split(" PRIVMSG ", 1)
                        if len(parts) < 2:
                            continue
                            
                        prefix, msg_parts = parts
                        user = prefix.split("!", 1)[0].replace(":", "")
                        
                        channel_part, message_text = msg_parts.split(" :", 1)
                        
                        # Extract user metadata
                        msg_id = tags.get("id", f"local-{time.time_ns()}")
                        display_name = tags.get("display-name", user)
                        badges = tags.get("badges", "")
                        
                        is_streamer = (user.lower() == TWITCH_CHANNEL)
                        is_mod = "moderator" in badges or "broadcaster" in badges
                        is_sub = "subscriber" in badges or "founder" in badges
                        is_vip = "vip" in badges
                        
                        parsed_msg = {
                            "id": msg_id,
                            "username": user,
                            "displayName": display_name,
                            "message": message_text,
                            "timestamp": new_iso_timestamp(),
                            "isStreamer": is_streamer,
                            "isMod": is_mod,
                            "isSub": is_sub,
                            "isVip": is_vip
                        }
                        
                        # Add to queue
                        chat_queue.put(parsed_msg)
                        
        except Exception as e:
            print(f"[Twitch IRC] Connection error: {e}. Retrying in 10 seconds...")
            time.sleep(10)
        finally:
            try:
                irc_sock.close()
            except:
                pass

def new_iso_timestamp():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

def chat_sender():
    """Periodically sends accumulated chat messages to backend API"""
    print("[Chat Sender] Started background queue sender.")
    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
    
    while not stop_flag.is_set():
        messages = []
        # Drain the queue
        while not chat_queue.empty():
            try:
                messages.append(chat_queue.get_nowait())
            except queue.Empty:
                break
        
        if messages:
            try:
                # 1. Send messages
                url = f"{API_URL}/api/log/messages"
                response = requests.post(url, json={"messages": messages}, headers=headers, timeout=5)
                if response.status_code != 200:
                    print(f"[Chat Sender] Failed to sync messages. API returned status {response.status_code}")
            except Exception as e:
                print(f"[Chat Sender] Network error sending messages: {e}")

            try:
                # 2. Send roles for users who have at least one badge
                roles = [
                    {
                        "username": m["username"],
                        "displayName": m.get("displayName", m["username"]),
                        "isMod": m.get("isMod", False),
                        "isSub": m.get("isSub", False),
                        "isVip": m.get("isVip", False),
                        "timestamp": m.get("timestamp")
                    }
                    for m in messages
                    if m.get("isMod") or m.get("isSub") or m.get("isVip")
                ]
                if roles:
                    requests.post(f"{API_URL}/api/log/roles", json={"roles": roles}, headers=headers, timeout=5)
            except Exception as e:
                print(f"[Chat Sender] Network error sending roles: {e}")
                
        # Wait 5 seconds before next sync
        time.sleep(5)


# =========================================================================
# AUDIO TRANSCRIBER (WHISPER)
# =========================================================================

def init_whisper_model():
    """Load Faster-Whisper model into memory"""
    print(f"[Whisper] Loading model '{WHISPER_MODEL_SIZE}' on {WHISPER_DEVICE} ({WHISPER_COMPUTE_TYPE}, threads={WHISPER_CPU_THREADS})...")
    # This might take a few minutes on first run as the model downloads
    model = WhisperModel(
        WHISPER_MODEL_SIZE, 
        device=WHISPER_DEVICE, 
        compute_type=WHISPER_COMPUTE_TYPE,
        cpu_threads=WHISPER_CPU_THREADS
    )
    print("[Whisper] Model loaded successfully.")
    return model

def transcribe_audio_segment(model, pcm_bytes):
    """Run Whisper transcription on PCM 16kHz 16-bit mono audio bytes"""
    if len(pcm_bytes) == 0:
        return []
    
    # Convert raw 16-bit PCM bytes to float32 numpy array normalized to [-1.0, 1.0]
    audio_np = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
    
    # Force language="ru" for better Russian accuracy and faster execution
    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, # Voice Activity Detection filters out silence
        vad_parameters=dict(threshold=0.5, min_silence_duration_ms=500)
    )
    
    words_list = []
    for segment in segments:
        text = segment.text.strip()
        if text:
            print(f"[Whisper Speech] Transcribed: \"{text}\"")
            # Split segment into clean words
            words = text.split()
            words_list.extend(words)
            
    return words_list

def send_voice_words(words):
    """Post transcribed words to backend server"""
    if not words:
        return
    
    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
    url = f"{API_URL}/api/log/voice"
    
    try:
        response = requests.post(url, json={"words": words, "timestamp": new_iso_timestamp()}, headers=headers, timeout=5)
        if response.status_code == 200:
            print(f"[Sync] Sent {len(words)} spoken words to cloud.")
        else:
            print(f"[Sync] Failed to send words. Status code: {response.status_code}")
    except Exception as e:
        print(f"[Sync] Error sending voice words to cloud: {e}")

def notify_stream_start():
    """Notify backend that the stream has started"""
    headers = {"x-api-key": API_KEY}
    url = f"{API_URL}/api/log/stream-start"
    try:
        response = requests.post(url, headers=headers, timeout=5)
        if response.status_code == 200:
            print("[Sync] Sent stream-start signal.")
        else:
            print(f"[Sync] Failed to send stream-start. Status: {response.status_code}")
    except Exception as e:
        print(f"[Sync] Error sending stream-start: {e}")

def get_recent_twitch_vods(channel_name):
    """Fetch recent Twitch VODs of a user using public Twitch GQL API"""
    url = "https://gql.twitch.tv/gql"
    headers = {
        "Client-Id": "kimne78kx3ncx6brgo4mv6wki5h1ko",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"
    }
    payload = {
        "query": """
        query($login: String!) {
            user(login: $login) {
                videos(first: 10, type: ARCHIVE) {
                    edges {
                        node {
                            id
                            title
                            createdAt
                            lengthSeconds
                        }
                    }
                }
            }
        }
        """,
        "variables": {
            "login": channel_name
        }
    }
    try:
        res = requests.post(url, json=payload, headers=headers, timeout=10)
        if res.status_code == 200:
            data = res.json()
            edges = data.get("data", {}).get("user", {}).get("videos", {}).get("edges", [])
            return [edge.get("node") for edge in edges if edge and edge.get("node")]
    except Exception as e:
        print(f"[VOD Resolve] Error fetching Twitch VODs: {e}")
    return []

def find_matching_vod(stream_start_time_iso, recent_vods):
    try:
        from datetime import datetime
        clean_stream = stream_start_time_iso.replace('Z', '+00:00')
        stream_dt = datetime.fromisoformat(clean_stream)
    except Exception as e:
        print(f"[Match] Error parsing stream start time: {e}")
        return None

    best_match = None
    min_diff = float('inf')

    for node in recent_vods:
        vod_id = node.get("id")
        created_at_raw = node.get("createdAt")
        if not vod_id or not created_at_raw:
            continue
        try:
            clean_vod = created_at_raw.replace('Z', '+00:00')
            vod_dt = datetime.fromisoformat(clean_vod)
        except Exception:
            continue
        
        diff = abs((stream_dt - vod_dt).total_seconds())
        # If the start times are within 2.5 hours (9000 seconds)
        if diff < 9000 and diff < min_diff:
            min_diff = diff
            best_match = vod_id

    return best_match

def resolve_missing_vods():
    """Find and resolve twitch_vod_id for pending streams that are missing it"""
    headers = {"x-api-key": API_KEY}
    try:
        # 1. Fetch missing VOD streams from backend
        res = requests.get(f"{API_URL}/api/streams/missing-vod", headers=headers, timeout=10)
        if res.status_code != 200:
            return
        
        streams = res.json().get("streams", [])
        if not streams:
            return
            
        print(f"[VOD Resolve] Found {len(streams)} pending stream(s) lacking VOD ID.")
        
        # 2. Get recent VODs from Twitch
        recent_vods = get_recent_twitch_vods(TWITCH_CHANNEL)
        if not recent_vods:
            print("[VOD Resolve] Could not retrieve recent Twitch VODs. Skipping resolve.")
            return
            
        # 3. Match each stream to a Twitch VOD
        for stream in streams:
            stream_id = stream.get("id")
            start_time_str = stream.get("start_time")
            end_time_str = stream.get("end_time")
            if not stream_id or not start_time_str:
                continue
                
            # Calculate stream age from end_time
            try:
                clean_end = end_time_str.replace('Z', '+00:00') if end_time_str else start_time_str.replace('Z', '+00:00')
                end_dt = datetime.fromisoformat(clean_end)
                now_utc = datetime.now(timezone.utc)
                age_hours = (now_utc - end_dt).total_seconds() / 3600.0
            except Exception as age_err:
                age_hours = 0

            matched_vod_id = find_matching_vod(start_time_str, recent_vods)
            if matched_vod_id:
                print(f"[VOD Resolve] Stream ID {stream_id} ({start_time_str}) matched with Twitch VOD {matched_vod_id}.")
                # Send update to server
                up_res = requests.post(f"{API_URL}/api/streams/{stream_id}/resolve-vod", 
                                       json={"twitchVodId": matched_vod_id}, 
                                       headers={"x-api-key": API_KEY, "Content-Type": "application/json"},
                                       timeout=10)
                if up_res.status_code == 200:
                    print(f"[VOD Resolve] Successfully updated VOD ID for stream {stream_id}!")
                else:
                    print(f"[VOD Resolve] Failed to update VOD ID: {up_res.status_code}")
            else:
                if age_hours > 3.0:
                    print(f"[VOD Resolve] Stream ID {stream_id} is older than 3 hours and has no matching VOD. Marking as completed to clear queue.")
                    try:
                        requests.post(f"{API_URL}/api/streams/mark-backfilled", json={"streamId": stream_id}, headers=headers, timeout=10)
                    except Exception as mark_e:
                        print(f"[VOD Resolve] Error marking stream as completed: {mark_e}")
                else:
                    print(f"[VOD Resolve] No matching VOD found for stream ID {stream_id} ({start_time_str}) within timeframe (will retry).")
    except Exception as e:
        print(f"[VOD Resolve] Exception during resolve loop: {e}")

def notify_stream_end(vod_id=None):
    """Notify backend that the stream has ended"""
    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
    url = f"{API_URL}/api/log/stream-end"
    payload = {}
    if vod_id:
        payload["twitchVodId"] = vod_id
    try:
        requests.post(url, json=payload, headers=headers, timeout=5)
        print(f"[Sync] Sent stream-end signal. VOD ID resolved: {vod_id}")
    except Exception as e:
        print(f"[Sync] Failed to send stream-end: {e}")

# Coverage window tracking
coverage_id = None
coverage_stop_flag = threading.Event()

def start_coverage(stream_id, source='live', covered_from=None, covered_to=None):
    """Register a new coverage window with the backend"""
    global coverage_id
    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
    payload = {"streamId": stream_id, "source": source}
    if covered_from:
        payload["coveredFrom"] = covered_from
    if covered_to:
        payload["coveredTo"] = covered_to
    try:
        res = requests.post(f"{API_URL}/api/log/coverage/start",
                            json=payload,
                            headers=headers, timeout=5)
        if res.status_code == 200:
            cid = res.json().get("coverageId")
            print(f"[Coverage] Started window #{cid} (source={source}) for stream {stream_id}")
            if source == 'live':
                coverage_id = cid
            return cid
    except Exception as e:
        print(f"[Coverage] Error starting coverage: {e}")
    return None

def stop_coverage(cid=None, covered_to=None):
    """Finalize a coverage window"""
    global coverage_id
    target_id = cid if cid is not None else coverage_id
    if not target_id:
        return
    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
    payload = {"coverageId": target_id}
    if covered_to:
        payload["coveredTo"] = covered_to
    try:
        requests.post(f"{API_URL}/api/log/coverage/end",
                      json=payload,
                      headers=headers, timeout=5)
        print(f"[Coverage] Ended window #{target_id}")
    except Exception as e:
        print(f"[Coverage] Error ending coverage: {e}")
    if cid is None or cid == coverage_id:
        coverage_id = None

def coverage_heartbeat_loop():
    """Background thread: ping coverage heartbeat every 2 minutes"""
    headers = {"x-api-key": API_KEY}
    while not coverage_stop_flag.is_set():
        time.sleep(120)
        if coverage_id:
            try:
                requests.patch(f"{API_URL}/api/log/coverage/{coverage_id}/heartbeat",
                               headers=headers, timeout=5)
            except Exception:
                pass

# =========================================================================
# CAPTURE METHODS
# =========================================================================

def check_stream_live():
    """Check if the Twitch channel is currently streaming using streamlink"""
    try:
        # Runs streamlink check without downloading stream
        result = subprocess.run(
            ["streamlink", f"twitch.tv/{TWITCH_CHANNEL}", "--stream-url"],
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        # If streamlink returns a stream URL, it's live!
        return result.returncode == 0 and result.stdout.strip().startswith("http")
    except FileNotFoundError:
        print("[Error] 'streamlink' utility is not installed or not in system PATH! Please install it.")
        time.sleep(10)
        return False

def run_stream_capture(model):
    """Capture Twitch stream audio using streamlink + ffmpeg and transcribe"""
    print(f"[Capture] Monitoring Twitch channel '{TWITCH_CHANNEL}'...")
    
    while not stop_flag.is_set():
        if not check_stream_live():
            # Try to resolve any missing VOD IDs once every 5 minutes (300s) to avoid log spam
            global last_resolve_time
            now_sec = time.time()
            if now_sec - last_resolve_time > 300:
                resolve_missing_vods()
                last_resolve_time = now_sec
            print(f"[Capture] Channel '{TWITCH_CHANNEL}' is offline. Checking for pending VOD backfill tasks...")
            try:
                headers_get = {"x-api-key": API_KEY}
                res = requests.get(f"{API_URL}/api/streams/pending-backfill", headers=headers_get, timeout=10)
                if res.status_code == 200:
                    data = res.json()
                    pending_streams = data.get("streams", [])
                    if pending_streams:
                        print(f"[Backfill] Found {len(pending_streams)} pending VOD backfill(s). Starting automated processing...")
                        
                        target = pending_streams[0]
                        stream_id = target.get("id")
                        vod_id = target.get("twitch_vod_id")
                        title = target.get("title", "Unknown Archive")
                        start_time = target.get("start_time")
                        
                        min_msg_time = target.get("min_message_time")
                        max_msg_time = target.get("max_message_time")
                        min_voice_time = target.get("min_voice_time")
                        max_voice_time = target.get("max_voice_time")

                        # Parse ISO strings to epoch timestamps
                        def parse_time(t_str):
                            if not t_str:
                                return None
                            try:
                                clean_t = t_str.split(".")[0].replace("Z", "").replace("+00:00", "")
                                return time.mktime(time.strptime(clean_t, "%Y-%m-%dT%H:%M:%S"))
                            except Exception as parse_err:
                                print(f"[Backfill] Error parsing time '{t_str}': {parse_err}")
                                return None

                        start_epoch = parse_time(start_time)

                        # Helper to compute ISO timestamp for given offset from start_time
                        def get_offset_iso_time(start_time_str, offset_seconds):
                            if not start_time_str or offset_seconds is None:
                                return None
                            try:
                                clean_str = start_time_str.replace('Z', '+00:00')
                                dt = datetime.fromisoformat(clean_str)
                                dt_offset = dt + timedelta(seconds=int(offset_seconds))
                                return dt_offset.strftime("%Y-%m-%dT%H:%M:%SZ")
                            except Exception as e:
                                print(f"[Coverage] Error calculating offset time: {e}")
                                return None

                        gaps = target.get("gaps", [])
                        if not gaps:
                            print(f"[Backfill] No gaps found for stream {stream_id}, marking complete.")
                            try:
                                requests.post(f"{API_URL}/api/streams/mark-backfilled",
                                              json={"streamId": stream_id}, headers=headers_get, timeout=10)
                            except Exception as e:
                                print(f"[Backfill] Error marking complete: {e}")
                            continue

                        print(f"[Backfill] Found {len(gaps)} gap(s) to fill:")
                        for g in gaps:
                            to_str = str(g['to_offset']) + 's' if g['to_offset'] is not None else 'end'
                            print(f"  Gap: {g['from_offset']}s → {to_str}")

                        # Download and upload each gap in order
                        for gap_idx, gap in enumerate(gaps, 1):
                            from_off = gap.get('from_offset', 0)
                            to_off   = gap.get('to_offset', None)
                            to_str   = str(to_off) + 's' if to_off is not None else 'end'
                            print(f"\n[Backfill] Gap {gap_idx}/{len(gaps)}: chat {from_off}s → {to_str}")

                            # 1. Register a coverage window for this gap (start)
                            covered_from_iso = get_offset_iso_time(start_time, from_off)
                            gap_coverage_id = start_coverage(stream_id, source='backfill', covered_from=covered_from_iso)

                            # 2. Download and upload chat comments
                            chat_comments = download_vod_chat(vod_id, start_offset=from_off, end_offset=to_off)
                            actual_to_offset = to_off
                            if chat_comments:
                                print(f"[Backfill] Uploading {len(chat_comments)} messages...")
                                batch_size = 100
                                headers_post = {"x-api-key": API_KEY, "Content-Type": "application/json"}
                                for i in range(0, len(chat_comments), batch_size):
                                    batch = chat_comments[i:i+batch_size]
                                    try:
                                        requests.post(f"{API_URL}/api/log/messages",
                                                      json={"messages": batch},
                                                      headers=headers_post, timeout=10)
                                    except Exception as e:
                                        print(f"[Backfill] Chat batch upload error: {e}")

                                # 2.5 Extract and upload roles
                                roles = [
                                    {
                                        "username": m["username"],
                                        "displayName": m.get("displayName", m["username"]),
                                        "isMod": m.get("isMod", False),
                                        "isSub": m.get("isSub", False),
                                        "isVip": m.get("isVip", False),
                                        "timestamp": m.get("timestamp")
                                    }
                                    for m in chat_comments
                                    if m.get("isMod") or m.get("isSub") or m.get("isVip")
                                ]
                                if roles:
                                    print(f"[Backfill] Uploading {len(roles)} roles...")
                                    for i in range(0, len(roles), 500):
                                        batch = roles[i:i+500]
                                        try:
                                            requests.post(f"{API_URL}/api/log/roles",
                                                          json={"roles": batch},
                                                          headers=headers_post, timeout=10)
                                        except Exception as e:
                                            print(f"[Backfill] Role batch upload error: {e}")

                                # If to_off was None, we can use the offset of the last downloaded message
                                if to_off is None:
                                    try:
                                        last_msg_time_str = chat_comments[-1]["timestamp"]
                                        last_msg_epoch = parse_time(last_msg_time_str)
                                        if last_msg_epoch and start_epoch:
                                            actual_to_offset = int(last_msg_epoch - start_epoch)
                                    except Exception as e:
                                        print(f"[Backfill] Error estimating end offset: {e}")
                            else:
                                print(f"[Backfill] No chat in this gap.")

                            # 3. Transcribe audio for this gap
                            print(f"[Backfill] Starting Whisper audio transcription for this gap...")
                            try:
                                start_time_iso = start_time.replace("+00:00", "").replace("Z", "") + "Z"
                                transcribe_vod_audio(vod_id, start_time_iso, model,
                                                     start_offset=from_off, end_offset=to_off)
                            except Exception as e:
                                print(f"[Backfill] Transcription error: {e}")

                            # 4. Register a coverage window for this gap (end)
                            covered_to_iso = get_offset_iso_time(start_time, actual_to_offset)
                            if not covered_to_iso:
                                covered_to_iso = get_offset_iso_time(start_time, from_off + 60)
                            stop_coverage(gap_coverage_id, covered_to=covered_to_iso)

                        # 3. Mark completed
                        print(f"[Backfill] Marking stream {stream_id} as backfilled...")
                        try:
                            res_mark = requests.post(f"{API_URL}/api/streams/mark-backfilled", json={"streamId": stream_id}, headers=headers_get, timeout=10)
                            if res_mark.status_code == 200:
                                print(f"[Backfill] Successfully marked stream {stream_id} as backfilled!")
                            else:
                                print(f"[Backfill] Failed to mark stream: {res_mark.status_code} - {res_mark.text}")
                        except Exception as e:
                            print(f"[Backfill] Network error marking stream: {e}")

                        # Loop again immediately to process next VOD or check if live
                        continue
                    else:
                        print("[Capture] No pending VOD backfills found.")
                else:
                    print(f"[Backfill] Failed to fetch pending backfills: {res.status_code}")
            except Exception as err:
                print(f"[Backfill] Error checking pending backfills: {err}")
                
            print(f"[Capture] Retrying stream check in 60 seconds...")
            time.sleep(60)
            continue
        
        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", "-"]
        
        p_streamlink = None
        p_ffmpeg = None
        
        try:
            p_streamlink = subprocess.Popen(streamlink_cmd, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
            p_ffmpeg = subprocess.Popen(ffmpeg_cmd, stdin=p_streamlink.stdout, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
            
            # Read in 10-second chunks
            # 16000Hz * 16-bit(2 bytes) * 10 seconds = 320,000 bytes
            chunk_size = 16000 * 2 * 10 
            
            while not stop_flag.is_set():
                pcm_data = p_ffmpeg.stdout.read(chunk_size)
                if not pcm_data:
                    print("[Capture] Stream audio stopped or disconnected.")
                    break
                
                # Transcribe in a background thread to prevent blocking audio read
                threading.Thread(
                    target=lambda data=pcm_data: send_voice_words(transcribe_audio_segment(model, data))
                ).start()
                
        except Exception as err:
            print(f"[Capture] Audio pipeline crashed: {err}")
        finally:
            # Clean up processes
            for p in [p_ffmpeg, p_streamlink]:
                if p:
                    try:
                        p.terminate()
                        p.wait(timeout=2)
                    except:
                        try: p.kill() 
                        except: pass
            
            # Try to get the latest VOD ID from Twitch GQL to auto-resolve it
            latest_vod_id = None
            try:
                vods = get_recent_twitch_vods(TWITCH_CHANNEL)
                if vods:
                    latest_vod_id = vods[0].get("id")
            except Exception as e:
                print(f"[Capture] Error pre-fetching latest VOD ID: {e}")
            
            notify_stream_end(latest_vod_id)
            print("[Capture] Pipeline stopped. Waiting 30 seconds before checking stream status...")
            time.sleep(30)

def run_microphone_capture(model):
    """Capture local microphone/system audio and transcribe"""
    try:
        import sounddevice as sd
    except ImportError:
        print("[Error] 'sounddevice' library is missing! Install it via pip install sounddevice.")
        sys.exit(1)
        
    print("[Capture] Starting microphone capture. Listening...")
    
    sample_rate = 16000
    duration = 10 # Transcribe every 10 seconds
    
    while not stop_flag.is_set():
        try:
            # Record float32 directly from default input device
            recording = sd.rec(
                int(duration * sample_rate), 
                samplerate=sample_rate, 
                channels=1, 
                dtype='float32'
            )
            sd.wait() # Wait until recording is finished
            
            # sounddevice records float32 directly, no need for raw conversion
            audio_np = recording.flatten()
            
            def process_mic(audio=audio_np):
                segments, info = model.transcribe(
                    audio, 
                    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:
                        print(f"[Whisper Mic] Transcribed: \"{text}\"")
                        words_list.extend(text.split())
                send_voice_words(words_list)
                
            threading.Thread(target=process_mic).start()
            
        except Exception as e:
            print(f"[Capture] Microphone error: {e}")
            time.sleep(5)

# =========================================================================
# MAIN EXECUTION
# =========================================================================

if __name__ == "__main__":
    print("=========================================================")
    print("      Twitch Analytics Local Logger & Voice Sync         ")
    print("=========================================================")
    print(f"Target Channel:  {TWITCH_CHANNEL}")
    print(f"Capture Method:  {CAPTURE_METHOD}")
    print(f"Whisper Model:   {WHISPER_MODEL_SIZE}")
    print(f"API Backend URL: {API_URL}")
    print("=========================================================")
    
    # 1. Initialize Whisper
    whisper_model = init_whisper_model()
    
    # 2. Start Twitch Chat background threads
    t_chat = threading.Thread(target=twitch_chat_listener, daemon=True)
    t_sender = threading.Thread(target=chat_sender, daemon=True)
    t_heartbeat = threading.Thread(target=coverage_heartbeat_loop, daemon=True)
    
    t_chat.start()
    t_sender.start()
    t_heartbeat.start()
    
    # 3. Start Audio Capture (Blocks main thread)
    try:
        if CAPTURE_METHOD == "stream":
            run_stream_capture(whisper_model)
        elif CAPTURE_METHOD == "microphone":
            run_microphone_capture(whisper_model)
        else:
            print(f"[Error] Unknown CAPTURE_METHOD '{CAPTURE_METHOD}'. Choose 'stream' or 'microphone'.")
    except KeyboardInterrupt:
        print("\n[Shutting Down] Gracefully stopping threads...")
    finally:
        # End coverage window and notify stream ended
        stop_coverage()
        coverage_stop_flag.set()
        
        # Try to get the latest VOD ID from Twitch GQL to auto-resolve it
        latest_vod_id = None
        if CAPTURE_METHOD == "stream":
            try:
                vods = get_recent_twitch_vods(TWITCH_CHANNEL)
                if vods:
                    latest_vod_id = vods[0].get("id")
            except Exception as e:
                print(f"[Capture] Error pre-fetching latest VOD ID: {e}")
        
        notify_stream_end(latest_vod_id)
        stop_flag.set()
        time.sleep(1)
        print("[Shutting Down] Done.")