import subprocess import random from pathlib import Path from moviepy.editor import VideoFileClip from pydub import AudioSegment class HorizontalVideoGenerator: def __init__(self, source_folder, temp_folder, preview_generator, resolution=(1920, 1080), fade_duration=1): """ Ініціалізація генератора горизонтальних відео Args: source_folder: Папка з фоновими відео temp_folder: Папка для тимчасових файлів preview_generator: Об'єкт PreviewGenerator для створення превью resolution: Розмір відео (ширина, висота) fade_duration: Тривалість fade ефектів """ self.source_folder = Path(source_folder) self.temp_folder = Path(temp_folder) self.preview_generator = preview_generator self.resolution = resolution self.fade_duration = fade_duration def apply_safe_fade_filters(self, use_dur, base_filters): """Застосовує fade фільтри безпечно, уникаючи негативних значень""" fade_in_duration = min(0.3, use_dur * 0.3) # Максимум 30% від тривалості кліпу fade_out_duration = min(0.3, use_dur * 0.3) # Переконуємося, що fade_in + fade_out не перевищує тривалість кліпу if fade_in_duration + fade_out_duration > use_dur: fade_in_duration = use_dur * 0.4 fade_out_duration = use_dur * 0.4 # Розраховуємо точку початку fade-out fade_out_start = max(0, use_dur - fade_out_duration) # Створюємо fade фільтри тільки якщо вони мають сенс fade_filters = [] if fade_in_duration > 0.05: # Мінімум 0.05 секунди для fade-in fade_filters.append(f"fade=t=in:st=0:d={fade_in_duration}") if fade_out_duration > 0.05 and fade_out_start > fade_in_duration: # Мінімум 0.05 секунди для fade-out fade_filters.append(f"fade=t=out:st={fade_out_start}:d={fade_out_duration}") # Об'єднуємо всі фільтри all_filters = [base_filters] + fade_filters return ",".join(all_filters) def create_preview_video_segments(self, story_id, preview_duration): """ Створює фонові відео сегменти для preview частини Args: story_id: ID історії для іменування файлів preview_duration: Тривалість preview в секундах Returns: Path: Шлях до об'єднаного preview відео """ bg_videos = list(self.source_folder.glob("*.mp4")) random.shuffle(bg_videos) bg_iter = iter(bg_videos) # Створюємо достатньо фонових відео для покриття всієї preview тривалості preview_clips = [] preview_duration_collected = 0 while preview_duration_collected < preview_duration: try: bg_video = next(bg_iter) except StopIteration: # Якщо відео закінчились, починаємо знову bg_videos_reset = list(self.source_folder.glob("*.mp4")) random.shuffle(bg_videos_reset) bg_iter = iter(bg_videos_reset) bg_video = next(bg_iter) clip = VideoFileClip(str(bg_video)) remaining_preview_duration = preview_duration - preview_duration_collected use_duration = min(clip.duration, remaining_preview_duration) clip.close() if use_duration <= 0: break preview_clip_tmp = self.temp_folder / f"{story_id}_preview_bg_{len(preview_clips):02d}.mp4" # Створюємо кліп з потрібною тривалістю subprocess.run([ "ffmpeg", "-y", "-i", str(bg_video), "-ss", "0", "-t", str(use_duration), "-vf", f"scale={self.resolution[0]}:{self.resolution[1]},fps=30", "-an", "-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "5M", str(preview_clip_tmp) ], check=True) preview_clips.append(preview_clip_tmp) preview_duration_collected += use_duration # Об'єднуємо всі preview кліпи if len(preview_clips) > 1: preview_concat_list = self.temp_folder / f"{story_id}_preview_concat.txt" with open(preview_concat_list, "w", encoding="utf-8") as f: for clip in preview_clips: f.write(f"file '{clip.as_posix()}'\n") preview_combined = self.temp_folder / f"{story_id}_preview_combined.mp4" subprocess.run([ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(preview_concat_list), "-c:v", "h264_nvenc", "-b:v", "5M", "-preset", "p4", str(preview_combined) ], check=True) else: preview_combined = preview_clips[0] return preview_combined, bg_iter def create_preview_with_overlay(self, story_id, preview_combined, preview_img_path, preview_duration): """ Накладає preview зображення на фонове відео Args: story_id: ID історії preview_combined: Шлях до об'єднаного фонового відео preview_img_path: Шлях до preview зображення preview_duration: Тривалість preview Returns: Path: Шлях до preview відео з накладеним зображенням """ preview_video = self.temp_folder / f"{story_id}_preview_video.mp4" subprocess.run([ "ffmpeg", "-y", "-i", str(preview_combined), "-i", str(preview_img_path), "-filter_complex", "[1:v]format=rgba,scale=iw*0.7:ih*0.7[overlay];[0:v][overlay]overlay=(W-w)/2:(H-h)/2:format=auto", "-t", str(preview_duration), "-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "5M", str(preview_video) ], check=True) return preview_video def create_story_video_segments(self, story_id, bg_iter, target_duration, duration_collected): """ Створює відео сегменти для основної частини історії Args: story_id: ID історії bg_iter: Ітератор фонових відео target_duration: Цільова тривалість всього відео duration_collected: Вже зібрана тривалість Returns: list: Список шляхів до створених відео сегментів """ used_clips = [] # Продовжуємо з того місця де зупинились в bg_iter для story частини for idx, bg in enumerate(bg_iter): clip = VideoFileClip(str(bg)) use_dur = min(clip.duration, target_duration - duration_collected) clip.close() if use_dur <= 0: break temp_out = self.temp_folder / f"clip_{idx:02d}.mp4" # Використовуємо безпечну функцію для fade фільтрів base_filters = f"scale={self.resolution[0]}:{self.resolution[1]},fps=30" filter_string = self.apply_safe_fade_filters(use_dur, base_filters) subprocess.run([ "ffmpeg", "-y", "-i", str(bg), "-ss", "0", "-t", str(use_dur), "-vf", filter_string, "-an", "-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "5M", str(temp_out) ], check=True) used_clips.append(temp_out) duration_collected += use_dur if duration_collected >= target_duration: break # Якщо все ще не достатньо відео, додаємо ще if duration_collected < target_duration: bg_videos_extra = list(self.source_folder.glob("*.mp4")) random.shuffle(bg_videos_extra) for idx, bg in enumerate(bg_videos_extra): if duration_collected >= target_duration: break clip = VideoFileClip(str(bg)) use_dur = min(clip.duration, target_duration - duration_collected) clip.close() if use_dur <= 0: break temp_out = self.temp_folder / f"clip_extra_{idx:02d}.mp4" # Використовуємо ту ж безпечну функцію для fade base_filters = f"scale={self.resolution[0]}:{self.resolution[1]},fps=30" filter_string = self.apply_safe_fade_filters(use_dur, base_filters) subprocess.run([ "ffmpeg", "-y", "-i", str(bg), "-ss", "0", "-t", str(use_dur), "-vf", filter_string, "-an", "-c:v", "h264_nvenc", "-preset", "p4", "-b:v", "5M", str(temp_out) ], check=True) used_clips.append(temp_out) duration_collected += use_dur return used_clips def combine_video_segments(self, story_id, preview_video, story_clips): """ Об'єднує всі відео сегменти в одне відео Args: story_id: ID історії preview_video: Шлях до preview відео story_clips: Список шляхів до story кліпів Returns: Path: Шлях до об'єднаного відео """ all_clips = [preview_video] + story_clips concat_list = self.temp_folder / f"{story_id}_concat.txt" with open(concat_list, "w", encoding="utf-8") as f: for clip in all_clips: f.write(f"file '{clip.as_posix()}'\n") combined_path = self.temp_folder / f"{story_id}_combined.mp4" subprocess.run([ "ffmpeg", "-y", "-f", "concat", "-safe", "0", "-i", str(concat_list), "-c:v", "h264_nvenc", "-b:v", "5M", "-preset", "p4", str(combined_path) ], check=True) return combined_path def create_horizontal_video(self, story_id, preview_txt, audio_preview, audio_story, selected_template): """ Створює повне горизонтальне відео Args: story_id: ID історії preview_txt: Текст preview audio_preview: AudioSegment з preview аудіо audio_story: AudioSegment з story аудіо selected_template: Назва вибраного template Returns: Path: Шлях до створеного відео """ print(f"????️ Створення preview зображення з template: {selected_template}...") preview_img_path = self.temp_folder / f"{story_id}_preview.png" self.preview_generator.draw_horizontal_preview(preview_txt, preview_img_path, selected_template) print(f"???? Створення preview відео...") preview_duration = audio_preview.duration_seconds # Створюємо фонові сегменти для preview preview_combined, bg_iter = self.create_preview_video_segments(story_id, preview_duration) # Накладаємо preview зображення preview_video = self.create_preview_with_overlay(story_id, preview_combined, preview_img_path, preview_duration) print(f"???? Створення основного відео...") # Створюємо story сегменти final_audio = audio_preview + AudioSegment.silent(duration=500) + audio_story story_clips = self.create_story_video_segments(story_id, bg_iter, final_audio.duration_seconds, preview_duration) # Об'єднуємо всі сегменти combined_video = self.combine_video_segments(story_id, preview_video, story_clips) return combined_video, preview_img_path def finalize_video(self, combined_video, audio_path, output_path): """ Фінальна збірка відео з аудіо Args: combined_video: Шлях до відео без аудіо audio_path: Шлях до аудіо файлу output_path: Шлях для збереження фінального відео """ print(f"????️ Фінальна збірка горизонтального відео...") subprocess.run([ "ffmpeg", "-y", "-i", str(combined_video), "-i", str(audio_path), "-c:v", "copy", "-c:a", "aac", "-shortest", "-movflags", "+faststart", str(output_path) ], check=True)