Step 4: enclosure downloads

Streaming download with progress, content sniffing that replaces the
long-dead detectFileType(), HTML-body rejection, and placement by rename
from an incomplete dir on the same filesystem.

Filters are applied once at discovery and recorded in enclosures.state,
so the download queue is the table rather than the parse result and
max_new_per_check defers work instead of dropping it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01RPyeapneuXrCdojsaiXGbe

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<String>,
+}
+
+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<Vec<Pending>> {
+        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::<rusqlite::Result<Vec<_>>>()?;
+        Ok(rows)
+    }
+}
+
 /// Unix seconds. Everything time-shaped in the DB is stored this way.
 pub fn now() -> i64 {
     std::time::SystemTime::now()

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<String> {
+    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<u64>),
+) -> Result<Downloaded> {
+    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<PathBuf> {
+    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<Sniffed> {
+    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("<!doctype html") || lead.starts_with("<html") {
+        return Ok(Sniffed::Html);
+    }
+    Ok(Sniffed::Other)
+}
+
+async fn read_head(path: &Path, n: usize) -> Result<Vec<u8>> {
+    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<u64>) -> 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");
+    }
+}

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<Option<(usize, usize)>> {
+) -> Result<Option<Scan>> {
     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<std::path::PathBuf> {
+    // 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<i64>) -> String {