Merge #214: sync manually grabbed subtitles with alass
Closes #214 # Conflicts: # crates/arr-api/src/state.rs # crates/arr-daemon/src/main.rs
This commit is contained in:
@@ -5,7 +5,7 @@ use std::sync::Arc;
|
|||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
|
|
||||||
use arr_db::Db;
|
use arr_db::Db;
|
||||||
use arr_subs::{Backend, Provider};
|
use arr_subs::{Backend, Provider, Syncer};
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
|
|
||||||
use crate::jellyfin::JellyfinClient;
|
use crate::jellyfin::JellyfinClient;
|
||||||
@@ -82,6 +82,7 @@ pub struct AppState {
|
|||||||
subtitle_providers: Arc<Vec<Arc<dyn Provider>>>,
|
subtitle_providers: Arc<Vec<Arc<dyn Provider>>>,
|
||||||
translation_backends: Arc<Vec<Arc<dyn Backend>>>,
|
translation_backends: Arc<Vec<Arc<dyn Backend>>>,
|
||||||
jellyfin: Option<JellyfinClient>,
|
jellyfin: Option<JellyfinClient>,
|
||||||
|
syncer: Syncer,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Work explicitly requested through the movie API.
|
/// Work explicitly requested through the movie API.
|
||||||
@@ -155,6 +156,7 @@ impl AppState {
|
|||||||
subtitle_providers: Arc::new(Vec::new()),
|
subtitle_providers: Arc::new(Vec::new()),
|
||||||
translation_backends: Arc::new(Vec::new()),
|
translation_backends: Arc::new(Vec::new()),
|
||||||
jellyfin: None,
|
jellyfin: None,
|
||||||
|
syncer: Syncer::default(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -217,6 +219,18 @@ impl AppState {
|
|||||||
.find(|backend| backend.id().as_str() == id)
|
.find(|backend| backend.id().as_str() == id)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Attach the `alass` binary this deployment runs (§15). Defaults to
|
||||||
|
/// resolving `alass` from `PATH`.
|
||||||
|
#[must_use]
|
||||||
|
pub fn with_syncer(mut self, syncer: Syncer) -> Self {
|
||||||
|
self.syncer = syncer;
|
||||||
|
self
|
||||||
|
}
|
||||||
|
|
||||||
|
pub(crate) fn syncer(&self) -> &Syncer {
|
||||||
|
&self.syncer
|
||||||
|
}
|
||||||
|
|
||||||
/// Wait for the next manual movie action in the daemon's reconcile loop.
|
/// Wait for the next manual movie action in the daemon's reconcile loop.
|
||||||
///
|
///
|
||||||
/// # Errors
|
/// # Errors
|
||||||
|
|||||||
@@ -645,13 +645,19 @@ pub async fn grab(
|
|||||||
let text = srt_text(&fetched)?;
|
let text = srt_text(&fetched)?;
|
||||||
write_sidecar(&destination, &text).await?;
|
write_sidecar(&destination, &text).await?;
|
||||||
|
|
||||||
|
let sync = state.syncer().settle(&target.path, &destination).await;
|
||||||
|
if let Some(synced) = &sync.content {
|
||||||
|
write_sidecar(&destination, synced).await?;
|
||||||
|
}
|
||||||
|
|
||||||
let mut record = arr_db::NewSubtitleFile::fetched(
|
let mut record = arr_db::NewSubtitleFile::fetched(
|
||||||
target.media_file_id,
|
target.media_file_id,
|
||||||
&language.to_string(),
|
&language.to_string(),
|
||||||
&input.provider,
|
&input.provider,
|
||||||
&input.candidate_id,
|
&input.candidate_id,
|
||||||
&destination.to_string_lossy(),
|
&destination.to_string_lossy(),
|
||||||
);
|
)
|
||||||
|
.sync(db_sync_state(sync.state));
|
||||||
if input.forced {
|
if input.forced {
|
||||||
record = record.forced();
|
record = record.forced();
|
||||||
}
|
}
|
||||||
@@ -860,6 +866,19 @@ fn srt_text(fetched: &arr_subs::Fetched) -> Result<String, ApiError> {
|
|||||||
.map_err(|error| ApiError::SubtitleUpstream(error.to_string()))
|
.map_err(|error| ApiError::SubtitleUpstream(error.to_string()))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Map `arr_subs`'s three-state sync result onto the column pair `arr_db`
|
||||||
|
/// stores it as. `Syncer::settle` already folded an unusable `alass` and an
|
||||||
|
/// implausible result together into "nothing changed" — this is just the
|
||||||
|
/// vocabulary switch between the crate that ran `alass` and the one that
|
||||||
|
/// persists what it decided.
|
||||||
|
const fn db_sync_state(state: arr_subs::SyncState) -> arr_db::SubtitleSync {
|
||||||
|
match state {
|
||||||
|
arr_subs::SyncState::NotRun => arr_db::SubtitleSync::NotRun,
|
||||||
|
arr_subs::SyncState::Synced => arr_db::SubtitleSync::Synced,
|
||||||
|
arr_subs::SyncState::Rejected => arr_db::SubtitleSync::Rejected,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Write a sidecar whole or not at all, so Jellyfin never reads a half file.
|
/// Write a sidecar whole or not at all, so Jellyfin never reads a half file.
|
||||||
async fn write_sidecar(destination: &Path, text: &str) -> Result<(), ApiError> {
|
async fn write_sidecar(destination: &Path, text: &str) -> Result<(), ApiError> {
|
||||||
let failure = |path: &Path, error: std::io::Error| {
|
let failure = |path: &Path, error: std::io::Error| {
|
||||||
@@ -920,6 +939,7 @@ async fn refresh_jellyfin(state: &AppState) {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
#[allow(clippy::too_many_lines)]
|
#[allow(clippy::too_many_lines)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
use std::os::unix::fs::PermissionsExt;
|
||||||
use std::path::PathBuf;
|
use std::path::PathBuf;
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
|
|
||||||
@@ -929,6 +949,7 @@ mod tests {
|
|||||||
ProviderId, SearchFuture, SearchRequest, SubtitleFormat, TranslateFuture, TranslatedCue,
|
ProviderId, SearchFuture, SearchRequest, SubtitleFormat, TranslateFuture, TranslatedCue,
|
||||||
};
|
};
|
||||||
use axum::http::StatusCode;
|
use axum::http::StatusCode;
|
||||||
|
use tokio::io::AsyncWriteExt;
|
||||||
|
|
||||||
use crate::{router, AppState, Upstreams};
|
use crate::{router, AppState, Upstreams};
|
||||||
|
|
||||||
@@ -1110,6 +1131,22 @@ mod tests {
|
|||||||
async fn application(
|
async fn application(
|
||||||
providers: Vec<Arc<dyn Provider>>,
|
providers: Vec<Arc<dyn Provider>>,
|
||||||
backends: Vec<Arc<dyn Backend>>,
|
backends: Vec<Arc<dyn Backend>>,
|
||||||
|
) -> Fixture {
|
||||||
|
build_fixture(providers, backends, arr_subs::Syncer::default()).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn application_with_syncer(
|
||||||
|
providers: Vec<Arc<dyn Provider>>,
|
||||||
|
backends: Vec<Arc<dyn Backend>>,
|
||||||
|
syncer: arr_subs::Syncer,
|
||||||
|
) -> Fixture {
|
||||||
|
build_fixture(providers, backends, syncer).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn build_fixture(
|
||||||
|
providers: Vec<Arc<dyn Provider>>,
|
||||||
|
backends: Vec<Arc<dyn Backend>>,
|
||||||
|
syncer: arr_subs::Syncer,
|
||||||
) -> Fixture {
|
) -> Fixture {
|
||||||
let dir = tempfile::tempdir().expect("tempdir");
|
let dir = tempfile::tempdir().expect("tempdir");
|
||||||
let database = arr_db::Db::connect(dir.path().join("arr.db"))
|
let database = arr_db::Db::connect(dir.path().join("arr.db"))
|
||||||
@@ -1151,7 +1188,8 @@ mod tests {
|
|||||||
.expect("state")
|
.expect("state")
|
||||||
.with_database(database)
|
.with_database(database)
|
||||||
.with_subtitle_providers(providers)
|
.with_subtitle_providers(providers)
|
||||||
.with_translation_backends(backends);
|
.with_translation_backends(backends)
|
||||||
|
.with_syncer(syncer);
|
||||||
|
|
||||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
|
||||||
.await
|
.await
|
||||||
@@ -1470,6 +1508,77 @@ mod tests {
|
|||||||
assert_eq!(fixture.subtitle_rows().await.len(), 1);
|
assert_eq!(fixture.subtitle_rows().await.len(), 1);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Write an executable fake `alass`. Its body receives the subtitle,
|
||||||
|
/// video and output paths as `$1`, `$2`, `$3`.
|
||||||
|
async fn fake_alass(dir: &std::path::Path, body: &str) -> PathBuf {
|
||||||
|
let path = dir.join("alass");
|
||||||
|
let mut file = tokio::fs::File::create(&path).await.expect("fake alass");
|
||||||
|
file.write_all(b"#!/bin/sh\n").await.expect("fake alass");
|
||||||
|
file.write_all(body.as_bytes()).await.expect("fake alass");
|
||||||
|
file.sync_all().await.expect("fake alass");
|
||||||
|
drop(file);
|
||||||
|
tokio::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o755))
|
||||||
|
.await
|
||||||
|
.expect("fake alass");
|
||||||
|
path
|
||||||
|
}
|
||||||
|
|
||||||
|
/// §15: `alass` runs on every fetched subtitle (#214), and an accepted
|
||||||
|
/// shift replaces the sidecar's content before the row is recorded.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_grab_writes_the_synced_content_when_alass_accepts_it() {
|
||||||
|
let dir = tempfile::tempdir().expect("tempdir");
|
||||||
|
// The fetched cue starts at 1s (`SRT`); a 5s shift is within §15's
|
||||||
|
// 60-second bound.
|
||||||
|
let binary = fake_alass(
|
||||||
|
dir.path(),
|
||||||
|
"printf '1\\n00:00:06,000 --> 00:00:07,000\\nola\\n' > \"$3\"\n",
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let fixture = application_with_syncer(
|
||||||
|
vec![Arc::new(StubProvider::new("opensubtitles"))],
|
||||||
|
vec![],
|
||||||
|
arr_subs::Syncer::new().with_binary(&binary),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
let (status, body) = grab(&fixture, pt()).await;
|
||||||
|
assert_eq!(status, StatusCode::CREATED, "{body}");
|
||||||
|
assert_eq!(body["sync"], "synced");
|
||||||
|
let sidecar = PathBuf::from(body["path"].as_str().expect("path"));
|
||||||
|
assert_eq!(
|
||||||
|
tokio::fs::read_to_string(&sidecar).await.expect("sidecar"),
|
||||||
|
"1\n00:00:06,000 --> 00:00:07,000\nola\n"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// §15: an implausible shift keeps the unsynced original and flags the
|
||||||
|
/// file rather than failing the grab.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn a_grab_keeps_the_original_when_alass_is_implausible() {
|
||||||
|
let dir = tempfile::tempdir().expect("tempdir");
|
||||||
|
let binary = fake_alass(
|
||||||
|
dir.path(),
|
||||||
|
"printf '1\\n00:05:00,000 --> 00:05:01,000\\nola\\n' > \"$3\"\n",
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let fixture = application_with_syncer(
|
||||||
|
vec![Arc::new(StubProvider::new("opensubtitles"))],
|
||||||
|
vec![],
|
||||||
|
arr_subs::Syncer::new().with_binary(&binary),
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
let (status, body) = grab(&fixture, pt()).await;
|
||||||
|
assert_eq!(status, StatusCode::CREATED, "{body}");
|
||||||
|
assert_eq!(body["sync"], "rejected");
|
||||||
|
let sidecar = PathBuf::from(body["path"].as_str().expect("path"));
|
||||||
|
assert_eq!(
|
||||||
|
tokio::fs::read_to_string(&sidecar).await.expect("sidecar"),
|
||||||
|
SRT
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
/// Sidecars are SRT (§15), so a provider serving ASS or VTT is converted
|
/// Sidecars are SRT (§15), so a provider serving ASS or VTT is converted
|
||||||
/// on the way to disk (#213) rather than refused.
|
/// on the way to disk (#213) rather than refused.
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
|
|||||||
@@ -151,7 +151,8 @@ async fn run() -> Result<(), Error> {
|
|||||||
config.opensubtitles_username,
|
config.opensubtitles_username,
|
||||||
config.opensubtitles_password,
|
config.opensubtitles_password,
|
||||||
))
|
))
|
||||||
.with_jellyfin(api_jellyfin);
|
.with_jellyfin(api_jellyfin)
|
||||||
|
.with_syncer(arr_subs::Syncer::new().with_binary(config.alass_path.clone()));
|
||||||
|
|
||||||
let app = arr_api::router(state.clone())
|
let app = arr_api::router(state.clone())
|
||||||
.merge(arr_compat::router(compat))
|
.merge(arr_compat::router(compat))
|
||||||
|
|||||||
@@ -54,7 +54,9 @@ pub use openai::{OpenAi, OpenAiConfig};
|
|||||||
pub use opensubtitles::{moviehash, OpenSubtitles, OpenSubtitlesConfig};
|
pub use opensubtitles::{moviehash, OpenSubtitles, OpenSubtitlesConfig};
|
||||||
pub use podnapisi::{Podnapisi, PodnapisiBuilder, DEFAULT_BASE_URL as PODNAPISI_DEFAULT_BASE_URL};
|
pub use podnapisi::{Podnapisi, PodnapisiBuilder, DEFAULT_BASE_URL as PODNAPISI_DEFAULT_BASE_URL};
|
||||||
pub use srt::Cue;
|
pub use srt::Cue;
|
||||||
pub use sync::{Outcome, Rejection, Syncer, DEFAULT_BINARY as ALASS_DEFAULT_BINARY};
|
pub use sync::{
|
||||||
|
Outcome, Rejection, Settled, SyncState, Syncer, DEFAULT_BINARY as ALASS_DEFAULT_BINARY,
|
||||||
|
};
|
||||||
pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, TranslatedCue};
|
pub use translate::{Backend, BackendId, Batch, BatchCue, TranslateFuture, TranslatedCue};
|
||||||
|
|
||||||
// `unused_crate_dependencies` is a per-target lint. Every provider and
|
// `unused_crate_dependencies` is a per-target lint. Every provider and
|
||||||
|
|||||||
+122
-1
@@ -131,6 +131,48 @@ impl Syncer {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Run [`Self::sync`] and settle the result into a form neither caller
|
||||||
|
/// has to branch on twice.
|
||||||
|
///
|
||||||
|
/// #199's manual grab handler and #196's reconcile loop both write a
|
||||||
|
/// fetched or translated subtitle, then call this before recording the
|
||||||
|
/// row — the seam issue #214 asked for, so the sync-then-classify logic
|
||||||
|
/// exists once. An [`Error`] from `sync` — a missing binary, a timeout, a
|
||||||
|
/// non-zero exit — degrades to [`SyncState::NotRun`] with a warning
|
||||||
|
/// logged, rather than failing the caller: a subtitle arr already has
|
||||||
|
/// beats one it refuses because `alass` could not run.
|
||||||
|
pub async fn settle(&self, video: impl AsRef<Path>, subtitle: impl AsRef<Path>) -> Settled {
|
||||||
|
let subtitle = subtitle.as_ref();
|
||||||
|
match self.sync(video, subtitle).await {
|
||||||
|
Ok(Outcome::Accepted { content }) => Settled {
|
||||||
|
content: Some(content),
|
||||||
|
state: SyncState::Synced,
|
||||||
|
},
|
||||||
|
Ok(Outcome::Rejected(reason)) => {
|
||||||
|
tracing::warn!(
|
||||||
|
subtitle = %subtitle.display(),
|
||||||
|
%reason,
|
||||||
|
"alass sync rejected as implausible, keeping the unsynced original"
|
||||||
|
);
|
||||||
|
Settled {
|
||||||
|
content: None,
|
||||||
|
state: SyncState::Rejected,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
Err(error) => {
|
||||||
|
tracing::warn!(
|
||||||
|
subtitle = %subtitle.display(),
|
||||||
|
%error,
|
||||||
|
"alass sync could not run, leaving the subtitle unsynced"
|
||||||
|
);
|
||||||
|
Settled {
|
||||||
|
content: None,
|
||||||
|
state: SyncState::NotRun,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Spawn `alass` and wait for it to write its output.
|
/// Spawn `alass` and wait for it to write its output.
|
||||||
async fn run(&self, video: &Path, subtitle: &Path, output: &Path) -> Result<()> {
|
async fn run(&self, video: &Path, subtitle: &Path, output: &Path) -> Result<()> {
|
||||||
let mut command = Command::new(&self.binary);
|
let mut command = Command::new(&self.binary);
|
||||||
@@ -177,6 +219,33 @@ impl Syncer {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// What settling a sync ended up recording, once an unusable `alass` and an
|
||||||
|
/// implausible result are both folded into "did not change anything" (§15).
|
||||||
|
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||||
|
pub enum SyncState {
|
||||||
|
/// `alass` could not be run to a plausible conclusion — binary missing,
|
||||||
|
/// timed out, or exited non-zero. Not the same as [`Self::Rejected`]:
|
||||||
|
/// nothing here says the *original* is implausible, only that nothing
|
||||||
|
/// checked.
|
||||||
|
NotRun,
|
||||||
|
/// Ran, and the result was accepted.
|
||||||
|
Synced,
|
||||||
|
/// Ran, and the result was implausible. The unsynced original stands and
|
||||||
|
/// the file is flagged.
|
||||||
|
Rejected,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The result of [`Syncer::settle`]: what belongs on disk, and what to
|
||||||
|
/// record.
|
||||||
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||||
|
pub struct Settled {
|
||||||
|
/// The synced text, only when it differs from what is already at the
|
||||||
|
/// subtitle path. `None` means the file `settle` was given is already
|
||||||
|
/// correct — the original, untouched.
|
||||||
|
pub content: Option<String>,
|
||||||
|
pub state: SyncState,
|
||||||
|
}
|
||||||
|
|
||||||
/// What one sync run decided.
|
/// What one sync run decided.
|
||||||
#[derive(Clone, Debug, PartialEq, Eq)]
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
||||||
pub enum Outcome {
|
pub enum Outcome {
|
||||||
@@ -250,7 +319,7 @@ mod tests {
|
|||||||
|
|
||||||
use tokio::io::AsyncWriteExt;
|
use tokio::io::AsyncWriteExt;
|
||||||
|
|
||||||
use super::{plausible, Outcome, Rejection, Syncer, MAX_SHIFT};
|
use super::{plausible, Outcome, Rejection, SyncState, Syncer, MAX_SHIFT};
|
||||||
use crate::error::Error;
|
use crate::error::Error;
|
||||||
use crate::srt::Cue;
|
use crate::srt::Cue;
|
||||||
|
|
||||||
@@ -505,6 +574,58 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn settle_returns_the_synced_content_to_write() {
|
||||||
|
let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
|
||||||
|
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 settled = Syncer::new()
|
||||||
|
.with_binary(&binary)
|
||||||
|
.settle(dir.path().join("video.mkv"), &subtitle)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
assert_eq!(settled.state, SyncState::Synced);
|
||||||
|
assert!(settled.content.is_some());
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn settle_on_rejection_keeps_the_original_and_flags_it() {
|
||||||
|
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 settled = Syncer::new()
|
||||||
|
.with_binary(&binary)
|
||||||
|
.settle(dir.path().join("video.mkv"), &subtitle)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
assert_eq!(settled.state, SyncState::Rejected);
|
||||||
|
assert_eq!(settled.content, None);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn settle_on_a_missing_binary_degrades_to_not_run() {
|
||||||
|
let dir = tempfile::tempdir().expect("test setup and fake binary succeed");
|
||||||
|
let subtitle = write_subtitle(dir.path(), "in.srt", PLAIN).await;
|
||||||
|
|
||||||
|
let settled = Syncer::new()
|
||||||
|
.with_binary("alass-not-installed")
|
||||||
|
.settle(dir.path().join("video.mkv"), &subtitle)
|
||||||
|
.await;
|
||||||
|
|
||||||
|
assert_eq!(settled.state, SyncState::NotRun);
|
||||||
|
assert_eq!(settled.content, None);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn rejections_render_as_a_sentence() {
|
fn rejections_render_as_a_sentence() {
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
|
|||||||
Reference in New Issue
Block a user