Spaces:
No application file
No application file
| import os | |
| import requests | |
| import cv2 | |
| import numpy as np | |
| from gtts import gTTS # Add this import | |
| from urllib.parse import quote as url_quote | |
| import shutil | |
| import logging | |
| from flask import Flask, request, jsonify | |
| from time import sleep | |
| import random | |
| import os | |
| # === Setup Logging === | |
| logging.basicConfig( | |
| level=logging.INFO, | |
| format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', | |
| handlers=[ | |
| logging.FileHandler('stoic_reel.log'), | |
| logging.StreamHandler() | |
| ] | |
| ) | |
| app = Flask(__name__, static_folder='static') | |
| # === Configure CORS for Cloudflare frontend === | |
| from flask_cors import CORS | |
| CORS(app, origins=["https://reeler.pages.dev"], supports_credentials=True, methods=["GET", "POST", "OPTIONS"], allow_headers=["Content-Type"]) | |
| # === Custom OPTIONS handler for CORS preflight === | |
| def handle_options(): | |
| return jsonify({}), 200 | |
| # === Hardcoded Config === | |
| API_URL = "https://generative.mdzaiduiux.workers.dev/?prompt=" | |
| OUTPUT_DIR = "stoic_images" | |
| REELS_DIR = "static/vdo" | |
| VOICEOVER_FILE = "stoic_voiceover.mp3" | |
| FPS = 40 | |
| RESOLUTION = (1080, 1920) | |
| PLACEHOLDER_IMAGE = "placeholder.jpg" | |
| USER_AGENT = "StoicReelApp (https://github.com/mzaid/stoicreel)" | |
| VOICE_TYPE = "male" | |
| TRANSITION_FRAMES = 6 | |
| # === Create Placeholder Image === | |
| def create_placeholder_image(size=(1080, 1920)): | |
| img = np.zeros((size[1], size[0], 3), dtype=np.uint8) | |
| cv2.imwrite(PLACEHOLDER_IMAGE, img) | |
| return PLACEHOLDER_IMAGE | |
| # === Split Quote into Segments === | |
| def split_quote(quote, num_segments): | |
| words = quote.split() | |
| words_per_segment = max(1, len(words) // num_segments) | |
| segments = [] | |
| for i in range(0, len(words), words_per_segment): | |
| segment = ' '.join(words[i:i+words_per_segment]) | |
| segments.append(segment) | |
| if len(segments) > num_segments: | |
| segments[-2] += ' ' + segments[-1] | |
| segments = segments[:-1] | |
| return segments[:num_segments] | |
| # === Estimate Voiceover Duration === | |
| def estimate_voiceover_duration(text, rate=140): | |
| words = len(text.split()) | |
| seconds = (words / (rate / 60)) * 1.1 | |
| return max(1, seconds) | |
| # === Calculate Number of Segments === | |
| def calculate_num_segments(quote): | |
| words = len(quote.split()) | |
| num_segments = max(3, min(words // 5, 8)) # Between 3 and 8 segments | |
| return num_segments | |
| # === Generate Prompt Variations === | |
| def generate_prompt_variations(base_prompt, num_variations): | |
| variations = [base_prompt] | |
| modifiers = [ | |
| "at sunrise, vibrant colors", | |
| "at sunset, warm tones", | |
| "under moonlight, serene", | |
| "in misty weather, ethereal", | |
| "with dramatic clouds, cinematic", | |
| "in autumn, golden hues", | |
| "at dusk, soft lighting" | |
| ] | |
| for i in range(num_variations - 1): | |
| modifier = random.choice(modifiers) | |
| variations.append(f"{base_prompt}, {modifier}, 4k, vertical") | |
| return variations | |
| # === Text Wrapper for Captions === | |
| def wrap_text(text, max_width, font, scale, thickness): | |
| words = text.split() | |
| lines = [] | |
| current_line = '' | |
| for word in words: | |
| test_line = f'{current_line} {word}'.strip() | |
| (w, _), _ = cv2.getTextSize(test_line, font, scale, thickness) | |
| if w <= max_width: | |
| current_line = test_line | |
| else: | |
| lines.append(current_line) | |
| current_line = word | |
| lines.append(current_line) | |
| return lines | |
| # === Fetch Images === | |
| def fetch_images(idx, prompts, retries=3): | |
| os.makedirs(OUTPUT_DIR, exist_ok=True) | |
| paths = [] | |
| for i, prompt in enumerate(prompts): | |
| success = False | |
| for attempt in range(retries): | |
| try: | |
| logging.info(f"Fetching image: {prompt} (Attempt {attempt+1})") | |
| response = requests.get(f"{API_URL}{url_quote(prompt)}", timeout=45, headers={"User-Agent": USER_AGENT}) | |
| if response.status_code == 200: | |
| path = os.path.join(OUTPUT_DIR, f"{idx}_{i}.jpg") | |
| with open(path, "wb") as f: | |
| f.write(response.content) | |
| paths.append(path) | |
| success = True | |
| logging.info(f"Fetched image: {path}") | |
| break | |
| else: | |
| logging.warning(f"Image fetch error: {response.status_code}") | |
| except Exception as e: | |
| logging.error(f"Image fetch error: {e}") | |
| sleep(3) | |
| if not success: | |
| paths.append(create_placeholder_image()) | |
| logging.warning(f"Using placeholder for {prompt}") | |
| return paths | |
| # === Generate Voiceover === | |
| def generate_voiceover(quote, author, file): | |
| try: | |
| full_text = f"{quote}" | |
| tts = gTTS(text=full_text, lang='en') | |
| tts.save(file) | |
| logging.info('Voiceover saved with gTTS') | |
| except Exception as e: | |
| logging.error(f"Voiceover error: {e}") | |
| # Create empty fallback file if TTS fails | |
| open(file, 'a').close() | |
| raise | |
| # === Image Resizer for 9:16 === | |
| def resize_and_pad(img, size=(1080, 1920)): | |
| h, w = img.shape[:2] | |
| scale = min(size[0]/w, size[1]/h) | |
| img = cv2.resize(img, (int(w*scale), int(h*scale))) | |
| top = (size[1] - img.shape[0]) // 2 | |
| bottom = size[1] - img.shape[0] - top | |
| left = (size[0] - img.shape[1]) // 2 | |
| right = size[0] - img.shape[1] - left | |
| return cv2.copyMakeBorder(img, top, bottom, left, right, cv2.BORDER_CONSTANT, value=[0, 0, 0]) | |
| def create_video_with_voice(paths, quote, output_file, voiceover_file, num_segments): | |
| if not any(paths): | |
| logging.error('No images for video') | |
| return False | |
| try: | |
| fourcc = cv2.VideoWriter_fourcc(*'mp4v') | |
| out = cv2.VideoWriter(output_file, fourcc, FPS, RESOLUTION) | |
| font = cv2.FONT_HERSHEY_SIMPLEX | |
| quote_segments = split_quote(quote, num_segments) | |
| quote_only_duration = estimate_voiceover_duration(quote, rate=140) | |
| segment_duration = quote_only_duration / num_segments | |
| frames_per_segment = int(segment_duration * FPS) | |
| for idx, segment in enumerate(quote_segments): | |
| img = cv2.imread(paths[idx % len(paths)]) | |
| if img is None: | |
| logging.warning(f"Failed to load image: {paths[idx % len(paths)]}") | |
| continue | |
| img = resize_and_pad(img, RESOLUTION) | |
| overlay = img.copy() | |
| lines = wrap_text(segment, 900, font, 1.3, 3) | |
| for i, line in enumerate(lines): | |
| (text_width, _), _ = cv2.getTextSize(line, font, 1.3, 3) | |
| x = (RESOLUTION[0] - text_width) // 2 | |
| cv2.putText(overlay, line, (x+3, 103 + i*50), font, 1.3, (0, 0, 0), 5, cv2.LINE_AA) | |
| cv2.putText(overlay, line, (x, 300 + i*50), font, 1.3, (255, 255, 255), 3, cv2.LINE_AA) | |
| # Watermark or credit line | |
| text = '@daily.stoic.dose' | |
| (text_width, _), _ = cv2.getTextSize(text, font, 1.0, 2) | |
| x = (RESOLUTION[0] - text_width) // 2 | |
| y = RESOLUTION[1] - 120 | |
| cv2.putText(overlay, text, (x, y), font, 1.0, (255, 255, 255), 2, cv2.LINE_AA) | |
| combined = cv2.addWeighted(overlay, 1, img, 0.3, 0) | |
| for _ in range(frames_per_segment): | |
| out.write(combined) | |
| if idx < len(quote_segments) - 1: | |
| for _ in range(TRANSITION_FRAMES): | |
| out.write(img) | |
| out.release() | |
| logging.info(f'Video saved: {output_file}') | |
| final_output = f'final_{output_file}' | |
| ffmpeg_cmd = ( | |
| f'ffmpeg -y -i "{output_file}" -i "{voiceover_file}" ' | |
| f'-c:v libx264 -preset ultrafast -crf 28 -pix_fmt yuv420p -c:a aac -b:a 128k -movflags +faststart -shortest "{final_output}"' | |
| ) | |
| logging.info(f'Running FFmpeg: {ffmpeg_cmd}') | |
| result = os.system(ffmpeg_cmd) | |
| if result == 0 and os.path.exists(final_output): | |
| logging.info(f'Instagram Reel ready: {final_output}') | |
| return True | |
| else: | |
| logging.error('FFmpeg failed or output file missing') | |
| return False | |
| except Exception as e: | |
| logging.error(f"Video creation error: {e}") | |
| return False | |
| # === Generate Unique Filename === | |
| def get_unique_filename(base_name, extension, directory): | |
| os.makedirs(directory, exist_ok=True) | |
| counter = 1 | |
| while os.path.exists(os.path.join(directory, f"{base_name}{counter}{extension}")): | |
| counter += 1 | |
| return os.path.join(directory, f"{base_name}{counter}{extension}") | |
| # === API Endpoint === | |
| def generate_video(): | |
| try: | |
| data = request.json | |
| quote_text = data.get('quote', '') | |
| author = data.get('author', '') | |
| base_prompt = data.get('prompt', '') | |
| if not quote_text or not author or not base_prompt: | |
| return jsonify({'success': False, 'error': 'Quote, author, and prompt are required'}) | |
| quote = f"{quote_text} - {author}" | |
| idx = random.randint(0, 1000) | |
| num_segments = calculate_num_segments(quote) | |
| prompts = generate_prompt_variations(base_prompt, num_segments) | |
| video_base = f"vdo_{idx}" | |
| video_file = get_unique_filename(video_base, '.mp4', REELS_DIR) | |
| temp_video_file = f'temp_{video_base}.mp4' | |
| voiceover_file = f"voiceover_{idx}.mp3" | |
| paths = fetch_images(idx, prompts) | |
| generate_voiceover(quote, author, voiceover_file) | |
| success = create_video_with_voice(paths, quote, temp_video_file, voiceover_file, num_segments) | |
| if success and os.path.exists(f'final_{temp_video_file}'): | |
| shutil.move(f'final_{temp_video_file}', video_file) | |
| try: | |
| if os.path.exists(OUTPUT_DIR): | |
| shutil.rmtree(OUTPUT_DIR) | |
| if os.path.exists(temp_video_file): | |
| os.remove(temp_video_file) | |
| if os.path.exists(voiceover_file): | |
| os.remove(voiceover_file) | |
| if os.path.exists(PLACEHOLDER_IMAGE): | |
| os.remove(PLACEHOLDER_IMAGE) | |
| except Exception as e: | |
| logging.error(f'Cleanup error: {e}') | |
| video_url = f"/static/vdo/{os.path.basename(video_file)}" # Relative URL for Render | |
| return jsonify({'success': True, 'video_url': video_url}) | |
| else: | |
| return jsonify({'success': False, 'error': 'Failed to generate video'}) | |
| except Exception as e: | |
| logging.error(f'Video generation error: {e}') | |
| return jsonify({'success': False, 'error': str(e)}) | |
| if __name__ == '__main__': | |
| port = int(os.environ.get('PORT', 7860)) # Hugging Face default port | |
| app.run(debug=False, host='0.0.0.0', port=port) |