Create app.py

app.py

@@ -0,0 +1,199 @@
+import os
+import shutil
+import subprocess
+import uuid
+import logging
+import asyncio
+from pathlib import Path
+
+import httpx
+from fastapi import FastAPI, HTTPException, Request, BackgroundTasks
+from fastapi.responses import JSONResponse
+from fastapi.staticfiles import StaticFiles
+
+# --- Basic Configuration ---
+# 1. Set up logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    handlers=[logging.StreamHandler()]
+)
+
+# 2. Get the Base URL for the Info API from secrets
+BASE_URL = os.getenv("BASE_URL")
+
+# 3. Define and create directories for temporary and public files
+TEMP_DIR = Path("/tmp/downloads")
+STATIC_DIR = Path("static")
+TEMP_DIR.mkdir(exist_ok=True)
+STATIC_DIR.mkdir(exist_ok=True)
+
+# 4. Cleanup configuration
+FILE_LIFETIME_SECONDS = 900  # 15 minutes
+
+# --- FastAPI App Initialization ---
+app = FastAPI(
+    title="Video Processing API v2",
+    description="Generates a temporary download link for a merged YouTube video."
+)
+
+# Mount the 'static' directory to serve files publicly
+app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
+
+
+# --- Helper Functions ---
+async def cleanup_file(filepath: Path):
+    """
+    Waits for a specified time and then deletes the file and its parent directory.
+    """
+    await asyncio.sleep(FILE_LIFETIME_SECONDS)
+    try:
+        if filepath.parent.exists():
+            shutil.rmtree(filepath.parent)
+            logging.info(f"Cleaned up directory: {filepath.parent}")
+    except Exception as e:
+        logging.error(f"Error during cleanup of {filepath.parent}: {e}")
+
+
+def get_best_formats_with_fallback(data: dict, requested_quality: int):
+    """
+    Parses the Info API response to find the best matching video format
+    with a fallback, and the best audio format.
+    """
+    if "formats" not in data:
+        raise ValueError("The 'formats' key is missing from the Info API response.")
+
+    # --- Video Selection with Fallback ---
+    video_url = None
+    # Filter for video-only, mp4 formats with a height attribute
+    video_formats = [
+        f for f in data["formats"]
+        if f.get("vcodec") != "none" and f.get("acodec") == "none" and f.get("ext") == "mp4" and f.get("height")
+    ]
+    # Sort by height, descending
+    video_formats.sort(key=lambda f: f["height"], reverse=True)
+
+    # Find the best format that is less than or equal to the requested quality
+    for f in video_formats:
+        if f["height"] <= requested_quality:
+            video_url = f["url"]
+            logging.info(f"Selected video quality: {f['height']}p (requested <= {requested_quality}p)")
+            break
+
+    # If no suitable format was found, fallback to the lowest available quality
+    if not video_url and video_formats:
+        f = video_formats[-1] # Smallest resolution
+        video_url = f["url"]
+        logging.warning(f"Requested quality not available. Falling back to lowest available: {f['height']}p")
+
+    # --- Audio Selection ---
+    audio_url = None
+    audio_formats = [
+        f for f in data["formats"]
+        if f.get("acodec") != "none" and f.get("vcodec") == "none" and "m4a" in f.get("format_id", "")
+    ]
+    audio_formats.sort(key=lambda f: f.get("abr", 0), reverse=True) # Sort by audio bitrate
+    if audio_formats:
+        audio_url = audio_formats[0]["url"]
+        logging.info(f"Selected best available audio with bitrate {audio_formats[0].get('abr', 'N/A')}k.")
+
+    if not video_url or not audio_url:
+        raise ValueError("Could not find suitable video and/or audio streams from the Info API.")
+
+    return video_url, audio_url
+
+
+# --- API Endpoints ---
+@app.get("/")
+def read_root():
+    return {"message": "Video Processing API v2 is running."}
+
+
+@app.post("/api/process")
+async def process_video(request: Request, background_tasks: BackgroundTasks):
+    """
+    Takes a URL and quality, processes the video, and returns a temporary download link.
+    """
+    if not BASE_URL:
+        logging.error("FATAL: BASE_URL is not configured in the server environment.")
+        raise HTTPException(status_code=500, detail="Server is not configured correctly.")
+
+    body = await request.json()
+    video_url = body.get("url")
+    try:
+        # Use a default quality of 1080p if not provided
+        quality = int(body.get("quality", "1080"))
+    except (ValueError, TypeError):
+        raise HTTPException(status_code=400, detail="'quality' must be a valid number (e.g., 1080, 720).")
+
+    if not video_url:
+        raise HTTPException(status_code=400, detail="A 'url' is required.")
+
+    logging.info(f"Received request for URL: {video_url} with quality: {quality}p")
+
+    # --- Step 1: Call the Info API ---
+    info_api_url = f"{BASE_URL}/api/info"
+    params = {"url": video_url, "playlist": "false"}
+    try:
+        async with httpx.AsyncClient() as client:
+            logging.info(f"Calling Info API: {info_api_url}")
+            response = await client.get(info_api_url, params=params, timeout=30.0)
+            response.raise_for_status()
+            video_data = response.json()
+            video_stream_url, audio_stream_url = get_best_formats_with_fallback(video_data, quality)
+    except httpx.RequestError as e:
+        logging.error(f"Info API connection failed: {e}")
+        raise HTTPException(status_code=502, detail=f"Failed to connect to the Info API: {e}")
+    except Exception as e:
+        logging.error(f"Error processing video info: {e}")
+        raise HTTPException(status_code=500, detail=f"An unexpected error occurred: {e}")
+
+    # --- Step 2: Download & Merge ---
+    task_id = str(uuid.uuid4())
+    # The final file will be in a public-facing static directory
+    final_output_dir = STATIC_DIR / task_id
+    final_output_dir.mkdir()
+    final_output_path = final_output_dir / "video.mp4"
+
+    # Temporary paths for raw downloads
+    video_path = TEMP_DIR / f"{task_id}_video.mp4"
+    audio_path = TEMP_DIR / f"{task_id}_audio.m4a"
+
+    try:
+        logging.info(f"Starting download for task {task_id}")
+        # Using yt-dlp to download is more robust for complex URLs
+        ydl_opts_video = {'outtmpl': str(video_path)}
+        ydl_opts_audio = {'outtmpl': str(audio_path)}
+        with YoutubeDL(ydl_opts_video) as ydl:
+            ydl.download([video_stream_url])
+        with YoutubeDL(ydl_opts_audio) as ydl:
+            ydl.download([audio_stream_url])
+
+        logging.info(f"Download complete. Starting FFmpeg merge for task {task_id}")
+        subprocess.run(
+            ['ffmpeg', '-i', str(video_path), '-i', str(audio_path), '-c', 'copy', str(final_output_path)],
+            check=True, capture_output=True, text=True
+        )
+        logging.info(f"Merge successful. Final file at: {final_output_path}")
+
+    except Exception as e:
+        logging.error(f"Download or Merge Failed for task {task_id}: {e}")
+        shutil.rmtree(final_output_dir) # Clean up public dir on failure
+        raise HTTPException(status_code=500, detail=f"Failed during file processing: {e}")
+    finally:
+        # Clean up raw downloads from /tmp
+        if video_path.exists(): video_path.unlink()
+        if audio_path.exists(): audio_path.unlink()
+
+    # --- Step 3: Generate Download URL and Schedule Cleanup ---
+    download_url = request.url_for('static', path=f"{task_id}/video.mp4")
+    background_tasks.add_task(cleanup_file, final_output_path)
+
+    logging.info(f"Responding with download URL: {download_url}")
+    return JSONResponse(
+        content={
+            "success": True,
+            "download_url": str(download_url),
+            "expires_in": f"{FILE_LIFETIME_SECONDS} seconds"
+        }
+    )