From 55d32379edab29e6bc44314508a6aed389311fd5 Mon Sep 17 00:00:00 2001 From: Ben_Kosytorz Date: Fri, 9 Oct 2026 15:30:27 +0200 Subject: [PATCH] feat(rust): port dlp.py orchestration + unified CLI (clap subcommands) - orchestrate.rs: process_channel, run_gofile (re-auth on AuthError, no-retry branches for dead/password/empty, cooldowns), run_0807, subscriptions, --assume-done (with note), failed_urls export - main.rs: dlp / catalogue / sorter / extract / gwshop subcommands, token resolution (--token -> $DISCORD_TOKEN -> hidden prompt) - globalstore: mark() gains a note field (python note= kwarg) - README: port status, CLI, deviations, untested live paths 77 tests green. --- version-rust/README.md | 109 +++- version-rust/src/globalstore.rs | 14 +- version-rust/src/main.rs | 617 ++++++++++++++++++- version-rust/src/orchestrate.rs | 1022 +++++++++++++++++++++++++++++++ version-rust/src/sorter.rs | 4 +- 5 files changed, 1747 insertions(+), 19 deletions(-) diff --git a/version-rust/README.md b/version-rust/README.md index 0416d49..5a68d50 100644 --- a/version-rust/README.md +++ b/version-rust/README.md @@ -13,26 +13,111 @@ the python tooling. (secrets, failure policy, idempotency, rate limiting) apply to the rewrite too and are the contract the Rust version must keep. -## Planned crate choices (initial, adjust as modules get ported) +## Crate choices (as ported) -| Concern | Python today | Rust plan | +| Concern | Python today | Rust | |---|---|---| | 7z extraction | 7z CLI, cp437 name repair, /var/tmp fallback | `sevenz-rust2` (pure Rust; LZMA/LZMA2/PPMD/AES-256 decrypt, v0.23, active) | | zip extraction | 7z CLI | `zip` crate | | RAR extraction (critical) | 7z CLI | `unrar.rs` (muja/unrar.rs — bindings to the official UnRAR library, RAR5 support, extraction only). Note: UnRAR license is freeware-not-OSI and forbids building a RAR compressor — acceptable, we only extract, and compression later is 7z-only (`sevenz-rust2` `ArchiveWriter`) anyway | +| tar / gz / bz2 / xz | 7z CLI | **deviation:** system binaries (`tar`, `gzip`, `bzip2`, `xz`) — no mature pure-Rust tar+filter combo that matches the 7z-pass behavior; contract (dest layout, quarantine, manifest) is identical | | HTTP client | `requests` | `reqwest` (blocking first, async where the webapp needs it) | | sha256 (gofile website token, --hash manifest) | hashlib | `sha2` | -| Fuzzy matching (sorter) | difflib-style ratios | `strsim` — port thresholds exactly, add unit tests | -| CLI | argparse | `clap` | +| Fuzzy matching (sorter) | difflib-style ratios | exact port of CPython `difflib.SequenceMatcher` in `src/difflib.rs` (NOT strsim — ratios must be bit-identical for the thresholds) | +| CLI | argparse | `clap` (subcommands below) | | Webapp | — | `axum` (CLI parity first, webapp second) | | Discord export | DCE docker container (external) | unchanged: shell out to docker | -## Port order (suggested) +## Port status -1. `state` / `globalstore` / manifest — **done**, see `src/state.rs`, - `src/globalstore.rs`, `src/manifest.rs` (schemas backward compatible). -2. hosters: `gofile` (one guest account per run!), `zero807` (UA + Referer). -3. extraction + manifest appender. -4. catalogue / sorter (hardest logic; port the matching rules documented in - `version-python/AGENTS.md` §7 with tests). -5. webapp on top. +All python entry points are ported; live network paths are tested offline +only (see "Untested live paths"). + +| Rust module | Python source | Tests | +|---|---|---| +| `src/state.rs` | state.py | ✓ | +| `src/globalstore.rs` | globalstore.py | ✓ | +| `src/manifest.rs` | state.py append_manifest | ✓ | +| `src/difflib.rs` | CPython difflib | bit-identical on generated vectors | +| `src/warhammer_data.rs` + `data/warhammer_data.json` | warhammer_data.py | ✓ (order-sensitive, IndexMap) | +| `src/discord.rs` | dlphelper/discord.py | ✓ (DCE docker export, URL extraction) | +| `src/gofile.rs` | dlphelper/gofile.py | ✓ (website-token goldens) | +| `src/zero807.rs` | dlphelper/zero807.py | ✓ (curl subprocess contract) | +| `src/extract.rs` | tools/extract_all.py | ✓ (7z/zip/rar/tar+bz2/xz, nested, sweep, `.broken/`) | +| `src/catalogue.rs` | dlphelper/catalogue.py | ✓ (all python test cases) | +| `src/sorter.rs` | dlphelper/sorter.py | ✓ (all python test_sorter_units cases) | +| `src/gwshop.rs` | dlphelper/gwshop.py | ✓ (Algolia client, facet partition) | +| `src/orchestrate.rs` | dlp.py pipeline | ✓ (record/failed-urls/subs/assume-done; retry loops offline) | +| `src/main.rs` | all CLIs | smoke-tested (see below) | + +## CLI + +``` +dlphelper dlp [--subs FILE] [--urls-file FILE] [--token] + [--password] [--retries N] [--retry-delay N] [--cooldown-failures N] + [--cooldown-seconds N] [--only gofile|0807] [--limit N] [--shuffle] + [--hash] [--extract] [--fresh|--no-fresh] [--dry-run] + [--global-state FILE] [--no-global] [--assume-done FILE] + +dlphelper catalogue [run_dir] [--scan DIR]... [--manifest FILE] + [--overrides FILE] [--out-md FILE] [--out-jsonl FILE] [--apply --library DIR] + +dlphelper sorter --library DIR [--catalog FILE] [--threshold 0.62] + [--apply] [--audit FILE] [--provenance FILE] [--json] + [--drop-system NAME]... [--keep-system NAME]... + +dlphelper extract ... --out DIR + +dlphelper gwshop [--index NAME] [--raw FILE] +``` + +Notes: + +- The Discord token goes `--token` → `$DISCORD_TOKEN` → hidden prompt, and is + only ever passed to the DCE container via env, never argv. +- `sorter --catalog` defaults to `version-python/data/gwshop_catalog.json` + (falling back to `data/gwshop_catalog.json`), both relative to the current + working directory — pass `--catalog` when running elsewhere. +- `sorter --audit` defaults to `/../_logs/sorter_applied.jsonl` + (bookkeeping stays out of the library, which may be a seeded torrent). +- Retry loops, cooldowns and the "one gofile guest account per run" rule are + ported verbatim from `dlp.py`; a failure in one URL never aborts the run. +- Ctrl-C: state.json is saved after every URL; re-run to resume (the python + "interrupted" record on the in-flight URL is not replicated). + +## Deviations from the python version (known, accepted) + +- `tar`/`gz`/`bz2`/`xz` extraction shells out to the system binaries + (7z CLI in python); 7z/zip/rar use native Rust. +- `fs::copy` fallback in the sorter does not preserve mtime + (python `shutil.copy2` does). +- state.json omits `"files": []` for records without files + (python writes the empty list); readers of both versions accept both. +- `sorter --json` prints keys alphabetically (serde_json, no preserve_order); + python printed them in dataclass order. Content is identical. +- Negative `--retries`/`--limit` values are rejected by clap (python accepted + them with odd semantics). + +## Untested live paths (validated offline only, per the failure-policy rules) + +- Actual Discord channel export via the DCE container (needs a token) — the + docker invocation and export.json parsing are unit-tested, the network path + is not. +- Live gofile downloads (guest account creation, `ensure_token`) — website + token derivation is golden-tested against python; the HTTP calls are not. +- Live 0807 downloads — the `curl -A "Mozilla/5.0" -e "https://0807.st/"` + contract is kept, timeouts are per python (`--max-time 1800`), but no live + fetch was made. +- Real `.rar` extraction — the extractor wiring is in place (`unrar` crate, + unrar runtime library present), but no real RAR archive was available for a + live test. +- `gwshop` was smoke-tested for request shaping only; a full Algolia browse + needs valid credentials (`algolia.json` or env). + +## Build/test on NFS + +The repo lives on NFS; rustc SIGBUSes there. Use: + +``` +CARGO_TARGET_DIR=/var/tmp/dlphelper-target cargo test +``` diff --git a/version-rust/src/globalstore.rs b/version-rust/src/globalstore.rs index 0fe06be..b2f3b05 100644 --- a/version-rust/src/globalstore.rs +++ b/version-rust/src/globalstore.rs @@ -101,13 +101,15 @@ impl GlobalStore { } /// Record a URL as done (plus attribution fields when known), then save. - /// No-ops when the store is disabled. + /// No-ops when the store is disabled. `note` mirrors the python `note=` + /// kwarg (e.g. "assumed done via --assume-done") and lands in `extra`. pub fn mark( &mut self, url: &str, channel: Option<&str>, host: Option<&str>, file_count: Option, + note: Option<&str>, ) -> anyhow::Result<()> { if self.path.is_none() { return Ok(()); @@ -123,6 +125,9 @@ impl GlobalStore { if let Some(n) = file_count { rec.extra.insert("file_count".into(), Value::from(n)); } + if let Some(note) = note { + rec.extra.insert("note".into(), Value::from(note)); + } rec.updated = Some(now_local_iso()); self.save() } @@ -143,7 +148,8 @@ mod tests { #[test] fn disabled_store_is_a_noop() { let mut gs = GlobalStore::disabled(); - gs.mark("u", Some("c"), Some("gofile"), Some(2)).unwrap(); + gs.mark("u", Some("c"), Some("gofile"), Some(2), None) + .unwrap(); assert!(gs.records.is_empty()); } @@ -153,9 +159,9 @@ mod tests { let path = dir.path().join("urls.json"); { let mut gs = GlobalStore::load(&path); - gs.mark("u1", Some("chan"), Some("gofile"), Some(3)) + gs.mark("u1", Some("chan"), Some("gofile"), Some(3), None) .unwrap(); - gs.mark("u2", None, None, None).unwrap(); + gs.mark("u2", None, None, None, None).unwrap(); } let gs = GlobalStore::load(&path); assert!(gs.is_done("u1")); diff --git a/version-rust/src/main.rs b/version-rust/src/main.rs index d6d050d..6bc201a 100644 --- a/version-rust/src/main.rs +++ b/version-rust/src/main.rs @@ -1,3 +1,21 @@ +//! dlphelper CLI — Rust rewrite of the python entry points: +//! +//! - `dlphelper dlp ...` <- version-python/dlp.py +//! - `dlphelper catalogue ...` <- version-python/catalogue.py +//! - `dlphelper sorter ...` <- version-python/sorter.py +//! - `dlphelper extract ...` <- version-python/tools/extract_all.py +//! - `dlphelper gwshop ...` <- version-python/dlphelper/gwshop.py +//! +//! Secrets rule: the Discord token is resolved via --token / $DISCORD_TOKEN / +//! hidden prompt and only ever handed to the DCE container via env. + +use std::io::IsTerminal; +use std::path::{Path, PathBuf}; +use std::process::exit; + +use anyhow::Context as _; +use clap::{Parser, Subcommand}; + mod catalogue; mod difflib; mod discord; @@ -13,7 +31,602 @@ mod util; mod warhammer_data; mod zero807; +#[derive(Parser)] +#[command( + name = "dlphelper", + version, + about = "Collect, catalogue and sort Warhammer downloads (Rust rewrite)." +)] +struct Cli { + #[command(subcommand)] + cmd: Cmd, +} + +#[derive(Subcommand)] +enum Cmd { + /// Collect downloads from Discord channels (gofile + 0807). + Dlp(DlpArgs), + /// Classify the collection into system / faction / class. + Catalogue(CatalogueArgs), + /// Fuzzysort collection files into the library (plan/apply). + Sorter(SorterArgs), + /// Extract archives into per-source directories. + Extract(ExtractArgs), + /// Dump the GW webshop catalog via Algolia. + Gwshop(GwshopArgs), +} + +#[derive(clap::Args)] +struct DlpArgs { + /// Discord channel or thread id + channel_id: Option, + /// directory for downloads and state + out_dir: Option, + /// process a subscription list instead of one channel + /// (entries: channel_id, out_dir, optional name) + #[arg(long, value_name = "FILE")] + subs: Option, + /// Discord token (else $DISCORD_TOKEN or prompt) + #[arg(long)] + token: Option, + /// skip the Discord export; read URLs from this file (e.g. failed_urls.txt) + #[arg(long, value_name = "FILE")] + urls_file: Option, + /// password for password-protected gofile links + #[arg(long)] + password: Option, + /// attempts per URL (default 3) + #[arg(long, default_value_t = 3)] + retries: u32, + /// base seconds between retries, multiplied per attempt (default 5) + #[arg(long, default_value_t = 5)] + retry_delay: u64, + /// consecutive failures before a cooldown pause (default 3) + #[arg(long, default_value_t = 3)] + cooldown_failures: u32, + /// length of the cooldown pause (default 60) + #[arg(long, default_value_t = 60)] + cooldown_seconds: u64, + /// process only one host class + #[arg(long)] + only: Option, + /// download at most N pending URLs per host class (test runs) + #[arg(long, value_name = "N")] + limit: Option, + /// randomize which URLs a --limit picks + #[arg(long)] + shuffle: bool, + /// record sha256 of each downloaded file + #[arg(long)] + hash: bool, + /// unzip downloaded archives into per-archive folders + #[arg(long)] + extract: bool, + /// re-export the Discord channel even if export.json exists + /// (subs mode does this by default) + #[arg(long)] + fresh: bool, + /// subs mode: reuse cached exports instead of re-exporting + #[arg(long)] + no_fresh: bool, + /// export/extract/classify only, download nothing + #[arg(long)] + dry_run: bool, + /// cross-run URL store FILE (default: $XDG_DATA_HOME/dlphelper/urls.json + /// or ~/.local/share/dlphelper/urls.json) + #[arg(long, value_name = "FILE")] + global_state: Option, + /// disable the cross-run URL store for this run + #[arg(long)] + no_global: bool, + /// mark every URL in the file as already downloaded, then exit + /// (manual import of old download lists) + #[arg(long, value_name = "FILE")] + assume_done: Option, +} + +#[derive(clap::Args)] +struct CatalogueArgs { + /// a dlp.py run dir (uses its manifest.jsonl and downloads/) + run_dir: Option, + /// additional directory to scan for archive files (repeatable) + #[arg(long, value_name = "DIR")] + scan: Vec, + /// explicit manifest.jsonl (default: /manifest.jsonl) + #[arg(long, value_name = "FILE")] + manifest: Option, + /// manual classification overrides (JSON) + #[arg(long, value_name = "FILE")] + overrides: Option, + #[arg(long, value_name = "FILE")] + out_md: Option, + #[arg(long, value_name = "FILE")] + out_jsonl: Option, + /// move files into the library structure (default: preview) + #[arg(long)] + apply: bool, + /// library root for --apply (required with --apply) + #[arg(long, value_name = "DIR")] + library: Option, +} + +#[derive(clap::Args)] +struct SorterArgs { + /// a directory to sort or a single file + source: PathBuf, + /// library root the files will be linked into + #[arg(long, value_name = "DIR")] + library: PathBuf, + /// catalog FILE (default: version-python/data/gwshop_catalog.json, + /// falling back to data/gwshop_catalog.json, both relative to the cwd) + #[arg(long, value_name = "FILE")] + catalog: Option, + /// fuzzy match threshold (default 0.62) + #[arg(long, default_value_t = sorter::DEFAULT_THRESHOLD)] + threshold: f64, + /// create the targets (hardlink, fallback copy) + #[arg(long)] + apply: bool, + /// audit FILE (default: /../_logs/sorter_applied.jsonl) + #[arg(long, value_name = "FILE")] + audit: Option, + /// extract_manifest.jsonl from extract: sort whole extraction + /// directories as units + #[arg(long, value_name = "FILE")] + provenance: Option, + /// print the plan as JSON instead of text + #[arg(long)] + json: bool, + /// never place into this game system (default-dropped: Middle-Earth) + #[arg(long, value_name = "NAME")] + drop_system: Vec, + /// re-enable a default-dropped game system + #[arg(long, value_name = "NAME")] + keep_system: Vec, +} + +#[derive(clap::Args)] +struct ExtractArgs { + /// archives or directories containing archives + sources: Vec, + /// directory to extract into + #[arg(long, required = true, value_name = "DIR")] + out: PathBuf, +} + +#[derive(clap::Args)] +struct GwshopArgs { + /// output catalog JSON path + out: PathBuf, + /// Algolia index name + #[arg(long, default_value = gwshop::DEFAULT_INDEX)] + index: String, + /// also write the raw Algolia hits here + #[arg(long, value_name = "FILE")] + raw: Option, +} + +/// argparse parser.error equivalent: message to stderr, exit code 2. +fn parser_error(msg: &str) -> ! { + eprintln!("error: {msg}"); + exit(2); +} + +/// python resolve_token: --token, then $DISCORD_TOKEN, then hidden prompt. +fn resolve_token(cli_value: Option<&str>) -> anyhow::Result { + if let Some(t) = cli_value { + return Ok(t.to_string()); + } + if let Ok(t) = std::env::var("DISCORD_TOKEN") { + if !t.is_empty() { + return Ok(t); + } + } + log::info!("No Discord token given ($DISCORD_TOKEN); prompting (input hidden)."); + if !std::io::stdin().is_terminal() { + anyhow::bail!( + "No Discord token available and stdin is not a terminal; \ + use --token or $DISCORD_TOKEN." + ); + } + let token = rpassword::prompt_password("Discord token: ").context("reading the token")?; + if token.is_empty() { + anyhow::bail!("No token provided, aborting."); + } + Ok(token) +} + +/// The global store for this run, honoring --no-global / --global-state. +fn make_gstore(args: &DlpArgs) -> globalstore::GlobalStore { + if args.no_global { + globalstore::GlobalStore::disabled() + } else if let Some(p) = &args.global_state { + globalstore::GlobalStore::load(p) + } else { + match globalstore::default_global_path() { + Some(p) => globalstore::GlobalStore::load(p), + None => globalstore::GlobalStore::disabled(), // no home dir: like python gstore=None + } + } +} + +fn default_catalog_path() -> PathBuf { + let v = PathBuf::from("version-python/data/gwshop_catalog.json"); + if v.exists() { + return v; + } + let alt = PathBuf::from("data/gwshop_catalog.json"); + if alt.exists() { + return alt; + } + v // surfaces a clear error in load_catalog +} + +fn cmd_dlp(args: &DlpArgs) -> anyhow::Result<()> { + if let Some(assume) = &args.assume_done { + if args.channel_id.is_some() || args.subs.is_some() { + parser_error("--assume-done cannot be combined with a channel or --subs"); + } + let mut gstore = make_gstore(args); + orchestrate::mark_assume_done(assume, &mut gstore)?; + return Ok(()); + } + + if let Some(subs_path) = &args.subs { + if args.channel_id.is_some() || args.out_dir.is_some() || args.urls_file.is_some() { + parser_error("--subs cannot be combined with channel_id/out_dir/--urls-file"); + } + let subs = orchestrate::load_subscriptions(subs_path)?; + let token = resolve_token(args.token.as_deref())?; + let mut gstore = make_gstore(args); + let mut total_failed = 0u64; + for (channel_id, out_dir, name) in &subs { + log::info!( + "=== subscription: {} ===", + name.clone().unwrap_or_else(|| channel_id.clone()) + ); + let opts = orchestrate::RunOpts::from(args); + // subs mode re-exports by default (python force_export=not --no-fresh) + total_failed += orchestrate::process_channel( + channel_id, + Path::new(out_dir), + &opts, + Some(&token), + Some(&mut gstore), + None, + !args.no_fresh, + )?; + } + if total_failed > 0 { + log::warn!("{total_failed} URL(s) failed in total across subscriptions."); + exit(1); + } + return Ok(()); + } + + let (Some(channel_id), Some(out_dir)) = (&args.channel_id, &args.out_dir) else { + parser_error("channel_id and out_dir are required (or use --subs/--assume-done)"); + }; + // A token is only needed when we will export a channel (--urls-file skips that) + let token = if args.urls_file.is_some() { + None + } else { + Some(resolve_token(args.token.as_deref())?) + }; + let opts = orchestrate::RunOpts::from(args); + let mut gstore = make_gstore(args); + let failures = orchestrate::process_channel( + channel_id, + out_dir, + &opts, + token.as_deref(), + Some(&mut gstore), + args.urls_file.as_deref(), + false, + )?; + if failures > 0 { + exit(1); + } + Ok(()) +} + +impl From<&DlpArgs> for orchestrate::RunOpts { + fn from(a: &DlpArgs) -> Self { + orchestrate::RunOpts { + retries: a.retries, + retry_delay: a.retry_delay, + cooldown_failures: a.cooldown_failures, + cooldown_seconds: a.cooldown_seconds, + hash: a.hash, + extract: a.extract, + password: a.password.clone(), + limit: a.limit, + shuffle: a.shuffle, + only: a.only.clone(), + dry_run: a.dry_run, + fresh: a.fresh, + } + } +} + +fn cmd_catalogue(args: &CatalogueArgs) -> anyhow::Result<()> { + let run_dir = args.run_dir.as_deref(); + let manifests: Vec = if let Some(m) = &args.manifest { + vec![m.clone()] + } else { + run_dir + .map(|d| vec![d.join("manifest.jsonl")]) + .unwrap_or_default() + }; + let mut scan_dirs = args.scan.clone(); + if let Some(rd) = run_dir { + if args.manifest.is_none() { + // run dir is scanned as a whole (covers files not yet in the manifest) + scan_dirs.push(rd.to_path_buf()); + } + } + + if manifests.is_empty() && scan_dirs.is_empty() { + parser_error("give a run_dir or at least one --scan directory"); + } + + let out_base: PathBuf = run_dir + .map(Path::to_path_buf) + .unwrap_or_else(|| std::env::current_dir().unwrap_or_else(|_| PathBuf::from("."))); + let out_md = args + .out_md + .clone() + .unwrap_or_else(|| out_base.join("catalogue.md")); + let out_jsonl = args + .out_jsonl + .clone() + .unwrap_or_else(|| out_base.join("catalogue.jsonl")); + + let mut entries = + catalogue::build_catalogue(&manifests, &scan_dirs, args.overrides.as_deref())?; + catalogue::write_reports(&entries, &out_md, &out_jsonl)?; + + let classified = entries + .iter() + .filter(|e| e.faction != catalogue::UNCLASSIFIED_FACTION) + .count(); + log::info!( + "Catalogued {} files ({} classified) -> {}", + entries.len(), + classified, + out_md.display() + ); + + // compact per-faction summary to the console + let mut summary: std::collections::BTreeMap<(String, String), (u64, u64)> = + std::collections::BTreeMap::new(); + for e in &entries { + let entry = summary + .entry((e.system.clone(), e.faction.clone())) + .or_insert((0, 0)); + entry.0 += 1; + entry.1 += e.size; + } + for ((system, faction), (n, size)) in &summary { + log::info!( + " {:<18} {:<32} {:>4} files {}", + system, + faction, + n, + catalogue::format_size(*size) + ); + } + + if args.apply { + let Some(library) = &args.library else { + parser_error("--apply needs --library DIR"); + }; + let Some(run_dir) = run_dir else { + parser_error("--apply needs a run_dir (files are moved out of it)"); + }; + let moved = catalogue::apply_moves( + &mut entries, + library, + &library.join("catalogue_moves.jsonl"), + )?; + log::info!( + "Moved {} files into {} (audit: catalogue_moves.jsonl)", + moved, + library.display() + ); + // rewrite reports with updated paths + catalogue::write_reports(&entries, &out_md, &out_jsonl)?; + let _ = run_dir; + } + Ok(()) +} + +fn load_provenance(path: &Path) -> anyhow::Result { + use std::io::BufRead; + let mut prov = sorter::Provenance::new(); + let fh = std::fs::File::open(path).with_context(|| format!("reading {}", path.display()))?; + for line in std::io::BufReader::new(fh).lines() { + let rec: serde_json::Value = serde_json::from_str(&line?)?; + let dest = rec["dest"] + .as_str() + .context("provenance row lacks 'dest'")?; + let unit = rec["unit"] + .as_str() + .context("provenance row lacks 'unit'")?; + prov.insert(dest.to_string(), unit.to_string()); + } + Ok(prov) +} + +fn cmd_sorter(args: &SorterArgs) -> anyhow::Result<()> { + let dropped = sorter::dropped_systems_default(&args.keep_system, &args.drop_system); + if !args.keep_system.is_empty() || !args.drop_system.is_empty() { + log::info!( + "System filter adjusted: dropping {}", + if dropped.is_empty() { + "nothing".to_string() + } else { + let mut s = dropped.clone(); + s.sort(); + s.join(", ") + } + ); + } + + let source = &args.source; + let files: Vec = if source.is_file() { + vec![source.clone()] + } else { + // hidden trees like .broken/ must never reach the plan + let mut out = Vec::new(); + collect_visible(source, source, &mut out)?; + out.sort(); + out + }; + let catalog_path = args.catalog.clone().unwrap_or_else(default_catalog_path); + let catalog = sorter::load_catalog(&catalog_path, &dropped)?; + if !dropped.is_empty() { + let mut s = dropped.clone(); + s.sort(); + log::info!("Dropped systems: {}", s.join(", ")); + } + // Bookkeeping must never land inside the library itself (it may be a + // seeded torrent whose file set is frozen): default to a _logs folder + // beside it. An explicit --audit overrides this. + let audit = args.audit.clone().unwrap_or_else(|| { + args.library + .parent() + .unwrap_or(Path::new(".")) + .join("_logs") + .join("sorter_applied.jsonl") + }); + log::info!("Audit file: {}", audit.display()); + + let provenance = match &args.provenance { + Some(p) => { + let prov = load_provenance(p)?; + log::info!("Provenance: {} extraction directories", prov.len()); + Some(prov) + } + None => None, + }; + + let mut items = sorter::sort_plan( + &files, + &args.library, + &catalog, + args.threshold, + provenance.as_ref(), + Some(source), + ); + + if args.json { + let rows: Vec = items + .iter() + .map(|i| { + serde_json::json!({ + "source": i.path.display().to_string(), + "name": i.name, + "size": i.size, + "system": i.system, + "group": i.group, + "faction": i.faction, + "subfaction": i.subfaction, + "class": i.klass, + "unit": i.unit, + "score": (i.match_score * 1000.0).round() / 1000.0, + "kind": i.match_kind, + "inner_count": i.inner_names.len(), + "target": i.target + .as_ref() + .map(|t| t.display().to_string()) + .unwrap_or_else(|| "None".into()), + }) + }) + .collect(); + println!("{}", serde_json::to_string_pretty(&rows)?); + } else { + println!("{}", sorter::format_plan(&items)); + } + + let n_catalog = items.iter().filter(|i| i.match_kind == "catalog").count(); + let n_implicit = items.iter().filter(|i| i.match_kind == "implicit").count(); + let n_none = items.iter().filter(|i| i.match_kind == "none").count(); + log::info!( + "Plan: {} files -> {} catalog matches [+] / {} implicit [~] / {} unclassifiable [?]", + items.len(), + n_catalog, + n_implicit, + n_none + ); + + if args.apply { + let (linked, copied, skipped) = sorter::apply_plan(&mut items, &audit)?; + log::info!( + "Applied: {} hardlinked, {} copied, {} skipped (audit: {})", + linked, + copied, + skipped, + audit.display() + ); + } + Ok(()) +} + +/// Recursively collect files under `root`, skipping any path whose parts +/// (relative to root) start with '.' (python rglob filter). +fn collect_visible(root: &Path, dir: &Path, out: &mut Vec) -> std::io::Result<()> { + for entry in std::fs::read_dir(dir)? { + let entry = entry?; + let path = entry.path(); + let name = entry.file_name().to_string_lossy().into_owned(); + if name.starts_with('.') { + continue; + } + if path.is_dir() { + collect_visible(root, &path, out)?; + } else { + out.push(path); + } + } + Ok(()) +} + +fn cmd_extract(args: &ExtractArgs) -> anyhow::Result<()> { + let ex = extract::Extractor { + out_root: args.out.clone(), + }; + ex.run(&args.sources) +} + +fn cmd_gwshop(args: &GwshopArgs) -> anyhow::Result<()> { + gwshop::run_cli(&args.out, &args.index, args.raw.as_deref()) +} + fn main() { - env_logger::init(); - println!("dlphelper {}", env!("CARGO_PKG_VERSION")); + env_logger::Builder::from_env(env_logger::Env::default().default_filter_or("info")) + .format(|buf, record| { + use std::io::Write as _; + writeln!( + buf, + "{} {:<7} {}", + chrono::Local::now().format("%H:%M:%S"), + record.level(), + record.args() + ) + }) + .init(); + + let cli = Cli::parse(); + let res = match &cli.cmd { + Cmd::Dlp(args) => cmd_dlp(args), + Cmd::Catalogue(args) => cmd_catalogue(args), + Cmd::Sorter(args) => cmd_sorter(args), + Cmd::Extract(args) => cmd_extract(args), + Cmd::Gwshop(args) => cmd_gwshop(args), + }; + if let Err(err) = res { + log::error!("{err:#}"); + exit(1); + } } diff --git a/version-rust/src/orchestrate.rs b/version-rust/src/orchestrate.rs index 96a221b..13b8ca6 100644 --- a/version-rust/src/orchestrate.rs +++ b/version-rust/src/orchestrate.rs @@ -4,3 +4,1025 @@ //! Failure policy (the core requirement): a failure in one URL must never //! abort the run. Catch, record in state.json, continue. Cooldown pauses //! are the only permitted stall. + +use std::collections::HashMap; +use std::fs; +use std::path::{Path, PathBuf}; +use std::thread; +use std::time::Duration; + +use anyhow::{Context as _, anyhow}; +use serde_json::Value; +use sha2::{Digest, Sha256}; + +use crate::discord::{self, FoundUrl}; +use crate::globalstore::GlobalStore; +use crate::gofile::{GoFileClient, GoFileError}; +use crate::manifest::{ManifestRow, append_manifest}; +use crate::state::{FileEntry, State}; +use crate::zero807::{self, Zero807Error}; + +/// Runtime options for a download pass (the subset of CLI flags the +/// orchestrator needs). +#[derive(Debug, Clone)] +pub struct RunOpts { + pub retries: u32, + pub retry_delay: u64, + pub cooldown_failures: u32, + pub cooldown_seconds: u64, + pub hash: bool, + pub extract: bool, + pub password: Option, + pub limit: Option, + pub shuffle: bool, + pub only: Option, + pub dry_run: bool, + pub fresh: bool, +} + +impl Default for RunOpts { + fn default() -> Self { + RunOpts { + retries: 3, + retry_delay: 5, + cooldown_failures: 3, + cooldown_seconds: 60, + hash: false, + extract: false, + password: None, + limit: None, + shuffle: false, + only: None, + dry_run: false, + fresh: false, + } + } +} + +/// sha256 of a file, streamed (python sha256_of). +pub fn sha256_of(path: &Path) -> anyhow::Result { + let mut file = + fs::File::open(path).with_context(|| format!("opening {} for hashing", path.display()))?; + let mut hasher = Sha256::new(); + std::io::copy(&mut file, &mut hasher)?; + Ok(hex::encode(hasher.finalize())) +} + +/// Lines from a URL list file; skips blanks and # comments. +pub fn read_url_lines(path: &Path) -> anyhow::Result> { + let text = fs::read_to_string(path).with_context(|| format!("reading {}", path.display()))?; + Ok(text + .lines() + .map(str::trim) + .filter(|l| !l.is_empty() && !l.starts_with('#')) + .map(str::to_string) + .collect()) +} + +/// Load a plain URL list (e.g. failed_urls.txt); no attribution available. +pub fn load_urls_file(path: &Path) -> anyhow::Result> { + let mut found: HashMap = HashMap::new(); + for url in read_url_lines(path)? { + let Some(host) = discord::classify(&url) else { + log::warn!("Unsupported host in {}: {url}", path.display()); + continue; + }; + found.entry(url.clone()).or_insert(FoundUrl { + url, + host, + message_id: None, + timestamp: None, + author: None, + encoded: false, + }); + } + let mut list: Vec = found.into_values().collect(); + list.sort_by(|a, b| a.url.cmp(&b.url)); + Ok(list) +} + +/// Record a successful download: state.mark(ok) + global store + manifest +/// rows (python record_files). +pub fn record_files( + state: &mut State, + gstore: Option<&mut GlobalStore>, + url: &str, + files: &[Value], + source: &SourceInfo, + manifest_path: &Path, +) -> anyhow::Result<()> { + let entries: Vec = files + .iter() + .map(|f| FileEntry { + path: f["path"].as_str().unwrap_or_default().to_string(), + size: f["size"].as_u64(), + extra: { + let mut m = std::collections::BTreeMap::new(); + if let Some(n) = f.get("name") { + m.insert("name".to_string(), n.clone()); + } + if let Some(h) = f.get("sha256") { + m.insert("sha256".to_string(), h.clone()); + } + m + }, + }) + .collect(); + state.mark(url, |r| { + r.status = "ok".into(); + r.files = entries.clone(); + // python passes error=None which is *skipped* (never clears) + })?; + if let Some(gs) = gstore { + gs.mark( + url, + Some(&source.channel), + source.host.as_deref(), + Some(files.len() as u64), + None, + )?; + } + for f in files { + let row = ManifestRow { + channel: source.channel.clone(), + url: url.to_string(), + host: source.host.clone(), + message_id: source.message_id.clone(), + timestamp: source.timestamp.clone(), + author: source.author.clone(), + path: f["path"].as_str().map(str::to_string), + size: f["size"].as_u64(), + extra: { + let mut m = std::collections::BTreeMap::new(); + if let Some(n) = f.get("name") { + m.insert("name".to_string(), n.clone()); + } + if let Some(h) = f.get("sha256") { + m.insert("sha256".to_string(), h.clone()); + } + m + }, + }; + append_manifest(manifest_path, &row)?; + } + Ok(()) +} + +/// Message-level attribution for one URL (python source_by_url values). +#[derive(Debug, Clone, Default)] +pub struct SourceInfo { + pub channel: String, + pub host: Option, + pub message_id: Option, + pub timestamp: Option, + pub author: Option, + pub manifest: PathBuf, +} + +fn source_of(channel_id: &str, f: &FoundUrl, manifest: &Path) -> SourceInfo { + SourceInfo { + channel: channel_id.to_string(), + host: Some(f.host.to_string()), + message_id: f.message_id.clone().map(Value::String), + timestamp: f.timestamp.clone(), + author: f.author.clone(), + manifest: manifest.to_path_buf(), + } +} + +/// Optional: unzip each downloaded archive next to itself (python +/// extract_zips in dlp.py). +pub fn extract_zips(files: &[Value]) { + for f in files { + let Some(path_str) = f["path"].as_str() else { + continue; + }; + let path = PathBuf::from(path_str); + if path.extension().map(|e| e.to_ascii_lowercase()) != Some("zip".into()) || !path.exists() + { + continue; + } + let target = path.with_extension(""); + if target.exists() + && fs::read_dir(&target) + .map(|d| d.count() > 0) + .unwrap_or(false) + { + log::info!("Skip extracting {} (target exists)", path.display()); + continue; + } + log::info!("Extracting {} -> {}", path.display(), target.display()); + if let Err(exc) = fs::create_dir_all(&target).and_then(|_| unzip_to(&path, &target)) { + log::warn!("Extraction of {} failed: {exc}", path.display()); + } + } +} + +fn unzip_to(zip_path: &Path, target: &Path) -> std::io::Result<()> { + let file = fs::File::open(zip_path)?; + let mut zf = zip::ZipArchive::new(file)?; + for i in 0..zf.len() { + let mut entry = zf.by_index(i)?; + if entry.is_dir() { + fs::create_dir_all(target.join(entry.name()))?; + continue; + } + let out_path = target.join(entry.name()); + if let Some(parent) = out_path.parent() { + fs::create_dir_all(parent)?; + } + let mut out = fs::File::create(&out_path)?; + std::io::copy(&mut entry, &mut out)?; + } + Ok(()) +} + +pub fn format_size(n: u64) -> String { + let mut size = n as f64; + for unit in ["B", "KiB", "MiB", "GiB", "TiB"] { + if size < 1024.0 || unit == "TiB" { + return format!("{size:.1} {unit}"); + } + size /= 1024.0; + } + format!("{n} B") +} + +pub fn write_failed_urls(out_dir: &Path, state: &State) -> anyhow::Result { + let dest = out_dir.join("failed_urls.txt"); + if let Some(parent) = dest.parent() { + fs::create_dir_all(parent)?; + } + let failures = state.failures(); + let text = if failures.is_empty() { + String::new() + } else { + format!("{}\n", failures.join("\n")) + }; + fs::write(&dest, text)?; + Ok(dest) +} + +/// 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. +pub fn mark_assume_done(path: &Path, gstore: &mut GlobalStore) -> anyhow::Result { + if gstore.path.is_none() { + anyhow::bail!("--assume-done needs the global store (drop --no-global)"); + } + let mut count = 0; + for url in read_url_lines(path)? { + gstore.mark( + &url, + None, + discord::classify(&url), + Some(0), + Some("assumed done via --assume-done"), + )?; + count += 1; + } + log::info!( + "Marked {count} URLs as done in the global store ({})", + gstore + .path + .as_ref() + .map(|p| p.display().to_string()) + .unwrap_or_default() + ); + Ok(count) +} + +/// state.is_done(url) or the global store says done (python is_done closure). +fn is_done(state: &State, gstore: Option<&GlobalStore>, url: &str) -> bool { + state.is_done(url) || gstore.is_some_and(|g| g.is_done(url)) +} + +/// One channel's pipeline; returns the number of failures (python +/// process_channel). +pub fn process_channel( + channel_id: &str, + out_dir: &Path, + opts: &RunOpts, + token: Option<&str>, + mut gstore: Option<&mut GlobalStore>, + urls_file: Option<&Path>, + force_export: bool, +) -> anyhow::Result { + let mut state = State::load(out_dir.join("state.json")); + let manifest = out_dir.join("manifest.jsonl"); + + let found = if let Some(urls_file) = urls_file { + load_urls_file(urls_file)? + } else { + let token = token.ok_or_else(|| anyhow!("internal error: no token resolved"))?; + let export_path = + discord::export_channel(channel_id, out_dir, token, force_export || opts.fresh)?; + discord::extract_urls(&export_path)? + }; + + let found: Vec = found + .into_iter() + .filter(|f| opts.only.as_ref().is_none_or(|o| o.as_str() == f.host)) + .collect(); + + let mut gofile_urls: Vec = Vec::new(); + let mut z0807_urls: Vec = Vec::new(); + for f in &found { + match f.host { + "gofile" => gofile_urls.push(f.clone()), + "0807" => z0807_urls.push(f.clone()), + // classify() already filtered, but be defensive + _ => { + log::warn!("Skipping unsupported URL: {}", f.url); + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some("unsupported host".into()); + })?; + } + } + } + + let mut source_by_url: HashMap = HashMap::new(); + for f in &gofile_urls { + source_by_url.insert(f.url.clone(), source_of(channel_id, f, &manifest)); + } + for f in &z0807_urls { + source_by_url.insert(f.url.clone(), source_of(channel_id, f, &manifest)); + } + + // --limit applies per host class to the *pending* URLs, so a test run + // downloads exactly N new things per host (done ones don't use up the + // quota). Python replaces the bucket with pending[:limit]. + if let Some(limit) = opts.limit { + for list in [&mut gofile_urls, &mut z0807_urls] { + let mut pending: Vec = list + .iter() + .filter(|f| !is_done(&state, gstore.as_deref(), &f.url)) + .cloned() + .collect(); + if opts.shuffle { + fisher_yates(&mut pending); + } + pending.truncate(limit); + *list = pending; + } + } + + let total = gofile_urls.len() + z0807_urls.len(); + let pending_len = gofile_urls + .iter() + .chain(&z0807_urls) + .filter(|f| !is_done(&state, gstore.as_deref(), &f.url)) + .count(); + log::info!( + "{cid}: {n} URLs ({g} gofile, {z} 0807); {d} done, {p} to download", + cid = channel_id, + n = total, + g = gofile_urls.len(), + z = z0807_urls.len(), + d = total - pending_len, + p = pending_len, + ); + + if opts.dry_run { + let selected: std::collections::HashSet<&str> = gofile_urls + .iter() + .chain(&z0807_urls) + .map(|f| f.url.as_str()) + .collect(); + for f in &found { + let flag = if selected.contains(f.url.as_str()) { + "download" + } else if is_done(&state, gstore.as_deref(), &f.url) { + "done" + } else { + "skip" // cut by --limit / --only + }; + println!("[{:<6}] [{:<8}] {}", f.host, flag, f.url); + } + return Ok(0); + } + + if !gofile_urls.is_empty() { + run_gofile( + &gofile_urls, + opts, + out_dir, + &mut state, + &source_by_url, + gstore.as_deref_mut(), + &manifest, + )?; + } + if !z0807_urls.is_empty() { + run_0807( + &z0807_urls, + opts, + out_dir, + &mut state, + &source_by_url, + gstore.as_deref_mut(), + &manifest, + )?; + } + + let failed_file = write_failed_urls(out_dir, &state)?; + + let ok = state.records.values().filter(|r| r.status == "ok").count(); + let bad = state.failures(); + let dead = state + .records + .values() + .filter(|r| { + r.error + .as_deref() + .map(|e| e.contains("dead link")) + .unwrap_or(false) + }) + .count(); + let total_files: u64 = state.records.values().map(|r| r.files.len() as u64).sum(); + let total_size: u64 = state + .records + .values() + .flat_map(|r| r.files.iter()) + .map(|f| f.size.unwrap_or(0)) + .sum(); + log::info!( + "{}: {} ok ({} files, {}) / {} failed ({} dead links) / {} seen", + out_dir.display(), + ok, + total_files, + format_size(total_size), + bad.len(), + dead, + state.records.len(), + ); + if !bad.is_empty() { + log::info!( + "Failed URLs written to {} - re-run this tool or use \ + dlphelper dlp --urls-file {} to retry them.", + failed_file.display(), + failed_file + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default() + ); + } + Ok(bad.len() as u64) +} + +/// Fisher-Yates over pending URL pointers (python random.shuffle). +fn fisher_yates(items: &mut [T]) { + use rand::Rng; + let mut rng = rand::rng(); + for i in (1..items.len()).rev() { + let j = rng.random_range(..=i); + items.swap(i, j); + } +} + +/// GoFile pipeline (python run_gofile): one guest account for the whole +/// pass, per-URL retry loop with dead-link/password/empty no-retry branches, +/// re-auth on auth errors, cooldowns on consecutive failures - the run never +/// stops. +pub fn run_gofile( + urls: &[FoundUrl], + opts: &RunOpts, + out_dir: &Path, + state: &mut State, + source_by_url: &HashMap, + mut gstore: Option<&mut GlobalStore>, + manifest: &Path, +) -> anyhow::Result<()> { + if !urls + .iter() + .any(|f| !is_done(state, gstore.as_deref(), &f.url)) + { + return Ok(()); // nothing pending: no guest account creation, no API calls + } + let mut client = GoFileClient::new(opts.password.as_deref()); + if let Err(exc) = client.ensure_token() { + log::error!("{exc} - marking all pending gofile links as failed."); + for f in urls { + if is_done(state, gstore.as_deref(), &f.url) { + continue; + } + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some(exc.to_string()); + r.files = Vec::new(); + })?; + } + return Ok(()); + } + + let mut consecutive_failures: u32 = 0; + for f in urls { + if is_done(state, gstore.as_deref(), &f.url) { + log::info!("Skip (already downloaded): {}", f.url); + continue; + } + let source = source_by_url.get(&f.url).cloned().unwrap_or_default(); + let mut resolved = false; + for attempt in 1..=opts.retries { + // One attempt: list, download all files, hash. GoFileError maps + // onto the python exception handlers below; anything else is an + // unexpected bug and is recorded as such. The immediately-invoked + // closure ends the shared borrow of `client` before the handler + // arms below may re-authenticate. + let files = (|| -> anyhow::Result> { + let files_spec = client.list_files(&f.url).map_err(anyhow::Error::new)?; + if files_spec.is_empty() { + return Err(anyhow::Error::new(GoFileError::Empty( + "content exists but contains no files".into(), + ))); + } + let mut files: Vec = Vec::new(); + for (relpath, name, link) in files_spec { + let dest = out_dir + .join("downloads") + .join("gofile") + .join(&relpath) + .join(&name); + let size = client + .download_file(&link, &dest) + .map_err(anyhow::Error::new)?; + let mut entry = json_entry(&dest, size, &name); + if opts.hash { + let h = sha256_of(&dest)?; + entry["sha256"] = Value::String(h); + } + files.push(entry); + } + Ok(files) + })(); + match files { + Ok(files) => { + log::info!("OK {} ({} files)", f.url, files.len()); + record_files( + state, + gstore.as_deref_mut(), + &f.url, + &files, + &source, + manifest, + )?; + if opts.extract { + extract_zips(&files); + } + consecutive_failures = 0; + resolved = true; + break; + } + Err(err) => { + let Some(ge) = err.downcast_ref::() else { + // a bug must not stop the run + log::error!("Unexpected error for {}: {err:#}", f.url); + let msg = format!("unexpected: {err}"); + state.mark(&f.url, |r| { + r.error = Some(msg); + r.files = Vec::new(); + })?; + if attempt < opts.retries { + thread::sleep(Duration::from_secs(opts.retry_delay * attempt as u64)); + } + continue; + }; + match ge { + GoFileError::PasswordRequired(msg) => { + // Retrying without a password never helps + log::error!("PASSWORD REQUIRED {}: {msg}", f.url); + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some(msg.clone()); + r.files = Vec::new(); + })?; + resolved = true; + break; + } + GoFileError::NotFound(msg) => { + // Dead links never come back - no retry + log::error!("DEAD LINK {}: {msg}", f.url); + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some(msg.clone()); + r.files = Vec::new(); + })?; + resolved = true; + break; + } + GoFileError::Empty(msg) => { + // Emptied content is stable - no retry + log::error!("EMPTY CONTENT {}: {msg}", f.url); + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some(msg.clone()); + r.files = Vec::new(); + })?; + resolved = true; + break; + } + GoFileError::AuthError(msg) => { + // Token may have died mid-run (rate limit); re-auth + log::warn!("Auth error for {}: {msg} - re-authenticating", f.url); + state.mark(&f.url, |r| { + r.error = Some(msg.clone()); + r.files = Vec::new(); + })?; + client.reset(); + if let Err(reauth) = client.ensure_token() { + log::error!("Re-authentication failed: {reauth}"); + } + if attempt < opts.retries { + thread::sleep(Duration::from_secs( + opts.retry_delay * attempt as u64, + )); + } + } + GoFileError::Error(msg) => { + log::warn!( + "Attempt {}/{} failed for {}: {msg}", + attempt, + opts.retries, + f.url + ); + state.mark(&f.url, |r| { + r.error = Some(msg.clone()); + r.files = Vec::new(); + })?; + if attempt < opts.retries { + thread::sleep(Duration::from_secs( + opts.retry_delay * attempt as u64, + )); + } + } + } + } + } + } + if !resolved { + let err = state + .records + .get(&f.url) + .and_then(|r| r.error.clone()) + .unwrap_or_else(|| "unknown".into()); + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some(err); + r.files = Vec::new(); + })?; + consecutive_failures += 1; + log::error!("FAILED {} after {} attempts", f.url, opts.retries); + if consecutive_failures >= opts.cooldown_failures { + log::warn!( + "{} failures in a row - cooling down {}s before continuing \ + (rate limit?). The run does not stop.", + consecutive_failures, + opts.cooldown_seconds, + ); + thread::sleep(Duration::from_secs(opts.cooldown_seconds)); + } + } + } + Ok(()) +} + +fn json_entry(dest: &Path, size: u64, name: &str) -> Value { + serde_json::json!({ + "path": dest.display().to_string(), + "size": size, + "name": name, + }) +} + +/// 0807 pipeline (python run_0807). +pub fn run_0807( + urls: &[FoundUrl], + opts: &RunOpts, + out_dir: &Path, + state: &mut State, + source_by_url: &HashMap, + mut gstore: Option<&mut GlobalStore>, + manifest: &Path, +) -> anyhow::Result<()> { + if !urls + .iter() + .any(|f| !is_done(state, gstore.as_deref(), &f.url)) + { + return Ok(()); + } + let out_dir = out_dir.join("downloads").join("0807"); + let mut consecutive_failures: u32 = 0; + for f in urls { + if is_done(state, gstore.as_deref(), &f.url) { + log::info!("Skip (already downloaded): {}", f.url); + continue; + } + let source = source_by_url.get(&f.url).cloned().unwrap_or_default(); + let mut resolved = false; + for attempt in 1..=opts.retries { + let downloaded = zero807::download(&f.url, &out_dir) + .map_err(anyhow::Error::new) + .and_then(|(dest, size)| -> anyhow::Result<(Value, u64)> { + let mut entry = json_entry( + &dest, + size, + &dest + .file_name() + .map(|n| n.to_string_lossy().into_owned()) + .unwrap_or_default(), + ); + if opts.hash { + let h = sha256_of(&dest)?; + entry["sha256"] = Value::String(h); + } + Ok((entry, size)) + }); + match downloaded { + Ok((entry, size)) => { + log::info!("OK {} ({})", f.url, format_size(size)); + record_files( + state, + gstore.as_deref_mut(), + &f.url, + std::slice::from_ref(&entry), + &source, + manifest, + )?; + if opts.extract { + extract_zips(std::slice::from_ref(&entry)); + } + consecutive_failures = 0; + resolved = true; + break; + } + Err(err) => { + let Some(ze) = err.downcast_ref::() else { + // a bug must not stop the run + log::error!("Unexpected error for {}: {err:#}", f.url); + let msg = format!("unexpected: {err}"); + state.mark(&f.url, |r| { + r.error = Some(msg); + r.files = Vec::new(); + })?; + if attempt < opts.retries { + thread::sleep(Duration::from_secs(opts.retry_delay * attempt as u64)); + } + continue; + }; + match ze { + Zero807Error::NotFound(msg) => { + // Dead links never come back - no retry + log::error!("DEAD LINK {}: {msg}", f.url); + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some(msg.clone()); + r.files = Vec::new(); + })?; + resolved = true; + break; + } + Zero807Error::Error(msg) => { + log::warn!( + "Attempt {}/{} failed for {}: {msg}", + attempt, + opts.retries, + f.url + ); + state.mark(&f.url, |r| { + r.error = Some(msg.clone()); + r.files = Vec::new(); + })?; + if attempt < opts.retries { + thread::sleep(Duration::from_secs( + opts.retry_delay * attempt as u64, + )); + } + } + } + } + } + } + if !resolved { + let err = state + .records + .get(&f.url) + .and_then(|r| r.error.clone()) + .unwrap_or_else(|| "unknown".into()); + state.mark(&f.url, |r| { + r.status = "failed".into(); + r.error = Some(err); + r.files = Vec::new(); + })?; + consecutive_failures += 1; + log::error!("FAILED {} after {} attempts", f.url, opts.retries); + if consecutive_failures >= opts.cooldown_failures { + log::warn!( + "{} failures in a row - cooling down {}s before continuing \ + (HTTP 403 storm?). The run does not stop.", + consecutive_failures, + opts.cooldown_seconds, + ); + thread::sleep(Duration::from_secs(opts.cooldown_seconds)); + } + } + } + Ok(()) +} + +/// Load subscriptions.json: [{"name", "channel_id", "out_dir"}, ...]. +pub fn load_subscriptions(path: &Path) -> anyhow::Result)>> { + let data: Value = serde_json::from_str( + &fs::read_to_string(path).with_context(|| format!("reading {}", path.display()))?, + )?; + let entries = match &data { + Value::Object(o) => o + .get("subscriptions") + .cloned() + .unwrap_or_else(|| Value::Array(vec![])), + Value::Array(_) => data.clone(), + _ => anyhow::bail!( + "{}: expected a list or {{\"subscriptions\": [...]}}", + path.display() + ), + }; + let mut subs = Vec::new(); + for entry in entries.as_array().context("subscriptions must be a list")? { + let bad = || { + anyhow!( + "{}: each subscription needs 'channel_id' and 'out_dir' (got: {entry})", + path.display() + ) + }; + let channel_id = entry + .get("channel_id") + .and_then(Value::as_str) + .ok_or_else(bad)?; + let out_dir = entry + .get("out_dir") + .and_then(Value::as_str) + .ok_or_else(bad)?; + let name = entry + .get("name") + .and_then(Value::as_str) + .map(str::to_string); + subs.push((channel_id.to_string(), out_dir.to_string(), name)); + } + Ok(subs) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn format_size_matches_python() { + // python format_size values + assert_eq!(format_size(0), "0.0 B"); + assert_eq!(format_size(512), "512.0 B"); + assert_eq!(format_size(1024), "1.0 KiB"); + assert_eq!(format_size(1536), "1.5 KiB"); + assert_eq!(format_size(1024 * 1024), "1.0 MiB"); + assert_eq!(format_size(1024_u64 * 1024 * 1024), "1.0 GiB"); + assert_eq!(format_size(1024_u64 * 1024 * 1024 * 1024), "1.0 TiB"); + } + + #[test] + fn read_url_lines_skips_comments_and_blanks() { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("urls.txt"); + fs::write(&p, "\n# comment\nhttps://a \n\nhttps://b\n# x\n").unwrap(); + let lines = read_url_lines(&p).unwrap(); + assert_eq!(lines, vec!["https://a", "https://b"]); + } + + #[test] + fn load_urls_file_sorts_and_dedupes() { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("urls.txt"); + fs::write( + &p, + "https://gofile.io/d/b\nhttps://0807.st/d/x\nhttps://gofile.io/d/a\n", + ) + .unwrap(); + let found = load_urls_file(&p).unwrap(); + let urls: Vec<&str> = found.iter().map(|f| f.url.as_str()).collect(); + assert_eq!( + urls, + vec![ + "https://0807.st/d/x", + "https://gofile.io/d/a", + "https://gofile.io/d/b" + ] + ); + assert!(found.iter().all(|f| f.message_id.is_none())); + } + + #[test] + fn load_urls_file_warns_unsupported() { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("urls.txt"); + fs::write(&p, "https://example.com/x\n").unwrap(); + let found = load_urls_file(&p).unwrap(); + assert!(found.is_empty()); + } + + #[test] + fn write_failed_urls_empty_and_filled() { + let dir = tempfile::tempdir().unwrap(); + let mut state = State::load(dir.path().join("state.json")); + let dest = write_failed_urls(dir.path(), &state).unwrap(); + assert_eq!(fs::read_to_string(&dest).unwrap(), ""); + state + .mark("https://gofile.io/d/a", |r| r.status = "failed".into()) + .unwrap(); + state + .mark("https://gofile.io/d/b", |r| r.status = "ok".into()) + .unwrap(); + let dest = write_failed_urls(dir.path(), &state).unwrap(); + assert_eq!( + fs::read_to_string(&dest).unwrap(), + "https://gofile.io/d/a\n" + ); + } + + #[test] + fn record_files_writes_state_global_and_manifest() { + let dir = tempfile::tempdir().unwrap(); + let mut state = State::load(dir.path().join("state.json")); + let manifest = dir.path().join("manifest.jsonl"); + let mut gs = GlobalStore::load(dir.path().join("urls.json")); + let source = SourceInfo { + channel: "123".into(), + host: Some("gofile".into()), + message_id: Some(Value::from(42)), + timestamp: Some("2024-01-01".into()), + author: Some("ben".into()), + manifest: manifest.clone(), + }; + let files = vec![serde_json::json!({"path": "/tmp/x.7z", "size": 10, "name": "x.7z"})]; + record_files( + &mut state, + Some(&mut gs), + "https://gofile.io/d/a", + &files, + &source, + &manifest, + ) + .unwrap(); + assert_eq!(state.records["https://gofile.io/d/a"].status, "ok"); + assert_eq!(state.records["https://gofile.io/d/a"].files.len(), 1); + assert!(gs.is_done("https://gofile.io/d/a")); + let mtext = fs::read_to_string(&manifest).unwrap(); + let row: Value = serde_json::from_str(mtext.trim()).unwrap(); + assert_eq!(row["channel"], "123"); + assert_eq!(row["message_id"], 42); + assert_eq!(row["path"], "/tmp/x.7z"); + assert_eq!(row["name"], "x.7z"); + } + + #[test] + fn extract_zips_skips_missing_and_non_zip() { + // non-zip entries are ignored, no panic + extract_zips(&[serde_json::json!({"path": "/definitely/not/there.7z"})]); + extract_zips(&[]); + } + + #[test] + fn fisher_yates_is_permutation() { + let mut v: Vec = (0..50).collect(); + fisher_yates(&mut v); + let mut sorted = v.clone(); + sorted.sort_unstable(); + assert_eq!(sorted, (0..50).collect::>()); + } + + #[test] + fn load_subscriptions_both_shapes() { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("subs.json"); + fs::write( + &p, + r#"{"subscriptions": [{"name": "A", "channel_id": "1", "out_dir": "out_a"}]}"#, + ) + .unwrap(); + let subs = load_subscriptions(&p).unwrap(); + assert_eq!( + subs, + vec![("1".to_string(), "out_a".to_string(), Some("A".to_string()))] + ); + fs::write(&p, r#"[{"channel_id": "2", "out_dir": "out_b"}]"#).unwrap(); + let subs = load_subscriptions(&p).unwrap(); + assert_eq!(subs, vec![("2".to_string(), "out_b".to_string(), None)]); + } + + #[test] + fn load_subscriptions_rejects_missing_fields() { + let dir = tempfile::tempdir().unwrap(); + let p = dir.path().join("subs.json"); + fs::write(&p, r#"[{"channel_id": "2"}]"#).unwrap(); + assert!(load_subscriptions(&p).is_err()); + } +} diff --git a/version-rust/src/sorter.rs b/version-rust/src/sorter.rs index 5f1a4f3..5db6389 100644 --- a/version-rust/src/sorter.rs +++ b/version-rust/src/sorter.rs @@ -19,6 +19,7 @@ use std::fs; use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; +use anyhow::Context as _; use serde_json::Value; use crate::catalogue::{clean_name, match_class, match_faction}; @@ -446,7 +447,8 @@ pub fn drop_dropped_system(item: &mut SortItem) { /// (pass an empty dropped set to keep everything). /// Accepts the catalog document ({..., "entries": [...]}) or a bare list. pub fn load_catalog(path: &Path, dropped: &[String]) -> anyhow::Result> { - let text = fs::read_to_string(path)?; + let text = + fs::read_to_string(path).with_context(|| format!("reading catalog {}", path.display()))?; let data: Value = serde_json::from_str(&text)?; let entries = match &data { Value::Object(o) => o.get("entries").cloned().unwrap_or(Value::Array(vec![])),