Best for YouTube Blogging in 2026: For Every Budget

#!/usr/bin/env python3
"""
YouTube Video Processing Pipeline — Free Tier
Dependencies: pip install pydub moviepy yt-dlp
System deps: ffmpeg 6.0+ (brew install ffmpeg or apt install ffmpeg)

This script automates the full post-production chain:
  1. Audio normalization to -16 LUFS (YouTube standard)
  2. Subtitle burn-in from SRT file
  3. Intro/outro concatenation
  4. Thumbnail timestamp extraction
  5. Final H.265 encode at CRF 20
"""

import subprocess
import sys
import os
import json
import logging
from pathlib import Path
from dataclasses import dataclass
from typing import Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


@dataclass
class PipelineConfig:
    """Configuration for the video processing pipeline."""
    input_file: str
    output_file: str
    intro_file: Optional[str] = None
    outro_file: Optional[str] = None
    subtitle_file: Optional[str] = None
    thumbnail_timestamp: float = 10.0  # seconds
    target_lufs: float = -16.0
    crf: int = 20  # Constant Rate Factor — lower = better quality, bigger file
    preset: str = "slow"  # encoding preset: ultrafast, fast, medium, slow, veryslow
    resolution: str = "1920x1080"
    max_bitrate: str = "12M"
    temp_dir: str = "./tmp_pipeline"


def run_ffmpeg(args: list[str], description: str) -> bool:
    """
    Execute an FFmpeg command with full error handling.
    Returns True on success, False on failure.
    """
    cmd = ["ffmpeg", "-y", "-hide_banner"] + args
    logger.info(f"Running: {' '.join(cmd[:10])}...")

    try:
        result = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=7200  # 2-hour timeout for long renders
        )
        if result.returncode != 0:
            logger.error(f"{description} FAILED (exit {result.returncode})")
            logger.error(f"STDERR: {result.stderr[-2000:]}")
            return False
        logger.info(f"{description} completed successfully")
        return True
    except subprocess.TimeoutExpired:
        logger.error(f"{description} timed out after 2 hours")
        return False
    except FileNotFoundError:
        logger.error("ffmpeg not found. Install with: brew install ffmpeg")
        return False
    except Exception as e:
        logger.error(f"{description} raised {type(e).__name__}: {e}")
        return False


def normalize_audio(input_path: str, output_path: str, target_lufs: float) -> bool:
    """
    Normalize audio to target LUFS using FFmpeg's loudnorm filter.
    YouTube normalizes to -14 LUFS, so we target -16 to leave headroom.
    """
    # First pass: measure audio levels
    args = [
        "-i", input_path,
        "-af", f"loudnorm=I={target_lufs}:TP=-1.5:LRA=11:print_format=json",
        "-f", "null", "-"
    ]

    result = subprocess.run(
        ["ffmpeg", "-y", "-hide_banner"] + args,
        capture_output=True, text=True, timeout=3600
    )

    # Parse measured values from stderr (FFmpeg outputs JSON before the null sink)
    measured = {"input_i": -23.0, "input_tp": -2.0, "input_lra": 7.0, "input_thresh": -34.0}
    for line in result.stderr.split("\n"):
        if line.strip().startswith("{"):
            try:
                parsed = json.loads(line.strip())
                if "input_i" in parsed:
                    measured = parsed
                    break
            except json.JSONDecodeError:
                pass

    # Second pass: apply normalization with measured values
    args = [
        "-i", input_path,
        "-af", (
            f"loudnorm=I={target_lufs}:TP=-1.5:LRA=11:"
            f"measured_I={measured['input_i']}:"
            f"measured_TP={measured['input_tp']}:"
            f"measured_LRA={measured['input_lra']}:"
            f"measured_thresh={measured['input_thresh']}:"
            f"linear=true"
        ),
        "-c:v", "copy",
        "-c:a", "aac", "-b:a", "192k",
        output_path
    ]

    return run_ffmpeg(args, "Audio normalization")


def burn_subtitles(input_path: str, subtitle_path: str, output_path: str) -> bool:
    """
   Hardcode SRT subtitles into the video stream.
    Uses libass for proper styling and timing.
    """
    args = [
        "-i", input_path,
        "-vf", f"subtitles={subtitle_path}:force_style='FontName=Inter,FontSize=18,PrimaryColour=&HFFFFFF,OutlineColour=&H80000000,BorderStyle=3,Outline=2,Shadow=0,MarginV=40'",
        "-c:v", "libx265",
        "-crf", "20",
        "-preset", "slow",
        "-c:a", "copy",
        "-tag:v", "hvc1",  # Required for QuickTime / Safari compatibility
        output_path
    ]
    return run_ffmpeg(args, "Subtitle burn-in")


def concatenate_videos(parts: list[str], output_path: str) -> bool:
    """
    Concatenate multiple video files (intro + main + outro) without re-encoding.
    Requires matching codecs and parameters.
    """
    concat_list = "|".join(parts)
    args = [
        "-i", f"concat:{concat_list}",
        "-c", "copy",
        "-movflags", "+faststart",
        output_path
    ]
    return run_ffmpeg(args, "Video concatenation")


def extract_thumbnail(input_path: str, timestamp: float, output_path: str) -> bool:
    """
    Extract a single frame at the given timestamp for use as thumbnail.
    """
    args = [
        "-ss", str(timestamp),
        "-i", input_path,
        "-vframes", "1",
        "-q:v", "2",  # JPEG quality 2 (very high)
        "-f", "image2",
        output_path
    ]
    return run_ffmpeg(args, "Thumbnail extraction")


