diff --git a/feeder-plugin/usenet-feeder/src/anchor.rs b/feeder-plugin/usenet-feeder/src/anchor.rs new file mode 100644 index 0000000..55e6976 --- /dev/null +++ b/feeder-plugin/usenet-feeder/src/anchor.rs @@ -0,0 +1,425 @@ +//! Name matching for anchored queries — how a self-scanned release lands on a +//! meta-watch show page **matched by title**, never anchored. +//! +//! A show page asks the gateway for `tmdbid: season: …` with no free +//! text, and keeps a release only when the release itself names that `tmdbid` +//! (and carries a poster + description). NNTmux knows release names, not TMDB +//! ids, and this feeder makes no TMDB calls. The show's identity is already on +//! the mesh, though: the card feeder stores every show it surfaced in the +//! gateway's meta-core as a **card record** keyed by its locator cid +//! (`compute_card_cid("tmdb", "tv:")`, the same address meta-watch routes +//! by). So: +//! +//! 1. read the card (title, AKAs, poster, description) from meta-core; +//! 2. search NNTmux by the card's title; +//! 3. keep the releases whose cleaned name resembles one of the card's names +//! and whose numbering agrees with the query; +//! 4. stamp them with the query's id, `anchored=true` + `anchorMethod=title`, +//! and the card's poster/description. +//! +//! `anchorMethod=title` is what keeps this honest: meta-watch grades it +//! *Matched* ("By title"), below an id-confirmed release, and the gateway files +//! it as a durable `matchedBy/tmdb:` claim rather than `anchoredBy/`. + +use std::collections::{BTreeMap, HashMap}; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use meta_feeder_sdk::filename_meta::{clean_title, extract_season_episode}; +use meta_feeder_sdk::hash::compute_card_cid; +use meta_feeder_sdk::query::GatewayQuery; +use tracing::{debug, warn}; + +/// A release name must be *about* the show, not merely mention it: the card's +/// tokens must all appear and cover this share of the release's tokens. Same +/// rule and threshold as the torznab feeder's anchor guard +/// (`tokens_resemble_anchor`), which is crate-private there. +const MIN_TITLE_COVERAGE: f64 = 0.6; + +/// A card found in meta-core is good for an hour; a miss is retried sooner, +/// since the card feeder may persist it any moment. +const CARD_TTL: Duration = Duration::from_secs(3600); +const MISS_TTL: Duration = Duration::from_secs(300); + +/// What a show page's query is about. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ShowKind { + Tv, + Movie, +} + +impl ShowKind { + fn locator_prefix(self) -> &'static str { + match self { + ShowKind::Tv => "tv", + ShowKind::Movie => "movie", + } + } +} + +/// The anchor a query carries: the TMDB id, the kinds it may be, and the +/// numbering it asks for. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct QueryAnchor { + pub tmdbid: String, + /// Most likely first. Both when the query does not say. + pub kinds: Vec, + pub season: Option, + pub episode: Option, +} + +fn first_filter<'a>(q: &'a GatewayQuery, key: &str) -> Option<&'a str> { + q.filters + .get(key) + .and_then(|v| v.first()) + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) +} + +/// `numeric` → normalised (`"02"` → `"2"`), else `None`. +fn number(s: &str) -> Option { + s.trim().parse::().ok().map(|n| n.to_string()) +} + +impl QueryAnchor { + /// `None` unless the query names a numeric `tmdbid`. + pub fn from_query(q: &GatewayQuery) -> Option { + let tmdbid = first_filter(q, "tmdbid").and_then(number)?; + let kinds_said: Vec<&str> = q + .filters + .get("contentKind") + .map(|v| v.iter().map(|s| s.trim()).collect()) + .unwrap_or_default(); + // `pack` says nothing: a film page asks for `movie OR pack`, a show + // page for `series OR episode OR pack`. + let tv = kinds_said.iter().any(|k| matches!(*k, "episode" | "series" | "tv")); + let movie = kinds_said.contains(&"movie"); + let kinds = match (tv, movie) { + (true, false) => vec![ShowKind::Tv], + (false, true) => vec![ShowKind::Movie], + _ => vec![ShowKind::Tv, ShowKind::Movie], + }; + Some(Self { + tmdbid, + kinds, + season: first_filter(q, "season").and_then(number), + episode: first_filter(q, "episode").and_then(number), + }) + } +} + +/// The parts of a card record this feeder uses. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct Card { + pub kind: ShowKind, + /// Display title first, then `searchTitle` and the `titles//` + /// AKAs. Deduped, never empty. + pub names: Vec, + /// Copied onto matched releases: poster, every `description/`, and + /// the cross-source ids. + pub stamp: BTreeMap, + /// The AKA the card feeder picked for searching (`3%` → `3 percent`). + pub search_title: Option, +} + +impl Card { + /// `None` when the record has no usable title. + pub fn from_fields(kind: ShowKind, f: &BTreeMap) -> Option { + let mut names: Vec = Vec::new(); + let mut push = |s: &str| { + let s = s.trim(); + if !s.is_empty() && !names.iter().any(|n| n.eq_ignore_ascii_case(s)) { + names.push(s.to_string()); + } + }; + push(f.get("title")?); + if let Some(s) = f.get("searchTitle") { + push(s); + } + for k in f.keys() { + if let Some(rest) = k.strip_prefix("titles/") { + if let Some((_, name)) = rest.split_once('/') { + push(name); + } + } + } + let stamp = f + .iter() + .filter(|(k, _)| { + matches!(k.as_str(), "poster" | "imdbid" | "tvdbid") || k.starts_with("description/") + }) + .map(|(k, v)| (k.clone(), v.clone())) + .collect(); + let search_title = f.get("searchTitle").map(|s| s.trim().to_string()).filter(|s| !s.is_empty()); + Some(Self { kind, names, stamp, search_title }) + } + + /// The name to search NNTmux with (the display title), plus `searchTitle` + /// when the card has one (a title NNTmux could not match as typed, `3%`). + pub fn search_terms(&self) -> Vec { + let mut out = vec![self.names[0].clone()]; + if let Some(s) = &self.search_title { + if !out.iter().any(|o| o.eq_ignore_ascii_case(s)) { + out.push(s.clone()); + } + } + out + } +} + +fn norm_tokens(s: &str) -> Vec { + s.split(|c: char| !c.is_alphanumeric()) + .filter(|t| !t.is_empty()) + .map(|t| t.to_lowercase()) + .collect() +} + +fn tokens_resemble(release: &[String], name: &str) -> bool { + let want = norm_tokens(name); + if want.is_empty() || release.is_empty() || !want.iter().all(|t| release.contains(t)) { + return false; + } + let covered = release.iter().filter(|t| want.contains(t)).count(); + covered as f64 >= release.len() as f64 * MIN_TITLE_COVERAGE +} + +/// Is this release name about one of these show names? +pub fn name_matches(release_name: &str, names: &[String]) -> bool { + let tokens = norm_tokens(&clean_title(release_name)); + names.iter().any(|n| tokens_resemble(&tokens, n)) +} + +/// Does the release's own numbering agree with what the query asks for? A +/// number the release does not state never disqualifies it. +pub fn numbering_agrees(release_name: &str, anchor: &QueryAnchor) -> bool { + let se = extract_season_episode(release_name); + if let (Some(want), Some(have)) = (&anchor.season, se.season.as_deref().and_then(number)) { + if se.season_explicit && *want != have { + return false; + } + } + if let Some(want) = &anchor.episode { + if let Some(have) = se.episode.as_deref().and_then(number) { + return *want == have; + } + // A pack carrying an explicit range: inside it or not. + if let (Some(a), Some(b)) = ( + se.episode_start.as_deref().and_then(|s| s.parse::().ok()), + se.episode_end.as_deref().and_then(|s| s.parse::().ok()), + ) { + let w: u32 = want.parse().unwrap_or(0); + return (a..=b).contains(&w); + } + } + true +} + +/// A cached lookup: the card, or `None` for a remembered miss. +type CardSlot = Option>; + +/// Reads card records out of the gateway's meta-core, with a small cache. +pub struct CardBook { + http: reqwest::Client, + meta_core_url: Option, + cache: Mutex>, + warned_unset: std::sync::atomic::AtomicBool, +} + +impl CardBook { + pub fn new(http: reqwest::Client, meta_core_url: Option) -> Self { + Self { + http, + meta_core_url: meta_core_url.map(|u| u.trim_end_matches('/').to_string()).filter(|u| !u.is_empty()), + cache: Mutex::new(HashMap::new()), + warned_unset: std::sync::atomic::AtomicBool::new(false), + } + } + + /// The card for this anchor, trying each kind the + /// query allows. `None` = no card on the mesh yet: nothing to match by. + pub async fn card_for(&self, anchor: &QueryAnchor) -> CardSlot { + let Some(base) = self.meta_core_url.as_deref() else { + if !self.warned_unset.swap(true, std::sync::atomic::Ordering::Relaxed) { + warn!(target: "meta-feeder", "usenet: META_CORE_URL is unset — show-page (tmdbid:) queries get no name matches"); + } + return None; + }; + for kind in &anchor.kinds { + let id = format!("{}:{}", kind.locator_prefix(), anchor.tmdbid); + let Some(locator) = compute_card_cid("tmdb", &id) else { + continue; + }; + if let Some(hit) = self.cached(&locator) { + match hit { + Some(card) => return Some(card), + None => continue, + } + } + let found = match meta_feeder_sdk::meta_core::get_record(&self.http, base, &locator).await { + Ok(Some(f)) => Card::from_fields(*kind, &f).map(Arc::new), + Ok(None) => None, + Err(e) => { + // Not cached: a meta-core blip should not hide a show for minutes. + debug!(target: "meta-feeder", %locator, error = %e, "usenet: card lookup failed"); + continue; + } + }; + self.cache.lock().unwrap().insert(locator.clone(), (Instant::now(), found.clone())); + if found.is_some() { + return found; + } + } + None + } + + fn cached(&self, locator: &str) -> Option { + let cache = self.cache.lock().unwrap(); + let (at, v) = cache.get(locator)?; + let ttl = if v.is_some() { CARD_TTL } else { MISS_TTL }; + (at.elapsed() < ttl).then(|| v.clone()) + } +} + +/// Stamp a release the anchor matched by name. +pub fn stamp_match(fields: &mut BTreeMap, anchor: &QueryAnchor, card: &Card) { + fields.insert("tmdbid".into(), anchor.tmdbid.clone()); + fields.insert("anchored".into(), "true".into()); + fields.insert("anchorMethod".into(), "title".into()); + for (k, v) in &card.stamp { + fields.entry(k.clone()).or_insert_with(|| v.clone()); + } + // A release NNTmux left uncategorised (a season pack in "Other") still + // belongs to the show page once its name matched: give it the routing axes + // the page filters on (`domain:screen`, a TV/movie kind). + if !fields.contains_key("contentKind") { + let kind = match card.kind { + ShowKind::Movie => "movie", + ShowKind::Tv if fields.contains_key("episode") => "episode", + ShowKind::Tv => "pack", + }; + set_kind(fields, kind); + } +} + +/// Write `contentKind` with its routing axes. `pack` is the one kind the SDK +/// cannot route on its own (a season pack and an album are the same shape), so +/// the writer says it: every pack this feeder emits is a screen work. +pub fn set_kind(fields: &mut BTreeMap, kind: &str) { + fields.insert("contentKind".into(), kind.into()); + let domain = meta_feeder_sdk::domain::domain_for_content_kind(kind).or((kind == "pack").then_some("screen")); + if let Some(d) = domain { + fields.insert("domain".into(), d.to_string()); + } + let form = meta_feeder_sdk::domain::work_form_for_content_kind(kind).or((kind == "pack").then_some("serial")); + if let Some(wf) = form { + fields.insert("workForm".into(), wf.to_string()); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn q(filters: &[(&str, &[&str])]) -> GatewayQuery { + let mut g = GatewayQuery::from_free_text(""); + for (k, vs) in filters { + g.filters.insert(k.to_string(), vs.iter().map(|s| s.to_string()).collect()); + } + g + } + + fn names(ns: &[&str]) -> Vec { + ns.iter().map(|s| s.to_string()).collect() + } + + #[test] + fn anchor_reads_id_kind_and_numbering() { + let a = QueryAnchor::from_query(&q(&[ + ("tmdbid", &["112442"]), + ("season", &["02"]), + ("contentKind", &["series", "episode", "pack"]), + ])) + .unwrap(); + assert_eq!(a.tmdbid, "112442"); + assert_eq!(a.kinds, vec![ShowKind::Tv]); + assert_eq!(a.season.as_deref(), Some("2")); + assert_eq!(a.episode, None); + + let film = QueryAnchor::from_query(&q(&[("tmdbid", &["603"]), ("contentKind", &["movie", "pack"])])).unwrap(); + assert_eq!(film.kinds, vec![ShowKind::Movie]); + let unsaid = QueryAnchor::from_query(&q(&[("tmdbid", &["603"])])).unwrap(); + assert_eq!(unsaid.kinds, vec![ShowKind::Tv, ShowKind::Movie]); + + assert!(QueryAnchor::from_query(&q(&[])).is_none()); + assert!(QueryAnchor::from_query(&q(&[("tmdbid", &["tt0133093"])])).is_none()); + } + + #[test] + fn release_names_match_the_show_not_a_mention_of_it() { + let n = names(&["Trash Truck"]); + assert!(name_matches("Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL.DDP5.1.H.264-NTb", &n)); + assert!(name_matches("Trash.Truck.S01E02.Slumber.Party.1080p.NF.WEB-DL.DDP5.1.x264-LAZY", &n)); + assert!(!name_matches("A.Trash.Truck.Christmas.Special.Behind.The.Scenes.Making.Of.2020.1080p", &n)); + assert!(!name_matches("Regular.Show.S02E19.Grave.Sights.1080p.DD2.0.VC-1.REMUX-FraMeSToR", &n)); + // An AKA matches as well as the display title. + assert!(name_matches("Pikappu.S01E03.1080p.WEB.h264-X", &names(&["Trash Truck", "Pikappu"]))); + } + + #[test] + fn numbering_must_agree_when_the_release_states_it() { + let a = |season: Option<&str>, episode: Option<&str>| QueryAnchor { + tmdbid: "1".into(), + kinds: vec![ShowKind::Tv], + season: season.map(str::to_string), + episode: episode.map(str::to_string), + }; + let rel = "Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL-NTb"; + assert!(numbering_agrees(rel, &a(Some("2"), None))); + assert!(numbering_agrees(rel, &a(Some("2"), Some("16")))); + assert!(!numbering_agrees(rel, &a(Some("1"), None))); + assert!(!numbering_agrees(rel, &a(Some("2"), Some("5")))); + // Says nothing about the episode → kept. + assert!(numbering_agrees("Trash.Truck.S02.1080p.NF.WEB-DL-NTb", &a(Some("2"), Some("5")))); + assert!(numbering_agrees(rel, &a(None, None))); + } + + #[test] + fn card_collects_names_and_stamp() { + let mut f = BTreeMap::new(); + f.insert("title".into(), "Trash Truck".into()); + f.insert("titles/jpn/Gomi Torakku".into(), "true".into()); + f.insert("titles/eng/trash truck".into(), "true".into()); + f.insert("poster".into(), "bafkpost".into()); + f.insert("description/eng".into(), "Hank and a truck.".into()); + f.insert("imdbid".into(), "tt9288860".into()); + f.insert("genres/Kids".into(), "true".into()); + let c = Card::from_fields(ShowKind::Tv, &f).unwrap(); + assert_eq!(c.names, names(&["Trash Truck", "Gomi Torakku"])); + assert_eq!(c.stamp.len(), 3); + assert!(!c.stamp.contains_key("genres/Kids")); + assert!(Card::from_fields(ShowKind::Tv, &BTreeMap::new()).is_none()); + } + + #[test] + fn stamp_marks_a_title_match_and_fills_display_fields_only_when_absent() { + let mut f = BTreeMap::new(); + f.insert("title".into(), "Trash Truck".into()); + f.insert("poster".into(), "bafkpost".into()); + f.insert("description/eng".into(), "Hank.".into()); + let card = Card::from_fields(ShowKind::Tv, &f).unwrap(); + let anchor = QueryAnchor { tmdbid: "112442".into(), kinds: vec![ShowKind::Tv], season: None, episode: None }; + + let mut rec = BTreeMap::new(); + rec.insert("poster".to_string(), "own".to_string()); + rec.insert("episode".to_string(), "16".to_string()); + stamp_match(&mut rec, &anchor, &card); + assert_eq!(rec["tmdbid"], "112442"); + assert_eq!(rec["anchored"], "true"); + assert_eq!(rec["anchorMethod"], "title"); + assert_eq!(rec["poster"], "own"); + assert_eq!(rec["description/eng"], "Hank."); + // Uncategorised by NNTmux → routed to the show page anyway. + assert_eq!(rec["contentKind"], "episode"); + assert_eq!(rec["domain"], "screen"); + } +} diff --git a/feeder-plugin/usenet-feeder/src/lib.rs b/feeder-plugin/usenet-feeder/src/lib.rs index 80422ab..f00ead3 100644 --- a/feeder-plugin/usenet-feeder/src/lib.rs +++ b/feeder-plugin/usenet-feeder/src/lib.rs @@ -34,4 +34,6 @@ pub mod nntmux { pub mod newznab; /// The `usenet` upstream itself. +pub mod anchor; + pub mod usenet; diff --git a/feeder-plugin/usenet-feeder/src/usenet.rs b/feeder-plugin/usenet-feeder/src/usenet.rs index 9a13728..dd19d69 100644 --- a/feeder-plugin/usenet-feeder/src/usenet.rs +++ b/feeder-plugin/usenet-feeder/src/usenet.rs @@ -49,6 +49,7 @@ use std::time::Duration; use async_trait::async_trait; use meta_feeder_sdk::common::build_http_client; use meta_feeder_sdk::config::{ConfigField as F, ConfigSchema}; +use meta_feeder_sdk::filename_meta::extract_season_episode; use meta_feeder_sdk::hash::{ codec_of, compute_ipfs_cid, compute_nzb_posting_cid, decode_nzb_release_cid, NZB_RELEASE_CODEC, }; @@ -57,6 +58,7 @@ use meta_feeder_sdk::query::GatewayQuery; use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth}; use tracing::{debug, info, warn}; +use crate::anchor::{self, CardBook, QueryAnchor}; use crate::newznab::{self, IndexerCred}; use crate::nntmux::api::{is_guid, NntmuxApi, DEFAULT_NNTMUX_URL}; use crate::nntmux::cache::{NzbCache, Posting}; @@ -76,6 +78,9 @@ const USER_AGENT: &str = concat!("meta-feeder-usenet/", env!("CARGO_PKG_VERSION" /// Wall-clock budget for one `.nzb` grab. The viewer is waiting on it. const GRAB_TIMEOUT_SECS: u64 = 60; +/// The `indexer` label when the config page sets none. +pub const DEFAULT_INDEXER_NAME: &str = "nntmux"; + /// How often the nntmux reachability probe (`t=caps`, free) runs. const PROBE_INTERVAL: Duration = Duration::from_secs(300); @@ -94,6 +99,10 @@ pub struct Settings { /// Per-host Newznab keys for redeeming `nzb-release` locators. Config page /// only — no env seed. pub indexers: Vec, + /// The source name every record carries (`indexer`), shown beside + /// "nzbgeek.info"-style names in a client's source list. It names who + /// indexed the release — the bytes are Usenet's for every NZB source alike. + pub indexer_name: String, } impl Settings { @@ -103,6 +112,7 @@ impl Settings { nntmux_url: env("NNTMUX_URL").unwrap_or_else(|| DEFAULT_NNTMUX_URL.to_string()), nntmux_api_key: env("NNTMUX_API_KEY").unwrap_or_default(), indexers: Vec::new(), + indexer_name: env("INDEXER_NAME").unwrap_or_else(|| DEFAULT_INDEXER_NAME.to_string()), } } @@ -122,6 +132,9 @@ impl Settings { if let Some(s) = text("nntmux_api_key") { self.nntmux_api_key = s.to_string(); } + if let Some(s) = text("indexer_name") { + self.indexer_name = s.to_string(); + } if text("db_url").is_some() && self.nntmux_api_key.is_empty() { // A pre-1.1 config: the feeder read nntmux's database directly. warn!( @@ -140,6 +153,20 @@ impl Settings { } } +/// `nntmux (watch)` → `nntmux-watch`: a source-member leaf safe to split on `:`. +fn source_slug(name: &str) -> String { + let mut out = String::new(); + for c in name.chars() { + if c.is_alphanumeric() || c == '.' { + out.extend(c.to_lowercase()); + } else if !out.ends_with('-') { + out.push('-'); + } + } + let out = out.trim_matches('-').to_string(); + if out.is_empty() { DEFAULT_INDEXER_NAME.to_string() } else { out } +} + pub struct UsenetPlugin { settings: Settings, /// `None` when no nntmux key is configured (grab-only deployment). @@ -157,6 +184,9 @@ pub struct UsenetPlugin { /// `https` in production — a locator carries no scheme and every public /// Newznab API is TLS. Tests point it at a plain-http mock. grab_scheme: &'static str, + /// Show cards from the gateway's meta-core, for name-matching `tmdbid:` + /// queries ([`crate::anchor`]). + cards: Arc, } impl Default for UsenetPlugin { @@ -174,6 +204,7 @@ impl UsenetPlugin { nntmux_problem: Arc::new(RwLock::new(None)), http: build_http_client(GRAB_TIMEOUT_SECS, USER_AGENT, None), grab_scheme: "https", + cards: Arc::new(CardBook::new(reqwest::Client::new(), None)), } } @@ -284,6 +315,88 @@ impl UsenetPlugin { .ok_or_else(|| GatewayError::Permanent(format!("nzb of release {guid} has no segments"))) } + /// Mint each hit's cid and build its record. + async fn mint_records(&self, releases: &[Release]) -> Vec { + // Mint each hit's cid here — see the `record_for` doc for why this + // cannot wait for compute. A release whose `.nzb` can't be had is + // DROPPED rather than surfaced uncid'd: an unplayable row in the + // client's "raw sources" list is worse than one fewer row. Once nntmux + // says a daily cap is spent, stop asking for this query. + let mut out = Vec::with_capacity(releases.len()); + let mut skipped = 0usize; + let mut capped = false; + for r in releases { + if capped { + skipped += 1; + continue; + } + match self.posting(&r.guid).await { + Ok(p) => out.push(self.record_for(r, Some(&p.cid))), + Err(GatewayError::RateLimited { .. }) => { + capped = true; + skipped += 1; + warn!(target: "meta-feeder", "usenet: nntmux download cap reached — remaining hits skipped (raise the feeder user's role caps)"); + } + Err(e) => { + skipped += 1; + debug!(target: "meta-feeder", guid = %r.guid, error = %e, "usenet: release skipped (no usable .nzb)"); + } + } + } + if skipped > 0 { + debug!(target: "meta-feeder", hits = out.len(), skipped, "usenet: releases skipped"); + } + out + } + + /// A `tmdbid:` query: read the show's card, search NNTmux by its title, + /// keep the releases whose names match and whose numbering agrees, and + /// stamp them as a title match. No card → nothing to match by → `[]`. + async fn handle_anchored( + &self, + anchor: &QueryAnchor, + max_results: usize, + ) -> Result, GatewayError> { + let Some(card) = self.cards.card_for(anchor).await else { + debug!(target: "meta-feeder", tmdbid = %anchor.tmdbid, "usenet: no card on meta-core for this id; nothing to match by name"); + return Ok(vec![]); + }; + let card = card.as_ref(); + let (api, _) = self.nntmux()?; + let mut seen = std::collections::HashSet::new(); + let mut matched: Vec = Vec::new(); + let mut looked = 0usize; + for term in card.search_terms() { + let releases = api.search(&term, max_results).await.inspect_err(|e| self.note_problem(e))?; + *self.nntmux_problem.write().unwrap() = None; + looked += releases.len(); + // Filter BEFORE minting: minting downloads the `.nzb`, and a + // release that is not this show must cost nothing. + for r in releases { + let kind_ok = !matches!( + (card.kind, content_kind_for_category(r.category_id)), + (anchor::ShowKind::Tv, Some("movie")) | (anchor::ShowKind::Movie, Some("episode")) + ); + if kind_ok + && anchor::name_matches(&r.search_name, &card.names) + && anchor::numbering_agrees(&r.search_name, anchor) + && seen.insert(r.guid.clone()) + { + matched.push(r); + } + } + } + let mut out = self.mint_records(&matched).await; + for rec in &mut out { + anchor::stamp_match(&mut rec.fields, anchor, card); + } + debug!( + target: "meta-feeder", tmdbid = %anchor.tmdbid, show = %card.names[0], looked, matched = out.len(), + "usenet: name-matched search" + ); + Ok(out) + } + /// Build the discovery record for one release. Metadata only — the bytes /// live on Usenet and are fetched by meta-share at playback, never by us. /// @@ -334,27 +447,43 @@ impl UsenetPlugin { "fileType".to_string(), file_type_for_category(r.category_id).to_string(), ); - if let Some(kind) = content_kind_for_category(r.category_id) { - fields.insert("contentKind".to_string(), kind.to_string()); + let kind = content_kind_for_category(r.category_id); + // Numbering from the release name, for TV releases (and uncategorised + // ones that plainly carry an `SxxEyy`). A season pack gets the season + // only and becomes `pack`, as the torznab feeder files them. + let se = extract_season_episode(&r.search_name); + let tv = kind == Some("episode") || (kind.is_none() && se.season_explicit); + let kind = if tv && se.is_pack { Some("pack") } else { kind }; + if tv { + if let Some(s) = &se.season { + fields.insert("season".to_string(), s.clone()); + } + if !se.is_pack { + if let Some(e) = &se.episode { + fields.insert("episode".to_string(), e.clone()); + } + } + } + if let Some(kind) = kind { // The routing axes, co-written with the kind (METADATA_KEYS.md §1): // `fileType` says what the bytes are, `domain` says which app wants // the record, `workForm` whether it stands alone or is one // instalment. Without the domain the row never reaches a wall. - if let Some(d) = meta_feeder_sdk::domain::domain_for_content_kind(kind) { - fields.insert("domain".to_string(), d.to_string()); - } - if let Some(wf) = meta_feeder_sdk::domain::work_form_for_content_kind(kind) { - fields.insert("workForm".to_string(), wf.to_string()); - } + anchor::set_kind(&mut fields, kind); } if let Some(ts) = r.post_date { fields.insert("publishedAt".to_string(), ts.to_string()); } - // Provenance: this catalog is ours, not an external indexer's. Kept - // distinct from indexer-feeder's `indexer` field on purpose — an - // operator reading a record should be able to tell at a glance whether - // its metadata came from a third party or from our own scan. + // Provenance: who indexed it. `indexer` is the name a client shows + // beside "nzbgeek.info", and a plain `source/` member carries it + // into the source facets the way the torznab feeder files its indexer + // (`source/nzb.life`). `source/gateway:usenet-scan` stays as the + // selector for everything this feeder published. + let name = self.settings.indexer_name.trim(); + let name = if name.is_empty() { DEFAULT_INDEXER_NAME } else { name }; + fields.insert("indexer".to_string(), name.to_string()); fields.insert("source/gateway:usenet-scan".to_string(), "true".to_string()); + fields.insert(format!("source/{}", source_slug(name)), "true".to_string()); DiscoveryRecord { upstream_id: UPSTREAM_ID.to_string(), record_id: r.guid.clone(), @@ -372,6 +501,10 @@ impl FeederPlugin for UsenetPlugin { fn configure(&mut self, cache_dir: &Path) -> Result<(), ConfigError> { let mut settings = Settings::from_env(); settings.merge_file(cache_dir); + // The gateway's meta-core, where show cards live (`crate::anchor`). + // Unset → show-page queries simply get no name matches. + let meta_core_url = std::env::var("META_CORE_URL").ok(); + self.cards = Arc::new(CardBook::new(build_http_client(10, USER_AGENT, None), meta_core_url)); if settings.nntmux_api_key.is_empty() { // Soft-skip, not a hard failure: serve /health, report Degraded. @@ -424,6 +557,15 @@ impl FeederPlugin for UsenetPlugin { query: &GatewayQuery, max_results: usize, ) -> Result, GatewayError> { + if self.nntmux.is_none() { + // Grab-only deployment (indexer keys, no nntmux): nothing to search. + return Ok(vec![]); + } + // A show page's query (`tmdbid: season: …`, no free text): match + // releases to the show BY NAME, against its card — see `crate::anchor`. + if let Some(anchor) = QueryAnchor::from_query(query) { + return self.handle_anchored(&anchor, max_results).await; + } // `free_text` is the bare-word leaves only — the structured filters // (contentKind:, fileType:) are the gateway's routing gate, not a search // term, and passing them to a LIKE would match nothing. @@ -431,47 +573,14 @@ impl FeederPlugin for UsenetPlugin { if term.trim().is_empty() { return Ok(vec![]); } - if self.nntmux.is_none() { - // Grab-only deployment (indexer keys, no nntmux): nothing to search. - return Ok(vec![]); - } let (api, _) = self.nntmux()?; let releases = api .search(term.trim(), max_results) .await .inspect_err(|e| self.note_problem(e))?; *self.nntmux_problem.write().unwrap() = None; - - // Mint each hit's cid here — see the `record_for` doc for why this - // cannot wait for compute. A release whose `.nzb` can't be had is - // DROPPED rather than surfaced uncid'd: an unplayable row in the - // client's "raw sources" list is worse than one fewer row. Once nntmux - // says a daily cap is spent, stop asking for this query. - let mut out = Vec::with_capacity(releases.len()); - let mut skipped = 0usize; - let mut capped = false; - for r in &releases { - if capped { - skipped += 1; - continue; - } - match self.posting(&r.guid).await { - Ok(p) => out.push(self.record_for(r, Some(&p.cid))), - Err(GatewayError::RateLimited { .. }) => { - capped = true; - skipped += 1; - warn!(target: "meta-feeder", "usenet: nntmux download cap reached — remaining hits skipped (raise the feeder user's role caps)"); - } - Err(e) => { - skipped += 1; - debug!(target: "meta-feeder", guid = %r.guid, error = %e, "usenet: release skipped (no usable .nzb)"); - } - } - } - debug!( - target: "meta-feeder", %term, hits = out.len(), skipped, - "usenet: nntmux search" - ); + let out = self.mint_records(&releases).await; + debug!(target: "meta-feeder", %term, hits = out.len(), "usenet: nntmux search"); Ok(out) } @@ -588,11 +697,17 @@ impl FeederPlugin for UsenetPlugin { ), F::secret("nntmux_api_key", "nntmux API key") .with_help( - "API key of the nntmux user this feeder searches as — the NNTmux app's \ - `metamesh` user, whose role has raised daily caps (see the NNTmux app's \ - install tips for how to read it). Blank → the self-scan search is off \ + "API key of the nntmux user this feeder searches as. Create one with \ + `docker exec nntmux-scanner nntmux-setup api-user metamesh` (raised daily \ + caps) and paste the key it prints. Blank → the self-scan search is off \ (indexer-key redeems still work).", ), + F::text("indexer_name", "Source name") + .with_help( + "How this feeder's releases are labelled in a client's source list, \ + beside names like nzbgeek.info (default: nntmux). Takes effect on the \ + next feeder restart.", + ), F::record_array( "indexers", "Indexer keys (redeem nzb-release)", @@ -619,7 +734,54 @@ impl FeederPlugin for UsenetPlugin { serde_json::json!({ "nntmux_url": self.settings.nntmux_url, "nntmux_api_key": self.settings.nntmux_api_key, + "indexer_name": self.settings.indexer_name, "indexers": self.settings.indexers, }) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn source_slug_keeps_a_leaf_safe_name() { + assert_eq!(source_slug("nntmux"), "nntmux"); + assert_eq!(source_slug("NNTmux (watch)"), "nntmux-watch"); + assert_eq!(source_slug("my:indexer"), "my-indexer"); + assert_eq!(source_slug(" "), "nntmux"); + } + + fn release(name: &str, category: i32) -> Release { + Release { + guid: "0f0e5b2c-1111-4c5e-9d6b-2a3b4c5d6e7f".into(), + search_name: name.into(), + category_id: category, + size: 1, + post_date: None, + } + } + + #[test] + fn records_carry_the_indexer_label_and_numbering() { + let mut p = UsenetPlugin::new(); + p.settings.indexer_name = "nntmux".into(); + let f = p.record_for(&release("Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL-NTb", 5040), None).fields; + assert_eq!(f["indexer"], "nntmux"); + assert_eq!(f["source/gateway:usenet-scan"], "true"); + assert_eq!(f["source/nntmux"], "true"); + assert_eq!(f["season"], "2"); + assert_eq!(f["episode"], "16"); + assert_eq!(f["contentKind"], "episode"); + + let pack = p.record_for(&release("Wonkas.The.Golden.Ticket.S01.1080p.WEB.H264-MIXED", 10), None).fields; + assert_eq!(pack["contentKind"], "pack"); + assert_eq!(pack["domain"], "screen"); + assert_eq!(pack["season"], "1"); + assert!(!pack.contains_key("episode")); + + let film = p.record_for(&release("Some.Film.2020.1080p.BluRay.x264-GRP", 2040), None).fields; + assert!(!film.contains_key("season")); + assert_eq!(film["contentKind"], "movie"); + } +} diff --git a/feeder-plugin/usenet-feeder/tests/anchored.rs b/feeder-plugin/usenet-feeder/tests/anchored.rs new file mode 100644 index 0000000..331d47f --- /dev/null +++ b/feeder-plugin/usenet-feeder/tests/anchored.rs @@ -0,0 +1,164 @@ +//! Show-page queries (`tmdbid: season:`, no free text): releases are +//! matched to the show BY NAME against its card in meta-core, and stamped as a +//! title match. Own binary because the card source is read from +//! `META_CORE_URL`, set once here. + +use std::collections::BTreeMap; + +use meta_feeder_sdk::hash::compute_card_cid; +use meta_feeder_sdk::serve::{configure_plugins, router}; +use meta_feeder_sdk::{GatewayQuery, QueryRequest, QueryResponse}; +use usenet_feeder::usenet::UsenetPlugin; +use wiremock::matchers::{method, path, query_param}; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +const KEY: &str = "metamesh-key"; + +fn item(guid: &str, title: &str, cat: u32) -> String { + format!( + r#"{title}http://localhost/details/{guid} +"# + ) +} + +fn rss(items: &[String]) -> String { + format!( + r#"{}"#, + items.join("") + ) +} + +const NZB: &str = r#" + + + alt.binaries.teevee + ttt@example.com + +"#; + +fn anchored(filters: &[(&str, &[&str])]) -> GatewayQuery { + let mut q = GatewayQuery::from_free_text(""); + q.filters = filters + .iter() + .map(|(k, vs)| (k.to_string(), vs.iter().map(|s| s.to_string()).collect())) + .collect::>(); + q +} + +async fn records(base: &str, q: GatewayQuery) -> Vec { + let r = reqwest::Client::new() + .post(format!("{base}/query")) + .json(&QueryRequest { upstream_id: "usenet".into(), query: q, max_results: 50 }) + .send() + .await + .expect("POST /query"); + assert!(r.status().is_success(), "query status {}", r.status()); + r.json::().await.expect("query json").records +} + +#[tokio::test] +async fn a_show_page_query_matches_releases_by_name_against_the_card() { + let nntmux = MockServer::start().await; + let core = MockServer::start().await; + std::env::set_var("META_CORE_URL", core.uri()); + + // The card the card feeder stored for tv 112442. + let locator = compute_card_cid("tmdb", "tv:112442").unwrap(); + Mock::given(method("GET")) + .and(path(format!("/meta/{locator}"))) + .respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ + "hashId": locator, + "metadata": { + "title": "Trash Truck", + "fileType": "card", + "poster": "bafkreiposter", + "description/eng": "Hank and his best pal, a giant trash truck.", + "imdbid": "tt9288860", + "genres/Kids": "true" + } + }))) + .mount(&core) + .await; + // No card for any other id. + Mock::given(method("GET")) + .respond_with(ResponseTemplate::new(404)) + .with_priority(10) + .mount(&core) + .await; + + Mock::given(method("GET")) + .and(path("/api/v1/api")) + .and(query_param("t", "caps")) + .respond_with(ResponseTemplate::new(200).set_body_string("")) + .mount(&nntmux) + .await; + Mock::given(method("GET")) + .and(path("/api/v1/api")) + .and(query_param("t", "search")) + .and(query_param("q", "Trash Truck")) + .respond_with(ResponseTemplate::new(200).set_body_string(rss(&[ + item("aaa111", "Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL.DDP5.1.H.264-NTb", 5040), + item("bbb222", "Trash.Truck.S01E02.Slumber.Party.1080p.NF.WEB-DL.DDP5.1.x264-LAZY", 5040), + item("ccc333", "Trash.Truck.Fan.Made.Toy.Unboxing.Compilation.Vol.3.1080p", 5040), + ]))) + .mount(&nntmux) + .await; + // Only the release that matches name AND season is ever downloaded. + Mock::given(method("GET")) + .and(path("/getnzb")) + .and(query_param("id", "aaa111.nzb")) + .respond_with(ResponseTemplate::new(200).set_body_string(NZB)) + .expect(1) + .mount(&nntmux) + .await; + Mock::given(method("GET")) + .and(path("/getnzb")) + .respond_with(ResponseTemplate::new(200).set_body_string(NZB)) + .with_priority(10) + .expect(0) + .mount(&nntmux) + .await; + + let dir = tempfile::tempdir().expect("tempdir"); + let cache = dir.path().join("gateway").join("usenet"); + std::fs::create_dir_all(&cache).expect("mkdir"); + std::fs::write( + cache.join("config.json"), + serde_json::json!({ "nntmux_url": nntmux.uri(), "nntmux_api_key": KEY }).to_string(), + ) + .expect("write config"); + let configured = configure_plugins(vec![Box::new(UsenetPlugin::new())], dir.path()).expect("configure"); + let app = router(configured, "test".to_string(), dir.path()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind"); + let base = format!("http://{}", listener.local_addr().expect("addr")); + tokio::spawn(async move { axum::serve(listener, app).await.expect("serve") }); + + // The season walk's shape: id + season + the TV kinds, no free text. + let recs = records( + &base, + anchored(&[ + ("tmdbid", &["112442"]), + ("season", &["2"]), + ("contentKind", &["series", "episode", "pack"]), + ]), + ) + .await; + assert_eq!(recs.len(), 1, "only the S02 release whose name is the show, got {recs:?}"); + let f = &recs[0].fields; + assert_eq!(recs[0].record_id, "aaa111"); + assert_eq!(f["tmdbid"], "112442"); + assert_eq!(f["anchored"], "true"); + assert_eq!(f["anchorMethod"], "title", "a name match, never an id-confirmed anchor"); + assert_eq!(f["imdbid"], "tt9288860"); + assert_eq!(f["poster"], "bafkreiposter"); + assert_eq!(f["description/eng"], "Hank and his best pal, a giant trash truck."); + assert!(!f.contains_key("genres/Kids")); + assert_eq!(f["season"], "2"); + assert_eq!(f["episode"], "16"); + assert_eq!(f["indexer"], "nntmux"); + assert_eq!(f["source/nntmux"], "true"); + assert_eq!(f["fileName"], "Trash.Truck.S02E16.Mint.Choco.Boom.1080p.NF.WEB-DL.DDP5.1.H.264-NTb"); + + // No card on the mesh for this id → nothing to match by name. + assert!(records(&base, anchored(&[("tmdbid", &["999"])])).await.is_empty()); +}