feat: implement RSS subscriptions, downloader routing and local media checks
This commit is contained in:
junior
+248
@@ -0,0 +1,248 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::{
|
||||
domain::{DownloaderInput, SubscriptionInput},
|
||||
error::{Error, Result},
|
||||
media::{self, DestinationInput, MappingInput},
|
||||
service::Service,
|
||||
};
|
||||
use actix_web::{web, HttpResponse};
|
||||
use std::sync::Arc;
|
||||
|
||||
type State = web::Data<Arc<Service>>;
|
||||
|
||||
pub fn configure(cfg: &mut web::ServiceConfig) {
|
||||
cfg.service(
|
||||
web::scope("/api")
|
||||
.route("/sources/mikan/search", web::get().to(mikan_search))
|
||||
.route("/sources/mikan/series/{id}", web::get().to(mikan_series))
|
||||
.route(
|
||||
"/sources/mikan/series/{id}/cover",
|
||||
web::get().to(mikan_cover),
|
||||
)
|
||||
.route("/sources/mikan/resolve", web::post().to(mikan_resolve))
|
||||
.route("/status", web::get().to(status))
|
||||
.route("/destinations", web::get().to(destinations))
|
||||
.route("/destinations", web::post().to(create_destination))
|
||||
.route("/destinations/active", web::put().to(activate_destination))
|
||||
.route("/destinations/{id}/mappings", web::get().to(mappings))
|
||||
.route(
|
||||
"/destinations/{id}/mappings",
|
||||
web::post().to(create_mapping),
|
||||
)
|
||||
.route(
|
||||
"/destinations/{id}/entries",
|
||||
web::get().to(directory_entries),
|
||||
)
|
||||
.route("/downloaders", web::get().to(downloaders))
|
||||
.route("/downloaders", web::post().to(create_downloader))
|
||||
.route("/downloaders/active", web::put().to(activate_downloader))
|
||||
.route(
|
||||
"/downloaders/{id}/claim-legacy-jobs",
|
||||
web::post().to(claim_legacy_jobs),
|
||||
)
|
||||
.route("/preview", web::post().to(preview))
|
||||
.route("/subscriptions", web::get().to(list))
|
||||
.route("/subscriptions", web::post().to(create))
|
||||
.route("/subscriptions/{id}", web::get().to(get))
|
||||
.route("/subscriptions/{id}", web::put().to(update))
|
||||
.route("/subscriptions/{id}", web::delete().to(delete))
|
||||
.route("/subscriptions/{id}/preview", web::get().to(preview_saved))
|
||||
.route("/subscriptions/{id}/scan", web::post().to(scan))
|
||||
.route("/subscriptions/{id}/releases", web::get().to(releases))
|
||||
.route("/jobs", web::get().to(jobs))
|
||||
.route("/jobs/{id}/files", web::get().to(job_files))
|
||||
.route("/jobs/{id}/files/check", web::post().to(check_job_files))
|
||||
.route("/jobs/{id}/retry", web::post().to(retry))
|
||||
.route("/run", web::post().to(run)),
|
||||
);
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
struct SearchQuery {
|
||||
q: String,
|
||||
}
|
||||
|
||||
async fn mikan_search(state: State, query: web::Query<SearchQuery>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.mikan.search(&query.q).await?))
|
||||
}
|
||||
|
||||
async fn mikan_series(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.mikan.series(&id).await?))
|
||||
}
|
||||
|
||||
async fn mikan_cover(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
let (mime, bytes) = state.mikan.cover(&id).await?;
|
||||
Ok(HttpResponse::Ok()
|
||||
.content_type(mime)
|
||||
.insert_header(("X-Content-Type-Options", "nosniff"))
|
||||
.insert_header(("Cache-Control", "private, max-age=60"))
|
||||
.body(bytes))
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct ResolveInput {
|
||||
url: String,
|
||||
}
|
||||
|
||||
async fn mikan_resolve(state: State, body: web::Json<ResolveInput>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.mikan.resolve(&body.url).await?))
|
||||
}
|
||||
|
||||
async fn status(state: State) -> Result<HttpResponse> {
|
||||
let active = state.store.active_downloader_id().await?;
|
||||
Ok(HttpResponse::Ok().json(serde_json::json!({
|
||||
"downloader_configured": active.is_some(),
|
||||
"active_downloader_id": active,
|
||||
"active_destination_id": state.store.active_destination_id().await?,
|
||||
"legacy_unassigned_jobs": state.store.legacy_unassigned_count().await?,
|
||||
"version": env!("CARGO_PKG_VERSION")
|
||||
})))
|
||||
}
|
||||
|
||||
async fn destinations(state: State) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.destinations().await?))
|
||||
}
|
||||
|
||||
async fn create_destination(
|
||||
state: State,
|
||||
body: web::Json<DestinationInput>,
|
||||
) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
Ok(HttpResponse::Created().json(state.store.create_destination(body.into_inner()).await?))
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct DestinationSelection {
|
||||
destination_id: String,
|
||||
}
|
||||
|
||||
async fn activate_destination(
|
||||
state: State,
|
||||
body: web::Json<DestinationSelection>,
|
||||
) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
state
|
||||
.store
|
||||
.activate_destination(&body.destination_id)
|
||||
.await?;
|
||||
Ok(HttpResponse::NoContent().finish())
|
||||
}
|
||||
|
||||
async fn mappings(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.mappings(&id).await?))
|
||||
}
|
||||
|
||||
async fn create_mapping(
|
||||
state: State,
|
||||
id: web::Path<String>,
|
||||
body: web::Json<MappingInput>,
|
||||
) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
Ok(HttpResponse::Created().json(state.store.create_mapping(&id, body.into_inner()).await?))
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
struct DirectoryQuery {
|
||||
#[serde(default)]
|
||||
path: String,
|
||||
}
|
||||
|
||||
async fn directory_entries(
|
||||
state: State,
|
||||
id: web::Path<String>,
|
||||
query: web::Query<DirectoryQuery>,
|
||||
) -> Result<HttpResponse> {
|
||||
media::validate_relative(&query.path, true)?;
|
||||
let root = state.store.destination(&id).await?.root_path;
|
||||
let path = query.into_inner().path;
|
||||
let entries = tokio::task::spawn_blocking(move || media::list_local(&root, &path))
|
||||
.await
|
||||
.map_err(|_| Error::Upstream("local directory query failed".into()))?;
|
||||
Ok(HttpResponse::Ok().json(entries))
|
||||
}
|
||||
|
||||
async fn job_files(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.file_report(&id).await?))
|
||||
}
|
||||
|
||||
async fn check_job_files(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.check_job_files(&id).await?))
|
||||
}
|
||||
|
||||
async fn downloaders(state: State) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.downloaders().await?))
|
||||
}
|
||||
|
||||
async fn create_downloader(state: State, body: web::Json<DownloaderInput>) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
Ok(HttpResponse::Created().json(state.store.create_downloader(body.into_inner()).await?))
|
||||
}
|
||||
|
||||
#[derive(serde::Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
struct DownloaderSelection {
|
||||
downloader_id: String,
|
||||
}
|
||||
|
||||
async fn activate_downloader(
|
||||
state: State,
|
||||
body: web::Json<DownloaderSelection>,
|
||||
) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
state.store.activate_downloader(&body.downloader_id).await?;
|
||||
Ok(HttpResponse::NoContent().finish())
|
||||
}
|
||||
|
||||
async fn claim_legacy_jobs(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
Ok(HttpResponse::Ok()
|
||||
.json(serde_json::json!({"claimed": state.store.claim_legacy_jobs(&id).await?})))
|
||||
}
|
||||
async fn list(state: State) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.subscriptions().await?))
|
||||
}
|
||||
async fn get(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.subscription(&id).await?))
|
||||
}
|
||||
async fn create(state: State, body: web::Json<SubscriptionInput>) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
Ok(HttpResponse::Created().json(state.store.create(body.into_inner()).await?))
|
||||
}
|
||||
async fn update(
|
||||
state: State,
|
||||
id: web::Path<String>,
|
||||
body: web::Json<SubscriptionInput>,
|
||||
) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
Ok(HttpResponse::Ok().json(state.store.update(&id, body.into_inner()).await?))
|
||||
}
|
||||
async fn delete(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
state.store.delete(&id).await?;
|
||||
Ok(HttpResponse::NoContent().finish())
|
||||
}
|
||||
async fn preview(state: State, body: web::Json<SubscriptionInput>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.preview(body.into_inner()).await?))
|
||||
}
|
||||
async fn preview_saved(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.preview_subscription(&id).await?))
|
||||
}
|
||||
async fn scan(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(serde_json::json!({"queued": state.scan_subscription(&id).await?})))
|
||||
}
|
||||
async fn releases(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.releases(&id).await?))
|
||||
}
|
||||
async fn jobs(state: State) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.store.jobs().await?))
|
||||
}
|
||||
async fn retry(state: State, id: web::Path<String>) -> Result<HttpResponse> {
|
||||
let _guard = state.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
state.store.retry(&id).await?;
|
||||
Ok(HttpResponse::NoContent().finish())
|
||||
}
|
||||
async fn run(state: State) -> Result<HttpResponse> {
|
||||
Ok(HttpResponse::Ok().json(state.cycle().await?))
|
||||
}
|
||||
+106
@@ -0,0 +1,106 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::error::{Error, Result};
|
||||
use std::{env, time::Duration};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Config {
|
||||
pub database_url: String,
|
||||
pub port: u16,
|
||||
pub poll_interval: Duration,
|
||||
pub qbit: Option<QbitConfig>,
|
||||
pub mikan_base_url: String,
|
||||
pub website: WebsiteConfig,
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
pub struct WebsiteConfig {
|
||||
pub proxy_url: Option<String>,
|
||||
pub direct_hosts: String,
|
||||
}
|
||||
|
||||
// Deliberately omit Debug: these fields can contain credentials.
|
||||
#[derive(Clone, serde::Serialize, serde::Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct QbitConfig {
|
||||
pub url: String,
|
||||
pub username: String,
|
||||
pub password: String,
|
||||
pub save_path: String,
|
||||
}
|
||||
|
||||
impl QbitConfig {
|
||||
pub fn validate(&mut self) -> Result<()> {
|
||||
self.url = self.url.trim().trim_end_matches('/').to_owned();
|
||||
crate::domain::validate_http_url(&self.url)?;
|
||||
let url = reqwest::Url::parse(&self.url).expect("validated URL");
|
||||
if url.query().is_some() || url.fragment().is_some() {
|
||||
return Err(Error::Invalid(
|
||||
"downloader URL cannot contain query parameters or a fragment".into(),
|
||||
));
|
||||
}
|
||||
self.url = url.to_string().trim_end_matches('/').to_owned();
|
||||
if self.username.len() > 1024
|
||||
|| self.password.len() > 4096
|
||||
|| self.save_path.trim().is_empty()
|
||||
|| self.save_path.len() > 4096
|
||||
|| self.save_path.chars().any(char::is_control)
|
||||
{
|
||||
return Err(Error::Invalid(
|
||||
"invalid downloader credentials or save path".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
impl Config {
|
||||
pub fn from_env() -> Result<Self> {
|
||||
let port = env::var("APP_PORT")
|
||||
.unwrap_or_else(|_| "8080".into())
|
||||
.parse()
|
||||
.map_err(|_| Error::Invalid("APP_PORT must be a valid port".into()))?;
|
||||
let seconds: u64 = env::var("POLL_INTERVAL_SECONDS")
|
||||
.unwrap_or_else(|_| "300".into())
|
||||
.parse()
|
||||
.map_err(|_| Error::Invalid("POLL_INTERVAL_SECONDS must be an integer".into()))?;
|
||||
if seconds < 10 {
|
||||
return Err(Error::Invalid(
|
||||
"POLL_INTERVAL_SECONDS must be at least 10".into(),
|
||||
));
|
||||
}
|
||||
let qbit = match env::var("QBIT_URL") {
|
||||
Ok(url) if !url.is_empty() => {
|
||||
crate::domain::validate_http_url(&url)?;
|
||||
let base = reqwest::Url::parse(&url).expect("validated URL");
|
||||
if base.query().is_some() || base.fragment().is_some() {
|
||||
return Err(Error::Invalid(
|
||||
"QBIT_URL cannot contain query parameters or a fragment".into(),
|
||||
));
|
||||
}
|
||||
Some(QbitConfig {
|
||||
url: url.trim_end_matches('/').into(),
|
||||
username: env::var("QBIT_USERNAME").unwrap_or_default(),
|
||||
password: env::var("QBIT_PASSWORD").unwrap_or_default(),
|
||||
save_path: env::var("QBIT_SAVE_PATH").unwrap_or_else(|_| "/downloads".into()),
|
||||
})
|
||||
}
|
||||
_ => None,
|
||||
};
|
||||
Ok(Self {
|
||||
database_url: env::var("DATABASE_URL")
|
||||
.unwrap_or_else(|_| "sqlite://data/ani.db".into()),
|
||||
port,
|
||||
poll_interval: Duration::from_secs(seconds),
|
||||
qbit,
|
||||
mikan_base_url: env::var("MIKAN_BASE_URL")
|
||||
.unwrap_or_else(|_| "https://mikanani.me".into()),
|
||||
website: WebsiteConfig {
|
||||
proxy_url: env::var("WEBSITE_PROXY_URL")
|
||||
.ok()
|
||||
.map(|s| s.trim().to_owned())
|
||||
.filter(|s| !s.is_empty()),
|
||||
direct_hosts: env::var("WEBSITE_DIRECT_HOSTS").unwrap_or_default(),
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
+160
@@ -0,0 +1,160 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::error::{Error, Result};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
#[derive(Clone, Debug, Deserialize, Serialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct SubscriptionInput {
|
||||
pub title: String,
|
||||
pub rss_url: String,
|
||||
#[serde(default = "one")]
|
||||
pub season: i32,
|
||||
#[serde(default)]
|
||||
pub episode_offset: i32,
|
||||
#[serde(default)]
|
||||
pub include: Vec<String>,
|
||||
#[serde(default)]
|
||||
pub exclude: Vec<String>,
|
||||
// Opt in after previewing: the first scan considers all entries in the feed.
|
||||
#[serde(default)]
|
||||
pub enabled: bool,
|
||||
}
|
||||
|
||||
fn one() -> i32 {
|
||||
1
|
||||
}
|
||||
|
||||
impl SubscriptionInput {
|
||||
pub fn validate(&mut self) -> Result<()> {
|
||||
self.title = self.title.trim().to_owned();
|
||||
self.rss_url = self.rss_url.trim().to_owned();
|
||||
if self.title.is_empty() || self.title.chars().count() > 200 {
|
||||
return Err(Error::Invalid("title must contain 1–200 characters".into()));
|
||||
}
|
||||
validate_http_url(&self.rss_url)?;
|
||||
if !(1..=999).contains(&self.season) || self.episode_offset.abs_diff(0) > 10000 {
|
||||
return Err(Error::Invalid(
|
||||
"season or episode_offset is out of range".into(),
|
||||
));
|
||||
}
|
||||
for rules in [&mut self.include, &mut self.exclude] {
|
||||
if rules.len() > 32 {
|
||||
return Err(Error::Invalid("at most 32 rules per list".into()));
|
||||
}
|
||||
for rule in rules {
|
||||
*rule = rule.trim().to_owned();
|
||||
if rule.is_empty() || rule.len() > 200 {
|
||||
return Err(Error::Invalid(
|
||||
"filter keywords must contain 1–200 bytes".into(),
|
||||
));
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
pub fn validate_http_url(value: &str) -> Result<()> {
|
||||
let url = reqwest::Url::parse(value).map_err(|_| Error::Invalid("invalid URL".into()))?;
|
||||
if !matches!(url.scheme(), "http" | "https")
|
||||
|| url.host_str().is_none()
|
||||
|| !url.username().is_empty()
|
||||
|| url.password().is_some()
|
||||
|| value.len() > 8192
|
||||
{
|
||||
return Err(Error::Invalid(
|
||||
"expected an HTTP(S) URL without embedded credentials".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct Subscription {
|
||||
pub id: String,
|
||||
#[serde(flatten)]
|
||||
pub input: SubscriptionInput,
|
||||
pub last_checked_at: Option<i64>,
|
||||
pub last_error: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum Decision {
|
||||
Ready,
|
||||
Filtered,
|
||||
NeedsReview,
|
||||
AlreadyQueued,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, Deserialize)]
|
||||
pub struct Release {
|
||||
pub key: String,
|
||||
pub title: String,
|
||||
pub download_url: Option<String>,
|
||||
pub episode: Option<i32>,
|
||||
pub decision: Decision,
|
||||
pub reason: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, Serialize, sqlx::FromRow)]
|
||||
pub struct DownloadJob {
|
||||
pub id: String,
|
||||
pub subscription_id: String,
|
||||
pub title: String,
|
||||
pub season: i32,
|
||||
pub episode: i32,
|
||||
pub download_url: String,
|
||||
pub info_hash: Option<String>,
|
||||
pub state: String,
|
||||
pub attempts: i32,
|
||||
pub last_error: Option<String>,
|
||||
pub next_attempt_at: i64,
|
||||
pub downloader_id: Option<String>,
|
||||
pub legacy_unassigned: bool,
|
||||
pub destination_id: Option<String>,
|
||||
pub mapping_id: Option<String>,
|
||||
pub download_save_path: Option<String>,
|
||||
pub file_status: String,
|
||||
pub file_checked_at: Option<i64>,
|
||||
pub file_last_success_at: Option<i64>,
|
||||
pub file_error: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, serde::Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct DownloaderInput {
|
||||
pub name: String,
|
||||
pub kind: DownloaderKind,
|
||||
pub connection: crate::config::QbitConfig,
|
||||
}
|
||||
|
||||
#[derive(Clone, serde::Deserialize, serde::Serialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum DownloaderKind {
|
||||
Qbittorrent,
|
||||
}
|
||||
|
||||
impl DownloaderInput {
|
||||
pub fn validate(&mut self) -> Result<()> {
|
||||
self.name = self.name.trim().to_owned();
|
||||
if self.name.is_empty() || self.name.chars().count() > 100 {
|
||||
return Err(Error::Invalid(
|
||||
"downloader name must contain 1–100 characters".into(),
|
||||
));
|
||||
}
|
||||
self.connection.validate()
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(serde::Serialize)]
|
||||
pub struct DownloaderView {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub kind: DownloaderKind,
|
||||
pub url: String,
|
||||
pub username: String,
|
||||
pub save_path: String,
|
||||
pub password_configured: bool,
|
||||
pub is_active: bool,
|
||||
}
|
||||
@@ -0,0 +1,265 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::{
|
||||
config::QbitConfig,
|
||||
domain::validate_http_url,
|
||||
error::{Error, Result},
|
||||
feeds::bounded_body,
|
||||
};
|
||||
use bendy::decoding::{Decoder, Object};
|
||||
use reqwest::{multipart, Client, Url};
|
||||
use serde::Deserialize;
|
||||
use sha1::{Digest, Sha1};
|
||||
use std::time::Duration;
|
||||
|
||||
pub struct PreparedTorrent {
|
||||
pub hash: String,
|
||||
pub bytes: Option<Vec<u8>>,
|
||||
}
|
||||
|
||||
pub fn magnet_hash(value: &str) -> Result<String> {
|
||||
let url = Url::parse(value).map_err(|_| Error::Invalid("invalid magnet URI".into()))?;
|
||||
if url.scheme() != "magnet" {
|
||||
return Err(Error::Invalid("expected magnet URI".into()));
|
||||
}
|
||||
for (key, value) in url.query_pairs() {
|
||||
if key != "xt" {
|
||||
continue;
|
||||
}
|
||||
if let Some(hash) = value.strip_prefix("urn:btih:") {
|
||||
if hash.len() == 40 && hash.bytes().all(|b| b.is_ascii_hexdigit()) {
|
||||
return Ok(hash.to_ascii_lowercase());
|
||||
}
|
||||
if hash.len() == 32 {
|
||||
if let Ok(bytes) =
|
||||
data_encoding::BASE32.decode(hash.to_ascii_uppercase().as_bytes())
|
||||
{
|
||||
return Ok(data_encoding::HEXLOWER.encode(&bytes));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Err(Error::Invalid(
|
||||
"magnet must contain a valid BitTorrent v1 info hash".into(),
|
||||
))
|
||||
}
|
||||
|
||||
pub fn torrent_hash(bytes: &[u8]) -> Result<String> {
|
||||
let invalid = |_| Error::Invalid("invalid or unsupported torrent metadata".into());
|
||||
let mut decoder = Decoder::new(bytes).with_max_depth(32);
|
||||
let mut root = decoder
|
||||
.next_object()
|
||||
.map_err(invalid)?
|
||||
.ok_or_else(|| Error::Invalid("empty torrent metadata".into()))?
|
||||
.try_into_dictionary()
|
||||
.map_err(invalid)?;
|
||||
let mut raw_info = None;
|
||||
while let Some((key, object)) = root.next_pair().map_err(invalid)? {
|
||||
if key == b"info" {
|
||||
raw_info = Some(
|
||||
object
|
||||
.try_into_dictionary()
|
||||
.map_err(invalid)?
|
||||
.into_raw()
|
||||
.map_err(invalid)?,
|
||||
);
|
||||
}
|
||||
}
|
||||
drop(root);
|
||||
if decoder.next_object().map_err(invalid)?.is_some() {
|
||||
return Err(Error::Invalid("trailing torrent data".into()));
|
||||
}
|
||||
let raw =
|
||||
raw_info.ok_or_else(|| Error::Invalid("torrent is missing info dictionary".into()))?;
|
||||
let mut info_decoder = Decoder::new(raw).with_max_depth(32);
|
||||
let mut info = info_decoder
|
||||
.next_object()
|
||||
.map_err(invalid)?
|
||||
.unwrap()
|
||||
.try_into_dictionary()
|
||||
.map_err(invalid)?;
|
||||
let mut v1 = false;
|
||||
while let Some((key, object)) = info.next_pair().map_err(invalid)? {
|
||||
if let (b"pieces", Object::Bytes(pieces)) = (key, object) {
|
||||
v1 = !pieces.is_empty() && pieces.len() % 20 == 0;
|
||||
}
|
||||
}
|
||||
if !v1 {
|
||||
return Err(Error::Invalid(
|
||||
"only BitTorrent v1/hybrid torrents are supported".into(),
|
||||
));
|
||||
}
|
||||
// Hash the exact encoded bytes, never a decoded/re-encoded dictionary.
|
||||
Ok(format!("{:x}", Sha1::digest(raw)))
|
||||
}
|
||||
|
||||
pub async fn prepare(client: &Client, url: &str) -> Result<PreparedTorrent> {
|
||||
if url.starts_with("magnet:") {
|
||||
return Ok(PreparedTorrent {
|
||||
hash: magnet_hash(url)?,
|
||||
bytes: None,
|
||||
});
|
||||
}
|
||||
validate_http_url(url)?;
|
||||
let response =
|
||||
crate::feeds::website_get(client, Url::parse(url).expect("validated URL")).await?;
|
||||
let bytes = bounded_body(response, 4 * 1024 * 1024).await?;
|
||||
Ok(PreparedTorrent {
|
||||
hash: torrent_hash(&bytes)?,
|
||||
bytes: Some(bytes),
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Qbit {
|
||||
client: Client,
|
||||
config: QbitConfig,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct TorrentStatus {
|
||||
pub hash: String,
|
||||
pub progress: f64,
|
||||
pub state: String,
|
||||
#[serde(default)]
|
||||
pub save_path: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
pub struct TorrentFile {
|
||||
pub name: String,
|
||||
pub size: i64,
|
||||
pub priority: i32,
|
||||
}
|
||||
|
||||
impl Qbit {
|
||||
pub fn new(config: QbitConfig) -> Result<Self> {
|
||||
// Keep downloader cookies in a separate client from untrusted RSS hosts.
|
||||
let client = Client::builder()
|
||||
.no_proxy()
|
||||
.cookie_store(true)
|
||||
.timeout(Duration::from_secs(30))
|
||||
.redirect(reqwest::redirect::Policy::none())
|
||||
.build()?;
|
||||
Ok(Self { client, config })
|
||||
}
|
||||
|
||||
pub async fn login(&self) -> Result<()> {
|
||||
let response = self
|
||||
.client
|
||||
.post(format!("{}/api/v2/auth/login", self.config.url))
|
||||
.header("Referer", format!("{}/", self.config.url))
|
||||
.form(&[
|
||||
("username", &self.config.username),
|
||||
("password", &self.config.password),
|
||||
])
|
||||
.send()
|
||||
.await?;
|
||||
let bytes = bounded_body(response, 1024).await?;
|
||||
if bytes != b"Ok." {
|
||||
return Err(Error::Upstream("qBittorrent login failed".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn torrents(&self) -> Result<Vec<TorrentStatus>> {
|
||||
let response = self
|
||||
.client
|
||||
.get(format!("{}/api/v2/torrents/info", self.config.url))
|
||||
.header("Referer", format!("{}/", self.config.url))
|
||||
.send()
|
||||
.await?;
|
||||
let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
|
||||
serde_json::from_slice(&bytes)
|
||||
.map_err(|_| Error::Upstream("invalid qBittorrent task response".into()))
|
||||
}
|
||||
|
||||
pub async fn files(&self, hash: &str) -> Result<Vec<TorrentFile>> {
|
||||
let response = self
|
||||
.client
|
||||
.get(format!("{}/api/v2/torrents/files", self.config.url))
|
||||
.header("Referer", format!("{}/", self.config.url))
|
||||
.query(&[("hash", hash)])
|
||||
.send()
|
||||
.await?;
|
||||
let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
|
||||
let files: Vec<TorrentFile> = serde_json::from_slice(&bytes)
|
||||
.map_err(|_| Error::Upstream("invalid qBittorrent file list".into()))?;
|
||||
if files.len() > 5000 || files.iter().any(|file| file.size < 0) {
|
||||
return Err(Error::Upstream(
|
||||
"qBittorrent file list exceeds limits or contains invalid sizes".into(),
|
||||
));
|
||||
}
|
||||
Ok(files)
|
||||
}
|
||||
|
||||
pub async fn add(
|
||||
&self,
|
||||
url: &str,
|
||||
bytes: Option<Vec<u8>>,
|
||||
save_path: Option<&str>,
|
||||
) -> Result<()> {
|
||||
let mut form = multipart::Form::new()
|
||||
.text(
|
||||
"savepath",
|
||||
save_path.unwrap_or(&self.config.save_path).to_owned(),
|
||||
)
|
||||
.text("tags", "rust-project")
|
||||
.text("autoTMM", "false");
|
||||
form = if let Some(bytes) = bytes {
|
||||
form.part(
|
||||
"torrents",
|
||||
multipart::Part::bytes(bytes)
|
||||
.file_name("release.torrent")
|
||||
.mime_str("application/x-bittorrent")?,
|
||||
)
|
||||
} else {
|
||||
form.text("urls", url.to_owned())
|
||||
};
|
||||
let response = self
|
||||
.client
|
||||
.post(format!("{}/api/v2/torrents/add", self.config.url))
|
||||
.header("Referer", format!("{}/", self.config.url))
|
||||
.multipart(form)
|
||||
.send()
|
||||
.await?;
|
||||
let bytes = bounded_body(response, 1024).await?;
|
||||
if bytes != b"Ok." {
|
||||
return Err(Error::Upstream("qBittorrent rejected torrent".into()));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
#[test]
|
||||
fn handles_hex_and_base32_magnets() {
|
||||
assert_eq!(
|
||||
magnet_hash("magnet:?xt=urn:btih:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA").unwrap(),
|
||||
"a".repeat(40)
|
||||
);
|
||||
assert_eq!(
|
||||
magnet_hash("magnet:?xt=urn:btih:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA").unwrap(),
|
||||
"0".repeat(40)
|
||||
);
|
||||
assert!(magnet_hash("magnet:?xt=urn:btih:bad").is_err());
|
||||
}
|
||||
#[test]
|
||||
fn hashes_original_info_bytes_and_rejects_bad_metadata() {
|
||||
let info = b"d6:lengthi1e4:name1:a12:piece lengthi16384e6:pieces20:aaaaaaaaaaaaaaaaaaaae";
|
||||
let torrent = [b"d4:info".as_slice(), info, b"e"].concat();
|
||||
assert_eq!(
|
||||
torrent_hash(&torrent).unwrap(),
|
||||
format!("{:x}", Sha1::digest(info))
|
||||
);
|
||||
for bad in [
|
||||
b"<html>error</html>".as_slice(),
|
||||
b"d4:infodee",
|
||||
b"d4:info",
|
||||
&[torrent.as_slice(), b"junk"].concat(),
|
||||
] {
|
||||
assert!(torrent_hash(bad).is_err());
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,47 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use actix_web::{http::StatusCode, HttpResponse, ResponseError};
|
||||
|
||||
pub type Result<T> = std::result::Result<T, Error>;
|
||||
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum Error {
|
||||
#[error("{0}")]
|
||||
Invalid(String),
|
||||
#[error("resource not found")]
|
||||
NotFound,
|
||||
#[error("a scan or download synchronization is already running")]
|
||||
Busy,
|
||||
#[error("too many Mikan requests; retry shortly")]
|
||||
SourceBusy,
|
||||
#[error("{0}")]
|
||||
Upstream(String),
|
||||
#[error("database operation failed")]
|
||||
Database(#[from] sqlx::Error),
|
||||
#[error("stored data could not be decoded")]
|
||||
Json(#[from] serde_json::Error),
|
||||
#[error("database migration failed")]
|
||||
Migration(#[from] sqlx::migrate::MigrateError),
|
||||
}
|
||||
|
||||
impl From<reqwest::Error> for Error {
|
||||
fn from(value: reqwest::Error) -> Self {
|
||||
Self::Upstream(value.without_url().to_string())
|
||||
}
|
||||
}
|
||||
|
||||
impl ResponseError for Error {
|
||||
fn status_code(&self) -> StatusCode {
|
||||
match self {
|
||||
Self::Invalid(_) => StatusCode::BAD_REQUEST,
|
||||
Self::NotFound => StatusCode::NOT_FOUND,
|
||||
Self::Busy => StatusCode::CONFLICT,
|
||||
Self::SourceBusy => StatusCode::TOO_MANY_REQUESTS,
|
||||
Self::Upstream(_) => StatusCode::BAD_GATEWAY,
|
||||
_ => StatusCode::INTERNAL_SERVER_ERROR,
|
||||
}
|
||||
}
|
||||
|
||||
fn error_response(&self) -> HttpResponse {
|
||||
HttpResponse::build(self.status_code()).json(serde_json::json!({"error": self.to_string()}))
|
||||
}
|
||||
}
|
||||
+183
@@ -0,0 +1,183 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::{
|
||||
domain::*,
|
||||
error::{Error, Result},
|
||||
parser,
|
||||
};
|
||||
use reqwest::{Client, Url};
|
||||
use sha1::{Digest, Sha1};
|
||||
use std::time::Duration;
|
||||
|
||||
pub fn http_client() -> Result<Client> {
|
||||
website_client(&crate::config::WebsiteConfig::default())
|
||||
}
|
||||
|
||||
pub fn website_client(config: &crate::config::WebsiteConfig) -> Result<Client> {
|
||||
let mut builder = Client::builder()
|
||||
.no_proxy()
|
||||
.timeout(Duration::from_secs(30))
|
||||
.connect_timeout(Duration::from_secs(10))
|
||||
.user_agent(concat!("rust_project/", env!("CARGO_PKG_VERSION")))
|
||||
// Rebuild each redirected request so proxy auth and direct-host rules are reapplied.
|
||||
.redirect(reqwest::redirect::Policy::none());
|
||||
if let Some(value) = &config.proxy_url {
|
||||
let invalid = || {
|
||||
Error::Invalid(
|
||||
"WEBSITE_PROXY_URL must be an HTTP(S), SOCKS5 or SOCKS5H proxy URL".into(),
|
||||
)
|
||||
};
|
||||
let url = Url::parse(value).map_err(|_| invalid())?;
|
||||
if !matches!(url.scheme(), "http" | "https" | "socks5" | "socks5h")
|
||||
|| url.host_str().is_none()
|
||||
|| url.query().is_some()
|
||||
|| url.fragment().is_some()
|
||||
|| !matches!(url.path(), "" | "/")
|
||||
{
|
||||
return Err(invalid());
|
||||
}
|
||||
let proxy = reqwest::Proxy::all(url)
|
||||
.map_err(|_| invalid())?
|
||||
.no_proxy(reqwest::NoProxy::from_string(&config.direct_hosts));
|
||||
builder = builder.proxy(proxy);
|
||||
}
|
||||
Ok(builder.build()?)
|
||||
}
|
||||
|
||||
pub async fn website_get(client: &Client, mut url: Url) -> Result<reqwest::Response> {
|
||||
tokio::time::timeout(Duration::from_secs(30), async {
|
||||
for hop in 0..=5 {
|
||||
validate_http_url(url.as_str())?;
|
||||
let response = client.get(url.clone()).send().await?;
|
||||
if !matches!(response.status().as_u16(), 301 | 302 | 303 | 307 | 308) {
|
||||
return Ok(response);
|
||||
}
|
||||
if hop == 5 {
|
||||
return Err(Error::Upstream("website redirect limit exceeded".into()));
|
||||
}
|
||||
let location = response
|
||||
.headers()
|
||||
.get(reqwest::header::LOCATION)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.ok_or_else(|| Error::Upstream("invalid website redirect".into()))?;
|
||||
url = url
|
||||
.join(location)
|
||||
.map_err(|_| Error::Upstream("invalid website redirect URL".into()))?;
|
||||
validate_http_url(url.as_str())
|
||||
.map_err(|_| Error::Upstream("unsupported website redirect URL".into()))?;
|
||||
}
|
||||
unreachable!("redirect count is bounded")
|
||||
})
|
||||
.await
|
||||
.map_err(|_| Error::Upstream("website request timed out".into()))?
|
||||
}
|
||||
|
||||
pub async fn bounded_body(mut response: reqwest::Response, limit: usize) -> Result<Vec<u8>> {
|
||||
response = response.error_for_status()?;
|
||||
if response.content_length().is_some_and(|n| n > limit as u64) {
|
||||
return Err(Error::Upstream("response exceeds size limit".into()));
|
||||
}
|
||||
let mut data = Vec::new();
|
||||
while let Some(chunk) = response.chunk().await? {
|
||||
if data.len() + chunk.len() > limit {
|
||||
return Err(Error::Upstream("response exceeds size limit".into()));
|
||||
}
|
||||
data.extend_from_slice(&chunk);
|
||||
}
|
||||
Ok(data)
|
||||
}
|
||||
|
||||
pub async fn fetch(client: &Client, sub: &SubscriptionInput) -> Result<Vec<Release>> {
|
||||
let response = website_get(
|
||||
client,
|
||||
Url::parse(&sub.rss_url).map_err(|_| Error::Invalid("invalid RSS URL".into()))?,
|
||||
)
|
||||
.await?;
|
||||
let base = response.url().clone();
|
||||
let bytes = bounded_body(response, 4 * 1024 * 1024).await?;
|
||||
parse(&bytes, &base, sub)
|
||||
}
|
||||
|
||||
pub fn parse(bytes: &[u8], base: &Url, sub: &SubscriptionInput) -> Result<Vec<Release>> {
|
||||
let feed = feed_rs::parser::Builder::new()
|
||||
.base_uri(Some(base.as_str()))
|
||||
.build()
|
||||
.parse(bytes)
|
||||
.map_err(|_| Error::Upstream("invalid RSS/Atom feed".into()))?;
|
||||
if feed.entries.len() > 2000 {
|
||||
return Err(Error::Upstream("feed exceeds 2000 entries".into()));
|
||||
}
|
||||
Ok(feed
|
||||
.entries
|
||||
.into_iter()
|
||||
.map(|entry| {
|
||||
let title = entry.title.map(|t| t.content).unwrap_or_default();
|
||||
let media = entry
|
||||
.media
|
||||
.iter()
|
||||
.flat_map(|m| m.content.iter())
|
||||
.filter_map(|c| c.url.as_ref().map(|u| u.as_str()));
|
||||
let links = entry
|
||||
.links
|
||||
.iter()
|
||||
.filter(|l| {
|
||||
l.rel.as_deref() == Some("enclosure")
|
||||
|| l.media_type.as_deref() == Some("application/x-bittorrent")
|
||||
|| l.href.starts_with("magnet:")
|
||||
|| Url::parse(&l.href)
|
||||
.ok()
|
||||
.is_some_and(|u| u.path().ends_with(".torrent"))
|
||||
})
|
||||
.map(|l| l.href.as_str());
|
||||
let download_url = media.chain(links).find_map(|link| {
|
||||
let url = base.join(link).ok()?;
|
||||
match url.scheme() {
|
||||
"magnet" => Some(url.to_string()),
|
||||
"http" | "https" if validate_http_url(url.as_str()).is_ok() => {
|
||||
Some(url.to_string())
|
||||
}
|
||||
_ => None,
|
||||
}
|
||||
});
|
||||
let key = format!(
|
||||
"{:x}",
|
||||
Sha1::digest(download_url.as_deref().unwrap_or(&entry.id).as_bytes())
|
||||
);
|
||||
let mut release = Release {
|
||||
key,
|
||||
title,
|
||||
download_url,
|
||||
episode: None,
|
||||
decision: Decision::Ready,
|
||||
reason: "matched subscription rules".into(),
|
||||
};
|
||||
let lower = release.title.to_lowercase();
|
||||
if sub
|
||||
.exclude
|
||||
.iter()
|
||||
.any(|v| lower.contains(&v.to_lowercase()))
|
||||
{
|
||||
release.decision = Decision::Filtered;
|
||||
release.reason = "matched an exclusion keyword".into();
|
||||
} else if !sub
|
||||
.include
|
||||
.iter()
|
||||
.all(|v| lower.contains(&v.to_lowercase()))
|
||||
{
|
||||
release.decision = Decision::Filtered;
|
||||
release.reason = "missing a required inclusion keyword".into();
|
||||
} else if release.download_url.is_none() {
|
||||
release.decision = Decision::NeedsReview;
|
||||
release.reason = "no supported torrent enclosure or magnet link".into();
|
||||
} else {
|
||||
match parser::episode(&release.title, sub.season, sub.episode_offset) {
|
||||
Ok(ep) => release.episode = Some(ep),
|
||||
Err(reason) => {
|
||||
release.decision = Decision::NeedsReview;
|
||||
release.reason = reason.into();
|
||||
}
|
||||
}
|
||||
}
|
||||
release
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
+13
@@ -0,0 +1,13 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
pub mod api;
|
||||
pub mod config;
|
||||
pub mod domain;
|
||||
pub mod downloader;
|
||||
pub mod error;
|
||||
pub mod feeds;
|
||||
pub mod media;
|
||||
mod media_store;
|
||||
pub mod mikan;
|
||||
pub mod parser;
|
||||
pub mod service;
|
||||
pub mod storage;
|
||||
+148
-41
@@ -1,45 +1,152 @@
|
||||
use std::io;
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use actix_web::{web, App, HttpResponse, HttpServer, Responder};
|
||||
use rust_project::{
|
||||
api,
|
||||
config::Config,
|
||||
service::{worker, Service},
|
||||
storage::Store,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::sync::Arc;
|
||||
|
||||
fn main() {
|
||||
println!("欢迎来到增强版 Rust 程序！");
|
||||
|
||||
// 演示变量声明和类型d
|
||||
let name = "Rustacean"; // Rustacean 是 Rust 开发者的昵称
|
||||
let age: u32 = 10;
|
||||
|
||||
println!("你好，{}！你使用 Rust 已经 {} 年了。", name, age);
|
||||
|
||||
// 演示数组和循环
|
||||
let numbers = [1, 2, 3, 4, 5];
|
||||
println!("\n数字列表:");
|
||||
for (index, &value) in numbers.iter().enumerate() {
|
||||
println!("索引 {}: 值 {}", index, value);
|
||||
}
|
||||
|
||||
// 计算数组元素总和
|
||||
let sum: i32 = numbers.iter().sum();
|
||||
println!("\n数组元素总和: {}", sum);
|
||||
|
||||
// 条件判断
|
||||
if sum > 10 {
|
||||
println!("总和大于 10");
|
||||
} else {
|
||||
println!("总和小于或等于 10");
|
||||
}
|
||||
|
||||
// 与用户交互
|
||||
println!("\n请输入您的名字:");
|
||||
let mut input = String::new();
|
||||
|
||||
match io::stdin().read_line(&mut input) {
|
||||
Ok(_) => {
|
||||
let input = input.trim(); // 移除换行符
|
||||
if !input.is_empty() {
|
||||
println!("很高兴认识您，{}！", input);
|
||||
} else {
|
||||
println!("您没有输入任何内容。");
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
struct Greeting {
|
||||
message: String,
|
||||
name: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
struct Health {
|
||||
status: String,
|
||||
version: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
struct CalcRequest {
|
||||
a: f64,
|
||||
b: f64,
|
||||
operation: String,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
struct CalcResponse {
|
||||
result: f64,
|
||||
operation: String,
|
||||
}
|
||||
|
||||
async fn hello() -> impl Responder {
|
||||
HttpResponse::Ok().body("Hello World! 欢迎使用 Rust Web 服务器")
|
||||
}
|
||||
|
||||
async fn health_check() -> impl Responder {
|
||||
let health = Health {
|
||||
status: "ok".to_string(),
|
||||
version: env!("CARGO_PKG_VERSION").to_string(),
|
||||
};
|
||||
HttpResponse::Ok().json(health)
|
||||
}
|
||||
|
||||
async fn greet(name: web::Path<String>) -> impl Responder {
|
||||
let greeting = Greeting {
|
||||
message: format!("你好，{}！", name),
|
||||
name: name.into_inner(),
|
||||
};
|
||||
HttpResponse::Ok().json(greeting)
|
||||
}
|
||||
|
||||
async fn calculate(web::Json(req): web::Json<CalcRequest>) -> impl Responder {
|
||||
let result = match req.operation.as_str() {
|
||||
"add" => req.a + req.b,
|
||||
"subtract" => req.a - req.b,
|
||||
"multiply" => req.a * req.b,
|
||||
"divide" => {
|
||||
if req.b == 0.0 {
|
||||
return HttpResponse::BadRequest().json("除数不能为 0");
|
||||
}
|
||||
req.a / req.b
|
||||
}
|
||||
Err(error) => println!("读取输入时发生错误: {}", error),
|
||||
}
|
||||
_ => return HttpResponse::BadRequest().json("不支持的操作"),
|
||||
};
|
||||
|
||||
let response = CalcResponse {
|
||||
result,
|
||||
operation: req.operation,
|
||||
};
|
||||
HttpResponse::Ok().json(response)
|
||||
}
|
||||
|
||||
async fn echo(data: web::Json<serde_json::Value>) -> impl Responder {
|
||||
HttpResponse::Ok().json(data)
|
||||
}
|
||||
|
||||
#[actix_web::main]
|
||||
async fn main() -> std::io::Result<()> {
|
||||
tracing_subscriber::fmt()
|
||||
.with_env_filter(
|
||||
tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
|
||||
)
|
||||
.init();
|
||||
let config = Config::from_env().map_err(std::io::Error::other)?;
|
||||
std::fs::create_dir_all("data")?;
|
||||
let store = Store::open(&config.database_url)
|
||||
.await
|
||||
.map_err(std::io::Error::other)?;
|
||||
let service = Arc::new(
|
||||
Service::new(store, &config)
|
||||
.await
|
||||
.map_err(std::io::Error::other)?,
|
||||
);
|
||||
let app_service = service.clone();
|
||||
let server = HttpServer::new(move || {
|
||||
App::new()
|
||||
.app_data(web::Data::new(app_service.clone()))
|
||||
.app_data(web::JsonConfig::default().limit(32 * 1024))
|
||||
.route("/", web::get().to(hello))
|
||||
.route("/health", web::get().to(health_check))
|
||||
.route("/greet/{name}", web::get().to(greet))
|
||||
.route("/api/echo", web::post().to(echo))
|
||||
.route("/api/calculate", web::post().to(calculate))
|
||||
.configure(api::configure)
|
||||
.route(
|
||||
"/info",
|
||||
web::get().to(|| async {
|
||||
HttpResponse::Ok().json(serde_json::json!({
|
||||
"project": env!("CARGO_PKG_NAME"),
|
||||
"version": env!("CARGO_PKG_VERSION"),
|
||||
"endpoints": [
|
||||
{ "method": "GET", "path": "/", "description": "欢迎页面" },
|
||||
{ "method": "GET", "path": "/health", "description": "健康检查" },
|
||||
{ "method": "GET", "path": "/greet/{name}", "description": "问候接口" },
|
||||
{ "method": "POST", "path": "/api/echo", "description": "回显数据" },
|
||||
{ "method": "POST", "path": "/api/calculate", "description": "计算接口" }
|
||||
]
|
||||
}))
|
||||
}),
|
||||
)
|
||||
})
|
||||
.bind(("127.0.0.1", config.port))?;
|
||||
|
||||
let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
|
||||
let background = tokio::spawn(worker(service, config.poll_interval, shutdown_rx));
|
||||
|
||||
println!("\n==================================================");
|
||||
println!("🚀 Rust Web 服务器已启动");
|
||||
println!("📡 监听地址: http://127.0.0.1:{}", config.port);
|
||||
println!(" GET/POST /api/subscriptions - 订阅管理");
|
||||
println!(" POST /api/preview - RSS 解析预览");
|
||||
println!(" POST /api/run - 执行订阅与下载同步");
|
||||
println!(" GET /api/jobs - 下载任务");
|
||||
println!("📋 可用端点:");
|
||||
println!(" GET / - 欢迎页面");
|
||||
println!(" GET /health - 健康检查");
|
||||
println!(" GET /info - API 信息");
|
||||
println!(" GET /greet/{{name}} - 问候接口");
|
||||
println!(" POST /api/echo - 回显数据");
|
||||
println!(" POST /api/calculate - 计算接口");
|
||||
println!("==================================================\n");
|
||||
|
||||
let result = server.run().await;
|
||||
let _ = shutdown_tx.send(true);
|
||||
let _ = background.await;
|
||||
result
|
||||
}
|
||||
|
||||
+269
@@ -0,0 +1,269 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::error::{Error, Result};
|
||||
use cap_std::{ambient_authority, fs::Dir};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use std::{
|
||||
io,
|
||||
path::{Component, Path},
|
||||
};
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct DestinationInput {
|
||||
pub name: String,
|
||||
pub kind: LocalKind,
|
||||
pub root_path: String,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum LocalKind {
|
||||
Local,
|
||||
}
|
||||
|
||||
impl DestinationInput {
|
||||
pub fn validate(&mut self) -> Result<()> {
|
||||
self.name = self.name.trim().to_owned();
|
||||
if self.name.is_empty()
|
||||
|| self.name.chars().count() > 100
|
||||
|| self.root_path.len() > 4096
|
||||
|| self.root_path.chars().any(char::is_control)
|
||||
|| !Path::new(&self.root_path).is_absolute()
|
||||
|| Path::new(&self.root_path)
|
||||
.components()
|
||||
.any(|c| matches!(c, Component::ParentDir))
|
||||
{
|
||||
return Err(Error::Invalid(
|
||||
"expected a name and an absolute local root without parent traversal".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize, sqlx::FromRow)]
|
||||
pub struct Destination {
|
||||
pub id: String,
|
||||
pub name: String,
|
||||
pub kind: String,
|
||||
pub root_path: String,
|
||||
pub is_active: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(deny_unknown_fields)]
|
||||
pub struct MappingInput {
|
||||
pub downloader_id: String,
|
||||
pub download_root: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize, sqlx::FromRow)]
|
||||
pub struct Mapping {
|
||||
pub id: String,
|
||||
pub destination_id: String,
|
||||
pub downloader_id: String,
|
||||
pub version: i64,
|
||||
pub download_root: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize, sqlx::FromRow)]
|
||||
pub struct MediaFile {
|
||||
pub relative_path: String,
|
||||
pub expected_size: i64,
|
||||
pub observed_size: Option<i64>,
|
||||
pub status: String,
|
||||
pub error: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct FileReport {
|
||||
pub job_id: String,
|
||||
pub destination_id: Option<String>,
|
||||
pub status: String,
|
||||
pub checked_at: Option<i64>,
|
||||
pub last_success_at: Option<i64>,
|
||||
pub error: Option<String>,
|
||||
pub files: Vec<MediaFile>,
|
||||
}
|
||||
|
||||
// Wire paths always use '/' and must be relative on both Unix and Windows.
|
||||
pub fn validate_relative(path: &str, allow_empty: bool) -> Result<()> {
|
||||
if (path.is_empty() && !allow_empty)
|
||||
|| path.len() > 4096
|
||||
|| path.contains(['\\', ':'])
|
||||
|| path.chars().any(char::is_control)
|
||||
|| (!path.is_empty()
|
||||
&& path
|
||||
.split('/')
|
||||
.any(|part| part.is_empty() || matches!(part, "." | "..")))
|
||||
{
|
||||
return Err(Error::Invalid(
|
||||
"expected a relative path without empty, dot, parent, drive or backslash components"
|
||||
.into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// qBittorrent may run on a different OS. Do not interpret its paths using the host OS.
|
||||
pub fn normalize_download_root(value: &str) -> Result<String> {
|
||||
let path = value.replace('\\', "/");
|
||||
let drive = path.len() >= 3
|
||||
&& path.as_bytes()[0].is_ascii_alphabetic()
|
||||
&& path.as_bytes()[1] == b':'
|
||||
&& path.as_bytes()[2] == b'/';
|
||||
if path.len() > 4096 || path.chars().any(char::is_control) || !(path.starts_with('/') || drive)
|
||||
{
|
||||
return Err(Error::Invalid(
|
||||
"downloader root must be an absolute Unix or Windows path".into(),
|
||||
));
|
||||
}
|
||||
let rest = if drive {
|
||||
&path[3..]
|
||||
} else {
|
||||
path.trim_start_matches('/')
|
||||
};
|
||||
if rest.contains(':') || rest.split('/').any(|s| matches!(s, "." | "..")) {
|
||||
return Err(Error::Invalid("invalid downloader root components".into()));
|
||||
}
|
||||
let path = path.trim_end_matches('/');
|
||||
Ok(if path.is_empty() {
|
||||
"/".into()
|
||||
} else if drive && path.len() == 2 {
|
||||
format!("{path}/")
|
||||
} else {
|
||||
path.into()
|
||||
})
|
||||
}
|
||||
|
||||
pub fn same_download_root(expected: &str, actual: &str) -> bool {
|
||||
match (
|
||||
normalize_download_root(expected),
|
||||
normalize_download_root(actual),
|
||||
) {
|
||||
(Ok(a), Ok(b)) => a == b,
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
fn access_error(error: &io::Error) -> String {
|
||||
match error.kind() {
|
||||
io::ErrorKind::PermissionDenied => "access denied or path escapes the configured root",
|
||||
io::ErrorKind::NotFound => "configured root or directory is unavailable",
|
||||
_ => "filesystem access failed",
|
||||
}
|
||||
.into()
|
||||
}
|
||||
|
||||
pub fn check_local(root: &str, mut files: Vec<MediaFile>) -> Vec<MediaFile> {
|
||||
let dir = Dir::open_ambient_dir(root, ambient_authority());
|
||||
for file in &mut files {
|
||||
file.observed_size = None;
|
||||
file.error = None;
|
||||
let result = match &dir {
|
||||
Ok(dir) => {
|
||||
if validate_relative(&file.relative_path, false).is_err() {
|
||||
file.status = "unavailable".into();
|
||||
file.error = Some("invalid stored relative path".into());
|
||||
continue;
|
||||
}
|
||||
dir.metadata(&file.relative_path)
|
||||
}
|
||||
Err(error) => {
|
||||
file.status = "unavailable".into();
|
||||
file.error = Some(access_error(error));
|
||||
continue;
|
||||
}
|
||||
};
|
||||
match result {
|
||||
Ok(metadata) if metadata.is_file() => {
|
||||
file.status = "present".into();
|
||||
file.observed_size = i64::try_from(metadata.len()).ok();
|
||||
}
|
||||
Ok(_) => {
|
||||
file.status = "unavailable".into();
|
||||
file.error = Some("expected a regular file".into());
|
||||
}
|
||||
Err(error) if error.kind() == io::ErrorKind::NotFound => file.status = "absent".into(),
|
||||
Err(error) => {
|
||||
file.status = "unavailable".into();
|
||||
file.error = Some(access_error(&error));
|
||||
}
|
||||
}
|
||||
}
|
||||
files
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct DirectoryEntry {
|
||||
pub name: String,
|
||||
pub kind: &'static str,
|
||||
pub size: Option<u64>,
|
||||
}
|
||||
#[derive(Serialize)]
|
||||
pub struct DirectoryListing {
|
||||
pub status: &'static str,
|
||||
pub error: Option<String>,
|
||||
pub entries: Vec<DirectoryEntry>,
|
||||
pub truncated: bool,
|
||||
}
|
||||
|
||||
pub fn list_local(root: &str, path: &str) -> DirectoryListing {
|
||||
if let Err(error) = validate_relative(path, true) {
|
||||
return DirectoryListing {
|
||||
status: "unavailable",
|
||||
error: Some(error.to_string()),
|
||||
entries: vec![],
|
||||
truncated: false,
|
||||
};
|
||||
}
|
||||
let result = (|| -> io::Result<_> {
|
||||
let dir = Dir::open_ambient_dir(root, ambient_authority())?;
|
||||
let mut entries = Vec::new();
|
||||
let mut truncated = false;
|
||||
for (index, item) in dir
|
||||
.read_dir(if path.is_empty() { "." } else { path })?
|
||||
.enumerate()
|
||||
{
|
||||
if index >= 1000 {
|
||||
truncated = true;
|
||||
break;
|
||||
}
|
||||
let item = item?;
|
||||
let kind = item.file_type()?;
|
||||
entries.push(DirectoryEntry {
|
||||
name: item.file_name().to_string_lossy().into_owned(),
|
||||
size: if kind.is_file() {
|
||||
Some(item.metadata()?.len())
|
||||
} else {
|
||||
None
|
||||
},
|
||||
kind: if kind.is_symlink() {
|
||||
"symlink"
|
||||
} else if kind.is_dir() {
|
||||
"directory"
|
||||
} else if kind.is_file() {
|
||||
"file"
|
||||
} else {
|
||||
"other"
|
||||
},
|
||||
});
|
||||
}
|
||||
entries.sort_by(|a, b| a.name.cmp(&b.name));
|
||||
Ok((entries, truncated))
|
||||
})();
|
||||
match result {
|
||||
Ok((entries, truncated)) => DirectoryListing {
|
||||
status: "available",
|
||||
error: None,
|
||||
entries,
|
||||
truncated,
|
||||
},
|
||||
Err(error) => DirectoryListing {
|
||||
status: "unavailable",
|
||||
error: Some(access_error(&error)),
|
||||
entries: vec![],
|
||||
truncated: false,
|
||||
},
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,153 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::{
|
||||
domain::DownloadJob,
|
||||
error::{Error, Result},
|
||||
media::*,
|
||||
storage::{now, Store},
|
||||
};
|
||||
|
||||
impl Store {
|
||||
pub async fn destinations(&self) -> Result<Vec<Destination>> {
|
||||
Ok(sqlx::query_as("SELECT d.*, COALESCE(s.destination_id = d.id, 0) AS is_active FROM media_destinations d CROSS JOIN destination_selection s ORDER BY d.rowid")
|
||||
.fetch_all(&self.pool).await?)
|
||||
}
|
||||
|
||||
pub async fn destination(&self, id: &str) -> Result<Destination> {
|
||||
self.destinations()
|
||||
.await?
|
||||
.into_iter()
|
||||
.find(|d| d.id == id)
|
||||
.ok_or(Error::NotFound)
|
||||
}
|
||||
|
||||
pub async fn create_destination(&self, mut input: DestinationInput) -> Result<Destination> {
|
||||
input.validate()?;
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
sqlx::query(
|
||||
"INSERT INTO media_destinations(id, name, kind, root_path) VALUES (?, ?, 'local', ?)",
|
||||
)
|
||||
.bind(&id)
|
||||
.bind(input.name)
|
||||
.bind(input.root_path)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
self.destination(&id).await
|
||||
}
|
||||
|
||||
pub async fn active_destination_id(&self) -> Result<Option<String>> {
|
||||
Ok(sqlx::query_scalar(
|
||||
"SELECT destination_id FROM destination_selection WHERE singleton = 1",
|
||||
)
|
||||
.fetch_one(&self.pool)
|
||||
.await?)
|
||||
}
|
||||
|
||||
pub async fn activate_destination(&self, id: &str) -> Result<()> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let exists: bool =
|
||||
sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM media_destinations WHERE id = ?)")
|
||||
.bind(id)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
if !exists {
|
||||
return Err(Error::NotFound);
|
||||
}
|
||||
let mapped: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM destination_mappings m JOIN downloader_selection s ON s.downloader_id = m.downloader_id WHERE m.destination_id = ?)")
|
||||
.bind(id).fetch_one(&mut *tx).await?;
|
||||
if !mapped {
|
||||
return Err(Error::Invalid(
|
||||
"select a downloader and configure its mapping before activating this destination"
|
||||
.into(),
|
||||
));
|
||||
}
|
||||
sqlx::query("UPDATE destination_selection SET destination_id = ? WHERE singleton = 1")
|
||||
.bind(id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn mappings(&self, id: &str) -> Result<Vec<Mapping>> {
|
||||
self.destination(id).await?;
|
||||
Ok(sqlx::query_as("SELECT * FROM destination_mappings WHERE destination_id = ? ORDER BY version DESC, downloader_id")
|
||||
.bind(id).fetch_all(&self.pool).await?)
|
||||
}
|
||||
|
||||
pub async fn create_mapping(
|
||||
&self,
|
||||
destination_id: &str,
|
||||
input: MappingInput,
|
||||
) -> Result<Mapping> {
|
||||
self.destination(destination_id).await?;
|
||||
self.downloader_config(&input.downloader_id).await?;
|
||||
let root = normalize_download_root(&input.download_root)?;
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let version: i64 = sqlx::query_scalar("SELECT COALESCE(MAX(version), 0) + 1 FROM destination_mappings WHERE destination_id = ? AND downloader_id = ?")
|
||||
.bind(destination_id).bind(&input.downloader_id).fetch_one(&mut *tx).await?;
|
||||
sqlx::query("INSERT INTO destination_mappings(id, destination_id, downloader_id, version, download_root) VALUES (?, ?, ?, ?, ?)")
|
||||
.bind(&id).bind(destination_id).bind(&input.downloader_id).bind(version).bind(&root).execute(&mut *tx).await?;
|
||||
tx.commit().await?;
|
||||
Ok(Mapping {
|
||||
id,
|
||||
destination_id: destination_id.into(),
|
||||
downloader_id: input.downloader_id,
|
||||
version,
|
||||
download_root: root,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn job(&self, id: &str) -> Result<DownloadJob> {
|
||||
sqlx::query_as("SELECT * FROM download_jobs WHERE id = ?")
|
||||
.bind(id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?
|
||||
.ok_or(Error::NotFound)
|
||||
}
|
||||
|
||||
pub async fn file_report(&self, id: &str) -> Result<FileReport> {
|
||||
let job = self.job(id).await?;
|
||||
let files = sqlx::query_as("SELECT relative_path, expected_size, observed_size, status, error FROM media_files WHERE job_id = ? ORDER BY relative_path")
|
||||
.bind(id).fetch_all(&self.pool).await?;
|
||||
Ok(FileReport {
|
||||
job_id: id.into(),
|
||||
destination_id: job.destination_id,
|
||||
status: job.file_status,
|
||||
checked_at: job.file_checked_at,
|
||||
last_success_at: job.file_last_success_at,
|
||||
error: job.file_error,
|
||||
files,
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) async fn save_manifest(&self, id: &str, files: &[MediaFile]) -> Result<()> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
for file in files {
|
||||
validate_relative(&file.relative_path, false)?;
|
||||
sqlx::query("INSERT INTO media_files(job_id, relative_path, expected_size) VALUES (?, ?, ?) ON CONFLICT(job_id, relative_path) DO NOTHING")
|
||||
.bind(id).bind(&file.relative_path).bind(file.expected_size).execute(&mut *tx).await?;
|
||||
}
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub(crate) async fn save_file_check(
|
||||
&self,
|
||||
id: &str,
|
||||
status: &str,
|
||||
error: Option<&str>,
|
||||
files: &[MediaFile],
|
||||
) -> Result<()> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let checked_at = now();
|
||||
for file in files {
|
||||
sqlx::query("UPDATE media_files SET observed_size = ?, status = ?, error = ? WHERE job_id = ? AND relative_path = ?")
|
||||
.bind(file.observed_size).bind(&file.status).bind(&file.error).bind(id).bind(&file.relative_path).execute(&mut *tx).await?;
|
||||
}
|
||||
sqlx::query("UPDATE download_jobs SET file_status = ?, file_error = ?, file_checked_at = ?, file_last_success_at = CASE WHEN ? IN ('present', 'absent') THEN ? ELSE file_last_success_at END WHERE id = ?")
|
||||
.bind(status).bind(error).bind(checked_at).bind(status).bind(checked_at).bind(id).execute(&mut *tx).await?;
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
+460
@@ -0,0 +1,460 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
//! Public Mikan discovery; HTML details stay inside this adapter.
|
||||
use crate::{
|
||||
domain::{validate_http_url, SubscriptionInput},
|
||||
error::{Error, Result},
|
||||
feeds::{bounded_body, website_get},
|
||||
};
|
||||
use reqwest::{Client, Url};
|
||||
use scraper::{Html, Selector};
|
||||
use serde::Serialize;
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
sync::OnceLock,
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
use tokio::sync::{Mutex, Semaphore};
|
||||
|
||||
pub struct Mikan {
|
||||
client: Client,
|
||||
base: Url,
|
||||
cache: Mutex<HashMap<String, (Instant, Cached)>>,
|
||||
requests: Semaphore,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
enum Cached {
|
||||
Search(SearchResult),
|
||||
Series(Box<SeriesDetail>),
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
pub struct SeriesSummary {
|
||||
pub series_id: String,
|
||||
pub title: String,
|
||||
pub page_url: String,
|
||||
pub cover_url: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
pub struct SearchResult {
|
||||
pub source_id: &'static str,
|
||||
pub query: String,
|
||||
pub series: Vec<SeriesSummary>,
|
||||
pub keyword_rss_url: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
pub struct SubtitleGroup {
|
||||
pub group_id: String,
|
||||
pub name: String,
|
||||
pub rss_url: String,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
pub struct SeriesDetail {
|
||||
pub source_id: &'static str,
|
||||
pub series_id: String,
|
||||
pub title: String,
|
||||
pub page_url: String,
|
||||
pub rss_url: String,
|
||||
pub groups: Vec<SubtitleGroup>,
|
||||
pub information: Vec<String>,
|
||||
pub bangumi_id: Option<String>,
|
||||
pub cover_url: Option<String>,
|
||||
#[serde(skip)]
|
||||
poster: Option<Url>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
pub struct Candidate {
|
||||
pub source_id: &'static str,
|
||||
pub scope: &'static str,
|
||||
pub series_id: Option<String>,
|
||||
pub group_id: Option<String>,
|
||||
pub page_url: String,
|
||||
pub rss_url: String,
|
||||
pub subscription: Option<SubscriptionInput>,
|
||||
pub warnings: Vec<&'static str>,
|
||||
}
|
||||
|
||||
fn text(element: scraper::ElementRef<'_>) -> String {
|
||||
element
|
||||
.text()
|
||||
.collect::<String>()
|
||||
.split_whitespace()
|
||||
.collect::<Vec<_>>()
|
||||
.join(" ")
|
||||
}
|
||||
|
||||
fn parameters(url: &Url) -> Result<HashMap<String, String>> {
|
||||
let mut params = HashMap::new();
|
||||
for (key, value) in url.query_pairs() {
|
||||
if params
|
||||
.insert(key.to_ascii_lowercase(), value.into_owned())
|
||||
.is_some()
|
||||
{
|
||||
return Err(Error::Invalid("duplicate URL parameter".into()));
|
||||
}
|
||||
}
|
||||
Ok(params)
|
||||
}
|
||||
|
||||
fn selector(value: &str) -> Selector {
|
||||
Selector::parse(value).expect("static selector")
|
||||
}
|
||||
|
||||
fn valid_id(value: &str) -> bool {
|
||||
!value.is_empty() && value.len() <= 20 && value.bytes().all(|b| b.is_ascii_digit())
|
||||
}
|
||||
|
||||
impl Mikan {
|
||||
pub fn new(client: Client, base: &str) -> Result<Self> {
|
||||
validate_http_url(base)?;
|
||||
let base = Url::parse(base).expect("validated URL");
|
||||
if base.path() != "/" || base.query().is_some() || base.fragment().is_some() {
|
||||
return Err(Error::Invalid(
|
||||
"MIKAN_BASE_URL must be an HTTP(S) origin without a path or query".into(),
|
||||
));
|
||||
}
|
||||
Ok(Self {
|
||||
client,
|
||||
base,
|
||||
cache: Mutex::new(HashMap::new()),
|
||||
requests: Semaphore::new(4),
|
||||
})
|
||||
}
|
||||
|
||||
async fn cached(&self, key: &str) -> Option<Cached> {
|
||||
self.cache
|
||||
.lock()
|
||||
.await
|
||||
.get(key)
|
||||
.filter(|(created, _)| created.elapsed() < Duration::from_secs(60))
|
||||
.map(|(_, value)| value.clone())
|
||||
}
|
||||
|
||||
async fn remember(&self, key: String, value: Cached) {
|
||||
let mut cache = self.cache.lock().await;
|
||||
cache.retain(|_, (created, _)| created.elapsed() < Duration::from_secs(60));
|
||||
if cache.len() >= 64 {
|
||||
if let Some(oldest) = cache
|
||||
.iter()
|
||||
.min_by_key(|(_, (time, _))| *time)
|
||||
.map(|(key, _)| key.clone())
|
||||
{
|
||||
cache.remove(&oldest);
|
||||
}
|
||||
}
|
||||
cache.insert(key, (Instant::now(), value));
|
||||
}
|
||||
|
||||
async fn html(&self, url: Url) -> Result<Html> {
|
||||
let response = website_get(&self.client, url).await?;
|
||||
let bytes = bounded_body(response, 8 * 1024 * 1024).await?;
|
||||
let text = std::str::from_utf8(&bytes)
|
||||
.map_err(|_| Error::Upstream("Mikan returned invalid UTF-8".into()))?;
|
||||
Ok(Html::parse_document(text))
|
||||
}
|
||||
|
||||
fn same_origin_url(&self, value: &str) -> Result<Url> {
|
||||
let url = self
|
||||
.base
|
||||
.join(value)
|
||||
.map_err(|_| Error::Invalid("invalid Mikan URL".into()))?;
|
||||
validate_http_url(url.as_str())?;
|
||||
if url.origin() != self.base.origin() {
|
||||
return Err(Error::Invalid(
|
||||
"URL must belong to the configured Mikan origin".into(),
|
||||
));
|
||||
}
|
||||
Ok(url)
|
||||
}
|
||||
|
||||
fn rss_link(
|
||||
&self,
|
||||
element: scraper::ElementRef<'_>,
|
||||
series_id: &str,
|
||||
) -> Option<(Url, Option<String>)> {
|
||||
let url = self.same_origin_url(element.value().attr("href")?).ok()?;
|
||||
if url.path() != "/RSS/Bangumi" {
|
||||
return None;
|
||||
}
|
||||
let params = parameters(&url).ok()?;
|
||||
if params.get("bangumiid").map(String::as_str) != Some(series_id) {
|
||||
return None;
|
||||
}
|
||||
let group = params.get("subgroupid").cloned();
|
||||
if group.as_ref().is_some_and(|id| !valid_id(id)) {
|
||||
return None;
|
||||
}
|
||||
Some((url, group))
|
||||
}
|
||||
|
||||
pub async fn series(&self, id: &str) -> Result<SeriesDetail> {
|
||||
if !valid_id(id) {
|
||||
return Err(Error::Invalid("invalid Mikan series ID".into()));
|
||||
}
|
||||
let key = format!("series:{id}");
|
||||
if let Some(Cached::Series(detail)) = self.cached(&key).await {
|
||||
return Ok(*detail);
|
||||
}
|
||||
let _permit = self.requests.try_acquire().map_err(|_| Error::SourceBusy)?;
|
||||
let page = self
|
||||
.base
|
||||
.join(&format!("/Home/Bangumi/{id}"))
|
||||
.expect("numeric ID");
|
||||
let html = self.html(page.clone()).await?;
|
||||
let title = html
|
||||
.select(&selector(".bangumi-title"))
|
||||
.next()
|
||||
.map(text)
|
||||
.filter(|name| !name.is_empty())
|
||||
.ok_or_else(|| {
|
||||
Error::Upstream("Mikan detail page is unrecognized or access is restricted".into())
|
||||
})?;
|
||||
let rss_url = html
|
||||
.select(&selector(".bangumi-title a.mikan-rss[href]"))
|
||||
.filter_map(|link| self.rss_link(link, id))
|
||||
.find(|(_, group)| group.is_none())
|
||||
.map(|(url, _)| url.to_string())
|
||||
.ok_or_else(|| Error::Upstream("Mikan series RSS link is missing or invalid".into()))?;
|
||||
let mut groups = Vec::new();
|
||||
let mut seen = HashSet::new();
|
||||
for block in html.select(&selector(".subgroup-text")) {
|
||||
let Some((url, Some(group_id))) = block
|
||||
.select(&selector("a.mikan-rss[href]"))
|
||||
.find_map(|link| self.rss_link(link, id))
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let Some(name) = block
|
||||
.select(&selector("a[href*='/Home/PublishGroup/']"))
|
||||
.next()
|
||||
.map(text)
|
||||
.filter(|name| !name.is_empty())
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
if seen.insert(group_id.clone()) {
|
||||
groups.push(SubtitleGroup {
|
||||
group_id,
|
||||
name,
|
||||
rss_url: url.into(),
|
||||
});
|
||||
}
|
||||
}
|
||||
let information = html.select(&selector(".bangumi-info")).map(text).collect();
|
||||
let bangumi_id = html
|
||||
.select(&selector(".bangumi-info a[href]"))
|
||||
.find_map(|link| {
|
||||
let url = Url::parse(link.value().attr("href")?).ok()?;
|
||||
if !matches!(url.host_str()?, "bgm.tv" | "bangumi.tv" | "chii.in") {
|
||||
return None;
|
||||
}
|
||||
url.path()
|
||||
.strip_prefix("/subject/")
|
||||
.filter(|id| valid_id(id))
|
||||
.map(str::to_owned)
|
||||
});
|
||||
static CSS_URL: OnceLock<regex::Regex> = OnceLock::new();
|
||||
let pattern = CSS_URL.get_or_init(|| {
|
||||
regex::Regex::new(r#"url\(\s*['"]?([^'"\s)]+)['"]?\s*\)"#).expect("static regex")
|
||||
});
|
||||
let poster = html
|
||||
.select(&selector(".bangumi-poster[style]"))
|
||||
.find_map(|element| {
|
||||
let captures = pattern.captures(element.value().attr("style")?)?;
|
||||
let url = self.same_origin_url(captures.get(1)?.as_str()).ok()?;
|
||||
url.path().starts_with("/images/").then_some(url)
|
||||
});
|
||||
let detail = SeriesDetail {
|
||||
source_id: "mikan",
|
||||
series_id: id.into(),
|
||||
title,
|
||||
page_url: page.into(),
|
||||
rss_url,
|
||||
groups,
|
||||
information,
|
||||
bangumi_id,
|
||||
cover_url: poster
|
||||
.as_ref()
|
||||
.map(|_| format!("/api/sources/mikan/series/{id}/cover")),
|
||||
poster,
|
||||
};
|
||||
self.remember(key, Cached::Series(Box::new(detail.clone())))
|
||||
.await;
|
||||
Ok(detail)
|
||||
}
|
||||
|
||||
pub async fn cover(&self, id: &str) -> Result<(String, Vec<u8>)> {
|
||||
let url = self.series(id).await?.poster.ok_or(Error::NotFound)?;
|
||||
let _permit = self.requests.try_acquire().map_err(|_| Error::SourceBusy)?;
|
||||
let response = website_get(&self.client, url).await?;
|
||||
let mime = response
|
||||
.headers()
|
||||
.get(reqwest::header::CONTENT_TYPE)
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.unwrap_or("")
|
||||
.split(';')
|
||||
.next()
|
||||
.unwrap_or("")
|
||||
.trim()
|
||||
.to_ascii_lowercase();
|
||||
if !matches!(
|
||||
mime.as_str(),
|
||||
"image/png" | "image/jpeg" | "image/webp" | "image/gif"
|
||||
) {
|
||||
return Err(Error::Upstream(
|
||||
"Mikan cover is not a supported raster image".into(),
|
||||
));
|
||||
}
|
||||
Ok((mime, bounded_body(response, 4 * 1024 * 1024).await?))
|
||||
}
|
||||
|
||||
pub async fn resolve(&self, value: &str) -> Result<Candidate> {
|
||||
if value.len() > 8192 {
|
||||
return Err(Error::Invalid("Mikan URL is too long".into()));
|
||||
}
|
||||
let url = self.same_origin_url(value.trim())?;
|
||||
let params = parameters(&url)?;
|
||||
if matches!(url.path(), "/RSS/Search" | "/Home/Search") {
|
||||
let query = params
|
||||
.get("searchstr")
|
||||
.map(|s| s.trim())
|
||||
.filter(|s| !s.is_empty() && s.chars().count() <= 100)
|
||||
.ok_or_else(|| {
|
||||
Error::Invalid("keyword RSS requires a query of 1–100 characters".into())
|
||||
})?;
|
||||
if params.len() != 1 {
|
||||
return Err(Error::Invalid("unsupported keyword RSS parameters".into()));
|
||||
}
|
||||
let mut rss = self.base.join("/RSS/Search").expect("static path");
|
||||
rss.query_pairs_mut().append_pair("searchstr", query);
|
||||
let mut page = rss.clone();
|
||||
page.set_path("/Home/Search");
|
||||
return Ok(Candidate { source_id: "mikan", scope: "keyword", series_id: None, group_id: None,
|
||||
page_url: page.into(), rss_url: rss.into(), subscription: None,
|
||||
warnings: vec!["Keyword RSS can contain multiple series. Select a specific series, or manually preview with filters that restrict the feed to one series before creating a subscription."] });
|
||||
}
|
||||
let (id, group) = if let Some(id) = url.path().strip_prefix("/Home/Bangumi/") {
|
||||
(id.to_owned(), None)
|
||||
} else if url.path() == "/RSS/Bangumi" {
|
||||
(
|
||||
params
|
||||
.get("bangumiid")
|
||||
.cloned()
|
||||
.ok_or_else(|| Error::Invalid("missing bangumiId".into()))?,
|
||||
params.get("subgroupid").cloned(),
|
||||
)
|
||||
} else {
|
||||
return Err(Error::Invalid(
|
||||
"expected a Mikan series page or series RSS URL".into(),
|
||||
));
|
||||
};
|
||||
if group.as_ref().is_some_and(|id| !valid_id(id)) {
|
||||
return Err(Error::Invalid("invalid subtitle group ID".into()));
|
||||
}
|
||||
let detail = self.series(&id).await?;
|
||||
let rss_url = match &group {
|
||||
Some(group) => detail
|
||||
.groups
|
||||
.iter()
|
||||
.find(|g| &g.group_id == group)
|
||||
.map(|g| g.rss_url.clone())
|
||||
.ok_or_else(|| {
|
||||
Error::Invalid("subtitle group is not available for this series".into())
|
||||
})?,
|
||||
None => detail.rss_url,
|
||||
};
|
||||
Ok(Candidate {
|
||||
source_id: "mikan", scope: if group.is_some() { "group" } else { "series" },
|
||||
series_id: Some(id), group_id: group, page_url: detail.page_url, rss_url: rss_url.clone(),
|
||||
subscription: Some(SubscriptionInput { title: detail.title, rss_url, season: 1, episode_offset: 0,
|
||||
include: Vec::new(), exclude: Vec::new(), enabled: false }),
|
||||
warnings: vec!["Confirm season, episode offset and filters in preview before enabling; season 1 is a default, not inferred metadata."],
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn search(&self, query: &str) -> Result<SearchResult> {
|
||||
let query = query.trim();
|
||||
if query.is_empty() || query.chars().count() > 100 {
|
||||
return Err(Error::Invalid(
|
||||
"search query must contain 1–100 characters".into(),
|
||||
));
|
||||
}
|
||||
let key = format!("search:{query}");
|
||||
if let Some(Cached::Search(result)) = self.cached(&key).await {
|
||||
return Ok(result);
|
||||
}
|
||||
let _permit = self.requests.try_acquire().map_err(|_| Error::SourceBusy)?;
|
||||
let mut url = self.base.join("/Home/Search").expect("static path");
|
||||
url.query_pairs_mut().append_pair("searchstr", query);
|
||||
let html = self.html(url.clone()).await?;
|
||||
let unrecognized =
|
||||
|| Error::Upstream("Mikan search page is unrecognized or access is restricted".into());
|
||||
let search_form = html
|
||||
.select(&selector(
|
||||
"form[action='/Home/Search'] input[name='searchstr']",
|
||||
))
|
||||
.next()
|
||||
.is_some();
|
||||
let results = html.select(&selector(".an-ul")).next().is_some()
|
||||
|| html
|
||||
.select(&selector(".header2-text"))
|
||||
.any(|element| text(element).contains("搜索结果"));
|
||||
if !search_form || !results {
|
||||
return Err(unrecognized());
|
||||
}
|
||||
let mut seen = HashSet::new();
|
||||
let mut series = Vec::new();
|
||||
for link in html.select(&selector(".an-ul a[href]")) {
|
||||
let Some(url) = link
|
||||
.value()
|
||||
.attr("href")
|
||||
.and_then(|s| self.base.join(s).ok())
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
if url.origin() != self.base.origin() {
|
||||
continue;
|
||||
}
|
||||
let Some(id) = url
|
||||
.path()
|
||||
.strip_prefix("/Home/Bangumi/")
|
||||
.filter(|id| valid_id(id))
|
||||
else {
|
||||
continue;
|
||||
};
|
||||
let Some(name) = link.select(&selector(".an-text")).next() else {
|
||||
return Err(unrecognized());
|
||||
};
|
||||
let text = name.text().collect::<String>();
|
||||
let title = name.value().attr("title").unwrap_or(&text).trim();
|
||||
if title.is_empty() {
|
||||
return Err(unrecognized());
|
||||
}
|
||||
if !seen.insert(id.to_owned()) {
|
||||
continue;
|
||||
}
|
||||
series.push(SeriesSummary {
|
||||
series_id: id.into(),
|
||||
title: title.into(),
|
||||
page_url: url.to_string(),
|
||||
cover_url: link
|
||||
.select(&selector("[data-src], img[src]"))
|
||||
.next()
|
||||
.map(|_| format!("/api/sources/mikan/series/{id}/cover")),
|
||||
});
|
||||
}
|
||||
url.set_path("/RSS/Search");
|
||||
let result = SearchResult {
|
||||
source_id: "mikan",
|
||||
query: query.into(),
|
||||
series,
|
||||
keyword_rss_url: url.into(),
|
||||
};
|
||||
self.remember(key, Cached::Search(result.clone())).await;
|
||||
Ok(result)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,89 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use regex::Regex;
|
||||
use std::{collections::BTreeSet, sync::LazyLock};
|
||||
|
||||
static UNSUPPORTED: LazyLock<Regex> = LazyLock::new(|| {
|
||||
Regex::new(
|
||||
r"(?i)(合集|全集|\bbatch\b|\bcomplete\b|\b(?:OVA|SP|NCOP|NCED)(?:[0-9]+\b|\b)|(?:\s-\s*|\[|第|\bEP?\s*)[0-9]+\.[0-9]+|[0-9]+\s*[-~～]\s*[0-9]+)"
|
||||
).unwrap()
|
||||
});
|
||||
static SEASON_EPISODE: LazyLock<Regex> =
|
||||
LazyLock::new(|| Regex::new(r"(?i)\bS([0-9]{1,3})E([0-9]{1,4})(?:v[0-9]+)?(?:\b|_)").unwrap());
|
||||
static EPISODE: LazyLock<Vec<Regex>> = LazyLock::new(|| {
|
||||
[
|
||||
r"(?i)\[([0-9]{1,3})(?:v[0-9]+)?(?:\s*END)?\]",
|
||||
r"第\s*([0-9]{1,4})\s*[话話集]",
|
||||
r"(?i)\bEP?\s*([0-9]{1,4})(?:v[0-9]+)?\b",
|
||||
r"(?i)\s-\s*([0-9]{1,4})(?:v[0-9]+)?(?:\s|\[|$)",
|
||||
]
|
||||
.into_iter()
|
||||
.map(|r| Regex::new(r).unwrap())
|
||||
.collect()
|
||||
});
|
||||
|
||||
/// Conservative single-episode parsing. Ambiguity is visible, never guessed.
|
||||
pub fn episode(title: &str, season: i32, offset: i32) -> std::result::Result<i32, &'static str> {
|
||||
if UNSUPPORTED.is_match(title) {
|
||||
return Err("batch, special episode or fractional episode requires manual review");
|
||||
}
|
||||
let mut candidates = BTreeSet::new();
|
||||
for captures in SEASON_EPISODE.captures_iter(title) {
|
||||
if captures[1].parse::<i32>().ok() != Some(season) {
|
||||
return Err("title season differs from subscription season");
|
||||
}
|
||||
candidates.insert(captures[2].parse::<i32>().unwrap());
|
||||
}
|
||||
for pattern in EPISODE.iter() {
|
||||
for captures in pattern.captures_iter(title) {
|
||||
candidates.insert(captures[1].parse::<i32>().unwrap());
|
||||
}
|
||||
}
|
||||
if candidates.len() != 1 {
|
||||
return Err("episode is missing or ambiguous");
|
||||
}
|
||||
let value = candidates
|
||||
.first()
|
||||
.copied()
|
||||
.unwrap()
|
||||
.checked_add(offset)
|
||||
.ok_or("episode offset overflow")?;
|
||||
if !(1..=10000).contains(&value) {
|
||||
return Err("episode is outside supported range after offset");
|
||||
}
|
||||
Ok(value)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
#[test]
|
||||
fn single_episode_formats() {
|
||||
for title in [
|
||||
"[Group] Example [03][1080p]",
|
||||
"Example - 03v2 [1080p]",
|
||||
"Example 第03话",
|
||||
"Example S01E03",
|
||||
"Example EP03",
|
||||
"Example - 03 [1080p HEVC AAC2.0]",
|
||||
] {
|
||||
assert_eq!(episode(title, 1, 0), Ok(3), "{title}");
|
||||
}
|
||||
assert_eq!(episode("Example - 13 [1080p]", 2, -12), Ok(1));
|
||||
}
|
||||
#[test]
|
||||
fn uncertain_titles_are_not_downloaded() {
|
||||
for title in [
|
||||
"Example [1080p]",
|
||||
"Example [01-12]",
|
||||
"Example - 12.5",
|
||||
"Example SP01",
|
||||
"Example SP01 [03]",
|
||||
"Example [03] [04]",
|
||||
"Example S02E03",
|
||||
"Example S01E03E04",
|
||||
] {
|
||||
assert!(episode(title, 1, 0).is_err(), "{title}");
|
||||
}
|
||||
assert!(episode("Example [01]", 1, -1).is_err());
|
||||
}
|
||||
}
|
||||
+380
@@ -0,0 +1,380 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::{
|
||||
config::Config,
|
||||
domain::*,
|
||||
downloader::{self, Qbit},
|
||||
error::{Error, Result},
|
||||
feeds,
|
||||
media::{self, FileReport, MediaFile},
|
||||
storage::Store,
|
||||
};
|
||||
use serde::Serialize;
|
||||
use std::{
|
||||
collections::{HashMap, HashSet},
|
||||
sync::Arc,
|
||||
};
|
||||
use tokio::sync::{watch, Mutex};
|
||||
|
||||
pub struct Service {
|
||||
pub store: Store,
|
||||
pub gate: Mutex<()>,
|
||||
pub mikan: crate::mikan::Mikan,
|
||||
client: reqwest::Client,
|
||||
}
|
||||
|
||||
#[derive(Default, Serialize)]
|
||||
pub struct CycleReport {
|
||||
pub scanned: usize,
|
||||
pub queued: usize,
|
||||
pub errors: Vec<String>,
|
||||
pub downloader_configured: bool,
|
||||
}
|
||||
|
||||
impl Service {
|
||||
pub async fn new(store: Store, config: &Config) -> Result<Self> {
|
||||
let client = feeds::website_client(&config.website)?;
|
||||
let mikan = crate::mikan::Mikan::new(client.clone(), &config.mikan_base_url)?;
|
||||
store.bootstrap_downloader(config.qbit.as_ref()).await?;
|
||||
Ok(Self {
|
||||
store,
|
||||
gate: Mutex::new(()),
|
||||
mikan,
|
||||
client,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn downloader_configured(&self) -> Result<bool> {
|
||||
Ok(self.store.active_downloader_id().await?.is_some())
|
||||
}
|
||||
|
||||
pub async fn preview(&self, mut input: SubscriptionInput) -> Result<Vec<Release>> {
|
||||
input.validate()?;
|
||||
feeds::fetch(&self.client, &input).await
|
||||
}
|
||||
|
||||
pub async fn preview_subscription(&self, id: &str) -> Result<Vec<Release>> {
|
||||
let sub = self.store.subscription(id).await?;
|
||||
let mut releases = self.preview(sub.input.clone()).await?;
|
||||
let mut episodes = HashSet::new();
|
||||
// Feed order determines the candidate selected in v0.1; priority rules come later.
|
||||
for release in &mut releases {
|
||||
if release.decision == Decision::Ready {
|
||||
let duplicate = self
|
||||
.store
|
||||
.already_queued(id, sub.input.season, release)
|
||||
.await?;
|
||||
if duplicate || !episodes.insert(release.episode) {
|
||||
release.decision = Decision::AlreadyQueued;
|
||||
release.reason = if duplicate {
|
||||
"resource or episode already has a task"
|
||||
} else {
|
||||
"an earlier feed entry already represents this episode"
|
||||
}
|
||||
.into();
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(releases)
|
||||
}
|
||||
|
||||
async fn scan(&self, sub: &Subscription) -> Result<usize> {
|
||||
let result = async {
|
||||
let releases = self.preview_subscription(&sub.id).await?;
|
||||
self.store.record_scan(sub, &releases).await
|
||||
}
|
||||
.await;
|
||||
self.store
|
||||
.checked(
|
||||
&sub.id,
|
||||
result.as_ref().err().map(ToString::to_string).as_deref(),
|
||||
)
|
||||
.await?;
|
||||
result
|
||||
}
|
||||
|
||||
pub async fn scan_subscription(&self, id: &str) -> Result<usize> {
|
||||
let _guard = self.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
let sub = self.store.subscription(id).await?;
|
||||
if !sub.input.enabled {
|
||||
return Err(Error::Invalid(
|
||||
"enable the subscription before scanning; preview is available while disabled"
|
||||
.into(),
|
||||
));
|
||||
}
|
||||
self.scan(&sub).await
|
||||
}
|
||||
|
||||
pub async fn cycle(&self) -> Result<CycleReport> {
|
||||
let _guard = self.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
let mut report = CycleReport {
|
||||
downloader_configured: self.downloader_configured().await?,
|
||||
..Default::default()
|
||||
};
|
||||
let legacy = self.store.legacy_unassigned_count().await?;
|
||||
if legacy > 0 {
|
||||
report.errors.push(format!("{legacy} legacy jobs have no recorded downloader; restore their original instance and explicitly claim them"));
|
||||
}
|
||||
for sub in self.store.subscriptions().await? {
|
||||
if !sub.input.enabled {
|
||||
continue;
|
||||
}
|
||||
match self.scan(&sub).await {
|
||||
Ok(count) => {
|
||||
report.scanned += 1;
|
||||
report.queued += count;
|
||||
}
|
||||
Err(error) => report
|
||||
.errors
|
||||
.push(format!("subscription {}: {error}", sub.id)),
|
||||
}
|
||||
}
|
||||
match self.sync_downloads().await {
|
||||
Ok(errors) => report.errors.extend(errors),
|
||||
Err(error) => report.errors.push(error.to_string()),
|
||||
}
|
||||
Ok(report)
|
||||
}
|
||||
|
||||
async fn sync_downloads(&self) -> Result<Vec<String>> {
|
||||
let jobs = self.store.active_jobs().await?;
|
||||
let mut groups: std::collections::BTreeMap<String, Vec<DownloadJob>> = Default::default();
|
||||
for job in jobs {
|
||||
if let Some(id) = &job.downloader_id {
|
||||
groups.entry(id.clone()).or_default().push(job);
|
||||
}
|
||||
}
|
||||
let mut errors = Vec::new();
|
||||
for (id, jobs) in groups {
|
||||
let result = async {
|
||||
let qbit = Qbit::new(self.store.downloader_config(&id).await?)?;
|
||||
self.sync_instance(&qbit, jobs).await
|
||||
}
|
||||
.await;
|
||||
if let Err(error) = result {
|
||||
if matches!(error, Error::Database(_)) {
|
||||
return Err(error);
|
||||
}
|
||||
errors.push(format!("downloader {id}: {error}"));
|
||||
}
|
||||
}
|
||||
Ok(errors)
|
||||
}
|
||||
|
||||
async fn sync_instance(&self, qbit: &Qbit, jobs: Vec<DownloadJob>) -> Result<()> {
|
||||
qbit.login().await?;
|
||||
// Never submit when querying the remote state fails.
|
||||
let remote: HashMap<_, _> = qbit
|
||||
.torrents()
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|torrent| (torrent.hash.to_ascii_lowercase(), torrent))
|
||||
.collect();
|
||||
let mut added: HashMap<String, Option<String>> = HashMap::new();
|
||||
for mut job in jobs {
|
||||
let sub = self.store.subscription(&job.subscription_id).await;
|
||||
let enabled = match sub {
|
||||
Ok(sub) => sub.input.enabled,
|
||||
Err(Error::NotFound) => false,
|
||||
Err(error) => return Err(error),
|
||||
};
|
||||
let result: Result<()> = async {
|
||||
if job.info_hash.is_none() {
|
||||
if !enabled { return Ok(()); }
|
||||
let prepared = downloader::prepare(&self.client, &job.download_url).await?;
|
||||
self.store.prepare(&job.id, &prepared.hash, prepared.bytes.as_deref()).await?;
|
||||
job.info_hash = Some(prepared.hash);
|
||||
}
|
||||
let hash = job.info_hash.as_ref().expect("prepared hash");
|
||||
if let Some(torrent) = remote.get(hash) {
|
||||
if job.destination_id.is_some() && !job.download_save_path.as_deref().zip(torrent.save_path.as_deref())
|
||||
.is_some_and(|(expected, actual)| media::same_download_root(expected, actual)) {
|
||||
return Err(Error::Invalid("remote save path differs from the job snapshot; refusing to move or reuse this torrent".into()));
|
||||
}
|
||||
if matches!(torrent.state.as_str(), "error" | "missingFiles") {
|
||||
return Err(Error::Invalid("qBittorrent reports an error or missing files; inspect the downloader before retrying".into()));
|
||||
}
|
||||
let state = if torrent.progress >= 1.0 { "completed" }
|
||||
else if torrent.progress > 0.0 { "downloading" } else { "submitted" };
|
||||
self.store.state(&job.id, state).await?;
|
||||
} else if let Some(path) = added.get(hash) {
|
||||
if job.destination_id.is_some() && path != &job.download_save_path {
|
||||
return Err(Error::Invalid("this torrent was submitted to a different path; refusing to reuse it".into()));
|
||||
}
|
||||
self.store.state(&job.id, "submitted").await?;
|
||||
} else if matches!(job.state.as_str(), "submitted" | "downloading") {
|
||||
// Respect removal in qBittorrent; recreating a task needs an explicit retry.
|
||||
self.store.state(&job.id, "missing").await?;
|
||||
} else if enabled {
|
||||
let bytes = self.store.torrent_bytes(&job.id).await?;
|
||||
// Commit the intent before the external side effect.
|
||||
self.store.state(&job.id, "submitting").await?;
|
||||
qbit.add(&job.download_url, bytes, job.download_save_path.as_deref()).await?;
|
||||
added.insert(hash.clone(), job.download_save_path.clone());
|
||||
self.store.state(&job.id, "submitted").await?;
|
||||
}
|
||||
Ok(())
|
||||
}.await;
|
||||
if let Err(error) = result {
|
||||
if matches!(error, Error::Database(_)) {
|
||||
return Err(error);
|
||||
}
|
||||
if matches!(error, Error::Invalid(_)) {
|
||||
self.store.state(&job.id, "failed").await?;
|
||||
}
|
||||
self.store.failed(&job, &error.to_string()).await?;
|
||||
tracing::warn!(job_id = %job.id, error = %error, "download job failed");
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn discover_files(&self, job: &DownloadJob) -> Result<Vec<MediaFile>> {
|
||||
let hash = job
|
||||
.info_hash
|
||||
.as_ref()
|
||||
.ok_or_else(|| Error::Invalid("torrent metadata is not available yet".into()))?;
|
||||
let owner = job
|
||||
.downloader_id
|
||||
.as_ref()
|
||||
.ok_or_else(|| Error::Invalid("downloader ownership is unknown".into()))?;
|
||||
let qbit = Qbit::new(self.store.downloader_config(owner).await?)?;
|
||||
qbit.login().await?;
|
||||
let torrent = qbit
|
||||
.torrents()
|
||||
.await?
|
||||
.into_iter()
|
||||
.find(|torrent| torrent.hash.eq_ignore_ascii_case(hash))
|
||||
.ok_or_else(|| {
|
||||
Error::Upstream("torrent is not available in its original downloader".into())
|
||||
})?;
|
||||
if !job
|
||||
.download_save_path
|
||||
.as_deref()
|
||||
.zip(torrent.save_path.as_deref())
|
||||
.is_some_and(|(expected, actual)| media::same_download_root(expected, actual))
|
||||
{
|
||||
return Err(Error::Upstream(
|
||||
"remote save path does not match the frozen destination mapping".into(),
|
||||
));
|
||||
}
|
||||
let mut files = Vec::new();
|
||||
let mut paths = HashSet::new();
|
||||
for file in qbit
|
||||
.files(hash)
|
||||
.await?
|
||||
.into_iter()
|
||||
.filter(|file| file.priority != 0)
|
||||
{
|
||||
media::validate_relative(&file.name, false).map_err(|_| {
|
||||
Error::Upstream("downloader returned an unsafe relative file path".into())
|
||||
})?;
|
||||
if !paths.insert(file.name.clone()) {
|
||||
return Err(Error::Upstream(
|
||||
"duplicate path in downloader file list".into(),
|
||||
));
|
||||
}
|
||||
files.push(MediaFile {
|
||||
relative_path: file.name,
|
||||
expected_size: file.size,
|
||||
observed_size: None,
|
||||
status: "unchecked".into(),
|
||||
error: None,
|
||||
});
|
||||
}
|
||||
Ok(files)
|
||||
}
|
||||
|
||||
pub async fn check_job_files(&self, id: &str) -> Result<FileReport> {
|
||||
// Serialize manifest discovery/checks with task synchronization and configuration changes.
|
||||
let _guard = self.gate.try_lock().map_err(|_| Error::Busy)?;
|
||||
let job = self.store.job(id).await?;
|
||||
let Some(destination) = &job.destination_id else {
|
||||
self.store
|
||||
.save_file_check(
|
||||
id,
|
||||
"unchecked",
|
||||
Some("job has no recorded media destination"),
|
||||
&[],
|
||||
)
|
||||
.await?;
|
||||
return self.store.file_report(id).await;
|
||||
};
|
||||
let root = self.store.destination(destination).await?.root_path;
|
||||
let mut files = self.store.file_report(id).await?.files;
|
||||
if files.is_empty() {
|
||||
if job.info_hash.is_none() {
|
||||
self.store
|
||||
.save_file_check(
|
||||
id,
|
||||
"unchecked",
|
||||
Some("torrent metadata is not available yet"),
|
||||
&[],
|
||||
)
|
||||
.await?;
|
||||
return self.store.file_report(id).await;
|
||||
}
|
||||
match self.discover_files(&job).await {
|
||||
Ok(manifest) => files = manifest,
|
||||
Err(error @ Error::Database(_)) => return Err(error),
|
||||
Err(error) => {
|
||||
self.store
|
||||
.save_file_check(id, "unavailable", Some(&error.to_string()), &[])
|
||||
.await?;
|
||||
return self.store.file_report(id).await;
|
||||
}
|
||||
}
|
||||
if files.is_empty() {
|
||||
self.store
|
||||
.save_file_check(
|
||||
id,
|
||||
"unchecked",
|
||||
Some("no selected files or torrent metadata is still pending"),
|
||||
&[],
|
||||
)
|
||||
.await?;
|
||||
return self.store.file_report(id).await;
|
||||
}
|
||||
self.store.save_manifest(id, &files).await?;
|
||||
}
|
||||
let files = tokio::task::spawn_blocking(move || media::check_local(&root, files))
|
||||
.await
|
||||
.map_err(|_| Error::Upstream("local file check failed".into()))?;
|
||||
let status = if files.iter().any(|f| f.status == "unavailable") {
|
||||
"unavailable"
|
||||
} else if files.iter().any(|f| f.status == "absent") {
|
||||
"absent"
|
||||
} else {
|
||||
"present"
|
||||
};
|
||||
let error = files.iter().find_map(|file| file.error.as_deref());
|
||||
self.store
|
||||
.save_file_check(id, status, error, &files)
|
||||
.await?;
|
||||
self.store.file_report(id).await
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn worker(
|
||||
service: Arc<Service>,
|
||||
interval: std::time::Duration,
|
||||
mut shutdown: watch::Receiver<bool>,
|
||||
) {
|
||||
let mut timer = tokio::time::interval(interval);
|
||||
timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
|
||||
loop {
|
||||
tokio::select! {
|
||||
_ = shutdown.changed() => break,
|
||||
_ = timer.tick() => {
|
||||
// Cancellation can interrupt network I/O; persisted submission intent allows recovery.
|
||||
tokio::select! {
|
||||
_ = shutdown.changed() => break,
|
||||
result = service.cycle() => match result {
|
||||
Ok(report) => for error in report.errors { tracing::warn!(%error, "background cycle issue"); },
|
||||
Err(Error::Busy) => {},
|
||||
Err(error) => tracing::error!(%error, "background cycle failed"),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
+367
@@ -0,0 +1,367 @@
|
||||
// SPDX-License-Identifier: GPL-2.0-only
|
||||
use crate::{
|
||||
domain::*,
|
||||
error::{Error, Result},
|
||||
};
|
||||
use sqlx::{
|
||||
sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions},
|
||||
SqlitePool,
|
||||
};
|
||||
use std::{
|
||||
str::FromStr,
|
||||
time::{Duration, SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct Store {
|
||||
pub pool: SqlitePool,
|
||||
}
|
||||
|
||||
pub fn now() -> i64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_secs() as i64
|
||||
}
|
||||
|
||||
impl Store {
|
||||
pub async fn active_downloader_id(&self) -> Result<Option<String>> {
|
||||
Ok(
|
||||
sqlx::query_scalar(
|
||||
"SELECT downloader_id FROM downloader_selection WHERE singleton = 1",
|
||||
)
|
||||
.fetch_one(&self.pool)
|
||||
.await?,
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn downloaders(&self) -> Result<Vec<DownloaderView>> {
|
||||
let rows: Vec<(String, String, String, bool)> = sqlx::query_as(
|
||||
"SELECT d.id, d.name, d.config, COALESCE(s.downloader_id = d.id, 0) FROM downloader_instances d CROSS JOIN downloader_selection s ORDER BY d.rowid")
|
||||
.fetch_all(&self.pool).await?;
|
||||
rows.into_iter()
|
||||
.map(|(id, name, json, is_active)| {
|
||||
let config: crate::config::QbitConfig = serde_json::from_str(&json)?;
|
||||
Ok(DownloaderView {
|
||||
id,
|
||||
name,
|
||||
kind: DownloaderKind::Qbittorrent,
|
||||
url: config.url,
|
||||
username: config.username,
|
||||
save_path: config.save_path,
|
||||
password_configured: !config.password.is_empty(),
|
||||
is_active,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn create_downloader(&self, mut input: DownloaderInput) -> Result<DownloaderView> {
|
||||
input.validate()?;
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
let result = sqlx::query("INSERT INTO downloader_instances(id, name, kind, endpoint, config) VALUES (?, ?, 'qbittorrent', ?, ?) ON CONFLICT(endpoint) DO NOTHING")
|
||||
.bind(&id).bind(&input.name).bind(&input.connection.url)
|
||||
.bind(serde_json::to_string(&input.connection)?).execute(&self.pool).await?;
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(Error::Invalid(
|
||||
"this downloader endpoint is already registered".into(),
|
||||
));
|
||||
}
|
||||
self.downloaders()
|
||||
.await?
|
||||
.into_iter()
|
||||
.find(|d| d.id == id)
|
||||
.ok_or(Error::NotFound)
|
||||
}
|
||||
|
||||
pub(crate) async fn downloader_config(&self, id: &str) -> Result<crate::config::QbitConfig> {
|
||||
let json: String =
|
||||
sqlx::query_scalar("SELECT config FROM downloader_instances WHERE id = ?")
|
||||
.bind(id)
|
||||
.fetch_optional(&self.pool)
|
||||
.await?
|
||||
.ok_or(Error::NotFound)?;
|
||||
Ok(serde_json::from_str(&json)?)
|
||||
}
|
||||
|
||||
pub async fn activate_downloader(&self, id: &str) -> Result<()> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let exists: bool =
|
||||
sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM downloader_instances WHERE id = ?)")
|
||||
.bind(id)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
if !exists {
|
||||
return Err(Error::NotFound);
|
||||
}
|
||||
let destination: Option<String> = sqlx::query_scalar(
|
||||
"SELECT destination_id FROM destination_selection WHERE singleton = 1",
|
||||
)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
if let Some(destination) = destination {
|
||||
let mapped: bool = sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM destination_mappings WHERE destination_id = ? AND downloader_id = ?)")
|
||||
.bind(destination).bind(id).fetch_one(&mut *tx).await?;
|
||||
if !mapped {
|
||||
return Err(Error::Invalid("configure a mapping to the active destination before selecting this downloader".into()));
|
||||
}
|
||||
}
|
||||
sqlx::query("UPDATE downloader_selection SET downloader_id = ? WHERE singleton = 1")
|
||||
.bind(id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
// Bind only new, never-owned work. A switch never rewrites an existing owner.
|
||||
sqlx::query("UPDATE download_jobs SET downloader_id = ? WHERE downloader_id IS NULL AND legacy_unassigned = 0")
|
||||
.bind(id).execute(&mut *tx).await?;
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn claim_legacy_jobs(&self, id: &str) -> Result<u64> {
|
||||
self.downloader_config(id).await?;
|
||||
Ok(sqlx::query("UPDATE download_jobs SET downloader_id = ?, legacy_unassigned = 0 WHERE downloader_id IS NULL AND legacy_unassigned = 1")
|
||||
.bind(id).execute(&self.pool).await?.rows_affected())
|
||||
}
|
||||
|
||||
pub async fn legacy_unassigned_count(&self) -> Result<i64> {
|
||||
Ok(sqlx::query_scalar("SELECT COUNT(*) FROM download_jobs WHERE legacy_unassigned = 1 AND downloader_id IS NULL")
|
||||
.fetch_one(&self.pool).await?)
|
||||
}
|
||||
|
||||
pub(crate) async fn bootstrap_downloader(
|
||||
&self,
|
||||
config: Option<&crate::config::QbitConfig>,
|
||||
) -> Result<()> {
|
||||
let Some(config) = config else {
|
||||
return Ok(());
|
||||
};
|
||||
let mut config = config.clone();
|
||||
config.validate()?;
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM downloader_instances")
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
if count == 0 {
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
sqlx::query("INSERT INTO downloader_instances(id, name, kind, endpoint, config) VALUES (?, 'Initial qBittorrent', 'qbittorrent', ?, ?)")
|
||||
.bind(&id).bind(&config.url).bind(serde_json::to_string(&config)?).execute(&mut *tx).await?;
|
||||
sqlx::query("UPDATE downloader_selection SET downloader_id = ? WHERE singleton = 1")
|
||||
.bind(&id)
|
||||
.execute(&mut *tx)
|
||||
.await?;
|
||||
// Initial import uses the operator-supplied original QBIT_* connection.
|
||||
sqlx::query("UPDATE download_jobs SET downloader_id = ?, legacy_unassigned = 0 WHERE downloader_id IS NULL")
|
||||
.bind(&id).execute(&mut *tx).await?;
|
||||
}
|
||||
tx.commit().await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn open(url: &str) -> Result<Self> {
|
||||
let options = SqliteConnectOptions::from_str(url)?
|
||||
.create_if_missing(true)
|
||||
.foreign_keys(true)
|
||||
.journal_mode(SqliteJournalMode::Wal)
|
||||
.busy_timeout(Duration::from_secs(5));
|
||||
// A single connection also makes sqlite::memory: deterministic in tests.
|
||||
let pool = SqlitePoolOptions::new()
|
||||
.max_connections(1)
|
||||
.connect_with(options)
|
||||
.await?;
|
||||
sqlx::migrate!().run(&pool).await?;
|
||||
Ok(Self { pool })
|
||||
}
|
||||
|
||||
pub async fn subscriptions(&self) -> Result<Vec<Subscription>> {
|
||||
let rows = sqlx::query_as::<_, (String, String, Option<i64>, Option<String>)>(
|
||||
"SELECT id, spec, last_checked_at, last_error FROM subscriptions WHERE deleted = 0 ORDER BY rowid"
|
||||
).fetch_all(&self.pool).await?;
|
||||
rows.into_iter()
|
||||
.map(|(id, spec, last_checked_at, last_error)| {
|
||||
Ok(Subscription {
|
||||
id,
|
||||
input: serde_json::from_str(&spec)?,
|
||||
last_checked_at,
|
||||
last_error,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn subscription(&self, id: &str) -> Result<Subscription> {
|
||||
self.subscriptions()
|
||||
.await?
|
||||
.into_iter()
|
||||
.find(|s| s.id == id)
|
||||
.ok_or(Error::NotFound)
|
||||
}
|
||||
|
||||
pub async fn create(&self, mut input: SubscriptionInput) -> Result<Subscription> {
|
||||
input.validate()?;
|
||||
let id = uuid::Uuid::new_v4().to_string();
|
||||
sqlx::query("INSERT INTO subscriptions(id, spec) VALUES (?, ?)")
|
||||
.bind(&id)
|
||||
.bind(serde_json::to_string(&input)?)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
self.subscription(&id).await
|
||||
}
|
||||
|
||||
pub async fn update(&self, id: &str, mut input: SubscriptionInput) -> Result<Subscription> {
|
||||
input.validate()?;
|
||||
let result = sqlx::query("UPDATE subscriptions SET spec = ? WHERE id = ? AND deleted = 0")
|
||||
.bind(serde_json::to_string(&input)?)
|
||||
.bind(id)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(Error::NotFound);
|
||||
}
|
||||
self.subscription(id).await
|
||||
}
|
||||
|
||||
pub async fn delete(&self, id: &str) -> Result<()> {
|
||||
let result =
|
||||
sqlx::query("UPDATE subscriptions SET deleted = 1 WHERE id = ? AND deleted = 0")
|
||||
.bind(id)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(Error::NotFound);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn checked(&self, id: &str, error: Option<&str>) -> Result<()> {
|
||||
sqlx::query("UPDATE subscriptions SET last_checked_at = ?, last_error = ? WHERE id = ?")
|
||||
.bind(now())
|
||||
.bind(error)
|
||||
.bind(id)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn already_queued(&self, id: &str, season: i32, release: &Release) -> Result<bool> {
|
||||
let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM download_jobs WHERE subscription_id = ? AND (resource_key = ? OR (season = ? AND episode = ?))")
|
||||
.bind(id).bind(&release.key).bind(season).bind(release.episode).fetch_one(&self.pool).await?;
|
||||
Ok(count > 0)
|
||||
}
|
||||
|
||||
pub async fn record_scan(&self, sub: &Subscription, releases: &[Release]) -> Result<usize> {
|
||||
let mut tx = self.pool.begin().await?;
|
||||
let downloader_id: Option<String> = sqlx::query_scalar(
|
||||
"SELECT downloader_id FROM downloader_selection WHERE singleton = 1",
|
||||
)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
let destination_id: Option<String> = sqlx::query_scalar(
|
||||
"SELECT destination_id FROM destination_selection WHERE singleton = 1",
|
||||
)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
let mut mapping_id = None;
|
||||
let mut download_save_path = None;
|
||||
if let Some(destination) = &destination_id {
|
||||
let mapping: crate::media::Mapping = sqlx::query_as("SELECT * FROM destination_mappings WHERE destination_id = ? AND downloader_id = ? ORDER BY version DESC LIMIT 1")
|
||||
.bind(destination).bind(&downloader_id).fetch_optional(&mut *tx).await?
|
||||
.ok_or_else(|| Error::Invalid("active destination has no mapping for the selected downloader".into()))?;
|
||||
mapping_id = Some(mapping.id);
|
||||
download_save_path = Some(mapping.download_root);
|
||||
} else if let Some(downloader) = &downloader_id {
|
||||
let config: String =
|
||||
sqlx::query_scalar("SELECT config FROM downloader_instances WHERE id = ?")
|
||||
.bind(downloader)
|
||||
.fetch_one(&mut *tx)
|
||||
.await?;
|
||||
download_save_path =
|
||||
Some(serde_json::from_str::<crate::config::QbitConfig>(&config)?.save_path);
|
||||
}
|
||||
let mut queued = 0;
|
||||
for release in releases {
|
||||
sqlx::query("INSERT INTO releases(subscription_id, resource_key, payload) VALUES (?, ?, ?) ON CONFLICT(subscription_id, resource_key) DO UPDATE SET payload = excluded.payload")
|
||||
.bind(&sub.id).bind(&release.key).bind(serde_json::to_string(release)?).execute(&mut *tx).await?;
|
||||
if release.decision == Decision::Ready {
|
||||
if let (Some(episode), Some(url)) = (release.episode, &release.download_url) {
|
||||
let result = sqlx::query("INSERT OR IGNORE INTO download_jobs(id, subscription_id, resource_key, title, season, episode, download_url, downloader_id, destination_id, mapping_id, download_save_path) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
|
||||
.bind(uuid::Uuid::new_v4().to_string()).bind(&sub.id).bind(&release.key)
|
||||
.bind(&release.title).bind(sub.input.season).bind(episode).bind(url).bind(&downloader_id)
|
||||
.bind(&destination_id).bind(&mapping_id).bind(&download_save_path).execute(&mut *tx).await?;
|
||||
queued += result.rows_affected() as usize;
|
||||
}
|
||||
}
|
||||
}
|
||||
tx.commit().await?;
|
||||
Ok(queued)
|
||||
}
|
||||
|
||||
pub async fn releases(&self, id: &str) -> Result<Vec<Release>> {
|
||||
self.subscription(id).await?;
|
||||
let rows: Vec<String> = sqlx::query_scalar(
|
||||
"SELECT payload FROM releases WHERE subscription_id = ? ORDER BY rowid DESC LIMIT 500",
|
||||
)
|
||||
.bind(id)
|
||||
.fetch_all(&self.pool)
|
||||
.await?;
|
||||
rows.into_iter()
|
||||
.map(|s| serde_json::from_str(&s).map_err(Error::from))
|
||||
.collect()
|
||||
}
|
||||
|
||||
pub async fn jobs(&self) -> Result<Vec<DownloadJob>> {
|
||||
Ok(
|
||||
sqlx::query_as("SELECT * FROM download_jobs ORDER BY rowid DESC LIMIT 500")
|
||||
.fetch_all(&self.pool)
|
||||
.await?,
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn active_jobs(&self) -> Result<Vec<DownloadJob>> {
|
||||
Ok(sqlx::query_as("SELECT * FROM download_jobs WHERE state NOT IN ('completed', 'missing', 'failed') AND next_attempt_at <= ? ORDER BY rowid")
|
||||
.bind(now()).fetch_all(&self.pool).await?)
|
||||
}
|
||||
|
||||
pub async fn prepare(&self, id: &str, hash: &str, bytes: Option<&[u8]>) -> Result<()> {
|
||||
sqlx::query("UPDATE download_jobs SET info_hash = ?, torrent_bytes = ? WHERE id = ?")
|
||||
.bind(hash)
|
||||
.bind(bytes)
|
||||
.bind(id)
|
||||
.execute(&self.pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn torrent_bytes(&self, id: &str) -> Result<Option<Vec<u8>>> {
|
||||
Ok(
|
||||
sqlx::query_scalar("SELECT torrent_bytes FROM download_jobs WHERE id = ?")
|
||||
.bind(id)
|
||||
.fetch_one(&self.pool)
|
||||
.await?,
|
||||
)
|
||||
}
|
||||
|
||||
pub async fn state(&self, id: &str, state: &str) -> Result<()> {
|
||||
sqlx::query("UPDATE download_jobs SET state = ?, last_error = NULL, next_attempt_at = 0 WHERE id = ?")
|
||||
.bind(state).bind(id).execute(&self.pool).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn failed(&self, job: &DownloadJob, error: &str) -> Result<()> {
|
||||
// Keep 'submitting' on an ambiguous network failure, so reconciliation comes first.
|
||||
let delay = 30 * (1_i64 << job.attempts.clamp(0, 7));
|
||||
sqlx::query("UPDATE download_jobs SET attempts = attempts + 1, last_error = ?, next_attempt_at = ? WHERE id = ?")
|
||||
.bind(error).bind(now() + delay.min(3600)).bind(&job.id).execute(&self.pool).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn retry(&self, id: &str) -> Result<()> {
|
||||
let result = sqlx::query("UPDATE download_jobs SET state = 'queued', attempts = 0, next_attempt_at = 0, last_error = NULL WHERE id = ? AND state IN ('failed', 'missing')")
|
||||
.bind(id).execute(&self.pool).await?;
|
||||
if result.rows_affected() == 0 {
|
||||
return Err(Error::Invalid(
|
||||
"only failed or missing jobs can be retried".into(),
|
||||
));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user