247 lines
8.5 KiB
Python
247 lines
8.5 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)
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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 = {
|
||
"quiet": True,
|
||
"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/<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)
|