import re import uuid import httpx import asyncio import base64 import subprocess import os import sys from typing import Optional, Dict, List, Any from datetime import datetime from sqlalchemy.orm import Session class TwitchService: def __init__(self): self.client_id = "kimne78kx3ncx6brgo4mv6wki5h1ko" # Twitch web client ID self.oauth_client_id = "7nfx52bjgr2kf5ltg6lu7dyu979pci" self.oauth_client_secret = "myoxudv1apj1uzgsbd4fsqiott65ir" self.cache_dir = "/app/twitch_cache" self.oauth_token = None import os os.makedirs(self.cache_dir, exist_ok=True) def extract_vod_id(self, url: str) -> Optional[str]: """Extract VOD ID from Twitch URL""" patterns = [ r'twitch\.tv/videos/(\d+)', r'twitch\.tv/(\w+)/video/(\d+)', r'clips\.twitch\.tv/([a-zA-Z0-9_-]+)' ] for pattern in patterns: match = re.search(pattern, url) if match: return match.group(1) if len(match.groups()) == 1 else match.group(2) if re.match(r'^\d+$', url): return url return None async def download_and_encode_image(self, url: str) -> str: """Download image and encode as base64""" try: async with httpx.AsyncClient() as client: response = await client.get(url) if response.status_code != 200: sys.stderr.write(f"[IMAGE ERROR] Failed to download image: {response.status_code}\n") sys.stderr.flush() return "" # Encode as base64 image_data = base64.b64encode(response.content).decode('utf-8') # Get content type content_type = response.headers.get("content-type", "image/png") return f"data:{content_type};base64,{image_data}" except Exception as e: sys.stderr.write(f"[IMAGE ERROR] Error downloading image: {e}\n") sys.stderr.flush() return "" async def get_oauth_token(self) -> str: """Generate OAuth token for Twitch Helix API""" if self.oauth_token: return self.oauth_token try: async with httpx.AsyncClient() as client: response = await client.post( "https://id.twitch.tv/oauth2/token", data={ "client_id": self.oauth_client_id, "client_secret": self.oauth_client_secret, "grant_type": "client_credentials" } ) if response.status_code != 200: sys.stderr.write(f"[OAUTH ERROR] Failed to generate OAuth token: {response.status_code}\n") sys.stderr.flush() return "" data = response.json() self.oauth_token = data.get("access_token", "") sys.stderr.write(f"[OAUTH] Generated OAuth token: {self.oauth_token[:10]}...\n") sys.stderr.flush() return self.oauth_token except Exception as e: sys.stderr.write(f"[OAUTH ERROR] Error generating OAuth token: {e}\n") sys.stderr.flush() return "" async def get_channel_badges(self, broadcaster_id: str) -> Dict[str, str]: """Fetch channel badges from Twitch Helix API and return mapping of set_id:version to badge_id""" try: oauth_token = await self.get_oauth_token() if not oauth_token: sys.stderr.write("[OAUTH ERROR] No OAuth token available\n") sys.stderr.flush() return {} async with httpx.AsyncClient() as client: response = await client.get( f"https://api.twitch.tv/helix/chat/badges", params={"broadcaster_id": broadcaster_id}, headers={ "Client-ID": self.oauth_client_id, "Authorization": f"Bearer {oauth_token}" } ) sys.stderr.write(f"[BADGES] Channel badges response status: {response.status_code}\n") sys.stderr.flush() if response.status_code != 200: sys.stderr.write(f"[BADGES ERROR] Failed to fetch channel badges: {response.status_code}\n") sys.stderr.write(f"[BADGES ERROR] Response body: {response.text}\n") sys.stderr.flush() return {} data = response.json() badges_data = data.get('data', []) sys.stderr.write(f"[BADGES] Channel badges data count: {len(badges_data)}\n") sys.stderr.flush() # Create mapping of set_id:version to image URL badge_mapping = {} for badge_set in badges_data: set_id = badge_set.get('set_id') versions = badge_set.get('versions', []) for version in versions: version_id = version.get('id') image_url = version.get('image_url_1x', '') # Store the direct image URL if image_url: key = f"{set_id}:{version_id}" badge_mapping[key] = image_url return badge_mapping except Exception as e: sys.stderr.write(f"[BADGES ERROR] Error fetching channel badges: {e}\n") sys.stderr.flush() return {} async def get_global_badges(self) -> Dict[str, str]: """Fetch global badges from Twitch Helix API and return mapping of set_id:version to badge_id""" try: oauth_token = await self.get_oauth_token() if not oauth_token: sys.stderr.write("[OAUTH ERROR] No OAuth token available\n") sys.stderr.flush() return {} async with httpx.AsyncClient() as client: response = await client.get( f"https://api.twitch.tv/helix/chat/badges/global", headers={ "Client-ID": self.oauth_client_id, "Authorization": f"Bearer {oauth_token}" } ) sys.stderr.write(f"[BADGES] Global badges response status: {response.status_code}\n") sys.stderr.flush() if response.status_code != 200: sys.stderr.write(f"[BADGES ERROR] Failed to fetch global badges: {response.status_code}\n") sys.stderr.write(f"[BADGES ERROR] Response body: {response.text}\n") sys.stderr.flush() return {} data = response.json() badges_data = data.get('data', []) sys.stderr.write(f"[BADGES] Global badges data count: {len(badges_data)}\n") sys.stderr.flush() # Create mapping of set_id:version to image URL badge_mapping = {} for badge_set in badges_data: set_id = badge_set.get('set_id') versions = badge_set.get('versions', []) for version in versions: version_id = version.get('id') image_url = version.get('image_url_1x', '') # Store the direct image URL if image_url: key = f"{set_id}:{version_id}" badge_mapping[key] = image_url return badge_mapping except Exception as e: sys.stderr.write(f"[BADGES ERROR] Error fetching global badges: {e}\n") sys.stderr.flush() return {} async def get_stream_info(self, url: str) -> Dict[str, Any]: """Get stream information using Twitch Helix API""" vod_id = self.extract_vod_id(url) if not vod_id: raise ValueError("Invalid Twitch URL or ID") try: async with httpx.AsyncClient() as client: # Get video info from GraphQL API (more reliable for public data) query = ''' { video(id: "%s") { title description createdAt lengthSeconds viewCount thumbnailURLs(width: 640, height: 360) owner { displayName } game { displayName boxArtURL(width: 100, height: 140) } } } ''' % vod_id response = await client.post( "https://gql.twitch.tv/gql", json={"query": query}, headers={"Client-ID": self.client_id} ) if response.status_code != 200: raise RuntimeError(f"Twitch API request failed: {response.status_code}") data = response.json() video_data = data.get('data', {}).get('video') if not video_data: raise RuntimeError("Video not found") return { 'title': video_data.get('title', ''), 'description': video_data.get('description', ''), 'created_at': video_data.get('createdAt'), 'duration_seconds': video_data.get('lengthSeconds', 0), 'view_count': 0, # Always start with 0 views 'thumbnail_url': video_data.get('thumbnailURLs', [''])[0] if video_data.get('thumbnailURLs') else None, 'streamer': video_data.get('owner', {}).get('displayName', ''), 'game': video_data.get('game', {}).get('displayName', ''), 'game_box_art': video_data.get('game', {}).get('boxArtURL', '') } except Exception as e: raise RuntimeError(f"Failed to get stream info: {str(e)}") async def get_video_chapters(self, vod_id: str) -> dict: """Get video chapters using TwitchDownloader's persistedQuery""" try: async with httpx.AsyncClient() as client: # TwitchDownloader's persistedQuery for VideoPlayer_ChapterSelectButtonVideo request_body = { "extensions": { "persistedQuery": { "sha256Hash": "71835d5ef425e154bf282453a926d99b328cdc5e32f36d3a209d0f4778b41203", "version": 1 } }, "operationName": "VideoPlayer_ChapterSelectButtonVideo", "variables": { "videoID": vod_id } } response = await client.post( "https://gql.twitch.tv/gql", json=request_body, headers={"Client-ID": "kimne78kx3ncx6brgo4mv6wki5h1ko"} ) if response.status_code != 200: raise RuntimeError(f"Twitch chapters API request failed: {response.status_code}") data = response.json() return data except Exception as e: raise RuntimeError(f"Failed to get video chapters: {str(e)}") def save_chapters_to_db(self, stream_id: str, chapters: list, db: Session): """Save chapters to the dedicated chapter table""" from ..models.chapter import Chapter # Delete existing chapters for this stream db.query(Chapter).filter(Chapter.stream_id == stream_id).delete() # Insert new chapters for chapter in chapters: db_chapter = Chapter( stream_id=stream_id, game_display_name=chapter.get('gameDisplayName', ''), game_id=chapter.get('gameId', ''), description=chapter.get('description', ''), start_milliseconds=chapter.get('startMilliseconds', 0), length_milliseconds=chapter.get('lengthMilliseconds', 0), thumbnail_url=chapter.get('thumbnailUrl', '') ) db.add(db_chapter) db.commit() sys.stderr.write(f"[CHAPTERS] Saved {len(chapters)} chapters to database for stream {stream_id}\n") sys.stderr.flush() async def download_chat(self, url: str, embed_emotes: bool = True, include_metadata: bool = False) -> Dict[str, Any]: """Download chat using Twitch GraphQL API with TwitchDownloader's approach Returns dict with 'messages' list and optional 'metadata' dict""" vod_id = self.extract_vod_id(url) if not vod_id: raise ValueError("Invalid Twitch URL or ID") # Use TwitchDownloader's client ID client_id = "kd1unb4b3q4t58fwlpcbzcbnm76a8fp" try: async with httpx.AsyncClient() as client: converted_messages = [] cursor = "" latest_offset = 0 video_start = 0 video_end = 999999 # Download entire video run_to_end = True error_count = 0 null_count = 0 broadcaster_id = None badge_mapping = {} global_badge_mapping = await self.get_global_badges() while run_to_end or latest_offset < video_end: # Build request using TwitchDownloader's approach if not cursor: # First request with offset payload = { "operationName": "VideoCommentsByOffsetOrCursor", "variables": { "videoID": vod_id, "contentOffsetSeconds": video_start }, "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a" } } } else: # Subsequent requests with cursor payload = { "operationName": "VideoCommentsByOffsetOrCursor", "variables": { "videoID": vod_id, "cursor": cursor }, "extensions": { "persistedQuery": { "version": 1, "sha256Hash": "b70a3591ff0f4e0313d126c6a1502d79a1c02baebb288227c582044aa76adf6a" } } } try: response = await client.post( "https://gql.twitch.tv/gql", json=payload, headers={"Client-ID": client_id} ) if response.status_code != 200: if error_count > 10: raise RuntimeError(f"Twitch API request failed: {response.status_code}") error_count += 1 continue data = response.json() video_data = data.get('data', {}).get('video', {}) if not video_data: raise RuntimeError("Video not found") # Extract broadcaster_id and fetch badges on first request if not broadcaster_id: broadcaster_id = video_data.get('owner', {}).get('id') sys.stderr.write(f"[CHAT] Extracted broadcaster_id: {broadcaster_id}\n") sys.stderr.flush() if broadcaster_id: sys.stderr.write(f"[CHAT] Fetching channel badges for broadcaster_id: {broadcaster_id}\n") sys.stderr.flush() channel_badges = await self.get_channel_badges(broadcaster_id) sys.stderr.write(f"[CHAT] Channel badges fetched: {len(channel_badges)} badges\n") sys.stderr.flush() # Merge global badges with channel badges (channel badges take precedence) badge_mapping = {**global_badge_mapping, **channel_badges} sys.stderr.write(f"[CHAT] Total badge mapping: {len(badge_mapping)} badges\n") sys.stderr.flush() else: sys.stderr.write("[CHAT] No broadcaster_id found, using only global badges\n") sys.stderr.flush() badge_mapping = global_badge_mapping comments_data = video_data.get('comments') # Handle null comments (Twitch sometimes returns null) if comments_data is None or comments_data.get('edges') is None: if null_count > 10: sys.stderr.write("[CHAT ERROR] Received too many null comment lists\n") sys.stderr.flush() break null_count += 1 continue null_count = max(0, null_count - 0.1) error_count = max(0, error_count - 0.1) comments_edges = comments_data.get('edges', []) page_info = comments_data.get('pageInfo', {}) # Process messages for edge in comments_edges: node = edge.get('node', {}) # Extract message fragments = node.get('message', {}).get('fragments', []) message_text = '' emotes = [] for fragment in fragments: if fragment.get('emote'): emote = fragment['emote'] emote_id = emote.get('emoteID', '') emotes.append({ 'id': emote_id, 'name': emote_id, 'url': f'https://static-cdn.jtvnw.net/emoticons/v2/{emote_id}/default/dark/1.0' }) if fragment.get('text'): message_text += fragment['text'] else: if fragment.get('text'): message_text += fragment['text'] # Get badges badges_data = node.get('message', {}).get('userBadges', []) badges = [] for badge in badges_data: if badge.get('setID'): set_id = badge.get('setID') version = badge.get('version') # Use badge_mapping to get image URL directly key = f"{set_id}:{version}" badge_url = badge_mapping.get(key, '') # Fallback to known global badge URLs if badge_mapping is empty if not badge_url: known_badge_urls = { 'broadcaster:1': 'https://static-cdn.jtvnw.net/badges/v1/5527c58c-fb7d-422d-b71b-f309dcb85cc1/1', 'moderator:1': 'https://static-cdn.jtvnw.net/badges/v1/3267646d-33f0-4b17-b3df-f923a41db1d0/1', 'vip:1': 'https://static-cdn.jtvnw.net/badges/v1/b817aba4-fad8-49e2-b88a-7cc744dfa6ec/1', 'subscriber:0': 'https://static-cdn.jtvnw.net/badges/v1/5d9f2208-5dd8-11e7-8513-2ff4adfae661/1', 'founder:1': 'https://static-cdn.jtvnw.net/badges/v1/3267646d-33f0-4b17-b3df-f923a41db1d0/1', 'premium:1': 'https://static-cdn.jtvnw.net/badges/v1/9r7hx2k0608i8x6s8u8g8/1', 'partner:1': 'https://static-cdn.jtvnw.net/badges/v1/d12a2e27-16f6-41d0-ab77-b780518f00a3/1', } badge_url = known_badge_urls.get(key, '') # If still no URL, try without version if not badge_url: base_key = f"{set_id}:1" badge_url = known_badge_urls.get(base_key, '') if badge_url: badges.append({ 'set_id': set_id, 'version': version, 'url': badge_url }) offset = node.get('contentOffsetSeconds', 0) converted_messages.append({ 'id': str(uuid.uuid4()), 'username': node.get('commenter', {}).get('displayName', 'Unknown'), 'message': message_text, 'timestamp_offset': offset, 'color': node.get('message', {}).get('userColor', '#9146FF'), 'emotes': emotes, 'badges': badges }) if offset > latest_offset: latest_offset = offset # Check if there are more pages if not page_info.get('hasNextPage', False): break cursor = comments_edges[-1].get('cursor', '') if comments_edges else "" if not cursor: break except Exception as e: if error_count > 10: raise RuntimeError(f"Failed to download chat: {str(e)}") error_count += 1 continue # Extract metadata if requested metadata = None if include_metadata: try: # Get video info for game/chapters sys.stderr.write(f"[METADATA] Extracting metadata for URL: {url}\n") sys.stderr.flush() stream_info = await self.get_stream_info(url) sys.stderr.write(f"[METADATA] Stream info game: {stream_info.get('game')}\n") sys.stderr.flush() # Get chapters using TwitchDownloader's persistedQuery chapters_data = await self.get_video_chapters(vod_id) sys.stderr.write(f"[METADATA] Chapters data keys: {chapters_data.keys()}\n") sys.stderr.flush() # Extract chapters from response chapters = [] moments_edges = chapters_data.get('data', {}).get('video', {}).get('moments', {}).get('edges', []) sys.stderr.write(f"[METADATA] Found {len(moments_edges)} chapter moments\n") sys.stderr.flush() for moment_edge in moments_edges: moment = moment_edge.get('node', {}) chapters.append({ 'id': moment.get('id', ''), 'startMilliseconds': moment.get('positionMilliseconds', 0), 'lengthMilliseconds': moment.get('durationMilliseconds', 0), 'type': moment.get('_type', 'GAME_CHANGE'), 'description': moment.get('details', {}).get('game', {}).get('displayName', ''), 'subDescription': '', 'thumbnailUrl': moment.get('thumbnailURL', ''), 'gameId': moment.get('details', {}).get('game', {}).get('id', ''), 'gameDisplayName': moment.get('details', {}).get('game', {}).get('displayName', ''), 'gameBoxArtUrl': moment.get('details', {}).get('game', {}).get('boxArtURL', '') }) sys.stderr.write(f"[METADATA] Extracted {len(chapters)} chapters\n") sys.stderr.flush() # If no chapters found, generate a bogus chapter with game info (like TwitchDownloader) if len(chapters) == 0: sys.stderr.write("[METADATA] No chapters found, generating bogus chapter with game info\n") sys.stderr.flush() chapters.append({ 'id': 'bogus_chapter_0', 'startMilliseconds': 0, 'lengthMilliseconds': stream_info.get('duration_seconds', 0) * 1000, 'type': 'GAME_CHANGE', 'description': stream_info.get('game', ''), 'subDescription': '', 'thumbnailUrl': '', 'gameId': '', 'gameDisplayName': stream_info.get('game', ''), 'gameBoxArtUrl': stream_info.get('game_box_art', '') }) sys.stderr.write(f"[METADATA] Generated bogus chapter: {chapters[0]}\n") sys.stderr.flush() metadata = { 'game': stream_info.get('game', ''), 'game_box_art': stream_info.get('game_box_art', ''), 'chapters': chapters } except Exception as e: sys.stderr.write(f"[METADATA ERROR] Failed to extract metadata: {e}\n") sys.stderr.flush() return { 'messages': converted_messages, 'metadata': metadata } except Exception as e: raise RuntimeError(f"Failed to download chat: {str(e)}") async def download_video(self, url: str, output_path: str, stream_id: str = None) -> Dict[str, Any]: """Download video using yt-dlp with progress tracking""" vod_id = self.extract_vod_id(url) if not vod_id: raise ValueError("Invalid Twitch URL or ID") try: # Ensure output directory exists os.makedirs(os.path.dirname(output_path), exist_ok=True) # Progress file for tracking progress_key = f"stream_progress:{stream_id or vod_id}" # Create progress hook script progress_script = f""" import sys import json import logging import sys logger = logging.getLogger(__name__) import redis r = redis.Redis(host='redis', port=6379, db=0, decode_responses=True) def progress_hook(d): if d['status'] == 'downloading': try: downloaded = d.get('downloaded_bytes', 0) total = d.get('total_bytes') or d.get('total_bytes_estimate', 0) speed = d.get('speed', 0) eta = d.get('eta', 0) progress_data = {{ 'status': 'downloading', 'downloaded_bytes': downloaded, 'total_bytes': total, 'percentage': (downloaded / total * 100) if total > 0 else 0, 'speed': speed, 'eta': eta }} r.setex('{progress_key}', 3600, json.dumps(progress_data)) except Exception as e: pass elif d['status'] == 'finished': progress_data = {{ 'status': 'finished', 'percentage': 100 }} r.setex('{progress_key}', 3600, json.dumps(progress_data)) # Hook into yt-dlp """ # Write progress hook to temp file import tempfile hook_file = tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) hook_file.write(progress_script) hook_file.close() sys.stderr.write(f"[DOWNLOAD] Starting VOD download: {url}\n") sys.stderr.write(f"[DOWNLOAD] Output path: {output_path}\n") sys.stderr.flush() # Use yt-dlp to download as MP4 with fragments # We'll rename fragments to .ts and create m3u8 playlist after download output_dir = os.path.dirname(output_path) output_name = os.path.splitext(os.path.basename(output_path))[0] cmd = [ 'yt-dlp', '-f', 'best', '-o', f'{output_dir}/{output_name}.mp4', '--no-playlist', '--newline', # Print progress on new lines '--progress', '--keep-fragments', # Keep downloaded fragments '--concurrent-fragments', '8', # Parallel download of fragments url ] # Run yt-dlp and capture output for progress process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1 ) # Parse progress from output import redis import json from ..config import settings r = redis.Redis(host='redis', port=6379, db=0, password=settings.REDIS_URL.split(':')[2].split('@')[0] if '@' in settings.REDIS_URL else None, decode_responses=True) last_logged_percentage = 0 for line in process.stdout: # Parse yt-dlp progress output if '[download]' in line and '%' in line: try: # Example: [download] 45.2% of 123.45MiB at 1.23MiB/s ETA 00:30 parts = line.split() percentage_str = next((p for p in parts if '%' in p), None) if percentage_str: percentage = float(percentage_str.replace('%', '')) # Terminate yt-dlp when download reaches 100% to skip FixupM3u8 if percentage >= 100: sys.stderr.write(f"[DOWNLOAD] Download complete at 100%, terminating yt-dlp to skip FixupM3u8\n") sys.stderr.flush() process.terminate() break # Extract downloaded size, total size, speed, and ETA downloaded_size = '' total_size = '' speed_info = '' eta_info = '' for i, part in enumerate(parts): # Look for pattern: "X.XXMiB of Y.YYMiB" or "X.XXMiB of ~ Y.YYMiB" if 'of' in part: if i > 0: downloaded_size = parts[i - 1] if i + 1 < len(parts): total_size = parts[i + 1] elif 'at' in part and i + 1 < len(parts): speed_info = parts[i + 1] elif 'ETA' in part and i + 1 < len(parts): eta_info = parts[i + 1] progress_data = { 'status': 'downloading', 'percentage': percentage, 'downloaded_size': downloaded_size, 'total_size': total_size, 'speed': speed_info, 'eta': eta_info } progress_key = f"stream_progress:{stream_id}" r.setex(progress_key, 3600, json.dumps(progress_data)) # Only log when reaching next 10% milestone current_milestone = int(percentage / 10) * 10 if current_milestone > last_logged_percentage or percentage >= 99: last_logged_percentage = current_milestone sys.stderr.write(f"[DOWNLOAD] {current_milestone}% ({downloaded_size} / {total_size}) - {speed_info} - ETA {eta_info}\n") sys.stderr.flush() except Exception as e: sys.stderr.write(f"[ERROR] Error parsing progress: {e}\n") sys.stderr.flush() # Wait for process to complete (2 hour timeout for large VODs) return_code = process.wait(timeout=7200) # Clean up hook file try: os.unlink(hook_file.name) except: pass # return_code -15 (SIGTERM) is expected when we terminate at 100% if return_code != 0 and return_code != -15: raise RuntimeError(f"yt-dlp failed with code {return_code}") if return_code == -15: sys.stderr.write(f"[DOWNLOAD] yt-dlp terminated at 100% (expected), continuing with HLS conversion\n") sys.stderr.flush() # Find all fragment files, rename to .ts, and create m3u8 playlist import glob import re from concurrent.futures import ThreadPoolExecutor fragment_pattern = f"{output_dir}/{output_name}.mp4.part-Frag*" fragment_files = glob.glob(fragment_pattern) # Sort numerically by fragment number def extract_fragment_number(filename): match = re.search(r'Frag(\d+)', filename) return int(match.group(1)) if match else 0 fragment_files.sort(key=extract_fragment_number) if fragment_files: # Rename fragments to .ts in parallel sys.stderr.write(f"[RENAME] Starting parallel rename of {len(fragment_files)} fragments\n") sys.stderr.flush() # Rename fragment files to standard HLS format def rename_fragment(fragment_file): # Extract fragment number match = re.search(r'Frag(\d+)', fragment_file) if match: frag_num = match.group(1) ts_file = fragment_file.replace('.mp4.part-Frag' + frag_num, f'-{frag_num}.ts') else: ts_file = fragment_file.replace('.part-Frag', '-').replace('.mp4', '.ts') os.rename(fragment_file, ts_file) return ts_file # Generate optimized m3u8 with exact timestamps def generate_optimized_m3u8(hls_dir, stream_id, progress_key=None): try: # Find all segment files segment_files = [f for f in os.listdir(hls_dir) if f.endswith('.ts') and stream_id in f and '-' in f] # Sort by segment number def extract_segment_number(filename): match = re.search(r'-(\d+)\.ts$', filename) return int(match.group(1)) if match else 0 segment_files.sort(key=extract_segment_number) total_segments = len(segment_files) # Extract timestamps for each segment using ffprobe segment_timestamps = [] for idx, seg_file in enumerate(segment_files): seg_num = extract_segment_number(seg_file) seg_path = os.path.join(hls_dir, seg_file) # Update progress for m3u8 optimization if progress_key and total_segments > 0: optimize_progress = 50 + (idx / total_segments * 10) # 50-60% eta_seconds = int((total_segments - idx) * 0.1) # Estimate 0.1s per segment progress_data = { 'status': 'processing', 'percentage': optimize_progress, 'step': 'generating_optimized_m3u8', 'step_name': 'Generating optimized m3u8', 'details': { 'current_segment': idx + 1, 'total_segments': total_segments, 'eta': f"{eta_seconds}s" } } r.setex(progress_key, 3600, json.dumps(progress_data)) result = subprocess.run( ['ffprobe', '-v', 'error', '-show_entries', 'format=start_time', '-of', 'default=noprint_wrappers=1:nokey=1', seg_path], capture_output=True, text=True ) if result.returncode == 0: start_time = float(result.stdout.strip()) segment_timestamps.append((seg_num, start_time)) if not segment_timestamps: sys.stderr.write("[TWITCH] No segment timestamps found, skipping optimized m3u8 generation\n") return # Read original m3u8 m3u8_file = os.path.join(hls_dir, f'{stream_id}.m3u8') with open(m3u8_file, 'r') as f: lines = f.readlines() # Create new m3u8 with #EXT-X-START-TIME tags new_m3u8 = [] seg_index = 0 for line in lines: if line.startswith('#EXTINF'): # Get current segment timestamp if seg_index < len(segment_timestamps): seg_num, start_time = segment_timestamps[seg_index] # Add #EXT-X-START-TIME tag before #EXTINF new_m3u8.append(f'#EXT-X-START-TIME:{start_time}\n') seg_index += 1 new_m3u8.append(line) # Write optimized m3u8 optimized_m3u8 = os.path.join(hls_dir, f'{stream_id}_optimized.m3u8') with open(optimized_m3u8, 'w') as f: f.writelines(new_m3u8) sys.stderr.write(f"[TWITCH] Optimized m3u8 created with {seg_index} #EXT-X-START-TIME tags\n") except Exception as e: sys.stderr.write(f"[TWITCH] Error generating optimized m3u8: {e}\n") with ThreadPoolExecutor(max_workers=8) as executor: ts_files = list(executor.map(rename_fragment, fragment_files)) # Already sorted, no need to sort again sys.stderr.write(f"[RENAME] Completed parallel rename of {len(ts_files)} fragments\n") sys.stderr.flush() # Create basic m3u8 playlist from fragments for 10-second segments m3u8_file = f"{output_dir}/{output_name}.m3u8" with open(m3u8_file, 'w') as f: f.write('#EXTM3U\n') f.write('#EXT-X-VERSION:3\n') f.write('#EXT-X-TARGETDURATION:10\n') f.write('#EXT-X-MEDIA-SEQUENCE:0\n') for ts_file in ts_files: f.write(f'#EXTINF:10.0,\n') f.write(f'{os.path.basename(ts_file)}\n') f.write('#EXT-X-ENDLIST\n') sys.stderr.write(f"[PLAYLIST] Created basic m3u8 playlist with {len(ts_files)} segments\n") # Save original playlist as backup before resegmentation original_playlist = f"{output_dir}/{output_name}.m3u8" backup_playlist = f"{output_dir}/{output_name}_10s.m3u8" if os.path.exists(original_playlist): import shutil shutil.copy(original_playlist, backup_playlist) sys.stderr.write(f"[PLAYLIST] Saved original playlist as backup: {backup_playlist}\n") # Update progress: Download complete, HLS available with 10-second segments # Update stream status to 'segmenting' to make it playable from app.database import get_db from app.models.stream import Stream as StreamModel from sqlalchemy.orm import Session try: db = next(get_db()) stream = db.query(StreamModel).filter(StreamModel.id == stream_id).first() if stream: stream.status = 'segmenting' stream.hls_path = f'/uploads/videos/{stream_id}/{stream_id}_10s.m3u8' stream.hls_url = f'/uploads/videos/{stream_id}/{stream_id}_10s.m3u8' stream.hls_available = True db.commit() sys.stderr.write(f"[STATUS] Stream {stream_id} status set to 'segmenting', using 10s playlist\n") except Exception as e: sys.stderr.write(f"[STATUS ERROR] Failed to update stream status: {e}\n") progress_data = { 'status': 'processing', 'percentage': 30, 'step': 'concatenating_segments', 'step_name': 'Concatenating segments' } r.setex(progress_key, 3600, json.dumps(progress_data)) # Re-segment video with FFmpeg to create actual 2-second segments sys.stderr.write(f"[FFMPEG] Re-segmenting video to 2-second segments\n") sys.stderr.flush() # First, concatenate all TS files into a single file concat_list_path = f"{output_dir}/concat_list.txt" with open(concat_list_path, 'w') as f: for ts_file in ts_files: f.write(f"file '{ts_file}'\n") # Concatenate segments temp_video = f"{output_dir}/{output_name}_temp.ts" total_files = len(ts_files) concat_cmd = [ 'ffmpeg', '-f', 'concat', '-safe', '0', '-i', concat_list_path, '-c', 'copy', temp_video, '-y' ] concat_process = subprocess.Popen( concat_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) # Parse FFmpeg progress for concatenation for line in concat_process.stdout: sys.stderr.write(f"[FFMPEG CONCAT] {line}") sys.stderr.flush() # Update progress periodically during concatenation if 'time=' in line: import re time_match = re.search(r'time=(\d{2}):(\d{2}):(\d{2}\.\d{2})', line) if time_match: hours = int(time_match.group(1)) minutes = int(time_match.group(2)) seconds = float(time_match.group(3)) total_seconds = hours * 3600 + minutes * 60 + seconds # Estimate total duration (17 minutes = 1020 seconds) duration = 1020 if duration > 0: progress_ratio = min(1.0, total_seconds / duration) files_processed = int(progress_ratio * total_files) remaining_seconds = duration - total_seconds eta_minutes = int(remaining_seconds / 60) if remaining_seconds > 0 else 0 eta_seconds = int(remaining_seconds % 60) if remaining_seconds > 0 else 0 progress_data = { 'status': 'processing', 'percentage': 15 + (progress_ratio * 5), # 15-20% 'step': 'concatenating_segments', 'step_name': 'Concatenating segments', 'details': { 'files_processed': files_processed, 'total_files': total_files, 'current_operation': 'Merging video segments', 'eta': f"{eta_minutes:02d}:{eta_seconds:02d}" } } r.setex(progress_key, 3600, json.dumps(progress_data)) concat_process.wait() if concat_process.returncode != 0: raise RuntimeError(f"FFmpeg concat failed with code {concat_process.returncode}") # Update progress: Concat complete, starting re-segmentation (40%) # Update stream status to 'resegmenting' try: db = next(get_db()) stream = db.query(StreamModel).filter(StreamModel.id == stream_id).first() if stream: stream.status = 'resegmenting' # Continue using the 10s playlist during resegmentation stream.hls_path = f'/uploads/videos/{stream_id}/{stream_id}_10s.m3u8' stream.hls_url = f'/uploads/videos/{stream_id}/{stream_id}_10s.m3u8' stream.hls_available = True db.commit() sys.stderr.write(f"[STATUS] Stream {stream_id} status set to 'resegmenting', still using 10s playlist\n") # Store actual duration for progress calculation actual_duration = stream.duration_seconds except Exception as e: sys.stderr.write(f"[STATUS ERROR] Failed to update stream status: {e}\n") actual_duration = 1020 # Fallback progress_data = { 'status': 'processing', 'percentage': 20, 'step': 'resegmenting', 'step_name': 'Re-segmenting to 2-second segments' } r.setex(progress_key, 3600, json.dumps(progress_data)) # Now re-segment to 2-second segments hls_dir = output_dir hls_path = f"{hls_dir}/{output_name}.m3u8" resegment_cmd = [ 'ffmpeg', '-i', temp_video, '-c', 'copy', '-f', 'hls', '-hls_time', '2', '-hls_list_size', '0', '-hls_segment_filename', f'{hls_dir}/{output_name}_%03d.ts', '-progress', 'pipe:1', '-y', hls_path ] resegment_process = subprocess.Popen( resegment_cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True ) # Parse FFmpeg progress for re-segmentation (40-70%) import re current_frame = 0 current_fps = 0 current_speed = 0 total_seconds = 0 for line in resegment_process.stdout: sys.stderr.write(f"[FFMPEG RESEGMENT] {line}") sys.stderr.flush() # Parse frame count from any line frame_match = re.search(r'frame=(\d+)', line) if frame_match: current_frame = int(frame_match.group(1)) # Parse fps from any line fps_match = re.search(r'fps=([\d.]+)', line) if fps_match: current_fps = float(fps_match.group(1)) # Parse speed from any line speed_match = re.search(r'speed=([\d.]+)x', line) if speed_match: current_speed = float(speed_match.group(1)) # Parse progress with detailed metrics when time is available if 'out_time=' in line: match = re.search(r'out_time=(\d{2}):(\d{2}):(\d{2}\.\d{2})', line) if match: hours = int(match.group(1)) minutes = int(match.group(2)) seconds = float(match.group(3)) total_seconds = hours * 3600 + minutes * 60 + seconds # Use actual duration from stream model duration = actual_duration if actual_duration > 0 else 1020 if duration > 0: resegment_progress = min(50, 20 + (total_seconds / duration * 30)) remaining_seconds = duration - total_seconds eta_minutes = int(remaining_seconds / 60) if remaining_seconds > 0 else 0 eta_seconds = int(remaining_seconds % 60) if remaining_seconds > 0 else 0 progress_data = { 'status': 'processing', 'percentage': resegment_progress, 'step': 'resegmenting', 'step_name': 'Re-segmenting to 2-second segments', 'details': { 'current_frame': current_frame, 'fps': round(current_fps, 1) if current_fps > 0 else 0, 'speed': round(current_speed, 2) if current_speed > 0 else 0, 'current_time': f"{hours:02d}:{minutes:02d}:{int(seconds):02d}", 'total_time': f"{int(duration // 60):02d}:{int(duration % 60):02d}", 'eta': f"{eta_minutes:02d}:{eta_seconds:02d}" } } r.setex(progress_key, 3600, json.dumps(progress_data)) resegment_process.wait() if resegment_process.returncode != 0: raise RuntimeError(f"FFmpeg resegment failed with code {resegment_process.returncode}") # Clean up temporary files os.remove(concat_list_path) os.remove(temp_video) for ts_file in ts_files: if os.path.exists(ts_file): os.remove(ts_file) sys.stderr.write(f"[FFMPEG] Re-segmenting completed\n") sys.stderr.flush() # Switch to the new 2-second playlist after resegmentation completes try: db = next(get_db()) stream = db.query(StreamModel).filter(StreamModel.id == stream_id).first() if stream: stream.hls_path = f'/uploads/videos/{stream_id}/{stream_id}.m3u8' stream.hls_url = f'/uploads/videos/{stream_id}/{stream_id}.m3u8' db.commit() sys.stderr.write(f"[STATUS] Stream {stream_id} switched to 2-second playlist\n") except Exception as e: sys.stderr.write(f"[STATUS ERROR] Failed to switch to 2-second playlist: {e}\n") # Regenerate master ffconcat since HLS path changed try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate after playlist switch: {e}\n") # Update progress: Resegment complete, generating optimized m3u8 (70%) progress_data = { 'status': 'processing', 'percentage': 70, 'step': 'generating_optimized_m3u8', 'step_name': 'Generating optimized m3u8', 'details': { 'current_operation': 'Creating HLS playlist with exact timestamps' } } r.setex(progress_key, 3600, json.dumps(progress_data)) # Generate optimized m3u8 with exact timestamps generate_optimized_m3u8(output_dir, output_name, progress_key) # Update progress: Optimized m3u8 complete, starting thumbnail generation (80%) # Update stream status to 'thumbnail_generation' try: db = next(get_db()) stream = db.query(StreamModel).filter(StreamModel.id == stream_id).first() if stream: stream.status = 'thumbnail_generation' db.commit() sys.stderr.write(f"[STATUS] Stream {stream_id} status set to 'thumbnail_generation'\n") except Exception as e: sys.stderr.write(f"[STATUS ERROR] Failed to update stream status: {e}\n") progress_data = { 'status': 'processing', 'percentage': 80, 'step': 'generating_thumbnails', 'step_name': 'Generating thumbnails', 'details': { 'current_operation': 'Preparing thumbnail generation' } } r.setex(progress_key, 3600, json.dumps(progress_data)) # Auto-start thumbnail generation (now at the end) sys.stderr.write(f"[THUMBNAILS] Auto-starting thumbnail generation for stream {stream_id}\n") sys.stderr.write(f"[THUMBNAILS] HLS directory: {output_dir}\n") sys.stderr.flush() import asyncio import threading from app.api.streams import generate_thumbnails_background def run_thumbnail_generation(): try: sys.stderr.write(f"[THUMBNAILS] Starting thumbnail thread for stream {stream_id}\n") sys.stderr.flush() asyncio.run(generate_thumbnails_background(stream_id, output_dir)) sys.stderr.write(f"[THUMBNAILS] Thumbnail thread completed for stream {stream_id}\n") sys.stderr.flush() except Exception as e: sys.stderr.write(f"[THUMBNAILS ERROR] Thumbnail thread failed: {e}\n") sys.stderr.flush() thumbnail_thread = threading.Thread(target=run_thumbnail_generation) thumbnail_thread.daemon = True thumbnail_thread.start() sys.stderr.write(f"[THUMBNAILS] Thumbnail thread started for stream {stream_id}\n") sys.stderr.flush() # Update progress: Thumbnail generation started (90%) progress_data = { 'status': 'processing', 'percentage': 90, 'step': 'generating_thumbnails', 'step_name': 'Generating thumbnails' } r.setex(progress_key, 3600, json.dumps(progress_data)) # Create m3u8 playlist with 2-second segments playlist_path = hls_path sys.stderr.write(f"[M3U8] Created playlist with 2-second segments\n") sys.stderr.flush() # Return playlist path instead of merged file output_path = playlist_path file_size = os.path.getsize(playlist_path) else: # Fallback: verify MP4 file exists if no fragments mp4_path = f"{output_dir}/{output_name}.mp4" if not os.path.exists(mp4_path): raise RuntimeError("Downloaded file not found") output_path = mp4_path file_size = os.path.getsize(mp4_path) # Mark as finished and set stream status to 'ready' try: db = next(get_db()) stream = db.query(StreamModel).filter(StreamModel.id == stream_id).first() if stream: stream.status = 'ready' stream.thumbnail_progress_generated = 0 stream.thumbnail_progress_total = 0 stream.thumbnail_progress_message = None stream.thumbnail_progress_error = None stream.thumbnail_progress_updated_at = None db.commit() sys.stderr.write(f"[STATUS] Stream {stream_id} status set to 'ready'\n") except Exception as e: sys.stderr.write(f"[STATUS ERROR] Failed to update stream status: {e}\n") # Regenerate master ffconcat try: from ..services.ffconcat_service import regenerate_master_ffconcat regenerate_master_ffconcat() except Exception as e: sys.stderr.write(f"[FFCONCAT ERROR] Failed to regenerate: {e}\n") progress_data = { 'status': 'finished', 'percentage': 100, 'file_size': file_size } r.setex(progress_key, 3600, json.dumps(progress_data)) sys.stderr.write(f"VOD download completed: {output_path} ({file_size} bytes)\n") sys.stderr.flush() return { 'file_path': output_path, 'file_size': file_size, 'vod_id': vod_id } except subprocess.TimeoutExpired: raise RuntimeError("VOD download timed out") except Exception as e: # Clean up partial download if os.path.exists(output_path): os.remove(output_path) raise RuntimeError(f"Failed to download VOD: {str(e)}") def cleanup_cache(self, vod_id: str): """Clean up cache files for a specific VOD""" import os files_to_remove = [ os.path.join(self.cache_dir, f"{vod_id}_chat.json"), os.path.join(self.cache_dir, f"{vod_id}.mp4") ] for file_path in files_to_remove: if os.path.exists(file_path): os.remove(file_path)