def youtube_encode(input_path: str, output_path: str, crf: int, preset: str) -> bool:
    """
    Final encode: H.265 at target CRF with YouTube-recommended settings.
    YouTube re-encodes everything, so we optimize for perceptual quality.
    """
    args = [
        "-i", input_path,
        "-c:v", "libx265",
        "-crf", str(crf),
        "-preset", preset,
        "-tune", "film",
        "-pix_fmt", "yuv420p",
        "-movflags", "+faststart",
        "-maxrate", "12M",
        "-bufsize", "24M",
        "-c:a", "aac",
        "-b:a", "192k",
        "-ac", "2",
        "-ar", "48000",
        "-tag:v", "hvc1",
        output_path
    ]
    return run_ffmpeg(args, "Final YouTube encode")


def run_pipeline(config: PipelineConfig) -> bool:
    """
    Execute the full video processing pipeline.
    Returns True if all stages complete successfully.
    """
    os.makedirs(config.temp_dir, exist_ok=True)

    stage_output = config.input_file

    # Stage 1: Normalize audio
    normalized = os.path.join(config.temp_dir, "normalized.mkv")
    if not normalize_audio(stage_output, normalized, config.target_lufs):
        logger.error("Pipeline failed at audio normalization")
        return False
    stage_output = normalized

    # Stage 2: Burn subtitles if provided
    if config.subtitle_file and os.path.exists(config.subtitle_file):
        subtitled = os.path.join(config.temp_dir, "subtitled.mkv")
        if not burn_subtitles(stage_output, config.subtitle_file, subtitled):
            logger.error("Pipeline failed at subtitle burn-in")
            return False
        stage_output = subtitled

    # Stage 3: Extract thumbnail
    if not extract_thumbnail(config.input_file, config.thumbnail_timestamp, 
                              os.path.join(config.temp_dir, "thumbnail.jpg")):
        logger.warning("Thumbnail extraction failed — continuing without thumbnail")

    # Stage 4: Final encode
    if not youtube_encode(stage_output, config.output_file, config.crf, config.preset):
        logger.error("Pipeline failed at final encode")
        return False

    logger.info(f"Pipeline complete: {config.output_file}")
    return True


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]}   [--intro FILE] [--outro FILE] [--srt FILE]")
        sys.exit(1)

    config = PipelineConfig(
        input_file=sys.argv[1],
        output_file=sys.argv[2]
    )

    # Parse optional arguments
    i = 3
    while i < len(sys.argv):
        if sys.argv[i] == "--intro" and i + 1 < len(sys.argv):
            config.intro_file = sys.argv[i + 1]
            i += 2
        elif sys.argv[i] == "--outro" and i + 1 < len(sys.argv):
            config.outro_file = sys.argv[i + 1]
            i += 2
        elif sys.argv[i] == "--srt" and i + 1 < len(sys.argv):
            config.subtitle_file = sys.argv[i + 1]
            i += 2
        else:
            i += 1

    success = run_pipeline(config)
    sys.exit(0 if success else 1)

#!/usr/bin/env python3
"""
AI-Powered Thumbnail Generator for YouTube
Dependencies: pip install openai pillow requests python-dotenv

Uses GPT-4o to analyze a video transcript and frame grabs,
then generates a composite thumbnail prompt for DALL-E 3.

Cost per thumbnail: ~$0.08 (DALL-E 3) + API calls
"""

import os
import json
import base64
import logging
import textwrap
from dataclasses import dataclass, field
from typing import List, Optional
from pathlib import Path
from PIL import Image, ImageDraw, ImageFont
import requests
from dotenv import load_dotenv

load_dotenv()

logger = logging.getLogger(__name__)

OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
YOUTUBE_THUMBNAIL_WIDTH = 1280
YOUTUBE_THUMBNAIL_HEIGHT = 720
MAX_TITLE_CHARS = 70  # YouTube truncates beyond this


@dataclass
class ThumbnailStyle:
    """Defines a repeatable thumbnail brand style."""
    name: str
    background_color: str  # hex
    accent_color: str      # hex
    font_path_bold: str
    font_path_regular: str
    title_font_size: int = 64
    badge_text: str = ""
    badge_color: str = "#FF0000"


# Pre-configured styles for different niches
STYLES = {
    "tech_tutorial": ThumbnailStyle(
        name="tech_tutorial",
        background_color="#0D1117",
        accent_color="#58A6FF",
        font_path_bold="./fonts/Inter-Bold.ttf",
        font_path_regular="./fonts/Inter-Regular.ttf",
        title_font_size=58,
        badge_text="TUTORIAL",
        badge_color="#FF7B72",
    ),
    "dev_opinion": ThumbnailStyle(
        name="dev_opinion",
        background_color="#1C1C1C",
        accent_color="#F0883E",
        font_path_bold="./fonts/Inter-Bold.ttf",
        font_path_regular="./fonts/Inter-Regular.ttf",
        title_font_size=52,
        badge_text="HOT TAKE",
        badge_color="#FF4D4D",
    ),
    "code_review": ThumbnailStyle(
        name="code_review",
        background_color="#0D1117",
        accent_color="#7EE787",
        font_path_bold="./fonts/Inter-Bold.ttf",
        font_path_regular="./fonts/Inter-Regular.ttf",
        title_font_size=56,
        badge_text="REVIEW",
        badge_color="#58A6FF",
    ),
}


def generate_thumbnail_prompt(transcript_excerpt: str, title: str, 
                               style: ThumbnailStyle) -> str:
    """
    Use GPT-4o to generate a detailed image prompt from transcript context.
    The prompt is designed for DALL-E 3 with specific aspect ratio and style constraints.
    """
    prompt_template = textwrap.dedent("""
    You are a YouTube thumbnail designer. Given a video title and a transcript
    excerpt, generate a detailed image prompt for DALL-E 3.

    Requirements:
    - Aspect ratio: 16:9 (1280x720)
    - Style: bold, high-contrast, clean composition suitable for a 120px-tall thumbnail
    - Include visual metaphor or key visual from the content
    - Use the accent color {accent_color} as a highlight
    - Background should be {bg_color}
    - The composition must leave space on the left third for text overlay
    - Format: return ONLY the prompt string, no preamble, no quotes
    """)

    user_message = prompt_template.format(
        accent_color=style.accent_color,
        bg_color=style.background_color
    ) + f"""
    Video Title: {title}
    Transcript Excerpt: {transcript_excerpt[:1500]}
    """

    try:
        response = requests.post(
            "https://api.openai.com/v1/chat/completions",
            headers={
                "Authorization": f"Bearer {OPENAI_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "model": "gpt-4o",
                "messages": [
                    {"role": "system", "content": "You are a thumbnail prompt engineer."},
                    {"role": "user", "content": user_message}
                ],
                "max_tokens": 200,
                "temperature": 0.7
            },
            timeout=30
        )
        response.raise_for_status()
        content = response.json()["choices"][0]["message"]["content"].strip()
        logger.info(f"Generated DALL-E prompt: {content[:80]}...")
        return content
    except requests.exceptions.RequestException as e:
        logger.error(f"GPT-4o API call failed: {e}")
        raise


