1056 lines
34 KiB
Rust
1056 lines
34 KiB
Rust
use std::{
|
|
collections::{BTreeSet, VecDeque},
|
|
ffi::OsStr,
|
|
fs::{self, OpenOptions},
|
|
io::{self, Read},
|
|
path::PathBuf,
|
|
process::{Child, Command, Stdio},
|
|
sync::mpsc::{self, Receiver, TryRecvError},
|
|
thread,
|
|
};
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
|
|
mod catalog;
|
|
|
|
pub const PAGE_SIZE: usize = 60;
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
pub struct FeedRequest {
|
|
pub channel: usize,
|
|
pub start: usize,
|
|
pub count: usize,
|
|
}
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
|
pub struct FeedProgress {
|
|
pub request: FeedRequest,
|
|
/// Raw rows, including unavailable, invalid and duplicate entries.
|
|
pub count: usize,
|
|
}
|
|
|
|
#[derive(Clone, Debug, Default, PartialEq, Eq)]
|
|
pub struct FeedCursor {
|
|
// None before planning; an exhausted channel has no next offset.
|
|
positions: Option<Vec<Option<usize>>>,
|
|
rotation: usize,
|
|
}
|
|
|
|
impl FeedCursor {
|
|
/// Planning is repeatable until a complete successful page is accepted.
|
|
pub fn plan(&mut self, channels: usize) -> io::Result<Vec<FeedRequest>> {
|
|
let positions = self
|
|
.positions
|
|
.get_or_insert_with(|| vec![Some(1); channels]);
|
|
if positions.len() != channels {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Feed channel membership changed without resetting its cursor",
|
|
));
|
|
}
|
|
self.planned().map(|(requests, _)| requests)
|
|
}
|
|
|
|
fn planned(&self) -> io::Result<(Vec<FeedRequest>, usize)> {
|
|
let Some(positions) = &self.positions else {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Feed is not planned",
|
|
));
|
|
};
|
|
let active = positions
|
|
.iter()
|
|
.filter(|position| position.is_some())
|
|
.count();
|
|
if active == 0 {
|
|
return Ok((Vec::new(), self.rotation));
|
|
}
|
|
let selected = active.min(PAGE_SIZE);
|
|
let mut requests = Vec::with_capacity(selected);
|
|
let mut next_rotation = self.rotation;
|
|
for step in 0..positions.len() {
|
|
let channel = (self.rotation + step) % positions.len();
|
|
let Some(start) = positions[channel] else {
|
|
continue;
|
|
};
|
|
let index = requests.len();
|
|
let count = PAGE_SIZE / selected + usize::from(index < PAGE_SIZE % selected);
|
|
range_end(start, count)?;
|
|
requests.push(FeedRequest {
|
|
channel,
|
|
start,
|
|
count,
|
|
});
|
|
if index == (PAGE_SIZE - 1) % selected {
|
|
next_rotation = (channel + 1) % positions.len();
|
|
}
|
|
if requests.len() == selected {
|
|
break;
|
|
}
|
|
}
|
|
requests.sort_unstable_by_key(|request| (request.start, request.count, request.channel));
|
|
Ok((requests, next_rotation))
|
|
}
|
|
|
|
pub fn accept(&mut self, progress: &[FeedProgress]) -> Result<(), String> {
|
|
let (requests, rotation) = self.planned().map_err(|error| error.to_string())?;
|
|
if progress.len() != requests.len() {
|
|
return Err("Feed progress did not match the planned channels".into());
|
|
}
|
|
// Preflight every request and offset before changing any channel or the rotation.
|
|
for (entry, request) in progress.iter().zip(&requests) {
|
|
if entry.request != *request || entry.count > request.count {
|
|
return Err("Feed progress did not match the planned request".into());
|
|
}
|
|
if entry.count == request.count && request.start.checked_add(request.count).is_none() {
|
|
return Err("Feed channel offset overflow".into());
|
|
}
|
|
}
|
|
let positions = self.positions.as_mut().unwrap();
|
|
for entry in progress {
|
|
positions[entry.request.channel] = if entry.count < entry.request.count {
|
|
None
|
|
} else {
|
|
Some(entry.request.start + entry.request.count)
|
|
};
|
|
}
|
|
self.rotation = rotation;
|
|
Ok(())
|
|
}
|
|
|
|
/// Unrequested channels remain active until their own response proves exhaustion.
|
|
pub fn has_more(&self) -> bool {
|
|
self.positions
|
|
.as_ref()
|
|
.is_some_and(|positions| positions.iter().any(Option::is_some))
|
|
}
|
|
}
|
|
|
|
pub struct Page {
|
|
pub videos: Vec<Video>,
|
|
pub has_more: bool,
|
|
pub feed_progress: Vec<FeedProgress>,
|
|
}
|
|
|
|
#[derive(Clone, Deserialize, Serialize)]
|
|
pub struct Video {
|
|
pub id: String,
|
|
pub title: String,
|
|
pub channel: Option<String>,
|
|
pub channel_id: Option<String>,
|
|
pub uploader_id: Option<String>,
|
|
pub timestamp: Option<i64>,
|
|
pub duration: Option<f64>,
|
|
/// Cached YYYY-MM-DD label; a leading ~ marks an estimated UTC date.
|
|
#[serde(default, rename = "upload_date")]
|
|
pub publication_date: Option<String>,
|
|
// Listing metadata only; offline collections keep their existing format.
|
|
#[serde(skip_serializing)]
|
|
pub availability: Option<Availability>,
|
|
}
|
|
|
|
#[derive(Clone, Copy, Deserialize)]
|
|
pub enum Availability {
|
|
#[serde(rename = "subscriber_only")]
|
|
MembersOnly,
|
|
#[serde(other)]
|
|
Other,
|
|
}
|
|
|
|
fn listed_video(entry: &serde_json::value::RawValue) -> Option<Video> {
|
|
let video: Video = serde_json::from_str(entry.get()).ok()?;
|
|
(!matches!(video.availability, Some(Availability::MembersOnly))).then_some(video)
|
|
}
|
|
|
|
#[derive(Deserialize)]
|
|
struct SearchResults<'a> {
|
|
// Borrow raw rows so one schema error cannot abort siblings or copy unused metadata.
|
|
#[serde(borrow)]
|
|
entries: Vec<&'a serde_json::value::RawValue>,
|
|
channel: Option<String>,
|
|
channel_id: Option<String>,
|
|
uploader_id: Option<String>,
|
|
}
|
|
|
|
pub fn search_results(output: &str) -> Result<Page, String> {
|
|
let results: SearchResults = serde_json::from_str(output)
|
|
.map_err(|error| format!("Invalid search response: {error}"))?;
|
|
let has_more = results.entries.len() >= PAGE_SIZE;
|
|
let videos = results
|
|
.entries
|
|
.into_iter()
|
|
.filter_map(|entry| {
|
|
let mut video = listed_video(entry)?;
|
|
validate_video(&mut video).ok()?;
|
|
Some(video)
|
|
})
|
|
.collect();
|
|
Ok(Page {
|
|
videos,
|
|
has_more,
|
|
feed_progress: Vec::new(),
|
|
})
|
|
}
|
|
|
|
fn valid_channel_id(id: &str) -> bool {
|
|
id.len() == 24
|
|
&& id.starts_with("UC")
|
|
&& id
|
|
.bytes()
|
|
.all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
|
|
}
|
|
|
|
pub(crate) fn valid_video_id(id: &str) -> bool {
|
|
id.len() == 11
|
|
&& id
|
|
.bytes()
|
|
.all(|c| c.is_ascii_alphanumeric() || c == b'-' || c == b'_')
|
|
}
|
|
|
|
pub(crate) fn validate_video(video: &mut Video) -> Result<(), String> {
|
|
if !valid_video_id(&video.id) {
|
|
return Err("an invalid YouTube video ID".into());
|
|
}
|
|
if video
|
|
.channel_id
|
|
.as_deref()
|
|
.is_some_and(|id| !valid_channel_id(id))
|
|
{
|
|
return Err("an invalid YouTube channel ID".into());
|
|
}
|
|
if video
|
|
.uploader_id
|
|
.as_deref()
|
|
.is_some_and(|handle| handle.chars().any(char::is_control))
|
|
{
|
|
return Err("an invalid YouTube channel handle".into());
|
|
}
|
|
video.title.retain(|c| !c.is_control());
|
|
if let Some(channel) = &mut video.channel {
|
|
channel.retain(|c| !c.is_control());
|
|
}
|
|
let exact_date = video.publication_date.as_deref().and_then(|date| {
|
|
if date.len() != 8 || !date.bytes().all(|byte| byte.is_ascii_digit()) {
|
|
return None;
|
|
}
|
|
time::Date::from_calendar_date(
|
|
date[..4].parse().ok()?,
|
|
date[4..6].parse::<u8>().ok()?.try_into().ok()?,
|
|
date[6..].parse().ok()?,
|
|
)
|
|
.ok()
|
|
});
|
|
video.publication_date = exact_date.map(|date| date.to_string()).or_else(|| {
|
|
video.timestamp.and_then(|timestamp| {
|
|
time::OffsetDateTime::from_unix_timestamp(timestamp)
|
|
.ok()
|
|
.map(|date| format!("~{}", date.date()))
|
|
})
|
|
});
|
|
Ok(())
|
|
}
|
|
|
|
pub fn feed_results(output: &str, requests: &[FeedRequest]) -> Result<Page, String> {
|
|
validate_feed_requests(requests).map_err(|error| error.to_string())?;
|
|
let mut videos = Vec::new();
|
|
let mut feed_progress = Vec::with_capacity(requests.len());
|
|
let mut playlists = serde_json::Deserializer::from_str(output).into_iter::<SearchResults>();
|
|
// Each URL produces one playlist envelope; grouped command output follows request order.
|
|
for &request in requests {
|
|
let playlist = playlists
|
|
.next()
|
|
.ok_or_else(|| "Feed returned fewer playlists than requested".to_owned())?
|
|
.map_err(|error| format!("Invalid feed response: {error}"))?;
|
|
if playlist
|
|
.channel_id
|
|
.as_deref()
|
|
.is_some_and(|id| !valid_channel_id(id))
|
|
{
|
|
return Err("Feed returned an invalid YouTube channel ID".into());
|
|
}
|
|
let count = playlist.entries.len();
|
|
if count > request.count {
|
|
return Err("Feed returned more rows than requested".into());
|
|
}
|
|
feed_progress.push(FeedProgress { request, count });
|
|
for entry in playlist.entries {
|
|
let Some(mut video) = listed_video(entry) else {
|
|
continue;
|
|
};
|
|
if video.channel_id.is_none() {
|
|
video.channel_id = playlist.channel_id.clone();
|
|
}
|
|
if video.uploader_id.is_none() {
|
|
video.uploader_id = playlist.uploader_id.clone();
|
|
}
|
|
if video.channel.is_none() {
|
|
video.channel = playlist.channel.clone();
|
|
}
|
|
if validate_video(&mut video).is_ok() {
|
|
videos.push(video);
|
|
}
|
|
}
|
|
}
|
|
if let Some(extra) = playlists.next() {
|
|
extra.map_err(|error| format!("Invalid feed response: {error}"))?;
|
|
return Err("Feed returned more playlists than requested".into());
|
|
}
|
|
// Keep the newest occurrence of each ID without copying IDs into a second collection.
|
|
videos.sort_unstable_by(|a, b| a.id.cmp(&b.id).then_with(|| b.timestamp.cmp(&a.timestamp)));
|
|
videos.dedup_by(|a, b| a.id == b.id);
|
|
videos.sort_by(|a, b| b.timestamp.cmp(&a.timestamp));
|
|
Ok(Page {
|
|
videos,
|
|
has_more: feed_progress
|
|
.iter()
|
|
.any(|entry| entry.count == entry.request.count),
|
|
feed_progress,
|
|
})
|
|
}
|
|
|
|
pub fn full_description(output: &str, expected_id: &str) -> Result<String, String> {
|
|
#[derive(Deserialize)]
|
|
struct Details {
|
|
id: String,
|
|
description: Option<String>,
|
|
}
|
|
let details: Details =
|
|
serde_json::from_str(output).map_err(|error| format!("Invalid video metadata: {error}"))?;
|
|
if details.id != expected_id {
|
|
return Err("Video metadata did not match the requested video".into());
|
|
}
|
|
let mut description = details.description.unwrap_or_default();
|
|
description.retain(|c| !c.is_control() || c == '\n' || c == '\t');
|
|
Ok(description)
|
|
}
|
|
|
|
pub fn channel_handle(output: &str, expected_id: &str) -> Result<String, String> {
|
|
#[derive(Deserialize)]
|
|
struct Channel {
|
|
channel_id: String,
|
|
uploader_id: Option<String>,
|
|
}
|
|
let channel: Channel = serde_json::from_str(output)
|
|
.map_err(|error| format!("Invalid channel metadata: {error}"))?;
|
|
if channel.channel_id != expected_id {
|
|
return Err("Channel metadata did not match the requested channel".into());
|
|
}
|
|
let handle = channel
|
|
.uploader_id
|
|
.filter(|handle| handle.starts_with('@'))
|
|
.ok_or_else(|| "This channel has no available handle; no channel was saved.".to_owned())?;
|
|
crate::config::normalize_channel(&handle).map_err(|error| error.to_string())?;
|
|
Ok(handle)
|
|
}
|
|
|
|
fn page_end(start: usize) -> io::Result<usize> {
|
|
range_end(start, PAGE_SIZE)
|
|
}
|
|
|
|
fn range_end(start: usize, count: usize) -> io::Result<usize> {
|
|
count
|
|
.checked_sub(1)
|
|
.and_then(|offset| start.checked_add(offset))
|
|
.filter(|_| start != 0)
|
|
.ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "Invalid playlist page range"))
|
|
}
|
|
|
|
fn validate_feed_requests(requests: &[FeedRequest]) -> io::Result<()> {
|
|
let mut remaining = PAGE_SIZE;
|
|
if requests.is_empty() {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"No channels to request",
|
|
));
|
|
}
|
|
for (index, request) in requests.iter().enumerate() {
|
|
range_end(request.start, request.count)?;
|
|
remaining = remaining.checked_sub(request.count).ok_or_else(|| {
|
|
io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Feed request exceeds 60 total rows",
|
|
)
|
|
})?;
|
|
if requests[..index]
|
|
.iter()
|
|
.any(|prior| prior.channel == request.channel)
|
|
{
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Duplicate feed channel",
|
|
));
|
|
}
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
const MAX_EXTRACTORS: usize = 4;
|
|
|
|
// Own running and queued extractors so replacing a request cancels the whole page.
|
|
pub struct Job {
|
|
children: Vec<Extraction>,
|
|
queued: VecDeque<Command>,
|
|
feed: Option<catalog::FeedExpansion>,
|
|
}
|
|
|
|
struct Extraction {
|
|
child: Option<Child>,
|
|
output: Receiver<io::Result<(String, String)>>,
|
|
status: Option<std::process::ExitStatus>,
|
|
completed: Option<String>,
|
|
}
|
|
|
|
fn encode_query(query: &str) -> String {
|
|
const HEX: &[u8; 16] = b"0123456789ABCDEF";
|
|
let unreserved =
|
|
|byte: u8| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'.' | b'_' | b'~');
|
|
let capacity = query
|
|
.bytes()
|
|
.map(|byte| if unreserved(byte) { 1 } else { 3 })
|
|
.sum();
|
|
let mut encoded = String::with_capacity(capacity);
|
|
for byte in query.bytes() {
|
|
if unreserved(byte) {
|
|
encoded.push(char::from(byte));
|
|
} else {
|
|
encoded.push('%');
|
|
encoded.push(char::from(HEX[usize::from(byte >> 4)]));
|
|
encoded.push(char::from(HEX[usize::from(byte & 0x0f)]));
|
|
}
|
|
}
|
|
encoded
|
|
}
|
|
|
|
fn feed_commands(
|
|
channels: &[String],
|
|
requests: &[FeedRequest],
|
|
query: &str,
|
|
) -> io::Result<Vec<Command>> {
|
|
validate_feed_requests(requests)?;
|
|
let query = (!query.is_empty()).then(|| encode_query(query));
|
|
let mut commands = Vec::new();
|
|
let mut bounds = None;
|
|
let batch_size = requests.len().div_ceil(MAX_EXTRACTORS);
|
|
let mut batch_len = 0;
|
|
for (index, request) in requests.iter().enumerate() {
|
|
let channel = channels.get(request.channel).ok_or_else(|| {
|
|
io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Feed channel index is out of range",
|
|
)
|
|
})?;
|
|
if requests[..index]
|
|
.iter()
|
|
.any(|prior| &channels[prior.channel] == channel)
|
|
{
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Duplicate feed channel URL",
|
|
));
|
|
}
|
|
if bounds != Some((request.start, request.count)) || batch_len == batch_size {
|
|
let mut command = extractor();
|
|
command.args([
|
|
"--flat-playlist",
|
|
"--playlist-start",
|
|
&request.start.to_string(),
|
|
"--playlist-end",
|
|
&range_end(request.start, request.count)?.to_string(),
|
|
"--extractor-args",
|
|
"youtubetab:approximate_date",
|
|
"--dump-single-json",
|
|
]);
|
|
commands.push(command);
|
|
bounds = Some((request.start, request.count));
|
|
batch_len = 0;
|
|
}
|
|
commands.last_mut().unwrap().arg(match &query {
|
|
Some(query) => format!("{channel}/search?query={query}"),
|
|
None => format!("{channel}/videos"),
|
|
});
|
|
batch_len += 1;
|
|
}
|
|
Ok(commands)
|
|
}
|
|
|
|
impl Job {
|
|
pub fn search(query: &str, start: usize) -> io::Result<Self> {
|
|
let end = page_end(start)?;
|
|
Self::spawn(extractor().args([
|
|
"--flat-playlist",
|
|
"--playlist-start",
|
|
&start.to_string(),
|
|
"--playlist-end",
|
|
&end.to_string(),
|
|
"--extractor-args",
|
|
"youtubetab:approximate_date",
|
|
"--dump-single-json",
|
|
&format!("ytsearchall:{query}"),
|
|
]))
|
|
}
|
|
|
|
pub fn feed(channels: &[String], requests: &[FeedRequest], query: &str) -> io::Result<Self> {
|
|
let mut job = Self::spawn_all(feed_commands(channels, requests, query)?)?;
|
|
if query.is_empty() {
|
|
job.feed = Some(catalog::FeedExpansion::new(channels, requests));
|
|
}
|
|
Ok(job)
|
|
}
|
|
|
|
pub fn channel(id: &str) -> io::Result<Self> {
|
|
if !valid_channel_id(id) {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Invalid YouTube channel ID",
|
|
));
|
|
}
|
|
Self::spawn(extractor().args([
|
|
"--flat-playlist",
|
|
"--playlist-end",
|
|
"1",
|
|
"--dump-single-json",
|
|
&format!("https://www.youtube.com/channel/{id}/videos"),
|
|
]))
|
|
}
|
|
|
|
pub fn description(id: &str) -> io::Result<Self> {
|
|
Self::spawn(extractor().args([
|
|
"--no-playlist",
|
|
"--skip-download",
|
|
"--dump-single-json",
|
|
&format!("https://www.youtube.com/watch?v={id}"),
|
|
]))
|
|
}
|
|
|
|
fn spawn(command: &mut Command) -> io::Result<Self> {
|
|
Ok(Self {
|
|
children: vec![Extraction::spawn(command)?],
|
|
queued: VecDeque::new(),
|
|
feed: None,
|
|
})
|
|
}
|
|
|
|
fn spawn_all(commands: impl IntoIterator<Item = Command>) -> io::Result<Self> {
|
|
let mut job = Self {
|
|
children: Vec::new(),
|
|
queued: VecDeque::new(),
|
|
feed: None,
|
|
};
|
|
// Dropping job also cancels started children if any initial spawn fails.
|
|
for mut command in commands {
|
|
if job.children.len() < MAX_EXTRACTORS {
|
|
job.children.push(Extraction::spawn(&mut command)?);
|
|
} else {
|
|
job.queued.push_back(command);
|
|
}
|
|
}
|
|
Ok(job)
|
|
}
|
|
|
|
pub fn poll(&mut self) -> Option<Result<String, String>> {
|
|
if let Some(mut feed) = self.feed.take() {
|
|
let result = feed.poll(self);
|
|
if result.is_none() {
|
|
self.feed = Some(feed);
|
|
}
|
|
return result;
|
|
}
|
|
self.poll_extractions()
|
|
}
|
|
|
|
fn poll_extractions(&mut self) -> Option<Result<String, String>> {
|
|
if self.children.is_empty() {
|
|
return None;
|
|
}
|
|
let mut active = 0;
|
|
for child in &mut self.children {
|
|
if child.completed.is_some() {
|
|
continue;
|
|
}
|
|
match child.poll() {
|
|
None => active += 1,
|
|
Some(Ok(output)) => child.completed = Some(output),
|
|
Some(Err(error)) => {
|
|
self.queued.clear();
|
|
self.children.clear();
|
|
return Some(Err(error));
|
|
}
|
|
}
|
|
}
|
|
while active < MAX_EXTRACTORS {
|
|
let Some(mut command) = self.queued.pop_front() else {
|
|
break;
|
|
};
|
|
match Extraction::spawn(&mut command) {
|
|
Ok(child) => {
|
|
self.children.push(child);
|
|
active += 1;
|
|
}
|
|
Err(error) => {
|
|
self.queued.clear();
|
|
self.children.clear();
|
|
return Some(Err(error.to_string()));
|
|
}
|
|
}
|
|
}
|
|
if active != 0 {
|
|
return None;
|
|
}
|
|
let mut children = self.children.drain(..);
|
|
// Retain command order even when a queued group completes ahead of an earlier one.
|
|
let mut output = children.next()?.completed.take().unwrap();
|
|
output.reserve(
|
|
children
|
|
.as_slice()
|
|
.iter()
|
|
.map(|child| child.completed.as_ref().unwrap().len() + 1)
|
|
.sum(),
|
|
);
|
|
for mut child in children {
|
|
let next = child.completed.take().unwrap();
|
|
if !output.is_empty() && !next.is_empty() {
|
|
output.push('\n');
|
|
}
|
|
output.push_str(&next);
|
|
}
|
|
Some(Ok(output))
|
|
}
|
|
}
|
|
|
|
impl Extraction {
|
|
fn spawn(command: &mut Command) -> io::Result<Self> {
|
|
let mut child = command
|
|
.stdin(Stdio::null())
|
|
.stdout(Stdio::piped())
|
|
.stderr(Stdio::piped())
|
|
.spawn()
|
|
.map_err(|error| {
|
|
io::Error::new(error.kind(), format!("Cannot start yt-dlp: {error}"))
|
|
})?;
|
|
let mut stdout = child.stdout.take().expect("stdout is piped");
|
|
let mut stderr = child.stderr.take().expect("stderr is piped");
|
|
let (sender, output) = mpsc::channel();
|
|
// Drain both pipes while the process runs; a full pipe must not stall the TUI.
|
|
thread::spawn(move || {
|
|
let stdout_reader = thread::spawn(move || {
|
|
let mut text = String::new();
|
|
stdout.read_to_string(&mut text).map(|_| text)
|
|
});
|
|
let mut errors = String::new();
|
|
let stderr_result = stderr.read_to_string(&mut errors);
|
|
let result = stdout_reader
|
|
.join()
|
|
.unwrap_or_else(|_| Err(io::Error::other("Output reader panicked")))
|
|
.and_then(|text| stderr_result.map(|_| (text, errors)));
|
|
let _ = sender.send(result);
|
|
});
|
|
Ok(Self {
|
|
child: Some(child),
|
|
output,
|
|
status: None,
|
|
completed: None,
|
|
})
|
|
}
|
|
|
|
fn poll(&mut self) -> Option<Result<String, String>> {
|
|
if self.status.is_none() {
|
|
self.status = match self.child.as_mut()?.try_wait() {
|
|
Ok(status) => status,
|
|
Err(error) => return Some(Err(format!("Cannot wait for yt-dlp: {error}"))),
|
|
};
|
|
}
|
|
let status = self.status?;
|
|
if !status.success() {
|
|
// Do not wait for inherited pipes held open by a failed child's descendants.
|
|
self.cancel();
|
|
}
|
|
let result = match self.output.try_recv() {
|
|
Ok(Ok((output, _))) if status.success() => Some(Ok(output)),
|
|
Ok(Ok((_, errors))) => Some(Err(if errors.trim().is_empty() {
|
|
format!("yt-dlp exited with {status}")
|
|
} else {
|
|
errors.trim().to_owned()
|
|
})),
|
|
Ok(Err(error)) => Some(Err(format!("Cannot read yt-dlp output: {error}"))),
|
|
Err(TryRecvError::Empty) => None,
|
|
Err(TryRecvError::Disconnected) => {
|
|
Some(Err("yt-dlp output reader disconnected".into()))
|
|
}
|
|
};
|
|
if result.is_some() {
|
|
self.cancel();
|
|
}
|
|
result
|
|
}
|
|
|
|
fn cancel(&mut self) {
|
|
if let Some(mut child) = self.child.take() {
|
|
// The leader may have exited while descendants still own its output pipes.
|
|
// SAFETY: every extraction owns an isolated process group.
|
|
#[cfg(unix)]
|
|
unsafe {
|
|
libc::kill(-(child.id() as i32), libc::SIGKILL);
|
|
}
|
|
stop(&mut child);
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Drop for Extraction {
|
|
fn drop(&mut self) {
|
|
self.cancel();
|
|
}
|
|
}
|
|
|
|
const MAX_THUMBNAIL_BYTES: usize = 2 * 1024 * 1024;
|
|
|
|
pub struct ThumbnailJob {
|
|
child: Child,
|
|
output: Receiver<io::Result<(Vec<u8>, String)>>,
|
|
}
|
|
|
|
impl ThumbnailJob {
|
|
pub fn start(id: &str) -> io::Result<Self> {
|
|
if !valid_video_id(id) {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Invalid YouTube video ID",
|
|
));
|
|
}
|
|
Self::from_url(&format!("https://i.ytimg.com/vi/{id}/hqdefault.jpg"))
|
|
}
|
|
|
|
pub fn from_url(url: &str) -> io::Result<Self> {
|
|
let uri: ureq::http::Uri = url
|
|
.parse()
|
|
.map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "Invalid artwork URL"))?;
|
|
if uri.scheme_str() != Some("https")
|
|
|| uri.host().is_none_or(str::is_empty)
|
|
|| uri
|
|
.authority()
|
|
.is_some_and(|authority| authority.as_str().contains('@'))
|
|
|| url.chars().any(char::is_control)
|
|
{
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Artwork URLs must be absolute HTTPS without credentials",
|
|
));
|
|
}
|
|
let mut child = isolated_command(OsStr::new("curl"))
|
|
.args([
|
|
"--disable",
|
|
"--silent",
|
|
"--show-error",
|
|
"--fail",
|
|
"--location",
|
|
"--globoff",
|
|
"--proto",
|
|
"=https",
|
|
"--proto-redir",
|
|
"=https",
|
|
"--connect-timeout",
|
|
"5",
|
|
"--max-time",
|
|
"15",
|
|
"--max-filesize",
|
|
"2097152", // 2 MiB; read_thumbnail also enforces the limit.
|
|
])
|
|
.arg(url)
|
|
.stdin(Stdio::null())
|
|
.stdout(Stdio::piped())
|
|
.stderr(Stdio::piped())
|
|
.spawn()
|
|
.map_err(|error| io::Error::new(error.kind(), format!("Cannot start curl: {error}")))?;
|
|
let stdout = child.stdout.take().expect("stdout is piped");
|
|
let errors = read_errors(child.stderr.take().expect("stderr is piped"));
|
|
let (sender, output) = mpsc::channel();
|
|
thread::spawn(move || {
|
|
let bytes = read_thumbnail(stdout);
|
|
let result = errors
|
|
.recv()
|
|
.unwrap_or_else(|_| Err(io::Error::other("curl error reader disconnected")))
|
|
.and_then(|errors| bytes.map(|bytes| (bytes, errors)));
|
|
let _ = sender.send(result);
|
|
});
|
|
Ok(Self { child, output })
|
|
}
|
|
|
|
pub fn poll(&mut self) -> Option<Result<Vec<u8>, String>> {
|
|
// EOF alone is not success: curl can write a partial body and then fail.
|
|
let status = match self.child.try_wait() {
|
|
Ok(Some(status)) => status,
|
|
Ok(None) => return None,
|
|
Err(error) => return Some(Err(format!("Cannot wait for curl: {error}"))),
|
|
};
|
|
match self.output.try_recv() {
|
|
Ok(Ok((bytes, _))) if status.success() => Some(Ok(bytes)),
|
|
Ok(Ok((_, errors))) => Some(Err(if errors.trim().is_empty() {
|
|
format!("curl exited with {status}")
|
|
} else {
|
|
errors.trim().to_owned()
|
|
})),
|
|
Ok(Err(error)) => Some(Err(format!("Cannot read thumbnail: {error}"))),
|
|
Err(TryRecvError::Empty) => None,
|
|
Err(TryRecvError::Disconnected) => Some(Err("curl output reader disconnected".into())),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Drop for ThumbnailJob {
|
|
fn drop(&mut self) {
|
|
stop(&mut self.child);
|
|
}
|
|
}
|
|
|
|
fn read_thumbnail(pipe: impl Read) -> io::Result<Vec<u8>> {
|
|
let mut bytes = Vec::new();
|
|
pipe.take((MAX_THUMBNAIL_BYTES + 1) as u64)
|
|
.read_to_end(&mut bytes)?;
|
|
if bytes.len() > MAX_THUMBNAIL_BYTES {
|
|
return Err(io::Error::other("Thumbnail exceeds 2 MiB"));
|
|
}
|
|
Ok(bytes)
|
|
}
|
|
|
|
fn isolated_command(program: &OsStr) -> Command {
|
|
let mut command = Command::new(program);
|
|
#[cfg(unix)]
|
|
{
|
|
use std::os::unix::process::CommandExt;
|
|
command.process_group(0);
|
|
}
|
|
command
|
|
}
|
|
|
|
fn extractor() -> Command {
|
|
let mut command = isolated_command(OsStr::new("yt-dlp"));
|
|
command.args([
|
|
"--ignore-config",
|
|
"--force-ipv4",
|
|
"--no-warnings",
|
|
"--socket-timeout",
|
|
"15",
|
|
]);
|
|
command
|
|
}
|
|
|
|
fn stop(child: &mut Child) {
|
|
if matches!(child.try_wait(), Ok(None)) {
|
|
// SAFETY: isolated_command gives this still-running child its own process group.
|
|
// Cancel the whole process group without signaling the TUI.
|
|
#[cfg(unix)]
|
|
unsafe {
|
|
libc::kill(-(child.id() as i32), libc::SIGKILL);
|
|
}
|
|
let _ = child.kill();
|
|
}
|
|
let _ = child.wait();
|
|
}
|
|
|
|
fn read_errors(mut pipe: impl Read + Send + 'static) -> Receiver<io::Result<String>> {
|
|
let (sender, receiver) = mpsc::channel();
|
|
thread::spawn(move || {
|
|
let mut text = String::new();
|
|
let _ = sender.send(pipe.read_to_string(&mut text).map(|_| text));
|
|
});
|
|
receiver
|
|
}
|
|
|
|
fn failure(errors: &Receiver<io::Result<String>>, fallback: String) -> Option<Result<(), String>> {
|
|
match errors.try_recv() {
|
|
Ok(Ok(text)) => Some(Err(if text.trim().is_empty() {
|
|
fallback
|
|
} else {
|
|
text.trim().to_owned()
|
|
})),
|
|
Ok(Err(error)) => Some(Err(format!("{fallback}: {error}"))),
|
|
Err(TryRecvError::Empty) => None,
|
|
Err(TryRecvError::Disconnected) => Some(Err(fallback)),
|
|
}
|
|
}
|
|
|
|
pub fn video_url(id: &str) -> io::Result<String> {
|
|
if valid_video_id(id) {
|
|
Ok(format!("https://www.youtube.com/watch?v={id}"))
|
|
} else {
|
|
tubular::anime::Identity::from_key(id)
|
|
.and_then(|identity| identity.watch_url())
|
|
.map_err(|error| io::Error::new(io::ErrorKind::InvalidInput, error))
|
|
}
|
|
}
|
|
|
|
pub fn open_in_browser(id: &str) -> io::Result<Receiver<Result<(), String>>> {
|
|
if !valid_video_id(id) {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Invalid YouTube video ID",
|
|
));
|
|
}
|
|
let opener = if cfg!(target_os = "macos") {
|
|
"open"
|
|
} else {
|
|
"xdg-open"
|
|
};
|
|
let child = isolated_command(OsStr::new(opener))
|
|
.arg(video_url(id)?)
|
|
.stdin(Stdio::null())
|
|
.stdout(Stdio::null())
|
|
.stderr(Stdio::piped())
|
|
.spawn()
|
|
.map_err(|error| {
|
|
io::Error::new(
|
|
error.kind(),
|
|
format!("Cannot start browser opener: {error}"),
|
|
)
|
|
})?;
|
|
let (sender, receiver) = mpsc::channel();
|
|
thread::spawn(move || {
|
|
let result = child
|
|
.wait_with_output()
|
|
.map_err(|error| format!("Cannot wait for browser opener: {error}"))
|
|
.and_then(|output| {
|
|
if output.status.success() {
|
|
Ok(())
|
|
} else {
|
|
let error = String::from_utf8_lossy(&output.stderr);
|
|
Err(format!(
|
|
"Browser opener exited with {}: {}",
|
|
output.status,
|
|
error.trim()
|
|
))
|
|
}
|
|
});
|
|
let _ = sender.send(result);
|
|
});
|
|
Ok(receiver)
|
|
}
|
|
|
|
pub struct PlayerLaunch {
|
|
player: Child,
|
|
player_errors: Receiver<io::Result<String>>,
|
|
}
|
|
|
|
impl PlayerLaunch {
|
|
pub fn start(
|
|
videos: &[Video],
|
|
player: &OsStr,
|
|
state: &mut crate::video_state::State,
|
|
enqueue: bool,
|
|
) -> io::Result<Self> {
|
|
if videos.is_empty() {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"No videos selected",
|
|
));
|
|
}
|
|
let mut command = isolated_command(player);
|
|
command.arg(if enqueue {
|
|
"--enqueue"
|
|
} else {
|
|
"--new-instance"
|
|
});
|
|
if let [video] = videos
|
|
&& !enqueue
|
|
{
|
|
let url = video_url(&video.id)?;
|
|
let report = state.progress_file(video)?;
|
|
crate::video_state::register_identity(&report, &video.id)?;
|
|
command
|
|
.args(["--url", &url])
|
|
.arg("--progress-file")
|
|
.arg(report);
|
|
return Self::launch(command);
|
|
}
|
|
let mut ids = BTreeSet::new();
|
|
let urls = videos
|
|
.iter()
|
|
.map(|video| {
|
|
if !ids.insert(video.id.as_str()) {
|
|
return Err(io::Error::new(
|
|
io::ErrorKind::InvalidInput,
|
|
"Duplicate queue identity",
|
|
));
|
|
}
|
|
video_url(&video.id)
|
|
})
|
|
.collect::<io::Result<Vec<_>>>()?;
|
|
let reports = state.progress_files(videos)?;
|
|
#[derive(Serialize)]
|
|
struct QueueItem {
|
|
url: String,
|
|
progress_file: PathBuf,
|
|
}
|
|
#[derive(Serialize)]
|
|
struct Queue {
|
|
version: u8,
|
|
items: Vec<QueueItem>,
|
|
}
|
|
let items = videos
|
|
.iter()
|
|
.zip(urls)
|
|
.zip(reports)
|
|
.map(|((video, url), report)| {
|
|
crate::video_state::register_identity(&report, &video.id)?;
|
|
let parent = fs::canonicalize(report.parent().expect("session directory"))?;
|
|
Ok(QueueItem {
|
|
url,
|
|
progress_file: parent.join("progress.jsonl"),
|
|
})
|
|
})
|
|
.collect::<io::Result<Vec<_>>>()?;
|
|
let manifest = items[0].progress_file.with_file_name("queue.json");
|
|
let mut options = OpenOptions::new();
|
|
options.write(true).create_new(true);
|
|
#[cfg(unix)]
|
|
{
|
|
use std::os::unix::fs::OpenOptionsExt;
|
|
options.mode(0o600);
|
|
}
|
|
let file = options.open(&manifest)?;
|
|
serde_json::to_writer(&file, &Queue { version: 1, items })?;
|
|
file.sync_all()?;
|
|
command.arg("--queue-file").arg(manifest);
|
|
Self::launch(command)
|
|
}
|
|
|
|
fn launch(mut command: Command) -> io::Result<Self> {
|
|
let mut player = command
|
|
.stdin(Stdio::null())
|
|
.stdout(Stdio::null())
|
|
.stderr(Stdio::piped())
|
|
.spawn()
|
|
.map_err(|error| {
|
|
io::Error::new(
|
|
error.kind(),
|
|
format!("Cannot start Terebi launcher: {error}"),
|
|
)
|
|
})?;
|
|
let player_errors = read_errors(player.stderr.take().expect("stderr is piped"));
|
|
Ok(Self {
|
|
player,
|
|
player_errors,
|
|
})
|
|
}
|
|
|
|
pub fn poll(&mut self) -> Option<Result<(), String>> {
|
|
match self.player.try_wait() {
|
|
Ok(Some(status)) if status.success() => Some(Ok(())),
|
|
Ok(Some(status)) => {
|
|
failure(&self.player_errors, format!("Player exited with {status}"))
|
|
}
|
|
Ok(None) => None,
|
|
Err(error) => Some(Err(format!("Cannot wait for player: {error}"))),
|
|
}
|
|
}
|
|
}
|
|
|
|
impl Drop for PlayerLaunch {
|
|
fn drop(&mut self) {
|
|
stop(&mut self.player);
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
#[path = "../tests/unit/youtube.rs"]
|
|
mod tests;
|