Create utils/ffmpeg.py

utils/ffmpeg.py

@@ -0,0 +1,83 @@
+# filename: utils/ffmpeg.py
+
+import asyncio
+import logging
+import os
+
+import config
+
+logger = logging.getLogger(__name__)
+
+async def _run_ffmpeg_command(command: list[str]) -> bool:
+    """A helper function to run an FFMPEG command asynchronously."""
+    if not config.ENABLE_FFMPEG:
+        logger.warning("FFMPEG is disabled in config, skipping command.")
+        return False
+        
+    process = await asyncio.create_subprocess_exec(
+        *command,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE
+    )
+    
+    stdout, stderr = await process.communicate()
+
+    if process.returncode != 0:
+        logger.error(f"FFMPEG command failed with exit code {process.returncode}")
+        logger.error(f"Stderr: {stderr.decode(errors='ignore').strip()}")
+        return False
+    
+    return True
+
+async def generate_thumbnail(video_path: str, thumbnail_path: str) -> str | None:
+    """
+    Generates a thumbnail from a video file.
+    Returns the path to the thumbnail on success, None on failure.
+    """
+    logger.info(f"Generating thumbnail for {video_path}...")
+    
+    # Command to seek 5 seconds into the video, grab one frame, resize it, and save as jpg
+    command = [
+        "ffmpeg",
+        "-i", video_path,
+        "-ss", "00:00:05",  # Seek to 5-second mark
+        "-vframes", "1",    # Extract only one frame
+        "-vf", "scale=320:-1", # Resize width to 320px, maintain aspect ratio
+        "-q:v", "3",        # Image quality (1-5, lower is better)
+        "-y",               # Overwrite output file if it exists
+        thumbnail_path
+    ]
+    
+    success = await _run_ffmpeg_command(command)
+
+    if success and os.path.exists(thumbnail_path):
+        logger.info(f"Thumbnail generated successfully: {thumbnail_path}")
+        return thumbnail_path
+    else:
+        logger.error(f"Failed to generate thumbnail for {video_path}")
+        return None
+
+async def remux_to_mp4(input_path: str, output_path: str) -> str | None:
+    """
+    Remuxes a video file (e.g., MKV) into an MP4 container without re-encoding.
+    This is a very fast operation.
+    Returns the path to the new MP4 file on success, None on failure.
+    """
+    logger.info(f"Remuxing {input_path} to MP4 format...")
+    
+    command = [
+        "ffmpeg",
+        "-i", input_path,
+        "-c", "copy",       # Copy all streams (video, audio, subtitles) without re-encoding
+        "-y",               # Overwrite output file
+        output_path
+    ]
+    
+    success = await _run_ffmpeg_command(command)
+    
+    if success and os.path.exists(output_path):
+        logger.info(f"Remuxed successfully: {output_path}")
+        return output_path
+    else:
+        logger.error(f"Failed to remux {input_path}")
+        return None