Add initial implementation of video encoder pipeline with FastAPI and FFmpeg

DOCS.md

@@ -0,0 +1,130 @@
+# Video Encoder Pipeline Documentation
+
+## Introduction
+The Video Encoder Pipeline is a robust solution for encoding videos into multiple resolutions and bitrates, suitable for Video on Demand (VOD) services. This pipeline leverages FastAPI for the API, Valkey (Redis) for job queuing, and FFmpeg for video encoding.
+
+## Quickstart Guide
+
+### Prerequisites
+- Python 3.8+
+- FFmpeg installed system-wide
+- Valkey (Redis) server running
+
+### Installation
+1. Clone the repository:
+   ```bash
+   git clone https://github.com/yourusername/video-encoder-pipeline.git
+   cd video-encoder-pipeline
+   ```
+
+2. Install the required dependencies:
+   ```bash
+   pip install -r requirements.txt
+   ```
+
+### Running the Server
+1. Start the FastAPI server:
+   ```bash
+   python run.py
+   ```
+
+2. The server will be available at `http://localhost:8000`.
+
+### Basic Usage
+1. Upload a video:
+   ```bash
+   curl -X POST -F "file=@input.mp4" http://localhost:8000/upload
+   ```
+
+2. Check encoding status:
+   ```bash
+   curl http://localhost:8000/status/{job_id}
+   ```
+
+3. Generate a token for playback:
+   ```bash
+   curl http://localhost:8000/token/{job_id}
+   ```
+
+4. Play encoded video:
+   ```bash
+   curl http://localhost:8000/play/{job_id}?token={valid_token}
+   ```
+
+## API Reference
+
+### Endpoints
+
+#### Upload Video
+- **Endpoint**: `/upload`
+- **Method**: `POST`
+- **Description**: Uploads a video file for encoding.
+- **Parameters**:
+  - `file`: The video file to upload.
+- **Response**:
+  - `job_id`: The ID of the encoding job.
+
+#### Get Encoding Status
+- **Endpoint**: `/status/{job_id}`
+- **Method**: `GET`
+- **Description**: Retrieves the encoding status of a job.
+- **Parameters**:
+  - `job_id`: The ID of the encoding job.
+- **Response**:
+  - `job_id`: The ID of the encoding job.
+  - `progress`: The encoding progress (0.0 to 1.0).
+
+#### Generate Token
+- **Endpoint**: `/token/{job_id}`
+- **Method**: `GET`
+- **Description**: Generates a token for secure playback.
+- **Parameters**:
+  - `job_id`: The ID of the encoding job.
+- **Response**:
+  - `token`: The generated token.
+
+#### Play Encoded Video
+- **Endpoint**: `/play/{job_id}`
+- **Method**: `GET`
+- **Description**: Streams the encoded video.
+- **Parameters**:
+  - `job_id`: The ID of the encoding job.
+  - `token`: The generated token.
+- **Response**:
+  - The HLS playlist for the encoded video.
+
+## Architecture Overview
+
+The video encoder pipeline consists of the following components:
+
+1. **FastAPI**: Handles API requests and responses.
+2. **Valkey (Redis)**: Manages job queuing and progress tracking.
+3. **FFmpeg**: Performs video encoding into multiple resolutions and bitrates.
+4. **HMAC**: Ensures secure access to encoded videos.
+
+## Encoding Profiles
+
+The pipeline supports the following encoding profiles:
+
+- **1080p**: 1920x1080, 5000 kbps video, 128 kbps audio
+- **720p**: 1280x720, 2500 kbps video, 128 kbps audio
+- **480p**: 854x480, 1000 kbps video, 96 kbps audio
+- **240p**: 426x240, 400 kbps video, 64 kbps audio
+
+## Security Considerations
+
+The pipeline uses HMAC for secure access to encoded videos. Each video playback request requires a valid token generated using the `/token/{job_id}` endpoint.
+
+## Troubleshooting
+
+### Common Issues
+
+1. **FFmpeg Not Found**: Ensure FFmpeg is installed and accessible in the system PATH.
+2. **Job Not Found**: Ensure the `job_id` is correct and the job has been enqueued.
+3. **Invalid Token**: Ensure the token is generated using the `/token/{job_id}` endpoint and is valid.
+
+### Solutions
+
+1. **FFmpeg Not Found**: Install FFmpeg and add it to the system PATH.
+2. **Job Not Found**: Verify the `job_id` and ensure the job has been enqueued.
+3. **Invalid Token**: Generate a new token using the `/token/{job_id}` endpoint.

