diff --git a/batch.py b/batch.py index 7e93c5c..ca699b2 100644 --- a/batch.py +++ b/batch.py @@ -3,18 +3,25 @@ import json from pathlib import Path import os -COOKIES = "cookies.txt" # Get them using the browser extension "Get cookies.txt" and export them while logged in to your account. -YTDLP = "yt-dlp.exe" # Change this if you aren't on Windows. Get the latest release from github +# Get cookies using the browser extension "Get cookies.txt" and export while +# logged in to your account. +COOKIES = "cookies.txt" + +# Change this if you aren't on Windows. Get the latest release from GitHub. +YTDLP = "yt-dlp.exe" OUTPUT_DIR = Path(__file__).parent / "YOUR_OUTPUT_FOLDER" # Input options: -# - If `urls.json` exists in the project root it will be used. The JSON should be an array of strings (URLs) -# - Otherwise set URLs below or pass a custom JSON path via the `URLS_JSON` environment variable. +# - If `urls.json` exists in the project root it will be used. +# The JSON should be an array of strings (URLs). +# - Otherwise set URLs below or pass a custom JSON path via the +# `URLS_JSON` environment variable. URLS = [ # fallback list — leave empty if you'll use urls.json ] -URLS_JSON = os.environ.get("URLS_JSON", str(Path(__file__).parent / "urls.json")) +default_urls = str(Path(__file__).parent / "urls.json") +URLS_JSON = os.environ.get("URLS_JSON", default_urls) RESET = "\033[0m" BOLD = "\033[1m" @@ -44,8 +51,13 @@ def fetch_info(url): "generic:impersonate", url, ] + cwd_dir = Path(__file__).parent result = subprocess.run( - cmd, capture_output=True, text=True, timeout=30, cwd=Path(__file__).parent + cmd, + capture_output=True, + text=True, + timeout=30, + cwd=cwd_dir, ) if result.returncode != 0: return None, result.stderr.strip() @@ -92,30 +104,40 @@ def main(): data = json.load(fh) if isinstance(data, list) and data: URLS.clear() - URLS.extend([u for u in data if isinstance(u, str) and u.strip()]) + for u in data: + if isinstance(u, str) and u.strip(): + URLS.append(u) total = len(URLS) except Exception as e: log(f"Failed to load URLs from {URLS_JSON}: {e}", YELLOW) ok, skipped, failed = [], [], [] - log(f"\n{'━'*55}", CYAN) - log(f" yt-dlp batch — {total} posts → {OUTPUT_DIR.name}/", BOLD) - log(f"{'━'*55}\n", CYAN) + log("\n" + "━" * 55, CYAN) + log(" yt-dlp batch — {} posts → {}/".format(total, OUTPUT_DIR.name), BOLD) + log("" + "━" * 55 + "\n", CYAN) for i, url in enumerate(URLS, 1): post_id = get_post_id(url) - log(f"[{i:02d}/{total}] {DIM}{url}{RESET}") + header = "[{:02d}/{:d}]".format(i, total) + log(header + " " + DIM + url + RESET) # Fetch metadata first to get the title - log(f" fetching info…", DIM) + log(" fetching info…", DIM) info, err = fetch_info(url) if info is None: # Likely a text post or unavailable - log( - f" {YELLOW}⚠ skipped — no media ({err[:80] if err else 'no video found'}){RESET}" + reason = err[:80] if err else "no video found" + msg = ( + " " + + YELLOW + + "⚠ skipped — no media (" + + reason + + ")" + + RESET ) + log(msg) skipped.append((url, err)) print() continue @@ -125,24 +147,27 @@ def main(): out_dir = OUTPUT_DIR / folder_name out_dir.mkdir(parents=True, exist_ok=True) - log(f" title: {BOLD}{title[:55]}{RESET}") - log(f" dir: {out_dir}") - log(f" downloading…", DIM) + log(" title: " + BOLD + title[:55] + RESET) + log(" dir: " + str(out_dir)) + log(" downloading…", DIM) success, output = download(url, out_dir) if success: # Find what was downloaded files = list(out_dir.iterdir()) - sizes = [f"{f.stat().st_size / 1e6:.1f} MB" for f in files if f.is_file()] - log( - f" {GREEN}✓ done — {', '.join(sizes) if sizes else 'file saved'}{RESET}" - ) + sizes = [] + for f in files: + if f.is_file(): + sizes.append("{:.1f} MB".format(f.stat().st_size / 1e6)) + msg = ", ".join(sizes) if sizes else "file saved" + log(" " + GREEN + "✓ done — " + msg + RESET) ok.append(url) else: - # Check if it's just no video (text post that slipped through info check) + # Check if it's just no video + # (text post that slipped through info check) if "no video" in output.lower() or "no formats" in output.lower(): - log(f" {YELLOW}⚠ skipped — text post{RESET}") + log(" " + YELLOW + "⚠ skipped — text post" + RESET) skipped.append((url, "no video content")) # Remove empty dir try: @@ -150,31 +175,43 @@ def main(): except OSError: pass else: - log(f" {RED}✗ failed{RESET}") + log(" " + RED + "✗ failed" + RESET) # Print last few lines of output for context for line in output.splitlines()[-3:]: - log(f" {DIM}{line}{RESET}") - failed.append( - (url, output.splitlines()[-1] if output else "unknown error") - ) + log(" " + DIM + line + RESET) + if output: + last_err = output.splitlines()[-1] + else: + last_err = "unknown error" + failed.append((url, last_err)) print() # Summary - log(f"{'━'*55}", CYAN) - log(f" Summary", BOLD) - log(f"{'━'*55}", CYAN) - log(f" {GREEN}✓ downloaded: {len(ok)}{RESET}") - log(f" {YELLOW}⚠ skipped (text/no media): {len(skipped)}{RESET}") - log(f" {RED}✗ failed: {len(failed)}{RESET}") + log("" + "━" * 55, CYAN) + log(" Summary", BOLD) + log("" + "━" * 55, CYAN) + log(" " + GREEN + "✓ downloaded: " + str(len(ok)) + RESET) + skipped_msg = ( + " " + YELLOW + "⚠ skipped (text/no media): " + + str(len(skipped)) + + RESET + ) + log(skipped_msg) + failed_msg = ( + " " + RED + "✗ failed: " + + str(len(failed)) + + RESET + ) + log(failed_msg) if failed: - log(f"\n Failed URLs:", RED) + log("\n Failed URLs:", RED) for url, reason in failed: - log(f" • {url}", RED) - log(f" {DIM}{reason[:80]}{RESET}") + log(" • " + url, RED) + log(" " + DIM + reason[:80] + RESET) - log(f"{'━'*55}\n", CYAN) + log("" + "━" * 55 + "\n", CYAN) if __name__ == "__main__": diff --git a/with_ui.py b/with_ui.py index 53dc279..c230563 100644 --- a/with_ui.py +++ b/with_ui.py @@ -71,15 +71,17 @@ def get_info(): cmd, capture_output=True, text=True, timeout=30, cwd=DOWNLOAD_DIR ) if result.returncode != 0: - return jsonify({"error": result.stderr or "Failed to fetch info"}), 400 + msg = result.stderr or "Failed to fetch info" + return jsonify({"error": msg}), 400 info = json.loads(result.stdout) formats = [] for f in info.get("formats", []): + res = f.get("resolution") or f.get("format_note") or "" formats.append( { "id": f.get("format_id"), "ext": f.get("ext"), - "resolution": f.get("resolution") or f.get("format_note") or "", + "resolution": res, "vcodec": f.get("vcodec", "none"), "acodec": f.get("acodec", "none"), "filesize": f.get("filesize") or f.get("filesize_approx"), @@ -125,7 +127,11 @@ def start_download(): cmd.append(url) job_id = str(uuid.uuid4()) - thread = threading.Thread(target=run_ytdlp, args=(job_id, cmd), daemon=True) + thread = threading.Thread( + target=run_ytdlp, + args=(job_id, cmd), + daemon=True, + ) thread.start() return jsonify({"job_id": job_id}) @@ -139,14 +145,17 @@ def job_status(job_id): with jobs_lock: job = jobs.get(job_id) if not job: - yield f"data: {json.dumps({'error': 'Job not found'})}\n\n" + payload = json.dumps({"error": "Job not found"}) + yield "data: " + payload + "\n\n" break lines = job["lines"] while sent < len(lines): - yield f"data: {json.dumps({'line': lines[sent]})}\n\n" + payload = json.dumps({"line": lines[sent]}) + yield "data: " + payload + "\n\n" sent += 1 if job["done"]: - yield f"data: {json.dumps({'done': True, 'error': job['error']})}\n\n" + payload = json.dumps({"done": True, "error": job["error"]}) + yield "data: " + payload + "\n\n" break time.sleep(0.2) @@ -158,6 +167,6 @@ def job_status(job_id): if __name__ == "__main__": - print(f"[yt-dlp UI] Serving on http://localhost:5000") - print(f"[yt-dlp UI] Download directory: {DOWNLOAD_DIR}") + print("[yt-dlp UI] Serving on http://localhost:5000") + print("[yt-dlp UI] Download directory: {}".format(DOWNLOAD_DIR)) app.run(debug=False, host="0.0.0.0", port=5000, threaded=True)