""" Physics Chapter Video Generator - Improved Proxy Implementation Creates educational videos by combining title cards with relevant content. """ import re, shutil, subprocess, textwrap, os, tempfile from pathlib import Path from typing import List, Optional import gradio as gr import random import requests import time from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry print("CWD:", os.getcwd()) print("cookies.txt exists:", os.path.exists("./cookies/cookies.txt")) # --- Oxylabs Proxy Configuration --- PROXY_USERNAME = "ujwal_CmiMZ" PROXY_PASSWORD = "xJv4DChht5P6y+u" PROXY_COUNTRY = "US" # List of proxy endpoints (Oxylabs DC Proxies) PROXY_PORTS = [8001, 8002, 8003, 8004, 8005] PROXY_HOST = "dc.oxylabs.io" # User agents to rotate for better bot detection avoidance USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36" ] def get_random_proxy(): """Get a random proxy with proper formatting""" port = random.choice(PROXY_PORTS) return f"http://user-{PROXY_USERNAME}-country-{PROXY_COUNTRY}:{PROXY_PASSWORD}@{PROXY_HOST}:{port}" def get_session_with_proxy(): """Create a requests session with proxy and anti-bot measures""" session = requests.Session() # Set up proxy proxy_url = get_random_proxy() session.proxies = { "http": proxy_url, "https": proxy_url, } # Set up headers to avoid bot detection session.headers.update({ 'User-Agent': random.choice(USER_AGENTS), 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8', 'Accept-Language': 'en-US,en;q=0.5', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive', 'Upgrade-Insecure-Requests': '1', 'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'none', 'Cache-Control': 'max-age=0' }) # Set up retry strategy retry_strategy = Retry( total=3, backoff_factor=1, status_forcelist=[429, 500, 502, 503, 504], ) adapter = HTTPAdapter(max_retries=retry_strategy) session.mount("http://", adapter) session.mount("https://", adapter) # Disable SSL warnings session.verify = False return session # ---------------- CONFIG ---------------- TITLE_DUR = 3 # seconds SIZE = "1280x720" # resolution for cards FPS = 30 CRF = 28 # Higher CRF for smaller files in HF Spaces PRESET = "ultrafast" # Fastest encoding for HF Spaces YT_MAX_RESULTS = 2 # Reduced for faster processing MAX_VIDEO_LENGTH = 30 # Max seconds per video clip MAX_TOPICS = 8 # Limit topics for HF Spaces resources # ---------------------------------------- # ---------- helpers ---------- def run_cmd(cmd: list[str], timeout: int = 120) -> bool: """Run command with timeout and proper error handling""" try: result = subprocess.run( cmd, check=True, timeout=timeout, capture_output=True, text=True ) return True except (subprocess.CalledProcessError, subprocess.TimeoutExpired) as e: print(f"Command failed:\n{' '.join(cmd)}\nError:\n{e.stderr if hasattr(e, 'stderr') else str(e)}") return False def yt_urls(query: str, max_results: int) -> List[str]: """Get YouTube URLs for search query via proxy with better bot avoidance""" try: # Add random delay to avoid rate limiting time.sleep(random.uniform(1, 3)) # Import here to avoid issues with monkey patching from youtube_search import YoutubeSearch # Create session with proxy session = get_session_with_proxy() # Monkey-patch requests to use our session original_get = requests.get original_post = requests.post def proxied_get(*args, **kwargs): # Remove conflicting parameters kwargs.pop('proxies', None) kwargs.pop('verify', None) kwargs.pop('headers', None) return session.get(*args, **kwargs) def proxied_post(*args, **kwargs): kwargs.pop('proxies', None) kwargs.pop('verify', None) kwargs.pop('headers', None) return session.post(*args, **kwargs) # Apply monkey patch requests.get = proxied_get requests.post = proxied_post try: # Perform search with modified query to avoid detection search_query = f"physics {query} education tutorial" results = YoutubeSearch(search_query, max_results=max_results).to_dict() urls = ["https://www.youtube.com" + r["url_suffix"] for r in results] return urls finally: # Always restore original functions requests.get = original_get requests.post = original_post session.close() except Exception as e: print(f"YouTube search failed: {e}") return [] def safe_filename(name: str) -> str: """Create safe filename""" return re.sub(r"[^\w\-\.]", "_", name)[:50] # Limit length def dl_video(url: str, out: Path) -> bool: """Download video with length limit using rotating proxy and better headers""" out.parent.mkdir(exist_ok=True) proxy = get_random_proxy() # Add random delay before download time.sleep(random.uniform(0.5, 2)) cmd = [ "yt-dlp", "--match-filter", f"duration<{MAX_VIDEO_LENGTH}", "-f", "best[height<=720][ext=mp4]/best[ext=mp4]/best", "--merge-output-format", "mp4", "-o", str(out), "--no-playlist", "--proxy", proxy, "--user-agent", random.choice(USER_AGENTS), "--referer", "https://www.google.com/", "--add-header", "Accept-Language:en-US,en;q=0.9", "--add-header", "Accept-Encoding:gzip, deflate, br", "--socket-timeout", "30", "--retries", "3", "--fragment-retries", "3", # "--cookies", "./cookies/cookies.txt", url, ] return run_cmd(cmd, timeout=60) def make_card(text: str, out: Path, dur: int = TITLE_DUR) -> bool: """Create title card with text""" # Wrap text for better display wrapped = textwrap.wrap(text, width=25) safe_text = "\\n".join(w.replace("'", r"\\'") for w in wrapped) cmd = [ "ffmpeg", "-loglevel", "error", "-f", "lavfi", "-i", f"color=c=navy:s={SIZE}:d={dur}", "-f", "lavfi", "-i", "anullsrc=r=44100:cl=stereo", "-vf", ( f"drawtext=text='{safe_text}':fontcolor=white:fontsize=60:" "x=(w-text_w)/2:y=(h-text_h)/2:fontfile=/usr/share/fonts/truetype/dejavu/DejaVuSans-Bold.ttf" ), "-shortest", "-r", str(FPS), "-c:v", "libx264", "-preset", PRESET, "-crf", str(CRF), "-c:a", "aac", "-b:a", "96k", "-movflags", "+faststart", "-y", str(out), ] return run_cmd(cmd) def extract_topics(text: str) -> List[str]: """Extract topics from input text""" topics = [] for line in text.splitlines(): line = line.strip() if not line or len(topics) >= MAX_TOPICS: continue # Match numbered lists if re.match(r"^\d+[\.)]\s+.+", line): topic = re.sub(r"^\d+[\.)]\s*", "", line) topics.append(topic) # Match markdown headers elif re.match(r"^#+\s+.+", line): topic = re.sub(r"^#+\s*", "", line) topics.append(topic) # Match all caps titles elif line.isupper() and 3 <= len(line) <= 50: topics.append(line.title()) # Match regular lines as topics elif len(line) > 3 and not line.startswith(('http', 'www')): topics.append(line) return topics[:MAX_TOPICS] def create_physics_video(chapter_text: str, progress=gr.Progress()) -> Optional[str]: """Generate educational physics video from chapter topics""" if not chapter_text.strip(): return None progress(0, desc="Extracting topics...") topics = extract_topics(chapter_text) if not topics: return None # Create temporary directory with tempfile.TemporaryDirectory() as temp_dir: temp_path = Path(temp_dir) concat_paths: List[Path] = [] total_steps = len(topics) * 2 + 3 # topics * (card + video) + opening + closing + concat current_step = 0 # Opening card progress(current_step/total_steps, desc="Creating opening card...") opening = temp_path / "00_opening.mp4" if make_card("Physics Chapter Overview", opening): concat_paths.append(opening) current_step += 1 # Process each topic for idx, topic in enumerate(topics, 1): # Create title card progress(current_step/total_steps, desc=f"Creating card for: {topic[:30]}...") card = temp_path / f"title_{idx:02d}.mp4" if make_card(topic, card): concat_paths.append(card) current_step += 1 # Try to download video with delays between attempts progress(current_step/total_steps, desc=f"Searching video for: {topic[:30]}...") video_found = False urls = yt_urls(f"{topic} physics explanation", YT_MAX_RESULTS) for url_idx, url in enumerate(urls): # Add delay between video download attempts if url_idx > 0: time.sleep(random.uniform(2, 5)) vid_id_match = re.search(r"(?:v=|be/|shorts/)([\w\-]{11})", url) if not vid_id_match: continue vid_path = temp_path / f"{safe_filename(vid_id_match.group(1))}.mp4" if dl_video(url, vid_path): concat_paths.append(vid_path) video_found = True break if not video_found: # Create a placeholder card if no video found placeholder = temp_path / f"placeholder_{idx:02d}.mp4" if make_card(f"Exploring: {topic}", placeholder, dur=5): concat_paths.append(placeholder) current_step += 1 # Closing card progress(current_step/total_steps, desc="Creating closing card...") closing = temp_path / "zz_closing.mp4" if make_card("Thank you for learning!", closing): concat_paths.append(closing) current_step += 1 if len(concat_paths) < 2: return None # Create concat file list_file = temp_path / "list.txt" list_file.write_text( "".join(f"file '{p.absolute()}'\n" for p in concat_paths), encoding="utf-8" ) # Final output path output_path = "physics_chapter_video.mp4" # Concatenate videos progress(current_step/total_steps, desc="Creating final video...") cmd = [ "ffmpeg", "-loglevel", "error", "-f", "concat", "-safe", "0", "-i", str(list_file), "-c:v", "libx264", "-preset", PRESET, "-crf", str(CRF), "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", "-y", output_path, ] if run_cmd(cmd, timeout=300): return output_path return None # Gradio Interface def create_interface(): """Setup the web interface""" with gr.Blocks(title="Physics Video Generator", theme=gr.themes.Soft()) as app: gr.Markdown(""" # Physics Video Generator Transform your physics topics into engaging educational videos! This tool will: - Create professional title slides for each topic - Find relevant educational content - Combine everything into a complete video **How to use:** Enter your topics one per line, or use numbered lists, or markdown headers. """) with gr.Row(): with gr.Column(): chapter_input = gr.Textbox( label="Chapter Topics", placeholder="""Enter topics like: 1. Newton's Laws of Motion 2. Force and Acceleration 3. Momentum and Impulse 4. Energy Conservation 5. Circular Motion Or: # Kinematics # Dynamics # Thermodynamics""", lines=10, max_lines=15 ) generate_btn = gr.Button("Create Physics Video", variant="primary", size="lg") with gr.Column(): video_output = gr.Video(label="Your Physics Video") gr.Markdown(""" ### Important Notes: - Processing typically takes 2-5 minutes - Videos are optimized for educational use - Limited to 8 topics per session - Each video segment is capped at 30 seconds """) generate_btn.click( fn=create_physics_video, inputs=[chapter_input], outputs=[video_output], show_progress=True ) # Examples gr.Examples( examples=[ ["1. Newton's First Law\n2. Newton's Second Law\n3. Newton's Third Law\n4. Applications of Newton's Laws"], ["# Wave Motion\n# Sound Waves\n# Light Waves\n# Electromagnetic Spectrum"], ["THERMODYNAMICS\nHEAT TRANSFER\nENTROPY\nCARNOT CYCLE"], ["Quantum Mechanics Basics\nWave-Particle Duality\nHeisenberg Uncertainty Principle\nQuantum Tunneling"] ], inputs=[chapter_input], label="Example Topics" ) return app if __name__ == "__main__": # Disable SSL warnings import urllib3 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) app = create_interface() app.queue(max_size=3) # Limit concurrent users for HF Spaces app.launch( share=False, server_name="0.0.0.0", server_port=7860 )