import os import re import time import threading import uuid import random 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") CACHE_DIR = os.path.join(os.path.dirname(__file__), ".yt-dlp-cache") os.makedirs(DOWNLOAD_DIR, exist_ok=True) os.makedirs(CACHE_DIR, exist_ok=True) jobs = {} AUDIO_FORMATS = ["mp3", "aac", "m4a", "opus", "flac", "wav", "vorbis", "alac"] VIDEO_FORMATS = ["mp4", "mkv", "webm", "mov", "avi", "flv"] # --------------------------------------------------------------------------- # In-memory info cache — avoids hitting YouTube again for the same URL # within INFO_CACHE_TTL seconds (10 minutes). # --------------------------------------------------------------------------- _info_cache = {} # url -> {"data": {...}, "ts": float} _info_lock = threading.Lock() INFO_CACHE_TTL = 600 # seconds def _cache_get(url): with _info_lock: entry = _info_cache.get(url) if entry and time.time() - entry["ts"] < INFO_CACHE_TTL: return entry["data"] return None def _cache_set(url, data): with _info_lock: _info_cache[url] = {"data": data, "ts": time.time()} # --------------------------------------------------------------------------- # Rotate through realistic browser User-Agents so repeated requests don't # look like a single bot fingerprint to YouTube. # --------------------------------------------------------------------------- _USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36", "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36", "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0", "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_4_1) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Safari/605.1.15", ] def _random_ua(): return random.choice(_USER_AGENTS) # --------------------------------------------------------------------------- # Base yt-dlp options shared by all calls. # # • cachedir – persists solved JS sig/nsig to disk; node only re-runs # when YouTube rotates their player (every few days). # • sleep_interval / # max_sleep_interval– small random sleep between fragment requests; # makes traffic pattern look more human. # • http_chunk_size – stream in 10 MB chunks rather than one giant request; # reduces the chance of a mid-transfer 429. # • js_runtimes – use local node v20+ for JS challenge solving. # --------------------------------------------------------------------------- def _base_opts(extra=None): opts = { "cachedir": CACHE_DIR, "js_runtimes": {"node": {"executable": "/usr/bin/node"}}, "http_headers": {"User-Agent": _random_ua()}, "sleep_interval": 1, "max_sleep_interval": 3, "http_chunk_size": 10 * 1024 * 1024, # 10 MB "socket_timeout": 30, "retries": 6, "fragment_retries": 6, } if extra: opts.update(extra) return opts def sanitize_filename(name): return re.sub(r'[\\/*?:"<>|]', "_", name) # --------------------------------------------------------------------------- # Background download worker # --------------------------------------------------------------------------- def run_download(job_id, url, fmt, chosen_format): 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: jobs[job_id]["progress"] = int(downloaded / total * 100) 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: common = _base_opts({"progress_hooks": [progress_hook]}) if fmt == "audio": codec = chosen_format or "mp3" ydl_opts = { **common, "format": "bestaudio/best", "outtmpl": os.path.join(DOWNLOAD_DIR, f"{job_id}_%(title)s.%(ext)s"), "postprocessors": [{ "key": "FFmpegExtractAudio", "preferredcodec": codec, "preferredquality": "192", }], } else: video_fmt = chosen_format or "mp4" ydl_opts = { **common, "format": "bestvideo+bestaudio/best", "outtmpl": os.path.join(DOWNLOAD_DIR, f"{job_id}_%(title)s.%(ext)s"), "merge_output_format": video_fmt, "postprocessors": [{ "key": "FFmpegVideoConvertor", "preferedformat": video_fmt, }], } with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=True) title = info.get("title", "video") 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) # --------------------------------------------------------------------------- # Routes # --------------------------------------------------------------------------- @app.route("/") def index(): return render_template("index.html") @app.route("/formats", methods=["GET"]) def get_formats(): return jsonify({"audio": AUDIO_FORMATS, "video": VIDEO_FORMATS}) @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 # Serve from cache if still fresh — no YouTube request at all cached = _cache_get(url) if cached: return jsonify(cached) try: ydl_opts = _base_opts({"skip_download": True}) with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=False) result = { "title": info.get("title"), "thumbnail": info.get("thumbnail"), "duration": info.get("duration"), "uploader": info.get("uploader"), } _cache_set(url, result) return jsonify(result) 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") chosen_format = data.get("chosen_format", None) 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, chosen_format), daemon=True, ) t.start() return jsonify({"job_id": job_id}) @app.route("/status/") 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/") 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__": # use_reloader=False prevents the Werkzeug reloader from forking the # process and accidentally firing double requests during development. app.run(host="0.0.0.0", port=5000, debug=True, use_reloader=False)