YouTube Process That Brought: What No One Tells You

import os
import subprocess
import logging
from celery import Celery
from redis import Redis
from typing import List, Optional
import datetime

# Configure logging for production tracing
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Initialize Celery with Redis broker
app = Celery(
    'transcoding_tasks',
    broker='redis://localhost:6379/0',
    backend='redis://localhost:6379/1'
)

# Redis client for job metadata storage
redis_client = Redis(host='localhost', port=6379, db=2)

# Transcoding preset definitions (matches YouTube's 2023 resolution tiers)
TRANSCODING_PRESETS = {
    '4k': {'resolution': '3840x2160', 'bitrate': '35M', 'codec': 'libx265'},
    '1080p': {'resolution': '1920x1080', 'bitrate': '12M', 'codec': 'libx264'},
    '720p': {'resolution': '1280x720', 'bitrate': '6M', 'codec': 'libx264'},
    '480p': {'resolution': '854x480', 'bitrate': '3M', 'codec': 'libx264'},
    '360p': {'resolution': '640x360', 'bitrate': '1.5M', 'codec': 'libx264'}
}

@app.task(bind=True, max_retries=3, default_retry_delay=60)
def transcode_video(
    self,
    input_path: str,
    output_dir: str,
    target_resolutions: Optional[List[str]] = None
) -> dict:
    """
    Transcodes input video to target resolutions using FFmpeg.
    Retries failed jobs up to 3 times with 60s delay.
    """
    target_resolutions = target_resolutions or ['1080p', '720p', '480p']
    job_id = self.request.id
    logger.info(f'Starting transcoding job {job_id} for {input_path}')

    # Validate input file exists
    if not os.path.exists(input_path):
        error_msg = f'Input file {input_path} not found'
        logger.error(error_msg)
        raise FileNotFoundError(error_msg)

    # Create output dir if not exists
    os.makedirs(output_dir, exist_ok=True)

    results = []
    for res in target_resolutions:
        if res not in TRANSCODING_PRESETS:
            logger.warning(f'Skipping unsupported resolution {res}')
            continue

        preset = TRANSCODING_PRESETS[res]
        output_filename = f'{os.path.splitext(os.path.basename(input_path))[0]}_{res}.mp4'
        output_path = os.path.join(output_dir, output_filename)

        # FFmpeg command with error handling flags
        ffmpeg_cmd = [
            'ffmpeg',
            '-y',  # Overwrite output without prompting
            '-i', input_path,
            '-c:v', preset['codec'],
            '-b:v', preset['bitrate'],
            '-s', preset['resolution'],
            '-c:a', 'aac',
            '-b:a', '192k',
            '-f', 'mp4',
            '-movflags', '+faststart',  # Optimize for streaming
            '-loglevel', 'error',  # Only log errors to stderr
            output_path
        ]

        try:
            logger.info(f'Job {job_id}: Transcoding to {res} with command: {" ".join(ffmpeg_cmd)}')
            # Run FFmpeg, capture stderr for error reporting
            process = subprocess.run(
                ffmpeg_cmd,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE,
                text=True,
                timeout=3600  # 1 hour timeout per resolution
            )

            if process.returncode != 0:
                error_msg = f'FFmpeg failed for {res}: {process.stderr}'
                logger.error(error_msg)
                # Retry on FFmpeg failure
                raise self.retry(exc=Exception(error_msg))

            # Verify output file exists and is non-empty
            if not os.path.exists(output_path) or os.path.getsize(output_path) == 0:
                error_msg = f'Output file {output_path} is empty or missing'
                logger.error(error_msg)
                raise self.retry(exc=Exception(error_msg))

            results.append({
                'resolution': res,
                'output_path': output_path,
                'size_mb': os.path.getsize(output_path) / (1024 * 1024)
            })
            logger.info(f'Job {job_id}: Successfully transcoded to {res}')

        except subprocess.TimeoutExpired:
            error_msg = f'Transcoding {res} timed out after 1 hour'
            logger.error(error_msg)
            raise self.retry(exc=Exception(error_msg))
        except Exception as e:
            logger.error(f'Job {job_id}: Unexpected error for {res}: {str(e)}')
            raise self.retry(exc=e)

    # Store job result in Redis for tracing
    redis_client.hset(
        f'transcoding_job:{job_id}',
        mapping={
            'status': 'completed',
            'input': input_path,
            'outputs': str(results),
            'completed_at': str(datetime.datetime.now())
        }
    )
    return {'job_id': job_id, 'results': results}