def generate_image_with_dalle(prompt: str) -> bytes:
    """
    Generate an image using DALL-E 3 via OpenAI's image generation API.
    Returns raw PNG bytes.
    """
    try:
        response = requests.post(
            "https://api.openai.com/v1/images/generations",
            headers={
                "Authorization": f"Bearer {OPENAI_API_KEY}",
                "Content-Type": "application/json"
            },
            json={
                "model": "dall-e-3",
                "prompt": prompt,
                "size": "1792x1024",  # Closest to 16:9 available
                "quality": "standard",
                "n": 1
            },
            timeout=60
        )
        response.raise_for_status()
        image_url = response.json()["data"][0]["url"]

        # Download the generated image
        img_response = requests.get(image_url, timeout=30)
        img_response.raise_for_status()
        logger.info("DALL-E image generated successfully")
        return img_response.content
    except requests.exceptions.RequestException as e:
        logger.error(f"DALL-E generation failed: {e}")
        raise


def composite_thumbnail(background_bytes: bytes, title: str,
                         style: ThumbnailStyle) -> Image.Image:
    """
    Composite the AI-generated background with text overlays.
    Applies title text, badge, and accent bar following YouTube best practices.
    """
    # Load background and resize to exact YouTube thumbnail dimensions
    background = Image.open(io.BytesIO(background_bytes))
    background = background.resize(
        (YOUTUBE_THUMBNAIL_WIDTH, YOUTUBE_THUMBNAIL_HEIGHT),
        Image.LANCZOS
    )

    draw = ImageDraw.Draw(background)

    # Apply dark gradient on left side for text readability
    for x in range(0, 520):
        alpha = int(180 * (1 - x / 520))
        gradient_bar = Image.new("RGBA", (1, YOUTUBE_THUMBNAIL_HEIGHT),
                                  (0, 0, 0, alpha))
        background.paste(gradient_bar, (x, 0), gradient_bar)

    # Draw accent bar on left edge
    bar_width = 8
    draw.rectangle(
        [0, 0, bar_width, YOUTUBE_THUMBNAIL_HEIGHT],
        fill=style.accent_color
    )

    # Load fonts (fallback to default if custom fonts unavailable)
    try:
        title_font = ImageFont.truetype(style.font_path_bold, style.title_font_size)
        badge_font = ImageFont.truetype(style.font_path_bold, 28)
    except (IOError, OSError):
        logger.warning("Custom fonts not found, using default")
        title_font = ImageFont.load_default()
        badge_font = ImageFont.load_default()

    # Truncate title if too long
    display_title = title
    if len(title) > MAX_TITLE_CHARS:
        display_title = title[:MAX_TITLE_CHARS - 3] + "..."

    # Draw title text (word-wrapped to 2 lines max)
    words = display_title.split()
    lines = []
    current_line = ""
    for word in words:
        test_line = f"{current_line} {word}".strip()
        bbox = draw.textbbox((0, 0), test_line, font=title_font)
        if bbox[2] - bbox[0] > 500:
            lines.append(current_line)
            current_line = word
        else:
            current_line = test_line
    if current_line:
        lines.append(current_line)

    # Position text in left third with padding
    y_start = 200
    for i, line in enumerate(lines):
        bbox = draw.textbbox((0, 0), line, font=title_font)
        text_height = bbox[3] - bbox[1]
        draw.text(
            (50, y_start + i * (text_height + 10)),
            line,
            fill="#FFFFFF",
            font=title_font
        )

    # Draw badge if configured
    if style.badge_text:
        badge_bbox = draw.textbbox((0, 0), style.badge_text, font=badge_font)
        badge_w = badge_bbox[2] - badge_bbox[0] + 24
        badge_h = badge_bbox[3] - badge_bbox[1] + 12
        badge_x, badge_y = 50, 28

        # Rounded rectangle badge
        draw.rounded_rectangle(
            [badge_x, badge_y, badge_x + badge_w, badge_y + badge_h],
            radius=6,
            fill=style.badge_color
        )
        draw.text(
            (badge_x + 12, badge_y + 2),
            style.badge_text,
            fill="#FFFFFF",
            font=badge_font
        )

    return background


def save_thumbnail(image: Image.Image, output_path: str) -> str:
    """
    Save thumbnail as high-quality JPEG.
    YouTube recommends under 2MB; we target ~500KB at quality 92.
    """
    image.convert("RGB").save(output_path, "JPEG", quality=92, optimize=True)
    file_size_kb = os.path.getsize(output_path) / 1024
    logger.info(f"Thumbnail saved to {output_path} ({file_size_kb:.0f} KB)")
    return output_path


