USER
have some code im trying to fix, its giving me errors, heres the code, then i will give the error:
Config.py
import os
# Base Directory and Storage Settings
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
STORAGE_PATH = os.path.join(BASE_DIR, 'storage')
FFMPEG_PATH = r"C:\ffmpeg\ffmpeg.exe" # Update this path to your ffmpeg installation
# Font settings - Updated to point to the existing font file
FONT_PATH = os.path.join(BASE_DIR, 'fonts', 'Roboto-Regular.ttf')
# API Keys and Headers
OPENAI_API_KEY = "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"
OPENAI_ORGANIZATION = "org-6rCVdDT1dTUsudtHt9ZjkxIe"
PROJECT_ID = "proj_yqEYl4a8Pyo6mHBASGran0sX"
# OpenAI API Headers
OPENAI_API_HEADERS = {
"Authorization": f"Bearer {OPENAI_API_KEY}",
"OpenAI-Organization": OPENAI_ORGANIZATION,
"OpenAI-Project": PROJECT_ID
}
# Ensure storage directory exists
if not os.path.exists(STORAGE_PATH):
os.makedirs(STORAGE_PATH)
# Voice Options
VOICE_OPTIONS = {
"NARRATOR": "echo",
"MALE_1": "echo",
"MALE_2": "onyx",
"MALE_3": "fable",
"FEMALE_1": "shimmer",
"FEMALE_2": "nova",
"FEMALE_3": "alloy",
}
# Rate Limiter Settings
MAX_CALLS_PER_MINUTE = 15
RATE_LIMIT_PERIOD = 60 # seconds
# Video Settings
VIDEO_WIDTH = 1920
VIDEO_HEIGHT = 1080
VIDEO_FPS = 24
# Font Settings
FONT_SIZE = 60
FONT_COLOR = (255, 255, 255) # White
STROKE_WIDTH = 2
STROKE_COLOR = (0, 0, 0) # Black
# Story Generation Settings
DEFAULT_GENRE = "revenge corporate business rags to riches"
DEFAULT_NUM_CHAPTERS = 3
# DALL-E Settings
DALLE_IMAGE_SIZE = "1792x1024"
# Transcription Settings
TRANSCRIPTION_MODEL = "whisper-1"
# TTS Settings
TTS_MODEL = "tts-1"
# GPT Model Settings
GPT_MODEL = "gpt-3.5-turbo-16k"
GPT_MAX_TOKENS = 2000
GPT_TEMPERATURE = 0.7
# Background Image Options
BG_IMAGE_OPTIONS = ['Prompt', 'Auto', 'File Storage', 'None']
# File Names
GPT_DATA_FILE = "gpt_data.json"
BACKGROUND_IMAGE_FILE = "background.jpg"
SPEECH_AUDIO_FILE = "speech.mp3"
TIMESTAMPS_FILE = "timestamps.csv"
FINAL_VIDEO_FILE = "final_video.mp4"
# Error Messages
ERROR_FFMPEG_NOT_FOUND = "FFmpeg not found. Please install FFmpeg and update the path in config.py"
ERROR_OPENAI_API_KEY_MISSING = "OpenAI API key is missing. Please add your API key to config.py"
# Only check for critical dependencies
if not os.path.exists(FFMPEG_PATH):
raise FileNotFoundError(ERROR_FFMPEG_NOT_FOUND)
if not OPENAI_API_KEY:
raise ValueError(ERROR_OPENAI_API_KEY_MISSING)
# Verify font exists without raising an error
if not os.path.exists(FONT_PATH):
print(f"Warning: Font file not found at {FONT_PATH}")
------------------------------------------------------------
create_template.py
# reddit_template.py
from PIL import Image, ImageDraw, ImageFont
import os
class RedditTemplateCreator:
def __init__(self):
self.COLORS = {
'background': '#DAE0E6', # Reddit's light gray background
'post_bg': '#FFFFFF', # Post background
'upvote': '#FF4500', # Reddit's orange
'downvote': '#7193FF', # Reddit's blue
'text': '#222222', # Main text color
'secondary': '#787C7E', # Secondary text
'divider': '#EDEFF1', # Line dividers
'link_color': '#0079D3' # Reddit's link blue
}
# Create necessary directories
os.makedirs('assets/templates', exist_ok=True)
def create_template(self, width=1920, height=1080):
"""Create a Reddit-style post template"""
# Create base image with Reddit's background color
image = Image.new('RGB', (width, height), self.COLORS['background'])
draw = ImageDraw.Draw(image)
# Calculate post dimensions
margin = 60
post_width = width - (margin * 2)
post_height = height - (margin * 2)
# Add shadow effect for post container
shadow_offset = 3
draw.rectangle(
[margin + shadow_offset, margin + shadow_offset,
margin + post_width + shadow_offset, margin + post_height + shadow_offset],
fill='#CCCCCC'
)
# Draw main post container
draw.rectangle(
[margin, margin, margin + post_width, margin + post_height],
fill=self.COLORS['post_bg']
)
# Add voting sidebar
sidebar_width = 40
sidebar_x = margin + 20
sidebar_y = margin + 20
# Draw upvote arrow
arrow_size = 24
points = [
(sidebar_x + arrow_size//2, sidebar_y), # Top point
(sidebar_x, sidebar_y + arrow_size), # Bottom left
(sidebar_x + arrow_size, sidebar_y + arrow_size) # Bottom right
]
draw.polygon(points, outline=self.COLORS['upvote'])
# Add placeholder for vote count
vote_y = sidebar_y + arrow_size + 10
draw.text((sidebar_x + 8, vote_y), "•", fill=self.COLORS['secondary'])
# Draw downvote arrow
down_y = vote_y + 30
down_points = [
(sidebar_x, down_y), # Top left
(sidebar_x + arrow_size, down_y), # Top right
(sidebar_x + arrow_size//2, down_y + arrow_size) # Bottom point
]
draw.polygon(down_points, outline=self.COLORS['downvote'])
# Add metadata bar (subreddit, posted by, etc.)
meta_x = margin + sidebar_width + 40
meta_y = margin + 20
draw.text((meta_x, meta_y), "r/", fill=self.COLORS['link_color'])
# Add divider line
divider_y = meta_y + 40
draw.line(
[meta_x, divider_y, margin + post_width - 20, divider_y],
fill=self.COLORS['divider'],
width=1
)
# Content area is now ready for dynamic text
# Save template
template_path = 'assets/templates/reddit_template.png'
image.save(template_path)
print(f"Template created successfully at: {template_path}")
return template_path
def verify_template(self):
"""Verify template exists and is valid"""
template_path = 'assets/templates/reddit_template.png'
if not os.path.exists(template_path):
print("Template not found, creating new one...")
return self.create_template()
try:
# Verify image can be opened
Image.open(template_path)
print("Template verified successfully!")
return template_path
except Exception as e:
print(f"Invalid template, creating new one... Error: {str(e)}")
return self.create_template()
# Test the template creation
if __name__ == "__main__":
creator = RedditTemplateCreator()
template_path = creator.verify_template()
print(f"Template ready at: {template_path}")
-------------------------------------------------------
main.py
import os
import asyncio
from pathlib import Path
import psutil
import shutil
from config import (
OPENAI_API_KEY,
OPENAI_API_HEADERS,
OPENAI_ORGANIZATION,
PROJECT_ID,
STORAGE_PATH
)
def verify_directory(path):
"""Create directory if it doesn't exist and return success message"""
Path(path).mkdir(parents=True, exist_ok=True)
return f"✓ Verified directory: {path}"
def check_system_resources():
"""Check and display system resource information"""
cpu_percent = psutil.cpu_percent()
memory = psutil.virtual_memory()
disk = psutil.disk_usage('/')
print("=== System Resources ===")
print(f"CPU Usage: {cpu_percent}%")
print(f"Memory Available: {memory.available // (1024*1024)}MB")
print(f"Disk Space Available: {disk.free // (1024*1024*1024):.1f}GB")
def verify_config():
"""Verify required configuration values are present"""
print("=== Checking Configuration ===")
if OPENAI_API_KEY:
print("✓ Verified OPENAI_API_KEY")
else:
raise ValueError("× Missing OPENAI_API_KEY")
if OPENAI_ORGANIZATION:
print("✓ Verified OPENAI_ORGANIZATION")
else:
raise ValueError("× Missing OPENAI_ORGANIZATION")
if PROJECT_ID:
print("✓ Verified PROJECT_ID")
else:
raise ValueError("× Missing PROJECT_ID")
async def main():
try:
print("=== Setting Up Environment ===")
# Verify required directories
directories = [
"assets",
"assets/fonts",
"assets/templates",
"assets/backgrounds",
"assets/temp",
"output"
]
for directory in directories:
print(verify_directory(directory))
# Check system resources
check_system_resources()
# Check background videos
bg_videos = list(Path("assets/backgrounds").glob("*.mp4"))
print(f"✓ Found {len(bg_videos)} background videos")
# Verify configuration
verify_config()
print("=== Initializing Video Generator ===")
from reddit_video_generator import RedditVideoGenerator
generator = RedditVideoGenerator(
output_dir=STORAGE_PATH,
openai_key=OPENAI_API_KEY,
openai_org=OPENAI_ORGANIZATION
)
await generator.create_full_video(
subreddit="AskReddit",
num_posts=3,
output_path="output/final_video.mp4"
)
print("✓ Video generation completed")
except Exception as e:
print(f"× Error during video generation: {str(e)}")
finally:
print("✓ Cleanup completed")
if __name__ == "__main__":
try:
asyncio.run(main())
except Exception as e:
print(f"× Fatal error: {str(e)}")
finally:
print("Process completed")
------------------------------------
reddit_video_generator.py
import os
import asyncio
import json
from pathlib import Path
from typing import List, Dict
from openai import AsyncOpenAI
from PIL import Image, ImageDraw, ImageFont
from moviepy.editor import VideoFileClip, AudioFileClip, ImageClip, concatenate_videoclips
from moviepy.video.fx.fadeout import fadeout
from moviepy.video.fx.fadein import fadein
import textwrap
class RedditVideoGenerator:
def __init__(self, output_dir: str = "assets/temp", openai_key: str = None, openai_org: str = None):
self.openai_client = AsyncOpenAI(
api_key=openai_key,
organization=openai_org
)
self.storage_path = Path(output_dir)
self.video_width = 1920
self.video_height = 1080
self.font_size = 60
# Updated font path handling
script_dir = Path(__file__).parent
self.font_path = str(script_dir / "assets" / "fonts" / "Roboto-Regular.ttf")
if not os.path.exists(self.font_path):
print(f"Warning: Font not found at {self.font_path}")
self.font_path = str(script_dir / "fonts" / "Roboto-Regular.ttf") # Try alternate location
self.ensure_directories()
def ensure_directories(self):
"""Create necessary directories if they don't exist"""
for dir_name in ["audio", "images", "videos"]:
(self.storage_path / dir_name).mkdir(parents=True, exist_ok=True)
async def generate_reddit_posts(self, subreddit: str, num_posts: int) -> List[Dict]:
"""Generate multiple Reddit posts using GPT"""
posts = []
for i in range(num_posts):
try:
prompt = f"""Create an engaging Reddit post for r/{subreddit}.
Format the response exactly as shown below:
Title: [An interesting and engaging post title]
Content: [A detailed and thought-provoking post content of at least 100 words]
Make the content engaging and natural, like a real Reddit post."""
response = await self.openai_client.chat.completions.create(
model="gpt-4",
messages=[{
"role": "system",
"content": "You are a Reddit post generator. Create engaging content that feels authentic and conversational."
},
{
"role": "user",
"content": prompt
}],
temperature=0.8
)
try:
response_text = response.choices[0].message.content
print(f"Raw GPT Response:\n{response_text}") # Debug print
# Parse the response into our required format
sections = response_text.split('\n\n') # Split by double newlines
post_data = {
"title": "",
"content": ""
}
# Process each section
for section in sections:
section = section.strip()
if section.startswith("Title:"):
post_data["title"] = section.replace("Title:", "").strip().strip('"')
elif section.startswith("Content:"):
post_data["content"] = section.replace("Content:", "").strip()
# Validate parsed data
if not post_data["title"]:
raise ValueError("Missing title")
if not post_data["content"]:
raise ValueError("Missing content")
print(f"Parsed post data:\n{json.dumps(post_data, indent=2)}") # Debug print
posts.append(post_data)
except Exception as e:
print(f"Error parsing response: {str(e)}")
print(f"Response text: {response_text}")
raise
await asyncio.sleep(1) # Rate limiting
except Exception as e:
print(f"Error generating post {i+1}: {str(e)}")
raise
return posts
async def create_post_image(self, post_data: Dict, index: int):
"""Create an image mimicking Reddit's interface"""
try:
# Create base image
img = Image.new('RGB', (self.video_width, self.video_height), color='white')
draw = ImageDraw.Draw(img)
try:
font = ImageFont.truetype(self.font_path, self.font_size)
except Exception as e:
print(f"Font error: {str(e)} - Falling back to default font")
font = ImageFont.load_default()
# Draw title
title_wrapped = textwrap.wrap(str(post_data.get("title", "")), width=50)
y_position = 100
for line in title_wrapped:
draw.text((100, y_position), line, font=font, fill='black')
y_position += self.font_size + 10
# Draw content
y_position += 50 # Add some space between title and content
content_wrapped = textwrap.wrap(str(post_data.get("content", "")), width=60)
for line in content_wrapped:
draw.text((100, y_position), line, font=font, fill='black')
y_position += self.font_size + 5
# Save image
img_path = self.storage_path / "images" / f"post_{index}.png"
img.save(img_path)
return img_path
except Exception as e:
print(f"Error creating post image: {str(e)}")
raise
async def generate_audio(self, text: str, index: int) -> str:
"""Generate audio narration using OpenAI TTS"""
try:
response = await self.openai_client.audio.speech.create(
model="tts-1",
voice="alloy",
input=text
)
audio_path = self.storage_path / "audio" / f"audio_{index}.mp3"
with open(audio_path, "wb") as f:
f.write(response.content)
return str(audio_path)
except Exception as e:
print(f"Error generating audio: {str(e)}")
raise
async def create_video_segment(self, post_data: Dict, index: int) -> str:
"""Create a video segment for a single post"""
try:
# Create image and audio concurrently
img_path, audio_path = await asyncio.gather(
self.create_post_image(post_data, index),
self.generate_audio(f"{post_data['title']}. {post_data['content']}", index)
)
# Create video clip
audio = AudioFileClip(str(audio_path))
image = ImageClip(str(img_path)).set_duration(audio.duration)
video = image.set_audio(audio)
# Save video segment
video_path = self.storage_path / "videos" / f"segment_{index}.mp4"
video.write_videofile(str(video_path), fps=24)
# Clean up
audio.close()
video.close()
return str(video_path)
except Exception as e:
print(f"Error creating video segment: {str(e)}")
raise
async def create_full_video(self, subreddit: str, num_posts: int, output_path: str):
"""Create a complete video with multiple Reddit posts"""
try:
# Generate posts
posts = await self.generate_reddit_posts(subreddit, num_posts)
# Create video segments concurrently
video_segment_paths = []
for i, post in enumerate(posts):
segment_path = await self.create_video_segment(post, i)
video_segment_paths.append(segment_path)
# Combine segments with transitions
video_clips = []
for i, path in enumerate(video_segment_paths):
clip = VideoFileClip(path)
if i > 0: # Add fade in for all clips except the first
clip = fadein(clip, duration=0.5)
if i < len(video_segment_paths) - 1: # Add fade out for all clips except the last
clip = fadeout(clip, duration=0.5)
video_clips.append(clip)
# Concatenate all clips
final_video = concatenate_videoclips(video_clips, method="compose")
# Write final video
final_video.write_videofile(output_path, fps=24)
# Cleanup
for clip in video_clips:
clip.close()
final_video.close()
print(f"Video successfully created at {output_path}")
except Exception as e:
print(f"Error creating video: {str(e)}")
raise
async def cleanup(self):
"""Clean up temporary files"""
try:
for dir_name in ["audio", "images", "videos"]:
dir_path = self.storage_path / dir_name
for file in dir_path.glob("*"):
file.unlink()
except Exception as e:
print(f"Error during cleanup: {str(e)}")
async def main():
# This main function is for testing purposes
from config import OPENAI_API_KEY, OPENAI_ORGANIZATION, STORAGE_PATH
generator = RedditVideoGenerator(
output_dir=STORAGE_PATH,
openai_key=OPENAI_API_KEY,
openai_org=OPENAI_ORGANIZATION
)
try:
subreddit = "AskReddit" # or any other subreddit theme
num_posts = 3 # Adjust based on desired video length
output_path = "final_video.mp4"
await generator.create_full_video(subreddit, num_posts, output_path)
finally:
await generator.cleanup()
if __name__ == "__main__":
asyncio.run(main())
---------------------------------
requirments.txt
# requirements.txt
openai
pillow
moviepy
numpy
pydub
opencv-python
python-dotenv
psutil
ffmpeg-python
aiohttp
requests
tqdm
--------------------------------------
setup.py
# setup.py
import os
from PIL import Image, ImageDraw
import shutil
def setup_project():
print("Starting setup process...")
# 1. Create directory structure
directories = [
'assets',
'assets/fonts',
'assets/templates',
'output',
'output/temp'
]
for directory in directories:
os.makedirs(directory, exist_ok=True)
print(f"✓ Created directory: {directory}")
# 2. Create template
template_path = os.path.join('assets', 'templates', 'reddit_template.png')
# Create basic template
WIDTH = 1920
HEIGHT = 1080
img = Image.new('RGB', (WIDTH, HEIGHT), '#FFFFFF')
draw = ImageDraw.Draw(img)
# Add basic Reddit styling
draw.rectangle([0, 0, WIDTH, HEIGHT], fill='#DAE0E6') # Background
draw.rectangle([60, 40, WIDTH-60, HEIGHT-40], fill='#FFFFFF') # Post area
# Save template
img.save(template_path)
print(f"✓ Created template at: {template_path}")
# 3. Check font
font_path = os.path.join('assets', 'fonts', 'bowlbyoneSC-Regular.ttf')
if not os.path.exists(font_path):
print(f"! Font file not found at: {font_path}")
print("Please copy bowlbyoneSC-Regular.ttf to the assets/fonts directory")
# 4. Verify files
print("\nVerifying setup...")
all_good = True
# Check template
if os.path.exists(template_path):
print(f"✓ Template exists at: {os.path.abspath(template_path)}")
else:
print("× Template file missing!")
all_good = False
# Check font
if os.path.exists(font_path):
print(f"✓ Font exists at: {os.path.abspath(font_path)}")
else:
print("× Font file missing!")
all_good = False
# Final status
if all_good:
print("\n✓ Setup completed successfully!")
else:
print("\n! Setup completed with warnings. Please check the messages above.")
return all_good
if __name__ == "__main__":
setup_project()
-------------------------------------------------------------
test_setup.py
# test_video.py
import asyncio
from reddit_video_generator import RedditVideoGenerator
import config
async def test_video():
"""Generate a test video with one post"""
print("\n=== Testing Video Generation ===")
# Use just one topic for testing
test_topics = ["funny cat stories"]
try:
generator = RedditVideoGenerator(
openai_api_key=config.OPENAI_API_KEY,
output_dir="output/test"
)
print("\nGenerating test video...")
output_path = await generator.generate_full_video(
topics=test_topics,
target_duration=60 # 1 minute test video
)
print(f"\n✓ Test video created: {output_path}")
except Exception as e:
print(f"\n× Error during test: {str(e)}")
if __name__ == "__main__":
asyncio.run(test_video())
-----------------------------------------------
test_video.py
# test_video.py
import asyncio
import os
import config
from reddit_video_generator import RedditVideoGenerator
import os
import asyncio
import json
from pathlib import Path
from typing import List, Dict
from openai import AsyncOpenAI
from PIL import Image, ImageDraw, ImageFont
from moviepy.editor import VideoFileClip, AudioFileClip, ImageClip, concatenate_videoclips, CompositeVideoClip, vfx
import textwrap
class RedditVideoGenerator:
def __init__(self, output_dir: str = "assets/temp"):
self.openai_client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
self.storage_path = Path(output_dir)
self.video_width = 1920
self.video_height = 1080
self.font_size = 60
self.font_path = str(Path("assets/fonts/Roboto-Regular.ttf"))
self.ensure_directories()
async def test_video():
"""Generate a test video with one post"""
print("\n=== Testing Video Generation ===")
# Use just one topic for testing
test_topics = ["funny cat stories"]
try:
# Initialize with OpenAI configuration from config.py
generator = RedditVideoGenerator(
openai_api_key=config.OPENAI_API_KEY,
output_dir=os.path.join("output", "test"),
font_path=os.path.join("assets", "fonts", "bowlbyoneSC-Regular.ttf"),
template_path=os.path.join("assets", "templates", "reddit_template.png")
)
print("\nGenerating test video...")
output_path = await generator.generate_full_video(
topics=test_topics,
target_duration=60 # 1 minute test video
)
print(f"\n✓ Test video created: {output_path}")
except Exception as e:
print(f"\n× Error during test: {str(e)}")
print(f"Error type: {type(e).__name__}")
import traceback
print("\nFull error traceback:")
print(traceback.format_exc())
if __name__ == "__main__":
# Create test output directory if it doesn't exist
os.makedirs(os.path.join("output", "test"), exist_ok=True)
print("Starting video test...")
print(f"OpenAI API Key: {'Set' if config.OPENAI_API_KEY else 'Not Set'}")
print(f"OpenAI Organization: {'Set' if config.OPENAI_ORGANIZATION else 'Not Set'}")
print(f"Project ID: {'Set' if config.PROJECT_ID else 'Not Set'}")
try:
asyncio.run(test_video())
except KeyboardInterrupt:
print("\nTest interrupted by user")
except Exception as e:
print(f"\nTest failed with error: {str(e)}")
finally:
print("\nTest completed")
input("Press Enter to exit…")
--------------------------------------------------------------------------
video_processor.py
import os
import time
import numpy as np
from typing import List
from pathlib import Path
import subprocess
from PIL import Image
from moviepy.editor import VideoFileClip, ImageClip, CompositeVideoClip, concatenate_videoclips, AudioFileClip
import psutil
class VideoProcessor:
def __init__(self, background_path="assets/backgrounds/"):
self.background_path = background_path
os.makedirs(background_path, exist_ok=True)
os.makedirs("assets/temp", exist_ok=True)
self.supported_formats = ['.mp4', '.mkv']
self.background_videos = {}
self.scan_and_convert_videos()
if not self.background_videos:
print("Warning: No background videos found in assets/backgrounds/")
print("Please add MP4 or MKV files to the backgrounds folder")
def scan_and_convert_videos(self):
print("Scanning for background videos...")
for file in os.listdir(self.background_path):
file_lower = file.lower()
if file_lower.endswith(tuple(self.supported_formats)):
file_path = os.path.join(self.background_path, file)
if file_lower.endswith('.mkv'):
print(f"Found MKV file: {file}")
mp4_path = self.convert_mkv_to_mp4(file_path)
if mp4_path:
name = Path(file).stem
self.background_videos[name] = Path(mp4_path).name
elif file_lower.endswith('.mp4'):
print(f"Found MP4 file: {file}")
name = Path(file).stem
self.background_videos[name] = file
print(f"Available background videos: {list(self.background_videos.keys())}")
def convert_mkv_to_mp4(self, mkv_path: str) -> str:
output_path = str(Path(mkv_path).with_suffix('.mp4'))
try:
print(f"Converting {mkv_path} to MP4...")
subprocess.run([
'ffmpeg', '-i', mkv_path,
'-c:v', 'copy',
'-c:a', 'aac',
output_path
], check=True, capture_output=True)
print(f"Conversion successful: {output_path}")
return output_path
except subprocess.CalledProcessError as e:
print(f"Error converting video: {e}")
return None
def check_resources(self):
cpu_usage = psutil.cpu_percent()
memory = psutil.virtual_memory().available / (1024 ** 2)
print(f"System Status:")
print(f"- CPU Usage: {cpu_usage}%")
print(f"- Memory Available: {memory:.0f}MB")
return cpu_usage < 85 and memory > 500
def chop_background(self, video_path: str, required_length: int) -> str:
try:
print(f"Preparing background video segment of {required_length} seconds...")
video = VideoFileClip(video_path)
video_length = video.duration
if video_length < required_length:
print(f"Background video ({video_length}s) shorter than required length ({required_length}s). Looping...")
repeats = int(np.ceil(required_length / video_length))
clips = [video] * repeats
video = concatenate_videoclips(clips)
video_length = video.duration
max_start = int(video_length - required_length)
start_time = np.random.randint(0, max_start) if max_start > 0 else 0
segment = video.subclip(start_time, start_time + required_length)
temp_path = os.path.join("assets/temp", f"bg_segment_{int(time.time())}.mp4")
segment.write_videofile(
temp_path,
codec="libx264",
audio_codec="aac",
fps=30,
threads=2,
logger=None
)
video.close()
segment.close()
return temp_path
except Exception as e:
print(f"Error chopping background video: {str(e)}")
if 'video' in locals(): video.close()
if 'segment' in locals(): segment.close()
raise
def create_overlay_clip(self, image_path: str, background_clip: VideoFileClip) -> ImageClip:
pil_image = Image.open(image_path)
new_width = int(background_clip.w * 0.8)
aspect_ratio = pil_image.width / pil_image.height
new_height = int(new_width / aspect_ratio)
pil_image = pil_image.resize((new_width, new_height), Image.Resampling.LANCZOS)
img_array = np.array(pil_image)
overlay = ImageClip(img_array)
x_pos = (background_clip.w - new_width) // 2
y_pos = (background_clip.h - new_height) // 2
overlay = overlay.set_position((x_pos, y_pos)).set_opacity(0.95)
return overlay
def create_video_segment(self, image_path: str, audio_path: str, background_video: str, duration: float = None) -> VideoFileClip:
try:
while not self.check_resources():
print("Waiting for resources to free up...")
time.sleep(1)
print(f"Creating video segment...")
bg_video = VideoFileClip(background_video, target_resolution=(720, 1280))
overlay = self.create_overlay_clip(image_path, bg_video)
composite = CompositeVideoClip(
[bg_video, overlay],
size=(bg_video.w, bg_video.h)
)
audio = AudioFileClip(audio_path)
duration = duration or audio.duration + 0.5
final_clip = composite.set_duration(duration).set_audio(audio)
return final_clip
except Exception as e:
print(f"Error in video processing: {str(e)}")
if 'bg_video' in locals(): bg_video.close()
if 'composite' in locals(): composite.close()
if 'audio' in locals(): audio.close()
raise
def merge_final_video(self, video_segments: List[VideoFileClip], output_path: str, background_video_path: str, total_duration: float) -> str:
try:
print("Starting final video merge...")
bg_path = self.chop_background(background_video_path, int(total_duration))
background = VideoFileClip(bg_path)
final_clips = []
for i, segment in enumerate(video_segments):
if i > 0:
segment = segment.crossfadein(1.0)
final_clips.append(segment)
print("Merging segments...")
merged = concatenate_videoclips(final_clips, method="compose")
print("Adding background...")
final = CompositeVideoClip([background, merged])
print(f"Exporting to {output_path}")
final.write_videofile(
output_path,
fps=30,
codec="libx264",
audio_codec="aac",
threads=2,
logger=None
)
background.close()
merged.close()
final.close()
os.remove(bg_path)
return output_path
except Exception as e:
print(f"Error in final merge: {str(e)}")
if 'background' in locals(): background.close()
if 'merged' in locals(): merged.close()
if 'final' in locals(): final.close()
raise
def cleanup(self):
temp_dir = "assets/temp"
if os.path.exists(temp_dir):
for file in os.listdir(temp_dir):
try:
os.remove(os.path.join(temp_dir, file))
except Exception as e:
print(f"Error removing temp file {file}: {e}")
if __name__ == "__main__":
processor = VideoProcessor()
print("Video processor initialized")
print("Available background videos:", processor.background_videos)
processor.check_resources()
-------------------------------