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 === @app.route('/api/generate_video', methods=['OPTIONS']) 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 === @app.route('/api/generate_video', methods=['POST']) 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)