255 lines
9.0 KiB
Python
255 lines
9.0 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)
|
|
|
|
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)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Cookies file — place a cookies.txt (Netscape format, exported from your
|
|
# browser while logged into YouTube) next to app.py to authenticate requests.
|
|
# Authenticated sessions are routed to faster CDN endpoints and bypass the
|
|
# throttling YouTube applies to anonymous traffic.
|
|
# 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.
|
|
#
|
|
# cookiefile - browser cookies for authenticated (fast) CDN routing.
|
|
# cachedir - persists solved JS sig/nsig to disk so node only re-runs
|
|
# when YouTube rotates their player (every few days).
|
|
# http_chunk_size - 10 MB chunks; plays nicer with YouTube CDN.
|
|
# js_runtimes - use local node v20+ for JS challenge solving.
|
|
#
|
|
# NOTE: sleep_interval intentionally removed — it was artificially capping
|
|
# download speed. Anti-throttle measures live in the frontend instead.
|
|
# ---------------------------------------------------------------------------
|
|
def _base_opts(extra=None):
|
|
opts = {
|
|
"cachedir": CACHE_DIR,
|
|
"js_runtimes": {"node": {"executable": "/usr/bin/node"}},
|
|
"http_headers": {"User-Agent": _random_ua()},
|
|
"http_chunk_size": 10 * 1024 * 1024,
|
|
"socket_timeout": 30,
|
|
"retries": 6,
|
|
"fragment_retries": 6,
|
|
}
|
|
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":
|
|
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/<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__":
|
|
# 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)
|