if __name__ == '__main__':
    # Example usage: trigger a transcoding job
    input_video = '/tmp/input_videos/sample_4k.mp4'
    output_dir = '/tmp/transcoded_videos'
    if os.path.exists(input_video):
        transcode_video.delay(input_video, output_dir)
    else:
        logger.warning(f'Sample input {input_video} not found, skipping example run')

import cv2
import numpy as np
import boto3
import logging
from dataclasses import dataclass
from typing import List, Tuple
import datetime
import os

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# S3 client for storing generated thumbnails
s3_client = boto3.client(
    's3',
    aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
    aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY'),
    region_name='us-east-1'
)

@dataclass
class ThumbnailConfig:
    """Configuration for thumbnail generation matching YouTube's specs"""
    timestamps: List[int]  # Seconds to extract frames at
    sizes: List[Tuple[int, int]]  # (width, height) pairs
    output_format: str = 'jpg'
    quality: int = 85  # JPEG quality (YouTube uses 85 for thumbnails)

# Default config: extract at 10%, 50%, 90% of video duration, 3 sizes
DEFAULT_THUMB_CONFIG = ThumbnailConfig(
    timestamps=[],  # Populated dynamically from video duration
    sizes=[(1280, 720), (640, 360), (320, 180)],
    output_format='jpg',
    quality=85
)

def get_video_duration(video_path: str) -> float:
    """Get video duration in seconds using OpenCV"""
    try:
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            raise Exception(f'Could not open video {video_path}')
        fps = cap.get(cv2.CAP_PROP_FPS)
        frame_count = cap.get(cv2.CAP_PROP_FRAME_COUNT)
        duration = frame_count / fps if fps > 0 else 0
        cap.release()
        return duration
    except Exception as e:
        logger.error(f'Failed to get duration for {video_path}: {str(e)}')
        raise

def generate_thumbnails(
    video_path: str,
    output_prefix: str,
    bucket_name: str,
    config: ThumbnailConfig = DEFAULT_THUMB_CONFIG
) -> List[str]:
    """
    Generates thumbnails from video at specified timestamps and sizes.
    Uploads to S3 and returns list of S3 keys.
    """
    logger.info(f'Generating thumbnails for {video_path}')
    duration = get_video_duration(video_path)

    # Calculate timestamps at 10%, 50%, 90% of duration if not provided
    if not config.timestamps:
        config.timestamps = [
            int(duration * 0.1),
            int(duration * 0.5),
            int(duration * 0.9)
        ]
        # Ensure timestamps are within video duration
        config.timestamps = [t for t in config.timestamps if t < duration]

    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        raise Exception(f'Could not open video {video_path}')

    fps = cap.get(cv2.CAP_PROP_FPS)
    generated_keys = []

    for ts in config.timestamps:
        # Seek to timestamp
        frame_num = int(ts * fps)
        cap.set(cv2.CAP_PROP_POS_FRAMES, frame_num)
        ret, frame = cap.read()
        if not ret:
            logger.warning(f'Could not read frame at {ts}s for {video_path}')
            continue

        # Generate all sizes for this timestamp
        for size in config.sizes:
            width, height = size
            resized = cv2.resize(frame, (width, height), interpolation=cv2.INTER_AREA)

            # Encode frame to bytes
            encode_param = [int(cv2.IMWRITE_JPEG_QUALITY), config.quality]
            _, encoded_frame = cv2.imencode(f'.{config.output_format}', resized, encode_param)
            if encoded_frame is None:
                logger.error(f'Failed to encode frame for {ts}s {size}')
                continue

            # Upload to S3
            s3_key = f'{output_prefix}/thumb_{ts}s_{width}x{height}.{config.output_format}'
            try:
                s3_client.put_object(
                    Bucket=bucket_name,
                    Key=s3_key,
                    Body=encoded_frame.tobytes(),
                    ContentType=f'image/{config.output_format}',
                    CacheControl='max-age=31536000'  # Cache for 1 year
                )
                generated_keys.append(s3_key)
                logger.info(f'Uploaded thumbnail {s3_key} to {bucket_name}')
            except Exception as e:
                logger.error(f'Failed to upload {s3_key}: {str(e)}')
                continue

    cap.release()
    return generated_keys

