Ben_Kosytorz/Warhammer_DLP_Helper
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial version: dlp.py orchestrates Discord export, link extraction and host-aware downloads

Browse Source 
- dlphelper/discord.py: DCE docker export (JSON) + base64/plain URL extraction
  (extraction regex kept compatible with parts/grab.sh)
- dlphelper/gofile.py: GoFile client adapted from Lysagxra/GoFileDownloader's
  API scheme, but ONE guest account per run and per-file failure tracking
  instead of sys.exit - fixes the rate-limit death after ~6 downloads
- dlphelper/zero807.py: curl downloader with the UA+Referer contract 0807.st
  requires; .part files for resumable downloads
- dlphelper/state.py: state.json (resume/skip) + manifest.jsonl (torrent stage
  later) + failed_urls.txt
- tools/new_repo.py: Gitea repo helper reading ~/.git-credentials itself
- AGENTS.md with agent rules, README, offline extraction test
This commit is contained in:
Ben_Kosytorz
2026-10-04 14:06:46 +02:00
commit db8a1c1b52
13 changed files with 1389 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+20
View File
@@ -0,0 +1,20 @@
# Python
__pycache__/
*.pyc
.venv/
*.egg-info/
# Run outputs (if OUT_DIR points into the repo)
thread/
downloads/
export.json
export.html
state.json
state.json.tmp
failed_urls.txt
manifest.jsonl
*.log
# Never commit credentials
*.token
.env
AGENTS.md
+177
View File
@@ -0,0 +1,177 @@
# AGENTS.md — Warhammer_DLP_Helper
Guidance for any AI agent (or human) working in this repository. Read this
fully before making changes.
## 1. What this project is
The user collects **Warhammer proxy models** (3rd-party / fan sculpts, never
official Games Workshop assets) that are shared in Discord channels. Links in
those channels point to two hosters and are often **base64-encoded** inside
message content (obfuscated by the posting bot).
The end goal is a **torrent-first platform** (think radarr/lidarr, but for
proxy files). This repo is the current stage of that journey: a tool that
1. exports a Discord channel/thread (DiscordChatExporter via docker),
2. extracts base64-encoded and plain download links,
3. picks the right downloader per host automatically,
4. downloads everything with per-file tracking, retries and resume,
5. records a manifest that will later feed the torrent stage.
Deliberately **not** in scope (yet): .torrent creation, tracker upload,
deseeding, rename-into-library conventions. Design decisions should not close
the door on these — especially `manifest.jsonl` and the `--hash` option exist
for dedupe/cross-seeding later.
## 2. Repository & environment
- Remote: `git.dieterunterstrich.de/Ben_Kosytorz/Warhammer_DLP_Helper` —
**private** repo, branch `main`. Push with the credentials already stored
via git's credential store (`~/.git-credentials`); never paste them into
commands or files.
- The server is Gitea 1.27. To create *another* repo, use
`python3 tools/new_repo.py list|create <name>` (reads `~/.git-credentials`
itself, never prints secrets). Never delete or overwrite existing repos.
- Machine: Linux, Python 3.12+ (3.14 available), `requests` is the only
runtime dependency (see `requirements.txt`), docker is available.
- Reference clone: `~/GoFileDownloader` (Lysagxra) — do not modify it; it is
upstream reading material only.
## 3. Architecture
```
dlp.py CLI orchestrator (entry point)
dlphelper/
discord.py DCE docker export (JSON) + base64/plain URL extraction
gofile.py GoFile API client (one guest account per run)
zero807.py curl-based downloader for 0807.st direct links
state.py state.json store + manifest.jsonl appender
parts/grab.sh legacy groundstone script (kept, not used by dlp.py)
tools/new_repo.py Gitea repo helper (reads ~/.git-credentials itself)
```
Data flow:
```mermaid
graph LR
A[channel id + token] --> B[DCE docker export -> export.json]
B --> C[decode base64 in message content]
C --> D{classify host}
D -->|gofile.io| E[GoFileClient: 1 guest account, list contents, download files]
D -->|0807.st| F[curl with UA + Referer]
D -->|other| G[recorded as failed: unsupported host]
E --> H[downloads/gofile/...]
F --> I[downloads/0807/...]
E & F --> J[state.json + manifest.jsonl + failed_urls.txt]
```
Key facts about the hosters (learned the hard way, do not re-learn):
- **gofile.io**: guest accounts are created via `POST api.gofile.io/accounts`.
Creation is aggressively rate-limited — create **at most one account per
run**. Content listing needs `Authorization: Bearer <token>`, an
`X-Website-Token` header (sha256 over `UA::en-US::<token>::<4h-window>::seed`,
see `gofile.py`), and downloads need `Cookie: accountToken=<token>` plus
Referer/Origin headers.
- **0807.st**: serves files directly but 403s anything that doesn't look like
a browser. `curl -A "Mozilla/5.0" -e "https://0807.st/"` works; aria2c's
default multi-connection request does not. curl is the tool here.
- Encoded links in the channels start as base64 strings matching
`aHR0cHM6Ly[89]...` — keep the extraction regex compatible with
`parts/grab.sh`.
## 4. Usage (what "done" looks like)
export DISCORD_TOKEN=...
python3 dlp.py <channel_id> /big/storage/warhammer/<channel-name>
Retry only failures: `python3 dlp.py <id> <out> --urls-file <out>/failed_urls.txt`
(or just re-run — state.json skips everything already marked ok).
## 5. Run artifacts (all in OUT_DIR, gitignored)
| File | Purpose |
|------|---------|
| `export.json` | DCE export; cached — `--fresh` re-exports |
| `state.json` | url → {status, attempts, error, files[]}; corrupt files are backed up and reset |
| `failed_urls.txt` | rewritten each run: every non-ok URL, usable as `--urls-file` |
| `manifest.jsonl` | append-only, one JSON line per downloaded file (channel, url, message_id, timestamp, author, path, size, optional sha256) |
| `downloads/` | actual content; `.part` files mean "interrupted, will be resumed" |
## 6. Rules
### Secrets and tokens
1. Never print, log, commit or echo a Discord token or GoFile account token.
Not even partial. Strip them from any output you paste into chats.
2. The Discord token goes: `--token` flag → `$DISCORD_TOKEN` → hidden prompt.
It is handed to the DCE container via the `TOKEN` env var, never as a
command-line argument.
3. `~/.git-credentials` may only be read by `tools/new_repo.py`. Never copy
credentials into scripts, tests, or fixtures.
4. Nothing downloaded and no export/state/manifest file is ever committed.
### Failure policy (the core requirement)
5. A failure in one URL or file must **never** abort the run. Catch, record in
`state.json`, continue. The loops in `dlp.py` therefore have catch-alls —
keep them.
6. Cooldowns are the only permitted "stall": after N consecutive failures,
pause, then continue. Circuit-breakers that exit the process need a
explicit justification in the PR/commit message.
7. Retries: exponential-ish (`retry_delay * attempt`), default 3 attempts.
Password-required gofile links are not retried (it can't help).
### Idempotency
8. Re-runs must be safe and cheap: skip URLs marked `ok`, skip existing
non-empty files, write partial downloads to `*.part` and rename on success.
9. `state.json` is the source of truth for resume; keep its schema backward
compatible when touching `state.py` (unknown keys are preserved, missing
keys defaulted).
### Rate limiting / network etiquette
10. One gofile guest account per run; `ensure_token()` backs off linearly
(20s × attempt). Do not raise the guest-account frequency "to make tests
pass". Testing account creation more than a couple times per hour is a
bug, not a test strategy.
11. Keep the 0807 referer/UA contract. Don't "optimize" it to aria2c without
re-verifying the 403 behavior.
### Code style
12. Python 3.12+, type hints on public functions, `logging` not `print` in
library code (prints are fine only for `--dry-run` output).
13. Stay on stdlib + `requests`. New dependencies need a real justification
and belong in `requirements.txt`.
14. `dlphelper/` modules must stay import-safe (no side effects at import).
### Validation
15. Minimum validation for any change: `python3 -m py_compile dlp.py
dlphelper/*.py`, plus the synthetic extraction test (below) if
`discord.py` changed. Document what you could *not* test (e.g. live
gofile/0807 downloads need real links and tokens).
16. Full live runs need a Discord token from the user — ask for it, never
ask for it to be pasted into a file or committed.
### Git
17. Branch `main`, descriptive commit messages, no force-push, no history
rewriting on the shared remote.
18. When creating a new repo use `tools/new_repo.py` (Gitea API); creation is
fine, deletion/renaming of existing repos requires the user.
## 7. Roadmap (future stages, keep compatible)
1. **Dedupe**: hash-based (sha256) duplicate detection across threads via
`manifest.jsonl`.
2. **Torrent stage**: `mktorrent` per downloaded folder, .torrent files into
a watch dir, cross-seeding between direct-download and torrent copies.
3. **Library organization**: naming conventions per game system / faction so
the collection is browsable like a media library.
4. **Watch mode**: periodic re-export of subscribed channels, notify on new
links.
## 8. Testing notes
- `tests/` currently holds a synthetic `export.json` test for extraction
(`tests/test_extract.py`, runnable with plain `python3`). Keep tests
dependency-free and offline — they must never hit gofile/0807/Discord.
- To exercise downloads without Discord: build a `--urls-file` from known
URLs and run with a scratch OUT_DIR.
README.md
+58
View File
@@ -0,0 +1,58 @@
# Warhammer DLP Helper
A small "radarr for Warhammer proxy files" tool: point it at a Discord
channel/thread and an output directory, and it exports the channel, extracts
the encoded download links, and downloads everything - automatically choosing
the right downloader per host:
| Host | Downloader | Notes |
|-------------|------------|-------|
| gofile.io | built-in GoFile API client | one guest account per run (rate-limit friendly) |
| 0807.st | curl | needs browser-like headers (`-A`, `-e https://0807.st/`) |
## Install
pip install -r requirements.txt # just `requests`
docker pull tyrrrz/discordchatexporter:stable
## Usage
export DISCORD_TOKEN="..."
python3 dlp.py <channel_or_thread_id> /path/to/output
The token can also be passed with `--token` (or you'll be prompted, input
hidden). Useful options:
--urls-file failed_urls.txt retry a previous failure list, no re-export
--password <pw> for password-protected gofile links
--only gofile|0807 process only one host
--extract unzip archives into per-archive folders
--hash record sha256 per file in the manifest
--dry-run just show what would be downloaded
--fresh force a fresh channel export
## Output layout
OUT_DIR/
export.json Discord channel export (cached between runs)
state.json per-URL status; re-runs skip everything marked "ok"
failed_urls.txt URLs that failed after all retries
manifest.jsonl one line per downloaded file (url, source message, path, size)
downloads/
gofile/<content>/<folders>/<files>
0807/<id>.zip
Runs are resumable and idempotent: interrupted downloads use `.part` files and
finished ones are skipped on the next run. Individual failures never stop the
run; after 3 consecutive failures it pauses for a cooldown (rate-limit
protection) and continues.
## Why not use Lysagxra/GoFileDownloader directly?
That tool creates a new GoFile guest account **per URL** and calls
`sys.exit(1)` on the first failure, which is exactly why runs died after ~6
downloads. This project reimplements its (working) API scheme - headers,
`X-Website-Token`, download links with the `accountToken` cookie - but creates
one guest account per run, records failures per file, and keeps going.
`parts/grab.sh` is the original groundstone script, kept for reference.
dlp.py
+379
View File
@@ -0,0 +1,379 @@
#!/usr/bin/env python3
"""Warhammer DLP Helper.
Give it a Discord channel/thread id and an output directory; it exports the
channel, extracts the encoded download links (gofile.io and 0807.st), picks
the right downloader for each link, and keeps going when individual links
fail. Everything is recorded in state.json; failures also land in
failed_urls.txt so a re-run (or --urls-file) retries exactly those.
Usage:
python3 dlp.py <channel_id> <out_dir> [options]
"""
from __future__ import annotations
import argparse
import getpass
import hashlib
import json
import logging
import os
import sys
import time
import zipfile
from pathlib import Path
from dlphelper import discord as dlp_discord
from dlphelper import gofile as dlph_gofile
from dlphelper import zero807 as dlph_zero807
from dlphelper.state import State, append_manifest
log = logging.getLogger("dlp")
def sha256_of(path: Path) -> str:
digest = hashlib.sha256()
with path.open("rb") as fh:
for chunk in iter(lambda: fh.read(1024 * 1024), b""):
digest.update(chunk)
return digest.hexdigest()
def resolve_token(cli_value: str | None) -> str:
if cli_value:
return cli_value
if os.environ.get("DISCORD_TOKEN"):
return os.environ["DISCORD_TOKEN"]
log.info("No Discord token given ($DISCORD_TOKEN); prompting (input hidden).")
token = getpass.getpass("Discord token: ")
if not token:
raise SystemExit("No token provided, aborting.")
return token
def load_urls_file(path: Path) -> list[dlp_discord.FoundUrl]:
"""Load a plain URL list (e.g. failed_urls.txt); no attribution available."""
found: dict[str, dlp_discord.FoundUrl] = {}
for line in path.read_text(encoding="utf-8").splitlines():
url = line.strip()
if not url or url.startswith("#"):
continue
host = dlp_discord.classify(url)
if host is None:
log.warning("Unsupported host in %s: %s", path.name, url)
continue
found.setdefault(url, dlp_discord.FoundUrl(url=url, host=host))
return sorted(found.values(), key=lambda f: f.url)
def record_files(state: State, url: str, files: list[dict], source: dict) -> None:
state.mark(url, status="ok", error=None, files=files)
for f in files:
append_manifest(Path(source["manifest"]), {
"channel": source["channel"],
"url": url,
"host": source["host"],
"message_id": source.get("message_id"),
"timestamp": source.get("timestamp"),
"author": source.get("author"),
**f,
})
def extract_zips(files: list[dict]) -> None:
"""Optional: unzip each downloaded archive next to itself."""
for f in files:
path = Path(f["path"])
if path.suffix.lower() != ".zip" or not path.exists():
continue
target = path.with_suffix("")
if target.exists() and any(target.iterdir()):
log.info("Skip extracting %s (target exists)", path.name)
continue
log.info("Extracting %s -> %s", path.name, target.name)
try:
target.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(path) as zf:
zf.extractall(target)
except (zipfile.BadZipFile, OSError) as exc:
log.warning("Extraction of %s failed: %s", path.name, exc)
def run_gofile(
urls: list[dlp_discord.FoundUrl],
args: argparse.Namespace,
state: State,
source_by_url: dict[str, dict],
) -> None:
client = dlph_gofile.GoFileClient(password=args.password)
try:
client.ensure_token()
except dlph_gofile.GoFileError as exc:
log.error("%s - marking all pending gofile links as failed.", exc)
for f in urls:
if state.is_done(f.url):
continue
state.mark(f.url, status="failed", error=str(exc),
files=[])
return
consecutive_failures = 0
for f in urls:
if state.is_done(f.url):
continue
source = source_by_url[f.url]
for attempt in range(1, args.retries + 1):
try:
files_spec = client.list_files(f.url)
if not files_spec:
raise dlph_gofile.GoFileError("no files found under URL")
files: list[dict] = []
for relpath, name, link in files_spec:
dest = Path(args.out_dir) / "downloads" / "gofile" / relpath / name
size = client.download_file(link, dest)
entry = {"path": str(dest), "size": size, "name": name}
if args.hash:
entry["sha256"] = sha256_of(dest)
files.append(entry)
log.info("OK %s (%d files)", f.url, len(files))
record_files(state, f.url, files, source)
if args.extract:
extract_zips(files)
consecutive_failures = 0
break
except dlph_gofile.GoFilePasswordRequired as exc:
# Retrying without a password never helps
log.error("PASSWORD REQUIRED %s: %s", f.url, exc)
state.mark(f.url, status="failed", error=str(exc), files=[])
break
except dlph_gofile.GoFileAuthError as exc:
# Token may have died mid-run (rate limit); re-auth once per attempt
log.warning("Auth error for %s: %s - re-authenticating", f.url, exc)
state.mark(f.url, error=str(exc), files=[])
client.reset()
try:
client.ensure_token()
except dlph_gofile.GoFileError as reauth_exc:
log.error("Re-authentication failed: %s", reauth_exc)
if attempt < args.retries:
time.sleep(args.retry_delay * attempt)
except dlph_gofile.GoFileError as exc:
log.warning("Attempt %d/%d failed for %s: %s",
attempt, args.retries, f.url, exc)
state.mark(f.url, error=str(exc), files=[])
if attempt < args.retries:
time.sleep(args.retry_delay * attempt)
except Exception as exc: # noqa: BLE001 - a bug must not stop the run
log.exception("Unexpected error for %s: %s", f.url, exc)
state.mark(f.url, error=f"unexpected: {exc}", files=[])
if attempt < args.retries:
time.sleep(args.retry_delay * attempt)
except KeyboardInterrupt:
state.mark(f.url, error="interrupted", files=[])
raise
else:
rec = state.records.get(f.url, {})
state.mark(f.url, status="failed", error=rec.get("error", "unknown"), files=[])
consecutive_failures += 1
log.error("FAILED %s after %d attempts", f.url, args.retries)
if consecutive_failures >= args.cooldown_failures:
log.warning(
"%d failures in a row - cooling down %ds before continuing "
"(rate limit?). The run does not stop.",
consecutive_failures, args.cooldown_seconds,
)
time.sleep(args.cooldown_seconds)
def run_0807(
urls: list[dlp_discord.FoundUrl],
args: argparse.Namespace,
state: State,
source_by_url: dict[str, dict],
) -> None:
out_dir = Path(args.out_dir) / "downloads" / "0807"
consecutive_failures = 0
for f in urls:
if state.is_done(f.url):
continue
source = source_by_url[f.url]
for attempt in range(1, args.retries + 1):
try:
dest, size = dlph_zero807.download(f.url, out_dir)
entry: dict = {"path": str(dest), "size": size, "name": dest.name}
if args.hash:
entry["sha256"] = sha256_of(dest)
log.info("OK %s (%s)", f.url, format_size(size))
record_files(state, f.url, [entry], source)
if args.extract:
extract_zips([entry])
consecutive_failures = 0
break
except dlph_zero807.Zero807Error as exc:
log.warning("Attempt %d/%d failed for %s: %s",
attempt, args.retries, f.url, exc)
state.mark(f.url, error=str(exc), files=[])
if attempt < args.retries:
time.sleep(args.retry_delay * attempt)
except Exception as exc: # noqa: BLE001 - a bug must not stop the run
log.exception("Unexpected error for %s: %s", f.url, exc)
state.mark(f.url, error=f"unexpected: {exc}", files=[])
if attempt < args.retries:
time.sleep(args.retry_delay * attempt)
except KeyboardInterrupt:
state.mark(f.url, error="interrupted", files=[])
raise
else:
rec = state.records.get(f.url, {})
state.mark(f.url, status="failed", error=rec.get("error", "unknown"), files=[])
consecutive_failures += 1
log.error("FAILED %s after %d attempts", f.url, args.retries)
if consecutive_failures >= args.cooldown_failures:
log.warning(
"%d failures in a row - cooling down %ds before continuing "
"(HTTP 403 storm?). The run does not stop.",
consecutive_failures, args.cooldown_seconds,
)
time.sleep(args.cooldown_seconds)
def format_size(n: int) -> str:
size = float(n)
for unit in ("B", "KiB", "MiB", "GiB", "TiB"):
if size < 1024 or unit == "TiB":
return f"{size:.1f} {unit}"
size /= 1024
return f"{n} B"
def write_failed_urls(out_dir: Path, state: State) -> Path:
dest = out_dir / "failed_urls.txt"
failures = state.failures()
dest.write_text("\n".join(failures) + ("\n" if failures else ""), encoding="utf-8")
return dest
def main() -> None:
parser = argparse.ArgumentParser(
description="Collect downloads from a Discord channel (gofile + 0807)."
)
parser.add_argument("channel_id", help="Discord channel or thread id")
parser.add_argument("out_dir", help="directory for downloads and state")
parser.add_argument("--token", help="Discord token (else $DISCORD_TOKEN or prompt)")
parser.add_argument("--urls-file", type=Path,
help="skip the Discord export; read URLs from this file "
"(e.g. failed_urls.txt)")
parser.add_argument("--password", help="password for password-protected gofile links")
parser.add_argument("--retries", type=int, default=3, help="attempts per URL (default 3)")
parser.add_argument("--retry-delay", type=int, default=5,
help="base seconds between retries, multiplied per attempt (default 5)")
parser.add_argument("--cooldown-failures", type=int, default=3,
help="consecutive failures before a cooldown pause (default 3)")
parser.add_argument("--cooldown-seconds", type=int, default=60,
help="length of the cooldown pause (default 60)")
parser.add_argument("--only", choices=("gofile", "0807"),
help="process only one host class")
parser.add_argument("--hash", action="store_true",
help="record sha256 of each downloaded file")
parser.add_argument("--extract", action="store_true",
help="unzip downloaded archives into per-archive folders")
parser.add_argument("--fresh", action="store_true",
help="re-export the Discord channel even if export.json exists")
parser.add_argument("--dry-run", action="store_true",
help="export/extract/classify only, download nothing")
args = parser.parse_args()
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(message)s",
datefmt="%H:%M:%S",
)
out_dir = Path(args.out_dir)
state = State(out_dir / "state.json")
channel = args.channel_id
if args.urls_file:
found = load_urls_file(args.urls_file)
else:
token = resolve_token(args.token)
export_path = dlp_discord.export_channel(
channel, out_dir, token, force=args.fresh
)
found = dlp_discord.extract_urls(export_path)
if args.only:
found = [f for f in found if f.host == args.only]
buckets: dict[str, list[dlp_discord.FoundUrl]] = {"gofile": [], "0807": []}
for f in found:
if f.host in buckets:
buckets[f.host].append(f)
else: # classify() already filtered, but be defensive
state.mark(f.url, status="failed", error=f"unsupported host: {f.url}")
unsupported = [
f for f in found if f.host not in buckets
]
for f in unsupported:
log.warning("Skipping unsupported URL: %s", f.url)
manifest = out_dir / "manifest.jsonl"
source_by_url: dict[str, dict] = {
f.url: {
"channel": channel,
"host": f.host,
"message_id": f.message_id,
"timestamp": f.timestamp,
"author": f.author,
"manifest": str(manifest),
}
for f in found
}
log.info(
"Found %d URLs (%d gofile, %d 0807); %d already done from earlier runs",
len(found),
len(buckets["gofile"]),
len(buckets["0807"]),
len(found) - len(state.pending([f.url for f in found])),
)
if args.dry_run:
for f in found:
print(f"[{f.host:6}] {f.url}")
return
if buckets["gofile"]:
run_gofile(buckets["gofile"], args, state, source_by_url)
if buckets["0807"]:
run_0807(buckets["0807"], args, state, source_by_url)
failed_file = write_failed_urls(out_dir, state)
# Summary
ok = [u for u, r in state.records.items() if r.get("status") == "ok"]
bad = state.failures()
total_files = sum(len(r.get("files") or []) for r in state.records.values())
total_size = sum(
(f_.get("size") or 0) for r in state.records.values() for f_ in (r.get("files") or [])
)
log.info("=" * 60)
log.info("Done: %d ok (%d files, %s) / %d failed / %d total seen",
len(ok), total_files, format_size(total_size), len(bad), len(state.records))
if bad:
log.info("Failed URLs written to %s - re-run this tool or use "
"python3 dlp.py <id> <out> --urls-file %s to retry them.",
failed_file, failed_file.name)
sys.exit(1)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
log.warning("Interrupted - state.json is saved; re-run to resume.")
sys.exit(130)
dlphelper/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Warhammer DLP Helper - collect and organize files shared in Discord channels."""
__version__ = "0.1.0"
dlphelper/discord.py
+157
View File
@@ -0,0 +1,157 @@
"""Export a Discord channel/thread with DiscordChatExporter and extract URLs.
The channels this tool targets encode their download links as base64 inside
message content, and occasionally as plain links. We export the channel as
JSON (richer metadata than the HTML export: message id, timestamp, author),
decode every base64 candidate, and keep only URLs for supported hosts.
"""
from __future__ import annotations
import base64
import json
import logging
import os
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlparse
log = logging.getLogger(__name__)
DCE_IMAGE = "tyrrrz/discordchatexporter:stable"
# The channels encode links as base64 that always starts with this prefix
# (https://...); grab.sh established this regex, keep it compatible.
B64_RE = re.compile(r"aHR0cHM6Ly[89][A-Za-z0-9+/_-]*")
URL_RE = re.compile(r"https://[^\s<>\"'\)\]\}]+")
SUPPORTED_HOSTS = {
"gofile": ("gofile.io", "www.gofile.io"),
"0807": ("0807.st", "www.0807.st"),
}
@dataclass(frozen=True)
class FoundUrl:
"""One download URL found in the channel, with source attribution."""
url: str
host: str # "gofile" | "0807" (set after classification)
message_id: str | None = None
timestamp: str | None = None
author: str | None = None
encoded: bool = False # found via base64 decode vs. plain link
def export_channel(channel_id: str, out_dir: Path, token: str, force: bool = False) -> Path:
"""Export the channel to out_dir/export.json using the DCE docker image.
The token is passed to the container via the TOKEN env var (DCE reads it
natively) so it never appears on the docker command line.
"""
dest = out_dir / "export.json"
if dest.exists() and not force:
log.info("Reusing existing export %s (pass --fresh to re-export)", dest)
return dest
out_dir.mkdir(parents=True, exist_ok=True)
env = dict(os.environ)
env["TOKEN"] = token
cmd = [
"docker", "run", "--rm",
"--user", f"{os.getuid()}:{os.getgid()}",
"-v", f"{out_dir.resolve()}:/out",
"-e", "TOKEN",
DCE_IMAGE,
"export", "-c", channel_id, "-f", "Json", "-o", "/out/export.json",
]
log.info("Exporting channel %s via DiscordChatExporter ...", channel_id)
try:
subprocess.run(cmd, env=env, check=True)
except FileNotFoundError as exc:
raise RuntimeError(
"docker not found - the Discord export needs the docker image "
f"{DCE_IMAGE}"
) from exc
except subprocess.CalledProcessError as exc:
raise RuntimeError(
f"DiscordChatExporter failed with exit code {exc.returncode} "
"(check the token and the channel id)"
) from exc
if not dest.exists():
raise RuntimeError("DiscordChatExporter produced no output file")
return dest
def classify(url: str) -> str | None:
"""Return the host key ('gofile' | '0807') for a URL, else None."""
try:
hostname = (urlparse(url).hostname or "").lower()
except ValueError:
return None
for key, hosts in SUPPORTED_HOSTS.items():
if hostname in hosts:
return key
return None
def _decode_b64(candidate: str) -> str | None:
s = candidate.replace("_", "/").replace("-", "+")
s = s.rstrip("=")
s += "=" * ((4 - len(s) % 4) % 4)
try:
decoded = base64.b64decode(s, validate=False).decode("utf-8", "strict")
except Exception: # noqa: BLE001 - malformed candidate, just skip it
return None
# One encoded blob can contain the URL plus padding/newlines
stripped = decoded.strip()
line = stripped.splitlines()[0] if stripped else ""
return line or None
def extract_urls(export_path: Path) -> list[FoundUrl]:
"""Extract supported URLs from a DCE JSON export, deduplicated."""
data = json.loads(export_path.read_text(encoding="utf-8"))
messages = data if isinstance(data, list) else data.get("messages", [])
found: dict[str, FoundUrl] = {}
stats = {"b64": 0, "plain": 0, "unsupported": 0}
for msg in messages:
content = str(msg.get("content") or "")
if not content:
continue
author = (msg.get("author") or {}).get("name")
attrib = {
"message_id": str(msg.get("id") or "") or None,
"timestamp": msg.get("timestamp"),
"author": author,
}
for match in B64_RE.finditer(content):
stats["b64"] += 1
decoded = _decode_b64(match.group(0))
if not decoded:
continue
host = classify(decoded)
if host is None:
stats["unsupported"] += 1
continue
found.setdefault(decoded, FoundUrl(url=decoded, host=host, encoded=True, **attrib))
for match in URL_RE.finditer(content):
url = match.group(0).rstrip(".,;)")
host = classify(url)
if host is None:
continue
stats["plain"] += 1
found.setdefault(url, FoundUrl(url=url, host=host, encoded=False, **attrib))
log.info(
"Extraction: %d base64 candidates, %d plain links, %d unsupported/undecodable, "
"%d unique supported URLs",
stats["b64"], stats["plain"], stats["unsupported"], len(found),
)
return sorted(found.values(), key=lambda f: f.url)
dlphelper/gofile.py
+219
View File
@@ -0,0 +1,219 @@
"""Minimal GoFile API client - one guest account per run.
The API interaction mirrors Lysagxra/GoFileDownloader (same headers, website
token scheme and endpoints), but with two deliberate differences:
1. A single guest account is created per run (GoFileDownloader creates one per
URL, which trips Gofile's rate limit after ~6 requests).
2. Failures raise exceptions instead of sys.exit(), so the orchestrator can
track them per file/URL and keep going.
"""
from __future__ import annotations
import hashlib
import logging
import os
import time
from pathlib import Path
from urllib.parse import urlparse
import requests
log = logging.getLogger(__name__)
API = "https://api.gofile.io"
USER_AGENT = "Mozilla/5.0"
LOCALE = "en-US"
TOKEN_SECRET = "12af056dacea0b" # public website-token seed, see GoFileDownloader # noqa: S105
TOKEN_WINDOW_SEC = 14400
CHUNK_SIZE = 1024 * 1024
COMMON_HEADERS = {
"User-Agent": USER_AGENT,
"Accept-Encoding": "gzip",
"Accept": "*/*",
"Connection": "keep-alive",
}
EXTENDED_HEADERS = {
**COMMON_HEADERS,
"Sec-Fetch-Dest": "empty",
"Sec-Fetch-Mode": "cors",
"Sec-Fetch-Site": "same-site",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
}
class GoFileError(Exception):
"""Base error for GoFile operations."""
class GoFilePasswordRequired(GoFileError):
"""The content is password protected and none (or a wrong one) was given."""
class GoFileAuthError(GoFileError):
"""The account token was rejected; the client can re-auth once."""
def content_id_from_url(url: str) -> str:
"""Extract the content id from a gofile.io/d/<id> URL."""
parts = [p for p in urlparse(url).path.split("/") if p]
if len(parts) >= 2 and parts[-2] == "d":
return parts[-1]
raise GoFileError(f"not a gofile content URL (expected gofile.io/d/<id>): {url}")
class GoFileClient:
"""GoFile API client that reuses one guest account token for the whole run."""
def __init__(self, password: str | None = None):
self._password_hash = (
hashlib.sha256(password.encode()).hexdigest() if password else None
)
self._token: str | None = None
self.session = requests.Session()
# -- authentication -----------------------------------------------------
def ensure_token(self, max_attempts: int = 8, wait_seconds: int = 20) -> str:
"""Create a guest account; backs off and retries on rate limiting."""
if self._token:
return self._token
last_status = "unknown"
for attempt in range(1, max_attempts + 1):
try:
resp = self.session.post(
f"{API}/accounts", headers=COMMON_HEADERS, timeout=15
)
data = resp.json()
except (ValueError, requests.RequestException) as exc:
last_status = f"request error: {exc}"
log.warning(
"GoFile account creation attempt %d/%d failed: %s",
attempt, max_attempts, last_status,
)
else:
if isinstance(data, dict) and data.get("status") == "ok":
self._token = data["data"]["token"]
log.info("GoFile guest account created (token reused for whole run)")
return self._token
last_status = str(data.get("status", data)[:200] if data else data)
log.warning(
"GoFile account creation attempt %d/%d failed: %s",
attempt, max_attempts, last_status,
)
if attempt < max_attempts:
time.sleep(wait_seconds * attempt)
raise GoFileError(
f"GoFile account creation failed after {max_attempts} attempts "
f"(last: {last_status})"
)
def reset(self) -> None:
"""Drop the current token so the next ensure_token() creates a new one."""
self._token = None
def _website_token(self) -> str:
window = int(time()) // TOKEN_WINDOW_SEC
seed = f"{USER_AGENT}::{LOCALE}::{self._token}::{window}::{TOKEN_SECRET}"
return hashlib.sha256(seed.encode()).hexdigest()
def _auth_headers(self) -> dict:
return {
**EXTENDED_HEADERS,
"Authorization": f"Bearer {self._token}",
"X-Website-Token": self._website_token(),
"X-BL": LOCALE,
}
# -- listing ------------------------------------------------------------
def list_files(self, url: str) -> list[tuple[str, str, str]]:
"""Return (relpath, filename, download_link) for every file under a URL."""
cid = content_id_from_url(url)
results: list[tuple[str, str, str]] = []
self._walk(cid, [], results)
# Direct files at the content root get their own folder named after the id
normalized = [
(relpath if relpath else cid, name, link) for relpath, name, link in results
]
return normalized
def _walk(self, cid: str, prefix: list[str], results: list[tuple[str, str, str]]) -> None:
query = "?cache=true&sortField=createTime&sortDirection=1"
if self._password_hash:
query += f"&password={self._password_hash}"
try:
resp = self.session.get(
f"{API}/contents/{cid}{query}", headers=self._auth_headers(), timeout=30
)
except requests.RequestException as exc:
raise GoFileError(f"content request for {cid} failed: {exc}") from exc
if resp.status_code in (401, 403):
raise GoFileAuthError(f"content request denied ({resp.status_code}) for {cid}")
try:
data = resp.json()
except ValueError as exc:
raise GoFileError(f"non-JSON response for {cid}: {resp.text[:200]}") from exc
if data.get("status") != "ok":
raise GoFileError(f"content request failed for {cid}: {data.get('status')}")
d = data["data"]
if d.get("passwordStatus") not in (None, "passwordOk"):
raise GoFilePasswordRequired(
f"content {cid} is password protected (use --password)"
)
if d.get("type") == "folder":
here = prefix + [d["name"]]
for child in (d.get("children") or {}).values():
if child.get("type") == "folder":
self._walk(child["id"], here, results)
else:
results.append(("/".join(here), child["name"], child["link"]))
else:
results.append(("/".join(prefix), d["name"], d["link"]))
# -- downloading ----------------------------------------------------------
def download_file(self, link: str, dest: Path) -> int:
"""Download one file; returns the file size. Skips existing files."""
dest.parent.mkdir(parents=True, exist_ok=True)
if dest.exists() and dest.stat().st_size > 0:
log.info("Skip existing: %s", dest)
return dest.stat().st_size
tmp = dest.with_name(dest.name + ".part")
headers = {
**EXTENDED_HEADERS,
"Cookie": f"accountToken={self._token}",
"Referer": link,
"Origin": link,
}
try:
with self.session.get(
link, headers=headers, stream=True, timeout=(10, 60), allow_redirects=True
) as resp:
if resp.status_code != 200:
raise GoFileError(f"HTTP {resp.status_code} while downloading {dest.name}")
total = int(resp.headers.get("Content-Length") or 0)
written = 0
with tmp.open("wb") as fh:
for chunk in resp.iter_content(chunk_size=CHUNK_SIZE):
fh.write(chunk)
written += len(chunk)
if written == 0:
raise GoFileError(f"empty body for {dest.name}")
os.replace(tmp, dest)
return written
finally:
if tmp.exists() and not dest.exists():
try:
tmp.unlink()
except OSError:
pass
dlphelper/state.py
+78
View File
@@ -0,0 +1,78 @@
"""Persistent per-URL state: makes runs resumable and failures trackable.
state.json lives in the run's OUT_DIR and maps every URL seen to a record:
{
"status": "ok" | "failed" | "error",
"attempts": <int>,
"error": "<last error message>",
"files": [{"path": ..., "size": ...}, ...],
"updated": "<iso timestamp>"
}
Re-running skips URLs whose status is "ok"; failed URLs are retried and also
written to failed_urls.txt for easy manual retry lists.
"""
from __future__ import annotations
import json
import logging
import time
from pathlib import Path
log = logging.getLogger(__name__)
class State:
"""JSON-backed state store for one run directory."""
def __init__(self, path: Path):
self.path = path
self.records: dict[str, dict] = {}
if path.exists():
try:
self.records = json.loads(path.read_text(encoding="utf-8"))
if not isinstance(self.records, dict):
raise ValueError("state root must be an object")
except (ValueError, OSError) as exc:
backup = path.with_suffix(".json.bak")
log.warning("Corrupt state file %s (%s); moving to %s and starting fresh",
path, exc, backup)
path.replace(backup)
self.records = {}
def is_done(self, url: str) -> bool:
rec = self.records.get(url)
return bool(rec) and rec.get("status") == "ok"
def pending(self, urls: list[str]) -> list[str]:
return [u for u in urls if not self.is_done(u)]
def mark(self, url: str, **fields) -> dict:
rec = self.records.setdefault(url, {"status": "pending", "attempts": 0, "files": []})
for key, value in fields.items():
if value is not None:
rec[key] = value
rec["attempts"] = rec.get("attempts", 0) + 1
rec["updated"] = time.strftime("%Y-%m-%dT%H:%M:%S")
self.save()
return rec
def failures(self) -> list[str]:
return [u for u, r in self.records.items() if r.get("status") != "ok"]
def save(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
tmp = self.path.with_suffix(".json.tmp")
tmp.write_text(
json.dumps(self.records, indent=2, sort_keys=True), encoding="utf-8"
)
tmp.replace(self.path)
def append_manifest(manifest_path: Path, row: dict) -> None:
"""Append one JSON line to manifest.jsonl (per downloaded file)."""
manifest_path.parent.mkdir(parents=True, exist_ok=True)
with manifest_path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
dlphelper/zero807.py
+83
View File
@@ -0,0 +1,83 @@
"""Downloader for 0807.st direct links via curl.
0807.st serves the files directly but rejects requests that don't look like a
browser (requires a User-Agent and the Referer of the site page), and it
rejects multi-connection/range abuse patterns differently than curl - which is
why curl is used instead of aria2c here (see chat history: curl worked,
aria2c's default request got 403).
Downloads go to a .part file first and are renamed on success, so an
interrupted download is resumable (--continue-at -) and partial files are
never mistaken for finished ones.
"""
from __future__ import annotations
import logging
import os
import subprocess
from pathlib import Path
from urllib.parse import urlparse
log = logging.getLogger(__name__)
USER_AGENT = "Mozilla/5.0"
REFERER = "https://0807.st/"
class Zero807Error(Exception):
"""Base error for 0807 downloads."""
def target_name(url: str) -> str:
"""Derive the filename from the URL path (e.g. /6saCNie.zip -> 6saCNie.zip)."""
name = Path(urlparse(url).path).name
if not name:
raise Zero807Error(f"cannot derive a filename from URL: {url}")
return name
def download(
url: str,
out_dir: Path,
timeout_seconds: int = 7200,
resume: bool = True,
) -> tuple[Path, int]:
"""Download one URL with curl; returns (final_path, size_bytes)."""
name = target_name(url)
out_dir.mkdir(parents=True, exist_ok=True)
dest = out_dir / name
tmp = dest.with_name(dest.name + ".part")
if dest.exists() and dest.stat().st_size > 0:
log.info("Skip existing: %s", dest)
return dest, dest.stat().st_size
cmd = [
"curl", "-fSL",
"--connect-timeout", "20",
"--max-time", str(timeout_seconds),
"--retry", "2", "--retry-delay", "5",
"-A", USER_AGENT,
"-e", REFERER,
"-o", str(tmp),
]
if resume:
cmd.append("--continue-at")
cmd.append("-")
cmd.append(url)
log.info("Downloading %s -> %s", url, dest.name)
proc = subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
if proc.returncode != 0:
tail = [line for line in proc.stderr.strip().splitlines() if line][-3:]
raise Zero807Error(f"curl exit code {proc.returncode}: {' | '.join(tail)}")
size = tmp.stat().st_size if tmp.exists() else 0
if size == 0:
tmp.unlink(missing_ok=True)
raise Zero807Error(f"empty file for {name}")
if dest.suffix.lower() == ".zip" and not tmp.read_bytes()[:2] == b"PK":
log.warning("%s does not look like a ZIP file (first bytes are not PK)", name)
os.replace(tmp, dest)
return dest, size
parts/grab.sh
Executable
+38
View File
@@ -0,0 +1,38 @@
#!/usr/bin/env bash
set -euo pipefail
THREAD_ID="${THREAD_ID:-1455993943471820842}"
OUT_DIR="${OUT_DIR:-$PWD/thread}"
HTML="$OUT_DIR/export.html"
URLS="$OUT_DIR/urls.txt" # aria2c input file: one URL per line
# Token from env, otherwise prompt (input hidden)
if [[ -z "${DISCORD_TOKEN:-}" ]]; then
read -rsp "Discord token: " DISCORD_TOKEN
echo
fi
mkdir -p "$OUT_DIR"
# 1. Export the thread to HTML
docker run --rm --user "$(id -u):$(id -g)" \
-v "$OUT_DIR:/out" \
tyrrrz/discordchatexporter:stable export \
-t "$DISCORD_TOKEN" -c "$THREAD_ID" -f HtmlDark -o /out/export.html
unset DISCORD_TOKEN
# 2. Find base64 strings starting with "https://", then decode them
grep -oP 'aHR0cHM6Ly[89][A-Za-z0-9+/_-]*' "$HTML" \
| sort -u \
| while read -r s; do
s=$(printf '%s' "$s" | tr '_-' '/+' | tr -d '=')
while (( ${#s} % 4 )); do s="$s="; done
printf '%s' "$s" | base64 -d 2>/dev/null | tr -d '\r\n' || true
echo
done \
| grep -E '^https://[[:graph:]]+$' \
| awk '!seen[$0]++' > "$URLS"
echo "Found $(wc -l < "$URLS") URLs -> $URLS"
echo "Download with: aria2c -i \"$URLS\" -d \"$OUT_DIR/downloads\""
requirements.txt
+1
View File
@@ -0,0 +1 @@
requests>=2.31
tests/test_extract.py
+68
View File
@@ -0,0 +1,68 @@
#!/usr/bin/env python3
"""Offline extraction test with a synthetic export.json (no network, no deps).
Run with: python3 tests/test_extract.py
"""
from __future__ import annotations
import base64
import json
import sys
import tempfile
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from dlphelper.discord import classify, extract_urls # noqa: E402
def b64(s: str) -> str:
return base64.b64encode(s.encode()).decode()
def main() -> None:
messages = [
{
"id": "1",
"timestamp": "2026-01-01T00:00:00",
"author": {"name": "poster"},
"content": f"chaos stuff: {b64('https://gofile.io/d/abc123')}",
},
{
"id": "2",
"timestamp": "2026-01-02T00:00:00",
"author": {"name": "poster"},
"content": (
f"0807 zip: {b64('https://0807.st/6saCNie.zip')}\n"
f"dupe of 1: {b64('https://gofile.io/d/abc123')}\n"
f"plain: https://0807.st/plainLink.zip and "
f"https://discord.com/channels/1/2 (unsupported)\n"
f"junk: {b64('https://pixeldrain.com/u/xyz')} "
"and not-our-domain base64: aHR0cHM6Ly9leGFtcGxlLm9yZw=="
),
},
]
with tempfile.TemporaryDirectory() as tmp:
export = Path(tmp) / "export.json"
export.write_text(json.dumps({"messages": messages}), encoding="utf-8")
found = extract_urls(export)
urls = {f.url for f in found}
expected = {"https://gofile.io/d/abc123", "https://0807.st/6saCNie.zip",
"https://0807.st/plainLink.zip"}
assert urls == expected, f"unexpected urls: {sorted(urls)}"
by_url = {f.url: f for f in found}
g = by_url["https://gofile.io/d/abc123"]
assert g.host == "gofile" and g.message_id == "1" and g.author == "poster"
assert classify("https://0807.st/x.zip") == "0807"
assert classify("https://gofile.io/d/x") == "gofile"
assert classify("https://example.org/x") is None
print("test_extract: OK")
if __name__ == "__main__":
main()
tools/new_repo.py
+108
View File
@@ -0,0 +1,108 @@
#!/usr/bin/env python3
"""Create or list repositories on the Gitea instance at git.dieterunterstrich.de.
Credentials are read from ~/.git-credentials (the line for that host); the
secret is never printed. Useful when you need a new repo without touching the
web UI.
Usage:
python3 tools/new_repo.py list
python3 tools/new_repo.py create <name> [--private] [--desc "..."]
"""
from __future__ import annotations
import argparse
import base64
import json
import sys
from pathlib import Path
from urllib.request import Request, urlopen
HOST = "git.dieterunterstrich.de"
API = f"https://{HOST}/api/v1"
CREDENTIALS_FILE = Path.home() / ".git-credentials"
def load_credentials() -> tuple[str, str]:
"""Return (username, password) for the Gitea host from ~/.git-credentials."""
for line in CREDENTIALS_FILE.read_text().splitlines():
line = line.strip()
if not line.startswith("https://") or HOST not in line:
continue
# https://user:pass@host
rest = line.removeprefix("https://")
userinfo = rest.split("@", 1)[0]
user, _, password = userinfo.partition(":")
return user, password
raise SystemExit(f"No credentials for {HOST} in {CREDENTIALS_FILE}")
def api(method: str, path: str, body: dict | None = None) -> tuple[int, dict]:
"""Call the Gitea API; returns (status_code, parsed_json)."""
user, password = load_credentials()
auth = base64.b64encode(f"{user}:{password}".encode()).decode()
data = json.dumps(body).encode() if body is not None else None
req = Request(
f"{API}{path}",
data=data,
method=method,
headers={
"Authorization": f"Basic {auth}",
"Content-Type": "application/json",
},
)
try:
resp = urlopen(req, timeout=30)
return resp.status, json.loads(resp.read() or b"{}")
except Exception as exc: # noqa: BLE001
# HTTPError carries the response body, useful for API errors
code = getattr(exc, "code", None)
if code is not None:
try:
return code, json.loads(exc.read() or b"{}")
except Exception: # noqa: BLE001
return code, {"message": str(exc)}
raise
def main() -> None:
parser = argparse.ArgumentParser(description="Gitea repo helper")
sub = parser.add_subparsers(dest="command", required=True)
sub.add_parser("list")
p_create = sub.add_parser("create")
p_create.add_argument("name")
p_create.add_argument("--desc", default=None)
p_create.add_argument("--public", action="store_true")
args = parser.parse_args()
if args.command == "list":
status, repos = api("GET", "/user/repos?limit=100")
print(f"HTTP {status}")
for repo in repos:
print(f" {repo['full_name']} (private={repo['private']})")
return
status, result = api(
"POST",
"/user/repos",
{
"name": args.name,
"description": args.desc or "",
"private": not args.public,
"auto_init": False,
"default_branch": "main",
},
)
if status in (201, 200):
print(f"Created: {result['full_name']} -> {result['clone_url']}")
else:
print(f"Failed (HTTP {status}): {result.get('message', result)}")
sys.exit(1)
if __name__ == "__main__":
main()