forge/backend/app/utils/video.py

"""Video utility functions"""
import subprocess
import json
from typing import Optional, Dict, Any


def extract_video_metadata(file_path: str) -> Dict[str, Any]:
    """Extract video metadata using ffprobe

    Args:
        file_path: Path to the video file

    Returns:
        Dictionary containing:
        - duration_seconds: Video duration in seconds
        - width: Video width in pixels
        - height: Video height in pixels
        - fps: Frames per second
        - codec: Video codec name
        - bitrate: Video bitrate

    Returns empty dict if extraction fails.
    """
    try:
        result = subprocess.run([
            'ffprobe',
            '-v', 'quiet',
            '-print_format', 'json',
            '-show_format',
            '-show_streams',
            file_path
        ], capture_output=True, text=True, timeout=30)

        if result.returncode != 0:
            print(f"ffprobe failed with return code {result.returncode}")
            return {}

        data = json.loads(result.stdout)

        # Initialize metadata dict
        metadata = {
            'duration_seconds': None,
            'width': None,
            'height': None,
            'fps': None,
            'codec': None,
            'bitrate': None
        }

        # Get duration from format
        if 'format' in data and 'duration' in data['format']:
            metadata['duration_seconds'] = float(data['format']['duration'])

        if 'format' in data and 'bit_rate' in data['format']:
            metadata['bitrate'] = int(data['format']['bit_rate'])

        # Get video stream info
        for stream in data.get('streams', []):
            if stream.get('codec_type') == 'video':
                metadata['width'] = stream.get('width')
                metadata['height'] = stream.get('height')
                metadata['codec'] = stream.get('codec_name')

                # Calculate FPS from r_frame_rate
                if 'r_frame_rate' in stream:
                    try:
                        num, den = map(int, stream['r_frame_rate'].split('/'))
                        if den > 0:
                            metadata['fps'] = num / den
                    except (ValueError, ZeroDivisionError):
                        pass

                # Some videos have avg_frame_rate instead
                if not metadata['fps'] and 'avg_frame_rate' in stream:
                    try:
                        num, den = map(int, stream['avg_frame_rate'].split('/'))
                        if den > 0:
                            metadata['fps'] = num / den
                    except (ValueError, ZeroDivisionError):
                        pass

                break  # Use first video stream

        return metadata

    except subprocess.TimeoutExpired:
        print(f"ffprobe timed out for file: {file_path}")
        return {}
    except FileNotFoundError:
        print("ffprobe not found. Please ensure ffmpeg is installed.")
        return {}
    except json.JSONDecodeError:
        print(f"Failed to parse ffprobe output for file: {file_path}")
        return {}
    except Exception as e:
        print(f"Failed to extract video metadata: {e}")
        return {}


def format_duration(seconds: float) -> str:
    """Format duration in seconds to HH:MM:SS

    Args:
        seconds: Duration in seconds

    Returns:
        Formatted duration string
    """
    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = int(seconds % 60)

    if hours > 0:
        return f"{hours:02d}:{minutes:02d}:{secs:02d}"
    else:
        return f"{minutes:02d}:{secs:02d}"