video_processing.py

"""Video processing utilities for OutofLipSync"""

import json
import os
import subprocess
from fractions import Fraction
from ffmpy import FFmpeg, FFRuntimeError

from config import (
    YOUTUBE_VIDEO_PRESET,
    YOUTUBE_VIDEO_CRF,
    YOUTUBE_VIDEO_PROFILE,
    YOUTUBE_VIDEO_LEVEL,
    YOUTUBE_VIDEO_PIX_FMT,
    YOUTUBE_AUDIO_CODEC,
    YOUTUBE_AUDIO_BITRATE,
    YOUTUBE_AUDIO_SAMPLE_RATE,
)


# def crop_video_duration(video_path: str, duration: int, output_dir: str) -> str:
#     """Crop video to specified duration using FFmpeg (DEPRECATED - merged into normalize_video_for_youtube)
#
#     Args:
#         video_path: Path to input video
#         duration: Duration in seconds
#         output_dir: Directory to save cropped video
#
#     Returns:
#         Path to cropped video
#     """
#     cropped_video_path = os.path.join(output_dir, "input_cropped.mp4")
#     ffmpeg = FFmpeg(
#         inputs={video_path: None},
#         outputs={
#             cropped_video_path: [
#                 "-t",
#                 f"{duration}",
#                 "-c",
#                 "copy",
#                 "-loglevel",
#                 "error",
#                 "-y",
#             ]
#         },
#     )
#     try:
#         ffmpeg.run()
#     except FFRuntimeError as e:
#         raise Exception(f"FFmpeg failed: {e}")
#     return cropped_video_path


def loop_video(video_path: str, output_path: str, loop_count: int) -> str:
    """Loop video bằng stream_loop với copy codec

    Args:
        video_path: Path video gốc
        output_path: Path output video đã loop
        loop_count: Số lần loop

    Returns:
        Path video đã loop
    """
    ffmpeg = FFmpeg(
        inputs={video_path: ["-stream_loop", f"{loop_count}"]},
        outputs={
            output_path: [
                "-c",
                "copy",
                "-loglevel",
                "error",
                "-y",
            ]
        },
    )
    try:
        ffmpeg.run()
    except FFRuntimeError as e:
        raise Exception(f"FFmpeg failed to loop video: {e}")
    return output_path


def encode_video_for_youtube(
    video_path: str, output_path: str, duration: float | None = None
) -> str:
    """Encode video theo tiêu chuẩn YouTube

    Args:
        video_path: Path video input
        output_path: Path output video
        duration: Duration crop (None = không crop)

    Returns:
        Path video đã encode
    """
    ffmpeg_args = [
        "-c:v",
        "libx264",
        "-preset",
        YOUTUBE_VIDEO_PRESET,
        "-crf",
        str(YOUTUBE_VIDEO_CRF),
        "-profile:v",
        YOUTUBE_VIDEO_PROFILE,
        "-level",
        YOUTUBE_VIDEO_LEVEL,
        "-pix_fmt",
        YOUTUBE_VIDEO_PIX_FMT,
        "-c:a",
        "copy",
        "-movflags",
        "+faststart",
        "-loglevel",
        "error",
        "-y",
    ]

    if duration is not None:
        ffmpeg_args = ["-t", f"{duration}"] + ffmpeg_args

    ffmpeg = FFmpeg(
        inputs={video_path: None},
        outputs={output_path: ffmpeg_args},
    )
    try:
        ffmpeg.run()
    except FFRuntimeError as e:
        raise Exception(f"FFmpeg failed to encode video: {e}")
    return output_path


