Compare commits

...

1 Commits

Author SHA1 Message Date
Connor Johnstone
93392db27c Formatting 2026-03-18 15:37:10 -04:00
8 changed files with 250 additions and 161 deletions

View File

@@ -1,5 +1,5 @@
use actix_cors::Cors; use actix_cors::Cors;
use actix_web::{web, App, HttpServer}; use actix_web::{App, HttpServer, web};
use clap::Parser; use clap::Parser;
use tracing_actix_web::TracingLogger; use tracing_actix_web::TracingLogger;
use tracing_subscriber::EnvFilter; use tracing_subscriber::EnvFilter;
@@ -64,7 +64,7 @@ async fn main() -> anyhow::Result<()> {
mb_client, mb_client,
search, search,
config: std::sync::Arc::new(tokio::sync::RwLock::new(config)), config: std::sync::Arc::new(tokio::sync::RwLock::new(config)),
config_path: config_path, config_path,
tasks: TaskManager::new(), tasks: TaskManager::new(),
}); });

View File

@@ -1,4 +1,4 @@
use actix_web::{web, HttpResponse}; use actix_web::{HttpResponse, web};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use shanty_db::entities::wanted_item::WantedStatus; use shanty_db::entities::wanted_item::WantedStatus;
@@ -15,7 +15,9 @@ pub struct PaginationParams {
#[serde(default)] #[serde(default)]
offset: u64, offset: u64,
} }
fn default_limit() -> u64 { 50 } fn default_limit() -> u64 {
50
}
#[derive(Deserialize)] #[derive(Deserialize)]
pub struct AddAlbumRequest { pub struct AddAlbumRequest {
@@ -40,10 +42,7 @@ pub fn configure(cfg: &mut web::ServiceConfig) {
.route(web::get().to(list_albums)) .route(web::get().to(list_albums))
.route(web::post().to(add_album)), .route(web::post().to(add_album)),
) )
.service( .service(web::resource("/albums/{mbid}").route(web::get().to(get_album)));
web::resource("/albums/{mbid}")
.route(web::get().to(get_album)),
);
} }
async fn list_albums( async fn list_albums(
@@ -69,7 +68,8 @@ async fn get_album(
Err(_) => { Err(_) => {
// Probably a release-group MBID. Browse releases for this group. // Probably a release-group MBID. Browse releases for this group.
let release_mbid = resolve_release_from_group(&state, &mbid).await?; let release_mbid = resolve_release_from_group(&state, &mbid).await?;
state.mb_client state
.mb_client
.get_release_tracks(&release_mbid) .get_release_tracks(&release_mbid)
.await .await
.map_err(|e| ApiError::Internal(format!("MusicBrainz error: {e}")))? .map_err(|e| ApiError::Internal(format!("MusicBrainz error: {e}")))?
@@ -112,7 +112,7 @@ async fn get_album(
/// Given a release-group MBID, find the first release MBID via the MB API. /// Given a release-group MBID, find the first release MBID via the MB API.
async fn resolve_release_from_group( async fn resolve_release_from_group(
state: &web::Data<AppState>, _state: &web::Data<AppState>,
release_group_mbid: &str, release_group_mbid: &str,
) -> Result<String, ApiError> { ) -> Result<String, ApiError> {
// Use the MB client's get_json (it's private, so we go through search) // Use the MB client's get_json (it's private, so we go through search)
@@ -152,7 +152,11 @@ async fn resolve_release_from_group(
.and_then(|r| r.get("id")) .and_then(|r| r.get("id"))
.and_then(|id| id.as_str()) .and_then(|id| id.as_str())
.map(String::from) .map(String::from)
.ok_or_else(|| ApiError::NotFound(format!("no releases found for release group {release_group_mbid}"))) .ok_or_else(|| {
ApiError::NotFound(format!(
"no releases found for release group {release_group_mbid}"
))
})
} }
async fn add_album( async fn add_album(

View File

@@ -1,4 +1,4 @@
use actix_web::{web, HttpResponse}; use actix_web::{HttpResponse, web};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use shanty_db::entities::wanted_item::WantedStatus; use shanty_db::entities::wanted_item::WantedStatus;
@@ -16,7 +16,9 @@ pub struct PaginationParams {
#[serde(default)] #[serde(default)]
offset: u64, offset: u64,
} }
fn default_limit() -> u64 { 50 } fn default_limit() -> u64 {
50
}
#[derive(Deserialize)] #[derive(Deserialize)]
pub struct AddArtistRequest { pub struct AddArtistRequest {
@@ -67,10 +69,7 @@ pub fn configure(cfg: &mut web::ServiceConfig) {
.route(web::get().to(list_artists)) .route(web::get().to(list_artists))
.route(web::post().to(add_artist)), .route(web::post().to(add_artist)),
) )
.service( .service(web::resource("/artists/{id}/full").route(web::get().to(get_artist_full)))
web::resource("/artists/{id}/full")
.route(web::get().to(get_artist_full)),
)
.service( .service(
web::resource("/artists/{id}") web::resource("/artists/{id}")
.route(web::get().to(get_artist)) .route(web::get().to(get_artist))
@@ -94,20 +93,23 @@ async fn list_artists(
// Check if we have cached artist-level totals from a prior detail page load // Check if we have cached artist-level totals from a prior detail page load
let cache_key = format!("artist_totals:{}", a.id); let cache_key = format!("artist_totals:{}", a.id);
let cached_totals: Option<(u32, u32, u32)> = if let Ok(Some(json)) = let cached_totals: Option<(u32, u32, u32)> =
queries::cache::get(state.db.conn(), &cache_key).await if let Ok(Some(json)) = queries::cache::get(state.db.conn(), &cache_key).await {
{
serde_json::from_str(&json).ok() serde_json::from_str(&json).ok()
} else { } else {
None None
}; };
let (total_watched, total_owned, total_items) = if let Some((avail, watched, owned)) = cached_totals { let (total_watched, total_owned, total_items) =
if let Some((avail, watched, owned)) = cached_totals {
(watched as usize, owned as usize, avail as usize) (watched as usize, owned as usize, avail as usize)
} else { } else {
// Fall back to wanted item counts // Fall back to wanted item counts
let total_items = artist_wanted.len(); let total_items = artist_wanted.len();
let total_owned = artist_wanted.iter().filter(|w| w.status == WantedStatus::Owned).count(); let total_owned = artist_wanted
.iter()
.filter(|w| w.status == WantedStatus::Owned)
.count();
(total_items, total_owned, total_items) (total_items, total_owned, total_items)
}; };
@@ -137,7 +139,9 @@ async fn get_artist(
"albums": albums, "albums": albums,
}))) })))
} else { } else {
Err(ApiError::BadRequest("use /artists/{id}/full for MBID lookups".into())) Err(ApiError::BadRequest(
"use /artists/{id}/full for MBID lookups".into(),
))
} }
} }
@@ -153,17 +157,24 @@ async fn get_cached_album_tracks(
let cache_key = format!("artist_rg_tracks:{rg_id}"); let cache_key = format!("artist_rg_tracks:{rg_id}");
// Check cache first // Check cache first
if let Some(json) = queries::cache::get(state.db.conn(), &cache_key).await if let Some(json) = queries::cache::get(state.db.conn(), &cache_key)
.await
.map_err(|e| ApiError::Internal(e.to_string()))? .map_err(|e| ApiError::Internal(e.to_string()))?
&& let Ok(cached) = serde_json::from_str::<CachedAlbumTracks>(&json)
{ {
if let Ok(cached) = serde_json::from_str::<CachedAlbumTracks>(&json) {
// Extend TTL if artist is now watched (upgrades 7-day browse cache to permanent) // Extend TTL if artist is now watched (upgrades 7-day browse cache to permanent)
if extend_ttl { if extend_ttl {
let _ = queries::cache::set(state.db.conn(), &cache_key, "musicbrainz", &json, ttl_seconds).await; let _ = queries::cache::set(
state.db.conn(),
&cache_key,
"musicbrainz",
&json,
ttl_seconds,
)
.await;
} }
return Ok(cached); return Ok(cached);
} }
}
// Not cached — resolve release MBID and fetch tracks // Not cached — resolve release MBID and fetch tracks
let release_mbid = if let Some(rid) = first_release_id { let release_mbid = if let Some(rid) = first_release_id {
@@ -173,7 +184,8 @@ async fn get_cached_album_tracks(
resolve_release_from_group(rg_id).await? resolve_release_from_group(rg_id).await?
}; };
let mb_tracks = state.mb_client let mb_tracks = state
.mb_client
.get_release_tracks(&release_mbid) .get_release_tracks(&release_mbid)
.await .await
.map_err(|e| ApiError::Internal(format!("MB error for release {release_mbid}: {e}")))?; .map_err(|e| ApiError::Internal(format!("MB error for release {release_mbid}: {e}")))?;
@@ -190,9 +202,15 @@ async fn get_cached_album_tracks(
}; };
// Cache with caller-specified TTL // Cache with caller-specified TTL
let json = serde_json::to_string(&cached) let json = serde_json::to_string(&cached).map_err(|e| ApiError::Internal(e.to_string()))?;
.map_err(|e| ApiError::Internal(e.to_string()))?; let _ = queries::cache::set(
let _ = queries::cache::set(state.db.conn(), &cache_key, "musicbrainz", &json, ttl_seconds).await; state.db.conn(),
&cache_key,
"musicbrainz",
&json,
ttl_seconds,
)
.await;
Ok(cached) Ok(cached)
} }
@@ -258,10 +276,14 @@ pub async fn enrich_artist(
let mbid = match &artist.musicbrainz_id { let mbid = match &artist.musicbrainz_id {
Some(m) => m.clone(), Some(m) => m.clone(),
None => { None => {
let results = state.search.search_artist(&artist.name, 1).await let results = state
.search
.search_artist(&artist.name, 1)
.await
.map_err(|e| ApiError::Internal(e.to_string()))?; .map_err(|e| ApiError::Internal(e.to_string()))?;
results.into_iter().next().map(|a| a.id) results.into_iter().next().map(|a| a.id).ok_or_else(|| {
.ok_or_else(|| ApiError::NotFound(format!("no MBID for artist '{}'", artist.name)))? ApiError::NotFound(format!("no MBID for artist '{}'", artist.name))
})?
} }
}; };
(artist, Some(local_id), mbid) (artist, Some(local_id), mbid)
@@ -272,7 +294,8 @@ pub async fn enrich_artist(
let local = { let local = {
// Check if any local artist has this MBID // Check if any local artist has this MBID
let all = queries::artists::list(state.db.conn(), 1000, 0).await?; let all = queries::artists::list(state.db.conn(), 1000, 0).await?;
all.into_iter().find(|a| a.musicbrainz_id.as_deref() == Some(&mbid)) all.into_iter()
.find(|a| a.musicbrainz_id.as_deref() == Some(&mbid))
}; };
if let Some(a) = local { if let Some(a) = local {
@@ -280,7 +303,8 @@ pub async fn enrich_artist(
(a, Some(local_id), mbid) (a, Some(local_id), mbid)
} else { } else {
// Look up artist name from MusicBrainz by MBID — don't create a local record // Look up artist name from MusicBrainz by MBID — don't create a local record
let (name, _disambiguation) = state.mb_client let (name, _disambiguation) = state
.mb_client
.get_artist_by_mbid(&mbid) .get_artist_by_mbid(&mbid)
.await .await
.map_err(|e| ApiError::NotFound(format!("artist MBID {mbid} not found: {e}")))?; .map_err(|e| ApiError::NotFound(format!("artist MBID {mbid} not found: {e}")))?;
@@ -299,7 +323,10 @@ pub async fn enrich_artist(
}; };
// Fetch release groups and filter by allowed secondary types // Fetch release groups and filter by allowed secondary types
let all_release_groups = state.search.get_release_groups(&mbid).await let all_release_groups = state
.search
.get_release_groups(&mbid)
.await
.map_err(|e| ApiError::Internal(e.to_string()))?; .map_err(|e| ApiError::Internal(e.to_string()))?;
let allowed = state.config.read().await.allowed_secondary_types.clone(); let allowed = state.config.read().await.allowed_secondary_types.clone();
let release_groups: Vec<_> = all_release_groups let release_groups: Vec<_> = all_release_groups
@@ -369,15 +396,21 @@ pub async fn enrich_artist(
// If artist has any watched items, cache permanently (10 years); // If artist has any watched items, cache permanently (10 years);
// otherwise cache for 7 days (just browsing) // otherwise cache for 7 days (just browsing)
let is_watched = !artist_wanted.is_empty(); let is_watched = !artist_wanted.is_empty();
let cache_ttl = if is_watched { 10 * 365 * 86400 } else { 7 * 86400 }; let cache_ttl = if is_watched {
10 * 365 * 86400
} else {
7 * 86400
};
let cached = match get_cached_album_tracks( let cached = match get_cached_album_tracks(
&state, state,
&rg.id, &rg.id,
rg.first_release_id.as_deref(), rg.first_release_id.as_deref(),
cache_ttl, cache_ttl,
is_watched, is_watched,
).await { )
.await
{
Ok(c) => c, Ok(c) => c,
Err(e) => { Err(e) => {
tracing::warn!(rg_id = %rg.id, title = %rg.title, error = %e, "failed to fetch tracks"); tracing::warn!(rg_id = %rg.id, title = %rg.title, error = %e, "failed to fetch tracks");
@@ -435,7 +468,10 @@ pub async fn enrich_artist(
.find(|a| a.name.to_lowercase() == rg.title.to_lowercase()); .find(|a| a.name.to_lowercase() == rg.title.to_lowercase());
let local_album_id = local.map(|a| a.id); let local_album_id = local.map(|a| a.id);
let local_tracks = if let Some(aid) = local_album_id { let local_tracks = if let Some(aid) = local_album_id {
queries::tracks::get_by_album(state.db.conn(), aid).await.unwrap_or_default().len() as u32 queries::tracks::get_by_album(state.db.conn(), aid)
.await
.unwrap_or_default()
.len() as u32
} else { } else {
0 0
}; };
@@ -468,9 +504,14 @@ pub async fn enrich_artist(
// Sort: owned first, then partial, then wanted, then unwatched; within each by date // Sort: owned first, then partial, then wanted, then unwatched; within each by date
albums.sort_by(|a, b| { albums.sort_by(|a, b| {
let order = |s: &str| match s { let order = |s: &str| match s {
"owned" => 0, "partial" => 1, "wanted" => 2, _ => 3, "owned" => 0,
"partial" => 1,
"wanted" => 2,
_ => 3,
}; };
order(&a.status).cmp(&order(&b.status)).then_with(|| a.date.cmp(&b.date)) order(&a.status)
.cmp(&order(&b.status))
.then_with(|| a.date.cmp(&b.date))
}); });
// Deduplicated artist-level totals // Deduplicated artist-level totals
@@ -478,7 +519,10 @@ pub async fn enrich_artist(
let total_artist_watched = seen_watched.len() as u32; let total_artist_watched = seen_watched.len() as u32;
let total_artist_owned = seen_owned.len() as u32; let total_artist_owned = seen_owned.len() as u32;
let artist_status = if total_artist_owned > 0 && total_artist_owned >= total_available_tracks && total_available_tracks > 0 { let artist_status = if total_artist_owned > 0
&& total_artist_owned >= total_available_tracks
&& total_available_tracks > 0
{
"owned" "owned"
} else if total_artist_watched > 0 { } else if total_artist_watched > 0 {
"partial" "partial"
@@ -487,16 +531,25 @@ pub async fn enrich_artist(
}; };
// Cache artist-level totals for the library listing page // Cache artist-level totals for the library listing page
if !skip_track_fetch { if !skip_track_fetch && let Some(local_id) = id {
if let Some(local_id) = id {
let cache_key = format!("artist_totals:{local_id}"); let cache_key = format!("artist_totals:{local_id}");
let totals = serde_json::json!([total_available_tracks, total_artist_watched, total_artist_owned]); let totals = serde_json::json!([
total_available_tracks,
total_artist_watched,
total_artist_owned
]);
let _ = queries::cache::set( let _ = queries::cache::set(
state.db.conn(), &cache_key, "computed", state.db.conn(),
&cache_key,
"computed",
&totals.to_string(), &totals.to_string(),
if artist_wanted.is_empty() { 7 * 86400 } else { 10 * 365 * 86400 }, if artist_wanted.is_empty() {
).await; 7 * 86400
} } else {
10 * 365 * 86400
},
)
.await;
} }
Ok(serde_json::json!({ Ok(serde_json::json!({
@@ -515,10 +568,7 @@ pub async fn enrich_all_watched_artists(state: &AppState) -> Result<u32, ApiErro
let all_wanted = queries::wanted::list(state.db.conn(), None).await?; let all_wanted = queries::wanted::list(state.db.conn(), None).await?;
// Collect unique artist IDs that have any wanted items // Collect unique artist IDs that have any wanted items
let mut artist_ids: Vec<i32> = all_wanted let mut artist_ids: Vec<i32> = all_wanted.iter().filter_map(|w| w.artist_id).collect();
.iter()
.filter_map(|w| w.artist_id)
.collect();
artist_ids.sort(); artist_ids.sort();
artist_ids.dedup(); artist_ids.dedup();

View File

@@ -1,4 +1,4 @@
use actix_web::{web, HttpResponse}; use actix_web::{HttpResponse, web};
use serde::Deserialize; use serde::Deserialize;
use shanty_db::entities::download_queue::DownloadStatus; use shanty_db::entities::download_queue::DownloadStatus;
@@ -19,30 +19,12 @@ pub struct EnqueueRequest {
} }
pub fn configure(cfg: &mut web::ServiceConfig) { pub fn configure(cfg: &mut web::ServiceConfig) {
cfg.service( cfg.service(web::resource("/downloads/queue").route(web::get().to(list_queue)))
web::resource("/downloads/queue") .service(web::resource("/downloads").route(web::post().to(enqueue_download)))
.route(web::get().to(list_queue)), .service(web::resource("/downloads/sync").route(web::post().to(sync_downloads)))
) .service(web::resource("/downloads/process").route(web::post().to(trigger_process)))
.service( .service(web::resource("/downloads/retry/{id}").route(web::post().to(retry_download)))
web::resource("/downloads") .service(web::resource("/downloads/{id}").route(web::delete().to(cancel_download)));
.route(web::post().to(enqueue_download)),
)
.service(
web::resource("/downloads/sync")
.route(web::post().to(sync_downloads)),
)
.service(
web::resource("/downloads/process")
.route(web::post().to(trigger_process)),
)
.service(
web::resource("/downloads/retry/{id}")
.route(web::post().to(retry_download)),
)
.service(
web::resource("/downloads/{id}")
.route(web::delete().to(cancel_download)),
);
} }
async fn list_queue( async fn list_queue(
@@ -69,9 +51,7 @@ async fn enqueue_download(
Ok(HttpResponse::Ok().json(item)) Ok(HttpResponse::Ok().json(item))
} }
async fn sync_downloads( async fn sync_downloads(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let stats = shanty_dl::sync_wanted_to_queue(state.db.conn(), false).await?; let stats = shanty_dl::sync_wanted_to_queue(state.db.conn(), false).await?;
Ok(HttpResponse::Ok().json(serde_json::json!({ Ok(HttpResponse::Ok().json(serde_json::json!({
"found": stats.found, "found": stats.found,
@@ -80,9 +60,7 @@ async fn sync_downloads(
}))) })))
} }
async fn trigger_process( async fn trigger_process(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let task_id = state.tasks.register("download"); let task_id = state.tasks.register("download");
let state = state.clone(); let state = state.clone();
let tid = task_id.clone(); let tid = task_id.clone();
@@ -90,9 +68,21 @@ async fn trigger_process(
tokio::spawn(async move { tokio::spawn(async move {
let cfg = state.config.read().await.clone(); let cfg = state.config.read().await.clone();
let cookies = cfg.download.cookies_path.clone(); let cookies = cfg.download.cookies_path.clone();
let format: shanty_dl::AudioFormat = cfg.download.format.parse().unwrap_or(shanty_dl::AudioFormat::Opus); let format: shanty_dl::AudioFormat = cfg
let source: shanty_dl::SearchSource = cfg.download.search_source.parse().unwrap_or(shanty_dl::SearchSource::YouTubeMusic); .download
let rate = if cookies.is_some() { cfg.download.rate_limit_auth } else { cfg.download.rate_limit }; .format
.parse()
.unwrap_or(shanty_dl::AudioFormat::Opus);
let source: shanty_dl::SearchSource = cfg
.download
.search_source
.parse()
.unwrap_or(shanty_dl::SearchSource::YouTubeMusic);
let rate = if cookies.is_some() {
cfg.download.rate_limit_auth
} else {
cfg.download.rate_limit
};
let backend = shanty_dl::YtDlpBackend::new(rate, source, cookies.clone()); let backend = shanty_dl::YtDlpBackend::new(rate, source, cookies.clone());
let backend_config = shanty_dl::BackendConfig { let backend_config = shanty_dl::BackendConfig {
output_dir: cfg.download_path.clone(), output_dir: cfg.download_path.clone(),
@@ -103,10 +93,20 @@ async fn trigger_process(
let task_state = state.clone(); let task_state = state.clone();
let progress_tid = tid.clone(); let progress_tid = tid.clone();
let on_progress: shanty_dl::ProgressFn = Box::new(move |current, total, msg| { let on_progress: shanty_dl::ProgressFn = Box::new(move |current, total, msg| {
task_state.tasks.update_progress(&progress_tid, current, total, msg); task_state
.tasks
.update_progress(&progress_tid, current, total, msg);
}); });
match shanty_dl::run_queue_with_progress(state.db.conn(), &backend, &backend_config, false, Some(on_progress)).await { match shanty_dl::run_queue_with_progress(
state.db.conn(),
&backend,
&backend_config,
false,
Some(on_progress),
)
.await
{
Ok(stats) => { Ok(stats) => {
state.tasks.complete(&tid, format!("{stats}")); state.tasks.complete(&tid, format!("{stats}"));
// Refresh artist data in background // Refresh artist data in background

View File

@@ -1,4 +1,4 @@
use actix_web::{web, HttpResponse}; use actix_web::{HttpResponse, web};
use serde::Deserialize; use serde::Deserialize;
use shanty_search::SearchProvider; use shanty_search::SearchProvider;
@@ -21,7 +21,9 @@ pub struct AlbumTrackSearchParams {
limit: u32, limit: u32,
} }
fn default_limit() -> u32 { 25 } fn default_limit() -> u32 {
25
}
pub fn configure(cfg: &mut web::ServiceConfig) { pub fn configure(cfg: &mut web::ServiceConfig) {
cfg.service(web::resource("/search/artist").route(web::get().to(search_artist))) cfg.service(web::resource("/search/artist").route(web::get().to(search_artist)))

View File

@@ -1,4 +1,4 @@
use actix_web::{web, HttpResponse}; use actix_web::{HttpResponse, web};
use serde::Deserialize; use serde::Deserialize;
use shanty_db::entities::download_queue::DownloadStatus; use shanty_db::entities::download_queue::DownloadStatus;
@@ -25,13 +25,14 @@ pub fn configure(cfg: &mut web::ServiceConfig) {
); );
} }
async fn get_status( async fn get_status(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let summary = shanty_watch::library_summary(state.db.conn()).await?; let summary = shanty_watch::library_summary(state.db.conn()).await?;
let pending_items = queries::downloads::list(state.db.conn(), Some(DownloadStatus::Pending)).await?; let pending_items =
let downloading_items = queries::downloads::list(state.db.conn(), Some(DownloadStatus::Downloading)).await?; queries::downloads::list(state.db.conn(), Some(DownloadStatus::Pending)).await?;
let failed_items = queries::downloads::list(state.db.conn(), Some(DownloadStatus::Failed)).await?; let downloading_items =
queries::downloads::list(state.db.conn(), Some(DownloadStatus::Downloading)).await?;
let failed_items =
queries::downloads::list(state.db.conn(), Some(DownloadStatus::Failed)).await?;
let tasks = state.tasks.list(); let tasks = state.tasks.list();
let mut queue_items = Vec::new(); let mut queue_items = Vec::new();
@@ -57,16 +58,16 @@ async fn get_status(
}))) })))
} }
async fn trigger_index( async fn trigger_index(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let task_id = state.tasks.register("index"); let task_id = state.tasks.register("index");
let state = state.clone(); let state = state.clone();
let tid = task_id.clone(); let tid = task_id.clone();
tokio::spawn(async move { tokio::spawn(async move {
let cfg = state.config.read().await.clone(); let cfg = state.config.read().await.clone();
state.tasks.update_progress(&tid, 0, 0, "Scanning library..."); state
.tasks
.update_progress(&tid, 0, 0, "Scanning library...");
let scan_config = shanty_index::ScanConfig { let scan_config = shanty_index::ScanConfig {
root: cfg.library_path.clone(), root: cfg.library_path.clone(),
dry_run: false, dry_run: false,
@@ -81,16 +82,16 @@ async fn trigger_index(
Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id }))) Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id })))
} }
async fn trigger_tag( async fn trigger_tag(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let task_id = state.tasks.register("tag"); let task_id = state.tasks.register("tag");
let state = state.clone(); let state = state.clone();
let tid = task_id.clone(); let tid = task_id.clone();
tokio::spawn(async move { tokio::spawn(async move {
let cfg = state.config.read().await.clone(); let cfg = state.config.read().await.clone();
state.tasks.update_progress(&tid, 0, 0, "Preparing tagger..."); state
.tasks
.update_progress(&tid, 0, 0, "Preparing tagger...");
let mb = match shanty_tag::MusicBrainzClient::new() { let mb = match shanty_tag::MusicBrainzClient::new() {
Ok(c) => c, Ok(c) => c,
Err(e) => { Err(e) => {
@@ -113,16 +114,16 @@ async fn trigger_tag(
Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id }))) Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id })))
} }
async fn trigger_organize( async fn trigger_organize(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let task_id = state.tasks.register("organize"); let task_id = state.tasks.register("organize");
let state = state.clone(); let state = state.clone();
let tid = task_id.clone(); let tid = task_id.clone();
tokio::spawn(async move { tokio::spawn(async move {
let cfg = state.config.read().await.clone(); let cfg = state.config.read().await.clone();
state.tasks.update_progress(&tid, 0, 0, "Organizing files..."); state
.tasks
.update_progress(&tid, 0, 0, "Organizing files...");
let org_config = shanty_org::OrgConfig { let org_config = shanty_org::OrgConfig {
target_dir: cfg.library_path.clone(), target_dir: cfg.library_path.clone(),
format: cfg.organization_format.clone(), format: cfg.organization_format.clone(),
@@ -149,9 +150,7 @@ async fn trigger_organize(
Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id }))) Ok(HttpResponse::Accepted().json(serde_json::json!({ "task_id": task_id })))
} }
async fn trigger_pipeline( async fn trigger_pipeline(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let sync_id = state.tasks.register_pending("sync"); let sync_id = state.tasks.register_pending("sync");
let download_id = state.tasks.register_pending("download"); let download_id = state.tasks.register_pending("download");
let index_id = state.tasks.register_pending("index"); let index_id = state.tasks.register_pending("index");
@@ -175,7 +174,9 @@ async fn trigger_pipeline(
// Step 1: Sync // Step 1: Sync
state.tasks.start(&sync_id); state.tasks.start(&sync_id);
state.tasks.update_progress(&sync_id, 0, 0, "Syncing watchlist to download queue..."); state
.tasks
.update_progress(&sync_id, 0, 0, "Syncing watchlist to download queue...");
match shanty_dl::sync_wanted_to_queue(state.db.conn(), false).await { match shanty_dl::sync_wanted_to_queue(state.db.conn(), false).await {
Ok(stats) => state.tasks.complete(&sync_id, format!("{stats}")), Ok(stats) => state.tasks.complete(&sync_id, format!("{stats}")),
Err(e) => state.tasks.fail(&sync_id, e.to_string()), Err(e) => state.tasks.fail(&sync_id, e.to_string()),
@@ -184,9 +185,21 @@ async fn trigger_pipeline(
// Step 2: Download // Step 2: Download
state.tasks.start(&download_id); state.tasks.start(&download_id);
let cookies = cfg.download.cookies_path.clone(); let cookies = cfg.download.cookies_path.clone();
let format: shanty_dl::AudioFormat = cfg.download.format.parse().unwrap_or(shanty_dl::AudioFormat::Opus); let format: shanty_dl::AudioFormat = cfg
let source: shanty_dl::SearchSource = cfg.download.search_source.parse().unwrap_or(shanty_dl::SearchSource::YouTubeMusic); .download
let rate = if cookies.is_some() { cfg.download.rate_limit_auth } else { cfg.download.rate_limit }; .format
.parse()
.unwrap_or(shanty_dl::AudioFormat::Opus);
let source: shanty_dl::SearchSource = cfg
.download
.search_source
.parse()
.unwrap_or(shanty_dl::SearchSource::YouTubeMusic);
let rate = if cookies.is_some() {
cfg.download.rate_limit_auth
} else {
cfg.download.rate_limit
};
let backend = shanty_dl::YtDlpBackend::new(rate, source, cookies.clone()); let backend = shanty_dl::YtDlpBackend::new(rate, source, cookies.clone());
let backend_config = shanty_dl::BackendConfig { let backend_config = shanty_dl::BackendConfig {
output_dir: cfg.download_path.clone(), output_dir: cfg.download_path.clone(),
@@ -196,9 +209,19 @@ async fn trigger_pipeline(
let task_state = state.clone(); let task_state = state.clone();
let progress_tid = download_id.clone(); let progress_tid = download_id.clone();
let on_progress: shanty_dl::ProgressFn = Box::new(move |current, total, msg| { let on_progress: shanty_dl::ProgressFn = Box::new(move |current, total, msg| {
task_state.tasks.update_progress(&progress_tid, current, total, msg); task_state
.tasks
.update_progress(&progress_tid, current, total, msg);
}); });
match shanty_dl::run_queue_with_progress(state.db.conn(), &backend, &backend_config, false, Some(on_progress)).await { match shanty_dl::run_queue_with_progress(
state.db.conn(),
&backend,
&backend_config,
false,
Some(on_progress),
)
.await
{
Ok(stats) => { Ok(stats) => {
let _ = queries::cache::purge_prefix(state.db.conn(), "artist_totals:").await; let _ = queries::cache::purge_prefix(state.db.conn(), "artist_totals:").await;
state.tasks.complete(&download_id, format!("{stats}")); state.tasks.complete(&download_id, format!("{stats}"));
@@ -208,7 +231,9 @@ async fn trigger_pipeline(
// Step 3: Index // Step 3: Index
state.tasks.start(&index_id); state.tasks.start(&index_id);
state.tasks.update_progress(&index_id, 0, 0, "Scanning library..."); state
.tasks
.update_progress(&index_id, 0, 0, "Scanning library...");
let scan_config = shanty_index::ScanConfig { let scan_config = shanty_index::ScanConfig {
root: cfg.library_path.clone(), root: cfg.library_path.clone(),
dry_run: false, dry_run: false,
@@ -221,7 +246,9 @@ async fn trigger_pipeline(
// Step 4: Tag // Step 4: Tag
state.tasks.start(&tag_id); state.tasks.start(&tag_id);
state.tasks.update_progress(&tag_id, 0, 0, "Tagging tracks..."); state
.tasks
.update_progress(&tag_id, 0, 0, "Tagging tracks...");
match shanty_tag::MusicBrainzClient::new() { match shanty_tag::MusicBrainzClient::new() {
Ok(mb) => { Ok(mb) => {
let tag_config = shanty_tag::TagConfig { let tag_config = shanty_tag::TagConfig {
@@ -239,7 +266,9 @@ async fn trigger_pipeline(
// Step 5: Organize // Step 5: Organize
state.tasks.start(&organize_id); state.tasks.start(&organize_id);
state.tasks.update_progress(&organize_id, 0, 0, "Organizing files..."); state
.tasks
.update_progress(&organize_id, 0, 0, "Organizing files...");
let org_config = shanty_org::OrgConfig { let org_config = shanty_org::OrgConfig {
target_dir: cfg.library_path.clone(), target_dir: cfg.library_path.clone(),
format: cfg.organization_format.clone(), format: cfg.organization_format.clone(),
@@ -249,7 +278,8 @@ async fn trigger_pipeline(
match shanty_org::organize_from_db(state.db.conn(), &org_config).await { match shanty_org::organize_from_db(state.db.conn(), &org_config).await {
Ok(stats) => { Ok(stats) => {
let promoted = queries::wanted::promote_downloaded_to_owned(state.db.conn()) let promoted = queries::wanted::promote_downloaded_to_owned(state.db.conn())
.await.unwrap_or(0); .await
.unwrap_or(0);
let msg = if promoted > 0 { let msg = if promoted > 0 {
format!("{stats}{promoted} items marked as owned") format!("{stats}{promoted} items marked as owned")
} else { } else {
@@ -262,9 +292,13 @@ async fn trigger_pipeline(
// Step 6: Enrich // Step 6: Enrich
state.tasks.start(&enrich_id); state.tasks.start(&enrich_id);
state.tasks.update_progress(&enrich_id, 0, 0, "Refreshing artist data..."); state
.tasks
.update_progress(&enrich_id, 0, 0, "Refreshing artist data...");
match enrich_all_watched_artists(&state).await { match enrich_all_watched_artists(&state).await {
Ok(count) => state.tasks.complete(&enrich_id, format!("{count} artists refreshed")), Ok(count) => state
.tasks
.complete(&enrich_id, format!("{count} artists refreshed")),
Err(e) => state.tasks.fail(&enrich_id, e.to_string()), Err(e) => state.tasks.fail(&enrich_id, e.to_string()),
} }
}); });
@@ -283,9 +317,7 @@ async fn get_task(
} }
} }
async fn list_watchlist( async fn list_watchlist(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let items = shanty_watch::list_items(state.db.conn(), None, None).await?; let items = shanty_watch::list_items(state.db.conn(), None, None).await?;
Ok(HttpResponse::Ok().json(items)) Ok(HttpResponse::Ok().json(items))
} }
@@ -299,9 +331,7 @@ async fn remove_watchlist(
Ok(HttpResponse::NoContent().finish()) Ok(HttpResponse::NoContent().finish())
} }
async fn get_config( async fn get_config(state: web::Data<AppState>) -> Result<HttpResponse, ApiError> {
state: web::Data<AppState>,
) -> Result<HttpResponse, ApiError> {
let config = state.config.read().await; let config = state.config.read().await;
Ok(HttpResponse::Ok().json(&*config)) Ok(HttpResponse::Ok().json(&*config))
} }
@@ -319,8 +349,9 @@ async fn save_config(
let new_config = body.into_inner().config; let new_config = body.into_inner().config;
// Persist to YAML // Persist to YAML
new_config.save(state.config_path.as_deref()) new_config
.map_err(|e| ApiError::Internal(e))?; .save(state.config_path.as_deref())
.map_err(ApiError::Internal)?;
// Update in-memory config // Update in-memory config
let mut config = state.config.write().await; let mut config = state.config.write().await;

View File

@@ -1,4 +1,4 @@
use actix_web::{web, HttpResponse}; use actix_web::{HttpResponse, web};
use serde::Deserialize; use serde::Deserialize;
use shanty_db::queries; use shanty_db::queries;
@@ -6,7 +6,9 @@ use shanty_db::queries;
use crate::error::ApiError; use crate::error::ApiError;
use crate::state::AppState; use crate::state::AppState;
fn default_limit() -> u64 { 50 } fn default_limit() -> u64 {
50
}
#[derive(Deserialize)] #[derive(Deserialize)]
pub struct SearchParams { pub struct SearchParams {
@@ -18,14 +20,8 @@ pub struct SearchParams {
} }
pub fn configure(cfg: &mut web::ServiceConfig) { pub fn configure(cfg: &mut web::ServiceConfig) {
cfg.service( cfg.service(web::resource("/tracks").route(web::get().to(list_tracks)))
web::resource("/tracks") .service(web::resource("/tracks/{id}").route(web::get().to(get_track)));
.route(web::get().to(list_tracks)),
)
.service(
web::resource("/tracks/{id}")
.route(web::get().to(get_track)),
);
} }
async fn list_tracks( async fn list_tracks(

View File

@@ -34,6 +34,12 @@ pub struct TaskManager {
tasks: Mutex<HashMap<String, TaskInfo>>, tasks: Mutex<HashMap<String, TaskInfo>>,
} }
impl Default for TaskManager {
fn default() -> Self {
Self::new()
}
}
impl TaskManager { impl TaskManager {
pub fn new() -> Self { pub fn new() -> Self {
Self { Self {