core/video_editor.py

# @title core/video_editor.py
import os
import json
import asyncio
from typing import List, Dict, Any, Optional
from pathlib import Path
import subprocess
from moviepy.editor import VideoFileClip, AudioFileClip, concatenate_videoclips
from utils.logger import SpotMakerLogger

class VideoEditor:
    """
    Montażysta video który łączy wygenerowane klipy z audio w finalny spot.
    """
    def __init__(self, output_dir: str, logger: SpotMakerLogger):
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        self.logger = logger

    async def create_spot(self, video_paths: List[str], audio_path: str, audio_duration: float) -> str:
        """
        Tworzy finalny spot łącząc klipy video ze ścieżką audio.
        
        Args:
            video_paths: Lista ścieżek do klipów video
            audio_path: Ścieżka do pliku audio
            audio_duration: Długość ścieżki audio w sekundach
            
        Returns:
            Ścieżka do finalnego spotu
        """
        try:
            self.logger.log("video_editor", "Rozpoczynam montaż spotu", "info")
            self.logger.update_progress("video_editor", 0, "Inicjalizacja...")
            
            # Oblicz długość pojedynczego klipu
            clip_duration = audio_duration / len(video_paths)
            self.logger.log("video_editor", 
                          f"Długość pojedynczego klipu: {clip_duration:.2f}s", 
                          "debug")

            # Wczytaj i przygotuj klipy
            clips = []
            for i, path in enumerate(video_paths, 1):
                self.logger.update_progress("video_editor", 
                                         i * 20, 
                                         f"Przetwarzam klip {i}/{len(video_paths)}")
                
                self.logger.log("video_editor", f"Wczytuję klip: {path}", "debug")
                clip = VideoFileClip(path)
                
                # Dostosuj długość klipu
                if clip.duration != clip_duration:
                    clip = clip.set_duration(clip_duration)
                
                clips.append(clip)

            # Połącz klipy
            self.logger.update_progress("video_editor", 80, "Łączę klipy...")
            final_video = concatenate_videoclips(clips)
            
            # Dodaj ścieżkę audio
            self.logger.update_progress("video_editor", 90, "Dodaję audio...")
            audio = AudioFileClip(audio_path)
            final_video = final_video.set_audio(audio)
            
            # Zapisz finalny spot
            output_path = self.output_dir / "final_spot.mp4"
            self.logger.log("video_editor", 
                          f"Zapisuję finalny spot: {output_path}", 
                          "debug")
            
            final_video.write_videofile(
                str(output_path),
                codec='libx264',
                audio_codec='aac',
                temp_audiofile='temp-audio.m4a',
                remove_temp=True
            )
            
            # Zamknij klipy aby zwolnić zasoby
            final_video.close()
            audio.close()
            for clip in clips:
                clip.close()
            
            self.logger.update_progress("video_editor", 100, "Montaż zakończony")
            return str(output_path)
            
        except Exception as e:
            self.logger.format_error("video_editor", e)
            raise
            
    def get_status(self) -> Dict[str, Any]:
        """Zwraca aktualny status montażysty."""
        return {
            'progress': self.logger.get_module_progress("video_editor"),
            'status': self.logger.get_module_status("video_editor")
        }