diff --git a/Cargo.lock b/Cargo.lock index acd4d94..084d2ec 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1503,9 +1503,11 @@ dependencies = [ "chrono", "clap", "dirs", + "futures-util", "infer", "librqbit", "opml", + "percent-encoding", "reqwest", "rss", "rusqlite", @@ -1515,6 +1517,7 @@ dependencies = [ "toml", "tracing", "tracing-subscriber", + "url", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index adbb79c..771f4b7 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,9 +9,11 @@ atom_syndication = "0.12.10" chrono = { version = "0.4.45", default-features = false, features = ["std", "clock"] } clap = { version = "4.6.6", features = ["derive"] } dirs = "7.0.0" +futures-util = { version = "0.3.34", default-features = false, features = ["std"] } infer = "0.22.0" librqbit = { version = "9.0.1", default-features = false, features = ["rust-tls", "http-api-client"] } opml = "1.1.6" +percent-encoding = "2.3.2" reqwest = { version = "0.13.5", default-features = false, features = ["rustls", "http2", "gzip", "stream", "json", "charset", "system-proxy"] } rss = "2.1.1" rusqlite = { version = "0.40.2", features = ["bundled"] } @@ -21,3 +23,4 @@ tokio = { version = "1.53.1", features = ["rt-multi-thread", "macros", "fs", "io toml = "1.1.5" tracing = "0.1.44" tracing-subscriber = { version = "0.3.23", features = ["env-filter"] } +url = "2.5.8" diff --git a/PROGRESS.md b/PROGRESS.md index 20aabf7..5224a08 100644 --- a/PROGRESS.md +++ b/PROGRESS.md @@ -10,9 +10,7 @@ The full design and step list live in the plan file at README, this file. - [x] **2. `config.rs` + `db.rs`** — TOML config structs + SQLite schema. - [x] **3. `feed.rs`** — conditional GET, RSS-then-Atom parse, persist entries. -- [ ] **4. `download.rs`** — filename derivation + sanitizer, streaming download, `infer` sniff, - HTML rejection, move into place, dedupe, keyword/explicit filters, `max_new_per_check`. - *Done when:* smoke 1, 2, 4 pass. +- [x] **4. `download.rs`** — downloads, filters, dedupe. - [ ] **5. `retention.rs`** — oldest-first quota + age reaper, `ipx reap [--dry-run]`. *Done when:* smoke 6 passes. - [ ] **6. `ipc.rs` + daemon** — broadcast event bus, UDS JSON-lines server, TTL scheduler, @@ -33,6 +31,38 @@ The full design and step list live in the plan file at --- +## 2026-09-09 — Step 4: download.rs + +`src/download.rs`: streaming download to `/.ipx-incomplete/` (same filesystem as the +destination, so filing it is a rename, not the original's copy-then-unlink), content sniffing, +then `place()`. Filename comes from the URL's last path segment, percent-decoded, unless +`Content-Disposition` names one (RFC 5987 `filename*=` preferred). The sanitizer keeps UTF-8 — +`latin1_to_ascii` existed because 2004 filesystems demanded ASCII — strips the same characters +`stringCleaning()` did plus control chars, and adds a real 255-byte cap the Python never had, +preserving the extension across truncation. + +Sniffing replaces `detectFileType()`, which called a `typeFile` module that was already missing in +2008 and so always answered `'data'`. Two rules survive: an HTML body is a failed download (login +wall/error page), and a torrent body is a torrent whatever the MIME claimed. + +Filters run once at discovery and are recorded in `enclosures.state`; the download queue is then +just "everything still `pending`", so an enclosure held back by `max_new_per_check` is picked up by +the next scan instead of being lost. Keywords are OR'd across keywords and AND'd within one — the +original's nested loop let a later keyword silently undo an earlier miss. + +Verified: `cargo test` 16/16. Smoke against a local server, five enclosures, each filter path hit: +`ep1.mp3 -> done`, `ep2.mp3 -> skipped (explicit)`, `ep2.mp3?v=3 -> skipped (no keyword match)`, +`paywall.html -> error (HTML page, not media)`, `ep5.torrent -> torrent (deferred to step 7)`. +Smoke 2 and 4 pass, and because a plain rerun 304s before parsing, dedupe was proved separately by +clearing the stored etag/last-modified and re-parsing all five entries: 0 downloaded, hand-deleted +file not refetched, `.ipx-incomplete` left empty. + +Known wart: `ipx list` counts `path IS NOT NULL`, so a hand-deleted file still reads as downloaded. +Reconciling rows against the filesystem belongs in step 5. +Next: step 5 — `retention.rs`. + +--- + ## 2026-09-09 — Step 3: feed.rs `src/feed.rs`: conditional GET (`If-None-Match` + `If-Modified-Since`, optional basic auth) and a diff --git a/src/db.rs b/src/db.rs index d88778d..b5e16e2 100644 --- a/src/db.rs +++ b/src/db.rs @@ -227,6 +227,28 @@ impl Db { /// Returns true when this enclosure URL is new. False means we have downloaded it /// before, or deliberately reaped it -- either way it is not fetched again. + + pub fn mark_downloaded(&self, url: &str, path: &std::path::Path, bytes: u64) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "UPDATE enclosures SET state = 'done', path = ?2, bytes_done = ?3, + downloaded_at = ?4, last_error = NULL WHERE url = ?1", + rusqlite::params![url, path.to_string_lossy(), bytes as i64, now()], + )?; + Ok(()) + } + + /// The row stays -- a failed URL is still a URL we have seen. `state` says why it has + /// no file, and a retry is an explicit act rather than something a rescan does silently. + pub fn mark_enclosure(&self, url: &str, state: &str, error: Option<&str>) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "UPDATE enclosures SET state = ?2, last_error = ?3 WHERE url = ?1", + rusqlite::params![url, state, error], + )?; + Ok(()) + } + pub fn record_enclosure( &self, feed_id: &str, @@ -243,6 +265,32 @@ impl Db { } } + +/// An enclosure waiting to be downloaded. +#[derive(Debug)] +pub struct Pending { + pub url: String, + pub mime: Option, +} + +impl Db { + /// The download queue is the table, not the parse result: an enclosure held back by + /// `max_new_per_check` is simply picked up by the next scan, in feed order. + pub fn pending(&self, feed_id: &str, limit: usize) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT url, mime FROM enclosures + WHERE feed_id = ?1 AND state = 'pending' ORDER BY id LIMIT ?2", + )?; + let rows = stmt + .query_map(rusqlite::params![feed_id, limit as i64], |r| { + Ok(Pending { url: r.get(0)?, mime: r.get(1)? }) + })? + .collect::>>()?; + Ok(rows) + } +} + /// Unix seconds. Everything time-shaped in the DB is stored this way. pub fn now() -> i64 { std::time::SystemTime::now() diff --git a/src/download.rs b/src/download.rs new file mode 100644 index 0000000..10f4f51 --- /dev/null +++ b/src/download.rs @@ -0,0 +1,344 @@ +//! Enclosure downloads. Replaces iPXDownloader.getFile / downloadFile. + +use anyhow::{Context, Result, bail}; +use futures_util::StreamExt; +use std::path::{Path, PathBuf}; +use tokio::io::AsyncWriteExt; + +use crate::config::{Config, Feed as FeedCfg, Organize}; + +/// Characters the original's stringCleaning() stripped, plus the control range and the +/// trailing dots/spaces it left in. A real length cap is new -- the Python had none. +const FORBIDDEN: &[char] = &['/', '\\', '?', '*', ':', '<', '>', '|', '"', '\'']; +const MAX_NAME_BYTES: usize = 255; + +/// Keeps UTF-8: the original transliterated to ASCII via latin1_to_ascii because 2004 +/// filesystems demanded it. Ours do not. +pub fn sanitize(name: &str) -> String { + let mut out: String = name + .chars() + .filter(|c| !c.is_control() && !FORBIDDEN.contains(c)) + .collect(); + out = out.trim().trim_matches('.').trim().to_owned(); + + if out.len() > MAX_NAME_BYTES { + // Truncate on a char boundary, keeping the extension if there is a plausible one. + let ext = Path::new(&out) + .extension() + .and_then(|e| e.to_str()) + .filter(|e| e.len() <= 8) + .map(|e| format!(".{e}")) + .unwrap_or_default(); + let keep = MAX_NAME_BYTES - ext.len(); + let mut cut = keep; + while !out.is_char_boundary(cut) { + cut -= 1; + } + out.truncate(cut); + out.push_str(&ext); + } + + if out.is_empty() { "download".into() } else { out } +} + +/// URL's last path segment, percent-decoded, unless the server names the file itself. +pub fn filename_for(url: &str, content_disposition: Option<&str>) -> String { + if let Some(name) = content_disposition.and_then(disposition_filename) { + return sanitize(&name); + } + let from_url = url::Url::parse(url) + .ok() + .and_then(|u| { + u.path_segments() + .and_then(|s| s.filter(|p| !p.is_empty()).next_back()) + .map(str::to_owned) + }) + .map(|seg| { + percent_encoding::percent_decode_str(&seg) + .decode_utf8_lossy() + .into_owned() + }) + .unwrap_or_default(); + sanitize(&from_url) +} + +fn disposition_filename(header: &str) -> Option { + for part in header.split(';') { + let part = part.trim(); + // RFC 5987 form wins when both are present. + if let Some(v) = part.strip_prefix("filename*=") { + let raw = v.rsplit('\'').next().unwrap_or(v); + return Some( + percent_encoding::percent_decode_str(raw) + .decode_utf8_lossy() + .into_owned(), + ); + } + } + for part in header.split(';') { + if let Some(v) = part.trim().strip_prefix("filename=") { + return Some(v.trim().trim_matches('"').to_owned()); + } + } + None +} + +/// What the download turned out to be, once the bytes were on disk. +pub enum Kind { + File, + /// A .torrent body, whatever the advertised MIME said. Step 7 takes it from here. + Torrent, +} + +pub struct Downloaded { + /// Still in the incomplete dir. Call `place()` to file it, or drop it. + pub tmp: PathBuf, + pub name: String, + pub bytes: u64, + pub kind: Kind, +} + +/// Streams to a temp file inside the download dir (so the final move is a rename, not a +/// cross-device copy), sniffs it, then places it. +pub async fn download( + client: &reqwest::Client, + cfg: &Config, + feed_cfg: &FeedCfg, + url: &str, + mut on_progress: impl FnMut(u64, Option), +) -> Result { + let mut req = client.get(url); + if let Some(user) = &feed_cfg.username { + req = req.basic_auth(user, feed_cfg.password()); + } + let resp = req.send().await.context("connecting")?; + if !resp.status().is_success() { + bail!("HTTP {}", resp.status()); + } + + let name = filename_for( + resp.url().as_str(), + resp.headers() + .get(reqwest::header::CONTENT_DISPOSITION) + .and_then(|v| v.to_str().ok()), + ); + let total = resp.content_length(); + + let tmp_dir = cfg.general.download_dir.join(".ipx-incomplete"); + tokio::fs::create_dir_all(&tmp_dir) + .await + .with_context(|| format!("creating {}", tmp_dir.display()))?; + let tmp = tmp_dir.join(&name); + + let mut file = tokio::fs::File::create(&tmp) + .await + .with_context(|| format!("creating {}", tmp.display()))?; + let mut done: u64 = 0; + let mut stream = resp.bytes_stream(); + while let Some(chunk) = stream.next().await { + let chunk = chunk.context("reading body")?; + file.write_all(&chunk).await?; + done += chunk.len() as u64; + on_progress(done, total); + } + file.flush().await?; + drop(file); + + let kind = match sniff(&tmp).await? { + Sniffed::Html => { + // "HTML Files are bad!" -- a login wall or an error page, never an episode. + let _ = tokio::fs::remove_file(&tmp).await; + bail!("server returned an HTML page, not a media file"); + } + Sniffed::Torrent => Kind::Torrent, + Sniffed::Other => Kind::File, + }; + + Ok(Downloaded { tmp, name, bytes: done, kind }) +} + +/// Moves a completed download into its folder. Same filesystem as the incomplete dir, so +/// this is a rename rather than the original's copy-then-unlink. +pub async fn place(d: &Downloaded, dir: &Path) -> Result { + tokio::fs::create_dir_all(dir) + .await + .with_context(|| format!("creating {}", dir.display()))?; + let dest = unique_path(dir, &d.name); + tokio::fs::rename(&d.tmp, &dest) + .await + .with_context(|| format!("moving into {}", dest.display()))?; + Ok(dest) +} + +/// Advertised as a torrent, before spending any bandwidth on it. +pub fn looks_like_torrent(url: &str, mime: Option<&str>) -> bool { + mime.is_some_and(|m| m.to_ascii_lowercase().contains("torrent")) + || url.split('?').next().unwrap_or(url).to_ascii_lowercase().ends_with(".torrent") +} + +enum Sniffed { + Html, + Torrent, + Other, +} + +/// Replaces detectFileType(), which called a `typeFile` module that was already missing in +/// 2008 and so always answered 'data'. +async fn sniff(path: &Path) -> Result { + let head = read_head(path, 512).await?; + if infer::is(&head, "torrent") || head.starts_with(b"d8:announce") || head.starts_with(b"d7:") { + return Ok(Sniffed::Torrent); + } + let text = String::from_utf8_lossy(&head); + let lead = text.trim_start().to_ascii_lowercase(); + if lead.starts_with(" Result> { + use tokio::io::AsyncReadExt; + let mut f = tokio::fs::File::open(path).await?; + let mut buf = vec![0u8; n]; + let read = f.read(&mut buf).await?; + buf.truncate(read); + Ok(buf) +} + +/// Two different URLs can yield the same filename; don't let the second clobber the first. +fn unique_path(dir: &Path, name: &str) -> PathBuf { + let candidate = dir.join(name); + if !candidate.exists() { + return candidate; + } + let stem = Path::new(name) + .file_stem() + .and_then(|s| s.to_str()) + .unwrap_or(name); + let ext = Path::new(name) + .extension() + .and_then(|e| e.to_str()) + .map(|e| format!(".{e}")) + .unwrap_or_default(); + for n in 2..1000 { + let candidate = dir.join(format!("{stem}-{n}{ext}")); + if !candidate.exists() { + return candidate; + } + } + dir.join(format!("{stem}-{}{ext}", crate::db::now())) +} + +/// Download folder for a feed: per-feed name, or per-day when organize = "date". +pub fn folder_for(cfg: &Config, id: &str, feed_cfg: &FeedCfg, title: Option<&str>) -> String { + match cfg.general.organize { + Organize::Date => chrono::Local::now().format("%m-%d-%Y").to_string(), + Organize::Feed => sanitize( + feed_cfg + .folder + .as_deref() + .or(title) + .filter(|s| !s.trim().is_empty()) + .unwrap_or(id), + ), + } +} + +/// Keywords are OR'd; the words within one keyword are AND'd. The original's nested loop +/// let a later keyword silently undo an earlier miss -- this is what it meant to do. +pub fn matches_keywords(keywords: &[String], haystacks: &[&str]) -> bool { + if keywords.is_empty() { + return true; + } + let hay = haystacks.join(" ").to_lowercase(); + keywords.iter().any(|kw| { + let mut words = kw.split_whitespace().filter(|w| w.len() > 1).peekable(); + words.peek().is_some() && words.all(|w| hay.contains(&w.to_lowercase())) + }) +} + +/// Formats a progress line, throttled by the caller to ~1% steps as the original's +/// lastDLStepSize guard did. +pub fn progress_line(name: &str, done: u64, total: Option) -> String { + match total { + Some(t) if t > 0 => format!( + " {name}: {:.1}% ({:.1}/{:.1} MB)", + done as f64 / t as f64 * 100.0, + done as f64 / 1_048_576.0, + t as f64 / 1_048_576.0 + ), + _ => format!(" {name}: {:.1} MB", done as f64 / 1_048_576.0), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sanitize_strips_path_and_control_characters() { + assert_eq!(sanitize("../../etc/passwd"), "etcpasswd"); + assert_eq!(sanitize("Ep 12: The \"Best\" One?"), "Ep 12 The Best One"); + assert_eq!(sanitize("bad\u{0}name\u{7}.mp3"), "badname.mp3"); + assert_eq!(sanitize(" spaced.mp3 "), "spaced.mp3"); + } + + #[test] + fn sanitize_never_yields_an_empty_or_dot_name() { + assert_eq!(sanitize(""), "download"); + assert_eq!(sanitize("..."), "download"); + assert_eq!(sanitize("/"), "download"); + assert_eq!(sanitize(" "), "download"); + } + + #[test] + fn sanitize_caps_length_and_keeps_the_extension() { + let long = format!("{}.mp3", "a".repeat(400)); + let out = sanitize(&long); + assert!(out.len() <= MAX_NAME_BYTES, "got {}", out.len()); + assert!(out.ends_with(".mp3"), "extension survives truncation"); + } + + #[test] + fn sanitize_keeps_unicode() { + assert_eq!(sanitize("Café Münster – ep1.mp3"), "Café Münster – ep1.mp3"); + } + + #[test] + fn filename_comes_from_the_url_then_the_header() { + assert_eq!( + filename_for("https://x.com/media/ep1.mp3?token=abc", None), + "ep1.mp3", + "query string is not part of the name" + ); + assert_eq!( + filename_for("https://x.com/media/My%20Show%20Ep%201.mp3", None), + "My Show Ep 1.mp3" + ); + assert_eq!( + filename_for("https://x.com/dl?id=9", Some("attachment; filename=\"real name.mp3\"")), + "real name.mp3", + "Content-Disposition overrides the URL" + ); + assert_eq!( + filename_for("https://x.com/dl", Some("attachment; filename*=UTF-8''caf%C3%A9.mp3")), + "café.mp3" + ); + assert_eq!(filename_for("https://x.com/", None), "download"); + assert_eq!( + filename_for("https://x.com/a/../../etc/passwd", None), + "passwd", + "a traversal in the URL cannot escape the download dir" + ); + } + + #[test] + fn keyword_matching_is_or_across_keywords_and_and_within_one() { + let kws = vec!["deep dive".to_string(), "interview".to_string()]; + assert!(matches_keywords(&kws, &["A deep and thorough dive"])); + assert!(matches_keywords(&kws, &["An Interview with someone"])); + assert!(!matches_keywords(&kws, &["Just a dive"]), "half a keyword is not a match"); + assert!(matches_keywords(&[], &["anything"]), "no keywords means take everything"); + } +} diff --git a/src/main.rs b/src/main.rs index 0f7b15e..8f67901 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,5 +1,6 @@ mod config; mod db; +mod download; mod feed; use anyhow::Result; @@ -91,10 +92,11 @@ async fn fetch( } } - match scan_one(&client, db, id, feed_cfg, &state).await { - Ok(Some((new_entries, new_encs))) => { - println!("{id}: {new_entries} new entries, {new_encs} new enclosures") - } + match scan_one(&client, cfg, db, id, feed_cfg, &state).await { + Ok(Some(s)) => println!( + "{id}: {} new entries, {} downloaded, {} failed, {} torrents deferred", + s.new_entries, s.downloaded, s.failed, s.torrents + ), Ok(None) => println!("{id}: not modified"), Err(e) => { // One bad feed must not end the scan. @@ -110,11 +112,12 @@ async fn fetch( /// Ok(None) means 304. async fn scan_one( client: &reqwest::Client, + cfg: &config::Config, db: &db::Db, id: &str, feed_cfg: &config::Feed, state: &db::HttpState, -) -> Result> { +) -> Result> { let fetched = feed::fetch( client, feed_cfg, @@ -141,19 +144,114 @@ async fn scan_one( parsed.ttl_mins, )?; - let mut new_entries = 0; - let mut new_encs = 0; + let mut scan = Scan::default(); for entry in &parsed.entries { if db.record_entry(id, entry)? { - new_entries += 1; + scan.new_entries += 1; } for enc in &entry.enclosures { - if db.record_enclosure(id, &entry.guid, enc)? { - new_encs += 1; + if !db.record_enclosure(id, &entry.guid, enc)? { + continue; // Seen before: downloaded, skipped or deliberately reaped. + } + // Filters run once, at discovery, and are recorded in `state`. The download + // queue below is then just "everything still pending". + if let Some(reason) = reject(feed_cfg, entry, &enc.url) { + db.mark_enclosure(&enc.url, "skipped", Some(reason))?; } } } - Ok(Some((new_entries, new_encs))) + + let budget = feed_cfg.max_new_per_check.unwrap_or(usize::MAX); + if feed_cfg.auto_download && budget > 0 { + let folder = download::folder_for(cfg, id, feed_cfg, parsed.title.as_deref()); + let dest_dir = cfg.general.download_dir.join(&folder); + + for item in db.pending(id, budget)? { + if download::looks_like_torrent(&item.url, item.mime.as_deref()) { + // Step 7 owns these; recorded so nothing re-queues them meanwhile. + db.mark_enclosure(&item.url, "torrent", None)?; + scan.torrents += 1; + continue; + } + match fetch_one(client, cfg, db, feed_cfg, &item.url, &dest_dir).await { + Ok(path) => { + println!(" saved {}", path.display()); + scan.downloaded += 1; + } + Err(e) => { + let msg = format!("{e:#}"); + println!(" failed {}: {msg}", item.url); + db.mark_enclosure(&item.url, "error", Some(&msg))?; + scan.failed += 1; + } + } + } + } + + Ok(Some(scan)) +} + +#[derive(Default)] +struct Scan { + new_entries: usize, + downloaded: usize, + failed: usize, + torrents: usize, +} + +/// Why this enclosure should not be downloaded, if it should not be. +fn reject(feed_cfg: &config::Feed, entry: &feed::Entry, url: &str) -> Option<&'static str> { + if !feed_cfg.auto_download { + return Some("auto_download is off"); + } + if entry.explicit && !feed_cfg.allow_explicit { + return Some("explicit"); + } + let categories = entry.categories.join(" "); + let haystacks = [ + url, + entry.title.as_deref().unwrap_or(""), + entry.description.as_deref().unwrap_or(""), + categories.as_str(), + ]; + if !download::matches_keywords(&feed_cfg.keywords, &haystacks) { + return Some("no keyword match"); + } + None +} + +async fn fetch_one( + client: &reqwest::Client, + cfg: &config::Config, + db: &db::Db, + feed_cfg: &config::Feed, + url: &str, + dest_dir: &std::path::Path, +) -> Result { + // Throttled to whole percents, as the original's lastDLStepSize guard did. + let mut last_pct = -1i64; + let name = download::filename_for(url, None); + let got = download::download(client, cfg, feed_cfg, url, |done, total| { + if let Some(t) = total.filter(|t| *t > 0) { + let pct = (done * 100 / t) as i64; + if pct > last_pct { + last_pct = pct; + println!("{}", download::progress_line(&name, done, total)); + } + } + }) + .await?; + + if matches!(got.kind, download::Kind::Torrent) { + // The MIME lied. Don't file a .torrent as an episode. + let _ = tokio::fs::remove_file(&got.tmp).await; + db.mark_enclosure(url, "torrent", None)?; + anyhow::bail!("body is a torrent, deferred to the torrent downloader"); + } + + let path = download::place(&got, dest_dir).await?; + db.mark_downloaded(url, &path, got.bytes)?; + Ok(path) } fn ago(t: Option) -> String {