Dockerfile

@@ -0,0 +1,15 @@
+FROM python:3.10
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+EXPOSE 7860
+
+COPY --chown=user . /app
+CMD ["python","run.py"]

requirements.txt

@@ -0,0 +1,7 @@
+fastapi[all]==0.115.12
+uvicorn==0.34.0
+valkey==6.1.0
+rq==2.3.2
+ffmpeg-python==0.2.0
+python-multipart==0.0.20
+python-dotenv==1.1.0

run.py

@@ -0,0 +1,9 @@
+import uvicorn
+from video_encoder.api.main import app
+
+if __name__ == "__main__":
+    uvicorn.run(
+        app,
+        host="0.0.0.0",
+        port=7860
+    )

video_encoder/api/__init__.py

@@ -0,0 +1 @@
+# Package initialization

video_encoder/api/main.py

@@ -0,0 +1,72 @@
+from fastapi import FastAPI, UploadFile, HTTPException, status
+from fastapi.responses import JSONResponse, FileResponse
+from pathlib import Path
+import uuid
+import os
+import hmac
+import hashlib
+from datetime import datetime
+from rq import Queue
+from redis import Redis
+from ..config import EncodingConfig
+from ..worker.tasks import encode_video_task, q
+
+app = FastAPI(title="Video Encoding Service")
+
+@app.post("/upload")
+async def upload_video(file: UploadFile):
+    job_id = str(uuid.uuid4())
+    temp_dir = os.path.join(EncodingConfig.TEMP_DIR, job_id)
+    os.makedirs(temp_dir, exist_ok=True)
+    
+    input_path = os.path.join(temp_dir, file.filename)
+    with open(input_path, "wb") as f:
+        f.write(await file.read())
+        
+    q.enqueue(encode_video_task, job_id, input_path)
+    
+    return JSONResponse(
+        content={"job_id": job_id},
+        status_code=status.HTTP_202_ACCEPTED
+    )
+
+@app.get("/status/{job_id}")
+async def get_status(job_id: str):
+    redis = Redis()
+    progress = redis.hget(f"job:{job_id}", "progress")
+    
+    if not progress:
+        raise HTTPException(status_code=404, detail="Job not found")
+    
+    return {"job_id": job_id, "progress": float(progress)}
+
+@app.get("/play/{job_id}")
+async def play_video(job_id: str, token: str):
+    expected_token = hmac.new(
+        EncodingConfig.HMAC_SECRET.encode(),
+        job_id.encode(),
+        hashlib.sha256
+    ).hexdigest()
+    
+    if not hmac.compare_digest(token, expected_token):
+        raise HTTPException(status_code=403, detail="Invalid token")
+    
+    master_playlist = os.path.join(EncodingConfig.TEMP_DIR, job_id, f"{job_id}_master.m3u8")
+    if not os.path.exists(master_playlist):
+        raise HTTPException(status_code=404, detail="Playlist not found")
+        
+    return FileResponse(
+        master_playlist,
+        media_type="application/vnd.apple.mpegurl"
+    )
+
+@app.get("/token/{job_id}")
+async def generate_token(job_id: str):
+    timestamp = datetime.now().isoformat()
+    signature = hmac.new(
+        EncodingConfig.HMAC_SECRET.encode(),
+        f"{job_id}{timestamp}".encode(),
+        hashlib.sha256
+    ).hexdigest()
+    
+    return {"token": f"{timestamp}:{signature}"}

video_encoder/config.py

