""" Vidsimplify Streamlit Application Supports two modes: 1. DIRECT MODE (for Hugging Face) - directly executes generation code 2. API MODE (for local development) - uses the FastAPI server Set DIRECT_MODE=true in environment to use direct execution. """ import streamlit as st import os import time import base64 import uuid import subprocess import asyncio import threading import shutil import logging from pathlib import Path from datetime import datetime # Setup logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Load environment variables from .env file from dotenv import load_dotenv load_dotenv() # Check if we're in direct mode (for Hugging Face deployment) DIRECT_MODE = os.getenv("DIRECT_MODE", "true").lower() == "true" # ============================================================================ # Background Cleanup Task (runs every 30 minutes) # ============================================================================ def cleanup_old_files(): """ Background task that runs every 10 minutes to delete files older than 10 minutes. This prevents running out of the 1GB storage limit on Hugging Face Spaces. """ import time while True: try: time.sleep(600) # Wait 10 minutes (600 seconds) cutoff_time = time.time() - 600 # 10 minutes ago deleted_count = 0 freed_mb = 0 logger.info("🧹 Starting periodic cleanup (files older than 10 minutes)...") # Clean up old video files media_videos = Path("media/videos") if media_videos.exists(): for scene_dir in media_videos.iterdir(): if scene_dir.is_dir() and scene_dir.name.startswith("scene_"): # Check if directory is older than 10 minutes if scene_dir.stat().st_mtime < cutoff_time: dir_size = sum(f.stat().st_size for f in scene_dir.rglob('*') if f.is_file()) shutil.rmtree(scene_dir) deleted_count += 1 freed_mb += dir_size / (1024 * 1024) logger.info(f" 🗑️ Deleted old scene: {scene_dir.name}") # Clean up old voiceover files (keep newest 50, delete older ones) for vo_dir in [Path("media/voiceover/edge_tts"), Path("media/voiceover/elevenlabs"), Path("media/voiceover/gtts")]: if vo_dir.exists(): voice_files = sorted(vo_dir.glob("*.mp3"), key=lambda x: x.stat().st_mtime, reverse=True) # Keep newest 50, delete the rest for old_file in voice_files[50:]: try: file_size = old_file.stat().st_size old_file.unlink() freed_mb += file_size / (1024 * 1024) logger.info(f" 🗑️ Deleted old voiceover: {old_file.name}") except: pass # Log disk usage total, used, free = shutil.disk_usage("/") free_mb = free // (1024 * 1024) logger.info(f"✅ Cleanup complete: Deleted {deleted_count} scenes, freed {freed_mb:.1f}MB") logger.info(f"💾 Disk space: {free_mb}MB free of {total // (1024 * 1024)}MB") except Exception as e: logger.error(f"❌ Cleanup error: {e}") # Start cleanup thread in DIRECT_MODE (Hugging Face) if DIRECT_MODE: cleanup_thread = threading.Thread(target=cleanup_old_files, daemon=True) cleanup_thread.start() logger.info("🧹 Background cleanup task started (runs every 10 minutes)") # Page config st.set_page_config( page_title="Vidsimplify - AI Video Generator", page_icon="🎬", layout="wide", initial_sidebar_state="expanded" ) # Custom CSS - Premium Techy Design st.markdown(""" """, unsafe_allow_html=True) # ============================================================================ # Direct Mode Functions (No API needed) # ============================================================================ def generate_video_direct(input_type: str, input_data: str, quality: str, category: str, progress_callback=None): """ Generate video directly without API. Used for Hugging Face deployment. """ from manimator.api.animation_generation import generate_animation_response from manimator.utils.code_fixer import CodeFixer job_id = str(uuid.uuid4()) base_dir = Path(__file__).parent # Quality flags for Manim quality_flags = { "low": "-pql", "medium": "-pqm", "high": "-pqh", "ultra": "-pqk" } quality_dirs = { "low": "480p15", "medium": "720p30", "high": "1080p60", "ultra": "2160p60" } try: # Stage 1: Generate code if progress_callback: progress_callback("generating_code", 20, "Generating Manim code with AI...") code = generate_animation_response( input_data=input_data, input_type=input_type, category=category, job_id=job_id ) # Save code to file scene_name = f"Scene_{uuid.uuid4().hex[:8]}" code_file = base_dir / f"scene_{job_id}.py" # Replace class name in code import re code = re.sub(r'class\s+\w+\s*\(\s*VoiceoverScene\s*\)', f'class {scene_name}(VoiceoverScene)', code) with open(code_file, 'w') as f: f.write(code) if progress_callback: progress_callback("code_generated", 40, "Code generated successfully!") # Stage 2: Render video with retry loop fixer = CodeFixer() max_retries = 3 video_path = None # Ensure media directories exist media_dir = base_dir / "media" (media_dir / "voiceover" / "elevenlabs").mkdir(parents=True, exist_ok=True) (media_dir / "voiceover" / "gtts").mkdir(parents=True, exist_ok=True) (media_dir / "voiceover" / "edge_tts").mkdir(parents=True, exist_ok=True) (media_dir / "videos").mkdir(parents=True, exist_ok=True) for attempt in range(max_retries): try: if progress_callback: progress_callback("rendering", 50 + (attempt * 10), f"Rendering video (attempt {attempt + 1}/{max_retries})...") # Run Manim cmd = [ "manim", quality_flags[quality], "--media_dir", str(media_dir), str(code_file), scene_name ] env = os.environ.copy() env["MEDIA_DIR"] = str(media_dir.resolve()) result = subprocess.run( cmd, capture_output=True, text=True, cwd=str(base_dir), env=env, timeout=300 # 5 minute timeout ) if result.returncode != 0: error_msg = result.stderr[-500:] if result.stderr else "Unknown render error" raise Exception(f"Manim rendering failed: {error_msg}") # Find video file video_dir = media_dir / "videos" / code_file.stem / quality_dirs[quality] video_files = list(video_dir.glob("*.mp4")) if not video_files: raise Exception(f"No video file found in {video_dir}") video_path = video_files[0] break # Success! except Exception as e: if attempt < max_retries - 1: # Try to fix code if progress_callback: progress_callback("fixing", 50 + (attempt * 10), "Fixing rendering error...") with open(code_file, 'r') as f: current_code = f.read() fixed_code = fixer.fix_runtime_error(current_code, str(e)) with open(code_file, 'w') as f: f.write(fixed_code) else: raise e # Cleanup code file try: code_file.unlink() except: pass if progress_callback: progress_callback("completed", 100, "Video generation completed!") return video_path except Exception as e: if progress_callback: progress_callback("failed", 0, f"Error: {str(e)}") raise e def generate_via_api(api_url: str, input_type: str, input_data: str, quality: str, category: str, progress_callback=None): """ Generate video via API (for local development). """ import requests payload = { "input_type": input_type, "input_data": input_data, "quality": quality, "category": category } response = requests.post(f"{api_url}/api/videos", json=payload, timeout=30) if response.status_code != 200: raise Exception(f"Failed to start job: {response.text}") job_data = response.json() job_id = job_data["job_id"] # Poll for status while True: status_response = requests.get(f"{api_url}/api/jobs/{job_id}", timeout=10) if status_response.status_code == 200: status_data = status_response.json() status = status_data["status"] progress = status_data.get("progress", {}) percentage = progress.get("percentage", 0) message = progress.get("message", "Processing...") if progress_callback: progress_callback(status, percentage, message) if status == "completed": return f"{api_url}/api/videos/{job_id}" elif status == "failed": raise Exception(status_data.get("error", "Unknown error")) time.sleep(2) # ============================================================================ # Sidebar # ============================================================================ with st.sidebar: # Logo and branding st.markdown("""