def generate_video_thumbnail(transcript_path: str, title: str,
                              output_path: str,
                              style_name: str = "tech_tutorial") -> Optional[str]:
    """
    Full pipeline: read transcript → generate AI prompt → create image → composite text.
    Returns path to saved thumbnail on success, None on failure.
    """
    style = STYLES.get(style_name)
    if not style:
        logger.error(f"Unknown style '{style_name}'. Available: {list(STYLES.keys())}")
        return None

    try:
        # Read transcript
        with open(transcript_path, "r", encoding="utf-8") as f:
            transcript = f.read()

        # Use middle section of transcript for context (avoids intro/outro fluff)
        mid_point = len(transcript) // 3
        excerpt = transcript[mid_point:mid_point + 1500]

        # Step 1: Generate AI image prompt
        prompt = generate_thumbnail_prompt(excerpt, title, style)

        # Step 2: Generate background image
        image_bytes = generate_image_with_dalle(prompt)

        # Step 3: Composite with text overlays
        final_image = composite_thumbnail(image_bytes, title, style)

        # Step 4: Save
        return save_thumbnail(final_image, output_path)

    except Exception as e:
        logger.error(f"Thumbnail generation failed: {e}")
        # Fallback: generate a simple text-based thumbnail
        logger.info("Falling back to text-only thumbnail")
        fallback = Image.new("RGB", (YOUTUBE_THUMBNAIL_WIDTH, YOUTUBE_THUMBNAIL_HEIGHT),
                             style.background_color)
        return save_thumbnail(fallback, output_path)


if __name__ == "__main__":
    import io

    result = generate_video_thumbnail(
        transcript_path="./transcript.txt",
        title="FFmpeg vs GStreamer: The Real Benchmarks (2026)",
        output_path="./thumbnail.jpg",
        style_name="tech_tutorial"
    )
    if result:
        print(f"Thumbnail generated: {result}")
    else:
        print("Thumbnail generation failed")
        sys.exit(1)

#!/usr/bin/env python3
"""
YouTube Batch Upload & Scheduling Automation
Dependencies: pip install google-api-python-client google-auth-oauthlib
              google-auth-httplib2 pydotenv

This script uploads videos to YouTube using the Data API v3 with
full metadata, chapters, and scheduled publish times.

Setup:
  1. Enable YouTube Data API v3 in Google Cloud Console
  2. Create OAuth 2.0 credentials (Desktop app type)
  3. Download credentials.json to ./auth/
  4. First run opens browser for auth; token.json is cached

Cost: $0 for API (within quota: 10,000 units/day, ~200 uploads)
"""

import os
import sys
import json
import time
import logging
import hashlib
from pathlib import Path
from datetime import datetime, timedelta, timezone
from dataclasses import dataclass, asdict
from typing import List, Optional, Dict, Any
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
from google_auth_oauthlib.flow import InstalledAppFlow
from google.auth.transport.requests import Request
from google.oauth2.credentials import Credentials
import dateutil.parser
from dotenv import load_dotenv

load_dotenv()

