Create app.py

app.py

@@ -0,0 +1,277 @@
+from fastapi import FastAPI, Request, Query, BackgroundTasks
+from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
+from fastapi.middleware.cors import CORSMiddleware
+import yt_dlp
+import os
+import asyncio
+from datetime import datetime, timedelta
+from urllib.parse import quote
+import io
+import logging
+import json
+import time
+from collections import defaultdict
+import uvicorn
+import gradio as gr
+
+OUTPUT_DIR = "output"
+os.makedirs(OUTPUT_DIR, exist_ok=True)
+PROXY_URL = os.environ.get("YTDLP_PROXY")
+BANNED_IP_FILE = "bannedip.json"
+MAX_REQUESTS_PER_MINUTE = 35
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+server_start_time = time.time()
+total_request_count = 0
+request_counter = defaultdict(list)
+
+
+def load_banned_ips():
+    try:
+        with open(BANNED_IP_FILE, "r") as f:
+            return set(json.load(f))
+    except (FileNotFoundError, json.JSONDecodeError):
+        return set()
+
+
+def save_banned_ips(banned_ips):
+    with open(BANNED_IP_FILE, "w") as f:
+        json.dump(list(banned_ips), f)
+
+
+async def delete_file_after_delay(file_path: str, delay: int = 600):
+    await asyncio.sleep(delay)
+    try:
+        os.remove(file_path)
+        logger.info(f"Deleted file {file_path} after {delay} seconds.")
+    except FileNotFoundError:
+        logger.warning(f"File {file_path} not found during deletion.")
+    except Exception as e:
+        logger.error(f"Error deleting file {file_path}: {e}", exc_info=True)
+
+
+app = FastAPI(
+    title="YouTube Downloader API",
+    description="API to download video/audio from YouTube.",
+    version="1.0.0",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+@app.middleware("http")
+async def log_requests(request: Request, call_next):
+    global total_request_count
+    total_request_count += 1
+
+    client_ip = (
+        request.headers.get("X-Forwarded-For", request.client.host)
+        .split(",")[0]
+        .strip()
+    )
+    banned_ips = load_banned_ips()
+
+    if client_ip in banned_ips:
+        return JSONResponse(
+            status_code=403, content={"message": "Blocked due to excessive requests."}
+        )
+
+    now = datetime.now()
+    request_counter[client_ip].append(now)
+    request_counter[client_ip] = [
+        t for t in request_counter[client_ip] if t > now - timedelta(minutes=1)
+    ]
+
+    if len(request_counter[client_ip]) > MAX_REQUESTS_PER_MINUTE:
+        logger.warning(f"IP {client_ip} is blocked due to rate limiting.")
+        banned_ips.add(client_ip)
+        save_banned_ips(banned_ips)
+        request_counter[client_ip] = []
+
+    logger.info(f"{client_ip} requested {request.method} {request.url}")
+    response = await call_next(request)
+    return response
+
+
+@app.get("/", summary="Root Endpoint")
+async def root():
+    return JSONResponse(status_code=200, content={"status": "Server Running"})
+
+
+@app.get("/ytv/", summary="Download YouTube Video")
+async def download_video(
+    background_tasks: BackgroundTasks,
+    url: str = Query(...),
+    quality: int = Query(720),
+    mode: str = Query("url"),
+):
+    if mode not in ["url", "buffer"]:
+        return JSONResponse(
+            status_code=400, content={"error": "Invalid mode. Use 'url' or 'buffer'."}
+        )
+
+    try:
+        resx = f"_{quality}p"
+        ydl_opts = {
+            "format": f"bestvideo[height<={quality}]+bestaudio/best",
+            "outtmpl": os.path.join(OUTPUT_DIR, "%(title)s" + resx + ".%(ext)s"),
+            "merge_output_format": "mp4",
+            "cookiefile": "yt.txt",
+            "postprocessors": [
+                {"key": "FFmpegVideoConvertor", "preferedformat": "mp4"}
+            ],
+        }
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+            file_path = ydl.prepare_filename(info)
+        file_ext = os.path.splitext(file_path)[-1].lstrip(".")
+        if not os.path.exists(file_path):
+            raise FileNotFoundError(f"File not found after download: {file_path}")
+        background_tasks.add_task(delete_file_after_delay, file_path)
+        if mode == "url":
+            return {
+                "status": "success",
+                "title": info["title"],
+                "thumb": info.get("thumbnail"),
+                "url": f"https://lordxdd-ytdlp.hf.space/cdn/video/{quote(os.path.basename(file_path))}",
+            }
+        media_type = f"video/{file_ext}" if file_ext else "video/mp4"
+        with open(file_path, "rb") as f:
+            return StreamingResponse(
+                io.BytesIO(f.read()),
+                media_type=media_type,
+                headers={
+                    "Content-Disposition": f"attachment; filename={os.path.basename(file_path)}"
+                },
+            )
+    except Exception as e:
+        logger.error(f"Download error: {e}", exc_info=True)
+        return JSONResponse(status_code=500, content={"error": str(e)})
+
+
+@app.get("/cdn/video/{filename}", summary="Serve Downloaded File")
+async def download_file(filename: str):
+    file_path = os.path.join(OUTPUT_DIR, filename)
+    if os.path.exists(file_path):
+        file_ext = os.path.splitext(filename)[-1].lstrip(".")
+        media_type = f"video/{file_ext}" if file_ext else "video/mp4"
+        return FileResponse(file_path, filename=filename, media_type=media_type)
+    return JSONResponse(status_code=404, content={"error": "File not found"})
+
+
+@app.get("/yta/", summary="Download YouTube Audio (.mp3)")
+async def download_audio(
+    background_tasks: BackgroundTasks,
+    url: str = Query(..., description="YouTube video URL"),
+    mode: str = Query("url", description="Response mode: 'url' or 'buffer'"),
+):
+    if mode not in {"url", "buffer"}:
+        return JSONResponse(
+            status_code=400, content={"error": "Invalid mode. Use 'url' or 'buffer'."}
+        )
+    ydl_opts = {
+        "format": "bestaudio/best",
+        "outtmpl": os.path.join(OUTPUT_DIR, "%(title)s.%(ext)s"),
+        "cookiefile": "yt.txt",
+        "noplaylist": True,
+        "no_warnings": True,
+        "proxy": PROXY_URL,
+        "postprocessors": [
+            {
+                "key": "FFmpegExtractAudio",
+                "preferredcodec": "mp3",
+                "preferredquality": "128",
+            }
+        ],
+        "postprocessor_args": ["-vn", "-preset", "ultrafast", "-threads", "4"],
+        "prefer_ffmpeg": True,
+        "noplaylist": True,
+        "no_warnings": True,
+    }
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+        if info.get("requested_downloads"):
+            file_path = info["requested_downloads"][0]["filepath"]
+        else:
+            base, _ = os.path.splitext(ydl.prepare_filename(info))
+            file_path = base + ".m4a"
+        if not os.path.exists(file_path):
+            raise FileNotFoundError(f"File not found: {file_path}")
+        background_tasks.add_task(delete_file_after_delay, file_path)
+        if mode == "url":
+            return {
+                "status": "success",
+                "title": info.get("title"),
+                "thumb": info.get("thumbnail"),
+                "url": f"https://lordofc-ytdlp.hf.space/cdn/audio/{quote(os.path.basename(file_path))}",
+            }
+
+        with open(file_path, "rb") as f:
+            data = f.read()
+        headers = {
+            "Content-Disposition": f'attachment; filename="{os.path.basename(file_path)}"'
+        }
+        return StreamingResponse(
+            io.BytesIO(data), media_type="audio/m4a", headers=headers
+        )
+
+    except Exception as e:
+        logger.error(f"Download error: {e}", exc_info=True)
+        return JSONResponse(status_code=500, content={"error": str(e)})
+
+
+@app.get("/cdn/audio/{filename}", summary="Serve Downloaded Audio File")
+async def serve_audio_file(filename: str):
+    file_path = os.path.join(OUTPUT_DIR, filename)
+    if os.path.exists(file_path):
+        ext = os.path.splitext(filename)[-1].lstrip(".")
+        return FileResponse(file_path, filename=filename, media_type=f"audio/{ext}")
+    return JSONResponse(status_code=404, content={"error": "File not found"})
+
+
+def run_gradio_ui():
+    def download_gradio(url, resolution):
+        try:
+            quality = int(resolution)
+            ydl_opts = {
+                "format": f"bestvideo[height<={quality}]+bestaudio/best",
+                "outtmpl": os.path.join(
+                    OUTPUT_DIR, "%(title)s" + f"_{quality}p" + ".%(ext)s"
+                ),
+                "merge_output_format": "mp4",
+                "postprocessors": [
+                    {"key": "FFmpegVideoConvertor", "preferedformat": "mp4"}
+                ],
+            }
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                info = ydl.extract_info(url, download=True)
+                file_path = ydl.prepare_filename(info)
+            return {
+                "status": "success",
+                "title": info["title"],
+                "url": f"/cdn/video/{quote(os.path.basename(file_path))}",
+            }
+        except Exception as e:
+            return {"error": str(e)}
+
+    interface = gr.Interface(
+        fn=download_gradio,
+        inputs=["text", gr.Slider(240, 1080)],
+        outputs="json",
+        title="YouTube Video Downloader",
+    )
+    interface.launch()
+
+
+if __name__ == "__main__":
+    uvicorn.run("app:app", host="0.0.0.0", port=7860)