304 lines
11 KiB
Python
304 lines
11 KiB
Python
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/<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
|
|
|
|
# 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)
|