@@ -0,0 +1,55 @@
+import os
+from typing import List, Dict
+
+class EncodingConfig:
+    TEMP_DIR = "./tmp"
+    HMAC_SECRET = os.getenv("VIDEO_HMAC_SECRET", "default-secret-change-me")
+    
+    RESOLUTIONS: List[Dict] = [
+        {
+            "name": "1080p",
+            "width": 1920,
+            "height": 1080,
+            "video_bitrate": "5000k",
+            "audio_bitrate": "128k",
+            "codec": "libx264",
+            "profile": "high",
+            "preset": "slow"
+        },
+        {
+            "name": "720p", 
+            "width": 1280,
+            "height": 720,
+            "video_bitrate": "2500k",
+            "audio_bitrate": "128k",
+            "codec": "libx264",
+            "profile": "main",
+            "preset": "medium"
+        },
+        {
+            "name": "480p",
+            "width": 854,
+            "height": 480,
+            "video_bitrate": "1000k",
+            "audio_bitrate": "96k",
+            "codec": "libx264",
+            "profile": "baseline",
+            "preset": "fast"
+        },
+        {
+            "name": "240p",
+            "width": 426,
+            "height": 240,
+            "video_bitrate": "400k",
+            "audio_bitrate": "64k", 
+            "codec": "libx264",
+            "profile": "baseline",
+            "preset": "veryfast"
+        }
+    ]
+    
+    @classmethod
+    def validate_resolutions(cls):
+        for res in cls.RESOLUTIONS:
+            if not all(key in res for key in ['name', 'width', 'height', 'video_bitrate']):
+                raise ValueError(f"Invalid resolution config: {res}")

video_encoder/worker/ffmpeg.py

@@ -0,0 +1,56 @@
+import subprocess
+import os
+from pathlib import Path
+from typing import List, Dict
+from ..config import EncodingConfig
+
+class FFmpegEncoder:
+    def __init__(self, input_path: str, output_dir: str):
+        self.input_path = input_path
+        self.output_dir = output_dir
+        self.base_name = Path(input_path).stem
+        os.makedirs(output_dir, exist_ok=True)
+
+    def generate_commands(self) -> List[List[str]]:
+        """Generate FFmpeg commands for all resolutions"""
+        commands = []
+        for res in EncodingConfig.RESOLUTIONS:
+            output_path = os.path.join(
+                self.output_dir,
+                f"{self.base_name}_{res['name']}.m3u8"
+            )
+            
+            cmd = [
+                "ffmpeg", "-i", self.input_path,
+                "-vf", f"scale={res['width']}:{res['height']}",
+                "-c:v", res["codec"],
+                "-profile:v", res["profile"],
+                "-preset", res["preset"],
+                "-b:v", res["video_bitrate"],
+                "-c:a", "aac",
+                "-b:a", res["audio_bitrate"],
+                "-f", "hls",
+                "-hls_time", "6",
+                "-hls_playlist_type", "vod",
+                "-hls_segment_filename", 
+                os.path.join(self.output_dir, f"{self.base_name}_{res['name']}_%03d.ts"),
+                output_path
+            ]
+            commands.append(cmd)
+        return commands
+
+    def encode(self) -> str:
+        """Execute encoding commands and return master playlist path"""
+        master_playlist = os.path.join(self.output_dir, f"{self.base_name}_master.m3u8")
+        
+        with open(master_playlist, "w") as f:
+            f.write("#EXTM3U\n")
+            for res in reversed(EncodingConfig.RESOLUTIONS):
+                f.write(f"#EXT-X-STREAM-INF:BANDWIDTH={res['video_bitrate'].replace('k', '000')},"
+                        f"RESOLUTION={res['width']}x{res['height']}\n")
+                f.write(f"{self.base_name}_{res['name']}.m3u8\n")
+        
+        for cmd in self.generate_commands():
+            subprocess.run(cmd, check=True, capture_output=True)
+            
+        return master_playlist

video_encoder/worker/tasks.py

@@ -0,0 +1,35 @@
+import os
+import logging
+import shutil
+from redis import Redis
+from rq import Queue
+from pathlib import Path
+from typing import Optional
+from .ffmpeg import FFmpegEncoder
+from ..config import EncodingConfig
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+q = Queue(connection=Redis(), default_timeout=3600)
+
+def update_job_progress(job_id: str, progress: float):
+    Redis().hset(f"job:{job_id}", "progress", progress)
+
+def encode_video_task(job_id: str, input_path: str) -> Optional[str]:
+    try:
+        output_dir = os.path.join(EncodingConfig.TEMP_DIR, job_id)
+        encoder = FFmpegEncoder(input_path, output_dir)
+        
+        update_job_progress(job_id, 0.1)
+        master_playlist = encoder.encode()
+        update_job_progress(job_id, 1.0)
+        
+        return master_playlist
+    except Exception as e:
+        logger.error(f"Encoding failed for job {job_id}: {str(e)}")
+        update_job_progress(job_id, -1.0)
+        raise
+    finally:
+        # Cleanup temporary files after 1 hour
+        pass  # Implement retention policy as needed