✨ Vidsimplify

Precision Animations

""", unsafe_allow_html=True) st.markdown("---") # Show mode indicator if DIRECT_MODE: st.success("🚀 Cloud Mode") else: st.info("🔌 Local Development") api_url = st.text_input("API URL", value="http://localhost:8000") st.markdown("### ⚙️ Settings") # Quality - default to HIGH (index=2) quality = st.selectbox( "Video Quality", ["low", "medium", "high"], index=2, # Default to HIGH format_func=lambda x: {"low": "⚡ Fast (480p)", "medium": "🎥 Standard (720p)", "high": "✨ Premium (1080p)"}[x] ) # Category with better labels category = st.selectbox( "Animation Style", ["tech_system", "product_startup", "mathematical"], index=0, format_func=lambda x: {"tech_system": "🔧 Tech & Systems", "product_startup": "🚀 Product & Startup", "mathematical": "📐 Mathematical"}[x] ) # Voice selection for TTS voice_options = { "af_heart": "❤️ Heart (Female, Warm)", "af_bella": "💼 Bella (Female, Professional)", "af_nicole": "🎤 Nicole (Female, Clear)", "af_sarah": "🌸 Sarah (Female, Natural)", "am_adam": "👔 Adam (Male, Professional)", "am_michael": "🎙️ Michael (Male, Deep)", "bf_emma": "🇬🇧 Emma (British Female)", "bf_isabella": "🇬🇧 Isabella (British Female)", "bm_george": "🇬🇧 George (British Male)", "bm_lewis": "🇬🇧 Lewis (British Male)", } voice = st.selectbox( "🎙️ Voice", list(voice_options.keys()), index=0, format_func=lambda x: voice_options[x] ) st.markdown("---") # About section - professional branding st.markdown("""

