krosh/rule34vault-downloader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

init commit

Browse Source
This commit is contained in:
krosh
2026-07-07 05:35:14 +03:00
parent 63352e1128
commit 77ff644f89
5 changed files with 1031 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
IDEA.md
+329
View File
@@ -0,0 +1,329 @@
# rule34vault downloader ideas
## User requirements / comments to keep
- Download bookmarks for a public username; no authorization required.
- Bookmark pages are available as `https://rule34vault.com/u/<username>?page=<n>`.
- Only username should be required for normal use.
- Use browser-like headers to reduce chance of request blocking/blocklisting.
- Do **not** rely on `HEAD` only. Some servers may reject or mis-handle `HEAD`; use `GET` with streaming/range checks when needed.
- For normal image posts, first try predictable original-file URLs:
- `https://rule34vault.com/posts/<prefix>/<post_id>/<post_id>.jpg`
- `https://r34xyz.b-cdn.net/posts/<prefix>/<post_id>/<post_id>.jpg`
- `<prefix>` is the first digits group from the post id path. Example: post `1263322` uses prefix `1263`.
- Some videos require opening the post page and extracting real `<video>/<source src=...>` URLs from the rendered HTML.
- Create a plain text list of downloaded files in the same order they appear in bookmarks.
- Plain txt state/log files are enough. Do **not** use SQLite.
- Use multiple tries and exponential sleep after failures.
- Project must use a new venv.
## Real site structure inspected
Inspected with browser-like headers from the project venv.
### Video post example
URL:
```text
https://rule34vault.com/post/738794
```
The page contains a real video element in returned HTML:
```html
<video controls loop class="video" style="aspect-ratio: 1280/720;">
<source type="video/mp4" src="https://r34xyz.b-cdn.net/posts/738/738794/738794.480.mp4">
</video>
```
So video extraction should parse the post HTML with BeautifulSoup and collect:
1. `video[src]`
2. `video source[src]`
3. fallback: any `source[src]` where `type` starts with `video/`
For post `996379`, real page had multiple video sources:
```text
https://r34xyz.b-cdn.net/posts/996/996379/996379.720.hevc.mp4
https://r34xyz.b-cdn.net/posts/996/996379/996379.480.mp4
```
Prefer broadly-compatible sources first, e.g. non-HEVC MP4 before HEVC if both exist.
### Image post examples
For:
```text
https://rule34vault.com/post/1263322
```
The page HTML contains a small display image:
```html
<img class="img" src="/posts/1263/1263322/1263322.small.jpg">
```
But the original image is available directly:
```text
https://rule34vault.com/posts/1263/1263322/1263322.jpg
https://r34xyz.b-cdn.net/posts/1263/1263322/1263322.jpg
```
Both returned `200 image/jpeg` during inspection.
Another inspected example:
```text
https://rule34vault.com/post/1263303
```
Sometimes the post page `img[src]` points only to a small/preview file:
```text
https://rule34vault.com/posts/1263/1263303/1263303.small.jpg
```
In this case the downloader must normalize the source URL to the original filename by removing the size suffix before the extension:
```text
https://rule34vault.com/posts/1263/1263303/1263303.jpg
```
This URL returned `200 image/jpeg` during inspection. Do not download `.small.jpg` when the corresponding original image exists.
### Bookmark page structure
On:
```text
https://rule34vault.com/u/krosh?page=6
```
Post links were **not** regular `<a href="/post/...">` links in the DOM. The bookmark data was inside:
```html
<script id="ng-state" type="application/json">...</script>
```
Relevant key looked like:
```json
"post:/api/v2/post/search/bookmarked/50311": {
"items": [
{"id": 1075647, "type": 0, ...},
{"id": 1081537, "type": 0, ...},
...
]
}
```
Therefore bookmark extraction should prefer parsing `script#ng-state` JSON and reading `items` from the key starting with:
```text
post:/api/v2/post/search/bookmarked/
```
Keep item order exactly as found in that `items` array.
The API also works directly as POST:
```text
POST https://rule34vault.com/api/v2/post/search/bookmarked/<user_id>
```
Examples observed:
- Body `{"skip": 100, "take": 20}` returned 20 bookmark items.
- Body `{"take": 20, "page": 6}` returned page-like bookmark items.
Simplest robust path:
1. Fetch profile page `/u/<username>?page=<page>`.
2. Parse `ng-state`.
3. Get user id from key `get:/api/v2/account/user/<username>`.
4. Get bookmark items from `post:/api/v2/post/search/bookmarked/<user_id>`.
5. Optionally use the direct POST API for pagination once user id is known.
## Headers
Use one shared `requests.Session()` with realistic headers:
```python
HEADERS = {
"User-Agent": "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/126.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Connection": "keep-alive",
"Referer": "https://rule34vault.com/",
}
```
For direct API POST, use:
```python
API_HEADERS = {
**HEADERS,
"Accept": "application/json,text/plain,*/*",
"Origin": "https://rule34vault.com",
"Content-Type": "application/json",
}
```
For media download, set referer to the post page:
```python
headers = {**HEADERS, "Referer": f"https://rule34vault.com/post/{post_id}"}
```
## Retry policy
Every network operation should use retry wrapper:
```python
for attempt in range(max_tries):
try:
...
except requests.RequestException:
sleep = base_delay * (2 ** attempt)
time.sleep(sleep)
```
Also retry on temporary HTTP statuses:
```text
429, 500, 502, 503, 504
```
Do not retry permanent `404` for candidate media URLs; just try the next candidate.
## Download order log
Create a txt manifest in download order:
```text
download_order.txt
```
Append one line per successfully downloaded post, exactly in bookmark order:
```text
000001 1075647 downloads/1075/1075647.jpg https://rule34vault.com/posts/1075/1075647/1075647.jpg
000002 1081537 downloads/1081/1081537.jpg https://r34xyz.b-cdn.net/posts/1081/1081537/1081537.jpg
000003 996379 downloads/996/996379.mp4 https://r34xyz.b-cdn.net/posts/996/996379/996379.480.mp4
```
Also keep simple txt logs:
```text
seen.txt # post ids already processed successfully
failed.txt # post id + reason
skipped.txt # already existing files
```
No SQLite.
## Media resolution strategy
For each bookmark item in order:
1. If already downloaded, skip and log to `skipped.txt`.
2. If `type == 0` image:
- Try original candidate URLs by extension and host.
- Use streaming `GET`, not only `HEAD`.
- Validate `status_code == 200` and `Content-Type` starts with `image/`.
3. If candidates fail, fetch post page and parse actual HTML.
- For image posts, inspect `img[src]` values from the real post page.
- If the only discovered image is a display derivative such as `<post_id>.small.jpg`, `<post_id>.preview.jpg`, `<post_id>.thumbnail.jpg`, etc., derive and try the original URL without the size suffix before downloading:
- `1263303.small.jpg` -> `1263303.jpg`
- `1263303.preview.jpg` -> `1263303.jpg`
- Download the higher-resolution original when it returns a valid image response; use the derivative only as a last fallback if the original does not exist.
4. For videos (`type == 1`) or if image candidates fail:
- Fetch `https://rule34vault.com/post/<post_id>`.
- Parse `<video>` and `<source>` tags.
- Choose best compatible source.
5. Download media to grouped directory:
```text
downloads/<prefix>/<post_id>.<ext>
```
## Candidate URL generation
For post id `1263322`:
```python
prefix = post_id[:4]
```
Candidate originals:
```python
hosts = [
"https://rule34vault.com",
"https://r34xyz.b-cdn.net",
]
extensions = ["jpg", "png", "webp", "gif"]
for host in hosts:
for ext in extensions:
yield f"{host}/posts/{prefix}/{post_id}/{post_id}.{ext}"
```
For video direct guesses, possible candidates can be tried after HTML parsing or as fallback:
```text
<post_id>.480.mp4
<post_id>.720.mp4
<post_id>.720.hevc.mp4
<post_id>.webm
```
But actual post HTML source extraction is preferred for videos.
## CLI idea
```bash
python -m venv .venv
. .venv/bin/activate
pip install requests beautifulsoup4 tqdm
python r34vault_downloader.py krosh --start-page 1 --end-page 10 --out downloads
```
Arguments:
```text
username required
--start-page default 1
--end-page optional; stop when no items if omitted
--out default downloads
--tries default 4
--delay default 0.5 seconds between posts
--base-retry-delay default 1.5 seconds
--dry-run print planned downloads without saving media
```
## Core functions
```python
make_session()
request_with_retries(method, url, **kwargs)
fetch_user_page(username, page)
parse_ng_state(html)
extract_user_id(ng_state, username)
extract_bookmark_items_from_state(ng_state, user_id)
fetch_bookmark_items_api(user_id, page=None, skip=None, take=20)
candidate_image_urls(post_id)
stream_download(url, dest, referer)
extract_media_sources_from_post_html(post_id)
choose_video_source(sources)
already_downloaded(post_id, out_dir)
append_download_order(index, post_id, path, source_url)
append_failed(post_id, reason)
main()
```
README.md
+71 -1
View File
@@ -1,2 +1,72 @@
# rule34vault-downloader # rule34vault-downloader
just python script to download your bookmarks from rule34.vault
Python script to download public bookmarks from `rule34vault.com` for a username. No authorization is required for public bookmark pages.
## Setup
```bash
python -m venv .venv
. .venv/bin/activate
pip install -r requirements.txt
```
## Usage
Download bookmarks for a user:
```bash
python r34vault_downloader.py <user> --out downloads
```
Download only a page range:
```bash
python r34vault_downloader.py <user> --start-page 1 --end-page 3 --out downloads
```
Process only the first N bookmark items, useful for testing:
```bash
python r34vault_downloader.py <user> --start-page 6 --end-page 6 --limit 1 --out downloads
```
## Output
Files are grouped by Rule34Vault post prefix:
```text
downloads/
1263/
1263322.jpg
996/
996379.mp4
download_order.txt
seen.txt
failed.txt
skipped.txt
```
`download_order.txt` records successful downloads in bookmark order:
```text
000001 1263322 downloads/1263/1263322.jpg https://rule34vault.com/posts/1263/1263322/1263322.jpg
```
## Behavior
- Uses browser-like headers and post-page `Referer` headers.
- Uses streaming `GET`; it does not rely on `HEAD` requests.
- Parses bookmark data from `script#ng-state` and preserves bookmark order.
- Tries original image URLs first, for example `1263322.jpg`.
- If a page image is only `1263303.small.jpg`, it tries `1263303.jpg` before falling back to the small file.
- For videos, parses real `<video>` and `<source src="...">` elements from the post page.
- Retries transient failures with exponential backoff.
- Uses plain text logs only; no SQLite database.
## Tests
```bash
. .venv/bin/activate
pip install pytest
pytest -q
```
r34vault_downloader.py
+543
View File
@@ -0,0 +1,543 @@
#!/usr/bin/env python3
"""Download public Rule34Vault bookmarks for a username.
Usage:
python r34vault_downloader.py krosh --start-page 1 --end-page 3 --out downloads
The downloader uses public pages/API only; no credentials are required.
"""
from __future__ import annotations
import argparse
import json
import mimetypes
import re
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable, Iterator
from urllib.parse import urljoin, urlparse
import requests
from bs4 import BeautifulSoup
from requests import Response, Session
from tqdm import tqdm
SITE = "https://rule34vault.com"
CDN = "https://r34xyz.b-cdn.net"
RETRY_STATUSES = {429, 500, 502, 503, 504}
IMAGE_EXTS = ("jpg", "png", "webp", "gif", "jpeg")
VIDEO_EXTS = ("mp4", "webm", "mov", "m4v")
DERIVATIVE_SUFFIXES = ("small", "preview", "thumbnail", "thumb", "medium", "large")
HEADERS = {
"User-Agent": (
"Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/126.0 Safari/537.36"
),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Connection": "keep-alive",
"Referer": f"{SITE}/",
}
API_HEADERS = {
**HEADERS,
"Accept": "application/json,text/plain,*/*",
"Origin": SITE,
"Content-Type": "application/json",
}
@dataclass(frozen=True)
class MediaSources:
images: list[str]
image_fallbacks: list[str]
videos: list[str]
def make_session() -> Session:
session = requests.Session()
session.headers.update(HEADERS)
return session
def request_with_retries(
session: Session,
method: str,
url: str,
*,
tries: int = 4,
base_retry_delay: float = 1.5,
retry_statuses: set[int] = RETRY_STATUSES,
**kwargs: Any,
) -> Response:
"""Run an HTTP request with exponential backoff for transient failures."""
last_exc: Exception | None = None
for attempt in range(tries):
try:
response = session.request(method, url, **kwargs)
if response.status_code not in retry_statuses:
return response
response.close()
last_exc = RuntimeError(f"HTTP {response.status_code} for {url}")
except requests.RequestException as exc:
last_exc = exc
if attempt < tries - 1:
time.sleep(base_retry_delay * (2**attempt))
if last_exc:
raise last_exc
raise RuntimeError(f"request failed without exception: {method} {url}")
def fetch_user_page(session: Session, username: str, page: int, **retry_kwargs: Any) -> str:
url = f"{SITE}/u/{username}?page={page}"
response = request_with_retries(session, "GET", url, headers={**HEADERS, "Referer": f"{SITE}/"}, timeout=30, **retry_kwargs)
response.raise_for_status()
return response.text
def parse_ng_state(html: str) -> dict[str, Any]:
soup = BeautifulSoup(html, "html.parser")
script = soup.find("script", id="ng-state", type="application/json") or soup.find("script", id="ng-state")
if not script:
raise ValueError("script#ng-state was not found")
text = script.string or script.get_text() or ""
if not text.strip():
raise ValueError("script#ng-state is empty")
return json.loads(text)
def extract_user_id(state: dict[str, Any], username: str) -> int:
key = f"get:/api/v2/account/user/{username}"
value = state.get(key)
if isinstance(value, dict):
for field in ("id", "userId"):
if isinstance(value.get(field), int):
return value[field]
data = value.get("data")
if isinstance(data, dict):
for field in ("userId", "id"):
if isinstance(data.get(field), int):
return data[field]
# Fallback for case variations or encoded usernames.
suffix = f"/api/v2/account/user/{username}".lower()
for candidate_key, candidate_value in state.items():
if candidate_key.lower().endswith(suffix) and isinstance(candidate_value, dict):
data = candidate_value.get("data") if isinstance(candidate_value.get("data"), dict) else candidate_value
for field in ("userId", "id"):
if isinstance(data.get(field), int):
return data[field]
raise ValueError(f"could not find public user id for {username!r} in ng-state")
def extract_bookmark_items_from_state(state: dict[str, Any], user_id: int) -> list[dict[str, Any]]:
exact = f"post:/api/v2/post/search/bookmarked/{user_id}"
value = state.get(exact)
if isinstance(value, dict) and isinstance(value.get("items"), list):
return list(value["items"])
prefix = "post:/api/v2/post/search/bookmarked/"
for key, candidate in state.items():
if key.startswith(prefix) and isinstance(candidate, dict) and isinstance(candidate.get("items"), list):
return list(candidate["items"])
return []
def fetch_bookmark_items_api(
session: Session,
user_id: int,
*,
page: int | None = None,
skip: int | None = None,
take: int = 20,
**retry_kwargs: Any,
) -> list[dict[str, Any]]:
body: dict[str, int] = {"take": take}
if page is not None:
body["page"] = page
if skip is not None:
body["skip"] = skip
url = f"{SITE}/api/v2/post/search/bookmarked/{user_id}"
response = request_with_retries(session, "POST", url, headers=API_HEADERS, json=body, timeout=30, **retry_kwargs)
response.raise_for_status()
payload = response.json()
items = payload.get("items", [])
if not isinstance(items, list):
raise ValueError(f"unexpected API response shape from {url}")
return items
def post_prefix(post_id: str) -> str:
"""Return Rule34Vault media path prefix: 1263322 -> 1263, 738794 -> 738."""
return post_id[:-3] if len(post_id) > 3 else post_id
def candidate_image_urls(post_id: str) -> Iterator[str]:
prefix = post_prefix(post_id)
for ext in IMAGE_EXTS:
yield f"{SITE}/posts/{prefix}/{post_id}/{post_id}.{ext}"
yield f"{CDN}/posts/{prefix}/{post_id}/{post_id}.{ext}"
def candidate_video_urls(post_id: str) -> Iterator[str]:
prefix = post_prefix(post_id)
# Observed videos use 480/720 variants; the path prefix is post_id without the last 3 digits.
names = [
f"{post_id}.480.mp4",
f"{post_id}.720.mp4",
f"{post_id}.720.hevc.mp4",
f"{post_id}.mp4",
f"{post_id}.webm",
]
for name in names:
yield f"{CDN}/posts/{prefix}/{post_id}/{name}"
yield f"{SITE}/posts/{prefix}/{post_id}/{name}"
def original_from_derivative_url(url: str, post_id: str) -> str:
"""Remove display-size suffix from Rule34Vault media filename if present."""
parsed = urlparse(url)
path = parsed.path
suffix_re = "|".join(map(re.escape, DERIVATIVE_SUFFIXES))
new_path = re.sub(rf"/{re.escape(post_id)}\.({suffix_re})(\.[A-Za-z0-9]+)$", rf"/{post_id}\2", path)
if new_path == path:
return url
if parsed.scheme and parsed.netloc:
return parsed._replace(path=new_path).geturl()
return new_path
def _dedupe(items: Iterable[str]) -> list[str]:
seen: set[str] = set()
result: list[str] = []
for item in items:
if item and item not in seen:
seen.add(item)
result.append(item)
return result
def extract_media_sources_from_html(html: str, post_id: str, page_url: str) -> MediaSources:
soup = BeautifulSoup(html, "html.parser")
images: list[str] = []
image_fallbacks: list[str] = []
videos: list[str] = []
for video in soup.find_all("video"):
src = video.get("src")
if src:
videos.append(urljoin(page_url, src))
for source in video.find_all("source"):
src = source.get("src")
if src:
videos.append(urljoin(page_url, src))
for source in soup.find_all("source"):
src = source.get("src")
media_type = source.get("type", "")
if src and (media_type.startswith("video/") or Path(urlparse(src).path).suffix.lower().lstrip(".") in VIDEO_EXTS):
videos.append(urljoin(page_url, src))
for img in soup.find_all("img"):
src = img.get("src")
if not src:
continue
abs_url = urljoin(page_url, src)
path = urlparse(abs_url).path
if f"/{post_id}/" not in path or not path.lower().endswith(tuple(f".{ext}" for ext in IMAGE_EXTS)):
continue
original = urljoin(page_url, original_from_derivative_url(abs_url, post_id))
if original != abs_url:
images.append(original)
image_fallbacks.append(abs_url)
else:
images.append(abs_url)
for meta_name in ("og:image", "twitter:image"):
meta = soup.find("meta", attrs={"property": meta_name}) or soup.find("meta", attrs={"name": meta_name})
content = meta.get("content") if meta else None
if content:
abs_url = urljoin(page_url, content)
original = urljoin(page_url, original_from_derivative_url(abs_url, post_id))
if original != abs_url:
images.append(original)
image_fallbacks.append(abs_url)
else:
images.append(abs_url)
return MediaSources(images=_dedupe(images), image_fallbacks=_dedupe(image_fallbacks), videos=_dedupe(videos))
def fetch_post_html(session: Session, post_id: str, **retry_kwargs: Any) -> str:
url = f"{SITE}/post/{post_id}"
response = request_with_retries(session, "GET", url, headers={**HEADERS, "Referer": SITE + "/"}, timeout=30, **retry_kwargs)
response.raise_for_status()
return response.text
def choose_video_source(sources: list[str]) -> str | None:
if not sources:
return None
def score(url: str) -> tuple[int, int, int]:
lower = url.lower()
hevc_penalty = 1 if "hevc" in lower or "hvc1" in lower else 0
ext_rank = 0 if lower.endswith(".mp4") else 1
# Prefer compatible/non-HEVC; for compatibility, 480 mp4 before heavier variants.
resolution_rank = 0 if ".480." in lower else 1 if ".720." in lower else 2
return hevc_penalty, ext_rank, resolution_rank
return sorted(sources, key=score)[0]
def response_extension(url: str, response: Response) -> str:
content_type = response.headers.get("content-type", "").split(";", 1)[0].strip().lower()
guessed = mimetypes.guess_extension(content_type) if content_type else None
if guessed == ".jpe":
guessed = ".jpg"
if guessed:
return guessed
suffix = Path(urlparse(url).path).suffix
return suffix or ".bin"
def is_media_response(response: Response, media: str) -> bool:
if response.status_code != 200:
return False
content_type = response.headers.get("content-type", "").lower()
if media == "image":
return content_type.startswith("image/")
if media == "video":
return content_type.startswith("video/") or "mp4" in content_type or "webm" in content_type
return content_type.startswith(("image/", "video/"))
def probe_media_url(
session: Session,
url: str,
*,
media: str,
post_id: str,
tries: int,
base_retry_delay: float,
) -> Response | None:
headers = {**HEADERS, "Accept": "*/*", "Referer": f"{SITE}/post/{post_id}"}
try:
response = request_with_retries(
session,
"GET",
url,
headers=headers,
stream=True,
timeout=45,
tries=tries,
base_retry_delay=base_retry_delay,
)
except Exception:
return None
if is_media_response(response, media):
return response
response.close()
return None
def destination_for(out_dir: Path, post_id: str, url: str, response: Response | None = None) -> Path:
ext = response_extension(url, response) if response is not None else Path(urlparse(url).path).suffix or ".bin"
return out_dir / post_prefix(post_id) / f"{post_id}{ext}"
def stream_download_response(response: Response, dest: Path, chunk_size: int = 1024 * 256) -> int:
dest.parent.mkdir(parents=True, exist_ok=True)
tmp = dest.with_suffix(dest.suffix + ".part")
total = 0
with tmp.open("wb") as fh:
for chunk in response.iter_content(chunk_size=chunk_size):
if not chunk:
continue
fh.write(chunk)
total += len(chunk)
tmp.replace(dest)
return total
def already_downloaded(post_id: str, out_dir: Path) -> Path | None:
prefix_dir = out_dir / post_prefix(post_id)
if not prefix_dir.exists():
return None
matches = sorted(path for path in prefix_dir.glob(f"{post_id}.*") if path.suffix != ".part")
return matches[0] if matches else None
def append_line(path: Path, line: str) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
with path.open("a", encoding="utf-8") as fh:
fh.write(line.rstrip("\n") + "\n")
def append_download_order(manifest: Path, index: int, post_id: str, path: Path, source_url: str) -> None:
append_line(manifest, f"{index:06d} {post_id} {path.as_posix()} {source_url}")
def read_seen(path: Path) -> set[str]:
if not path.exists():
return set()
return {line.strip().split()[0] for line in path.read_text(encoding="utf-8").splitlines() if line.strip()}
def resolve_media(
session: Session,
item: dict[str, Any],
*,
tries: int,
base_retry_delay: float,
) -> tuple[str, Response] | None:
post_id = str(item["id"])
post_type = int(item.get("type", 0))
if post_type == 0:
for url in candidate_image_urls(post_id):
response = probe_media_url(session, url, media="image", post_id=post_id, tries=tries, base_retry_delay=base_retry_delay)
if response:
return url, response
html = fetch_post_html(session, post_id, tries=tries, base_retry_delay=base_retry_delay)
sources = extract_media_sources_from_html(html, post_id, f"{SITE}/post/{post_id}")
if post_type == 1:
ordered_video_sources = []
chosen = choose_video_source(sources.videos)
if chosen:
ordered_video_sources.append(chosen)
ordered_video_sources.extend(url for url in sources.videos if url != chosen)
ordered_video_sources.extend(candidate_video_urls(post_id))
for url in _dedupe(ordered_video_sources):
response = probe_media_url(session, url, media="video", post_id=post_id, tries=tries, base_retry_delay=base_retry_delay)
if response:
return url, response
else:
for url in _dedupe([*sources.images, *sources.image_fallbacks]):
response = probe_media_url(session, url, media="image", post_id=post_id, tries=tries, base_retry_delay=base_retry_delay)
if response:
return url, response
return None
def bookmark_items_for_pages(
session: Session,
username: str,
*,
start_page: int,
end_page: int | None,
take: int,
tries: int,
base_retry_delay: float,
) -> Iterator[dict[str, Any]]:
"""Yield bookmark items from public user pages, preserving page/DOM order.
The direct POST API exists, but observed `page` semantics did not exactly match
`/u/<username>?page=N`, so the downloader treats the public page as the
source of truth and parses its `script#ng-state` payload for each page.
"""
del take # kept as CLI/API option for future direct-API use; page HTML is authoritative now.
user_id: int | None = None
page = start_page
while True:
if end_page is not None and page > end_page:
break
html = fetch_user_page(session, username, page, tries=tries, base_retry_delay=base_retry_delay)
state = parse_ng_state(html)
if user_id is None:
user_id = extract_user_id(state, username)
items = extract_bookmark_items_from_state(state, user_id)
if not items:
break
for item in items:
yield item
page += 1
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description="Download public Rule34Vault bookmarks for a username")
parser.add_argument("username", help="Rule34Vault username, e.g. krosh")
parser.add_argument("--start-page", type=int, default=1, help="first bookmark page to fetch")
parser.add_argument("--end-page", type=int, help="last bookmark page to fetch; omitted means until empty")
parser.add_argument("--out", type=Path, default=Path("downloads"), help="output directory")
parser.add_argument("--tries", type=int, default=4, help="request attempts for transient failures")
parser.add_argument("--delay", type=float, default=0.5, help="sleep between posts")
parser.add_argument("--base-retry-delay", type=float, default=1.5, help="initial exponential retry sleep")
parser.add_argument("--dry-run", action="store_true", help="list bookmark ids that would be processed; do not download files")
parser.add_argument("--limit", type=int, help="maximum number of bookmark items to process after pagination")
return parser.parse_args(argv)
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
out_dir: Path = args.out
out_dir.mkdir(parents=True, exist_ok=True)
manifest = out_dir / "download_order.txt"
seen_path = out_dir / "seen.txt"
failed_path = out_dir / "failed.txt"
skipped_path = out_dir / "skipped.txt"
seen = read_seen(seen_path)
session = make_session()
success_index = sum(1 for line in manifest.read_text(encoding="utf-8").splitlines()) if manifest.exists() else 0
items = bookmark_items_for_pages(
session,
args.username,
start_page=args.start_page,
end_page=args.end_page,
take=20,
tries=args.tries,
base_retry_delay=args.base_retry_delay,
)
processed_count = 0
for item in tqdm(items, desc="bookmarks", unit="post"):
if args.limit is not None and processed_count >= args.limit:
break
processed_count += 1
post_id = str(item.get("id"))
if not post_id or post_id == "None":
append_line(failed_path, f"unknown invalid bookmark item {item!r}")
continue
existing = already_downloaded(post_id, out_dir)
if post_id in seen or existing:
append_line(skipped_path, f"{post_id} already downloaded {existing or ''}".rstrip())
continue
try:
if args.dry_run:
append_line(skipped_path, f"{post_id} dry-run")
continue
resolved = resolve_media(
session,
item,
tries=args.tries,
base_retry_delay=args.base_retry_delay,
)
if not resolved:
append_line(failed_path, f"{post_id} no media found")
continue
source_url, response = resolved
try:
dest = destination_for(out_dir, post_id, source_url, response)
size = stream_download_response(response, dest)
finally:
response.close()
success_index += 1
append_download_order(manifest, success_index, post_id, dest, source_url)
append_line(seen_path, f"{post_id} {dest.as_posix()} {size}")
except Exception as exc: # keep batch running and log plain text failure
append_line(failed_path, f"{post_id} {type(exc).__name__}: {exc}")
time.sleep(args.delay)
return 0
if __name__ == "__main__":
raise SystemExit(main())
requirements.txt
+3
View File
@@ -0,0 +1,3 @@
requests>=2.34
beautifulsoup4>=4.15
tqdm>=4.68
tests/test_r34vault_downloader.py
+85
View File
@@ -0,0 +1,85 @@
import json
from pathlib import Path
import pytest
import r34vault_downloader as dl
def test_extract_bookmark_items_from_ng_state_preserves_order():
state = {
"get:/api/v2/account/user/krosh": {"id": 50311, "userName": "krosh"},
"post:/api/v2/post/search/bookmarked/50311": {
"items": [
{"id": 1075647, "type": 0},
{"id": 996379, "type": 1},
{"id": 1263322, "type": 0},
]
},
}
html = f'<script id="ng-state" type="application/json">{json.dumps(state)}</script>'
parsed = dl.parse_ng_state(html)
user_id = dl.extract_user_id(parsed, "krosh")
items = dl.extract_bookmark_items_from_state(parsed, user_id)
assert user_id == 50311
assert [item["id"] for item in items] == [1075647, 996379, 1263322]
def test_candidate_image_urls_prefers_original_rule34vault_then_cdn():
assert dl.post_prefix("1263322") == "1263"
assert dl.post_prefix("738794") == "738"
assert list(dl.candidate_image_urls("1263322"))[:4] == [
"https://rule34vault.com/posts/1263/1263322/1263322.jpg",
"https://r34xyz.b-cdn.net/posts/1263/1263322/1263322.jpg",
"https://rule34vault.com/posts/1263/1263322/1263322.png",
"https://r34xyz.b-cdn.net/posts/1263/1263322/1263322.png",
]
def test_original_from_derivative_url_removes_small_preview_suffix():
assert dl.original_from_derivative_url(
"https://rule34vault.com/posts/1263/1263303/1263303.small.jpg", "1263303"
) == "https://rule34vault.com/posts/1263/1263303/1263303.jpg"
assert dl.original_from_derivative_url(
"/posts/1263/1263303/1263303.preview.webp", "1263303"
) == "/posts/1263/1263303/1263303.webp"
def test_extract_media_sources_from_post_html_finds_video_and_originalized_images():
html = """
<html><body>
<img class="img" src="/posts/1263/1263303/1263303.small.jpg">
<video controls>
<source type="video/mp4; codecs=hvc1" src="https://r34xyz.b-cdn.net/posts/996/996379/996379.720.hevc.mp4">
<source type="video/mp4" src="https://r34xyz.b-cdn.net/posts/996/996379/996379.480.mp4">
</video>
</body></html>
"""
sources = dl.extract_media_sources_from_html(html, "1263303", "https://rule34vault.com/post/1263303")
assert "https://rule34vault.com/posts/1263/1263303/1263303.jpg" in sources.images
assert "https://rule34vault.com/posts/1263/1263303/1263303.small.jpg" in sources.image_fallbacks
assert sources.videos == [
"https://r34xyz.b-cdn.net/posts/996/996379/996379.720.hevc.mp4",
"https://r34xyz.b-cdn.net/posts/996/996379/996379.480.mp4",
]
def test_choose_video_source_prefers_non_hevc_mp4():
sources = [
"https://r34xyz.b-cdn.net/posts/996/996379/996379.720.hevc.mp4",
"https://r34xyz.b-cdn.net/posts/996/996379/996379.480.mp4",
]
assert dl.choose_video_source(sources) == "https://r34xyz.b-cdn.net/posts/996/996379/996379.480.mp4"
def test_append_download_order_writes_plain_text_manifest(tmp_path):
manifest = tmp_path / "download_order.txt"
dl.append_download_order(manifest, 3, "996379", Path("downloads/996/996379.mp4"), "https://example/file.mp4")
assert manifest.read_text() == "000003 996379 downloads/996/996379.mp4 https://example/file.mp4\n"