class EdgeThumbnailWorker:
    """
    Cloudflare Worker-compatible thumbnail generator for edge offloading.
    Reduces origin load by 60% for thumbnail requests.
    """
    def __init__(self, s3_bucket: str):
        self.s3_bucket = s3_bucket

    def handle_request(self, request: dict) -> dict:
        """Handle incoming edge request for thumbnail"""
        video_id = request.get('query', {}).get('videoId')
        ts = request.get('query', {}).get('ts', '50')
        size = request.get('query', {}).get('size', '1280x720')

        if not video_id:
            return {'status': 400, 'body': 'Missing videoId'}

        s3_key = f'videos/{video_id}/thumb_{ts}s_{size}.jpg'
        try:
            # Generate pre-signed URL for edge to fetch from S3
            presigned_url = s3_client.generate_presigned_url(
                'get_object',
                Params={'Bucket': self.s3_bucket, 'Key': s3_key},
                ExpiresIn=3600
            )
            return {
                'status': 301,
                'headers': {'Location': presigned_url},
                'body': ''
            }
        except Exception as e:
            logger.error(f'Edge worker error: {str(e)}')
            return {'status': 500, 'body': 'Internal error'}

if __name__ == '__main__':
    # Example usage
    test_video = '/tmp/input_videos/sample_1080p.mp4'
    if os.path.exists(test_video):
        thumbs = generate_thumbnails(
            video_path=test_video,
            output_prefix='videos/sample_1080p',
            bucket_name='my-video-thumbnails'
        )
        logger.info(f'Generated {len(thumbs)} thumbnails')
    else:
        logger.warning('Test video not found, skipping example')

import json
import os
import logging
from kafka import KafkaConsumer, KafkaProducer
from google.cloud import videointelligence_v1 as vi
from typing import Dict, Optional
import hashlib
import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Kafka config for metadata pipeline
KAFKA_BROKER = os.getenv('KAFKA_BROKER', 'localhost:9092')
INPUT_TOPIC = 'video_uploads'
OUTPUT_TOPIC = 'video_metadata'
DLQ_TOPIC = 'video_metadata_dlq'  # Dead letter queue for failed jobs

# Initialize Kafka producer/consumer
producer = KafkaProducer(
    bootstrap_servers=[KAFKA_BROKER],
    value_serializer=lambda v: json.dumps(v).encode('utf-8')
)

consumer = KafkaConsumer(
    INPUT_TOPIC,
    bootstrap_servers=[KAFKA_BROKER],
    value_deserializer=lambda v: json.loads(v.decode('utf-8')),
    group_id='metadata_extraction_group',
    auto_offset_reset='earliest'
)

# Google Cloud Video Intelligence client
video_client = vi.VideoIntelligenceServiceClient()

def extract_video_metadata(video_path: str, video_id: str) -> Dict:
    """Extract technical metadata from video file"""
    try:
        import cv2
        cap = cv2.VideoCapture(video_path)
        if not cap.isOpened():
            raise Exception(f'Could not open {video_path}')

        metadata = {
            'video_id': video_id,
            'duration_seconds': cap.get(cv2.CAP_PROP_FRAME_COUNT) / cap.get(cv2.CAP_PROP_FPS),
            'width': int(cap.get(cv2.CAP_PROP_FRAME_WIDTH)),
            'height': int(cap.get(cv2.CAP_PROP_FRAME_HEIGHT)),
            'fps': cap.get(cv2.CAP_PROP_FPS),
            'codec': cap.get(cv2.CAP_PROP_FOURCC),
            'file_size_mb': os.path.getsize(video_path) / (1024 * 1024),
            'file_hash': hashlib.md5(open(video_path, 'rb').read()).hexdigest()
        }
        cap.release()
        return metadata
    except Exception as e:
        logger.error(f'Metadata extraction failed for {video_id}: {str(e)}')
        raise