About Vidsimplify

Transform your idea into precision animation.

""", unsafe_allow_html=True) # Footer in sidebar st.markdown("""

Made with 💜 by Vidsimplify Team

""", unsafe_allow_html=True) # ============================================================================ # Main Content # ============================================================================ # Hero section st.markdown("""

Create Precision Animations

Transform text, blogs, and documents into beautiful animated videos

""", unsafe_allow_html=True) # Input tabs tab1, tab2, tab3 = st.tabs(["📝 Text / Script", "🔗 Blog / URL", "📄 PDF Document"]) input_type = "text" input_data = "" with tab1: st.header("Text Input") st.markdown("Paste your script, blog post, or long text here.") text_input = st.text_area("Content", height=300, placeholder="Enter your text here...") if text_input: input_type = "text" input_data = text_input with tab2: st.header("URL Input") st.markdown("Enter the URL of a blog post or article.") url_input = st.text_input("URL", placeholder="https://example.com/blog-post") if url_input: input_type = "url" input_data = url_input with tab3: st.header("PDF Upload") st.markdown("Upload a PDF document to generate a video from.") uploaded_file = st.file_uploader("Choose a PDF file", type="pdf") if uploaded_file: input_type = "pdf" bytes_data = uploaded_file.getvalue() base64_pdf = base64.b64encode(bytes_data).decode('utf-8') input_data = base64_pdf # ============================================================================ # Generate Button & Logic # ============================================================================ # Add spacing before button st.markdown("
", unsafe_allow_html=True) if st.button("✨ Create Animation", type="primary", use_container_width=True): if not input_data: st.error("Please provide input data.") else: # Create status containers progress_bar = st.progress(0) status_container = st.empty() video_container = st.empty() def update_progress(status, percentage, message): progress_bar.progress(min(percentage / 100, 1.0)) status_class = { "generating_code": "status-generating", "code_generated": "status-generating", "rendering": "status-rendering", "fixing": "status-rendering", "completed": "status-complete", "failed": "status-error" }.get(status, "status-generating") status_container.markdown( f'

📊 {status.upper()}: {message}

', unsafe_allow_html=True ) try: if DIRECT_MODE: # Direct execution (Hugging Face) update_progress("starting", 5, "Initializing video generation...") video_path = generate_video_direct( input_type=input_type, input_data=input_data, quality=quality, category=category, progress_callback=update_progress ) # Display video from local file st.balloons() with open(video_path, 'rb') as video_file: video_bytes = video_file.read() video_container.video(video_bytes) # Offer download st.download_button( label="📥 Download Video", data=video_bytes, file_name=f"vidsimplify_{datetime.now().strftime('%Y%m%d_%H%M%S')}.mp4", mime="video/mp4" ) else: # API mode (local development) import requests update_progress("submitting", 5, "Submitting job to API...") video_url = generate_via_api( api_url=api_url, input_type=input_type, input_data=input_data, quality=quality, category=category, progress_callback=update_progress ) st.balloons() video_container.video(video_url) except Exception as e: error_msg = str(e) st.error(f"❌ Generation failed: {error_msg}") # Show helpful message for common errors if "Could not connect" in error_msg or "Connection" in error_msg: st.warning("💡 If running locally, make sure the API server is running: `python api_server.py`") # ============================================================================ # Footer # ============================================================================ st.markdown("""

""", unsafe_allow_html=True)