// resolve.rs — Tier 1 (rustypipe) → Tier 2 (yt-dlp -j fallback) // SPDX-License-Identifier: GPL-3.0-or-later use serde_json::Value; use crate::{run_yt_dlp, HandlerError}; /// Search YouTube via rustypipe. Returns `SearchResult` as JSON — /// the Python addon picks the fields it needs (id, name, channel, duration, /// thumbnail, view_count, …) to build a Kodi directory listing. pub(crate) async fn search(query: &str, limit: u32) -> Result { use rustypipe::client::RustyPipe; use rustypipe::model::VideoItem; let rp = RustyPipe::new(); let result = rp .query() .search::(query) .await .map_err(|e| classify_rustypipe_error(&e))?; // SearchResult.items is a Paginator — take the first // page truncated to `limit` items. let items_json: Vec = result .items .items .iter() .take(limit as usize) .filter_map(|v| serde_json::to_value(v).ok()) .collect(); tracing::info!(query, count = items_json.len(), "search ok via rustypipe"); Ok(serde_json::json!({ "source": "rustypipe", "query": query, "items": items_json, "corrected_query": result.corrected_query, })) } /// List a channel's recent videos. Returns the same VideoItem shape as /// `search`, plus channel metadata (name, subscribers, description, banner). pub(crate) async fn channel_videos(channel_id: &str, limit: u32) -> Result { use rustypipe::client::RustyPipe; let rp = RustyPipe::new(); let ch = rp .query() .channel_videos(channel_id) .await .map_err(|e| classify_rustypipe_error(&e))?; let items_json: Vec = ch .content .items .iter() .take(limit as usize) .filter_map(|v| serde_json::to_value(v).ok()) .collect(); tracing::info!(channel_id, count = items_json.len(), "channel_videos ok"); Ok(serde_json::json!({ "source": "rustypipe", "channel": { "id": ch.id, "name": ch.name, "description": ch.description, "subscribers": ch.subscriber_count, "video_count": ch.video_count, "avatar": ch.avatar, "banner": ch.banner, }, "items": items_json, })) } /// Fetch a subscriptions feed: pull the most recent N videos from each /// channel in parallel, merge + sort by publish-date newest-first, cap /// the returned union at `limit`. A channel that fails (404, region block, /// rustypipe error) is silently dropped — one dead subscription shouldn't /// kill the whole feed. pub(crate) async fn subscriptions_feed( channel_ids: &[String], per_channel: u32, limit: u32, ) -> Result { use rustypipe::client::RustyPipe; use std::collections::HashMap; let rp = std::sync::Arc::new(RustyPipe::new()); // Spawn one tokio task per channel; merge results once they all finish. // Fan-out is capped at 200 by main.rs; per_channel capped at 50. let tasks: Vec<_> = channel_ids .iter() .map(|cid| { let rp = rp.clone(); let cid = cid.clone(); tokio::spawn(async move { let res = rp.query().channel_videos(&cid).await; (cid, res) }) }) .collect(); let mut all_items: Vec = Vec::new(); let mut channel_names: HashMap = HashMap::new(); let mut failed: Vec = Vec::new(); for handle in tasks { let (cid, res) = match handle.await { Ok(t) => t, Err(e) => { tracing::warn!(error = %e, "subscriptions feed task panicked"); continue; } }; match res { Ok(ch) => { if !ch.name.is_empty() { channel_names.insert(cid.clone(), ch.name.clone()); } for vi in ch.content.items.iter().take(per_channel as usize) { if let Ok(v) = serde_json::to_value(vi) { all_items.push(v); } } } Err(e) => { tracing::warn!(channel = %cid, error = %e, "subscriptions feed channel failed"); failed.push(cid); } } } // Sort newest-first by publish_date. rustypipe VideoItem has an optional // `publish_date` as an RFC3339-ish string when known; fall back to // `publish_date_txt` (relative) ordering by string is meaningless, so we // keep channel-order as a stable secondary sort by leaving channel order // as insertion order and using a stable sort. all_items.sort_by(|a, b| { let ad = a.get("publish_date").and_then(Value::as_str).unwrap_or(""); let bd = b.get("publish_date").and_then(Value::as_str).unwrap_or(""); bd.cmp(ad) // descending }); all_items.truncate(limit as usize); tracing::info!( channels = channel_ids.len(), items = all_items.len(), failed = failed.len(), "subscriptions_feed ok" ); Ok(serde_json::json!({ "source": "rustypipe", "items": all_items, "channels_total": channel_ids.len(), "channels_failed": failed, "channel_names": channel_names, })) } /// List a playlist's videos. Returns the same VideoItem shape as search/channel. pub(crate) async fn playlist(playlist_id: &str, limit: u32) -> Result { use rustypipe::client::RustyPipe; let rp = RustyPipe::new(); let pl = rp .query() .playlist(playlist_id) .await .map_err(|e| classify_rustypipe_error(&e))?; let items_json: Vec = pl .videos .items .iter() .take(limit as usize) .filter_map(|v| serde_json::to_value(v).ok()) .collect(); tracing::info!(playlist_id, count = items_json.len(), "playlist ok"); Ok(serde_json::json!({ "source": "rustypipe", "playlist": { "id": pl.id, "name": pl.name, "description": pl.description, "video_count": pl.video_count, "channel": pl.channel, "thumbnail": pl.thumbnail, }, "items": items_json, })) } /// DASH-ready resolve: returns rustypipe's full `video_only_streams` + /// `audio_streams` arrays + `details`. The Python addon builds an MPD /// from these and hands it to inputstream.adaptive — unlocks 1080p+ via /// H.264 hardware decode on the RPi (vs the 360p ceiling on progressive). pub(crate) async fn resolve_dash(id: &str) -> Result { use rustypipe::client::RustyPipe; let rp = RustyPipe::new(); let player = rp .query() .player(id) .await .map_err(|e| classify_rustypipe_error(&e))?; let details_json = serde_json::to_value(&player.details) .map_err(|e| HandlerError::Internal(format!("serialize details: {e}")))?; let video_streams = serde_json::to_value(&player.video_only_streams) .map_err(|e| HandlerError::Internal(format!("serialize video_only_streams: {e}")))?; let audio_streams = serde_json::to_value(&player.audio_streams) .map_err(|e| HandlerError::Internal(format!("serialize audio_streams: {e}")))?; tracing::info!(id, "resolve_dash ok via rustypipe"); Ok(serde_json::json!({ "source": "rustypipe", "details": details_json, "video_only_streams": video_streams, "audio_streams": audio_streams, "expires_in_seconds": player.expires_in_seconds, })) } /// Playback-ready single-URL resolve. Asks yt-dlp for `best[ext=mp4]/best` — /// a combined audio+video format that Kodi can play as a plain HTTP URL. /// Slower than `resolve()` (~3-5s) but guarantees a working stream. pub(crate) async fn resolve_play(id: &str) -> Result { let url = format!("https://www.youtube.com/watch?v={id}"); // -f best[ext=mp4]/best — prefer mp4 progressive, else any best combined. // We use -j to get the full info dump; the selected format's URL appears // as the top-level "url" field. let stdout = run_yt_dlp(&[ "-j", "--no-warnings", "--no-playlist", "-f", "best[ext=mp4]/best", &url, ]) .await .map_err(|e| classify_yt_dlp_error(&e))?; let dump: Value = serde_json::from_slice(&stdout) .map_err(|e| HandlerError::Extractor(format!("yt-dlp json parse: {e}")))?; let stream_url = dump .get("url") .and_then(Value::as_str) .ok_or_else(|| HandlerError::Extractor("yt-dlp: no top-level url".into()))? .to_string(); tracing::info!(id, "resolve_play ok via yt-dlp combined"); Ok(serde_json::json!({ "source": "yt-dlp", "stream_url": stream_url, "title": dump.get("title"), "duration_s": dump.get("duration"), "channel_name": dump.get("channel"), "channel_id": dump.get("channel_id"), "thumbnail": dump.get("thumbnail"), "format_id": dump.get("format_id"), "ext": dump.get("ext"), })) } /// Metadata-rich resolve. Tries Tier 1 (rustypipe), falls back to Tier 2 (yt-dlp -j). /// Returns the full extractor response including separate audio + video streams. /// Use `resolve_play` instead for direct playback (returns a single combined URL). pub(crate) async fn resolve(id: &str) -> Result { match tier1_rustypipe(id).await { Ok(v) => { tracing::info!(id, source = "rustypipe", "resolve ok"); Ok(v) } Err(e) => { tracing::warn!(id, error = %e, "rustypipe failed; falling back to yt-dlp"); // Typed errors that mean "video can't be played by anyone" — don't retry yt-dlp, // it'll just hit the same wall. if matches!( e, HandlerError::AgeRestricted | HandlerError::PrivateVideo | HandlerError::NotFound ) { return Err(e); } tier2_yt_dlp(id).await } } } /// Tier 1 — native rustypipe. Serializes the whole player.details + selected streams as /// opaque pass-through JSON. The Python addon parses the fields it needs; this keeps us /// resilient to rustypipe shape evolution and unblocks tier-2 normalization later. async fn tier1_rustypipe(id: &str) -> Result { use rustypipe::client::RustyPipe; use rustypipe::param::StreamFilter; let rp = RustyPipe::new(); let player = rp .query() .player(id) .await .map_err(|e| classify_rustypipe_error(&e))?; let (video, audio) = player.select_video_audio_stream(&StreamFilter::default()); let details_json = serde_json::to_value(&player.details) .map_err(|e| HandlerError::Internal(format!("serialize details: {e}")))?; let video_json = video .map(|v| serde_json::to_value(v)) .transpose() .map_err(|e| HandlerError::Internal(format!("serialize video: {e}")))? .unwrap_or(Value::Null); let audio_json = audio .map(|a| serde_json::to_value(a)) .transpose() .map_err(|e| HandlerError::Internal(format!("serialize audio: {e}")))? .unwrap_or(Value::Null); Ok(serde_json::json!({ "source": "rustypipe", "details": details_json, "video_stream": video_json, "audio_stream": audio_json, })) } /// Classify a yt-dlp shell-out error into one of our typed handler errors. /// yt-dlp's stderr is freeform English; we match on **word-boundary** patterns /// so "private" matches the standalone word, not e.g. "private network" inside /// a TLS error. Preserve the original message verbatim in the returned error. fn classify_yt_dlp_error(e: &anyhow::Error) -> HandlerError { let original = e.to_string(); let lower = original.to_lowercase(); if matches_word(&lower, &["age-restrict", "age restricted", "age restriction"]) { HandlerError::AgeRestricted } else if matches_word(&lower, &["private video", "video is private"]) { HandlerError::PrivateVideo } else if lower.contains("not available") || lower.contains("does not exist") { HandlerError::NotFound } else if matches_word(&lower, &["geo-restrict", "geo restrict", "region-restrict", "region restrict"]) { HandlerError::RegionBlocked } else { HandlerError::Extractor(original) } } /// Classify a rustypipe error into one of our typed handler errors. /// rustypipe's error enum varies by version; we match on the Display string for resilience. fn classify_rustypipe_error(e: &dyn std::fmt::Display) -> HandlerError { let original = e.to_string(); let msg = original.to_lowercase(); if matches_word(&msg, &["age-restrict", "age restricted", "age restriction"]) { HandlerError::AgeRestricted } else if matches_word(&msg, &["region-restrict", "region restrict", "geo-restrict", "geo restrict", "country restrict"]) { HandlerError::RegionBlocked } else if matches_word(&msg, &["private video", "video is private"]) { HandlerError::PrivateVideo } else if matches_word(&msg, &["not found", "unavailable", "does not exist"]) { HandlerError::NotFound } else if matches_word(&msg, &["network error", "timeout", "connection refused", "dns error"]) { HandlerError::Network(original) } else { HandlerError::Extractor(original) } } fn matches_word(haystack: &str, needles: &[&str]) -> bool { needles.iter().any(|n| haystack.contains(n)) } /// Tier 2 — shell out to yt-dlp -j. async fn tier2_yt_dlp(id: &str) -> Result { let url = format!("https://www.youtube.com/watch?v={id}"); let stdout = run_yt_dlp(&["-j", "--no-warnings", "--no-playlist", &url]) .await .map_err(|e| classify_yt_dlp_error(&e))?; let dump: Value = serde_json::from_slice(&stdout) .map_err(|e| HandlerError::Extractor(format!("yt-dlp json parse: {e}")))?; // yt-dlp's JSON has a `formats` array. We pass it through largely as-is — the addon // can pick what inputstream.adaptive wants. Shape it to match our protocol. let streams: Vec = dump .get("formats") .and_then(Value::as_array) .cloned() .unwrap_or_default() .into_iter() .filter_map(|f| { let url = f.get("url")?.as_str()?.to_string(); let vcodec = f.get("vcodec").and_then(Value::as_str).unwrap_or("none"); let acodec = f.get("acodec").and_then(Value::as_str).unwrap_or("none"); let is_audio_only = vcodec == "none" && acodec != "none"; let is_video_only = vcodec != "none" && acodec == "none"; Some(serde_json::json!({ "url": url, "itag": f.get("format_id").and_then(|v| v.as_str()).and_then(|s| s.parse::().ok()), "mime": f.get("ext"), "width": f.get("width"), "height": f.get("height"), "bitrate": f.get("tbr"), "is_audio_only": is_audio_only, "is_video_only": is_video_only, })) }) .collect(); Ok(serde_json::json!({ "source": "yt-dlp", "title": dump.get("title"), "duration_s": dump.get("duration"), "channel_name": dump.get("channel"), "channel_id": dump.get("channel_id"), "streams": streams, })) }