logger = logging.getLogger(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")

SCOPES = ["https://www.googleapis.com/auth/youtube.upload"]
CLIENT_SECRET_FILE = "./auth/credentials.json"
TOKEN_FILE = "./auth/token.json"
API_SERVICE_NAME = "youtube"
API_VERSION = "v3"


@dataclass
class VideoMetadata:
    """Complete metadata for a YouTube video upload."""
    file_path: str
    title: str
    description: str
    tags: List[str] = field(default_factory=list)
    category_id: str = "28"  # Science & Technology
    privacy_status: str = "private"  # or "public", "unlisted", "private"
    publish_at: Optional[str] = None  # ISO 8601 datetime for scheduling
    chapters: List[Dict[str, str]] = field(default_factory=list)
    # chapters: [{"title": "Intro", "start": "0:00"}, ...]
    thumbnail_path: Optional[str] = None
    made_for_kids: bool = False
    default_language: str = "en"


class YouTubeUploader:
    """Handles authentication and uploads to YouTube Data API v3."""

    def __init__(self, client_secret: str = CLIENT_SECRET_FILE,
                 token_file: str = TOKEN_FILE):
        self.client_secret = client_secret
        self.token_file = token_file
        self.service = None

    def authenticate(self) -> bool:
        """
        Authenticate with OAuth 2.0. Caches token to disk.
        Returns True on success.
        """
        creds = None

        # Load cached token
        if os.path.exists(self.token_file):
            try:
                creds = Credentials.from_authorized_user_file(
                    self.token_file, SCOPES
                )
                logger.info("Loaded cached credentials")
            except Exception as e:
                logger.warning(f"Failed to load token: {e}")

        # Refresh or initiate OAuth flow
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                try:
                    creds.refresh(Request())
                    logger.info("Refreshed OAuth token")
                except Exception as e:
                    logger.error(f"Token refresh failed: {e}")
                    creds = None
            else:
                flow = InstalledAppFlow.from_client_secrets_file(
                    self.client_secret, SCOPES
                )
                creds = flow.run_local_server(port=0)
                logger.info("Completed OAuth flow")

            # Save refreshed/new token
            with open(self.token_file, "w") as f:
                f.write(creds.to_json())

        try:
            self.service = build(
                API_SERVICE_NAME, API_VERSION,
                credentials=creds,
                cache_discovery=False
            )
            return True
        except Exception as e:
            logger.error(f"Failed to build YouTube service: {e}")
            return False

    def _build_chapter_description(self, chapters: List[Dict]) -> str:
        """
        Generate timestamped chapters for the video description.
        YouTube auto-detects chapters from this format.
        """
        lines = []
        for chapter in chapters:
            lines.append(f"{chapter['start']} - {chapter['title']}")
        return "\n".join(lines)

    def _append_chapters_to_description(self, existing: str,
                                          chapters: List[Dict]) -> str:
        """
        Append chapter timestamps to description, preserving existing content.
        """
        chapter_section = "\n\n--- CHAPTERS ---\n" + self._build_chapter_description(chapters)

        if "--- CHAPTERS ---" in existing:
            # Replace existing chapters section
            before = existing.split("--- CHAPTERS ---")[0]
            return before.rstrip() + chapter_section
        else:
            return existing.rstrip() + chapter_section

    def _compute_video_id(self, file_path: str) -> str:
        """Compute deterministic upload ID for idempotency."""
        return hashlib.sha256(file_path.encode()).hexdigest()[:16]

    def upload(self, metadata: VideoMetadata) -> Optional[str]:
        """
        Upload a video with full metadata. Returns video ID on success.

        Implements retry logic with exponential backoff for quota errors.
        """
        if not self.service:
            logger.error("Not authenticated. Call authenticate() first.")
            return None

        # Build request body
        body = {
            "snippet": {
                "title": metadata.title[:100],  # YouTube caps at 100 chars
                "description": metadata.description,
                "tags": metadata.tags[:50],  # YouTube caps at 500 chars total
                "categoryId": metadata.category_id,
                "defaultLanguage": metadata.default_language,
                "defaultAudioLanguage": metadata.default_language,
            },
            "status": {
                "privacyStatus": metadata.privacy_status,
                "selfDeclaredMadeForKids": metadata.made_for_kids,
            },
        }

        # Add publish schedule if specified
        if metadata.publish_at:
            body["status"]["publishAt"] = metadata.publish_at

        # Append chapters to description
        if metadata.chapters:
            body["snippet"]["description"] = self._append_chapters_to_description(
                body["snippet"]["description"],
                metadata.chapters
            )

        # Prepare media upload
        media = MediaFileUpload(
            metadata.file_path,
            chunksize=10 * 1024 * 1024,  # 10MB chunks for resumable upload
            resumable=True,
            mimetype="video/mp4"
        )

        max_retries = 3
        for attempt in range(max_retries):
            try:
                logger.info(f"Uploading {metadata.file_path} "
                          f"(attempt {attempt + 1}/{max_retries})")

                request = self.service.videos().insert(
                    part="snippet,status",
                    body=body,
                    media_body=media
                )

                response = None
                while response is None:
                    status, response = request.next_chunk(num_retries=5)
                    if status:
                        progress = int(status.progress() * 100)
                        logger.info(f"Upload progress: {progress}%")

                video_id = response["id"]
                logger.info(f"Upload successful! Video ID: {video_id}")

                # Set thumbnail if provided
                if metadata.thumbnail_path:
                    self._set_thumbnail(video_id, metadata.thumbnail_path)

                return video_id

            except Exception as e:
                error_msg = str(e)
                if "quota" in error_msg.lower() and attempt < max_retries - 1:
                    wait_seconds = 2 ** attempt * 60  # Exponential backoff
                    logger.warning(f"Quota exceeded. Retrying in {wait_seconds}s")
                    time.sleep(wait_seconds)
                else:
                    logger.error(f"Upload failed: {e}")
                    return None

        return None

    def _set_thumbnail(self, video_id: str, thumbnail_path: str) -> bool:
        """Set custom thumbnail for an uploaded video."""
        try:
            from googleapiclient.http import MediaFileUpload
            media = MediaFileUpload(thumbnail_path, mimetype="image/jpeg")
            self.service.thumbnails().set(
                videoId=video_id,
                media_body=media
            ).execute()
            logger.info(f"Thumbnail set for video {video_id}")
            return True
        except Exception as e:
            logger.error(f"Failed to set thumbnail: {e}")
            return False

    def schedule_batch(self, videos: List[VideoMetadata],
                       start_time: datetime,
                       gap_hours: int = 48) -> List[Dict[str, Any]]:
        """
        Schedule multiple videos with even spacing.
        Returns list of {title, video_id, publish_at} dicts.
        """
        results = []
        current_time = start_time

        for i, video in enumerate(videos):
            video.publish_at = current_time.isoformat()
            video_id = self.upload(video)

            if video_id:
                results.append({
                    "title": video.title,
                    "video_id": video_id,
                    "publish_at": current_time.isoformat()
                })
                logger.info(f"Scheduled '{video.title}' for {current_time}")

            current_time += timedelta(hours=gap_hours)

        return results


def load_video_manifest(manifest_path: str) -> List[VideoMetadata]:
    """
    Load video metadata from a JSON manifest file.
    Enables fully automated batch processing from a declarative config.
    """
    with open(manifest_path, "r", encoding="utf-8") as f:
        data = json.load(f)

    videos = []
    for entry in data["videos"]:
        metadata = VideoMetadata(
            file_path=entry["file"],
            title=entry["title"],
            description=entry.get("description", ""),
            tags=entry.get("tags", []),
            category_id=entry.get("category_id", "28"),
            privacy_status=entry.get("privacy_status", "private"),
            publish_at=entry.get("publish_at"),
            chapters=entry.get("chapters", []),
            thumbnail_path=entry.get("thumbnail"),
        )
        videos.append(metadata)

    return videos


if __name__ == "__main__":
    uploader = YouTubeUploader()

    if not uploader.authenticate():
        print("Authentication failed")
        sys.exit(1)

    # Upload a single video
    metadata = VideoMetadata(
        file_path="./output/final_video.mp4",
        title="FFmpeg vs GStreamer: The Real Benchmarks (2026)",
        description="A deep-dive comparison of FFmpeg and GStreamer for real-time video processing...",
        tags=["ffmpeg", "gstreamer", "video", "benchmarks", "linux"],
        chapters=[
            {"start": "0:00", "title": "Introduction"},
            {"start": "1:30", "title": "Test Methodology"},
            {"start": "4:00", "title": "Encoding Benchmarks"},
            {"start": "8:15", "title": "Latency Comparison"},
            {"start": "11:00", "title": "Verdict"},
        ],
        privacy_status="private",
        publish_at=datetime.now(timezone.utc).isoformat(),
    )

    video_id = uploader.upload(metadata)
    if video_id:
        print(f"Published: https://youtube.com/watch?v={video_id}")
    else:
        print("Upload failed")
        sys.exit(1)

#!/usr/bin/env python3
"""
Two-pass FFmpeg encoding with progress tracking and error handling.
Optimal for YouTube uploads where bitrate efficiency matters.
"""

import subprocess
import sys
import os
import re
import logging
from pathlib import Path

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)


