local_worker/chat_mod_worker.py

from datetime import datetime, timedelta, timezone
import os
import sys
import time
last_resolve_time = 0
import socket
import select
import threading
import queue
import random
import requests
from dotenv import load_dotenv
from vod_backfiller import download_vod_chat

# 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()

# Local storage file to remember which stream chats we have already backfilled
PROCESSED_FILE = "processed_chat_streams.txt"

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

# Thread-safe queues
chat_queue = queue.Queue()
mod_action_queue = queue.Queue()

# Flag to signal thread termination
stop_flag = threading.Event()

# 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

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 new_iso_timestamp():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())

# =========================================================================
# TWITCH CHAT & MOD ACTION IRC LISTENER
# =========================================================================

def twitch_irc_listener():
    """Background thread to connect to Twitch IRC and read chat & mod actions"""
    server = "irc.chat.twitch.tv"
    port = 6667
    
    anon_nick = f"justinfan{random.randint(10000, 99999)}"
    
    print(f"[Twitch IRC] Connecting to chat 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))
            
            irc_sock.send(f"PASS oauth:anonymous\r\n".encode("utf-8"))
            irc_sock.send(f"NICK {anon_nick}\r\n".encode("utf-8"))
            irc_sock.send("CAP REQ :twitch.tv/tags twitch.tv/commands twitch.tv/membership\r\n".encode("utf-8"))
            irc_sock.send(f"JOIN #{TWITCH_CHANNEL}\r\n".encode("utf-8"))
            
            print(f"[Twitch IRC] Joined channel #{TWITCH_CHANNEL}. Listening for chat and moderation events...")
            
            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

                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)
                    
                    if line.startswith("PING"):
                        irc_sock.send("PONG :tmi.twitch.tv\r\n".encode("utf-8"))
                        continue
                    
                    if "PRIVMSG" in line:
                        tags = {}
                        tags_str = ""
                        
                        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)
                        
                        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
                        }
                        
                        chat_queue.put(parsed_msg)
                    
                    elif "CLEARCHAT" in line:
                        tags = {}
                        tags_str = ""
                        
                        if line.startswith("@"):
                            tags_str, remainder = line[1:].split(" ", 1)
                            tags = parse_irc_tags(tags_str)
                            line = remainder
                            
                        parts = line.split(" CLEARCHAT ")
                        if len(parts) >= 2:
                            channel_part = parts[1]
                            if " :" in channel_part:
                                _, target_user = channel_part.split(" :", 1)
                                target_user = target_user.strip()
                                
                                ban_duration = tags.get("ban-duration")
                                action_type = "timeout" if ban_duration else "ban"
                                duration = int(ban_duration) if ban_duration else None
                                
                                mod_action = {
                                    "actionType": action_type,
                                    "moderator": "TwitchIRC",
                                    "targetUser": target_user,
                                    "duration": duration,
                                    "reason": tags.get("ban-reason", "No reason provided via IRC"),
                                    "timestamp": new_iso_timestamp()
                                }
                                
                                print(f"[Twitch IRC] Detected moderation event: {action_type.upper()} for user '{target_user}'" + (f" (duration: {duration}s)" if duration else ""))
                                mod_action_queue.put(mod_action)
                                
                    elif "CLEARMSG" in line:
                        tags = {}
                        tags_str = ""
                        
                        if line.startswith("@"):
                            tags_str, remainder = line[1:].split(" ", 1)
                            tags = parse_irc_tags(tags_str)
                            line = remainder
                            
                        parts = line.split(" CLEARMSG ")
                        if len(parts) >= 2:
                            channel_part = parts[1]
                            if " :" in channel_part:
                                _, message_text = channel_part.split(" :", 1)
                                message_text = message_text.strip()
                                target_user = tags.get("login", "")
                                
                                mod_action = {
                                    "actionType": "delete",
                                    "moderator": "TwitchIRC",
                                    "targetUser": target_user,
                                    "messageText": message_text,
                                    "timestamp": new_iso_timestamp()
                                }
                                
                                print(f"[Twitch IRC] Detected moderation event: DELETE message of '{target_user}': \"{message_text}\"")
                                mod_action_queue.put(mod_action)
                                
        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

# =========================================================================
# BACKGROUND SENDERS
# =========================================================================

def chat_sender():
    """Periodically sends accumulated chat messages to backend API"""
    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
    url = f"{API_URL}/api/log/messages"
    
    while not stop_flag.is_set():
        messages = []
        while not chat_queue.empty():
            try:
                messages.append(chat_queue.get_nowait())
            except queue.Empty:
                break
        
        if messages:
            try:
                # 1. Send 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}")
                
        time.sleep(3)


