improve compute API
This commit is contained in:
Worph
@@ -6,10 +6,15 @@ gateway queries), not a meta-sort *plugin*. The enrichment plugin that attaches
|
||||
subtitles to local files is the separate `metamesh-plugin-opensubtitle`. See
|
||||
`docs/project-architecture/plugins-vs-feeders.md` in the meta-root.
|
||||
|
||||
**Search only.** Nothing is downloaded, so the account's download quota is never
|
||||
spent. Each result is one subtitle **file** addressed by a `provider-file`
|
||||
locator (`0x100A`, `docs/cid-formats.md` §8) that embeds
|
||||
`("opensubtitles", "file:<file_id>")`. Retrieving the bytes is meta-share's job.
|
||||
**Search is free.** Each result is one subtitle **file** addressed by a
|
||||
`provider-file` locator (`0x100A`, `docs/cid-formats.md` §8) that embeds
|
||||
`("opensubtitles", "file:<file_id>")`; searching never spends download quota.
|
||||
|
||||
**Downloads are redeems.** When a viewer plays a subtitle, the gateway asks this
|
||||
feeder to redeem the locator (`POST /compute`), which calls OpenSubtitles
|
||||
`/download` — one unit of the daily quota: the API key's anonymous quota, or an
|
||||
account's rank quota when a login is configured. The gateway stores the file, so
|
||||
each file is downloaded once. See [Downloads](#downloads).
|
||||
|
||||
## Queries
|
||||
|
||||
@@ -43,9 +48,37 @@ There is no `contentKind` and no `domain`, because a subtitle is a sidecar. Resu
|
||||
|---|---|---|
|
||||
| API key (required) | `apiKey` | `OPENSUBTITLES_API_KEY` |
|
||||
| User-Agent (`App vX.Y`) | `userAgent` | `OPENSUBTITLES_USER_AGENT` |
|
||||
| Account username (optional, opensubtitles.com) | `username` | `OPENSUBTITLES_USERNAME` |
|
||||
| Account password (optional) | `password` | `OPENSUBTITLES_PASSWORD` |
|
||||
|
||||
The config page (`/config`) wins over env. Changes apply on restart. Without a key
|
||||
the feeder stays up, reports degraded on `/health`, and answers nothing.
|
||||
the feeder stays up, reports degraded on `/health`, and answers nothing. Without a
|
||||
login it still downloads, on the API key's anonymous quota; a login (a
|
||||
**.com** account — .org accounts are a different API) switches to that account's
|
||||
rank quota.
|
||||
|
||||
## Downloads
|
||||
|
||||
`POST /compute {upstream_id: "opensubtitles", record_id}` where `record_id` is the
|
||||
`provider-file` cid (or `opensubtitles:file:<id>`):
|
||||
|
||||
1. Not an `opensubtitles` file locator → `404` (not ours).
|
||||
2. `POST /download {file_id}` with the `Api-Key` header → a short-lived link,
|
||||
fetched by a second client that follows redirects and sends only the
|
||||
User-Agent. With a login configured the call also carries a Bearer token from
|
||||
`POST /login` (cached; re-login once on a `401`/`403`).
|
||||
3. Answer ONE `sha2_256` outcome: the bytes, `file_extension` from the file name
|
||||
(`srt` when unknown), and a record holding only what the download knows —
|
||||
`extension`, `fileName`, `sizeByte`. The gateway merges those onto the search
|
||||
record (which already has title, language and ids) next to the `file` pointer.
|
||||
|
||||
Quota: a `406`, or `remaining: 0` on the previous download, answers `429` with the
|
||||
reset time (`reset_time_utc`, else `reset_time`, else 1 h) and makes no further
|
||||
`/download` call until then. A file under 10 bytes → `422`.
|
||||
|
||||
`GET /redeems` and `/manifest` advertise `{codec: provider-file, field: file,
|
||||
sources: ["opensubtitles"]}` whenever an API key is configured; `package` is
|
||||
`meta-feeder-opensubtitles` (`/files/plugin/meta-feeder-opensubtitles/`).
|
||||
|
||||
Also `META_FEEDER_HTTP_LISTEN` (default `0.0.0.0:8080`), `META_FEEDER_STATE_DIR`
|
||||
(default `/data/meta-feeder`), and `RUST_LOG`.
|
||||
@@ -70,3 +103,18 @@ docker build -f feeder-plugin/opensubtitles-feeder/Dockerfile -t meta-feeder-ope
|
||||
|
||||
`tests/contract.rs` drives the real SDK router against a wiremock OpenSubtitles,
|
||||
using fixtures captured from the live API.
|
||||
|
||||
The SDK is pinned by tag (`v1.3.0`). Until that tag is pushed, cargo cannot
|
||||
resolve it — not even under a `[patch]` override — so the recipe above fails. To
|
||||
verify against a local SDK checkout, run it on a scratch copy with the
|
||||
dependency swapped for a path (never commit that form):
|
||||
|
||||
```bash
|
||||
SCRATCH=/d/workspace/tmp-claude/os-feeder; SDK=/d/workspace/MetaMesh/meta-root-v2/packages/plugins/meta-feeder-sdk
|
||||
rm -rf $SCRATCH && mkdir -p $SCRATCH && tar cf - --exclude=./target --exclude=./.git . | (cd $SCRATCH && tar xf -)
|
||||
sed -i "s#meta-feeder-sdk = { git = .*#meta-feeder-sdk = { path = \"$SDK\" }#" $SCRATCH/feeder-plugin/*/Cargo.toml
|
||||
# then the docker `cargo test` above with -v /d/workspace:/d/workspace -w $SCRATCH
|
||||
```
|
||||
|
||||
After the tag is pushed, run `cargo update -p meta-feeder-sdk` to move
|
||||
`Cargo.lock` off the v1.2.1 revision.
|
||||
|
||||
@@ -6,9 +6,10 @@ rust-version.workspace = true
|
||||
license.workspace = true
|
||||
publish = false
|
||||
|
||||
# OpenSubtitles subtitle search. Answers `fileType:subtitle` queries — keyword
|
||||
# and id-anchored — with `provider-file` (0x100A) locator records. Search only:
|
||||
# no download, no quota spend. Retrieval is meta-share's job.
|
||||
# OpenSubtitles subtitles. Answers `fileType:subtitle` queries — keyword and
|
||||
# id-anchored — with `provider-file` (0x100A) locator records (free), and
|
||||
# REDEEMS those locators through `/compute` with an account login (`/download`,
|
||||
# which spends the account's download quota) when the gateway asks on a play.
|
||||
|
||||
[lib]
|
||||
name = "opensubtitles_feeder"
|
||||
@@ -21,7 +22,7 @@ path = "src/main.rs"
|
||||
[dependencies]
|
||||
# Pinned by TAG, never by path. A path dep resolves OUTSIDE this repo, and this
|
||||
# repo is the Docker build context — see meta-feeder-anilist's manifest.
|
||||
meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.2.1" }
|
||||
meta-feeder-sdk = { git = "https://github.com/worph/meta-feeder-sdk", tag = "v1.3.0" }
|
||||
|
||||
tokio = { version = "1.40", features = ["full"] }
|
||||
axum = { version = "0.7", features = ["json"] }
|
||||
@@ -36,4 +37,5 @@ tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
|
||||
[dev-dependencies]
|
||||
wiremock = "0.6"
|
||||
tempfile = "3"
|
||||
base64 = "0.22"
|
||||
tokio = { version = "1.40", features = ["full", "test-util"] }
|
||||
|
||||
@@ -1,15 +1,18 @@
|
||||
//! OpenSubtitles REST v1 search client — `GET /subtitles` only.
|
||||
//! OpenSubtitles REST v1 client — `GET /subtitles`, plus `POST /login` and
|
||||
//! `POST /download` for redeeming a file.
|
||||
//!
|
||||
//! Search is free: OpenSubtitles meters `/download`, which this feeder never
|
||||
//! calls. What *is* limited is the request rate (5/s, measured), hence the
|
||||
//! shared [`RateBudget`].
|
||||
//! Search is free. `/download` is **metered per account** (a daily quota), so
|
||||
//! it runs only from `compute_outcomes`, which the gateway calls on a real play
|
||||
//! and at most once per file (it stores the bytes). Every API call shares the
|
||||
//! request-rate [`RateBudget`] (5/s, measured).
|
||||
//!
|
||||
//! ⚠ **OpenSubtitles 301s any query string that is not in its canonical form**
|
||||
//! (keys sorted, values lowercased). Redirects are disabled on this client, so
|
||||
//! (keys sorted, values lowercased). Redirects are disabled on the API client, so
|
||||
//! a drift in [`SearchParams::to_pairs`] surfaces as a loud `Permanent` error
|
||||
//! instead of a silent extra round-trip per search.
|
||||
//! instead of a silent extra round-trip per search. The short-lived download
|
||||
//! *link* is fetched by a second client that does follow redirects.
|
||||
|
||||
use std::sync::Arc;
|
||||
use std::sync::{Arc, Mutex as StdMutex};
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
use meta_feeder_sdk::budget::{Lease, RateBudget};
|
||||
@@ -17,9 +20,12 @@ use meta_feeder_sdk::cache::{search_key, MidhashCache};
|
||||
use meta_feeder_sdk::common::{build_http_client, map_status};
|
||||
use meta_feeder_sdk::types::GatewayError;
|
||||
use serde::{Deserialize, Deserializer};
|
||||
use tokio::sync::Mutex;
|
||||
use tracing::debug;
|
||||
|
||||
use crate::consts::{HTTP_TIMEOUT_SECS, SEARCH_CACHE_TTL_SECS, SEARCH_DEADLINE_SECS};
|
||||
use crate::consts::{
|
||||
DEFAULT_QUOTA_RETRY_SECS, HTTP_TIMEOUT_SECS, SEARCH_CACHE_TTL_SECS, SEARCH_DEADLINE_SECS,
|
||||
};
|
||||
|
||||
/// One search, already resolved from a gateway query.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
@@ -228,14 +234,129 @@ fn fresh_body(entry: &str, now: u64, ttl_secs: u64) -> Option<String> {
|
||||
v.get("body")?.as_str().map(str::to_string)
|
||||
}
|
||||
|
||||
// --- download shapes --------------------------------------------------------
|
||||
|
||||
#[derive(Debug, Clone, Default, Deserialize)]
|
||||
struct LoginResponse {
|
||||
#[serde(default)]
|
||||
token: Option<String>,
|
||||
}
|
||||
|
||||
/// `POST /download` answer (also the body of its `406` quota refusal).
|
||||
#[derive(Debug, Clone, Default, Deserialize)]
|
||||
#[serde(default)]
|
||||
pub struct DownloadResponse {
|
||||
pub link: Option<String>,
|
||||
pub file_name: Option<String>,
|
||||
#[serde(deserialize_with = "lenient_u64")]
|
||||
pub remaining: Option<u64>,
|
||||
/// Human text, e.g. `23 hours and 59 minutes`.
|
||||
pub reset_time: Option<String>,
|
||||
/// e.g. `2022-04-08T13:03:16.000Z`.
|
||||
pub reset_time_utc: Option<String>,
|
||||
pub message: Option<String>,
|
||||
}
|
||||
|
||||
/// One redeemed subtitle file.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Downloaded {
|
||||
pub bytes: Vec<u8>,
|
||||
pub file_name: Option<String>,
|
||||
pub remaining: Option<u64>,
|
||||
}
|
||||
|
||||
/// Seconds until the download quota resets: `reset_time_utc` if parseable,
|
||||
/// else `reset_time` text, else [`DEFAULT_QUOTA_RETRY_SECS`]. Clamped to a day.
|
||||
pub fn quota_retry_secs(reset_time_utc: Option<&str>, reset_time: Option<&str>, now: u64) -> u32 {
|
||||
let secs = reset_time_utc
|
||||
.and_then(parse_rfc3339_utc)
|
||||
.filter(|at| *at > now)
|
||||
.map(|at| at - now)
|
||||
.or_else(|| reset_time.and_then(parse_duration_text).filter(|s| *s > 0));
|
||||
match secs {
|
||||
Some(s) => s.clamp(1, 86_400) as u32,
|
||||
None => DEFAULT_QUOTA_RETRY_SECS,
|
||||
}
|
||||
}
|
||||
|
||||
/// `YYYY-MM-DDTHH:MM:SS[.fff](Z|+00:00)` → unix seconds. UTC only — which is
|
||||
/// all OpenSubtitles sends in `reset_time_utc`.
|
||||
fn parse_rfc3339_utc(s: &str) -> Option<u64> {
|
||||
let s = s.trim();
|
||||
let (date, time) = s.split_once('T')?;
|
||||
let time = time
|
||||
.strip_suffix('Z')
|
||||
.or_else(|| time.strip_suffix("+00:00"))?;
|
||||
let time = time.split('.').next()?;
|
||||
let mut d = date.split('-').map(|p| p.parse::<i64>());
|
||||
let (y, mo, da) = (d.next()?.ok()?, d.next()?.ok()?, d.next()?.ok()?);
|
||||
let mut t = time.split(':').map(|p| p.parse::<i64>());
|
||||
let (h, mi, se) = (t.next()?.ok()?, t.next()?.ok()?, t.next()?.ok()?);
|
||||
if !(1..=12).contains(&mo) || !(1..=31).contains(&da) || h > 23 || mi > 59 || se > 60 {
|
||||
return None;
|
||||
}
|
||||
let secs = days_from_civil(y, mo, da) * 86_400 + h * 3600 + mi * 60 + se;
|
||||
u64::try_from(secs).ok()
|
||||
}
|
||||
|
||||
/// Days since 1970-01-01 for a proleptic Gregorian date (Howard Hinnant).
|
||||
fn days_from_civil(y: i64, m: i64, d: i64) -> i64 {
|
||||
let y = if m <= 2 { y - 1 } else { y };
|
||||
let era = if y >= 0 { y } else { y - 399 } / 400;
|
||||
let yoe = y - era * 400;
|
||||
let mp = (m + 9) % 12;
|
||||
let doy = (153 * mp + 2) / 5 + d - 1;
|
||||
let doe = yoe * 365 + yoe / 4 - yoe / 100 + doy;
|
||||
era * 146_097 + doe - 719_468
|
||||
}
|
||||
|
||||
/// `23 hours and 59 minutes` → seconds. Unknown shapes → `None`.
|
||||
fn parse_duration_text(s: &str) -> Option<u64> {
|
||||
let mut total = 0u64;
|
||||
let mut pending: Option<u64> = None;
|
||||
let mut matched = false;
|
||||
for word in s.split(|c: char| !c.is_ascii_alphanumeric()).filter(|w| !w.is_empty()) {
|
||||
if let Ok(n) = word.parse::<u64>() {
|
||||
pending = Some(n);
|
||||
continue;
|
||||
}
|
||||
let Some(n) = pending.take() else { continue };
|
||||
let unit = word.to_ascii_lowercase();
|
||||
let mult = if unit.starts_with("hour") {
|
||||
3600
|
||||
} else if unit.starts_with("minute") {
|
||||
60
|
||||
} else if unit.starts_with("second") {
|
||||
1
|
||||
} else {
|
||||
continue;
|
||||
};
|
||||
total += n * mult;
|
||||
matched = true;
|
||||
}
|
||||
matched.then_some(total)
|
||||
}
|
||||
|
||||
// --- client -----------------------------------------------------------------
|
||||
|
||||
pub struct OsClient {
|
||||
http: reqwest::Client,
|
||||
/// Fetches the pre-signed download link: redirects allowed (it may bounce
|
||||
/// to a CDN), User-Agent only — no Api-Key, no Bearer.
|
||||
download_http: reqwest::Client,
|
||||
api_base: String,
|
||||
api_key: String,
|
||||
budget: Arc<RateBudget>,
|
||||
cache: Option<MidhashCache>,
|
||||
/// Optional account login for `/download`. `None` → downloads run on the
|
||||
/// API key alone (OpenSubtitles' anonymous quota); `Some` → a Bearer token
|
||||
/// from `/login`, so the account's rank quota applies.
|
||||
login: Option<(String, String)>,
|
||||
/// Cached JWT from `/login`; dropped and re-fetched once on a 401/403.
|
||||
token: Mutex<Option<String>>,
|
||||
/// Unix time until which the download quota is known to be spent — no
|
||||
/// `/download` call is made before it.
|
||||
quota_until: StdMutex<Option<u64>>,
|
||||
}
|
||||
|
||||
impl OsClient {
|
||||
@@ -251,15 +372,209 @@ impl OsClient {
|
||||
user_agent,
|
||||
Some(reqwest::redirect::Policy::none()),
|
||||
);
|
||||
let download_http = build_http_client(HTTP_TIMEOUT_SECS, user_agent, None);
|
||||
Self {
|
||||
http,
|
||||
download_http,
|
||||
api_base: api_base.trim_end_matches('/').to_string(),
|
||||
api_key,
|
||||
budget,
|
||||
cache,
|
||||
login: None,
|
||||
token: Mutex::new(None),
|
||||
quota_until: StdMutex::new(None),
|
||||
}
|
||||
}
|
||||
|
||||
/// Download as an account instead of on the bare API key. Blank credentials
|
||||
/// keep the anonymous (API key only) mode.
|
||||
pub fn with_login(mut self, username: &str, password: &str) -> Self {
|
||||
let (u, p) = (username.trim(), password.trim());
|
||||
if !u.is_empty() && !p.is_empty() {
|
||||
self.login = Some((u.to_string(), p.to_string()));
|
||||
}
|
||||
self
|
||||
}
|
||||
|
||||
/// True when downloads go through an account login rather than the bare
|
||||
/// API key.
|
||||
pub fn has_login(&self) -> bool {
|
||||
self.login.is_some()
|
||||
}
|
||||
|
||||
/// One token from the shared request-rate budget. Busy is Transient.
|
||||
async fn api_lease(&self) -> Result<(), GatewayError> {
|
||||
if matches!(
|
||||
self.budget.acquire(Duration::from_secs(SEARCH_DEADLINE_SECS)).await,
|
||||
Lease::DeadlineExceeded
|
||||
) {
|
||||
return Err(GatewayError::Transient(
|
||||
"opensubtitles: rate budget deadline exceeded".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Freeze the whole bucket on a 429, not just this caller.
|
||||
fn note_error(&self, e: &GatewayError) {
|
||||
if let GatewayError::RateLimited { retry_after_s } = e {
|
||||
self.budget
|
||||
.note_throttled(Duration::from_secs(u64::from(*retry_after_s)));
|
||||
}
|
||||
}
|
||||
|
||||
fn quota_retry_after(&self, now: u64) -> Option<u32> {
|
||||
let until = (*self.quota_until.lock().expect("quota mutex"))?;
|
||||
(until > now).then(|| (until - now).min(u64::from(u32::MAX)) as u32)
|
||||
}
|
||||
|
||||
fn block_quota_for(&self, secs: u32, now: u64) {
|
||||
*self.quota_until.lock().expect("quota mutex") = Some(now + u64::from(secs));
|
||||
}
|
||||
|
||||
async fn login(&self) -> Result<String, GatewayError> {
|
||||
let Some((username, password)) = &self.login else {
|
||||
return Err(GatewayError::Permanent(
|
||||
"opensubtitles: login requested without configured credentials".into(),
|
||||
));
|
||||
};
|
||||
self.api_lease().await?;
|
||||
let resp = self
|
||||
.http
|
||||
.post(format!("{}/login", self.api_base))
|
||||
.header("Api-Key", &self.api_key)
|
||||
.header(reqwest::header::ACCEPT, "application/json")
|
||||
.json(&serde_json::json!({ "username": username, "password": password }))
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| GatewayError::Transient(format!("opensubtitles login: {e}")))?;
|
||||
if let Err(e) = map_status(&resp) {
|
||||
self.note_error(&e);
|
||||
return Err(match e {
|
||||
// A missing login endpoint is a broken config, not "not mine".
|
||||
GatewayError::NotFound => {
|
||||
GatewayError::Permanent("opensubtitles login: endpoint not found".into())
|
||||
}
|
||||
other => other,
|
||||
});
|
||||
}
|
||||
let body: LoginResponse = resp
|
||||
.json()
|
||||
.await
|
||||
.map_err(|e| GatewayError::Transient(format!("opensubtitles login: bad json: {e}")))?;
|
||||
body.token
|
||||
.filter(|t| !t.is_empty())
|
||||
.ok_or_else(|| GatewayError::Permanent("opensubtitles login returned no token".into()))
|
||||
}
|
||||
|
||||
/// The cached token, logging in when there is none (or `refresh` drops it).
|
||||
async fn bearer(&self, refresh: bool) -> Result<String, GatewayError> {
|
||||
let mut slot = self.token.lock().await;
|
||||
if refresh {
|
||||
*slot = None;
|
||||
}
|
||||
if let Some(t) = slot.as_ref() {
|
||||
return Ok(t.clone());
|
||||
}
|
||||
let t = self.login().await?;
|
||||
*slot = Some(t.clone());
|
||||
Ok(t)
|
||||
}
|
||||
|
||||
async fn post_download(&self, token: Option<&str>, file_id: u64) -> Result<reqwest::Response, GatewayError> {
|
||||
self.api_lease().await?;
|
||||
let mut req = self
|
||||
.http
|
||||
.post(format!("{}/download", self.api_base))
|
||||
.header("Api-Key", &self.api_key)
|
||||
.header(reqwest::header::ACCEPT, "application/json")
|
||||
.json(&serde_json::json!({ "file_id": file_id }));
|
||||
if let Some(token) = token {
|
||||
req = req.bearer_auth(token);
|
||||
}
|
||||
req.send()
|
||||
.await
|
||||
.map_err(|e| GatewayError::Transient(format!("opensubtitles download: {e}")))
|
||||
}
|
||||
|
||||
/// Redeem one subtitle file: `/download` for a short-lived link, then fetch
|
||||
/// it. ⚠ Spends one unit of the daily download quota — the API key's
|
||||
/// anonymous quota, or the account's when a login is configured.
|
||||
///
|
||||
/// A spent quota (`406`, or `remaining: 0` on the previous call) is
|
||||
/// `RateLimited` with the reset time, and blocks further `/download` calls
|
||||
/// until then.
|
||||
pub async fn download(&self, file_id: u64) -> Result<Downloaded, GatewayError> {
|
||||
if let Some(retry_after_s) = self.quota_retry_after(now_secs()) {
|
||||
return Err(GatewayError::RateLimited { retry_after_s });
|
||||
}
|
||||
|
||||
let resp = if self.has_login() {
|
||||
let token = self.bearer(false).await?;
|
||||
let resp = self.post_download(Some(&token), file_id).await?;
|
||||
if matches!(resp.status().as_u16(), 401 | 403) {
|
||||
debug!(target: "opensubtitles", status = %resp.status(), "download refused the token; logging in again");
|
||||
let token = self.bearer(true).await?;
|
||||
self.post_download(Some(&token), file_id).await?
|
||||
} else {
|
||||
resp
|
||||
}
|
||||
} else {
|
||||
self.post_download(None, file_id).await?
|
||||
};
|
||||
if resp.status().as_u16() == 406 {
|
||||
let body: DownloadResponse = resp.json().await.unwrap_or_default();
|
||||
let now = now_secs();
|
||||
let retry = quota_retry_secs(body.reset_time_utc.as_deref(), body.reset_time.as_deref(), now);
|
||||
self.block_quota_for(retry, now);
|
||||
return Err(GatewayError::RateLimited { retry_after_s: retry });
|
||||
}
|
||||
if let Err(e) = map_status(&resp) {
|
||||
self.note_error(&e);
|
||||
return Err(e);
|
||||
}
|
||||
let body: DownloadResponse = resp
|
||||
.json()
|
||||
.await
|
||||
.map_err(|e| GatewayError::Transient(format!("opensubtitles download: bad json: {e}")))?;
|
||||
let link = body
|
||||
.link
|
||||
.clone()
|
||||
.filter(|l| !l.is_empty())
|
||||
.ok_or_else(|| GatewayError::Transient("opensubtitles download returned no link".into()))?;
|
||||
if body.remaining == Some(0) {
|
||||
// This file is served; the next one would be refused.
|
||||
let now = now_secs();
|
||||
let retry = quota_retry_secs(body.reset_time_utc.as_deref(), body.reset_time.as_deref(), now);
|
||||
self.block_quota_for(retry, now);
|
||||
}
|
||||
|
||||
// The link is pre-signed: keep it out of error messages.
|
||||
let got = self
|
||||
.download_http
|
||||
.get(&link)
|
||||
.send()
|
||||
.await
|
||||
.map_err(|e| GatewayError::Transient(format!("opensubtitles file fetch: {}", e.without_url())))?;
|
||||
let status = got.status();
|
||||
if !status.is_success() {
|
||||
return Err(if status.is_server_error() {
|
||||
GatewayError::Transient(format!("opensubtitles file fetch: HTTP {status}"))
|
||||
} else {
|
||||
GatewayError::Permanent(format!("opensubtitles file fetch: HTTP {status}"))
|
||||
});
|
||||
}
|
||||
let bytes = got
|
||||
.bytes()
|
||||
.await
|
||||
.map_err(|e| GatewayError::Transient(format!("opensubtitles file fetch: {}", e.without_url())))?;
|
||||
Ok(Downloaded {
|
||||
bytes: bytes.to_vec(),
|
||||
file_name: body.file_name,
|
||||
remaining: body.remaining,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn search(&self, params: &SearchParams) -> Result<Vec<OsSubtitle>, GatewayError> {
|
||||
let pairs = params.to_pairs();
|
||||
let key = cache_key(&pairs);
|
||||
@@ -376,6 +691,21 @@ mod tests {
|
||||
assert_eq!(fresh_body("garbage", 0, 60), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn quota_reset_prefers_the_utc_timestamp() {
|
||||
// 2022-04-08T13:03:16Z = 1649422996.
|
||||
assert_eq!(parse_rfc3339_utc("2022-04-08T13:03:16.000Z"), Some(1_649_422_996));
|
||||
assert_eq!(parse_rfc3339_utc("1970-01-01T00:00:00Z"), Some(0));
|
||||
assert_eq!(parse_rfc3339_utc("yesterday"), None);
|
||||
let now = 1_649_422_996 - 90;
|
||||
assert_eq!(quota_retry_secs(Some("2022-04-08T13:03:16.000Z"), None, now), 90);
|
||||
// Falls back to the text, then to the default.
|
||||
assert_eq!(quota_retry_secs(None, Some("23 hours and 59 minutes"), now), 86_340);
|
||||
assert_eq!(quota_retry_secs(Some("garbage"), Some("soon"), now), DEFAULT_QUOTA_RETRY_SECS);
|
||||
// A reset already in the past is not a retry time.
|
||||
assert_eq!(quota_retry_secs(Some("1970-01-01T00:00:00Z"), None, now), DEFAULT_QUOTA_RETRY_SECS);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn lenient_numbers_survive_odd_upstream_values() {
|
||||
let a: OsAttributes = serde_json::from_str(
|
||||
|
||||
@@ -32,6 +32,22 @@ pub(crate) const HTTP_TIMEOUT_SECS: u64 = 20;
|
||||
/// and download counts, and download count is half of the ranking.
|
||||
pub(crate) const SEARCH_CACHE_TTL_SECS: u64 = 24 * 60 * 60;
|
||||
|
||||
/// `/files/plugin/<PACKAGE>/` — where the gateway stores downloaded subtitles.
|
||||
pub const PACKAGE: &str = "meta-feeder-opensubtitles";
|
||||
|
||||
/// `retry_after_s` when the download quota is spent and OpenSubtitles gave no
|
||||
/// parseable reset time. The quota is daily.
|
||||
pub(crate) const DEFAULT_QUOTA_RETRY_SECS: u32 = 3600;
|
||||
|
||||
/// A "subtitle" shorter than this is an error page or an empty file, not a
|
||||
/// subtitle — refuse it rather than seed it to the mesh.
|
||||
pub(crate) const MIN_SUBTITLE_BYTES: usize = 10;
|
||||
|
||||
/// Extension used when the downloaded file name has no known subtitle suffix.
|
||||
pub(crate) const DEFAULT_SUBTITLE_EXTENSION: &str = "srt";
|
||||
|
||||
/// Env seeds; the config page's saved values win over them.
|
||||
pub(crate) const ENV_API_KEY: &str = "OPENSUBTITLES_API_KEY";
|
||||
pub(crate) const ENV_USER_AGENT: &str = "OPENSUBTITLES_USER_AGENT";
|
||||
pub(crate) const ENV_USERNAME: &str = "OPENSUBTITLES_USERNAME";
|
||||
pub(crate) const ENV_PASSWORD: &str = "OPENSUBTITLES_PASSWORD";
|
||||
|
||||
@@ -1,10 +1,13 @@
|
||||
//! `opensubtitles-feeder` — subtitle search over OpenSubtitles.
|
||||
//! `opensubtitles-feeder` — subtitle search and download over OpenSubtitles.
|
||||
//!
|
||||
//! A **sidecar** tier: it answers "which subtitles exist for X", never "give me
|
||||
//! the bytes". Each result is one subtitle *file*, addressed by a
|
||||
//! `provider-file` locator (`0x100A`, `docs/cid-formats.md` §8) that embeds
|
||||
//! `("opensubtitles", "file:<file_id>")`. meta-share resolves that locator later;
|
||||
//! nothing here downloads, so searching never spends the account's download quota.
|
||||
//! A **sidecar** tier. Search answers "which subtitles exist for X": each result
|
||||
//! is one subtitle *file*, addressed by a `provider-file` locator (`0x100A`,
|
||||
//! `docs/cid-formats.md` §8) that embeds `("opensubtitles", "file:<file_id>")`.
|
||||
//! Searching never spends the account's download quota.
|
||||
//!
|
||||
//! Redeeming a locator — `POST /compute` with the cid, which the gateway sends
|
||||
//! on a real play — logs in and calls `/download`, spending one unit of quota;
|
||||
//! the gateway stores the bytes so each file is downloaded once.
|
||||
//!
|
||||
//! ## The contracts that must not drift
|
||||
//!
|
||||
|
||||
@@ -10,24 +10,15 @@
|
||||
//! digest : varint(len(source)) ‖ source ‖ id (both UTF-8, ≤ 64 bytes)
|
||||
//! ```
|
||||
//!
|
||||
//! The framing is byte-identical to `card` (`0x1007`) — only the codec differs —
|
||||
//! which is why the tests below re-derive the existing `card-locator` vector
|
||||
//! with the same encoder.
|
||||
//!
|
||||
//! ⚠ **This is a local copy.** The SDK's framing helpers (`kinded_locator_cid`,
|
||||
//! `write_pb_varint`, `base32_lower_no_padding`) are private, so it could not be
|
||||
//! reused. Move it into `meta-feeder-sdk` as `encode_source_id_locator(codec,
|
||||
//! source, id)` (§8 trap 7) and delete this file; the golden vectors are what
|
||||
//! prove the move changed no bytes.
|
||||
//! The encoder/decoder live in `meta-feeder-sdk` (`hash::compute_provider_file_cid`
|
||||
//! / `decode_provider_file_cid`, with the §8.2 golden vector); this module only
|
||||
//! binds them to OpenSubtitles' `file:<file_id>` id form.
|
||||
|
||||
use crate::consts::UPSTREAM_ID;
|
||||
|
||||
/// Custom multicodec for a fetchable `(source, id)` locator.
|
||||
pub const PROVIDER_FILE_CODEC: u64 = 0x100A;
|
||||
|
||||
/// meta-share's `MAX_MULTIHASH_SIZE` — its `CidGeneric<64>` rejects a longer
|
||||
/// multihash, so a longer digest would mint a CID no peer can parse.
|
||||
const MAX_DIGEST_BYTES: usize = 64;
|
||||
pub use meta_feeder_sdk::hash::{
|
||||
compute_provider_file_cid as provider_file_cid, decode_provider_file_cid, PROVIDER_FILE_CODEC,
|
||||
};
|
||||
|
||||
/// The locator for one OpenSubtitles **file** (`/download` takes a `file_id`;
|
||||
/// one subtitle listing can hold several files, one per CD).
|
||||
@@ -35,156 +26,28 @@ pub fn file_cid(file_id: u64) -> Option<String> {
|
||||
provider_file_cid(UPSTREAM_ID, &format!("file:{file_id}"))
|
||||
}
|
||||
|
||||
/// Encode a `provider-file` CID. `None` when the digest would exceed 64 bytes —
|
||||
/// the caller drops that row rather than emit an unparseable address.
|
||||
pub fn provider_file_cid(source: &str, id: &str) -> Option<String> {
|
||||
source_id_locator(PROVIDER_FILE_CODEC, source, id)
|
||||
}
|
||||
|
||||
/// Inverse of [`provider_file_cid`]: `(source, id)`, or `None` for anything that
|
||||
/// is not exactly a well-formed `0x100A` CID. Matches the **codec slot** — a
|
||||
/// `card` CID has the same framing and must not decode here.
|
||||
pub fn decode_provider_file_cid(cid: &str) -> Option<(String, String)> {
|
||||
let wire = base32_lower_no_pad_decode(cid.strip_prefix('b')?)?;
|
||||
let mut pos = 0;
|
||||
if read_varint(&wire, &mut pos)? != 1 {
|
||||
/// The OpenSubtitles `file_id` a `/compute` record id names, if it is ours.
|
||||
///
|
||||
/// Accepts both shapes a caller holds: the `provider-file` cid (what the
|
||||
/// gateway's redeem route sends) and this feeder's own record id
|
||||
/// `opensubtitles:file:<id>`. A cid for another source is `None` — not ours.
|
||||
pub fn file_id_of(record_id: &str) -> Option<u64> {
|
||||
if let Some(id) = record_id.strip_prefix(&format!("{UPSTREAM_ID}:file:")) {
|
||||
return id.parse().ok();
|
||||
}
|
||||
let (source, id) = decode_provider_file_cid(record_id)?;
|
||||
if source != UPSTREAM_ID {
|
||||
return None;
|
||||
}
|
||||
if read_varint(&wire, &mut pos)? != PROVIDER_FILE_CODEC {
|
||||
return None;
|
||||
}
|
||||
if read_varint(&wire, &mut pos)? != 0x00 {
|
||||
return None;
|
||||
}
|
||||
let len = read_varint(&wire, &mut pos)? as usize;
|
||||
let digest = wire.get(pos..)?;
|
||||
if digest.len() != len {
|
||||
return None;
|
||||
}
|
||||
let mut dpos = 0;
|
||||
let source_len = read_varint(digest, &mut dpos)? as usize;
|
||||
let source = digest.get(dpos..dpos.checked_add(source_len)?)?;
|
||||
let id = digest.get(dpos + source_len..)?;
|
||||
Some((
|
||||
String::from_utf8(source.to_vec()).ok()?,
|
||||
String::from_utf8(id.to_vec()).ok()?,
|
||||
))
|
||||
}
|
||||
|
||||
fn source_id_locator(codec: u64, source: &str, id: &str) -> Option<String> {
|
||||
let source = source.as_bytes();
|
||||
let id = id.as_bytes();
|
||||
|
||||
// identity-multihash digest = varint(source_len) ‖ source ‖ id
|
||||
let mut digest = Vec::with_capacity(2 + source.len() + id.len());
|
||||
write_varint(source.len() as u64, &mut digest);
|
||||
digest.extend_from_slice(source);
|
||||
digest.extend_from_slice(id);
|
||||
if digest.len() > MAX_DIGEST_BYTES {
|
||||
return None;
|
||||
}
|
||||
|
||||
// CIDv1: [version=0x01][codec varint][mh code=0x00 identity][len varint][digest]
|
||||
let mut wire = Vec::with_capacity(6 + digest.len());
|
||||
wire.push(0x01);
|
||||
write_varint(codec, &mut wire);
|
||||
wire.push(0x00);
|
||||
write_varint(digest.len() as u64, &mut wire);
|
||||
wire.extend_from_slice(&digest);
|
||||
Some(format!("b{}", base32_lower_no_pad(&wire)))
|
||||
}
|
||||
|
||||
fn write_varint(mut value: u64, out: &mut Vec<u8>) {
|
||||
loop {
|
||||
let byte = (value & 0x7f) as u8;
|
||||
value >>= 7;
|
||||
if value == 0 {
|
||||
out.push(byte);
|
||||
return;
|
||||
}
|
||||
out.push(byte | 0x80);
|
||||
}
|
||||
}
|
||||
|
||||
fn read_varint(bytes: &[u8], pos: &mut usize) -> Option<u64> {
|
||||
let mut value = 0u64;
|
||||
let mut shift = 0u32;
|
||||
loop {
|
||||
let byte = *bytes.get(*pos)?;
|
||||
*pos += 1;
|
||||
value |= u64::from(byte & 0x7f) << shift;
|
||||
if byte & 0x80 == 0 {
|
||||
return Some(value);
|
||||
}
|
||||
shift += 7;
|
||||
if shift >= 64 {
|
||||
return None;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const BASE32_ALPHABET: &[u8; 32] = b"abcdefghijklmnopqrstuvwxyz234567";
|
||||
|
||||
fn base32_lower_no_pad(input: &[u8]) -> String {
|
||||
let mut out = String::with_capacity((input.len() * 8).div_ceil(5));
|
||||
let mut acc = 0u64;
|
||||
let mut bits = 0u32;
|
||||
for &byte in input {
|
||||
acc = (acc << 8) | u64::from(byte);
|
||||
bits += 8;
|
||||
while bits >= 5 {
|
||||
bits -= 5;
|
||||
out.push(BASE32_ALPHABET[((acc >> bits) & 31) as usize] as char);
|
||||
}
|
||||
acc &= (1u64 << bits) - 1;
|
||||
}
|
||||
if bits > 0 {
|
||||
out.push(BASE32_ALPHABET[((acc << (5 - bits)) & 31) as usize] as char);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
fn base32_lower_no_pad_decode(input: &str) -> Option<Vec<u8>> {
|
||||
let mut out = Vec::with_capacity(input.len() * 5 / 8);
|
||||
let mut acc = 0u64;
|
||||
let mut bits = 0u32;
|
||||
for c in input.bytes() {
|
||||
let v = match c {
|
||||
b'a'..=b'z' => c - b'a',
|
||||
b'2'..=b'7' => c - b'2' + 26,
|
||||
_ => return None,
|
||||
};
|
||||
acc = (acc << 5) | u64::from(v);
|
||||
bits += 5;
|
||||
if bits >= 8 {
|
||||
bits -= 8;
|
||||
out.push((acc >> bits) as u8);
|
||||
acc &= (1u64 << bits) - 1;
|
||||
}
|
||||
}
|
||||
Some(out)
|
||||
id.strip_prefix("file:")?.parse().ok()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
/// The same framing under `card`'s codec reproduces the existing
|
||||
/// `card-locator` golden vector — proof the encoder is the family's framing,
|
||||
/// not a look-alike.
|
||||
#[test]
|
||||
fn framing_reproduces_the_card_locator_vector() {
|
||||
assert_eq!(
|
||||
source_id_locator(0x1007, "tmdb", "tv:95479").as_deref(),
|
||||
Some("bagdsaaanar2g2zdcor3duojvgq3ts")
|
||||
);
|
||||
assert_eq!(
|
||||
source_id_locator(0x1008, "video", "4D7u5KF7SP8").as_deref(),
|
||||
Some("bagecaaarav3gszdfn42ein3vgvfumn2tka4a")
|
||||
);
|
||||
}
|
||||
|
||||
/// `docs/cid-formats.md` §8.2 worked example.
|
||||
/// `docs/cid-formats.md` §8.2 worked example — still minted identically now
|
||||
/// that the encoder lives in the SDK.
|
||||
#[test]
|
||||
fn file_cid_matches_the_documented_vector() {
|
||||
assert_eq!(
|
||||
@@ -203,13 +66,15 @@ mod tests {
|
||||
// Identical framing, different codec: a card must not decode as a file.
|
||||
assert_eq!(decode_provider_file_cid("bagdsaaanar2g2zdcor3duojvgq3ts"), None);
|
||||
assert_eq!(decode_provider_file_cid("not-a-cid"), None);
|
||||
assert_eq!(decode_provider_file_cid(""), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn digest_ceiling_is_64_bytes() {
|
||||
// varint(13) + "opensubtitles" = 14 bytes, so the id may use 50.
|
||||
assert!(provider_file_cid("opensubtitles", &"x".repeat(50)).is_some());
|
||||
assert!(provider_file_cid("opensubtitles", &"x".repeat(51)).is_none());
|
||||
fn file_id_of_accepts_both_record_id_shapes() {
|
||||
assert_eq!(file_id_of(&file_cid(7061834).unwrap()), Some(7061834));
|
||||
assert_eq!(file_id_of("opensubtitles:file:7061834"), Some(7061834));
|
||||
// Another provider's file, a card, and garbage are not ours.
|
||||
assert_eq!(file_id_of(&provider_file_cid("other", "file:1").unwrap()), None);
|
||||
assert_eq!(file_id_of("bagdsaaanar2g2zdcor3duojvgq3ts"), None);
|
||||
assert_eq!(file_id_of("opensubtitles:file:abc"), None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,13 +2,15 @@
|
||||
//!
|
||||
//! Serves `fileType:subtitle`. Needs an OpenSubtitles API key (config page or
|
||||
//! `OPENSUBTITLES_API_KEY`); without one it stays registered but degraded and
|
||||
//! answers every query with nothing.
|
||||
//! answers every query with nothing. An account login additionally lets the
|
||||
//! gateway redeem (download) subtitle files.
|
||||
//!
|
||||
//! Env:
|
||||
//! - `META_FEEDER_HTTP_LISTEN` — listen addr (default `0.0.0.0:8080`)
|
||||
//! - `META_FEEDER_STATE_DIR` — per-plugin cache root (default `/data/meta-feeder`)
|
||||
//! - `OPENSUBTITLES_API_KEY` — seed API key (the config page overrides it)
|
||||
//! - `OPENSUBTITLES_USER_AGENT` — seed User-Agent (OpenSubtitles wants `App vX.Y`)
|
||||
//! - `OPENSUBTITLES_USERNAME` / `OPENSUBTITLES_PASSWORD` — seed account login (downloads)
|
||||
//! - `RUST_LOG` — tracing filter (default `info`)
|
||||
|
||||
use std::net::SocketAddr;
|
||||
|
||||
@@ -8,26 +8,30 @@
|
||||
//!
|
||||
//! plus an optional `languages:` narrowing on either.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::path::Path;
|
||||
|
||||
use async_trait::async_trait;
|
||||
use meta_feeder_sdk::budget::RateBudget;
|
||||
use meta_feeder_sdk::common::open_midhash_cache;
|
||||
use meta_feeder_sdk::config::{ConfigField as F, ConfigSchema};
|
||||
use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashOutcome};
|
||||
use meta_feeder_sdk::hash::compute_ipfs_cid;
|
||||
use meta_feeder_sdk::plugin::{ConfigError, FeederPlugin, HashKind, HashOutcome, RedeemClaim};
|
||||
use meta_feeder_sdk::query::GatewayQuery;
|
||||
use meta_feeder_sdk::query_eval::query_accepts_plugin;
|
||||
use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, PluginHealth};
|
||||
use meta_feeder_sdk::types::{DiscoveryRecord, GatewayError, Hash, PluginHealth};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tracing::{debug, warn};
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
use crate::client::{OsClient, SearchParams};
|
||||
use crate::consts::{
|
||||
DEFAULT_USER_AGENT, ENV_API_KEY, ENV_USER_AGENT, OS_API_BASE, OS_BURST, OS_RATE_PER_SEC,
|
||||
DEFAULT_SUBTITLE_EXTENSION, DEFAULT_USER_AGENT, ENV_API_KEY, ENV_PASSWORD, ENV_USERNAME,
|
||||
ENV_USER_AGENT, MIN_SUBTITLE_BYTES, OS_API_BASE, OS_BURST, OS_RATE_PER_SEC, PACKAGE,
|
||||
UPSTREAM_ID,
|
||||
};
|
||||
use crate::lang::to_os_codes;
|
||||
use crate::record::project;
|
||||
use crate::locator::file_id_of;
|
||||
use crate::record::{extension_of, project};
|
||||
|
||||
/// Keys match [`FeederPlugin::config_schema`], which is what the SDK writes to
|
||||
/// `config.json` from the dashboard.
|
||||
@@ -37,6 +41,12 @@ struct OsConfig {
|
||||
api_key: String,
|
||||
#[serde(default, rename = "userAgent")]
|
||||
user_agent: String,
|
||||
/// Optional account login for `/download`. Blank → downloads run on the API
|
||||
/// key alone (anonymous quota).
|
||||
#[serde(default)]
|
||||
username: String,
|
||||
#[serde(default)]
|
||||
password: String,
|
||||
}
|
||||
|
||||
pub struct OpenSubtitlesPlugin {
|
||||
@@ -54,6 +64,8 @@ impl OpenSubtitlesPlugin {
|
||||
config: OsConfig {
|
||||
api_key: std::env::var(ENV_API_KEY).unwrap_or_default(),
|
||||
user_agent: std::env::var(ENV_USER_AGENT).unwrap_or_default(),
|
||||
username: std::env::var(ENV_USERNAME).unwrap_or_default(),
|
||||
password: std::env::var(ENV_PASSWORD).unwrap_or_default(),
|
||||
},
|
||||
api_base: OS_API_BASE.to_string(),
|
||||
client: None,
|
||||
@@ -65,13 +77,20 @@ impl OpenSubtitlesPlugin {
|
||||
Self {
|
||||
config: OsConfig {
|
||||
api_key: api_key.to_string(),
|
||||
user_agent: String::new(),
|
||||
..OsConfig::default()
|
||||
},
|
||||
api_base,
|
||||
client: None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Test builder: an account login (downloads use the account's quota).
|
||||
pub fn with_login(mut self, username: &str, password: &str) -> Self {
|
||||
self.config.username = username.to_string();
|
||||
self.config.password = password.to_string();
|
||||
self
|
||||
}
|
||||
|
||||
/// File wins over env, like every other feeder: `config.json` in the
|
||||
/// per-plugin cache dir, read at `configure` time. No hot reload.
|
||||
fn load_config(&mut self, cache_dir: &Path) {
|
||||
@@ -88,6 +107,12 @@ impl OpenSubtitlesPlugin {
|
||||
if !file_cfg.user_agent.trim().is_empty() {
|
||||
self.config.user_agent = file_cfg.user_agent;
|
||||
}
|
||||
if !file_cfg.username.trim().is_empty() {
|
||||
self.config.username = file_cfg.username;
|
||||
}
|
||||
if !file_cfg.password.trim().is_empty() {
|
||||
self.config.password = file_cfg.password;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -212,13 +237,22 @@ impl FeederPlugin for OpenSubtitlesPlugin {
|
||||
"" => DEFAULT_USER_AGENT,
|
||||
ua => ua,
|
||||
};
|
||||
self.client = Some(OsClient::new(
|
||||
let client = OsClient::new(
|
||||
self.api_base.clone(),
|
||||
api_key,
|
||||
user_agent,
|
||||
RateBudget::new(OS_RATE_PER_SEC, OS_BURST),
|
||||
Some(cache),
|
||||
));
|
||||
)
|
||||
.with_login(&self.config.username, &self.config.password);
|
||||
if !client.has_login() {
|
||||
info!(
|
||||
target: "opensubtitles",
|
||||
"no OpenSubtitles account login ({ENV_USERNAME}/{ENV_PASSWORD} or config page); \
|
||||
subtitle downloads use the API key's anonymous quota"
|
||||
);
|
||||
}
|
||||
self.client = Some(client);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
@@ -264,16 +298,85 @@ impl FeederPlugin for OpenSubtitlesPlugin {
|
||||
Ok(records)
|
||||
}
|
||||
|
||||
/// Not implemented, on purpose: a `provider-file` locator is resolved by
|
||||
/// meta-share's fetch tier (`docs/cid-formats.md` §8), and the SDK's
|
||||
/// `HashKind` has no variant to carry one — adding it is an SDK + gateway
|
||||
/// change. Searching never needs this path.
|
||||
async fn compute_outcomes(&self, _record_id: &str) -> Result<Vec<HashOutcome>, GatewayError> {
|
||||
Err(GatewayError::Permanent(
|
||||
"opensubtitles: provider-file locators are resolved by meta-share, not /compute \
|
||||
(docs/cid-formats.md §8)"
|
||||
.into(),
|
||||
))
|
||||
/// **Redeem** a `provider-file` locator: download the subtitle file it names.
|
||||
///
|
||||
/// `record_id` is the `0x100A` cid (what the gateway's redeem route sends)
|
||||
/// or this feeder's `opensubtitles:file:<id>` record id. Anything else —
|
||||
/// including another source's `provider-file` — is `NotFound`: not ours.
|
||||
///
|
||||
/// ⚠ Spends one unit of the daily download quota (the API key's anonymous
|
||||
/// quota, or the account's with a login). Returns ONE `Sha2_256` outcome
|
||||
/// with the bytes and a record of what only the download knows —
|
||||
/// `extension`, `fileName`, `sizeByte`. The gateway merges those onto the
|
||||
/// locator record next to the `file` pointer it writes; everything else
|
||||
/// (title, language, ids) is already there from the search.
|
||||
async fn compute_outcomes(&self, record_id: &str) -> Result<Vec<HashOutcome>, GatewayError> {
|
||||
let Some(file_id) = file_id_of(record_id) else {
|
||||
return Err(GatewayError::NotFound);
|
||||
};
|
||||
let Some(client) = self.client.as_ref() else {
|
||||
return Err(GatewayError::Permanent(
|
||||
"opensubtitles: no API key configured".into(),
|
||||
));
|
||||
};
|
||||
|
||||
info!(
|
||||
target: "opensubtitles",
|
||||
file_id,
|
||||
login = client.has_login(),
|
||||
"downloading subtitle (spends download quota)"
|
||||
);
|
||||
let got = client.download(file_id).await?;
|
||||
if got.bytes.len() < MIN_SUBTITLE_BYTES {
|
||||
return Err(GatewayError::Permanent(format!(
|
||||
"opensubtitles: file {file_id} is {} bytes — not a subtitle",
|
||||
got.bytes.len()
|
||||
)));
|
||||
}
|
||||
if let Some(remaining) = got.remaining {
|
||||
debug!(target: "opensubtitles", file_id, remaining, "download quota remaining");
|
||||
}
|
||||
let extension = got
|
||||
.file_name
|
||||
.as_deref()
|
||||
.and_then(extension_of)
|
||||
.unwrap_or(DEFAULT_SUBTITLE_EXTENSION)
|
||||
.to_string();
|
||||
|
||||
// Facts only the download knows. The search record may lack
|
||||
// `extension` (a listing whose name has no subtitle suffix) and never
|
||||
// knows the real size.
|
||||
let mut fields = BTreeMap::new();
|
||||
fields.insert("extension".to_string(), extension.clone());
|
||||
if let Some(name) = got.file_name.as_deref().map(str::trim).filter(|n| !n.is_empty()) {
|
||||
fields.insert("fileName".to_string(), name.to_string());
|
||||
}
|
||||
fields.insert("sizeByte".to_string(), got.bytes.len().to_string());
|
||||
|
||||
Ok(vec![HashOutcome {
|
||||
hash: Hash(compute_ipfs_cid(&got.bytes)),
|
||||
hash_kind: HashKind::Sha2_256,
|
||||
bytes: Some(got.bytes.into()),
|
||||
record: Some(DiscoveryRecord {
|
||||
upstream_id: UPSTREAM_ID.to_string(),
|
||||
record_id: format!("{UPSTREAM_ID}:file:{file_id}"),
|
||||
fields,
|
||||
}),
|
||||
file_extension: Some(extension),
|
||||
}])
|
||||
}
|
||||
|
||||
fn package(&self) -> Option<&'static str> {
|
||||
Some(PACKAGE)
|
||||
}
|
||||
|
||||
/// Claims `provider-file` for `opensubtitles` whenever an API key is
|
||||
/// configured — a login only changes which quota a download spends.
|
||||
fn redeems(&self) -> Vec<RedeemClaim> {
|
||||
if self.client.is_none() {
|
||||
return Vec::new();
|
||||
}
|
||||
vec![RedeemClaim::provider_file(vec![UPSTREAM_ID.to_string()])]
|
||||
}
|
||||
|
||||
fn config_schema(&self) -> ConfigSchema {
|
||||
@@ -282,14 +385,23 @@ impl FeederPlugin for OpenSubtitlesPlugin {
|
||||
F::secret("apiKey", "OpenSubtitles API key")
|
||||
.required()
|
||||
.with_help(
|
||||
"Consumer API key from opensubtitles.com. Search only — this feeder \
|
||||
never downloads, so no account login is needed and no download \
|
||||
quota is spent. Takes effect on the next feeder restart.",
|
||||
"Consumer API key from opensubtitles.com. Enough for search (which \
|
||||
spends no quota) and for downloads, on the key's anonymous download \
|
||||
quota. Takes effect on the next feeder restart.",
|
||||
),
|
||||
F::text("userAgent", "User-Agent").with_help(
|
||||
"OpenSubtitles expects the registered app name in `App vX.Y` form. \
|
||||
Blank keeps the built-in default.",
|
||||
),
|
||||
F::text("username", "Account username (optional)").with_help(
|
||||
"opensubtitles.com account (not .org) to download as, so the account's \
|
||||
rank quota applies instead of the anonymous one. Each download spends \
|
||||
one unit — once per file, since the gateway stores what it fetched. \
|
||||
Blank → downloads on the API key alone.",
|
||||
),
|
||||
F::secret("password", "Account password").with_help(
|
||||
"Password for the account above. Takes effect on the next feeder restart.",
|
||||
),
|
||||
],
|
||||
}
|
||||
}
|
||||
@@ -298,6 +410,8 @@ impl FeederPlugin for OpenSubtitlesPlugin {
|
||||
serde_json::json!({
|
||||
"apiKey": self.config.api_key,
|
||||
"userAgent": self.config.user_agent,
|
||||
"username": self.config.username,
|
||||
"password": self.config.password,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -138,7 +138,7 @@ fn title_for(a: &OsAttributes, fd: &FeatureDetails, file: &OsFile) -> Option<Str
|
||||
}
|
||||
}
|
||||
|
||||
fn extension_of(name: &str) -> Option<&'static str> {
|
||||
pub(crate) fn extension_of(name: &str) -> Option<&'static str> {
|
||||
let (_, ext) = name.rsplit_once('.')?;
|
||||
let ext = ext.to_ascii_lowercase();
|
||||
SUBTITLE_EXTENSIONS.iter().copied().find(|e| *e == ext)
|
||||
|
||||
@@ -12,29 +12,45 @@
|
||||
//! own `record_matches` against the query that produced them;
|
||||
//! - the upstream sees OpenSubtitles' canonical query form (sorted keys) and the
|
||||
//! API key header;
|
||||
//! - a query this feeder cannot answer never reaches the upstream.
|
||||
//! - a query this feeder cannot answer never reaches the upstream;
|
||||
//! - redeeming a `provider-file` cid through `/compute` downloads the file on
|
||||
//! the API key alone, or as an account (re-logging in once on a stale token),
|
||||
//! returns the download-only facts as record fields, and honours the quota.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
|
||||
use base64::Engine as _;
|
||||
use meta_feeder_sdk::hash::compute_ipfs_cid;
|
||||
use meta_feeder_sdk::query_eval::record_matches;
|
||||
use meta_feeder_sdk::{
|
||||
configure_plugins, router, ComputeRequest, GatewayQuery, ManifestResponse, QueryRequest,
|
||||
QueryResponse,
|
||||
configure_plugins, router, ComputeRequest, ComputeResponse, GatewayQuery, HashKindDto,
|
||||
ManifestResponse, QueryRequest, QueryResponse, RedeemsResponse,
|
||||
};
|
||||
use opensubtitles_feeder::locator::decode_provider_file_cid;
|
||||
use opensubtitles_feeder::locator::{decode_provider_file_cid, file_cid, provider_file_cid};
|
||||
use opensubtitles_feeder::plugin::OpenSubtitlesPlugin;
|
||||
use wiremock::matchers::{header, method, path, query_param};
|
||||
use wiremock::matchers::{body_partial_json, header, method, path, query_param};
|
||||
use wiremock::{Match, Mock, MockServer, Request, ResponseTemplate};
|
||||
|
||||
const KEY: &str = "test-key";
|
||||
const USER: &str = "viewer";
|
||||
const PASS: &str = "hunter2";
|
||||
const FILE_ID: u64 = 7061834;
|
||||
const SRT: &str = "1\n00:00:01,000 --> 00:00:02,000\nWake up, Neo.\n";
|
||||
|
||||
const KEYWORD: &str = include_str!("fixtures/search_keyword_matrix.json");
|
||||
const MOVIE_FR: &str = include_str!("fixtures/search_imdb_133093_fr.json");
|
||||
const EPISODE: &str = include_str!("fixtures/search_episode_1399_s01e01.json");
|
||||
|
||||
async fn spawn_feeder(api_key: &str, upstream: &MockServer) -> (String, tempfile::TempDir) {
|
||||
spawn_plugin(OpenSubtitlesPlugin::with_api_base(api_key, upstream.uri())).await
|
||||
}
|
||||
|
||||
async fn spawn_feeder_with_login(upstream: &MockServer) -> (String, tempfile::TempDir) {
|
||||
spawn_plugin(OpenSubtitlesPlugin::with_api_base(KEY, upstream.uri()).with_login(USER, PASS)).await
|
||||
}
|
||||
|
||||
async fn spawn_plugin(plugin: OpenSubtitlesPlugin) -> (String, tempfile::TempDir) {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let plugin = OpenSubtitlesPlugin::with_api_base(api_key, upstream.uri());
|
||||
let plugins = configure_plugins(vec![Box::new(plugin)], dir.path()).expect("configure plugins");
|
||||
let app = router(plugins, "opensubtitles-feeder-test".to_string(), dir.path());
|
||||
|
||||
@@ -129,6 +145,14 @@ async fn manifest_advertises_subtitle_only() {
|
||||
assert_eq!(m.plugins[0].id, "opensubtitles");
|
||||
assert_eq!(m.plugins[0].served_file_types, vec!["subtitle".to_string()]);
|
||||
assert!(m.plugins[0].served_content_kinds.is_empty());
|
||||
assert_eq!(m.plugins[0].package.as_deref(), Some("meta-feeder-opensubtitles"));
|
||||
// An API key alone downloads (anonymous quota), so the source is claimed.
|
||||
assert_eq!(
|
||||
serde_json::to_value(&m.plugins[0].redeems).unwrap(),
|
||||
serde_json::json!([{
|
||||
"codec": "provider-file", "field": "file", "hosts": [], "sources": ["opensubtitles"]
|
||||
}])
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -293,19 +317,239 @@ async fn missing_api_key_degrades_and_answers_nothing() {
|
||||
.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn compute_is_refused() {
|
||||
let upstream = MockServer::start().await;
|
||||
let (base, _dir) = spawn_feeder(KEY, &upstream).await;
|
||||
// --- redeeming a provider-file locator ---------------------------------------
|
||||
|
||||
let resp = reqwest::Client::new()
|
||||
async fn post_compute(base: &str, record_id: &str) -> reqwest::Response {
|
||||
reqwest::Client::new()
|
||||
.post(format!("{base}/compute"))
|
||||
.json(&ComputeRequest {
|
||||
upstream_id: "opensubtitles".into(),
|
||||
record_id: "opensubtitles:file:12658754".into(),
|
||||
record_id: record_id.into(),
|
||||
})
|
||||
.send()
|
||||
.await
|
||||
.expect("POST /compute")
|
||||
}
|
||||
|
||||
async fn mount_login(upstream: &MockServer, token: &str, times: Option<u64>) {
|
||||
let mock = Mock::given(method("POST"))
|
||||
.and(path("/login"))
|
||||
.and(header("Api-Key", KEY))
|
||||
.and(body_partial_json(serde_json::json!({ "username": USER, "password": PASS })))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_json(serde_json::json!({ "token": token })));
|
||||
match times {
|
||||
Some(n) => mock.up_to_n_times(n).expect(n).mount(upstream).await,
|
||||
None => mock.mount(upstream).await,
|
||||
}
|
||||
}
|
||||
|
||||
fn download_ok(upstream: &MockServer, remaining: u64) -> ResponseTemplate {
|
||||
ResponseTemplate::new(200).set_body_json(serde_json::json!({
|
||||
"link": format!("{}/files/{FILE_ID}.srt", upstream.uri()),
|
||||
"file_name": "The.Matrix.1999.en.srt",
|
||||
"requests": 1,
|
||||
"remaining": remaining,
|
||||
"reset_time_utc": "2099-01-01T00:00:00.000Z"
|
||||
}))
|
||||
}
|
||||
|
||||
async fn mount_file(upstream: &MockServer, body: &str) {
|
||||
Mock::given(method("GET"))
|
||||
.and(path(format!("/files/{FILE_ID}.srt")))
|
||||
.respond_with(ResponseTemplate::new(200).set_body_string(body))
|
||||
.mount(upstream)
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_provider_file_cid_redeems_to_the_subtitle_bytes() {
|
||||
let upstream = MockServer::start().await;
|
||||
mount_login(&upstream, "t1", None).await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/download"))
|
||||
.and(header("Authorization", "Bearer t1"))
|
||||
.and(body_partial_json(serde_json::json!({ "file_id": FILE_ID })))
|
||||
.respond_with(download_ok(&upstream, 99))
|
||||
.expect(2)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
mount_file(&upstream, SRT).await;
|
||||
let (base, _dir) = spawn_feeder_with_login(&upstream).await;
|
||||
|
||||
let resp = post_compute(&base, &file_cid(FILE_ID).unwrap()).await;
|
||||
assert_eq!(resp.status(), 200, "{:?}", resp.text().await);
|
||||
let body: ComputeResponse = resp.json().await.unwrap();
|
||||
assert_eq!(body.outcomes.len(), 1);
|
||||
let o = &body.outcomes[0];
|
||||
assert_eq!(o.hash_kind, HashKindDto::Sha2_256);
|
||||
assert_eq!(o.file_extension.as_deref(), Some("srt"));
|
||||
let rec = o.record.as_ref().expect("download-only facts for the locator record");
|
||||
assert_eq!(rec.fields.get("extension").map(String::as_str), Some("srt"));
|
||||
assert_eq!(rec.fields.get("fileName").map(String::as_str), Some("The.Matrix.1999.en.srt"));
|
||||
assert_eq!(rec.fields.get("sizeByte"), Some(&SRT.len().to_string()));
|
||||
assert!(
|
||||
!rec.fields.keys().any(|k| k.starts_with("cids/")),
|
||||
"a redeem never re-addresses the locator record"
|
||||
);
|
||||
let bytes = base64::engine::general_purpose::STANDARD
|
||||
.decode(o.bytes_b64.as_deref().unwrap())
|
||||
.unwrap();
|
||||
assert_eq!(resp.status(), 422);
|
||||
assert_eq!(bytes, SRT.as_bytes());
|
||||
assert_eq!(o.hash, compute_ipfs_cid(&bytes));
|
||||
|
||||
// The feeder's own record id form redeems the same file.
|
||||
assert_eq!(post_compute(&base, &format!("opensubtitles:file:{FILE_ID}")).await.status(), 200);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_stale_token_is_refreshed_once() {
|
||||
let upstream = MockServer::start().await;
|
||||
// First login hands out t1, the second t2.
|
||||
mount_login(&upstream, "t1", Some(1)).await;
|
||||
mount_login(&upstream, "t2", Some(1)).await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/download"))
|
||||
.and(header("Authorization", "Bearer t1"))
|
||||
.respond_with(ResponseTemplate::new(401))
|
||||
.expect(1)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/download"))
|
||||
.and(header("Authorization", "Bearer t2"))
|
||||
.respond_with(download_ok(&upstream, 99))
|
||||
.expect(1)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
mount_file(&upstream, SRT).await;
|
||||
let (base, _dir) = spawn_feeder_with_login(&upstream).await;
|
||||
|
||||
assert_eq!(post_compute(&base, &file_cid(FILE_ID).unwrap()).await.status(), 200);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_spent_quota_is_rate_limited_and_stops_calling_download() {
|
||||
let upstream = MockServer::start().await;
|
||||
mount_login(&upstream, "t1", None).await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/download"))
|
||||
.respond_with(ResponseTemplate::new(406).set_body_json(serde_json::json!({
|
||||
"message": "You have downloaded your allowed 20 subtitles for 24h.",
|
||||
"reset_time": "23 hours and 59 minutes",
|
||||
"reset_time_utc": "2099-01-01T00:00:00.000Z"
|
||||
})))
|
||||
.expect(1)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
let (base, _dir) = spawn_feeder_with_login(&upstream).await;
|
||||
|
||||
let cid = file_cid(FILE_ID).unwrap();
|
||||
assert_eq!(post_compute(&base, &cid).await.status(), 429);
|
||||
// Blocked locally until the reset: the upstream is not asked again.
|
||||
assert_eq!(post_compute(&base, &cid).await.status(), 429);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn the_last_allowed_download_blocks_the_next() {
|
||||
let upstream = MockServer::start().await;
|
||||
mount_login(&upstream, "t1", None).await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/download"))
|
||||
.respond_with(download_ok(&upstream, 0))
|
||||
.expect(1)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
mount_file(&upstream, SRT).await;
|
||||
let (base, _dir) = spawn_feeder_with_login(&upstream).await;
|
||||
|
||||
let cid = file_cid(FILE_ID).unwrap();
|
||||
assert_eq!(post_compute(&base, &cid).await.status(), 200);
|
||||
assert_eq!(post_compute(&base, &cid).await.status(), 429);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn not_our_locator_is_not_found_and_spends_nothing() {
|
||||
let upstream = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.respond_with(ResponseTemplate::new(200))
|
||||
.expect(0)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
let (base, _dir) = spawn_feeder_with_login(&upstream).await;
|
||||
|
||||
let other = provider_file_cid("otherprovider", "file:1").unwrap();
|
||||
assert_eq!(post_compute(&base, &other).await.status(), 404);
|
||||
// A card locator shares the framing but not the codec.
|
||||
assert_eq!(post_compute(&base, "bagdsaaanar2g2zdcor3duojvgq3ts").await.status(), 404);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_empty_file_is_refused() {
|
||||
let upstream = MockServer::start().await;
|
||||
mount_login(&upstream, "t1", None).await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/download"))
|
||||
.respond_with(download_ok(&upstream, 99))
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
mount_file(&upstream, "oops").await;
|
||||
let (base, _dir) = spawn_feeder_with_login(&upstream).await;
|
||||
|
||||
assert_eq!(post_compute(&base, &file_cid(FILE_ID).unwrap()).await.status(), 422);
|
||||
}
|
||||
|
||||
/// Anonymous mode sends the API key and no Bearer token.
|
||||
struct NoAuthorization;
|
||||
|
||||
impl Match for NoAuthorization {
|
||||
fn matches(&self, request: &Request) -> bool {
|
||||
!request.headers.contains_key("authorization")
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn an_api_key_alone_downloads_without_logging_in() {
|
||||
let upstream = MockServer::start().await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/login"))
|
||||
.respond_with(ResponseTemplate::new(200))
|
||||
.expect(0)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
Mock::given(method("POST"))
|
||||
.and(path("/download"))
|
||||
.and(header("Api-Key", KEY))
|
||||
.and(NoAuthorization)
|
||||
.and(body_partial_json(serde_json::json!({ "file_id": FILE_ID })))
|
||||
.respond_with(download_ok(&upstream, 4))
|
||||
.expect(1)
|
||||
.mount(&upstream)
|
||||
.await;
|
||||
mount_file(&upstream, SRT).await;
|
||||
let (base, _dir) = spawn_feeder(KEY, &upstream).await;
|
||||
|
||||
let resp = post_compute(&base, &file_cid(FILE_ID).unwrap()).await;
|
||||
assert_eq!(resp.status(), 200, "{:?}", resp.text().await);
|
||||
let body: ComputeResponse = resp.json().await.unwrap();
|
||||
assert_eq!(body.outcomes.len(), 1);
|
||||
assert_eq!(body.outcomes[0].hash, compute_ipfs_cid(SRT.as_bytes()));
|
||||
|
||||
let r: RedeemsResponse = reqwest::get(format!("{base}/redeems")).await.unwrap().json().await.unwrap();
|
||||
assert_eq!(r.redeems["opensubtitles"].len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn with_a_login_provider_file_is_claimed() {
|
||||
let upstream = MockServer::start().await;
|
||||
let (base, _dir) = spawn_feeder_with_login(&upstream).await;
|
||||
|
||||
let r: RedeemsResponse = reqwest::get(format!("{base}/redeems")).await.unwrap().json().await.unwrap();
|
||||
assert_eq!(
|
||||
serde_json::to_value(&r.redeems["opensubtitles"]).unwrap(),
|
||||
serde_json::json!([{
|
||||
"codec": "provider-file", "field": "file", "hosts": [], "sources": ["opensubtitles"]
|
||||
}])
|
||||
);
|
||||
let m: ManifestResponse = reqwest::get(format!("{base}/manifest")).await.unwrap().json().await.unwrap();
|
||||
assert_eq!(m.plugins[0].redeems, r.redeems["opensubtitles"]);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user