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

Add subscription mode, global cross-run URL store and --assume-done import

Browse Source 
- dlphelper/globalstore.py: ~/.local/share/dlphelper/urls.json records every
  successfully downloaded URL once, regardless of OUT_DIR - reposts in other
  channels and re-runs of moved channels no longer re-download
- --subs subscriptions.json: batch mode over a configurable channel list;
  re-exports each channel by default (new links get picked up), --no-fresh
  reuses cached exports
- --assume-done <file>: manual import step that marks old URL lists as done
  in the global store (deliberate, unverified, never automatic)
- lazy downloader startup: no gofile guest account / no curl runs when all
  URLs are already done; token prompt skipped in --urls-file mode
- write_failed_urls creates OUT_DIR if needed
This commit is contained in:
Ben_Kosytorz
2026-10-04 17:19:11 +02:00
parent db8a1c1b52
commit 72ef7bc465
4 changed files with 357 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
dlp.py
+213 -88
View File
@@ -7,8 +7,16 @@ 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.
Already-downloaded URLs are recorded in a global store (default
~/.local/share/dlphelper/urls.json), so a link that was downloaded for one
channel is never re-downloaded for another - and re-runs of the same channel
skip everything finished (see also --assume-done for importing old lists).
Usage:
python3 dlp.py <channel_id> <out_dir> [options]
python3 dlp.py --subs subscriptions.json [options]
python3 dlp.py <channel_id> <out_dir> --urls-file failed_urls.txt
python3 dlp.py --assume-done old_urls.txt # import past downloads
"""
from __future__ import annotations
@@ -27,6 +35,7 @@ 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.globalstore import GlobalStore, default_global_path
from dlphelper.state import State, append_manifest
log = logging.getLogger("dlp")
@@ -52,13 +61,18 @@ def resolve_token(cli_value: str | None) -> str:
return token
def read_url_lines(path: Path) -> list[str]:
return [
line.strip()
for line in path.read_text(encoding="utf-8").splitlines()
if line.strip() and not line.strip().startswith("#")
]
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
for url in read_url_lines(path):
host = dlp_discord.classify(url)
if host is None:
log.warning("Unsupported host in %s: %s", path.name, url)
@@ -67,13 +81,22 @@ def load_urls_file(path: Path) -> list[dlp_discord.FoundUrl]:
return sorted(found.values(), key=lambda f: f.url)
def record_files(state: State, url: str, files: list[dict], source: dict) -> None:
def record_files(
state: State,
gstore: GlobalStore | None,
url: str,
files: list[dict],
source: dict,
) -> None:
state.mark(url, status="ok", error=None, files=files)
if gstore is not None:
gstore.mark(url, channel=source.get("channel"), host=source.get("host"),
file_count=len(files))
for f in files:
append_manifest(Path(source["manifest"]), {
"channel": source["channel"],
"url": url,
"host": source["host"],
"host": source.get("host"),
"message_id": source.get("message_id"),
"timestamp": source.get("timestamp"),
"author": source.get("author"),
@@ -105,22 +128,26 @@ def run_gofile(
args: argparse.Namespace,
state: State,
source_by_url: dict[str, dict],
is_done,
gstore: GlobalStore | None,
) -> None:
if not [f for f in urls if not is_done(f.url)]:
return # nothing pending: no guest account creation, no API calls
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):
if is_done(f.url):
continue
state.mark(f.url, status="failed", error=str(exc),
files=[])
state.mark(f.url, status="failed", error=str(exc), files=[])
return
consecutive_failures = 0
for f in urls:
if state.is_done(f.url):
if is_done(f.url):
log.info("Skip (already downloaded): %s", f.url)
continue
source = source_by_url[f.url]
@@ -138,7 +165,7 @@ def run_gofile(
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)
record_files(state, gstore, f.url, files, source)
if args.extract:
extract_zips(files)
consecutive_failures = 0
@@ -192,11 +219,16 @@ def run_0807(
args: argparse.Namespace,
state: State,
source_by_url: dict[str, dict],
is_done,
gstore: GlobalStore | None,
) -> None:
if not [f for f in urls if not is_done(f.url)]:
return
out_dir = Path(args.out_dir) / "downloads" / "0807"
consecutive_failures = 0
for f in urls:
if state.is_done(f.url):
if is_done(f.url):
log.info("Skip (already downloaded): %s", f.url)
continue
source = source_by_url[f.url]
@@ -207,7 +239,7 @@ def run_0807(
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)
record_files(state, gstore, f.url, [entry], source)
if args.extract:
extract_zips([entry])
consecutive_failures = 0
@@ -251,17 +283,141 @@ def format_size(n: int) -> str:
def write_failed_urls(out_dir: Path, state: State) -> Path:
dest = out_dir / "failed_urls.txt"
dest.parent.mkdir(parents=True, exist_ok=True)
failures = state.failures()
dest.write_text("\n".join(failures) + ("\n" if failures else ""), encoding="utf-8")
return dest
def mark_assume_done(path: Path, gstore: GlobalStore | None) -> int:
"""Record URLs as already downloaded (no files) so they are never fetched.
Manual import step for the pre-dlp.py era: point it at an old URL list you
know is fully downloaded. There is no verification - garbage in, skipped
forever.
"""
if gstore is None:
raise SystemExit("--assume-done needs the global store (drop --no-global)")
count = 0
for url in read_url_lines(path):
gstore.mark(url, channel=None, host=dlp_discord.classify(url), file_count=0,
note="assumed done via --assume-done")
count += 1
log.info("Marked %d URLs as done in the global store (%s)", count, gstore.path)
return count
def process_channel(
channel_id: str,
out_dir: Path,
args: argparse.Namespace,
token: str | None,
gstore: GlobalStore | None,
force_export: bool = False,
) -> int:
"""Run the full pipeline for one channel; returns the number of failures."""
state = State(out_dir / "state.json")
manifest = out_dir / "manifest.jsonl"
def is_done(url: str) -> bool:
return state.is_done(url) or (gstore is not None and gstore.is_done(url))
if args.urls_file:
found = load_urls_file(args.urls_file)
else:
if token is None:
raise SystemExit("internal error: no token resolved")
export_path = dlp_discord.export_channel(
channel_id, out_dir, token, force=force_export or 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
log.warning("Skipping unsupported URL: %s", f.url)
state.mark(f.url, status="failed", error="unsupported host")
source_by_url: dict[str, dict] = {
f.url: {
"channel": channel_id,
"host": f.host,
"message_id": f.message_id,
"timestamp": f.timestamp,
"author": f.author,
"manifest": str(manifest),
}
for f in found
}
pending = [f for f in found if not is_done(f.url)]
log.info(
"%s: %d URLs (%d gofile, %d 0807); %d done, %d to download",
channel_id, len(found),
len(buckets["gofile"]), len(buckets["0807"]),
len(found) - len(pending), len(pending),
)
if args.dry_run:
for f in found:
flag = "done" if is_done(f.url) else "download"
print(f"[{f.host:6}] [{flag:8}] {f.url}")
return 0
if buckets["gofile"]:
run_gofile(buckets["gofile"], args, state, source_by_url, is_done, gstore)
if buckets["0807"]:
run_0807(buckets["0807"], args, state, source_by_url, is_done, gstore)
failed_file = write_failed_urls(out_dir, state)
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("%s: %d ok (%d files, %s) / %d failed / %d seen",
out_dir, 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)
return len(bad)
def load_subscriptions(path: Path) -> list[dict]:
"""Load subscriptions.json: [{"name": ..., "channel_id": ..., "out_dir": ...}]."""
data = json.loads(path.read_text(encoding="utf-8"))
entries = data.get("subscriptions", []) if isinstance(data, dict) else data
if not isinstance(entries, list):
raise SystemExit(f"{path}: expected a list or {'{'}\"subscriptions\": [...]{'}'}")
subs = []
for entry in entries:
if not isinstance(entry, dict) or "channel_id" not in entry or "out_dir" not in entry:
raise SystemExit(
f"{path}: each subscription needs 'channel_id' and 'out_dir' "
f"(got: {entry})"
)
subs.append(entry)
return subs
def main() -> None:
parser = argparse.ArgumentParser(
description="Collect downloads from a Discord channel (gofile + 0807)."
description="Collect downloads from Discord channels (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("channel_id", nargs="?", help="Discord channel or thread id")
parser.add_argument("out_dir", nargs="?", help="directory for downloads and state")
parser.add_argument("--subs", type=Path, metavar="FILE",
help="process a subscription list instead of one channel "
"(entries: channel_id, out_dir, optional name)")
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 "
@@ -281,9 +437,19 @@ def main() -> None:
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")
help="re-export the Discord channel even if export.json exists "
"(subs mode does this by default)")
parser.add_argument("--no-fresh", action="store_true",
help="subs mode: reuse cached exports instead of re-exporting")
parser.add_argument("--dry-run", action="store_true",
help="export/extract/classify only, download nothing")
parser.add_argument("--global-state", type=Path, metavar="FILE",
help=f"cross-run URL store (default: {default_global_path()})")
parser.add_argument("--no-global", action="store_true",
help="disable the cross-run URL store for this run")
parser.add_argument("--assume-done", type=Path, metavar="FILE",
help="mark every URL in the file as already downloaded, "
"then exit (manual import of old download lists)")
args = parser.parse_args()
logging.basicConfig(
@@ -292,82 +458,41 @@ def main() -> None:
datefmt="%H:%M:%S",
)
out_dir = Path(args.out_dir)
state = State(out_dir / "state.json")
gstore = None
if not args.no_global:
gstore = GlobalStore(args.global_state or default_global_path())
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}")
if args.assume_done:
if args.channel_id or args.subs:
parser.error("--assume-done cannot be combined with a channel or --subs")
mark_assume_done(args.assume_done, gstore)
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)
if args.subs:
if args.channel_id or args.out_dir or args.urls_file:
parser.error("--subs cannot be combined with channel_id/out_dir/--urls-file")
subs = load_subscriptions(args.subs)
token = resolve_token(args.token)
total_failed = 0
for entry in subs:
out_dir = Path(entry["out_dir"])
log.info("=== subscription: %s ===", entry.get("name") or entry["channel_id"])
total_failed += process_channel(
entry["channel_id"], out_dir, args, token, gstore,
force_export=not args.no_fresh, # subs mode re-exports by default
)
if total_failed:
log.warning("%d URL(s) failed in total across subscriptions.", total_failed)
sys.exit(1)
return
failed_file = write_failed_urls(out_dir, state)
if not args.channel_id or not args.out_dir:
parser.error("channel_id and out_dir are required (or use --subs/--assume-done)")
# 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)
# A token is only needed when we will export a channel (--urls-file skips that)
token = None if args.urls_file else resolve_token(args.token)
failures = process_channel(args.channel_id, Path(args.out_dir), args, token, gstore)
if failures:
sys.exit(1)