import os import subprocess import uuid import imageio_ffmpeg from typing import List, Optional, Tuple, Any # Import existing functions - avoid circular imports from ffmpeg_utils import extract_clip_ffmpeg, get_video_info_ffmpeg def should_use_ffmpeg_processing(timestamps, custom_dims, export_audio, bg_music, output_format, subtitle_path=None) -> bool: # FFmpeg can now handle background music mixing and all ShortsStyle! # Default to True to use FFmpeg optimization for most cases return True def process_single_clip_hybrid(video_path: str, start_time: float, end_time: float, clip_id: str, output_format: Any, custom_dims: Optional[Any] = None, export_audio: bool = True, bg_music: Optional[str] = None, aspect_ratio: Optional[Any] = None, subtitle_path: Optional[str] = None) -> str: """ Hybrid clip processing: FFmpeg for speed, MoviePy for features """ try: from schemas import ShortsStyle, AspectRatio from routers.video import PROCESSED_DIR # Generate output path in the PROCESSED folder output_path = os.path.join(PROCESSED_DIR, f"clip_{clip_id}.mp4") # Check if we can use FFmpeg for this clip can_use_ffmpeg = should_use_ffmpeg_processing( [type('obj', (object,), {'start_time': start_time, 'end_time': end_time})], custom_dims, export_audio, bg_music, output_format, subtitle_path=subtitle_path ) if can_use_ffmpeg: # Use FFmpeg for speed print(f"🚀 Using FFmpeg for clip {clip_id} (Style: {output_format})") # Map custom_dims or output_format to style style = output_format if isinstance(output_format, ShortsStyle) else None return extract_clip_ffmpeg( video_path, start_time, end_time, output_path, target_width=custom_dims.width if hasattr(custom_dims, 'width') else None, target_height=custom_dims.height if hasattr(custom_dims, 'height') else None, include_audio=True, # Video should always have audio if it exists style=style, aspect_ratio=aspect_ratio, bg_music_path=bg_music, video_volume=custom_dims.video_volume if hasattr(custom_dims, 'video_volume') else 1.0, music_volume=custom_dims.music_volume if hasattr(custom_dims, 'music_volume') else 0.2, loop_music=custom_dims.loop_music if hasattr(custom_dims, 'loop_music') else True, subtitle_path=subtitle_path ) else: # Use MoviePy for features (fallback) print(f"🎬 Using MoviePy for clip {clip_id}") # Import here to avoid circular imports from video_processor import process_single_clip as moviepy_process_single_clip return moviepy_process_single_clip(video_path, start_time, end_time, clip_id, output_format, custom_dims, export_audio, bg_music, subtitle_path=subtitle_path) except Exception as e: print(f"❌ Hybrid processing failed for clip {clip_id}: {e}") raise def process_video_hybrid(video_path: str, timestamps, output_format, custom_dims=None, export_audio=True, bg_music=None, aspect_ratio=None, subtitle_paths: Optional[List[str]] = None) -> tuple: """ Process video using hybrid approach (FFmpeg + MoviePy) Returns: (clip_paths, audio_paths) """ clip_paths = [] audio_paths = [] try: # Create temp directory os.makedirs("temp_videos", exist_ok=True) # Check if we can use FFmpeg for the entire video # Passing None for subtitle_path here as it's checked per-clip in process_single_clip_hybrid can_use_ffmpeg = should_use_ffmpeg_processing(timestamps, custom_dims, export_audio, bg_music, output_format) if can_use_ffmpeg: print(f"🚀 Using FFmpeg optimization for entire video (Style: {output_format})") # Process all clips with FFmpeg for i, ts in enumerate(timestamps): current_subtitle = subtitle_paths[i] if subtitle_paths and i < len(subtitle_paths) else None clip_path = process_single_clip_hybrid( video_path, ts.start_time, ts.end_time, f"hybrid_{i}_{uuid.uuid4().hex[:4]}", output_format, custom_dims, export_audio, bg_music, aspect_ratio=aspect_ratio, subtitle_path=current_subtitle ) clip_paths.append(clip_path) # If separate audio export is requested if export_audio: from routers.video import AUDIO_DIR audio_path = os.path.join(AUDIO_DIR, f"audio_hybrid_{i}_{uuid.uuid4().hex[:4]}.mp3") # Extract audio using FFmpeg cmd = [ imageio_ffmpeg.get_ffmpeg_exe(), '-i', clip_path, '-vn', '-acodec', 'libmp3lame', '-y', audio_path ] subprocess.run(cmd, check=True, capture_output=True) audio_paths.append(audio_path) else: print("🎬 Using MoviePy for complex processing") # Fallback to MoviePy for complex cases # This will be handled by the calling function return [], [] return clip_paths, audio_paths except Exception as e: print(f"❌ Hybrid video processing failed: {e}") # Clean up any partial results for path in clip_paths + audio_paths: if os.path.exists(path): os.remove(path) return [], []