turns-00044.parquet:21382
8395a541a52bd87b2cbadb44degenerate_repetitionAbsentFinal dense release
USER
ok its still placing it on the reddit_tmplate.png and not the title_template.png
ASSISTANT
To ensure that you are using the correct template file (`title_template.png`), you'll need to explicitly specify the path to `title_template.png` in your code where the template image is opened. Here's how you can adjust your `create_post_image` method within `reddit_video_generator.py` to use `title_template.png`:
### Updated Code for Using `title_template.png`
```python
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
# 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")
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
def fit_text_to_width(self, draw, text, max_width, initial_font_size):
"""
Dynamically resize the font size to fit the text within a given width.
"""
font_path = self.font_path # Ensure that the font path is correctly set
font_size = initial_font_size
font = ImageFont.truetype(font_path, font_size)
wrapped_text = textwrap.wrap(text, width=50)
# Decrease the font size until the text fits the width
while True:
# Check the width of the text block using getbbox()
width = 0
for line in wrapped_text:
bbox = draw.textbbox((0, 0), line, font=font)
line_width = bbox[2] - bbox[0]
if line_width > width:
width = line_width
if width <= max_width or font_size <= 10:
break
font_size -= 2
font = ImageFont.truetype(font_path, font_size)
wrapped_text = textwrap.wrap(text, width=50)
return font, wrapped_text
async def create_post_image(self, post_data: Dict, index: int):
"""Create an image mimicking Reddit's interface, overlaying text on a template"""
try:
# Load the correct template image
template_path = 'assets/templates/title_template.png'
if not os.path.exists(template_path):
raise FileNotFoundError(f"Template not found at {template_path}")
img = Image.open(template_path)
draw = ImageDraw.Draw(img)
initial_font_size = self.font_size
max_text_width = img.width - 200 # Consider margins
# Resizing title text to fit
title = str(post_data.get("title", ""))
font_title, wrapped_title = self.fit_text_to_width(draw, title, max_text_width, initial_font_size)
y_position = 150 # Start a bit lower to avoid the top edge
for line in wrapped_title:
draw.text((100, y_position), line, font=font_title, fill='black')
bbox = draw.textbbox((100, y_position), line, font=font_title)
y_position += (bbox[3] - bbox[1]) + 10
# Resizing content text to fit
content = str(post_data.get("content", ""))
font_content, wrapped_content = self.fit_text_to_width(draw, content, max_text_width, initial_font_size)
y_position += 50 # Space between title and content
for line in wrapped_content:
draw.text((100, y_position), line, font=font_content, fill='gray') # Maybe a different color for separation
bbox = draw.textbbox((100, y_position), line, font=font_content)
y_position += (bbox[3] - bbox[1]) + 5
# Save the 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())
```
### Explanation:
- **Template Path**: The `template_path` variable is explicitly set to `'assets/templates/title_template.png'`. Ensure this matches the actual file path and filename in your assets folder.
- **Error Handling**: An additional check is made to ensure the template file exists, raising a `FileNotFoundError` if it does not, which assists in debugging path issues.
This code adjustment ensures that the `title_template.png` is used for creating post images, which should align with your requirement to overlay text on the correct template. Make sure that `title_template.png` is in the specified directory.