import subprocess import os from typing import Optional class ThumbnailService: def __init__(self): pass def generate_thumbnail_from_timestamp( self, video_path: str, timestamp_seconds: float, output_path: str, width: int = 1280, height: int = 720, segment_duration: int = 10 ) -> str: """ Generate a thumbnail from a specific timestamp in a video or HLS playlist. For HLS, FFmpeg can directly read the m3u8 playlist and seek to the timestamp. Args: video_path: Path to video file or HLS playlist timestamp_seconds: Timestamp in seconds to capture thumbnail from output_path: Path where thumbnail should be saved width: Thumbnail width (default: 1280) height: Thumbnail height (default: 720) segment_duration: Duration of each HLS segment in seconds (default: 10) Returns: Path to generated thumbnail """ # Ensure output directory exists os.makedirs(os.path.dirname(output_path), exist_ok=True) # Check if this is an HLS playlist (.m3u8) if video_path.endswith('.m3u8'): # Check if MP4 file exists (easier to handle) hls_dir = os.path.dirname(video_path) stream_id = os.path.basename(video_path).replace('.m3u8', '') mp4_path = f"{hls_dir}/{stream_id}.mp4" if os.path.exists(mp4_path): # Use MP4 file directly video_path = mp4_path else: # Try to use optimized m3u8 with #EXT-X-START-TIME tags import re optimized_m3u8 = f"{hls_dir}/{stream_id}_optimized.m3u8" if os.path.exists(optimized_m3u8): # Parse optimized m3u8 to find segment for timestamp with open(optimized_m3u8, 'r') as f: lines = f.readlines() segment_timestamps = [] current_start_time = None for line in lines: if line.startswith('#EXT-X-START-TIME:'): current_start_time = float(line.split(':')[1].strip()) elif line.startswith(f'{stream_id}-') and line.endswith('.ts\n'): if current_start_time is not None: seg_num = int(line.split('-')[-1].replace('.ts\n', '')) segment_timestamps.append((seg_num, current_start_time)) current_start_time = None if segment_timestamps: # Find segment that contains the target timestamp target_segment = None for i, (seg_num, start_time) in enumerate(segment_timestamps): if i < len(segment_timestamps) - 1: next_start_time = segment_timestamps[i + 1][1] else: next_start_time = start_time + segment_duration if start_time <= timestamp_seconds < next_start_time: target_segment = (seg_num, start_time, next_start_time) break if target_segment: seg_num, start_time, end_time = target_segment timestamp_within_segment = timestamp_seconds - start_time # Try standard HLS format first: stream_id-N.ts segment_file = f"{hls_dir}/{stream_id}-{seg_num}.ts" if not os.path.exists(segment_file): # Fallback to Twitch format: stream_id.ts-FragN segment_file = f"{hls_dir}/{stream_id}.ts-Frag{seg_num}" if os.path.exists(segment_file): video_path = segment_file timestamp_seconds = timestamp_within_segment else: # Fall back to ffprobe scanning pass # If optimized m3u8 not available or not successful, use ffprobe scanning if video_path.endswith('.m3u8'): import subprocess # 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) # Extract timestamps for each segment segment_timestamps = [] for seg_file in segment_files: seg_num = extract_segment_number(seg_file) seg_path = os.path.join(hls_dir, seg_file) 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: raise Exception(f"No segment timestamps found for {stream_id}") # Find segment that contains the target timestamp target_segment = None for i, (seg_num, start_time) in enumerate(segment_timestamps): if i < len(segment_timestamps) - 1: next_start_time = segment_timestamps[i + 1][1] else: next_start_time = start_time + segment_duration if start_time <= timestamp_seconds < next_start_time: target_segment = (seg_num, start_time, next_start_time) break if not target_segment: raise Exception(f"No segment found for timestamp {timestamp_seconds}s") seg_num, start_time, end_time = target_segment timestamp_within_segment = timestamp_seconds - start_time # Try standard HLS format first: stream_id-N.ts segment_file = f"{hls_dir}/{stream_id}-{seg_num}.ts" if not os.path.exists(segment_file): # Fallback to Twitch format: stream_id.ts-FragN segment_file = f"{hls_dir}/{stream_id}.ts-Frag{seg_num}" if not os.path.exists(segment_file): raise Exception(f"Segment file not found for timestamp {timestamp_seconds}s. Tried: {segment_file}") # Use the segment file directly video_path = segment_file # Calculate timestamp within the segment (0-10 seconds) timestamp_seconds = timestamp_within_segment # Build FFmpeg command cmd = [ 'ffmpeg', '-ss', str(timestamp_seconds), '-i', video_path, '-frames:v', '1', '-q:v', '2', '-strict', '-2', '-y', output_path ] try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=30 ) if result.returncode != 0: raise Exception(f"FFmpeg failed: {result.stderr}") if not os.path.exists(output_path): raise Exception("Thumbnail file was not created") return output_path except subprocess.TimeoutExpired: raise Exception("FFmpeg thumbnail generation timed out") except Exception as e: raise Exception(f"Failed to generate thumbnail: {str(e)}")