Implement OTF ecoding for videos

app/routers/videos.py

@@ -1,7 +1,8 @@
 import os
 import random
+import asyncio
 from fastapi import APIRouter, Depends, HTTPException, status
-from fastapi.responses import FileResponse
+from fastapi.responses import FileResponse, StreamingResponse
 from typing import List
 from ..models import Vote, VoteCreate, Video
 from ..security import is_admin, is_user
@@ -16,14 +17,14 @@ VIDEO_DIRECTORY = os.environ.get("VIDEO_DIRECTORY")
 async def scan_videos_directory(current_user: dict = Depends(is_admin)):
     """
     Scans the video directory for new clips and adds them to the database.
-    Handles structures: 
+    Handles structures:
     - /videos/person/game/clip[.mp4|.mkv]
     - /videos/person/clip[.mp4|.mkv]
     """
     new_videos_count = 0
     if not VIDEO_DIRECTORY or not os.path.isdir(VIDEO_DIRECTORY):
         raise HTTPException(status_code=500, detail="Video directory not found or not configured on server")
-    
+
     db = get_db()
     videos_collection = db.collection('videos')
 
@@ -37,7 +38,7 @@ async def scan_videos_directory(current_user: dict = Depends(is_admin)):
         for item_name in os.listdir(person_path):
             item_path = os.path.join(person_path, item_name)
             game_name = None
-            
+
             # Structure 1: /person/game/clip.mkv
             if os.path.isdir(item_path):
                 game_name = item_name
@@ -47,7 +48,7 @@ async def scan_videos_directory(current_user: dict = Depends(is_admin)):
                     if os.path.isfile(clip_path) and clip_name.lower().endswith(('.mp4', '.mkv')):
                         # Construct relative path for DB: person/game/clip.mkv
                         relative_path = os.path.join(person_name, game_name, clip_name)
-                        
+
                         # Check for duplicates
                         existing = videos_collection.where('file_path', '==', relative_path).limit(1).get()
                         if not existing:
@@ -61,7 +62,7 @@ async def scan_videos_directory(current_user: dict = Depends(is_admin)):
                             }
                             new_video_ref.set(video_data)
                             new_videos_count += 1
-            
+
             # Structure 2: /person/clip.mp4
             elif os.path.isfile(item_path) and item_name.lower().endswith(('.mp4', '.mkv')):
                 clip_name = item_name
@@ -88,15 +89,20 @@ async def scan_videos_directory(current_user: dict = Depends(is_admin)):
 @router.get("/vote-next", response_model=Video)
 async def get_random_unvoted_video(current_user: dict = Depends(is_user)):
     """
-    Retrieves a random, unvoted video document from Firestore.
+    Retrieves a random, unvoted video document from Firestore efficiently using a small batch approach.
     """
     db = get_db()
-    videos_stream = db.collection('videos').where('has_been_voted', '==', False).limit(1).stream()
-    unvoted_videos = [doc.to_dict() for doc in videos_stream]
-    
+
+    limit_count = 5   
+    unvoted_videos_query = db.collection('videos').where('has_been_voted', '==', False).limit(limit_count)
+    unvoted_videos = [doc.to_dict() for doc in unvoted_videos_query.stream()]
+
     if not unvoted_videos:
         raise HTTPException(status_code=404, detail="No more videos to vote on!")
-    
+
+    # Randomly select one video from the fetched batch
+    random_video_data = random.choice(unvoted_videos)
+
     return Video(**random_video_data)
 
 
