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.")