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) # Sanity-check write permissions at startup _test = os.path.join(DOWNLOAD_DIR, ".write_test") try: open(_test, "w").close() os.remove(_test) except OSError as e: raise RuntimeError(f"DOWNLOAD_DIR is not writable: {e}") jobs = {} AUDIO_FORMATS = ["mp3", "aac", "m4a", "opus", "flac", "wav", "vorbis", "alac"] VIDEO_FORMATS = ["mp4", "mkv", "webm", "mov", "avi", "flv"] # --------------------------------------------------------------------------- # Maps each output container to the audio codec FFmpeg should use when # merging. "copy" means the audio stream is passed through without # re-encoding (safe when the container supports the source codec). # Containers like mp4/mov/avi are strict about which audio codecs they # accept, so we re-encode to a known-compatible codec for those. # --------------------------------------------------------------------------- CONTAINER_AUDIO = { "mp4": "aac", # opus/webm audio is incompatible with mp4 "mov": "aac", # same restriction as mp4 "avi": "mp3", # avi supports mp3 broadly "flv": "aac", # flv works with aac "mkv": "copy", # mkv supports virtually every codec "webm": "copy", # webm+opus/vorbis is fine as-is } # --------------------------------------------------------------------------- # In-memory info cache — avoids hitting YouTube again for the same URL # within INFO_CACHE_TTL seconds (10 minutes). # --------------------------------------------------------------------------- _info_cache = {} _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) # --------------------------------------------------------------------------- # Cookies file — place a cookies.txt (Netscape format, exported from your # browser while logged into YouTube) next to app.py to authenticate requests. # The file is optional; if absent yt-dlp falls back to anonymous mode. # --------------------------------------------------------------------------- COOKIES_FILE = os.path.join(os.path.dirname(__file__), "cookies.txt") # --------------------------------------------------------------------------- # Base yt-dlp options shared by all calls. # --------------------------------------------------------------------------- def _base_opts(extra=None): opts = { "cachedir": CACHE_DIR, "js_runtimes": {"node": {"executable": "/usr/bin/node"}}, # MUST be a list. Allows yt-dlp to download the JS solver via Node. "remote_components": ["ejs:github"], # Bypass the broken 'tv_downgraded' client when using cookies. # Forces fallback to default or web_embedded clients. "extractor_args": { "youtube": ["player_client=default,web_embedded"] }, "http_headers": {"User-Agent": _random_ua()}, "http_chunk_size": 10 * 1024 * 1024, "socket_timeout": 30, "retries": 6, "fragment_retries": 6, "source_address": "0.0.0.0", } if os.path.isfile(COOKIES_FILE): opts["cookiefile"] = COOKIES_FILE 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": # Sanitize codec choice — fall back to mp3 if unrecognised codec = chosen_format if chosen_format in AUDIO_FORMATS else "mp3" ydl_opts = { **common, "format": "bestaudio/best", # Use only the job_id in the filename to avoid OS-level # permission errors caused by special characters in titles. "outtmpl": os.path.join(DOWNLOAD_DIR, f"{job_id}.%(ext)s"), "tmpdir": DOWNLOAD_DIR, "postprocessors": [{ "key": "FFmpegExtractAudio", "preferredcodec": codec, "preferredquality": "192", }], } else: # Sanitize container choice — fall back to mp4 if unrecognised video_fmt = chosen_format if chosen_format in VIDEO_FORMATS else "mp4" # Look up the audio codec compatible with this container. # mkv/webm streams are copied as-is; mp4/mov/avi/flv re-encode # audio so FFmpeg doesn't refuse the merge. audio_codec = CONTAINER_AUDIO.get(video_fmt, "aac") ydl_opts = { **common, # Let yt-dlp freely pick the best video+audio streams without # restricting by extension — quality is maximised this way. "format": "bestvideo+bestaudio/best", "outtmpl": os.path.join(DOWNLOAD_DIR, f"{job_id}.%(ext)s"), "tmpdir": DOWNLOAD_DIR, "merge_output_format": video_fmt, # Pass codec flags directly to the FFmpeg merge step: # -c:v copy — video is never re-encoded (fast, lossless) # -c:a ... — audio copied or re-encoded as needed "postprocessor_args": { "merger": ["-c:v", "copy", "-c:a", audio_codec] }, } with yt_dlp.YoutubeDL(ydl_opts) as ydl: info = ydl.extract_info(url, download=True) title = sanitize_filename(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 # Serve the file with the original (sanitized) video title as the # download name, preserving the correct extension. ext = os.path.splitext(job["filename"])[1] download_name = f"{job['title']}{ext}" return send_file( filepath, as_attachment=True, download_name=download_name, ) 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)