python_scripts/youtube_short.py

#!/usr/bin/env python3
import os
import json
import linux
import tempfile
from pathlib import Path
from moviepy import VideoFileClip, ColorClip, CompositeVideoClip
from moviepy.video.VideoClip import TextClip

from pathlib import Path
import numpy as np
from PIL import Image, ImageFilter

def get_video_info(input_path: Path) -> tuple:
    """Uses ffprobe to instantly read input video dimensions and frame rate."""
    cmd = f"ffprobe -v error -select_streams v:0 -show_entries stream=width,height,r_frame_rate -of json {input_path}"
    # Run command and capture output (assumes linux.run_command prints or you use subprocess)
    import subprocess
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    #linux.run_command(cmd)
    try:
        data = json.loads(result.stdout)
        stream = data['streams'][0]
        w = int(stream['width'])
        h = int(stream['height'])
        # Convert fractional FPS string (e.g. "60/1" or "30000/1001") to float
        fps_parts = stream['r_frame_rate'].split('/')
        fps = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) > 1 else float(fps_parts[0])
        return w, h, fps
    except Exception:
        return 1920, 1080, 60.0  # Safe defaults if probe fails

def fit_to_9_16_letterbox(input_path: str, top_text: str = "TOP TEXT", bottom_text: str = "BOTTOM TEXT", use_blur: bool = True, force: bool = False):
    threads = "8"
    input_file = Path(input_path)
    output_suffix = "gaussian_9_16" if use_blur else "black_9_16"
    output_path = input_file.parent / f"{input_file.stem}_{output_suffix}{input_file.suffix}"

    if os.path.exists(output_path):
        if not force:
            print(f"❌ Error: Short video already exists: {output_path}")
            return
        else:
            os.remove(output_path)

    # 1. Probe input metadata instantly
    orig_w, orig_h, fps = get_video_info(input_file)
    canvas_w = 1080
    canvas_h = 1920

    print("✍️ Generating text overlay graphics via MoviePy...")
    # Render static images for text instead of running a video context
    title_clip = TextClip(
        text=top_text, font_size=55, color="white", font="DejaVuSans-Bold",
        text_align="center", size=(canvas_w - 100, 300), method="caption"
    )
    bottom_clip = TextClip(
        text=bottom_text, font_size=55, color="white", font="DejaVuSans-Bold",
        text_align="center", size=(canvas_w - 100, 300), method="caption"
    )

    # Save text layers to temporary PNGs
    top_png = tempfile.NamedTemporaryFile(suffix=".png", delete=False).name
    bottom_png = tempfile.NamedTemporaryFile(suffix=".png", delete=False).name
    title_clip.save_frame(top_png)
    bottom_clip.save_frame(bottom_png)
    
    title_clip.close()
    bottom_clip.close()

    print("🎬 Dispatching compilation workload to FFmpeg filtergraph...")
    
    # 2. Build the complex FFmpeg filtergraph
    # [0:v] is the raw input video stream
    filter_complex = []

    if use_blur:
        # Scale height to 1920, crop center 1080x1920, apply fast boxblur (power of 3 approximates Gaussian)
        filter_complex.append(
            f"[0:v]scale=-1:{canvas_h},crop={canvas_w}:{canvas_h}:(iw-{canvas_w})/2:0,boxblur=luma_radius=35:luma_power=3[bg];"
        )
    else:
        # Generate a pure black background canvas matching video frame specs
        filter_complex.append(
            f"color=c=black:s={canvas_w}x{canvas_h}:r={fps}[bg];"
        )

    # Scale the foreground video to a clean 1080 width, keeping aspect ratio
    filter_complex.append(
        f"[0:v]scale={canvas_w}:-1[fg];"
    )

    # Layer composition chain:
    # Overlay 1: Put scaled foreground onto background (centered vertically)
    filter_complex.append(
        f"[bg][fg]overlay=0:(H-h)/2[tmp1];"
    )
    # Overlay 2: Drop top text asset onto position Y=180
    filter_complex.append(
        f"[tmp1][1:v]overlay=(W-w)/2:180[tmp2];"
    )
    # Overlay 3: Drop bottom text asset onto position Y=1430
    filter_complex.append(
        f"[tmp2][2:v]overlay=(W-w)/2:1430[finalv]"
    )

    filter_graph = "".join(filter_complex)

    # 3. Execute the native assembly command
    # -map_chapters -1 -sn: Strips unnecessary metadata chunks instantly
    # -c:a copy: Safely pulls original digital audio directly without decompression cycles
    # -threads 0: Forces FFmpeg to auto-consume all available processing cores
    ffmpeg_cmd = (
        f'ffmpeg -y -v error -i "{input_file}" -i "{top_png}" -i "{bottom_png}" '
        f'-filter_complex "{filter_graph}" '
        f'-map "[finalv]" -map 0:a? -c:v libx264 -crf 18 -preset slow -pix_fmt yuv420p '
        f'-c:a copy -map_chapters -1 -sn -threads {threads} "{output_path}"'
    )

    try:
        linux.run_command(ffmpeg_cmd)
        print(f"🎉 High-speed processing complete! Video saved to: {output_path}")
    finally:
        # Clean up temporary PNG picture files safely
        for path in (top_png, bottom_png):
            if os.path.exists(path):
                os.unlink(path)


if __name__ == "__main__":
    import time

    start_time = time.perf_counter()

    creator_name = "greenskiesbluegrass"
    top_txt = "TeamPGP Live on Twitch...\n Every Friday and Sunday @7:30 ET.\n twitch.tv/teampgp"
    bottom_txt = f"Clipped By: {creator_name}."
    fit_to_9_16_letterbox("download/clips/AbnegateAgitatedGrassPJSalt/AbnegateAgitatedGrassPJSalt.mp4", top_txt, bottom_txt, True, True)

    end_time = time.perf_counter()
    execution_time = end_time - start_time

    print(f"The function took {execution_time:.6f} seconds to complete.")