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)