diff --git a/crates/arr-subs/Cargo.toml b/crates/arr-subs/Cargo.toml index a2ca035..7ff83d5 100644 --- a/crates/arr-subs/Cargo.toml +++ b/crates/arr-subs/Cargo.toml @@ -25,12 +25,13 @@ encoding_rs.workspace = true reqwest.workspace = true serde.workspace = true serde_json.workspace = true +tempfile.workspace = true thiserror.workspace = true +tokio.workspace = true tracing.workspace = true zip.workspace = true [dev-dependencies] -tempfile.workspace = true tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } wiremock.workspace = true diff --git a/crates/arr-subs/src/error.rs b/crates/arr-subs/src/error.rs index f0e5d56..0185f98 100644 --- a/crates/arr-subs/src/error.rs +++ b/crates/arr-subs/src/error.rs @@ -111,6 +111,53 @@ pub enum Error { /// What went wrong, short enough to log. detail: String, }, + + /// `alass` could not be started at all — the binary is not installed + /// where the configuration points (#194). A health condition (#200), + /// folded into the operator message §9.5 defines; never a panic. + #[error("could not run {binary}")] + Spawn { + /// The binary that was attempted. + binary: String, + /// The underlying IO failure. + #[source] + source: std::io::Error, + }, + + /// One `alass` run took longer than its timeout. + #[error("syncing {path} timed out")] + Timeout { + /// The subtitle being synced. + path: PathBuf, + }, + + /// `alass` ran and exited non-zero. + /// + /// Its own variant rather than [`Error::Io`]: the process answered, so + /// this is about the sync, and `status` plus a trimmed stderr is what an + /// operator needs to see. + #[error("syncing {path} failed (status {status:?}): {stderr}")] + SyncFailed { + /// The subtitle being synced. + path: PathBuf, + /// The exit code, when it exited rather than died on a signal. + status: Option, + /// The first part of what it wrote to stderr. + stderr: String, + }, + + /// A file that should be SRT — the subtitle going into `alass`, or what + /// came out of it — does not parse as SRT. + /// + /// An implausible output is *not* here: that is [`crate::sync::Outcome::Rejected`], + /// a decision, while a file nothing can parse is a fact worth failing on. + #[error("{path} is not SRT: {detail}")] + NotSrt { + /// The file involved. + path: PathBuf, + /// What was wrong with it, short enough to log. + detail: String, + }, } impl Error { @@ -120,7 +167,10 @@ impl Error { /// response do not fix themselves, and retrying them only burns budget. #[must_use] pub const fn is_transient(&self) -> bool { - matches!(self, Self::Transport { .. } | Self::RateLimited { .. }) + matches!( + self, + Self::Transport { .. } | Self::RateLimited { .. } | Self::Timeout { .. } + ) } } @@ -139,4 +189,24 @@ mod tests { .is_transient()); assert!(!Error::Unauthorized { provider }.is_transient()); } + + #[test] + fn a_timeout_is_worth_retrying_and_a_missing_binary_is_not() { + let timeout = Error::Timeout { + path: std::path::PathBuf::from("/m/a.srt"), + }; + let spawn = Error::Spawn { + binary: "alass".to_owned(), + source: std::io::Error::new(std::io::ErrorKind::NotFound, "no such file"), + }; + let not_srt = Error::NotSrt { + path: std::path::PathBuf::from("/m/a.srt"), + detail: "line 1: expected a timing line".to_owned(), + }; + assert!(timeout.is_transient()); + // A missing binary needs installing, and a broken subtitle does not + // heal; neither resolves by retrying. + assert!(!spawn.is_transient()); + assert!(!not_srt.is_transient()); + } } diff --git a/crates/arr-subs/src/lib.rs b/crates/arr-subs/src/lib.rs index a22dde2..c41a92c 100644 --- a/crates/arr-subs/src/lib.rs +++ b/crates/arr-subs/src/lib.rs @@ -33,6 +33,7 @@ pub mod model; pub mod opensubtitles; pub mod podnapisi; pub mod srt; +pub mod sync; pub mod translate; pub use error::{Error, Result}; @@ -42,6 +43,7 @@ pub use model::{ pub use opensubtitles::{moviehash, OpenSubtitles, OpenSubtitlesConfig}; pub use podnapisi::{Podnapisi, PodnapisiBuilder, DEFAULT_BASE_URL as PODNAPISI_DEFAULT_BASE_URL}; pub use srt::Cue; +pub use sync::{Outcome, Rejection, Syncer, DEFAULT_BINARY as ALASS_DEFAULT_BINARY}; pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, TranslatedCue}; // The OpenSubtitles provider's HTTP paths are exercised by the integration diff --git a/crates/arr-subs/src/sync.rs b/crates/arr-subs/src/sync.rs new file mode 100644 index 0000000..79e53d4 --- /dev/null +++ b/crates/arr-subs/src/sync.rs @@ -0,0 +1,519 @@ +//! The `alass` wrapper — timing sync for every fetched and translated +//! subtitle, DESIGN.md §15. +//! +//! The same shape as `arr-probe`'s [`crate::Prober`]: one small binary, +//! invoked and discarded, with a configurable path and a timeout. It adds +//! nothing at rest. +//! +//! `alass` reports no confidence value, so trust is bounded structurally: the +//! output is parsed back and compared against the input it was made from. A +//! cue that moved more than [`MAX_SHIFT`] from where it started, or cues that +//! went missing between input and output, make the whole result implausible — +//! and an implausible result is not an error but a [`Outcome::Rejected`], so +//! the caller keeps the unsynced original and flags the row (#186). The +//! reference is always the media file itself; subtitle and video are both on +//! disk by the time this runs. + +use std::{ffi::OsString, path::Path, process::Stdio, time::Duration}; + +use tokio::process::Command; + +use crate::{error::Error, srt, Result}; + +/// The binary invoked when nothing else is configured. +pub const DEFAULT_BINARY: &str = "alass"; + +/// How long one sync may take. Generous: `alass` correlates audio against the +/// whole file, and a feature-length remux over a network mount is slow. +pub const DEFAULT_TIMEOUT: Duration = Duration::from_secs(180); + +/// How far a cue may sit from where it started and still be plausible. +/// +/// §15's number. Wide enough for a genuine PAL-speedup or a different cut of +/// the same film; narrow enough that a sync against the wrong file cannot +/// sneak through. +pub const MAX_SHIFT: Duration = Duration::from_secs(60); + +/// How much of `alass`'s stderr is kept in an error. +const STDERR_LIMIT: usize = 512; + +/// Runs `alass` over one subtitle against its video. +#[derive(Clone, Debug)] +pub struct Syncer { + binary: OsString, + timeout: Duration, +} + +impl Default for Syncer { + fn default() -> Self { + Self { + binary: DEFAULT_BINARY.into(), + timeout: DEFAULT_TIMEOUT, + } + } +} + +impl Syncer { + /// A syncer that runs `alass` from `PATH`. + #[must_use] + pub fn new() -> Self { + Self::default() + } + + /// Run a specific binary instead of whatever `PATH` resolves to. + #[must_use] + pub fn with_binary(mut self, binary: impl Into) -> Self { + self.binary = binary.into(); + self + } + + /// Change how long one sync may take. + #[must_use] + pub fn with_timeout(mut self, timeout: Duration) -> Self { + self.timeout = timeout; + self + } + + /// Sync `subtitle` against `video`, in place on disk. + /// + /// Returns the synced SRT text on acceptance; the input file is never + /// touched either way, so a rejection leaves the caller holding the + /// unsynced original it already has. + /// + /// # Errors + /// + /// [`Error::NotSrt`] when either side does not parse as SRT, + /// [`Error::Spawn`] when the binary is missing (a health condition, + /// #200), [`Error::Timeout`], [`Error::Failed`] when `alass` exits + /// non-zero, and [`Error::Io`] when a file cannot be read or written. + pub async fn sync( + &self, + video: impl AsRef, + subtitle: impl AsRef, + ) -> Result { + let video = video.as_ref(); + let subtitle = subtitle.as_ref(); + + let text = tokio::fs::read_to_string(subtitle) + .await + .map_err(|source| Error::Io { + path: subtitle.to_path_buf(), + source, + })?; + let input = srt::parse(&text).map_err(|error| Error::NotSrt { + path: subtitle.to_path_buf(), + detail: error.to_string(), + })?; + + let output_dir = tempfile::tempdir().map_err(|source| Error::Io { + path: std::env::temp_dir(), + source, + })?; + let synced_path = output_dir.path().join("synced.srt"); + self.run(video, subtitle, &synced_path).await?; + + let synced_text = tokio::fs::read_to_string(&synced_path) + .await + .map_err(|source| Error::Io { + path: synced_path.clone(), + source, + })?; + let output = srt::parse(&synced_text).map_err(|error| Error::NotSrt { + path: synced_path, + detail: error.to_string(), + })?; + + Ok(match plausible(&input, &output) { + Some(rejection) => Outcome::Rejected(rejection), + None => Outcome::Accepted { + content: synced_text, + }, + }) + } + + /// Spawn `alass` and wait for it to write its output. + async fn run(&self, video: &Path, subtitle: &Path, output: &Path) -> Result<()> { + let mut command = Command::new(&self.binary); + command + .args([subtitle, video]) + .arg(output) + .stdin(Stdio::null()) + .stdout(Stdio::null()) + .stderr(Stdio::piped()) + .kill_on_drop(true); + + let child = command.spawn().map_err(|source| Error::Spawn { + binary: self.binary.to_string_lossy().into_owned(), + source, + })?; + + let waited = tokio::time::timeout(self.timeout, child.wait_with_output()).await; + let output_handle = match waited { + Ok(output) => output.map_err(|source| Error::Io { + path: subtitle.to_path_buf(), + source, + })?, + Err(_) => { + return Err(Error::Timeout { + path: subtitle.to_path_buf(), + }) + } + }; + + if output_handle.status.success() { + return Ok(()); + } + + let stderr = String::from_utf8_lossy(&output_handle.stderr) + .trim() + .chars() + .take(STDERR_LIMIT) + .collect(); + Err(Error::SyncFailed { + path: subtitle.to_path_buf(), + status: output_handle.status.code(), + stderr, + }) + } +} + +/// What one sync run decided. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Outcome { + /// Plausible; `content` is the synced SRT to write over the original. + Accepted { + /// The synced subtitle, as `alass` wrote it. + content: String, + }, + /// Implausible. The unsynced original stands and the row is flagged. + Rejected(Rejection), +} + +/// Why a sync's output was judged implausible (§15). +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Rejection { + /// Cues disappeared between the input and the output. + CueLost { + /// How many the input had. + input: usize, + /// How many survived. + output: usize, + }, + /// A cue moved further than [`MAX_SHIFT`] from where it started. + Shifted { + /// The cue's position in the file, 0-based. + index: usize, + /// How far it moved. + shift: Duration, + }, +} + +impl std::fmt::Display for Rejection { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::CueLost { input, output } => { + write!(formatter, "{output} of {input} cues survived") + } + Self::Shifted { index, shift } => { + write!(formatter, "cue {index} moved {shift:.1?}") + } + } + } +} + +/// Compare a sync's output against its input. +/// +/// Cues are paired positionally: `alass` shifts timings without reordering or +/// merging, so position is the identity across the two files. Loss is checked +/// first because a shifted comparison means nothing once counts diverge. +fn plausible(input: &[srt::Cue], output: &[srt::Cue]) -> Option { + if output.len() < input.len() { + return Some(Rejection::CueLost { + input: input.len(), + output: output.len(), + }); + } + + input + .iter() + .zip(output) + .enumerate() + .find_map(|(index, (before, after))| { + let shift = after.start.abs_diff(before.start); + (shift > MAX_SHIFT).then_some(Rejection::Shifted { index, shift }) + }) +} + +#[cfg(test)] +mod tests { + use std::{os::unix::fs::PermissionsExt, time::Duration}; + + use tokio::io::AsyncWriteExt; + + use super::{plausible, Outcome, Rejection, Syncer, MAX_SHIFT}; + use crate::error::Error; + use crate::srt::Cue; + + const PLAIN: &str = + "1\n00:00:01,000 --> 00:00:02,500\nol\u{e1}\n\n2\n00:00:03,000 --> 00:00:04,000\ntwo lines\nof text\n"; + + /// Write an executable fake `alass`. Its body receives the three paths as + /// `$1` subtitle, `$2` video and `$3` output. + async fn fake_binary(dir: &std::path::Path, body: &str) -> std::path::PathBuf { + let path = dir.join("alass"); + let mut file = tokio::fs::File::create(&path) + .await + .expect("test setup and fake binary succeed"); + file.write_all(b"#!/bin/sh\n") + .await + .expect("test setup and fake binary succeed"); + file.write_all(body.as_bytes()) + .await + .expect("test setup and fake binary succeed"); + file.sync_all() + .await + .expect("test setup and fake binary succeed"); + drop(file); + tokio::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755)) + .await + .expect("test setup and fake binary succeed"); + path + } + + async fn write_subtitle( + dir: &std::path::Path, + name: &str, + content: &str, + ) -> std::path::PathBuf { + let path = dir.join(name); + tokio::fs::write(&path, content) + .await + .expect("test setup and fake binary succeed"); + path + } + + fn cue(start: u64) -> Cue { + Cue { + start: Duration::from_millis(start * 1_000), + end: Duration::from_millis(start * 1_000 + 1_500), + text: "line".to_owned(), + } + } + + #[tokio::test] + async fn an_unchanged_output_is_accepted_with_alass_text() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + let binary = fake_binary(dir.path(), "cp \"$1\" \"$3\"\n").await; + let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await; + + let outcome = Syncer::new() + .with_binary(&binary) + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect("test setup and fake binary succeed"); + + assert_eq!( + outcome, + Outcome::Accepted { + content: PLAIN.to_owned() + } + ); + } + + #[tokio::test] + async fn a_shift_within_the_bound_is_accepted() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + // A five-second delay, the classic PAL-to-film speedup correction. + let binary = fake_binary( + dir.path(), + "printf '1\\n00:00:06,000 --> 00:00:07,500\\nola\\n\\n2\\n00:00:08,000 --> 00:00:09,000\\ntext\\n' > \"$3\"\n", + ) + .await; + let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await; + + let outcome = Syncer::new() + .with_binary(&binary) + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect("test setup and fake binary succeed"); + + assert!(matches!(outcome, Outcome::Accepted { .. })); + } + + #[tokio::test] + async fn a_shift_beyond_sixty_seconds_is_rejected_and_the_original_stands() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + let binary = fake_binary( + dir.path(), + "printf '1\\n00:01:40,000 --> 00:01:41,500\\nola\\n\\n2\\n00:01:42,000 --> 00:01:43,000\\ntext\\n' > \"$3\"\n", + ) + .await; + let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await; + + let outcome = Syncer::new() + .with_binary(&binary) + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect("test setup and fake binary succeed"); + + assert_eq!( + outcome, + Outcome::Rejected(Rejection::Shifted { + index: 0, + shift: Duration::from_secs(99) + }) + ); + // The rejection never touched the input file. + assert_eq!( + tokio::fs::read_to_string(&subtitle) + .await + .expect("test setup and fake binary succeed"), + PLAIN + ); + } + + #[tokio::test] + async fn lost_cues_are_rejected() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + let binary = fake_binary( + dir.path(), + "printf '1\\n00:00:01,000 --> 00:00:02,500\\nola\\n' > \"$3\"\n", + ) + .await; + let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await; + + let outcome = Syncer::new() + .with_binary(&binary) + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect("test setup and fake binary succeed"); + + assert_eq!( + outcome, + Outcome::Rejected(Rejection::CueLost { + input: 2, + output: 1 + }) + ); + } + + #[tokio::test] + async fn a_failing_run_names_the_stderr() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + let binary = fake_binary(dir.path(), "echo 'no such stream' >&2\nexit 3\n").await; + let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await; + + let error = Syncer::new() + .with_binary(&binary) + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect_err("the run must fail"); + + match error { + Error::SyncFailed { status, stderr, .. } => { + assert_eq!(status, Some(3)); + assert_eq!(stderr, "no such stream"); + } + other => panic!("expected SyncFailed, got {other:?}"), + } + } + + #[tokio::test] + async fn a_missing_binary_is_a_spawn_error_not_a_panic() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await; + + let error = Syncer::new() + .with_binary("alass-not-installed") + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect_err("the run must fail"); + + assert!(matches!(error, Error::Spawn { .. })); + } + + #[tokio::test] + async fn an_input_that_is_not_srt_never_reaches_alass() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + // Any run would leave this marker behind; its absence proves the + // binary was never invoked. + let marker = dir.path().join("marker"); + let binary = + fake_binary(dir.path(), &format!("touch {}\nexit 0\n", marker.display())).await; + let subtitle = write_subtitle(dir.path(), "in.txt", "{1}{50}not srt at all").await; + + let error = Syncer::new() + .with_binary(&binary) + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect_err("the run must fail"); + + assert!(matches!(error, Error::NotSrt { .. })); + assert!(!marker.exists(), "alass must not run on unparseable input"); + } + + #[tokio::test] + async fn an_output_that_is_not_srt_is_an_error_rather_than_a_rejection() { + let dir = tempfile::tempdir().expect("test setup and fake binary succeed"); + let binary = fake_binary(dir.path(), "echo garbage > \"$3\"\n").await; + let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await; + + let error = Syncer::new() + .with_binary(&binary) + .sync(dir.path().join("video.mkv"), &subtitle) + .await + .expect_err("the run must fail"); + + assert!(matches!(error, Error::NotSrt { .. })); + } + + #[test] + fn exactly_sixty_seconds_is_still_plausible() { + let before = [cue(10)]; + let mut after = before[0].clone(); + after.start += MAX_SHIFT; + assert_eq!(plausible(&before, &[after]), None); + } + + #[test] + fn a_grown_output_is_not_loss_but_a_shift_is_still_caught() { + let before = [cue(10), cue(20)]; + let grown = vec![cue(10), cue(20), cue(30)]; + assert_eq!(plausible(&before, &grown), None, "extra cues are not loss"); + + let mut moved = vec![cue(10), cue(20), cue(30)]; + moved[1].start += MAX_SHIFT + Duration::from_secs(1); + assert_eq!( + plausible(&before, &moved), + Some(Rejection::Shifted { + index: 1, + shift: MAX_SHIFT + Duration::from_secs(1) + }) + ); + } + + #[test] + fn loss_is_reported_before_any_shift() { + let before = [cue(10), cue(600)]; + let after = [cue(700)]; + assert_eq!( + plausible(&before, &after), + Some(Rejection::CueLost { + input: 2, + output: 1 + }) + ); + } + + #[test] + fn rejections_render_as_a_sentence() { + assert_eq!( + Rejection::CueLost { + input: 2, + output: 1 + } + .to_string(), + "1 of 2 cues survived" + ); + } +}