Design a Video Streaming Platform

import boto3
import hashlib

class VideoUploadService:
    """
    Handles chunked, resumable video uploads.

    Flow:
    1. Client requests upload URL (pre-signed S3 multipart)
    2. Client uploads in 5MB chunks
    3. On complete: trigger transcoding pipeline
    """

    def __init__(self):
        self.s3 = boto3.client('s3')
        self.BUCKET = 'raw-videos'
        self.CHUNK_SIZE = 5 * 1024 * 1024  # 5MB

    def initiate_upload(self, video_id: str,
                         filename: str) -> dict:
        """Start a multipart upload and return upload_id."""
        key = f"uploads/{video_id}/{filename}"
        response = self.s3.create_multipart_upload(
            Bucket=self.BUCKET, Key=key,
            ContentType='video/mp4'
        )
        return {
            "upload_id": response["UploadId"],
            "video_id": video_id,
            "chunk_size": self.CHUNK_SIZE
        }

    def get_chunk_url(self, video_id: str,
                      upload_id: str,
                      part_number: int) -> str:
        """Generate a pre-signed URL for uploading one chunk."""
        return self.s3.generate_presigned_url(
            'upload_part',
            Params={
                'Bucket': self.BUCKET,
                'Key': f"uploads/{video_id}/video.mp4",
                'UploadId': upload_id,
                'PartNumber': part_number
            },
            ExpiresIn=3600  # 1 hour
        )

    def complete_upload(self, video_id: str, upload_id: str):
        """Finalize upload and trigger transcoding."""
        # Complete multipart upload in S3
        self.s3.complete_multipart_upload(...)
        # Publish job to transcoding queue
        queue.publish("video-transcode", {
            "video_id": video_id,
            "source": f"s3://raw-videos/uploads/{video_id}/"
        })

import subprocess
import json

class TranscodeWorker:
    """
    Video transcoding worker using FFmpeg.
    Produces multiple resolutions + HLS segments.
    """

    PROFILES = [
        {"name": "360p",  "width": 640,  "height": 360,  "bitrate": "800k"},
        {"name": "480p",  "width": 854,  "height": 480,  "bitrate": "1500k"},
        {"name": "720p",  "width": 1280, "height": 720,  "bitrate": "3000k"},
        {"name": "1080p", "width": 1920, "height": 1080, "bitrate": "6000k"},
    ]

    def transcode(self, source_path: str,
                   output_dir: str, video_id: str):
        """Transcode video into multiple HLS streams."""
        for profile in self.PROFILES:
            output = f"{output_dir}/{profile['name']}"
            cmd = [
                "ffmpeg", "-i", source_path,
                "-vf", f"scale={profile['width']}:{profile['height']}",
                "-b:v", profile["bitrate"],
                "-codec:v", "h264",
                "-codec:a", "aac",
                "-hls_time", "6",       # 6-second segments
                "-hls_list_size", "0",  # Keep all segments
                "-f", "hls",
                f"{output}/playlist.m3u8"
            ]
            subprocess.run(cmd, check=True)

        # Generate master playlist
        self._create_master_playlist(output_dir, video_id)

#EXTM3U
#EXT-X-VERSION:3

# 360p variant
#EXT-X-STREAM-INF:BANDWIDTH=800000,RESOLUTION=640x360
360p/playlist.m3u8

# 480p variant
#EXT-X-STREAM-INF:BANDWIDTH=1500000,RESOLUTION=854x480
480p/playlist.m3u8

# 720p variant
#EXT-X-STREAM-INF:BANDWIDTH=3000000,RESOLUTION=1280x720
720p/playlist.m3u8

# 1080p variant
#EXT-X-STREAM-INF:BANDWIDTH=6000000,RESOLUTION=1920x1080
1080p/playlist.m3u8

# The player downloads this master playlist first,
# then selects the appropriate quality level and
# fetches 6-second .ts segments from that stream.

class RecommendationPipeline:
    """
    Simplified recommendation pipeline.
    In production, each stage is a separate microservice.
    """

    def get_recommendations(self, user_id: str,
                             count: int = 20) -> list:
        # Stage 1: Candidate generation (fast, broad)
        candidates = []
        candidates += self._from_watch_history(user_id, 200)
        candidates += self._from_subscriptions(user_id, 200)
        candidates += self._collaborative_filter(user_id, 200)
        candidates += self._trending_videos(100)

        # Deduplicate and filter already-watched
        candidates = self._deduplicate(candidates)
        candidates = self._filter_watched(user_id, candidates)

        # Stage 2: Ranking (slower, precise)
        scored = []
        for video in candidates:
            score = self._predict_watch_time(user_id, video)
            scored.append((score, video))
        scored.sort(reverse=True)

        # Stage 3: Re-ranking (business rules)
        final = self._apply_diversity(scored[:count * 2])
        final = self._apply_content_policy(final)
        return final[:count]

    def _predict_watch_time(self, user_id, video) -> float:
        """ML model predicts expected watch time (in seconds)."""
        features = {
            "user_history": get_user_embedding(user_id),
            "video_features": get_video_embedding(video.id),
            "freshness": video.age_hours,
            "channel_affinity": get_affinity(user_id, video.channel_id),
        }
        return self.model.predict(features)