def two_pass_encode(input_file: str, output_file: str,
                    crf: int = 20, preset: str = "medium",
                    video_codec: str = "libx265") -> bool:
    """
    Perform two-pass H.265 encoding for optimal quality-per-bit.

    Pass 1: Analyze video complexity, write stats file.
    Pass 2: Use stats to allocate bits efficiently.

    Args:
        input_file: Path to source video
        output_file: Path for encoded output
        crf: Constant Rate Factor (0=lossless, 51=worst, 20=good)
        preset: Encoding speed preset (ultrafast to veryslow)
        video_codec: Encoder to use (libx265 or h265_videotoolbox)

    Returns:
        True if both passes succeed, False otherwise
    """

    if not os.path.exists(input_file):
        logger.error(f"Input file not found: {input_file}")
        return False

    pass_log_file = "ffmpeg2pass-0.log"

    # Clean up stale pass log
    if os.path.exists(pass_log_file):
        os.remove(pass_log_file)

    # --- Pass 1: Analysis ---
    pass1_args = [
        "ffmpeg", "-y",
        "-i", input_file,
        "-c:v", video_codec,
        "-preset", preset,
        "-b:v", "0",  # Let CRF control quality
        "-crf", str(crf),
        "-pass", "1",
        "-an",  # No audio in video pass
        "-f", "null",  # Discard output
        "/dev/null" if sys.platform != "win32" else "NUL"
    ]

    logger.info(f"Pass 1/2: Analyzing {input_file}")

    try:
        result = subprocess.run(
            pass1_args,
            capture_output=True,
            text=True,
            timeout=14400  # 4-hour timeout
        )

        if result.returncode != 0:
            logger.error(f"Pass 1 failed with code {result.returncode}")
            logger.error(f"STDERR: {result.stderr[-1500:]}")
            return False

        logger.info("Pass 1 complete")

    except subprocess.TimeoutExpired:
        logger.error("Pass 1 timed out")
        return False
    except FileNotFoundError:
        logger.error("ffmpeg not found in PATH")
        return False

    # --- Pass 2: Actual encoding ---
    pass2_args = [
        "ffmpeg", "-y",
        "-i", input_file,
        "-c:v", video_codec,
        "-preset", preset,
        "-b:v", "0",
        "-crf", str(crf),
        "-pass", "2",
        "-c:a", "aac", "-b:a", "192k",
        "-movflags", "+faststart",
        "-pix_fmt", "yuv420p",
        "-tag:v", "hvc1",
        output_file
    ]

    logger.info(f"Pass 2/2: Encoding to {output_file}")

    try:
        result = subprocess.run(
            pass2_args,
            capture_output=True,
            text=True,
            timeout=14400
        )

        if result.returncode != 0:
            logger.error(f"Pass 2 failed with code {result.returncode}")
            logger.error(f"STDERR: {result.stderr[-1500:]}")
            return False

        # Extract final stats from stderr
        for line in result.stderr.split("\n"):
            if "speed=" in line and "fps=" in line:
                logger.info(f"Encoding stats: {line.strip()}")
                break

        logger.info(f"Two-pass encoding complete: {output_file}")
        return True

    except subprocess.TimeoutExpired:
        logger.error("Pass 2 timed out")
        return False


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print(f"Usage: {sys.argv[0]}  ")
        sys.exit(1)

    success = two_pass_encode(sys.argv[1], sys.argv[2])
    sys.exit(0 if success else 1)

#!/usr/bin/env python3
"""
Automated chapter generation using OpenAI Whisper with diarization.
Dependencies: pip install openai-whisper torch pyannote.audio
              pip install https://github.com/pyannote/pyannote-audio
License: pyannote.audio requires accepting the license at
         https://huggingface.co/pyannote/speaker-diarization-3.1
"""

import whisper
import json
import logging
import subprocess
from pathlib import Path
from dataclasses import dataclass, field
from typing import List
import torch
import numpy as np

logger = logging.getLogger(__name__)


@dataclass
class TranscriptSegment:
    """A single segment from Whisper transcription."""
    start: float    # seconds
    end: float      # seconds
    text: str
    speaker: str = "unknown"


