eric/pmomusic
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Update la web app pour tirer partie du nouveau systeme de cache

Browse Source
This commit is contained in:
Eric Coissac
2025-10-28 00:10:51 +01:00
parent b4a8925281
commit a6ed30e0c7
14 changed files with 459 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
pmocache/src/api.rs
View File

@@ -15,6 +15,7 @@ use axum::{
Json,
};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::sync::Arc;
#[cfg(feature = "openapi")]
@@ -39,6 +40,21 @@ pub struct DownloadStatus {
pub finished: bool,
/// Erreur éventuelle
pub error: Option<String>,
/// Informations sur la conversion
pub conversion: Option<ConversionStatus>,
}
/// Informations sur la conversion en cours ou réalisée
#[derive(Debug, Serialize, Deserialize, Clone)]
#[cfg_attr(feature = "openapi", derive(ToSchema))]
pub struct ConversionStatus {
/// Mode de conversion (ex: "passthrough", "transcode")
#[cfg_attr(feature = "openapi", schema(example = "passthrough"))]
pub mode: String,
/// Codec source détecté (si disponible)
pub input_codec: Option<String>,
/// Informations complémentaires lisibles (optionnel)
pub details: Option<String>,
}
/// Requête pour ajouter un item au cache
@@ -134,30 +150,74 @@ pub async fn get_download_status<C: CacheConfig>(
State(cache): State<Arc<Cache<C>>>,
Path(pk): Path<String>,
) -> impl IntoResponse {
// Vérifier que l'item existe dans la DB
if cache.db.get(&pk, false).is_err() {
return (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: "NOT_FOUND".to_string(),
message: format!("Item with pk '{}' not found in cache", pk),
}),
)
.into_response();
}
let entry = match cache.db.get(&pk, false) {
Ok(entry) => entry,
Err(_) => {
return (
StatusCode::NOT_FOUND,
Json(ErrorResponse {
error: "NOT_FOUND".to_string(),
message: format!("Item with pk '{}' not found in cache", pk),
}),
)
.into_response();
}
};
let in_progress = cache.get_download(&pk).await.is_some();
let current_size = cache.current_size(&pk).await;
let transformed_size = cache.transformed_size(&pk).await;
let expected_size = cache.expected_size(&pk).await;
let finished = cache.is_finished(&pk).await;
let download = cache.get_download(&pk).await;
let file_path = cache.get_file_path(&pk);
let file_size = if file_path.exists() {
std::fs::metadata(&file_path).ok().map(|m| m.len())
} else {
None
};
let error = if let Some(download) = cache.get_download(&pk).await {
let in_progress = download.is_some();
let current_size = if let Some(download) = download.as_ref() {
Some(download.current_size().await)
} else {
file_size
};
let transformed_size = if let Some(download) = download.as_ref() {
Some(download.transformed_size().await)
} else {
file_size
};
let expected_size = if let Some(download) = download.as_ref() {
download.expected_size().await
} else {
file_size
};
let finished = if let Some(download) = download.as_ref() {
download.finished().await
} else {
file_path.exists()
};
let error = if let Some(download) = download.as_ref() {
download.error().await
} else {
None
};
let mut conversion = if let Some(download) = download.as_ref() {
download
.transform_metadata()
.await
.map(ConversionStatus::from)
} else {
None
};
if conversion.is_none() {
if let Some(meta) = entry.metadata.as_ref() {
conversion = conversion_from_json(meta);
}
}
let status = DownloadStatus {
pk,
in_progress,
@@ -166,11 +226,28 @@ pub async fn get_download_status<C: CacheConfig>(
expected_size,
finished,
error,
conversion,
};
(StatusCode::OK, Json(status)).into_response()
}
impl From<crate::download::TransformMetadata> for ConversionStatus {
fn from(value: crate::download::TransformMetadata) -> Self {
Self {
mode: value.mode.unwrap_or_else(|| "unknown".to_string()),
input_codec: value.input_codec,
details: value.details,
}
}
}
fn conversion_from_json(value: &Value) -> Option<ConversionStatus> {
value
.get("conversion")
.and_then(|conv| serde_json::from_value(conv.clone()).ok())
}
/// Ajoute un item au cache depuis une URL
///
/// Télécharge l'item depuis l'URL fournie et l'ajoute au cache.

12
pmocache/src/cache.rs
View File

@@ -94,9 +94,10 @@ impl<C: CacheConfig> Cache<C> {
///
/// let transformer_factory = Arc::new(|| {
/// // Créer un transformer qui convertit les données
/// Box::new(|input, file, progress| {
/// Box::new(|input, file, ctx| {
/// Box::pin(async move {
/// // Transformation personnalisée
/// ctx.report_progress(0);
/// Ok(())
/// })
/// }) as StreamTransformer
@@ -610,6 +611,15 @@ impl<C: CacheConfig> Cache<C> {
}
}
/// Retourne les métadonnées de transformation (si disponibles)
pub async fn transform_metadata(&self, pk: &str) -> Option<crate::download::TransformMetadata> {
if let Some(download) = self.get_download(pk).await {
download.transform_metadata().await
} else {
None
}
}
/// Indique si le téléchargement est terminé
///
/// # Arguments

51
pmocache/src/download.rs
View File

@@ -15,18 +15,20 @@ use tokio_util::io::ReaderStream;
/// La fonction reçoit :
/// - Un `CacheInput` abstrait (HTTP ou lecteur en streaming)
/// - Un writer pour écrire les données transformées
/// - Un callback pour mettre à jour la progression
/// - Un contexte fournissant des utilitaires (progression, métadonnées)
///
/// Elle retourne un `Future` qui se résout en `Result`.
pub type StreamTransformer = Box<
dyn FnOnce(
CacheInput,
tokio::fs::File,
Arc<dyn Fn(u64) + Send + Sync>,
TransformContextHandle,
) -> Pin<Box<dyn Future<Output = Result<(), String>> + Send>>
+ Send,
>;
pub type TransformContextHandle = Arc<TransformContext>;
type ByteStream = Pin<Box<dyn Stream<Item = Result<Bytes, String>> + Send>>;
/// Source générique (HTTP ou lecteur) exposée aux transformers.
@@ -180,6 +182,7 @@ struct DownloadState {
finished: bool,
read_position: u64,
error: Option<String>,
transform_metadata: Option<TransformMetadata>,
}
/// Objet représentant un téléchargement en cours
@@ -200,6 +203,7 @@ impl Download {
finished: false,
read_position: 0,
error: None,
transform_metadata: None,
})),
})
}
@@ -274,6 +278,45 @@ impl Download {
let state = self.state.read().await;
state.error.clone()
}
pub async fn transform_metadata(&self) -> Option<TransformMetadata> {
let state = self.state.read().await;
state.transform_metadata.clone()
}
}
#[derive(Debug, Clone, Default)]
pub struct TransformMetadata {
pub mode: Option<String>,
pub input_codec: Option<String>,
pub details: Option<String>,
}
pub struct TransformContext {
state: Arc<RwLock<DownloadState>>,
progress_cb: Arc<dyn Fn(u64) + Send + Sync>,
}
impl TransformContext {
fn new(state: Arc<RwLock<DownloadState>>, progress_cb: Arc<dyn Fn(u64) + Send + Sync>) -> Self {
Self { state, progress_cb }
}
/// Reports progress (in bytes) to the download state.
pub fn report_progress(&self, bytes: u64) {
(self.progress_cb)(bytes);
}
/// Returns the underlying progress callback (useful for piping into other APIs).
pub fn progress_callback(&self) -> Arc<dyn Fn(u64) + Send + Sync> {
Arc::clone(&self.progress_cb)
}
/// Stores metadata describing the transformation that occurred.
pub async fn set_metadata(&self, metadata: TransformMetadata) {
let mut state = self.state.write().await;
state.transform_metadata = Some(metadata);
}
}
/// Lance le téléchargement d'une URL dans un fichier.
@@ -402,7 +445,9 @@ async fn process_input(
});
});
match transformer(input, file, Arc::clone(&progress_callback)).await {
let context = Arc::new(TransformContext::new(Arc::clone(&state), progress_callback));
match transformer(input, file, Arc::clone(&context)).await {
Ok(_) => {
let mut s = state.write().await;
if s.current_size == 0 {

2
pmocache/src/lib.rs
View File

@@ -134,7 +134,7 @@ pub use cache_trait::{pk_from_content_header, FileCache};
pub use db::{CacheEntry, DB};
pub use download::{
download, download_with_transformer, ingest_with_transformer, peek_header, peek_reader_header,
Download, StreamTransformer,
Download, StreamTransformer, TransformContextHandle, TransformMetadata,
};
#[cfg(feature = "pmoserver")]