def normalize_video_for_youtube(
    video_path: str, audio_duration: float, output_dir: str
) -> str:
    """Chuẩn hóa video theo tiêu chuẩn YouTube

    - Loop nếu ngắn hơn audio, crop nếu dài hơn
    - Re-encode: libx264, preset slow, CRF 18, profile high, level 4.2, yuv420p

    Args:
        video_path: Path video gốc
        audio_duration: Duration audio (seconds)
        output_dir: Output directory

    Returns:
        Path video đã chuẩn hóa
    """
    video_info = get_video_info(video_path)
    video_duration = video_info["duration"]

    output_path = os.path.join(output_dir, "video_normalized.mp4")

    if video_duration >= audio_duration:
        encode_video_for_youtube(video_path, output_path, audio_duration)
    else:
        loop_count = int(audio_duration // video_duration) + 1
        temp_looped = os.path.join(output_dir, "video_looped_temp.mp4")

        loop_video(video_path, temp_looped, loop_count)
        encode_video_for_youtube(temp_looped, output_path, audio_duration)
        os.remove(temp_looped)

    return output_path


def merge_audio_video(video_path: str, audio_path: str, output_dir: str) -> str:
    """Merge video with audio track using FFmpeg with YouTube-optimized encoding

    Args:
        video_path: Path to input video
        audio_path: Path to audio track
        output_dir: Directory to save merged video

    Returns:
        Path to merged video
    """
    video_out = os.path.join(output_dir, "output_final.mp4")
    ffmpeg = FFmpeg(
        inputs={video_path: None, audio_path: None},
        outputs={
            video_out: [
                "-c:v",
                "libx264",
                "-preset",
                YOUTUBE_VIDEO_PRESET,
                "-crf",
                str(YOUTUBE_VIDEO_CRF),
                "-profile:v",
                YOUTUBE_VIDEO_PROFILE,
                "-level",
                YOUTUBE_VIDEO_LEVEL,
                "-pix_fmt",
                YOUTUBE_VIDEO_PIX_FMT,
                "-map",
                "0:v:0",
                "-map",
                "1:a:0",
                "-c:a",
                YOUTUBE_AUDIO_CODEC,
                "-b:a",
                YOUTUBE_AUDIO_BITRATE,
                "-ar",
                str(YOUTUBE_AUDIO_SAMPLE_RATE),
                "-shortest",
                "-movflags",
                "+faststart",
                "-loglevel",
                "error",
                "-y",
            ]
        },
    )

    try:
        ffmpeg.run()
    except FFRuntimeError as e:
        raise Exception(f"FFmpeg failed: {e}")
    return video_out


def get_video_info(video_path: str) -> dict:
    """Get video information: resolution, duration, fps

    Args:
        video_path: Path to video

    Returns:
        Dict with keys: width, height, duration, fps
    """
    cmd = [
        "ffprobe",
        "-v",
        "error",
        "-select_streams",
        "v:0",
        "-show_entries",
        "stream=width,height,r_frame_rate",
        "-show_entries",
        "format=duration",
        "-of",
        "json",
        video_path,
    ]
    result = subprocess.run(cmd, capture_output=True, text=True, check=True)
    data = json.loads(result.stdout)

    width = data["streams"][0]["width"]
    height = data["streams"][0]["height"]
    fps = float(Fraction(data["streams"][0]["r_frame_rate"]))
    duration = float(data["format"]["duration"])

    return {"width": width, "height": height, "fps": fps, "duration": duration}


def loop_video_to_match_audio(
    video_path: str, audio_duration: float, output_dir: str
) -> str:
    """Loop video to match audio target duration

    Args:
        video_path: Path to video source
        audio_duration: Audio target duration (seconds)
        output_dir: Output directory

    Returns:
        Path to looped/cropped video
    """
    video_info = get_video_info(video_path)
    video_duration = video_info["duration"]

    output_path = os.path.join(output_dir, "video_looped.mp4")

    if video_duration >= audio_duration:
        ffmpeg = FFmpeg(
            inputs={video_path: None},
            outputs={
                output_path: [
                    "-t",
                    f"{audio_duration}",
                    "-c",
                    "copy",
                    "-loglevel",
                    "error",
                    "-y",
                ]
            },
        )
    else:
        loop_count = int(audio_duration // video_duration) + 1
        ffmpeg = FFmpeg(
            inputs={video_path: ["-stream_loop", f"{loop_count}"]},
            outputs={
                output_path: [
                    "-t",
                    f"{audio_duration}",
                    "-c",
                    "copy",
                    "-loglevel",
                    "error",
                    "-y",
                ]
            },
        )
    try:
        ffmpeg.run()
    except FFRuntimeError as e:
        raise Exception(f"FFmpeg failed: {e}")
    return output_path