app.py

"""
Physics Chapter Video Generator
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

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"

def get_random_proxy():
    port = random.choice(PROXY_PORTS)
    return f"http://user-{PROXY_USERNAME}-country-{PROXY_COUNTRY}:{PROXY_PASSWORD}@{PROXY_HOST}:{port}"


# ---------------- 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"""
    try:
        import requests
        from youtube_search import YoutubeSearch

        proxy_url = get_random_proxy()
        proxies = {
            "http": proxy_url,
            "https": proxy_url,
        }

        # Monkey-patch YoutubeSearch to use proxy
        original_get = requests.get

        def proxied_get(*args, **kwargs):
            kwargs["proxies"] = proxies
            kwargs["verify"] = False
            return original_get(*args, **kwargs)

        requests.get = proxied_get
        results = YoutubeSearch(query, max_results=max_results).to_dict()
        requests.get = original_get  # Restore

        return ["https://www.youtube.com" + r["url_suffix"] for r in results]
    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"""
    out.parent.mkdir(exist_ok=True)
    proxy = get_random_proxy()

    cmd = [
        "yt-dlp",
        "--match-filter", f"duration<{MAX_VIDEO_LENGTH}",
        "-f", "mp4",
        "--merge-output-format", "mp4",
        "-o", str(out),
        "--no-playlist",
        # "--no-check-certificate",
        "--proxy", proxy,
        "--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
            progress(current_step/total_steps, desc=f"Searching video for: {topic[:30]}...")
            video_found = False
            
            for url in yt_urls(f"{topic} physics explanation", YT_MAX_RESULTS):
                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__":
    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
    )