@@ -132,7 +138,7 @@ async def submit_vote(video_id: str, vote_data: VoteCreate, current_user: dict =
 @router.get("/{video_id}/stream")
 async def stream_video(video_id: str, current_user: dict = Depends(is_user)):
     """
-    Streams a video file from the server based on its Firestore document.
+    Streams a video file from the server with on-the-fly transcoding and chunking using FFmpeg.
     """
     db = get_db()
     video_doc = db.collection('videos').document(video_id).get()
@@ -140,12 +146,81 @@ async def stream_video(video_id: str, current_user: dict = Depends(is_user)):
     if not video_doc.exists:
         raise HTTPException(status_code=404, detail="Video not found")
     
-    full_path = os.path.join(VIDEO_DIRECTORY, video_doc.to_dict()["file_path"])
-    print(full_path)
+    video_data = video_doc.to_dict()
+    full_path = os.path.join(VIDEO_DIRECTORY, video_data["file_path"])
+
     if not os.path.exists(full_path):
         raise HTTPException(status_code=404, detail="Video file not found on disk")
         
-    return FileResponse(full_path, media_type="video/mp4")
+    async def generate_video_chunks():
+        # FFmpeg command to transcode to MP4 (H.264/AAC) and output to stdout
+        # -i: input file
+        # -movflags frag_keyframe+empty_moov: essential for fragmented MP4,
+        #                                       allowing playback before full file is downloaded
+        # -f mp4: output format is MP4
+        # -codec:v libx264: video codec H.264
+        # -preset veryfast: transcoding speed vs quality trade-off (adjust as needed)
+        # -crf 28: constant rate factor for H.264 quality (higher is lower quality/smaller file, 23 is default for libx264)
+        # -codec:a aac: audio codec AAC
+        # -b:a 128k: audio bitrate
+        # -vf scale=-1:720: scale video height to 720 pixels, maintain aspect ratio (adjust or remove if not needed)
+        # -loglevel warning -hide_banner: suppress verbose FFmpeg output for cleaner logs
+        # pipe:1: output to stdout
+        ffmpeg_command = [
+            'ffmpeg',
+            '-i', full_path,
+            '-movflags', 'frag_keyframe+empty_moov',
+            '-f', 'mp4',
+            '-codec:v', 'libx264',
+            '-preset', 'veryfast',
+            '-crf', '28',
+            '-codec:a', 'aac',
+            '-b:a', '128k',
+            '-vf', 'scale=-1:720',
+            '-loglevel', 'warning',
+            '-hide_banner',
+            'pipe:1'
+        ]
+
+        process = None
+        try:
+            process = await asyncio.create_subprocess_exec(
+                *ffmpeg_command,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE # Capture stderr for potential error logging
+            )
+
+            # Read chunks from stdout and yield them
+            while True:
+                chunk = await process.stdout.read(8192) # Read 8KB chunks
+                if not chunk:
+                    break
+                yield chunk
+
+        except asyncio.CancelledError:
+            # This exception is raised if the client disconnects before the stream is complete
+            print(f"[{video_id}] Streaming cancelled by client.")
+        except Exception as e:
+            print(f"[{video_id}] Error during video streaming: {e}")
+            if process and process.returncode is None:
+                # If FFmpeg process is still running, try to read any remaining stderr
+                stderr_output = await process.stderr.read()
+                print(f"[{video_id}] FFmpeg stderr: {stderr_output.decode()}")
+            raise HTTPException(status_code=500, detail="Error processing video stream.")
+        finally:
+            if process and process.returncode is None:
+                # Terminate the FFmpeg process if it's still running
+                print(f"[{video_id}] Terminating FFmpeg process.")
+                process.terminate() # Send SIGTERM
+                await process.wait() # Wait for process to exit
+            elif process and process.returncode != 0:
+                # If FFmpeg exited with an error code
+                stderr_output = await process.stderr.read()
+                print(f"[{video_id}] FFmpeg exited with error code {process.returncode}: {stderr_output.decode()}")
+                # You might want to log this error more formally or send a more specific client error
+            print(f"[{video_id}] Video streaming finished or terminated.")
+
+    return StreamingResponse(generate_video_chunks(), media_type="video/mp4")
 
 
 @router.get("/votes", response_model=List[Vote])