def mod_action_sender():
    """Periodically sends accumulated mod actions to backend API"""
    headers = {"x-api-key": API_KEY, "Content-Type": "application/json"}
    url = f"{API_URL}/api/log/mod-action"
    
    while not stop_flag.is_set():
        actions = []
        while not mod_action_queue.empty():
            try:
                actions.append(mod_action_queue.get_nowait())
            except queue.Empty:
                break
        
        for action in actions:
            try:
                response = requests.post(url, json=action, headers=headers, timeout=5)
                if response.status_code == 200:
                    print(f"[Mod Sender] Successfully logged {action['actionType']} action to backend.")
                else:
                    print(f"[Mod Sender] Failed to sync mod action. API returned status {response.status_code}")
            except Exception as e:
                print(f"[Mod Sender] Network error sending mod action: {e}")
                
        time.sleep(1)

# =========================================================================
# VOD ID AUTO-RESOLUTION & CHAT BACKFILLING LOOP (NO AUDIO)
# =========================================================================

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 load_processed_streams():
    """Load successfully processed stream IDs from a local file"""
    if not os.path.exists(PROCESSED_FILE):
        return set()
    try:
        with open(PROCESSED_FILE, "r") as f:
            return set(line.strip() for line in f if line.strip())
    except Exception as e:
        print(f"[Backfill] Error loading processed streams file: {e}")
        return set()

def save_processed_stream(stream_id):
    """Save a successfully processed stream ID to the local file"""
    try:
        with open(PROCESSED_FILE, "a") as f:
            f.write(f"{stream_id}\n")
    except Exception as e:
        print(f"[Backfill] Error writing to processed streams file: {e}")

def run_backfill_loop():
    """Main loop to check and backfill VOD chat comments ONLY"""
    
    while not stop_flag.is_set():
        # 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"[Backfill] Checking for pending VOD backfill tasks...")
        processed_streams = load_processed_streams()
        
        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", [])
                
                # Filter out streams we have already backfilled chat for
                unprocessed_streams = [s for s in pending_streams if str(s.get("id")) not in processed_streams]
                
                if unprocessed_streams:
                    print(f"[Backfill] Found {len(unprocessed_streams)} pending VOD chat backfill(s). Starting automated processing...")
                    
                    target = unprocessed_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")
                    
                    gaps = target.get("gaps", [])
                    if not gaps:
                        print(f"[Backfill] No gaps found for stream {stream_id}, skipping.")
                        save_processed_stream(stream_id)
                        continue

                    print(f"\n=========================================================")
                    print(f"[Backfill] Processing Chat Gaps: {title}")
                    print(f"[Backfill] VOD ID:                {vod_id}")
                    print(f"[Backfill] Stream ID:            {stream_id}")
                    print(f"[Backfill] Gaps count:            {len(gaps)}")
                    print(f"=========================================================")
                    
                    # 1. Download and upload chat segments for each gap
                    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}")

                        chat_comments = download_vod_chat(vod_id, start_offset=from_off, end_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}")

                            # 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}")
                        else:
                            print(f"[Backfill] No chat in this gap.")
                    
                    # 2. Mark locally as processed (does NOT mark completed on server, so worker.py can do voice)
                    save_processed_stream(stream_id)
                    print(f"[Backfill] Successfully backfilled chat for stream {stream_id}. Recorded to local processed file.")
                    
                    # Loop again immediately
                    continue
                else:
                    print("[Backfill] No pending VOD chat 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"[Backfill] Retrying check in 60 seconds...")
        time.sleep(60)

# =========================================================================
# MAIN ENTRYPOINT
# =========================================================================

if __name__ == "__main__":
    print("=========================================================")
    print("  Twitch Chat & Moderation Actions Logger & Chat-Sync    ")
    print("=========================================================")
    print(f"Target Channel:  {TWITCH_CHANNEL}")
    print(f"API Backend URL: {API_URL}")
    print("=========================================================")
    
    # Start Twitch IRC background threads (runs 24/7 to catch chat/mod actions)
    t_irc = threading.Thread(target=twitch_irc_listener, daemon=True)
    t_chat_send = threading.Thread(target=chat_sender, daemon=True)
    t_mod_send = threading.Thread(target=mod_action_sender, daemon=True)
    t_heartbeat = threading.Thread(target=coverage_heartbeat_loop, daemon=True)
    
    t_irc.start()
    t_chat_send.start()
    t_mod_send.start()
    t_heartbeat.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 VOD backfilling (GQL chat ONLY) in the main thread
    try:
        run_backfill_loop()
    except KeyboardInterrupt:
        print("\n[Shutting Down] Gracefully stopping threads...")
    finally:
        stop_coverage()
        coverage_stop_flag.set()
        stop_flag.set()
        time.sleep(1)
        print("[Shutting Down] Done.")