def run_content_id_check(video_path: str, video_id: str) -> Dict:
    """
    Run Content ID check using Google Cloud Video Intelligence API.
    Matches YouTube's content ID system for copyright detection.
    """
    try:
        # Read video file as bytes
        with open(video_path, 'rb') as f:
            input_content = f.read()

        # Configure video intelligence request
        features = [vi.Feature.LOGO_RECOGNITION, vi.Feature.TEXT_DETECTION]
        request = vi.AnnotateVideoRequest(
            input_content=input_content,
            features=features,
            video_context=vi.VideoContext(
                segments=[vi.VideoSegment(start_time_offset=0, end_time_offset=300)]  # Check first 5 mins
            )
        )

        # Run synchronous annotation (use async for production scale)
        result = video_client.annotate_video(request=request)
        response = result.result()

        # Parse logo detection results
        logos = []
        for annotation in response.annotation_results[0].logo_recognition_annotations:
            logos.append({
                'name': annotation.entity.description,
                'confidence': annotation.tracks[0].confidence if annotation.tracks else 0.0
            })

        # Parse text detection results
        text = []
        for annotation in response.annotation_results[0].text_annotations:
            text.append({
                'text': annotation.text,
                'confidence': annotation.tracks[0].confidence if annotation.tracks else 0.0
            })

        return {
            'video_id': video_id,
            'content_id_matches': logos,
            'text_detections': text[:10],  # Limit to top 10 text detections
            'copyright_risk': 'high' if any(l['confidence'] > 0.9 for l in logos) else 'low'
        }
    except Exception as e:
        logger.error(f'Content ID check failed for {video_id}: {str(e)}')
        raise

def process_video_message(message: Dict) -> None:
    """Process single video upload message from Kafka"""
    video_id = message.get('video_id')
    video_path = message.get('video_path')
    logger.info(f'Processing video {video_id} at {video_path}')

    if not video_id or not video_path:
        logger.error('Missing video_id or video_path in message')
        producer.send(DLQ_TOPIC, value={'original': message, 'error': 'Missing fields'})
        return

    if not os.path.exists(video_path):
        logger.error(f'Video file {video_path} not found')
        producer.send(DLQ_TOPIC, value={'original': message, 'error': 'File not found'})
        return

    try:
        # Step 1: Extract technical metadata
        metadata = extract_video_metadata(video_path, video_id)
        # Step 2: Run Content ID check
        content_id = run_content_id_check(video_path, video_id)
        # Step 3: Combine results and send to output topic
        combined = {**metadata, **content_id, 'processed_at': str(datetime.datetime.now())}
        producer.send(OUTPUT_TOPIC, value=combined)
        logger.info(f'Successfully processed video {video_id}')
    except Exception as e:
        logger.error(f'Failed to process {video_id}: {str(e)}')
        producer.send(DLQ_TOPIC, value={'original': message, 'error': str(e)})

if __name__ == '__main__':
    logger.info(f'Starting metadata consumer on topic {INPUT_TOPIC}')
    for msg in consumer:
        try:
            process_video_message(msg.value)
        except Exception as e:
            logger.error(f'Consumer error: {str(e)}')

ffmpeg -vaapi_device /dev/dri/renderD128 -i input.mp4 -vf 'format=nv12,hwupload' -c:v hevc_vaapi -b:v 35M -s 3840x2160 -c:a aac -b:a 192k output_4k.mp4

@app.task(bind=True, max_retries=3, default_retry_delay=60, deadletter_queue='celery_dlq')
def transcode_video(self, input_path, output_dir):
    # Task logic here
    pass

addEventListener('fetch', event => {
  event.respondWith(handleRequest(event.request))
})

async function handleRequest(request) {
  const url = new URL(request.url)
  const videoId = url.searchParams.get('videoId')
  if (!videoId) return new Response('Missing videoId', { status: 400 })
  const s3Url = `https://my-s3-bucket.s3.amazonaws.com/videos/${videoId}/thumb_50s_1280x720.jpg`
  return Response.redirect(s3Url, 301)
}