from fastapi import FastAPI, HTTPException, Header from pydantic import BaseModel from typing import Optional import uvicorn import subprocess import os import uuid import base64 import requests import shutil from pathlib import Path import edge_tts import re import emoji import time app = FastAPI() API_KEY = os.environ.get("API_KEY", "default-key") TEMP_DIR = Path("/tmp/ffmpeg_processing") TEMP_DIR.mkdir(parents=True, exist_ok=True) # Set HOME for libass/fontconfig to ensure cache can be written os.environ["HOME"] = "/tmp" os.environ["XDG_CACHE_HOME"] = "/tmp/.cache" # List available filters and fonts for debugging def run_diagnostics(): print("📋 Checking FFmpeg environment...") try: filters_out = subprocess.run( ["ffmpeg", "-filters"], capture_output=True, text=True ).stdout has_subtitles = "subtitles" in filters_out has_drawtext = "drawtext" in filters_out print(f"✅ Filters found: subtitles={has_subtitles}, drawtext={has_drawtext}") print("📋 Available fonts (fc-list):") subprocess.run(["fc-list"], check=True) except Exception as e: print(f"⚠️ Failed to check FFmpeg environment: {e}") run_diagnostics() @app.get("/debug-ffmpeg") async def debug_ffmpeg(): try: filters = subprocess.run( ["ffmpeg", "-filters"], capture_output=True, text=True ).stdout fonts = subprocess.run(["fc-list"], capture_output=True, text=True).stdout return { "filters_summary": { "subtitles": "subtitles" in filters, "drawtext": "drawtext" in filters, }, "env": {k: v for k, v in os.environ.items() if "API" not in k}, "fonts": fonts.splitlines()[:50], # First 50 "raw_filters_hint": filters[:500], } except Exception as e: return {"error": str(e)} # Robust font detection def get_font_path(): possible_paths = [ "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", "/usr/share/fonts/truetype/dejavu-core/DejaVuSans.ttf", "/usr/share/fonts/TTF/DejaVuSans.ttf", ] for path in possible_paths: if os.path.exists(path): print(f"✅ Found font at: {path}") return path # Fallback search print("⚠️ Specific font not found, searching recursively in /usr/share/fonts...") try: found = [] for root, dirs, files in os.walk("/usr/share/fonts"): for file in files: if file.endswith(".ttf"): found.append(os.path.join(root, file)) if found: print(f"✅ Found {len(found)} fonts, using first: {found[0]}") return found[0] except Exception as e: print(f"⚠️ Error searching for fonts: {e}") return "Sans" # Generic fallback FONT_PATH = get_font_path() class ReelRequest(BaseModel): video_base64: Optional[str] = None video_url: Optional[str] = None text: Optional[str] = None music_id: Optional[str] = None music_url: Optional[str] = None word_duration: float = 0.6 font_size: int = 64 music_volume: float = 0.25 tts_enabled: bool = False tts_voice: str = "fr-FR-VivienneMultilingualNeural" draw_text: bool = True stabilize: bool = False # Stabilisation vidéo via vidstab def clean_text_for_tts(text: str) -> str: # 1. Remove emojis text = emoji.replace_emoji(text, replace="") # 2. Remove hashtags (e.g. #viral #reels) text = re.sub(r"#\w+", "", text) # 3. Cleanup whitespace return " ".join(text.split()) async def generate_tts_with_subs( text: str, voice: str, audio_path: Path, ass_path: Path ): """Generate TTS audio with subtitles, with retry and fallback voices.""" import asyncio # Determine gender of requested voice to choose appropriate fallbacks is_male = any(name in voice for name in ["Remi", "Henri", "Paul"]) if is_male: fallback_voices = [ voice, # Try requested voice first "fr-FR-HenriNeural", # Primary Male fallback "fr-FR-PaulNeural", # Secondary Male fallback ] else: fallback_voices = [ voice, # Try requested voice first "fr-FR-VivienneNeural", # Primary Female fallback "fr-FR-DeniseNeural", # Secondary Female fallback ] # Always add English fallback as last resort fallback_voices.append("en-US-JennyNeural") # Remove duplicates while preserving order fallback_voices = list(dict.fromkeys(fallback_voices)) last_error = None for attempt_voice in fallback_voices: try: print(f"🔊 TTS attempt with voice: {attempt_voice}") communicate = edge_tts.Communicate(text, attempt_voice) # Use the simple save() method which is more reliable await asyncio.wait_for(communicate.save(str(audio_path)), timeout=60.0) # Check if file was created and has content if audio_path.exists() and audio_path.stat().st_size > 0: print(f"✅ TTS audio saved: {audio_path.stat().st_size} bytes") # Measure audio duration for perfect sync try: duration_cmd = [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(audio_path), ] dur_proc = subprocess.run( duration_cmd, stdout=subprocess.PIPE, text=True ) audio_duration = float(dur_proc.stdout.strip()) print(f"⏱️ TTS Audio Duration: {audio_duration:.2f}s") except Exception as e: print(f"⚠️ Could not measure TTS duration, using fallback: {e}") audio_duration = None # Generate a high-quality ASS file with sync generate_simple_ass(text, ass_path, total_duration=audio_duration) print(f"✅ TTS success with voice: {attempt_voice}") return # Success! else: print(f"⚠️ Audio file empty or missing with voice: {attempt_voice}") except asyncio.TimeoutError: print(f"⚠️ TTS timeout with voice: {attempt_voice}") last_error = "Timeout" except Exception as e: print(f"⚠️ TTS failed with voice {attempt_voice}: {e}") last_error = e # If all voices failed, raise the last error raise Exception(f"All TTS voices failed. Last error: {last_error}") def generate_simple_ass( text: str, ass_path: Path, font_size: int = 50, total_duration: float = None ): """Generate a high-quality ASS subtitle file with embedded styling.""" # Split text into chunks words = text.split() chunks = [] current_chunk = [] for word in words: current_chunk.append(word) if len(current_chunk) >= 5 or word.endswith((".", "!", "?", ":")): chunks.append(" ".join(current_chunk)) current_chunk = [] if current_chunk: chunks.append(" ".join(current_chunk)) # ASS Header with explicit resolution and style # Alignment 5 = Middle Center header = f"""[Script Info] ScriptType: v4.00+ PlayResX: 1080 PlayResY: 1920 ScaledBorderAndShadow: yes [V4+ Styles] Format: Name, Fontname, Fontsize, PrimaryColour, SecondaryColour, OutlineColour, BackColour, Bold, Italic, Underline, StrikeOut, ScaleX, ScaleY, Spacing, Angle, BorderStyle, Outline, Shadow, Alignment, MarginL, MarginR, MarginV, Encoding Style: Default,DejaVu Sans,{font_size},&H00FFFFFF,&H000000FF,&H00000000,&H00000000,1,0,0,0,100,100,0,0,1,2,0,5,50,50,0,1 [Events] Format: Layer, Start, End, Style, Name, MarginL, MarginR, MarginV, Effect, Text """ events = "" current_time = 0.0 # Calculate timing strategy if total_duration: # Dynamic timing based on characters count (more accurate than word count) total_chars = sum(len(chunk) for chunk in chunks) # Avoid division by zero if total_chars == 0: total_chars = 1 time_per_char = total_duration / total_chars else: # Fallback static timing using word count time_per_char = None for i, chunk in enumerate(chunks): if time_per_char: # Duration proportional to character length including spaces # Ensure minimum readability of 0.5s per chunk chunk_len = len(chunk) duration = chunk_len * time_per_char # Optional: Enforce min duration, but we must respect total_duration # So purely proportional is safer for sync else: # Fallback: 0.4s per word word_count = len(chunk.split()) duration = word_count * 0.4 start_time = format_ass_time(current_time) end_time = format_ass_time(current_time + duration) # Escape special characters if any sanitized_chunk = chunk.replace("{", "(").replace("}", ")") events += ( f"Dialogue: 0,{start_time},{end_time},Default,,0,0,0,,{sanitized_chunk}\n" ) current_time += duration with open(ass_path, "w", encoding="utf-8") as f: f.write(header + events) print( f"📄 Generated ASS file: {ass_path.stat().st_size} bytes with {len(chunks)} chunks" ) def format_ass_time(seconds: float) -> str: """Format seconds as ASS timestamp (H:MM:SS.ss).""" hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) secs = int(seconds % 60) centis = int((seconds % 1) * 100) return f"{hours}:{minutes:02d}:{secs:02d}.{centis:02d}" def generate_simple_srt(text: str, srt_path: Path): # This was a stub, but let's fix the internal helper if it were called def format_srt_time(seconds: float) -> str: hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) secs = int(seconds % 60) millis = int((seconds % 1) * 1000) return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}" pass def format_vtt_time(seconds: float) -> str: """Format seconds as VTT timestamp (HH:MM:SS.mmm).""" hours = int(seconds // 3600) minutes = int((seconds % 3600) // 60) secs = int(seconds % 60) millis = int((seconds % 1) * 1000) return f"{hours:02d}:{minutes:02d}:{secs:02d}.{millis:03d}" class ReelResponse(BaseModel): success: bool output_base64: Optional[str] = None duration: Optional[float] = None detail: Optional[str] = None processing_stats: Optional[dict] = None @app.get("/health") def health_check(x_api_key: str = Header(None)): if x_api_key != API_KEY: raise HTTPException(status_code=401, detail="Invalid API Key") return {"status": "healthy"} @app.post("/process-reel") async def process_reel(request: ReelRequest, x_api_key: str = Header(None)): if x_api_key != API_KEY: raise HTTPException(status_code=401, detail="Invalid API Key") start_total = time.time() stats = { "download_duration": 0, "tts_duration": 0, "stabilize_duration": 0, "encoding_duration": 0, "total_duration": 0, } try: job_id = str(uuid.uuid4()) job_dir = TEMP_DIR / job_id job_dir.mkdir() input_video_path = job_dir / "input.mp4" input_audio_path = job_dir / "music.mp3" tts_audio_path = job_dir / "tts.mp3" tts_ass_path = job_dir / "tts.ass" output_video_path = job_dir / "output.mp4" start_step = time.time() # 1. Save Input Video if request.video_base64: with open(input_video_path, "wb") as f: f.write(base64.b64decode(request.video_base64)) elif request.video_url: response = requests.get(request.video_url, stream=True) response.raise_for_status() with open(input_video_path, "wb") as f: shutil.copyfileobj(response.raw, f) else: raise HTTPException(status_code=400, detail="No video source provided") # 2. Download Music (if present) has_music = False if request.music_url: try: # Add User-Agent to avoid 403 on some CDNs headers = {"User-Agent": "Mozilla/5.0"} response = requests.get(request.music_url, headers=headers, stream=True) response.raise_for_status() with open(input_audio_path, "wb") as f: shutil.copyfileobj(response.raw, f) has_music = True except Exception as e: print(f"Failed to download music: {e}") # We continue without music if it fails stats["download_duration"] = time.time() - start_step start_step = time.time() # 3. Generate TTS (if enabled) has_tts = False tts_clean_text = "" if request.tts_enabled and request.text: try: # Clean text for TTS (remove hashtags/emojis) tts_clean_text = clean_text_for_tts(request.text) print(f"🔊 TTS enabled. Original: '{request.text}'") print(f"🔊 TTS cleaned: '{tts_clean_text}'") # Check for male/female voice map voice = request.tts_voice if voice == "male": voice = "fr-FR-RemyMultilingualNeural" elif voice == "female": voice = "fr-FR-VivienneMultilingualNeural" elif not voice or "Neural" not in voice: # Default if invalid voice = "fr-FR-VivienneMultilingualNeural" print(f"🔊 Using voice: {voice}") if tts_clean_text: print(f"🔊 Generating TTS audio to: {tts_audio_path}") await generate_tts_with_subs( tts_clean_text, voice, tts_audio_path, tts_ass_path ) # Verify files were created if tts_audio_path.exists() and tts_audio_path.stat().st_size > 0: print( f"✅ TTS audio generated: {tts_audio_path.stat().st_size} bytes" ) has_tts = True else: print(f"❌ TTS audio file missing or empty!") else: print("⚠️ TTS text is empty after cleaning, skipping.") except Exception as e: import traceback print(f"❌ Failed to generate TTS: {e}") traceback.print_exc() stats["tts_duration"] = time.time() - start_step start_step = time.time() # 4. Build FFmpeg Command with Unified filter_complex cmd = ["ffmpeg", "-y", "-i", str(input_video_path)] # --- Stability Pass 1 (if requested) --- vidstab_filter = "" if request.stabilize: print("📐 Starting video stabilization (Pass 1: Detection)...") transforms_path = job_dir / "transforms.trf" # Run detection pass # Aggressive stabilization settings: # - shakiness=10: Max sensitivity to shake # - accuracy=15: High accuracy # - stepsize=32: Larger search window for bigger shakes detect_cmd = [ "ffmpeg", "-y", "-i", str(input_video_path), "-vf", f"vidstabdetect=stepsize=32:shakiness=10:accuracy=15:result={transforms_path}", "-f", "null", "-", ] detect_proc = subprocess.run( detect_cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE ) if detect_proc.returncode == 0 and transforms_path.exists(): print( "✅ Stabilization Pass 1 complete. Integrating Pass 2 into main filter chain." ) # We will add vidstabtransform to the video chain below # optzoom=2 -> Adaptive zoom to avoid black borders # zoom=0 -> Auto zoom # smoothing=30 -> Heavy smoothing (default is 10) for handheld feel vidstab_filter = f"vidstabtransform=input={transforms_path}:smoothing=30:optzoom=2:zoom=0:interpol=bicubic,unsharp=5:5:0.8:3:3:0.4," else: print( f"⚠️ Stabilization Pass 1 failed: {detect_proc.stderr.decode()[:500]}" ) stats["stabilize_duration"] = time.time() - start_step start_step = time.time() # --- Audio Checks --- has_original_audio = False try: probe_cmd = [ "ffprobe", "-v", "error", "-select_streams", "a:0", "-show_entries", "stream=codec_type", "-of", "csv=p=0", str(input_video_path), ] probe_out = subprocess.check_output(probe_cmd).decode().strip() if probe_out == "audio": has_original_audio = True except Exception: pass # --- Inputs --- # 0: Video (already added) # 1: Music (optional) # 2: TTS (optional) input_count = 1 music_idx = -1 tts_idx = -1 if has_music: cmd.extend(["-i", str(input_audio_path)]) music_idx = input_count input_count += 1 if has_tts: cmd.extend(["-i", str(tts_audio_path)]) tts_idx = input_count input_count += 1 # --- Filter Complex Construction --- fc_parts = [] # A. Video Chain # Chain: [0:v] -> [stabilize] -> [scale/crop] -> [text] -> [vout] # 1. Stabilization (if enabled) + Scaling/Cropping # We apply stabilization FIRST on raw video, THEN crop to 9:16 # Start of video chain v_chain = "[0:v]" if vidstab_filter: v_chain += vidstab_filter # Note: vidstabtransform output is same res as input # Scale & Crop to Fill 1080x1920 (Vertical Reel) v_chain += "scale=1080:1920:force_original_aspect_ratio=increase,crop=1080:1920" # 2. Text Overlay if request.text and request.draw_text: text_filter = "" if has_tts: # Subtitles (TikTok style) using ASS (already generated in TTS block) print(f"🎬 Overlaying subtitles from TTS ASS: {tts_ass_path}") ass_path_str = str(tts_ass_path).replace("\\", "/").replace(":", "\\:") text_filter = f",subtitles='{ass_path_str}'" else: # Standard Text (without TTS) print( f"🎬 Overlaying subtitles from standard text: {request.text[:30]}..." ) std_ass_path = job_dir / "std_text.ass" # Use fontsize 40 by default for standard text generate_simple_ass(request.text, std_ass_path, font_size=40) ass_path_str = str(std_ass_path).replace("\\", "/").replace(":", "\\:") text_filter = f",subtitles='{ass_path_str}'" # Combine formatting + text v_chain += text_filter # End of video chain v_chain += "[vout]" fc_parts.append(v_chain) # B. Audio Chain audio_mapped = False inputs_for_mix = 0 audio_mix_str = "" # Strategy: # If no music and no TTS -> Copy original audio (if exists) or silent # If music or TTS -> Mix everything if has_music or has_tts: # Prepare inputs if has_original_audio: audio_mix_str += "[0:a]" inputs_for_mix += 1 if has_music: # Adjust volume fc_parts.append( f"[{music_idx}:a]volume={request.music_volume}[a_music]" ) audio_mix_str += "[a_music]" inputs_for_mix += 1 if has_tts: # TTS louder fc_parts.append(f"[{tts_idx}:a]volume=1.5[a_tts]") audio_mix_str += "[a_tts]" inputs_for_mix += 1 # Mix if inputs_for_mix > 0: fc_parts.append( f"{audio_mix_str}amix=inputs={inputs_for_mix}:duration=first:dropout_transition=2:normalize=0[aout]" ) audio_mapped = True else: # No external audio added if has_original_audio: # Just pass through original audio # We can map 0:a directly, no filter needed for audio audio_mapped = False else: # No audio at all audio_mapped = False # Apply Filter Complex cmd.extend(["-filter_complex", ";".join(fc_parts)]) # Maps cmd.extend(["-map", "[vout]"]) # Map processed video if audio_mapped: cmd.extend(["-map", "[aout]"]) # Map mixed audio elif has_original_audio: cmd.extend(["-map", "0:a"]) # Map original audio directly cmd.extend(["-shortest"]) # Quality settings cmd.extend( [ "-c:v", "libx264", "-profile:v", "high", "-r", "30", "-preset", "slow", "-crf", "17", "-b:v", "10M", "-maxrate", "12M", "-bufsize", "20M", "-c:a", "aac", "-b:a", "192k", "-pix_fmt", "yuv420p", "-movflags", "+faststart", ] ) cmd.append(str(output_video_path)) print(f"🚀 Executing FFmpeg command: {' '.join(cmd)}") # execute process = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE) # Log detailed output on failure OR success for debugging font issues if process.returncode != 0: print(f"❌ FFmpeg failed. Stderr:\n{process.stderr.decode()}") else: # Check stderr for font warnings even on success stderr_last_lines = "\n".join(process.stderr.decode().splitlines()[-20:]) print(f"✅ FFmpeg executed. Stderr (last 20 lines):\n{stderr_last_lines}") stats["encoding_duration"] = time.time() - start_step stats["total_duration"] = time.time() - start_total if process.returncode != 0: raise Exception(f"FFmpeg encoding failed: {process.stderr.decode()}") # 4. Get Duration (ffprobe) duration_cmd = [ "ffprobe", "-v", "error", "-show_entries", "format=duration", "-of", "default=noprint_wrappers=1:nokey=1", str(output_video_path), ] dur_proc = subprocess.run(duration_cmd, stdout=subprocess.PIPE) duration = float(dur_proc.stdout.decode().strip() or 0) # 5. Read Output with open(output_video_path, "rb") as f: out_bytes = f.read() out_b64 = base64.b64encode(out_bytes).decode("utf-8") # Cleanup shutil.rmtree(job_dir) print(f"📊 Processing Stats: {stats}") return { "success": True, "output_base64": out_b64, "duration": duration, "processing_stats": stats, } except Exception as e: if "job_dir" in locals(): shutil.rmtree(job_dir, ignore_errors=True) return {"success": False, "detail": str(e)} @app.post("/preview-tts") async def preview_tts(request: ReelRequest, x_api_key: str = Header(None)): if x_api_key != API_KEY: raise HTTPException(status_code=401, detail="Invalid API Key") try: job_id = str(uuid.uuid4()) job_dir = TEMP_DIR / job_id job_dir.mkdir() tts_audio_path = job_dir / "preview.mp3" tts_srt_path = job_dir / "preview.srt" # Consistent with rest of app if not request.text: raise HTTPException(status_code=400, detail="Text required for preview") clean_text = clean_text_for_tts(request.text) # Determine voice (reuse logic) voice = request.tts_voice if voice == "male": voice = "fr-FR-RemyMultilingualNeural" elif voice == "female": voice = "fr-FR-VivienneMultilingualNeural" elif not voice or "Neural" not in voice: voice = "fr-FR-VivienneMultilingualNeural" await generate_tts_with_subs(clean_text, voice, tts_audio_path, tts_srt_path) if not tts_audio_path.exists(): raise Exception("TTS generation failed (file missing)") with open(tts_audio_path, "rb") as f: audio_bytes = f.read() audio_b64 = base64.b64encode(audio_bytes).decode("utf-8") shutil.rmtree(job_dir) return {"success": True, "audio_base64": audio_b64} except Exception as e: if "job_dir" in locals(): shutil.rmtree(job_dir, ignore_errors=True) return {"success": False, "detail": str(e)} if __name__ == "__main__": uvicorn.run(app, host="0.0.0.0", port=8000)