@dataclass 
class Chapter:
    """A video chapter with start time and title."""
    start_seconds: float
    title: str

    @property
    def timestamp(self) -> str:
        """Format as HH:MM:SS for YouTube chapters."""
        hours = int(self.start_seconds // 3600)
        minutes = int((self.start_seconds % 3600) // 60)
        seconds = int(self.start_seconds % 60)
        return f"{hours:02d}:{minutes:02d}:{seconds:02d}"


class ChapterGenerator:
    """
    Generates YouTube chapters from audio using Whisper + diarization.
    """

    def __init__(self, whisper_model: str = "medium"):
        """
        Args:
            whisper_model: Whisper model size. 'medium' is the sweet spot
                           for chapters — 3x faster than 'large' with
                           comparable timestamp accuracy.
        """
        logger.info(f"Loading Whisper model: {whisper_model}")
        self.model = whisper.load_model(whisper_model)
        self.diarizer = None

    def _load_diarization_model(self):
        """Lazily load diarization model (requires ~2GB VRAM)."""
        if self.diarizer is None:
            try:
                from pyannote.audio import Pipeline
                self.diarizer = Pipeline.from_pretrained(
                    "pyannote/speaker-diarization-3.1",
                    use_auth_token=os.environ.get("HF_TOKEN", "")
                )
                self.diarizer.to(torch.device("cuda" if torch.cuda.is_available() else "cpu"))
                logger.info("Diarization model loaded")
            except ImportError:
                logger.warning("pyannote.audio not installed — skipping diarization")
                self.diarizer = False  # Sentinel: tried and unavailable

    def transcribe(self, audio_path: str) -> dict:
        """Run Whisper transcription."""
        logger.info(f"Transcribing: {audio_path}")
        result = self.model.transcribe(
            audio_path,
            word_timestamps=True,
            verbose=False
        )
        return result

    def diarize(self, audio_path: str) -> List[dict]:
        """Run speaker diarization. Returns list of {'start', 'end', 'speaker'}."""
        self._load_diarization_model()

        if self.diarizer is False:
            return []  # Diarization unavailable

        from pyannote.core import Segment
        diarization = self.diarizer(audio_path)

        speakers = {}
        speaker_count = 0
        segments = []

        for turn, _, speaker in diarization.itertracks(yield_label=True):
            if speaker not in speakers:
                speaker_count += 1
                speakers[speaker] = f"Speaker {speaker_count}"

            segments.append({
                "start": turn.start,
                "end": turn.end,
                "speaker": speakers[speaker]
            })

        logger.info(f"Diarization found {speaker_count} speakers")
        return segments

    def _merge_speakers(self, segments: List[TranscriptSegment]) -> List[TranscriptSegment]:
        """Merge consecutive segments from the same speaker."""
        if not segments:
            return []

        merged = [segments[0]]
        for seg in segments[1:]:
            if seg.speaker == merged[-1].speaker:
                merged[-1].end = seg.end
                merged[-1].text += " " + seg.text
            else:
                merged.append(seg)
        return merged

    def _find_chapter_boundaries(self, segments: List[TranscriptSegment],
                                  min_duration: float = 30.0,
                                  max_duration: float = 300.0) -> List[Chapter]:
        """
        Identify chapter boundaries based on topic shifts.

        Strategy: A new chapter starts when:
          1. Speaker changes AND the current segment > min_duration
          2. A gap > 2 seconds exists between segments
          3. Text contains strong topic-shift keywords
        """
        TOPIC_KEYWORDS = {
            "introduction", "overview", "agenda",
            "first", "next", "finally", "conclusion",
            "summary", "recap", "demo", "example",
            "setup", "installation", "configuration",
            "benchmark", "result", "comparison"
        }

        chapters = [Chapter(0.0, "Introduction")]
        current_start = 0.0

        for i, seg in enumerate(segments):
            duration = seg.end - current_start
            words = set(seg.text.lower().split())

            should_split = False
            reason = ""

            # Check duration bounds
            if duration > max_duration:
                should_split = True
                reason = "max_duration"

            # Check for speaker change (after minimum segment)
            if i > 0 and seg.speaker != segments[i-1].speaker and duration > min_duration:
                should_split = True
                reason = "speaker_change"

            # Check for topic keywords at segment start
            if words & TOPIC_KEYWORDS and duration > min_duration and i > 0:
                should_split = True
                reason = "topic_shift"

            if should_split and chapters[-1].start_seconds != current_start:
                # Use first sentence as chapter title
                title = seg.text.split(".")[0].strip()[:60]
                if not title:
                    title = f"Section {len(chapters) + 1}"
                chapters.append(Chapter(current_start, title))
                current_start = seg.start

        return chapters

    def generate(self, audio_path: str, speaker_names: dict = None) -> List[Chapter]:
        """
        Full pipeline: transcribe → diarize → identify chapters.

        Args:
            audio_path: Path to audio file (MP3, WAV, M4A)
            speaker_names: Optional dict mapping 'Speaker 1' -> 'Alice'

        Returns:
            List of Chapter objects with timestamps and titles
        """
        # Transcribe
        result = self.transcribe(audio_path)

        # Build segments
        segments = []
        for seg in result["segments"]:
            segments.append(TranscriptSegment(
                start=seg["start"],
                end=seg["end"],
                text=seg["text"].strip()
            ))

        # Diarize and assign speakers
        diarization = self.diarize(audio_path)
        if diarization:
            for seg in segments:
                for diar_seg in diarization:
                    if diar_seg["start"] <= seg.start < diar_seg["end"]:
                        seg.speaker = diar_seg["speaker"]
                        break

        # Merge same-speaker segments
        segments = self._merge_speakers(segments)

        # Apply custom speaker names
        if speaker_names:
            for seg in segments:
                if seg.speaker in speaker_names:
                    seg.speaker = speaker_names[seg.speaker]

        # Find chapter boundaries
        chapters = self._find_chapter_boundaries(segments)

        logger.info(f"Generated {len(chapters)} chapters")
        for ch in chapters:
            logger.info(f"  {ch.timestamp} — {ch.title}")

        return chapters

    def format_for_youtube(self, chapters: List[Chapter]) -> str:
        """Format chapters as YouTube-compatible text."""
        lines = []
        for ch in chapters:
            lines.append(f"{ch.timestamp} {ch.title}")
        return "\n".join(lines)


if __name__ == "__main__":
    import sys

    if len(sys.argv) < 2:
        print(f"Usage: {sys.argv[0]}  [--names Speaker1:Alice]")
        sys.exit(1)

    generator = ChapterGenerator(whisper_model="medium")

    # Parse optional speaker name mappings
    speaker_names = {}
    for arg in sys.argv[2:]:
        if arg.startswith("--names"):
            continue
        if ":" in arg:
            key, val = arg.split(":", 1)
            speaker_names[key] = val

    chapters = generator.generate(sys.argv[1], speaker_names or None)
    print("\nYouTube Chapters:\n")
    print(generator.format_for_youtube(chapters))

#!/usr/bin/env python3
"""
GitHub Actions-compatible video processing & upload script.
Designed to run on ubuntu-latest runners.

Usage in .github/workflows/youtube-pipeline.yml:
  on:
    workflow_dispatch:
      inputs:
        video_file: ...
  OR
    s3-event trigger (via S3 → SNS → SQS → Actions)

This script handles the CI/CD portion of the pipeline.
"""

import os
import sys
import json
import subprocess
import logging
import boto3
from pathlib import Path
from datetime import datetime, timezone
from dataclasses import dataclass
from typing import Optional
import requests

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s"
)
logger = logging.getLogger(__name__)


@dataclass
class PipelineResult:
    """Result of a pipeline execution."""
    success: bool
    output_file: Optional[str] = None
    video_id: Optional[str] = None
    error: Optional[str] = None
    duration_seconds: float = 0
    file_size_mb: float = 0


class VideoPipeline:
    """
    Complete video processing pipeline for YouTube.
    Runs on GitHub Actions or any Linux environment with FFmpeg.
    """

    def __init__(self, config: dict):
        self.config = config
        self.s3_client = None
        self.work_dir = Path(config.get("work_dir", "/tmp/video-pipeline"))
        self.work_dir.mkdir(parents=True, exist_ok=True)

    def _run(self, args: list, description: str, timeout: int = 7200) -> tuple[bool, str]:
        """Run a shell command and return (success, stderr)."""
        cmd = ["ffmpeg", "-y"] + args
        logger.info(f"{description}: {' '.join(cmd[:8])}...")
        try:
            result = subprocess.run(
                cmd,
                capture_output=True,
                text=True,
                timeout=timeout,
                cwd=str(self.work_dir)
            )
            if result.returncode != 0:
                return False, result.stderr[-2000:]
            return True, ""
        except subprocess.TimeoutExpired:
            return False, "Command timed out"
        except FileNotFoundError:
            return False, "ffmpeg not found — install with: apt-get install ffmpeg"

    def download_from_s3(self, bucket: str, key: str) -> Optional[str]:
        """Download source file from S3 using OIDC credentials."""
        if not self.s3_client:
            self.s3_client = boto3.client("s3")

        local_path = self.work_dir / Path(key).name
        try:
            logger.info(f"Downloading s3://{bucket}/{key}")
            self.s3_client.download_file(bucket, key, str(local_path))
            size_mb = local_path.stat().st_size / (1024 * 1024)
            logger.info(f"Downloaded: {size_mb:.1f} MB")
            return str(local_path)
        except Exception as e:
            logger.error(f"S3 download failed: {e}")
            return None

    def process(self, input_path: str) -> PipelineResult:
        """
        Execute the full pipeline:
        1. Audio normalization
        2. Scale to 1080p
        3. Two-pass H.265 encode
        4. Move faststart
        """
        result = PipelineResult(success=False)
        start_time = datetime.now(timezone.utc)

        # Stage 1: Audio normalization
        normalized = self.work_dir / "step1_normalized.mkv"
        ok, err = self._run([
            "-i", input_path,
            "-af", "loudnorm=I=-16:TP=-1.5:LRA=11",
            "-c:v", "copy",
            "-c:a", "aac", "-b:a", "192k",
            str(normalized)
        ], "Audio normalization")
        if not ok:
            result.error = f"Audio normalization failed: {err}"
            return result

        # Stage 2: Scale and encode
        scaled = self.work_dir / "step2_scaled.mkv"
        ok, err = self._run([
            "-i", str(normalized),
            "-vf", "scale='min(1920,iw)':'min(1080,ih)':force_original_aspect_ratio=decrease,pad=1920:1080:(ow-iw)/2:(oh-ih)/2",
            "-c:v", "libx265",
            "-crf", "20",
            "-preset", "medium",
            "-tune", "film",
            "-c:a", "copy",
            str(scaled)
        ], "Scaling & encoding", timeout=14400)
        if not ok:
            result.error = f"Encoding failed: {err}"
            return result

        # Stage 3: Final output with faststart
        final = self.work_dir / "final_video.mp4"
        ok, err = self._run([
            "-i", str(scaled),
            "-c", "copy",
            "-movflags", "+faststart",
            str(final)
        ], "Final mux with faststart")
        if not ok:
            result.error = f"Final mux failed: {err}"
            return result

        result.success = True
        result.output_file = str(final)
        result.duration_seconds = (datetime.now(timezone.utc) - start_time).total_seconds()
        result.file_size_mb = final.stat().st_size / (1024 * 1024)

        logger.info(f"Pipeline complete in {result.duration_seconds:.0f}s, "
                    f"output: {result.file_size_mb:.1f} MB")
        return result

    def upload_to_youtube(self, video_path: str,
                          title: str, description: str) -> Optional[str]:
        """Upload processed video to YouTube via Data API v3."""
        # Implementation uses google-api-python-client (see uploader script above)
        # Returns video ID or None
        logger.info(f"Uploading to YouTube: {title}")
        # ... authentication and upload logic from the uploader script
        return None  # Placeholder — full code in uploader section

    def notify_slack(self, webhook_url: str, result: PipelineResult):
        """Send pipeline result to Slack channel."""
        if not webhook_url:
            return

        status = "✅" if result.success else "❌"
        payload = {
            "text": f"{status} Video Pipeline {'Succeeded' if result.success else 'Failed'}",
            "blocks": [
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"{status} *YouTube Pipeline* {'complete' if result.success else 'failed'}\n"
                                f"File: `{result.output_file or 'N/A'}`\n"
                                f"Size: {result.file_size_mb:.1f} MB\n"
                                f"Time: {result.duration_seconds:.0f}s\n"
                                f"Error: {result.error or 'None'}"
                    }
                }
            ]
        }

        try:
            requests.post(webhook_url, json=payload, timeout=10)
        except requests.exceptions.RequestException as e:
            logger.warning(f"Slack notification failed: {e}")


if __name__ == "__main__":
    config = {
        "work_dir": "/tmp/video-pipeline",
    }

    pipeline = VideoPipeline(config)

    input_file = sys.argv[1] if len(sys.argv) > 1 else None
    if not input_file:
        print("Usage: pipeline.py ")
        sys.exit(1)

    result = pipeline.process(input_file)

    if result.success:
        print(f"Output: {result.output_file}")
        print(f"Processing time: {result.duration_seconds:.0f}s")
        print(f"File size: {result.file_size_mb:.1f} MB")
    else:
        print(f"Pipeline failed: {result.error}")
        sys.exit(1)