From 77204e263a9fe18b810ae8cb9d3092f5c7860a46 Mon Sep 17 00:00:00 2001 From: Ayush Rai Date: Fri, 28 Aug 2026 14:15:22 +0545 Subject: [PATCH 1/7] Update README.md --- README.md | 249 ++++++++++++++++++++++-------------------------------- 1 file changed, 101 insertions(+), 148 deletions(-) diff --git a/README.md b/README.md index 54a17f3..20c73ef 100644 --- a/README.md +++ b/README.md @@ -1,170 +1,123 @@ -# scraper +# scraper.py — video downloader with Cloudflare bypass -Video downloader with Cloudflare bypass. Paste a URL, pick a format, get the file. +A single-file CLI tool that pulls video files down from a page URL, falling +through a chain of strategies until one works — direct fetch, real-browser +Cloudflare bypass, HTML/iframe scraping, or a yt-dlp fallback. YouTube links +skip straight to yt-dlp. -YouTube routes directly to yt-dlp. Everything else goes through a 4-layer extraction stack: direct HTTP fetch, real Chrome with CF bypass, HTML/iframe scanning, then yt-dlp as a last resort. +## How it works ---- +Requests are tried in order, stopping at the first success: + +1. **Direct fetch** — `curl_cffi` (or `requests` if that's unavailable) with a + Chrome TLS fingerprint, no browser needed. +2. **Real Chrome via DrissionPage** — launches an actual Chrome instance to + clear Cloudflare's JS challenge and sniff the media URL off the network + tab. +3. **HTML / iframe scan** — regex + base64 decoding over the raw page (and + any player iframe it finds) to dig out a direct media URL. +4. **yt-dlp fallback** — handed off whenever nothing above finds a URL, the + URL is YouTube, or the CDN URL turns out to be token-bound (short-lived + signed URL) or in a domain-mismatch situation. + +Direct downloads stream with retries; `.m3u8` sources are pulled through +`ffmpeg`. Progress renders as a live-redrawing rainbow bar in the terminal, +including through yt-dlp's silent merge/remux step and through Chrome's fully +blocking page-load/Cloudflare-bypass calls — both used to just go quiet and +look frozen; now a marquee bar keeps animating through them. ## Requirements -**Python 3.10 or newer** - -External tools (must be on PATH): - -| Tool | Purpose | Required | -|------|---------|----------| -| ffmpeg | HLS/DASH muxing, format conversion | Strongly recommended | -| Chrome | CF bypass and token-bound CDN intercept | Required for protected sites | -| yt-dlp | YouTube and generic fallback | Recommended | - -Python packages: - -``` -pip install DrissionPage curl_cffi yt-dlp -``` - -Install ffmpeg: -- Windows: https://ffmpeg.org/download.html, add the bin folder to PATH -- Or via winget: `winget install ffmpeg` - ---- +- **Python 3.10+** (the code uses `X | None` union-type hints, which need + 3.10 or newer) +- **ffmpeg** on `PATH` — required for `.m3u8` downloads and yt-dlp's + audio/video mux step + - Windows: `winget install ffmpeg` + - macOS: `brew install ffmpeg` + - Linux: `apt install ffmpeg` +- **Google Chrome** installed — used by DrissionPage for the browser-bypass + layer -## Install +### Python packages ```bash -git clone https://github.com/yourusername/scraper.git -cd scraper -pip install DrissionPage curl_cffi yt-dlp +pip install -r requirements.txt ``` -No virtual environment required, but use one if you prefer. - ---- +| Package | Why it's needed | +|-------------|------------------| +| `curl_cffi` | Preferred HTTP backend — impersonates a real Chrome TLS fingerprint so Layer 1 isn't trivially blocked. Falls back to `requests` if not installed. | +| `DrissionPage` | Drives real Chrome for the Cloudflare-bypass / network-intercept layer (Layer 2). | +| `yt-dlp` | Handles YouTube and the generic fallback layer (Layer 4); also used internally for `.m3u8`/DASH merges. | +| `requests` | Fallback HTTP backend if `curl_cffi` fails to install (e.g. no prebuilt wheel for your platform). | ## Usage ```bash -python scraper.py https://example.com/video +python scraper.py ``` -Or run without arguments and paste the URL when prompted: +Or run it with no arguments to get an interactive prompt (with an idle-logo +animation) for the URL and output format: ```bash python scraper.py ``` -You will then be asked for an output format: - -``` -Output format: - 1. mp4 - 2. mp3 - 3. mkv - 4. webm - 5. original <- keeps original container/quality -Choice [1]: -``` - -Press Enter for mp4. Type a number or a custom extension (flac, opus, avi, etc). - -Output lands in `./videos/`. - ---- - -## How it works - -**YouTube / Shorts / Live** — detected by URL, handed straight to yt-dlp with best quality up to 1080p merged to the chosen format. No browser, no scraping. - -**Everything else** runs through four layers in order: - -1. Direct HTTP fetch via curl_cffi (Chrome TLS fingerprint) -2. Real Chrome via DrissionPage if step 1 hits a 403 or CF challenge -3. HTML scan for media URLs, iframe player fetch, base64 decode -4. yt-dlp generic fallback - -If a token-bound CDN URL is detected (pipe-signature pattern), the tool opens the player in Chrome, intercepts the live CDN request, then downloads with ffmpeg. - ---- - -## Output formats - -When you pick mp4, mkv, or webm: ffmpeg remuxes the stream into that container. - -When you pick mp3, aac, flac, opus, m4a: audio is extracted, video discarded. - -When you pick original: downloaded as-is, no remux. - -Custom extensions work too: type `avi`, `mov`, `ts`, whatever ffmpeg supports. - ---- - -## Config - -All tunable constants are at the top of the file: - -```python -OUTPUT_DIR = "videos" # output folder -MAX_RETRIES = 3 # retry count on direct download failures -MIN_MB = 2 # files smaller than this are rejected -YTDLP_TIMEOUT = 3600 # max seconds for yt-dlp (1 hour) -FFMPEG_TIMEOUT = 3600 # max seconds for ffmpeg -STREAM_TIMEOUT = 30 # per-chunk connect/read timeout -``` - ---- - -## Planned - -- GUI with queue, progress bar, output folder picker -- 4K / quality selector flag -- Batch mode: read URLs from a text file -- YouTube playlist support -- Resume support via Range header -- `--dry-run` flag -- Structured log file per session -- Twitter/X dedicated path (currently works via intercept) -- Instagram Reels -- Bilibili with cookie injection - ---- - -## Repo setup (first time) - -Create a new repo on GitHub with no README, no gitignore, no license (you will add these yourself). - -Then in your project folder: - -```bash -git init -git add scraper.py README.md .gitignore -git commit -m "init" -git branch -M main -git remote add origin https://github.com/yourusername/scraper.git -git push -u origin main -``` - -Suggested `.gitignore`: - -``` -videos/ -__pycache__/ -*.pyc -*.part -*.part.mp4 -.env -``` - -For future changes: - -```bash -git add scraper.py -git commit -m "what you changed" -git push -``` - ---- - -## License - -MIT +Downloaded files land in `./videos/`. + +## Terminal experience + +The whole CLI is built around one rule: nothing animated should ever freeze +mid-way and look dead. + +- **Breathing menu box** — the output-format picker has a continuously + animating rainbow border, with the "thinking" mascot looping above it the + whole time you're choosing. +- **Rainbow progress bars** — byte-accurate where possible, falling back to + a time-based or marquee bar when the source doesn't report a real size. +- **Mascots** — a single happy/sad face animates through to the very end of + the run (through the final "press any key to close" wait), instead of + playing a couple of loops and freezing partway. +- **Browser-intercept marquee** — Chrome's page-load and Cloudflare-bypass + steps are fully blocking with no progress hooks of their own; those now + run on a background thread while the main thread keeps a marquee bar + animating, so the CLI never goes silently unresponsive during a Cloudflare + clear. + +All animation is single-writer: any background thread only touches data +(subprocess pipes, the browser driver) and never the terminal directly, to +avoid the redraw races that come from two things trying to draw at once. + +## Configuration + +A few constants near the top of `scraper.py` control behavior: + +| Constant | Default | Purpose | +|------------------|---------|----------| +| `OUTPUT_DIR` | `videos`| Where downloaded files are saved | +| `MAX_RETRIES` | `3` | Retry attempts for a failed direct download | +| `MIN_MB` | `2` | Minimum acceptable file size (guards against corrupt/partial downloads) | +| `YTDLP_TIMEOUT` | `3600` | Max seconds to let a yt-dlp subprocess run | +| `FFMPEG_TIMEOUT` | `3600` | Max seconds to let an ffmpeg subprocess run | +| `STREAM_TIMEOUT` | `30` | Socket timeout for streamed direct downloads | + +## Known limitations / possible next steps + +- **Windows-only cookie handling** — the intercept path notes that DPAPI + cookie decryption is unreliable on Windows, so it skips cookies rather than + looping; worth revisiting if you need authenticated sessions. +- **Chrome-only bypass** — DrissionPage is hard-wired to Chrome; no + Firefox/WebKit fallback if Chrome isn't installed. +- **No proxy support** — neither the direct-fetch nor browser layers accept + a proxy URL; add one if you're scraping from a blocked network. +- **No concurrent downloads** — `scrape()` handles one URL per run; batching + a list of URLs would need a thin wrapper around it. +- **Single output filename scheme** — `safe_filename()` numbers files + sequentially per run; a resumable/skip-if-exists mode isn't implemented. + +## Code health + +Checked with `pyflakes` — no unused imports, no dead functions, no unused +variables. Every top-level function is reachable from `scrape()` or the +`if __name__ == "__main__"` entry point. From d7a14f7f651fba00359a7fa1ff7657b80ce97eb6 Mon Sep 17 00:00:00 2001 From: Ayush Rai Date: Fri, 28 Aug 2026 14:15:49 +0545 Subject: [PATCH 2/7] Update package versions in requirements.txt --- requirements.txt | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/requirements.txt b/requirements.txt index 2dafa27..50bd9ab 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ -DrissionPage -curl_cffi -yt-dlp +curl_cffi>=0.7.0 +DrissionPage>=4.1.0 +yt-dlp>=2024.8.6 +requests>=2.31.0 From d00b3959d2aec6820d2195b38549a84654f3eb1e Mon Sep 17 00:00:00 2001 From: Ayush Rai Date: Fri, 28 Aug 2026 15:17:33 +0545 Subject: [PATCH 3/7] Revise README for clarity and feature updates Updated project description and features in README.md. --- README.md | 187 +++++++++++++++++++++++++----------------------------- 1 file changed, 86 insertions(+), 101 deletions(-) diff --git a/README.md b/README.md index 20c73ef..80ebacc 100644 --- a/README.md +++ b/README.md @@ -1,123 +1,108 @@ -# scraper.py — video downloader with Cloudflare bypass +# scrape -A single-file CLI tool that pulls video files down from a page URL, falling -through a chain of strategies until one works — direct fetch, real-browser -Cloudflare bypass, HTML/iframe scraping, or a yt-dlp fallback. YouTube links -skip straight to yt-dlp. +Video downloader with Cloudflare bypass. Paste a link, pick a format, done. -## How it works +![Python](https://img.shields.io/badge/python-3.10%2B-blue) +![License](https://img.shields.io/badge/license-MIT-green) -Requests are tried in order, stopping at the first success: - -1. **Direct fetch** — `curl_cffi` (or `requests` if that's unavailable) with a - Chrome TLS fingerprint, no browser needed. -2. **Real Chrome via DrissionPage** — launches an actual Chrome instance to - clear Cloudflare's JS challenge and sniff the media URL off the network - tab. -3. **HTML / iframe scan** — regex + base64 decoding over the raw page (and - any player iframe it finds) to dig out a direct media URL. -4. **yt-dlp fallback** — handed off whenever nothing above finds a URL, the - URL is YouTube, or the CDN URL turns out to be token-bound (short-lived - signed URL) or in a domain-mismatch situation. - -Direct downloads stream with retries; `.m3u8` sources are pulled through -`ffmpeg`. Progress renders as a live-redrawing rainbow bar in the terminal, -including through yt-dlp's silent merge/remux step and through Chrome's fully -blocking page-load/Cloudflare-bypass calls — both used to just go quiet and -look frozen; now a marquee bar keeps animating through them. +--- -## Requirements +## Features + +- **YouTube & Twitter/X** — routed straight to yt-dlp, no scraping needed +- **Cloudflare-protected sites** — real Chrome via DrissionPage handles the JS challenge +- **Token-bound CDN URLs** — browser network interception captures the live stream URL +- **Auto-update** — checks yt-dlp on startup, updates silently if stale +- **Rainbow progress bar** — because why not +- **ffmpeg post-processing** — download in whatever format yt-dlp gives, convert to what you asked for -- **Python 3.10+** (the code uses `X | None` union-type hints, which need - 3.10 or newer) -- **ffmpeg** on `PATH` — required for `.m3u8` downloads and yt-dlp's - audio/video mux step - - Windows: `winget install ffmpeg` - - macOS: `brew install ffmpeg` - - Linux: `apt install ffmpeg` -- **Google Chrome** installed — used by DrissionPage for the browser-bypass - layer +--- + +## Requirements ### Python packages +``` +pip install -r requirements.txt +``` + +| Package | Purpose | +|---|---| +| `yt-dlp` | YouTube, Twitter/X, and generic video extraction | +| `curl_cffi` | Chrome TLS fingerprint for Cloudflare bypass | +| `DrissionPage` | Real Chrome automation for JS-heavy sites | + +### System dependencies + +| Tool | Install | +|---|---| +| **Python 3.10+** | [python.org](https://python.org) | +| **ffmpeg** | `winget install ffmpeg` (Windows) · `brew install ffmpeg` (Mac) · `apt install ffmpeg` (Linux) | +| **Chrome** | Must be installed — DrissionPage drives it | + +--- + +## Install + ```bash +git clone https://github.com/yourname/scrape +cd scrape pip install -r requirements.txt ``` -| Package | Why it's needed | -|-------------|------------------| -| `curl_cffi` | Preferred HTTP backend — impersonates a real Chrome TLS fingerprint so Layer 1 isn't trivially blocked. Falls back to `requests` if not installed. | -| `DrissionPage` | Drives real Chrome for the Cloudflare-bypass / network-intercept layer (Layer 2). | -| `yt-dlp` | Handles YouTube and the generic fallback layer (Layer 4); also used internally for `.m3u8`/DASH merges. | -| `requests` | Fallback HTTP backend if `curl_cffi` fails to install (e.g. no prebuilt wheel for your platform). | +--- ## Usage +**Double-click** `scraper.py` or run from terminal: + ```bash -python scraper.py +python scraper.py ``` -Or run it with no arguments to get an interactive prompt (with an idle-logo -animation) for the URL and output format: +Paste a URL when prompted, pick output format (mp4 / mp3 / mkv / webm / original), wait. + +You can also pass the URL as an argument: ```bash -python scraper.py +python scraper.py https://www.youtube.com/watch?v=dQw4w9WgXcQ +``` + +Output lands in a `videos/` folder next to the script. + +--- + +## How it works + +Sites go through layers in order, stopping at the first success: + ``` +URL + │ + ├─ YouTube / Twitter? ──► yt-dlp (native extractor) + │ + ├─ [1] curl_cffi direct fetch (Chrome TLS fingerprint) + ├─ [2] Real Chrome + Cloudflare bypass (DrissionPage) + ├─ [3] HTML / iframe scan + base64 decode + └─ [4] Browser network interception → CDN URL → ffmpeg/yt-dlp +``` + +--- + +## YouTube & 403 errors + +YouTube enforces Proof of Origin (PO) tokens on stream downloads. If you hit a 403: + +1. The script tries plain yt-dlp first (works for most videos) +2. Falls back to Edge → Chrome → Firefox cookies automatically +3. Make sure you're **logged into YouTube** in at least one browser + +Keeping yt-dlp up to date (handled automatically on startup) is usually enough. + +--- + +## Notes -Downloaded files land in `./videos/`. - -## Terminal experience - -The whole CLI is built around one rule: nothing animated should ever freeze -mid-way and look dead. - -- **Breathing menu box** — the output-format picker has a continuously - animating rainbow border, with the "thinking" mascot looping above it the - whole time you're choosing. -- **Rainbow progress bars** — byte-accurate where possible, falling back to - a time-based or marquee bar when the source doesn't report a real size. -- **Mascots** — a single happy/sad face animates through to the very end of - the run (through the final "press any key to close" wait), instead of - playing a couple of loops and freezing partway. -- **Browser-intercept marquee** — Chrome's page-load and Cloudflare-bypass - steps are fully blocking with no progress hooks of their own; those now - run on a background thread while the main thread keeps a marquee bar - animating, so the CLI never goes silently unresponsive during a Cloudflare - clear. - -All animation is single-writer: any background thread only touches data -(subprocess pipes, the browser driver) and never the terminal directly, to -avoid the redraw races that come from two things trying to draw at once. - -## Configuration - -A few constants near the top of `scraper.py` control behavior: - -| Constant | Default | Purpose | -|------------------|---------|----------| -| `OUTPUT_DIR` | `videos`| Where downloaded files are saved | -| `MAX_RETRIES` | `3` | Retry attempts for a failed direct download | -| `MIN_MB` | `2` | Minimum acceptable file size (guards against corrupt/partial downloads) | -| `YTDLP_TIMEOUT` | `3600` | Max seconds to let a yt-dlp subprocess run | -| `FFMPEG_TIMEOUT` | `3600` | Max seconds to let an ffmpeg subprocess run | -| `STREAM_TIMEOUT` | `30` | Socket timeout for streamed direct downloads | - -## Known limitations / possible next steps - -- **Windows-only cookie handling** — the intercept path notes that DPAPI - cookie decryption is unreliable on Windows, so it skips cookies rather than - looping; worth revisiting if you need authenticated sessions. -- **Chrome-only bypass** — DrissionPage is hard-wired to Chrome; no - Firefox/WebKit fallback if Chrome isn't installed. -- **No proxy support** — neither the direct-fetch nor browser layers accept - a proxy URL; add one if you're scraping from a blocked network. -- **No concurrent downloads** — `scrape()` handles one URL per run; batching - a list of URLs would need a thin wrapper around it. -- **Single output filename scheme** — `safe_filename()` numbers files - sequentially per run; a resumable/skip-if-exists mode isn't implemented. - -## Code health - -Checked with `pyflakes` — no unused imports, no dead functions, no unused -variables. Every top-level function is reachable from `scrape()` or the -`if __name__ == "__main__"` entry point. +- Downloads are saved to `videos/` — created automatically if it doesn't exist +- Existing files are skipped (no re-download) +- ffmpeg is optional but strongly recommended — without it format conversion is limited From 7fa453c0042f20b96f4bc6503a4b4c1dec54584f Mon Sep 17 00:00:00 2001 From: Ayush Rai Date: Fri, 28 Aug 2026 15:17:57 +0545 Subject: [PATCH 4/7] Update requirements to remove version constraints Removed version constraints from dependencies. --- requirements.txt | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/requirements.txt b/requirements.txt index 50bd9ab..6fc21df 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,3 @@ -curl_cffi>=0.7.0 -DrissionPage>=4.1.0 -yt-dlp>=2024.8.6 -requests>=2.31.0 +curl_cffi +DrissionPage +yt-dlp From ea58674c96042eed0953764631b3207e0cdb662d Mon Sep 17 00:00:00 2001 From: Ayush Rai Date: Fri, 28 Aug 2026 15:18:22 +0545 Subject: [PATCH 5/7] Update scraper.py --- scraper.py | 1590 +++++++++++++++++++++++++++++++++++----------------- 1 file changed, 1078 insertions(+), 512 deletions(-) diff --git a/scraper.py b/scraper.py index 1607e45..3dc2762 100644 --- a/scraper.py +++ b/scraper.py @@ -1,175 +1,611 @@ """ -scraper.py — video downloader with Cloudflare bypass - -Architecture: - Layer 1: DrissionPage (real Chrome, no webdriver signals) - → CloudflareBypasser logic inlined - → network listener catches mp4/m3u8/mpd as they load - → browser-intercept path for IP/token-bound CDNs - Layer 2: curl_cffi (Chrome TLS fingerprint) for non-CF sites - Layer 3: requests fallback - Layer 4: yt-dlp fallback if all extraction fails - -Install deps (run once): +scraper.py — video downloader with Cloudflare bypass + +Layers (in order): + 1. curl_cffi direct fetch (Chrome TLS fingerprint) + 2. Real Chrome via DrissionPage (CF bypass + network interception) + 3. HTML/iframe scan + base64 decode + 4. yt-dlp generic fallback + +YouTube is detected early and routed straight to yt-dlp, skipping all layers. +Token-bound CDN URLs trigger the browser-intercept path automatically. + +Install: pip install DrissionPage curl_cffi yt-dlp -Chrome must be installed — DrissionPage drives your real Chrome. + ffmpeg must be on PATH (winget install ffmpeg) + Chrome must be installed """ -import os, sys, re, time, shutil, base64, subprocess, threading +import os, sys, re, time, math, shutil, base64, subprocess, logging, colorsys from urllib.parse import urlparse, urljoin, unquote -# ── curl_cffi / requests backend ────────────────────────────────────────────── +# ── HTTP backend: curl_cffi (preferred) or plain requests ───────────────────── try: from curl_cffi import requests as cffi_requests - IMPERSONATE = "chrome124" - def _make_session(referer=""): - s = cffi_requests.Session(impersonate=IMPERSONATE) + _IMPERSONATE = "chrome124" + + def _make_session(referer: str = "") -> cffi_requests.Session: + s = cffi_requests.Session(impersonate=_IMPERSONATE) if referer: s.headers["Referer"] = referer return s - def _raw_get(url, headers, stream=False, timeout=30): + + def _raw_get(url: str, headers: dict, stream: bool = False, timeout: int = 30): return cffi_requests.get(url, headers=headers, stream=stream, - timeout=timeout, impersonate=IMPERSONATE, + timeout=timeout, impersonate=_IMPERSONATE, allow_redirects=True) USING_CFFI = True + except ImportError: import requests as _req - def _make_session(referer=""): + + def _make_session(referer: str = "") -> _req.Session: s = _req.Session() if referer: s.headers["Referer"] = referer return s - def _raw_get(url, headers, stream=False, timeout=30): + + def _raw_get(url: str, headers: dict, stream: bool = False, timeout: int = 30): return _req.get(url, headers=headers, stream=stream, timeout=timeout, allow_redirects=True) USING_CFFI = False -OUTPUT_DIR = "videos" -MAX_RETRIES = 3 -MIN_MB = 2 -YTDLP_TIMEOUT = 3600 # 1 hour — yt-dlp can be slow on large files -FFMPEG_TIMEOUT = 3600 # same -STREAM_TIMEOUT = 30 # connect+read timeout per chunk window for _raw_get +# ── Config ──────────────────────────────────────────────────────────────────── +OUTPUT_DIR = "videos" +MAX_RETRIES = 3 +MIN_MB = 2 +YTDLP_TIMEOUT = 3600 +FFMPEG_TIMEOUT = 3600 +STREAM_TIMEOUT = 30 UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36") +AUDIO_FMTS = frozenset(("mp3", "aac", "flac", "opus", "m4a", "wav")) + +# ── ASCII banner (solid block glyphs, animated rainbow sweep) ───────────────── LOGO = r""" - ______ ______ ______ ________ ______ ______ -/_____/\ /_____/\ /_____/\ /_______/\ /_____/\ /_____/\ -\::::_\/_\:::__\/ \:::_ \ \ \::: _ \ \\:::_ \ \\::::_\/_ - \:\/___/\\:\ \ __\:(_) ) )_\::(_) \ \\:(_) \ \\:\/___/\ - \_::._\:\\:\ \/_/\\: __ `\ \\:: __ \ \\: ___\/ \::___\/_ - /____\:\\:\_\ \ \\ \ `\ \ \\:.\ \ \ \\ \ \ \:\____/\ - \_____\/ \_____\/ \_\/ \_\/ \__\/\__\/ \_\/ \_____\/ +███████╗ ██████╗██████╗ █████╗ ██████╗ ███████╗ +██╔════╝██╔════╝██╔══██╗██╔══██╗██╔══██╗██╔════╝ +███████╗██║ ██████╔╝███████║██████╔╝█████╗ +╚════██║██║ ██╔══██╗██╔══██║██╔═══╝ ██╔══╝ +███████║╚██████╗██║ ██║██║ ██║██║ ███████╗ +╚══════╝ ╚═════╝╚═╝ ╚═╝╚═╝ ╚═╝╚═╝ ╚══════╝ """ +# 256-color ramp the sweep animation cycles through (warm -> cool -> warm) +_WAVE_COLORS = [196, 202, 208, 214, 220, 226, 190, 154, 118, 82, + 46, 47, 48, 49, 50, 51, 45, 39, 33, 27, + 21, 57, 93, 129, 165, 201, 199, 198, 197] + +def _ansi_ready() -> bool: + """True if the terminal can render ANSI escapes; enables them on Windows.""" + if os.name == "nt": + os.system("") # no-op that flips on VT100 processing in modern conhost + return sys.stdout.isatty() + +def print_logo(frames: int = 16, delay: float = 0.035) -> None: + """Print the banner. Animates a color sweep across it in a real terminal, + falls back to a plain static print anywhere ANSI isn't supported (piped + output, dumb terminals, etc.).""" + if not _ansi_ready(): + print(LOGO) + return + lines = LOGO.strip("\n").splitlines() + n = len(_WAVE_COLORS) + out = sys.stdout + try: + out.write("\033[?25l") # hide cursor + for frame in range(frames): + buf = [] + for row, line in enumerate(lines): + chars = [] + for col, ch in enumerate(line): + if ch == " ": + chars.append(" ") + else: + idx = (col // 2 + row - frame) % n + chars.append(f"\033[38;5;{_WAVE_COLORS[idx]}m{ch}") + buf.append("".join(chars) + "\033[0m") + out.write("\033[H" + "\n".join(buf) + "\n") + out.flush() + time.sleep(delay) + finally: + out.write("\033[0m\033[?25h") # reset color, restore cursor + out.flush() + +# ── ASCII mascot (in-place looping frame animation) ──────────────────────────── +def _play_frames(frames: list, loops: int = 1, delay: float = 0.12) -> None: + """Redraw a sequence of multi-line ASCII frames in place, looping `loops` + times, then leave the final frame on screen. Falls back to a single + static print of the last frame when ANSI cursor moves aren't supported.""" + frame_lines = [f.splitlines() for f in frames] + if not _ansi_ready(): + print("\n".join(frame_lines[-1])) + return + out = sys.stdout + try: + out.write("\033[?25l") + first = True + for _ in range(loops): + for lines in frame_lines: + if not first: + out.write(f"\033[{len(lines)}A") # cursor up to overwrite + first = False + for line in lines: + out.write("\033[2K" + line + "\n") # clear line, redraw + out.flush() + time.sleep(delay) + finally: + out.write("\033[0m\033[?25h") + out.flush() + +# tears wiggle side to side and drip down, ends in a little splash +_CRY_FRAMES = [ + " (╥﹏╥) \n , \n ", + "(╥﹏╥) \n ' \n ", + " (╥﹏╥)\n , \n . ", + " (╥﹏╥) \n ` \n . ", + " (╥﹏╥) \n \n ~*~*~ ", +] + +# cartoonish squeeze-and-stretch jump for hops +_HAPPY_FRAMES = [ + " ___ \n (^▽^) \n ▔▔▔▔▔ ", + " \\(^▽^)/\n | | \n ", + " \\(★▽★)/\n ✧ \n ", + " ___ \n (^▽^) \n ▔▔▔▔▔ ", +] + +# idle chin-scrub "pondering" loop +_THINKING_FRAMES = [ + " (≖‿≖ )⌐\n ", + " (≖‿≖ )~\n ", + " ( ≖o≖ )\n ⌐ ", +] + +def print_mascot_fail() -> None: + _play_frames(_CRY_FRAMES, loops=2, delay=0.15) + +def print_mascot_success() -> None: + _play_frames(_HAPPY_FRAMES, loops=2, delay=0.11) + +def print_mascot_thinking() -> None: + _play_frames(_THINKING_FRAMES, loops=2, delay=0.25) + +# ── HSV -> truecolor helper (shared by border / progress bar / press-key) ──── +def _rgb(h: float, s: float, v: float) -> str: + r, g, b = colorsys.hsv_to_rgb(h % 1.0, s, max(0.0, min(1.0, v))) + return f"\033[38;2;{int(r*255)};{int(g*255)};{int(b*255)}m" + +# ── Single-owner key polling (no threads, no race — see mascot_demo.py notes) ─ +class _RawStdin: + """Puts the terminal in cbreak mode (unix) so keys are readable one at a + time without waiting for Enter, and without the tty auto-echoing them. + No-op on Windows; msvcrt already reads raw per-key.""" + def __enter__(self): + self.enabled = False + if os.name != "nt" and sys.stdin.isatty(): + import termios, tty + self.fd = sys.stdin.fileno() + self.old = termios.tcgetattr(self.fd) + tty.setcbreak(self.fd) + self.enabled = True + return self + + def __exit__(self, *exc): + if self.enabled: + import termios + termios.tcsetattr(self.fd, termios.TCSADRAIN, self.old) + +def _poll_key(): + """Non-blocking: return one character if a key is waiting, else None.""" + if os.name == "nt": + import msvcrt + if msvcrt.kbhit(): + return msvcrt.getwch() + return None + else: + import select + if not sys.stdin.isatty(): + return None + r, _, _ = select.select([sys.stdin], [], [], 0) + if r: + return sys.stdin.read(1) + return None + +_BACKSPACE = {"\x08", "\x7f"} +_ENTER = {"\r", "\n"} + +def _live_prompt(render_fn, n_lines: int, on_submit, frame_delay: float = 0.05) -> str: + """The core engine: one loop, one thread, one writer. Every tick it + either processes a waiting keystroke or redraws the next animation + frame — never both racing each other (this is the fix for the + duplicate-border bug from the old background-thread version). + `render_fn(buf, error)` returns the full box text. `on_submit(buf)` + returns None to accept, or an error string to reject and re-prompt.""" + out = sys.stdout + if not _ansi_ready(): + while True: + buf = input(render_fn("", "") + "\n> ") + err = on_submit(buf) + if err is None: + return buf + print(f"[!] {err}") + + buf, error = "", "" + out.write("\033[?25l") + printed_once = False + try: + while True: + frame = render_fn(buf, error) + lines = frame.splitlines() + if printed_once: + out.write(f"\033[{n_lines}A") + for line in lines: + out.write("\033[2K" + line + "\n") + out.flush() + printed_once = True + + key = _poll_key() + if key is None: + time.sleep(frame_delay) + continue + if key in _ENTER: + err = on_submit(buf) + if err is None: + return buf + error = err + buf = "" + elif key in _BACKSPACE: + buf = buf[:-1] + error = "" + elif key.isprintable(): + buf += key + error = "" + finally: + out.write("\033[0m\033[?25h") + out.flush() + +def input_with_breathing_menu(title: str, options: list, valid: set, default: str = "1") -> str: + """Bordered menu box with a continuously breathing rainbow border while + it waits on input. Invalid choices re-prompt in place with an inline + error instead of silently accepting garbage or crashing.""" + body_width = max(len(title), max(len(o) for o in options), 24) + 2 + t0 = time.time() + + def side(text: str) -> str: + return f"║ {text.ljust(body_width - 1)}║" + + def border_row(is_top: bool, elapsed: float) -> str: + hue_shift = (elapsed * 0.12) % 1.0 + breath = 0.55 + 0.45 * math.sin(elapsed * 2.2) + corners = ("╔", "╗") if is_top else ("╚", "╝") + chars = [corners[0]] + for i in range(body_width): + hue = (i / body_width + hue_shift) % 1.0 + chars.append(_rgb(hue, 0.85, breath) + "═") + chars.append("\033[0m" + corners[1]) + return "".join(chars) + + def render(buf: str, error: str) -> str: + elapsed = time.time() - t0 + rows = [title, ""] + options + ["", f"Choice [{default}]: {buf}", "", + (f"\033[91m{error}\033[0m" if error else "")] + body = [side(r) for r in rows] + return "\n".join([border_row(True, elapsed)] + body + [border_row(False, elapsed)]) + + n_lines = len(render("", "").splitlines()) + + def on_submit(buf: str): + choice = buf.strip() or default + if choice not in valid: + return f"use 1–{max(valid, key=int)}, got '{choice}'" + return None + + return _live_prompt(render, n_lines, on_submit).strip() or default + +# ── Rainbow byte-accurate download progress bar ────────────────────────────── +def render_progress_bar(done: int, total: int, width: int = 40, elapsed: float = 0.0) -> str: + pct = 0.0 if total <= 0 else min(1.0, done / total) + filled = int(width * pct) + hue_shift = (elapsed * 0.15) % 1.0 + bar_chars = [] + for i in range(width): + if i < filled: + hue = (i / width + hue_shift) % 1.0 + bar_chars.append(_rgb(hue, 0.85, 0.95) + "█") + else: + bar_chars.append("\033[38;5;238m░") + bar = "".join(bar_chars) + "\033[0m" + + def _fmt_mb(n: int) -> str: + return f"{n / (1024 * 1024):.1f}MB" + + return f"[{bar}] {pct*100:5.1f}% {_fmt_mb(done)}/{_fmt_mb(total)}" + +def render_time_progress_bar(done_sec: float, total_sec: float, width: int = 40, elapsed: float = 0.0) -> str: + """Same visual style as render_progress_bar, but driven by playback + time processed rather than bytes — for sources (m3u8/HLS) where the + real byte total isn't knowable up front but ffmpeg reports duration.""" + pct = 0.0 if total_sec <= 0 else min(1.0, done_sec / total_sec) + filled = int(width * pct) + hue_shift = (elapsed * 0.15) % 1.0 + bar_chars = [] + for i in range(width): + if i < filled: + hue = (i / width + hue_shift) % 1.0 + bar_chars.append(_rgb(hue, 0.85, 0.95) + "█") + else: + bar_chars.append("\033[38;5;238m░") + bar = "".join(bar_chars) + "\033[0m" + + def _fmt_t(s: float) -> str: + s = max(0, int(s)) + return f"{s // 60:02d}:{s % 60:02d}" + + return f"[{bar}] {pct*100:5.1f}% {_fmt_t(done_sec)}/{_fmt_t(total_sec)}" + +def _probe_duration(url: str, referer: str) -> float: + """Best-effort ffprobe duration lookup, in seconds. Returns 0.0 if it + can't be determined (some m3u8 sources refuse to report it too, at + which point we just fall back to indeterminate/no-bar).""" + if not shutil.which("ffprobe"): + return 0.0 + try: + cmd = ["ffprobe", "-v", "error", "-headers", _ffmpeg_hdr_block(referer), + "-show_entries", "format=duration", "-of", "csv=p=0", url] + r = subprocess.run(cmd, capture_output=True, timeout=15, text=True) + return float(r.stdout.strip()) + except Exception: + return 0.0 + +# ── Animated "press any key to close" ───────────────────────────────────────── +def press_any_key_to_close(message: str = "Press any key to close...", frame_delay: float = 0.05) -> None: + if not _ansi_ready(): + input(message + " ") + return + out = sys.stdout + t0 = time.time() + out.write("\033[?25l") + try: + with _RawStdin(): + printed_once = False + while True: + elapsed = time.time() - t0 + breath = 0.5 + 0.5 * math.sin(elapsed * 3.0) + hue = (elapsed * 0.2) % 1.0 + colored = _rgb(hue, 0.8, 0.5 + 0.5 * breath) + message + "\033[0m" + if printed_once: + out.write("\033[2K\r") + out.write(colored) + out.flush() + printed_once = True + key = _poll_key() + if key is not None: + out.write("\n") + return + time.sleep(frame_delay) + finally: + out.write("\033[0m\033[?25h") + out.flush() + +# ── Idle logo replay (single-loop — waits on a key, replays sweep every ~5s) ── +def wait_for_site_input_with_idle_logo(replay_every: float = 5.0) -> str: + """Sits at 'Site URL: ' prompt. If the user hasn't typed anything for + `replay_every` seconds, the logo does one rainbow sweep in place above + the prompt line, then returns to waiting — all in the same single + loop/single writer that reads keystrokes, so no race with typing.""" + if not _ansi_ready(): + return input("Site URL: ").strip() + + lines = LOGO.strip("\n").splitlines() + n_logo = len(lines) + n_wave = len(_WAVE_COLORS) + out = sys.stdout + buf = "" + t0 = time.time() + last_key_t = t0 + sweep_frame = 0 + printed_once = False + out.write("\033[?25l") + try: + while True: + now = time.time() + idle = now - last_key_t + sweeping = idle >= replay_every + + logo_rows = [] + for row, line in enumerate(lines): + chars = [] + for col, ch in enumerate(line): + if ch == " ": + chars.append(" ") + else: + idx = (col // 2 + row - (sweep_frame if sweeping else 0)) % n_wave + chars.append(f"\033[38;5;{_WAVE_COLORS[idx]}m{ch}") + logo_rows.append("".join(chars) + "\033[0m") + prompt_row = f"Site URL: {buf}" + frame = "\n".join(logo_rows + [prompt_row]) + n_lines = n_logo + 1 + + if printed_once: + out.write(f"\033[{n_lines}A") + for line in frame.splitlines(): + out.write("\033[2K" + line + "\n") + out.flush() + printed_once = True + + if sweeping: + sweep_frame += 1 + if sweep_frame >= 16: # one full sweep, then rest until idle timer resets + sweep_frame = 0 + last_key_t = now # restart the 5s idle countdown after a replay + + key = _poll_key() + if key is None: + time.sleep(0.035 if sweeping else 0.05) + continue + last_key_t = time.time() + sweep_frame = 0 + if key in _ENTER: + out.write("\n") + return buf.strip() + elif key in _BACKSPACE: + buf = buf[:-1] + elif key.isprintable(): + buf += key + finally: + out.write("\033[0m\033[?25h") + out.flush() + +# ── Compiled regexes (module-level — compiled once) ─────────────────────────── MEDIA_RE = re.compile( r'https?://[^\s"\'<>{}\[\]]+\.(?:mp4|m3u8|mpd)(?:[?#][^\s"\'<>]*)?', re.I ) - -# Analytics/tracking domains to skip (but still mine for mu= params) SKIP_DOMAINS_RE = re.compile( r'jwpltx\.com|google-analytics|doubleclick|googlesyndication' r'|facebook\.com|twitter\.com|scorecardresearch|omtrdc\.net', re.I ) +TOKEN_BOUND_RE = re.compile(r'\|\d{9,10}\|[0-9a-f]{16,}', re.I) +IFRAME_RE = re.compile(r']+src=["\']([^"\']+)["\']', re.I) +IFRAME_SKIP_RE = re.compile( + r'google\.com/recaptcha|accounts\.google|facebook\.com/plugins' + r'|twitter\.com/i/|disqus\.com|google|facebook|disqus', + re.I +) +_DIRECT_RE = re.compile( + r'(?:(?:file|src|source|href|data-src)["\s]*[:=]["\s]*|["\'])' + r'["\']?(https?://[^\s"\'<>{}\[\]]+\.(?:mp4|m3u8|mpd)(?:[?#][^\s"\'<>]*)?)', + re.I, +) +YT_RE = re.compile( + r'(?:https?://)?(?:www\.|m\.)?' + r'(?:youtube\.com/(?:watch|shorts|live|embed)|youtu\.be/)', + re.I +) -# Pipe-delimited token signatures — always IP-bound, direct download = 403 -TOKEN_BOUND_RE = re.compile(r'\|\d{9,10}\|[0-9a-f]{16,}', re.I) - -# ── Cached tool checks (avoid shutil.which on every call) ───────────────────── -_FFMPEG_AVAILABLE: bool | None = None -_YTDLP_AVAILABLE: bool | None = None +# ── Cached tool checks ──────────────────────────────────────────────────────── +_FFMPEG: bool | None = None +_YTDLP: bool | None = None def ffmpeg_ok() -> bool: - global _FFMPEG_AVAILABLE - if _FFMPEG_AVAILABLE is None: - _FFMPEG_AVAILABLE = shutil.which("ffmpeg") is not None - return _FFMPEG_AVAILABLE + global _FFMPEG + if _FFMPEG is None: + _FFMPEG = shutil.which("ffmpeg") is not None + return _FFMPEG def ytdlp_ok() -> bool: - global _YTDLP_AVAILABLE - if _YTDLP_AVAILABLE is None: - _YTDLP_AVAILABLE = shutil.which("yt-dlp") is not None - return _YTDLP_AVAILABLE - - -def is_token_bound(url: str) -> bool: - return bool(TOKEN_BOUND_RE.search(url)) - -def _extract_media_url(raw_url: str): - """ - Given a captured network URL, return the real media URL or None. - Handles: - 1. URL is mp4/m3u8/mpd directly - 2. JWPlayer analytics ping with mu= param containing the real URL - """ - # JWPlayer ping and other analytics — check mu= param first - if SKIP_DOMAINS_RE.search(raw_url): - mu = re.search(r'[?&]mu=([^&]+)', raw_url) - if mu: - media = unquote(mu.group(1)) - if MEDIA_RE.search(media): - print(f"[listen] Extracted mu= media URL: {media}") - return media - return None - if MEDIA_RE.search(raw_url): - return raw_url - return None + global _YTDLP + if _YTDLP is None: + _YTDLP = shutil.which("yt-dlp") is not None + return _YTDLP + +# ── Dependency update check ─────────────────────────────────────────────────── +def quick_update_check() -> None: + """Fast update check: ask yt-dlp if it needs updating (single process, + no PyPI round-trips). Skips silently if yt-dlp not installed or offline.""" + if not ytdlp_ok(): + return + try: + # --update-to stable checks GitHub releases — fast, single request + r = subprocess.run( + ["yt-dlp", "--update-to", "stable"], + capture_output=True, text=True, timeout=10 + ) + out = (r.stdout + r.stderr).strip() + # yt-dlp prints "yt-dlp is up to date" or "Updated to " + if "up to date" not in out.lower() and "updated" in out.lower(): + print(f"[update] {out.splitlines()[0]}") + except Exception: + pass # offline or timeout — silent skip + +# ── Shared header builders ──────────────────────────────────────────────────── +_BASE_HEADERS_STATIC = { + "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8", + "Accept-Language": "en-US,en;q=0.9", + "Accept-Encoding": "gzip, deflate, br", + "Connection": "keep-alive", + "Upgrade-Insecure-Requests": "1", + "Sec-Fetch-Dest": "document", + "Sec-Fetch-Mode": "navigate", + "Sec-Fetch-Site": "none", + "Sec-Fetch-User": "?1", + "Sec-CH-UA": '"Chromium";v="124","Google Chrome";v="124","Not-A.Brand";v="99"', + "Sec-CH-UA-Mobile": "?0", + "Sec-CH-UA-Platform": '"Windows"', + "Cache-Control": "max-age=0", +} + +def _base_headers(referer: str = "") -> dict: + return {"User-Agent": UA, "Referer": referer, **_BASE_HEADERS_STATIC} + +def _cdn_headers(referer: str) -> dict: + parsed = urlparse(referer) + origin = f"{parsed.scheme}://{parsed.netloc}" + return { + "User-Agent": UA, + "Referer": referer, + "Origin": origin, + "Sec-Fetch-Dest": "video", + "Sec-Fetch-Mode": "no-cors", + "Sec-Fetch-Site": "cross-site", + } +def _ffmpeg_hdr_block(referer: str) -> str: + h = _cdn_headers(referer) + return ( + f"Referer: {h['Referer']}\r\n" + f"Origin: {h['Origin']}\r\n" + f"User-Agent: {UA}\r\n" + f"Sec-Fetch-Dest: video\r\n" + f"Sec-Fetch-Mode: no-cors\r\n" + f"Sec-Fetch-Site: cross-site\r\n" + ) -# ── inlined CloudflareBypasser ──────────────────────────────────────────────── +# ── yt-dlp format args builder (shared across all call sites) ───────────────── +def _yt_fmt_args(out_fmt: str) -> tuple[str, list]: + """Return (format_selector, extra_args) for yt-dlp.""" + if out_fmt in AUDIO_FMTS: + return ("bestaudio/best", + ["--extract-audio", "--audio-format", out_fmt, "--audio-quality", "0"]) + if not out_fmt: + return ("bestvideo+bestaudio/best[height<=1080]/best", []) + sel = "bestvideo+bestaudio/best" if ffmpeg_ok() else "best" + return (sel, ["--merge-output-format", out_fmt]) -def _cf_bypass(driver, max_attempts=10): - import logging - log = logging.getLogger("CFBypass") - def _is_cf_page(d): - title = d.title or "" - return ("just a moment" in title.lower() or - "checking your browser" in title.lower() or - "cloudflare" in title.lower()) +# ── Cloudflare bypass ───────────────────────────────────────────────────────── +_cf_log = logging.getLogger("CFBypass") - def _click_verify(d): - try: - for iframe in d.get_frames(): - try: - cb = iframe.ele("tag:input@type=checkbox", timeout=1) - if cb: - cb.click() - log.info("Clicked Turnstile checkbox inside iframe") - return True - except Exception: - pass - except Exception: - pass - try: - cb = d.ele("tag:input@type=checkbox", timeout=1) - if cb: - cb.click() - log.info("Clicked checkbox (direct)") - return True - except Exception: - pass +def _cf_bypass(driver, max_attempts: int = 10) -> bool: + def _is_cf(d): + t = (d.title or "").lower() + return "just a moment" in t or "checking your browser" in t or "cloudflare" in t + + def _click(d): + for src in ([d] + list(d.get_frames() or [])): + try: + cb = src.ele("tag:input@type=checkbox", timeout=1) + if cb: + cb.click() + return True + except Exception: + pass return False - log.info("Starting CF bypass") for attempt in range(1, max_attempts + 1): - if not _is_cf_page(driver): - log.info("CF page gone — bypass succeeded") + if not _is_cf(driver): return True - log.info(f"Attempt {attempt}: CF page detected, trying to click...") - _click_verify(driver) + _cf_log.info(f"CF attempt {attempt}") + _click(driver) time.sleep(2) - - log.warning("CF bypass: max attempts reached, proceeding anyway") return False -# ── shared Chrome options factory ───────────────────────────────────────────── - +# ── Chrome options ──────────────────────────────────────────────────────────── def _chrome_opts(): from DrissionPage import ChromiumOptions opts = ChromiumOptions() @@ -180,34 +616,26 @@ def _chrome_opts(): return opts -# ── DrissionPage network listener (4.1.x API) ──────────────────────────────── -# -# Confirmed API from version 4.1.1.4: -# listen.start(targets=None, is_regex=None, method=None, res_type=None) -# listen.wait(count=1, timeout=None, fit_count=True, raise_err=None) -# listen.steps(count=None, timeout=None, gap=1) — generator -# -# Strategy: start() with no filter (catch everything), then wait() with a -# short timeout in a loop, check each packet's URL ourselves. - -def _start_listener(driver, captured: dict, lock: threading.Lock): - """Start listener with no filter — we'll check URLs ourselves.""" +# ── Network listener helpers ────────────────────────────────────────────────── +def _start_listener(driver): try: - listener = driver.listen - listener.start() # no targets filter — catch all requests - print("[listen] Network listener started") - return listener + driver.listen.start() + return driver.listen except Exception as e: - print(f"[listen] Listener unavailable: {e}") + _cprint(f"[listen] Unavailable: {e}", 196) return None +def _extract_media_url(raw_url: str) -> str | None: + if SKIP_DOMAINS_RE.search(raw_url): + mu = re.search(r'[?&]mu=([^&]+)', raw_url) + if mu: + media = unquote(mu.group(1)) + if MEDIA_RE.search(media): + return media + return None + return raw_url if MEDIA_RE.search(raw_url) else None -def _poll_listener(listener, captured: dict, lock: threading.Lock, timeout=15): - """ - Call listener.wait() in a loop until we get a media URL or timeout. - Checks each packet URL through _extract_media_url() which handles - both direct media URLs and JWPlayer analytics pings with mu= params. - """ +def _poll_listener(listener, captured: dict, timeout: int = 15) -> None: if listener is None: return deadline = time.time() + timeout @@ -220,210 +648,200 @@ def _poll_listener(listener, captured: dict, lock: threading.Lock, timeout=15): fit_count=True, raise_err=False) if packet is None: continue - packets = packet if isinstance(packet, (list, tuple)) else [packet] - for p in packets: - raw_url = getattr(p, "url", "") or "" - media_url = _extract_media_url(raw_url) - if media_url: - with lock: - if not captured["url"]: - captured["url"] = media_url - print(f"[listen] Captured: {media_url}") + for p in (packet if isinstance(packet, (list, tuple)) else [packet]): + url = _extract_media_url(getattr(p, "url", "") or "") + if url: + captured["url"] = url + _cprint(f"[listen] Captured: {url}", 45) return except Exception: time.sleep(0.3) -# ── DrissionPage browser fetch ──────────────────────────────────────────────── - -def _drission_fetch(site: str): +# ── Browser fetch (layer 2) ─────────────────────────────────────────────────── +def _drission_fetch(site: str) -> tuple: try: from DrissionPage import ChromiumPage except ImportError: - print("[browser] DrissionPage not installed — pip install DrissionPage") + _cprint("[browser] DrissionPage not installed — pip install DrissionPage", 196) return None, None - print("[browser] Launching Chrome (DrissionPage)...") + print("[browser] Launching Chrome...") captured = {"url": None} - lock = threading.Lock() driver = ChromiumPage(addr_or_opts=_chrome_opts()) - listener = _start_listener(driver, captured, lock) + listener = _start_listener(driver) try: - print(f"[browser] Navigating to {site}") driver.get(site) time.sleep(3) _cf_bypass(driver) time.sleep(2) - - # Poll for intercepted media — longer window for slow embeds - _poll_listener(listener, captured, lock, timeout=15) + _poll_listener(listener, captured, timeout=15) html = driver.html - media_url = captured["url"] - - if not media_url: + if not captured["url"]: m = MEDIA_RE.search(html) if m: - media_url = m.group(0) - print(f"[browser] Found media in page HTML: {media_url}") - - if not media_url: - iframes = re.findall(r']+src=["\']([^"\']+)["\']', html, re.I) - for src in iframes: - if src.startswith("http") and not re.search( - r'google|facebook|disqus', src, re.I): - print(f"[browser] Checking iframe: {src}") - driver.get(src) - time.sleep(3) - _poll_listener(listener, captured, lock, timeout=15) - if captured["url"]: - media_url = captured["url"] - break - frame_html = driver.html - m = MEDIA_RE.search(frame_html) - if m: - media_url = m.group(0) - print(f"[browser] Found media in iframe: {media_url}") - break - - return html, media_url + captured["url"] = m.group(0) + print(f"[browser] Found in HTML: {captured['url']}") + + if not captured["url"]: + for m in IFRAME_RE.finditer(html): + src = m.group(1).strip() + if not src.startswith("http") or IFRAME_SKIP_RE.search(src): + continue + print(f"[browser] Checking iframe: {src}") + driver.get(src) + time.sleep(3) + _poll_listener(listener, captured, timeout=15) + if captured["url"]: + break + fm = MEDIA_RE.search(driver.html) + if fm: + captured["url"] = fm.group(0) + print(f"[browser] Found in iframe HTML: {captured['url']}") + break + + return html, captured["url"] except Exception as e: print(f"[browser] Error: {e}") return None, None finally: - try: - driver.quit() - except Exception: - pass + try: driver.quit() + except Exception: pass -# ── Browser-intercept CDN download (IP/token-bound) ─────────────────────────── - +# ── Browser-intercept CDN download (token-bound) ────────────────────────────── def _browser_intercept_and_download(player_url: str, site_referer: str, out_fmt: str = "mp4") -> bool: - """ - Open player_url in real Chrome → CF Worker issues token bound to our IP. - Poll network listener for the CDN request in-flight. - Download with yt-dlp --cookies-from-browser chrome, fall back to ffmpeg. - """ try: from DrissionPage import ChromiumPage except ImportError: print("[intercept] DrissionPage not installed.") return False - print(f"[intercept] Opening player in Chrome: {player_url}") - intercepted: dict = {"url": None} - lock = threading.Lock() + print(f"[intercept] Opening in Chrome: {player_url}") + captured = {"url": None} driver = ChromiumPage(addr_or_opts=_chrome_opts()) - listener = _start_listener(driver, intercepted, lock) + listener = _start_listener(driver) try: driver.get(player_url) time.sleep(3) _cf_bypass(driver) - # Nudge play button while polling for CDN request - deadline = time.time() + 20 - while not intercepted["url"] and time.time() < deadline: - # Poll one tick - _poll_listener(listener, intercepted, lock, timeout=1) - if intercepted["url"]: + # X.com / Twitter needs extra time — CDN URL only fires after play + is_twitter = any(h in player_url for h in ("x.com", "twitter.com", "t.co")) + deadline = time.time() + (40 if is_twitter else 20) + + # X.com player selectors (aria-label is the most reliable, others as fallbacks) + _XCOM_SELECTORS = [ + "css:[data-testid='videoPlayer'] video", + "css:[data-testid='videoComponent']", + "css:div[aria-label='Embedded video']", + "css:div[role='progressbar']", # timeline bar — click triggers play + "css:video", # bare