VideoDownloader/app.py

import os
import re
import json
import threading
import uuid
from flask import Flask, request, jsonify, send_file, render_template
import yt_dlp

app = Flask(__name__)

DOWNLOAD_DIR = os.path.join(os.path.dirname(__file__), "downloads")
os.makedirs(DOWNLOAD_DIR, exist_ok=True)

# Track job progress
jobs = {}

def sanitize_filename(name):
    return re.sub(r'[\\/*?:"<>|]', "_", name)

def run_download(job_id, url, fmt, quality):
    jobs[job_id]["status"] = "downloading"
    jobs[job_id]["progress"] = 0

    def progress_hook(d):
        if d["status"] == "downloading":
            total = d.get("total_bytes") or d.get("total_bytes_estimate", 0)
            downloaded = d.get("downloaded_bytes", 0)
            if total:
                pct = int(downloaded / total * 100)
                jobs[job_id]["progress"] = pct
                jobs[job_id]["speed"] = d.get("_speed_str", "")
                jobs[job_id]["eta"] = d.get("_eta_str", "")
        elif d["status"] == "finished":
            jobs[job_id]["progress"] = 99

    try:
        import shutil

        common = {
            "progress_hooks": [progress_hook],
            "quiet": True,
            "extractor_args": {"youtube": {
                "js_runtimes": [f"node:/usr/bin/node"]
            }},
        }

        if fmt == "audio":
            ydl_opts = {
                **common,
                "format": "bestaudio/best",
                "outtmpl": os.path.join(DOWNLOAD_DIR, f"{job_id}_%(title)s.%(ext)s"),
                "postprocessors": [{
                    "key": "FFmpegExtractAudio",
                    "preferredcodec": quality or "mp3",
                    "preferredquality": "192",
                }],
            }
        else:
            res = quality or "best"
            if res == "best":
                fmt_str = "bestvideo[ext=mp4]+bestaudio[ext=m4a]/best[ext=mp4]/best"
            else:
                h = res.replace("p", "")
                fmt_str = f"bestvideo[height<={h}][ext=mp4]+bestaudio[ext=m4a]/best[height<={h}][ext=mp4]/best"

            ydl_opts = {
                **common,
                "format": fmt_str,
                "outtmpl": os.path.join(DOWNLOAD_DIR, f"{job_id}_%(title)s.%(ext)s"),
                "merge_output_format": "mp4",
            }

        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=True)
            title = info.get("title", "video")

        # Find the downloaded file
        files = [f for f in os.listdir(DOWNLOAD_DIR) if f.startswith(job_id)]
        if files:
            jobs[job_id]["status"] = "done"
            jobs[job_id]["progress"] = 100
            jobs[job_id]["filename"] = files[0]
            jobs[job_id]["title"] = title
        else:
            jobs[job_id]["status"] = "error"
            jobs[job_id]["error"] = "File not found after download."

    except Exception as e:
        jobs[job_id]["status"] = "error"
        jobs[job_id]["error"] = str(e)


@app.route("/")
def index():
    return render_template("index.html")


@app.route("/info", methods=["POST"])
def get_info():
    data = request.get_json()
    url = data.get("url", "").strip()
    if not url:
        return jsonify({"error": "No URL provided"}), 400
    try:
        import shutil
        node_bin = shutil.which("node") or shutil.which("nodejs") or "/usr/bin/node"
        ydl_opts = {
            "quiet": True,
            "skip_download": True,
            "extractor_args": {"youtube": {
                "js_runtimes": [f"node:{node_bin}"]
            }},
        }
        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
            info = ydl.extract_info(url, download=False)
        formats = info.get("formats", [])
        heights = sorted(set(
            f.get("height") for f in formats
            if f.get("height") and f.get("vcodec") != "none"
        ), reverse=True)
        return jsonify({
            "title": info.get("title"),
            "thumbnail": info.get("thumbnail"),
            "duration": info.get("duration"),
            "uploader": info.get("uploader"),
            "resolutions": [f"{h}p" for h in heights[:6]],
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 400


@app.route("/download", methods=["POST"])
def start_download():
    data = request.get_json()
    url = data.get("url", "").strip()
    fmt = data.get("format", "video")
    quality = data.get("quality", "best")

    if not url:
        return jsonify({"error": "No URL provided"}), 400

    job_id = str(uuid.uuid4())[:8]
    jobs[job_id] = {"status": "queued", "progress": 0}

    t = threading.Thread(target=run_download, args=(job_id, url, fmt, quality), daemon=True)
    t.start()

    return jsonify({"job_id": job_id})


@app.route("/status/<job_id>")
def job_status(job_id):
    job = jobs.get(job_id)
    if not job:
        return jsonify({"error": "Job not found"}), 404
    return jsonify(job)


@app.route("/file/<job_id>")
def download_file(job_id):
    job = jobs.get(job_id)
    if not job or job.get("status") != "done":
        return jsonify({"error": "File not ready"}), 404
    filepath = os.path.join(DOWNLOAD_DIR, job["filename"])
    if not os.path.exists(filepath):
        return jsonify({"error": "File missing"}), 404
    return send_file(filepath, as_attachment=True, download_name=job["filename"].split("_", 1)[-1])


if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000, debug=True)