diff --git a/.gitignore b/.gitignore index 0dda1217..05ec847c 100644 --- a/.gitignore +++ b/.gitignore @@ -14,6 +14,8 @@ **/*.wav **/*.opus **/*.mp4 +**/*.mp3 +**/*.ogg xxx /dcai/ **/.pmomusic.yml diff --git a/Cargo.lock b/Cargo.lock index 12b54da7..ddf544a0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -758,30 +758,15 @@ dependencies = [ "libc", ] -[[package]] -name = "crc" -version = "2.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "49fc9a695bca7f35f5f4c15cddc84415f66a74ea78eef08e90c5024f2b540e23" -dependencies = [ - "crc-catalog 1.1.1", -] - [[package]] name = "crc" version = "3.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9710d3b3739c2e349eb44fe848ad0b7c8cb1e42bd87ee49371df2f7acaf3e675" dependencies = [ - "crc-catalog 2.4.0", + "crc-catalog", ] -[[package]] -name = "crc-catalog" -version = "1.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ccaeedb56da03b09f598226e25e80088cb4cd25f316e6e4df7d695f0feeb1403" - [[package]] name = "crc-catalog" version = "2.4.0" @@ -1162,24 +1147,6 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ced73b1dacfc750a6db6c0a0c3a3853c8b41997e2e2c563dc90804ae6867959" -[[package]] -name = "flacenc" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fb6da14d3c6605689b5c9ed5187a5218a6d3888e14b747bc18fd4e4bafd452bd" -dependencies = [ - "built", - "crc 2.1.0", - "crossbeam-channel", - "heapless", - "log", - "md-5", - "num-traits", - "rustversion", - "seq-macro", - "serde", -] - [[package]] name = "flacenc" version = "0.5.0" @@ -1187,7 +1154,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "22846e369206f5c3caa4c59071041fe70c1c2863ebdd2341e6ad17490d53772b" dependencies = [ "built", - "crc 3.3.0", + "crc", "crossbeam-channel", "heapless", "log", @@ -2701,8 +2668,6 @@ dependencies = [ "axum 0.8.6", "bytes", "chrono", - "claxon", - "flacenc 0.4.0", "futures-util", "lofty", "paste", @@ -2712,11 +2677,9 @@ dependencies = [ "pmoflac", "pmoserver", "quick-xml 0.37.5", - "reqwest", "rusqlite", "serde", "serde_json", - "symphonia", "tokio", "tracing", "tracing-subscriber", @@ -2861,7 +2824,7 @@ dependencies = [ "chrono", "claxon", "ffmpeg-next", - "flacenc 0.5.0", + "flacenc", "futures", "hex", "hound", diff --git a/pmoaudiocache/Cargo.toml b/pmoaudiocache/Cargo.toml index 4dd76992..1a7b6a22 100644 --- a/pmoaudiocache/Cargo.toml +++ b/pmoaudiocache/Cargo.toml @@ -20,15 +20,9 @@ chrono = "0.4" # Métadonnées audio lofty = "0.22" -# Encodage/décodage audio -symphonia = { version = "0.5", features = ["all"] } -claxon = "0.4" # Décodeur FLAC (legacy) -flacenc = "0.4" # Encodeur FLAC (legacy) -futures-util = "0.3" # Pour le streaming -bytes = "1.0" # Pour la conversion de streams - -# HTTP client -reqwest = { version = "0.12", features = ["blocking"] } +# Outils de streaming +futures-util = "0.3" +bytes = "1.0" # Utilitaires anyhow = "1.0" diff --git a/pmoaudiocache/src/cache.rs b/pmoaudiocache/src/cache.rs index 43dfd3ab..731d091a 100644 --- a/pmoaudiocache/src/cache.rs +++ b/pmoaudiocache/src/cache.rs @@ -5,7 +5,7 @@ //! des métadonnées en JSON dans la base de données. use anyhow::Result; -use pmocache::{CacheConfig, StreamTransformer}; +use pmocache::CacheConfig; use serde_json::Value; use std::sync::Arc; @@ -33,293 +33,6 @@ impl CacheConfig for AudioConfig { /// Type alias pour le cache audio avec conversion FLAC pub type Cache = pmocache::Cache; -/// Créateur de transformer FLAC (legacy) -/// -/// ⚠️ DEPRECATED: Cette fonction est conservée pour compatibilité mais utilise -/// une approche avec buffer complet. Utilisez `create_streaming_flac_transformer()` -/// pour de meilleures performances et un vrai streaming. -/// -/// Convertit automatiquement tout fichier audio téléchargé en format FLAC -/// en bufferisant d'abord tout le fichier. -/// -/// # Workflow -/// -/// 1. Buffer tous les bytes du stream -/// 2. Décoder l'audio en PCM via Symphonia -/// 3. Encoder le PCM en FLAC via flacenc -/// 4. Écrire le fichier FLAC complet -/// -/// Note: Cette approche nécessite de charger tout le fichier en mémoire. -#[allow(dead_code)] -fn create_flac_transformer_legacy() -> StreamTransformer { - Box::new(|input, mut file, progress| { - Box::pin(async move { - use futures_util::StreamExt; - use tokio::io::AsyncWriteExt; - - // 1. Collecter tous les bytes du stream - // Note: Symphonia nécessite un MediaSource avec Read + Seek, - // ce qui n'est pas compatible avec un vrai streaming HTTP. - // Nous devons donc bufferiser les données. - let mut buffer = Vec::new(); - let mut stream = input.into_byte_stream(); - - while let Some(chunk) = stream.next().await { - let chunk = chunk.map_err(|e| format!("Stream error: {}", e))?; - buffer.extend_from_slice(&chunk); - } - - tracing::debug!( - "Downloaded {} bytes total, starting FLAC conversion", - buffer.len() - ); - - // 2. Si c'est déjà du FLAC, on l'écrit directement - if buffer.len() >= 4 && &buffer[0..4] == b"fLaC" { - tracing::debug!("Input is already FLAC, writing directly"); - file.write_all(&buffer).await.map_err(|e| e.to_string())?; - file.flush().await.map_err(|e| e.to_string())?; - progress(buffer.len() as u64); - return Ok(()); - } - - tracing::debug!("Converting to FLAC with Symphonia + flacenc"); - - // 3. Décoder l'audio avec Symphonia - let (samples, channels, sample_rate, bits_per_sample) = { - use std::io::Cursor; - use symphonia::core::audio::SampleBuffer; - use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL}; - use symphonia::core::errors::Error as SymphoniaError; - use symphonia::core::formats::FormatOptions; - use symphonia::core::io::MediaSourceStream; - use symphonia::core::meta::MetadataOptions; - use symphonia::core::probe::Hint; - - let cursor = Cursor::new(buffer); - let mss = MediaSourceStream::new(Box::new(cursor), Default::default()); - - let hint = Hint::new(); - let probed = symphonia::default::get_probe() - .format( - &hint, - mss, - &FormatOptions::default(), - &MetadataOptions::default(), - ) - .map_err(|e| { - tracing::error!("Symphonia failed to detect audio format: {}", e); - format!( - "Unable to detect audio format. Error: {}. \ - Supported formats: MP3, WAV, OGG, FLAC, AAC, ALAC. \ - Please verify the URL points to a valid audio file.", - e - ) - })?; - - let mut format = probed.format; - - let track = format - .tracks() - .iter() - .find(|t| t.codec_params.codec != CODEC_TYPE_NULL) - .ok_or_else(|| { - tracing::error!("No audio track found in the file"); - "No audio track found in the file. The file may be corrupted or not a valid audio file.".to_string() - })?; - - let codec_name = format!("{:?}", track.codec_params.codec); - tracing::debug!("Detected codec: {}", codec_name); - - let mut decoder = symphonia::default::get_codecs() - .make(&track.codec_params, &DecoderOptions::default()) - .map_err(|e| { - tracing::error!("Failed to create decoder for codec {}: {}", codec_name, e); - format!( - "Codec '{}' is not supported or failed to initialize. Error: {}", - codec_name, e - ) - })?; - - let channels = track - .codec_params - .channels - .ok_or_else(|| { - tracing::error!("Audio file missing channel information"); - "Audio file is missing channel information. The file may be corrupted." - .to_string() - })? - .count(); - - let sample_rate = track.codec_params.sample_rate.ok_or_else(|| { - tracing::error!("Audio file missing sample rate information"); - "Audio file is missing sample rate information. The file may be corrupted." - .to_string() - })?; - - let bits_per_sample = track.codec_params.bits_per_sample.unwrap_or(16); - - let mut samples_i32 = Vec::new(); - let track_id = track.id; - - // Décoder tous les packets - loop { - let packet = match format.next_packet() { - Ok(packet) => packet, - Err(SymphoniaError::ResetRequired) => { - decoder.reset(); - continue; - } - Err(SymphoniaError::IoError(e)) - if e.kind() == std::io::ErrorKind::UnexpectedEof => - { - break; - } - Err(e) => { - tracing::error!("Failed to read audio packet: {}", e); - return Err(format!( - "Failed to read audio data: {}. The file may be corrupted.", - e - )); - } - }; - - if packet.track_id() != track_id { - continue; - } - - match decoder.decode(&packet) { - Ok(decoded) => { - let spec = *decoded.spec(); - let duration = decoded.capacity() as u64; - - // Convertir en i32 pour flacenc - // Note: Symphonia retourne des samples i32, nous devons les convertir - // en fonction du bits_per_sample réel - let mut sample_buf = SampleBuffer::::new(duration, spec); - sample_buf.copy_interleaved_ref(decoded); - samples_i32.extend_from_slice(sample_buf.samples()); - } - Err(SymphoniaError::DecodeError(e)) => { - tracing::warn!("Skipping corrupted audio packet: {}", e); - continue; - } - Err(e) => { - tracing::error!("Fatal decode error: {}", e); - return Err(format!("Failed to decode audio: {}. The file may be corrupted or use an unsupported codec variant.", e)); - } - } - } - - if samples_i32.is_empty() { - tracing::error!("No audio samples could be decoded from the file"); - return Err( - "No audio samples could be decoded. The file may be corrupted or empty." - .to_string(), - ); - } - - tracing::debug!( - "Decoded {} samples (i32), {} channels, {} Hz, {} bits", - samples_i32.len(), - channels, - sample_rate, - bits_per_sample - ); - - // Normaliser les samples i32 vers la plage appropriée pour flacenc - // Symphonia retourne des samples i32 en pleine échelle (32 bits), - // nous devons les normaliser selon le bits_per_sample réel - let (normalized_samples, target_bits): (Vec, u32) = match bits_per_sample { - 0..=16 => { - // Pour 16 bits ou moins, normaliser vers la plage i16 - tracing::debug!("Normalizing to 16-bit"); - let samples = samples_i32.iter().map(|&s| (s >> 16) as i32).collect(); - (samples, 16) - } - 17..=24 => { - // Pour 17-24 bits, normaliser vers la plage 24-bit - tracing::debug!("Normalizing to 24-bit"); - let samples = samples_i32.iter().map(|&s| (s >> 8) as i32).collect(); - (samples, 24) - } - _ => { - // Pour 25-32 bits, garder la pleine échelle i32 - tracing::debug!("Keeping 32-bit"); - (samples_i32, 32) - } - }; - - (normalized_samples, channels, sample_rate, target_bits) - }; - - tracing::debug!( - "Encoding to FLAC: {} samples, {} channels, {} Hz, {} bits", - samples.len(), - channels, - sample_rate, - bits_per_sample - ); - - // 4. Encoder en FLAC avec flacenc - // Note: L'encodage FLAC est une opération bloquante/CPU-intensive, - // donc nous l'exécutons dans un thread bloquant pour ne pas bloquer le runtime Tokio - let flac_data = tokio::task::spawn_blocking(move || { - use flacenc::bitsink::ByteSink; - use flacenc::component::BitRepr; - use flacenc::error::Verify; - - let config = flacenc::config::Encoder::default() - .into_verified() - .map_err(|e| { - tracing::error!("Failed to create FLAC encoder config: {:?}", e); - format!("Internal error: FLAC encoder configuration failed: {:?}", e) - })?; - - let source = flacenc::source::MemSource::from_samples( - &samples, - channels, - bits_per_sample as usize, - sample_rate as usize, - ); - - let flac_stream = - flacenc::encode_with_fixed_block_size(&config, source, config.block_size) - .map_err(|e| { - tracing::error!("FLAC encoding failed: {:?}", e); - format!("Failed to encode audio to FLAC format: {:?}", e) - })?; - - let mut sink = ByteSink::new(); - flac_stream.write(&mut sink).map_err(|e| { - tracing::error!("Failed to write FLAC stream: {:?}", e); - format!("Failed to write FLAC data: {:?}", e) - })?; - - Ok::, String>(sink.into_inner()) - }) - .await - .map_err(|e| { - tracing::error!("FLAC encoding task panicked: {}", e); - format!("Internal error: FLAC encoding task failed: {}", e) - })??; - - tracing::debug!("FLAC encoding complete: {} bytes", flac_data.len()); - - // 5. Écrire le fichier FLAC - file.write_all(&flac_data) - .await - .map_err(|e| e.to_string())?; - file.flush().await.map_err(|e| e.to_string())?; - - // 6. Mettre à jour la progression finale - progress(flac_data.len() as u64); - - Ok(()) - }) - }) -} - /// Crée un cache audio avec conversion FLAC automatique /// /// # Arguments diff --git a/pmoaudiocache/src/flac.rs b/pmoaudiocache/src/flac.rs deleted file mode 100644 index 1cc4d760..00000000 --- a/pmoaudiocache/src/flac.rs +++ /dev/null @@ -1,150 +0,0 @@ -// //! Module de conversion audio en FLAC -// //! -// //! Ce module gère la conversion de divers formats audio vers FLAC -// //! pour standardiser le stockage dans le cache. - -// use anyhow::{anyhow, Result}; -// use std::io::Cursor; -// use symphonia::core::audio::SampleBuffer; -// use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL}; -// use symphonia::core::errors::Error as SymphoniaError; -// use symphonia::core::formats::FormatOptions; -// use symphonia::core::io::MediaSourceStream; -// use symphonia::core::meta::MetadataOptions; -// use symphonia::core::probe::Hint; - -// /// Convertit des données audio en FLAC -// /// -// /// Cette fonction accepte n'importe quel format audio supporté par Symphonia -// /// et le convertit en FLAC pour un stockage standardisé. -// /// -// /// # Arguments -// /// -// /// * `data` - Données audio brutes (n'importe quel format) -// /// * `extension` - Extension du fichier source (optionnel, aide à la détection) -// /// -// /// # Returns -// /// -// /// Données audio au format FLAC -// /// -// /// # Exemple -// /// -// /// ```rust,no_run -// /// use pmoaudiocache::flac::convert_to_flac; -// /// -// /// let mp3_data = std::fs::read("track.mp3").unwrap(); -// /// let flac_data = convert_to_flac(&mp3_data, Some("mp3")).unwrap(); -// /// ``` -// pub fn convert_to_flac(data: &[u8], extension: Option<&str>) -> Result> { -// // Si c'est déjà du FLAC, on le retourne tel quel -// if is_flac(data) { -// return Ok(data.to_vec()); -// } - -// // Créer un MediaSource depuis les données (en clonant pour avoir 'static) -// let data_owned = data.to_vec(); -// let cursor = Cursor::new(data_owned); -// let mss = MediaSourceStream::new(Box::new(cursor), Default::default()); - -// // Créer un hint si on a l'extension -// let mut hint = Hint::new(); -// if let Some(ext) = extension { -// hint.with_extension(ext); -// } - -// // Prober le format -// let probed = symphonia::default::get_probe() -// .format( -// &hint, -// mss, -// &FormatOptions::default(), -// &MetadataOptions::default(), -// ) -// .map_err(|e| anyhow!("Impossible de détecter le format audio: {}", e))?; - -// let mut format = probed.format; - -// // Obtenir le premier track audio -// let track = format -// .tracks() -// .iter() -// .find(|t| t.codec_params.codec != CODEC_TYPE_NULL) -// .ok_or_else(|| anyhow!("Aucune piste audio trouvée"))?; - -// // Créer un décodeur -// let mut decoder = symphonia::default::get_codecs() -// .make(&track.codec_params, &DecoderOptions::default()) -// .map_err(|e| anyhow!("Impossible de créer le décodeur: {}", e))?; - -// // Buffer pour stocker les samples décodés -// let mut samples = Vec::new(); -// let track_id = track.id; - -// // Décoder tous les packets -// loop { -// let packet = match format.next_packet() { -// Ok(packet) => packet, -// Err(SymphoniaError::ResetRequired) => { -// // Reset du décodeur requis -// decoder.reset(); -// continue; -// } -// Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { -// break; -// } -// Err(e) => return Err(anyhow!("Erreur lors de la lecture: {}", e)), -// }; - -// // Ignorer les packets qui ne sont pas de notre track -// if packet.track_id() != track_id { -// continue; -// } - -// match decoder.decode(&packet) { -// Ok(decoded) => { -// // Convertir les samples en format standard -// let spec = *decoded.spec(); -// let duration = decoded.capacity() as u64; - -// let mut sample_buf = SampleBuffer::::new(duration, spec); -// sample_buf.copy_interleaved_ref(decoded); - -// samples.extend_from_slice(sample_buf.samples()); -// } -// Err(SymphoniaError::DecodeError(_)) => continue, -// Err(e) => return Err(anyhow!("Erreur de décodage: {}", e)), -// } -// } - -// if samples.is_empty() { -// return Err(anyhow!("Aucun sample décodé")); -// } - -// // Note: Pour l'encodage FLAC, on aurait besoin d'une bibliothèque comme -// // `flacenc` qui n'existe pas encore en Rust. Pour l'instant, on stocke -// // les données telles quelles si c'est déjà du FLAC, sinon on retourne -// // les données originales avec un warning. - -// // TODO: Implémenter l'encodage FLAC quand une bibliothèque sera disponible -// tracing::warn!("Encodage FLAC non implémenté, stockage du format original"); -// Ok(data.to_vec()) -// } - -// /// Vérifie si les données sont déjà au format FLAC -// fn is_flac(data: &[u8]) -> bool { -// data.len() >= 4 && &data[0..4] == b"fLaC" -// } - -// #[cfg(test)] -// mod tests { -// use super::*; - -// #[test] -// fn test_is_flac() { -// let flac_header = b"fLaC\x00\x00\x00\x22"; -// assert!(is_flac(flac_header)); - -// let not_flac = b"RIFF\x00\x00\x00\x00"; -// assert!(!is_flac(not_flac)); -// } -// } diff --git a/pmoaudiocache/src/streaming.rs b/pmoaudiocache/src/streaming.rs index 803f566a..8007adf9 100644 --- a/pmoaudiocache/src/streaming.rs +++ b/pmoaudiocache/src/streaming.rs @@ -1,454 +1,91 @@ -//! Module de streaming audio avec conversion FLAC progressive +//! Audio streaming transformer built on pmoflac. //! -//! Ce module fournit un transformer qui utilise pmoflac pour traiter -//! les fichiers audio en vrai streaming, permettant de rendre les fichiers -//! FLAC disponibles progressivement pendant le téléchargement. +//! This module wires the generic `pmocache` download pipeline with the new +//! streaming transcode helper provided by `pmoflac`. Any supported codec +//! (FLAC, MP3, OGG/Vorbis, Opus, WAV, AIFF) is converted to FLAC on the fly, +//! while native FLAC input is forwarded byte-for-byte without re-encoding. -use futures_util::StreamExt; +use bytes::Bytes; use pmocache::StreamTransformer; -use std::sync::Arc; -use tokio::io::AsyncWriteExt; +use pmoflac::{transcode_to_flac_stream, AudioCodec, TranscodeOptions}; +use tokio::io::{AsyncReadExt, AsyncWriteExt}; -/// Créateur de transformer FLAC avec streaming progressif -/// -/// Ce transformer améliore significativement la disponibilité des fichiers : -/// -/// # Pour les sources FLAC -/// -/// Pipeline streaming complet : -/// 1. Télécharger les chunks HTTP au fur et à mesure -/// 2. Décoder FLAC → PCM avec pmoflac (streaming) -/// 3. Re-encoder PCM → FLAC avec pmoflac (streaming) -/// 4. Écrire les frames FLAC immédiatement dans le fichier -/// -/// Résultat : Les fichiers sont disponibles pour lecture en ~100ms au lieu -/// d'attendre le téléchargement complet ! -/// -/// # Pour les autres formats (MP3, OGG, AAC, etc.) -/// -/// Pipeline hybride : -/// 1. Buffer complet du fichier (nécessaire pour Symphonia) -/// 2. Décoder avec Symphonia (Read+Seek requis) -/// 3. Encoder avec pmoflac en streaming -/// 4. Écrire progressivement le FLAC -/// -/// Résultat : Pas de gain sur la latence initiale, mais meilleures performances -/// d'encodage et usage mémoire optimisé. +/// Creates the transformer consumed by the audio cache. pub fn create_streaming_flac_transformer() -> StreamTransformer { - Box::new(|input, file, progress| { + Box::new(|input, mut file, progress| { Box::pin(async move { - // TODO: Implémenter la détection de format pour utiliser le pipeline - // streaming complet pour les sources FLAC - // Pour l'instant, on utilise toujours le pipeline hybride qui fonctionne - // pour tous les formats + let byte_stream = input.into_byte_stream(); + let reader = StreamToAsyncRead::new(byte_stream); - tracing::debug!("Using buffered decode + streaming encode pipeline"); - buffer_and_convert_to_flac(input, file, progress).await + let transcode = transcode_to_flac_stream(reader, TranscodeOptions::default()) + .await + .map_err(|e| format!("Audio transcode error: {}", e))?; + + log_stream_info(transcode.input_codec(), transcode.input_stream_info()); + + let mut flac_stream = transcode.into_stream(); + let mut buffer = vec![0u8; 64 * 1024]; + let mut total_written = 0u64; + + loop { + let read = flac_stream + .read(&mut buffer) + .await + .map_err(|e| format!("Failed to read FLAC data: {}", e))?; + + if read == 0 { + break; + } + + file.write_all(&buffer[..read]) + .await + .map_err(|e| format!("Failed to write FLAC file: {}", e))?; + + total_written += read as u64; + progress(total_written); + } + + file.flush() + .await + .map_err(|e| format!("Failed to flush FLAC file: {}", e))?; + + flac_stream + .wait() + .await + .map_err(|e| format!("FLAC encoder error: {}", e))?; + + Ok(()) }) }) } -#[derive(Debug, Clone, Copy)] -enum AudioFormat { - Flac, - Other, -} - -/// Détecte le format audio en analysant les premiers bytes -async fn detect_format(_input: &pmocache::download::CacheInput) -> Result { - // On ne peut pas peek CacheInput directement sans le consommer, - // donc on va détecter pendant le traitement du stream - // Pour l'instant, on retourne Other par défaut - Ok(AudioFormat::Other) -} - -/// Pipeline streaming complet pour FLAC → FLAC -/// -/// Cette fonction implémente le vrai streaming sans buffer : -/// - Lecture progressive du stream HTTP -/// - Décodage FLAC → PCM au fur et à mesure -/// - Re-encodage PCM → FLAC au fur et à mesure -/// - Écriture progressive dans le fichier -async fn stream_flac_to_flac( - input: pmocache::download::CacheInput, - mut file: tokio::fs::File, - progress: Arc, -) -> Result<(), String> { - use pmoflac::{decode_flac_stream, encode_flac_stream, EncoderOptions, PcmFormat}; - use tokio::io::AsyncReadExt; - - // Convertir le CacheInput en stream - let stream = input.into_byte_stream(); - - // Créer un lecteur depuis le stream de bytes - let reader = StreamToAsyncRead::new(stream); - - // Décoder le FLAC en PCM (streaming) - let decoded_stream = decode_flac_stream(reader) - .await - .map_err(|e| format!("FLAC decode error: {}", e))?; - - let info = decoded_stream.info().clone(); +fn log_stream_info(codec: AudioCodec, info: &pmoflac::StreamInfo) { tracing::debug!( - "FLAC stream info: {} Hz, {} channels, {} bits/sample", + "Detected codec {:?}: {} Hz, {} channels, {} bits/sample (passthrough={})", + codec, info.sample_rate, info.channels, - info.bits_per_sample + info.bits_per_sample, + codec == AudioCodec::Flac ); - - // Créer le format PCM depuis les infos du stream - let pcm_format = PcmFormat { - sample_rate: info.sample_rate, - channels: info.channels, - bits_per_sample: info.bits_per_sample, - }; - - // Options d'encodage - let encoder_options = EncoderOptions { - compression_level: 5, - verify: false, - total_samples: info.total_samples, - block_size: Some(info.max_block_size as u32), - }; - - // Re-encoder PCM → FLAC (streaming) - // Note: encode_flac_stream consomme decoded_stream - let mut encoded_stream = encode_flac_stream(decoded_stream, pcm_format, encoder_options) - .await - .map_err(|e| format!("FLAC encode error: {}", e))?; - - // Écrire le FLAC encodé dans le fichier au fur et à mesure - let mut total_written = 0u64; - let mut buffer = vec![0u8; 64 * 1024]; // Buffer de 64 KB - - loop { - let n = encoded_stream - .read(&mut buffer) - .await - .map_err(|e| format!("Failed to read encoded FLAC: {}", e))?; - - if n == 0 { - break; - } - - file.write_all(&buffer[..n]) - .await - .map_err(|e| format!("Failed to write to file: {}", e))?; - - total_written += n as u64; - progress(total_written); - } - - file.flush() - .await - .map_err(|e| format!("Failed to flush file: {}", e))?; - - // Attendre que la tâche d'encodage se termine - // (l'encoder attend automatiquement que le decoder se termine) - encoded_stream - .wait() - .await - .map_err(|e| format!("Encoder/Decoder error: {}", e))?; - - tracing::debug!( - "Streaming FLAC conversion complete: {} bytes", - total_written - ); - - Ok(()) } -/// Pipeline hybride pour autres formats → FLAC -/// -/// Buffer le fichier complet (nécessaire pour Symphonia), puis encode en streaming -async fn buffer_and_convert_to_flac( - input: pmocache::download::CacheInput, - mut file: tokio::fs::File, - progress: Arc, -) -> Result<(), String> { - use pmoflac::{encode_flac_stream, EncoderOptions, PcmFormat}; - - // 1. Collecter tous les bytes du stream - let mut buffer = Vec::new(); - let mut stream = input.into_byte_stream(); - - while let Some(chunk) = stream.next().await { - let chunk = chunk.map_err(|e| format!("Stream error: {}", e))?; - buffer.extend_from_slice(&chunk); - } - - tracing::debug!( - "Downloaded {} bytes total, starting conversion", - buffer.len() - ); - - // 2. Si c'est déjà du FLAC, on l'écrit directement - if buffer.len() >= 4 && &buffer[0..4] == b"fLaC" { - tracing::debug!("Input is already FLAC, writing directly"); - file.write_all(&buffer).await.map_err(|e| e.to_string())?; - file.flush().await.map_err(|e| e.to_string())?; - progress(buffer.len() as u64); - return Ok(()); - } - - tracing::debug!("Converting to FLAC with Symphonia + pmoflac"); - - // 3. Décoder l'audio avec Symphonia (dans un blocking task car c'est CPU-intensive) - let (samples, channels, sample_rate, bits_per_sample) = - tokio::task::spawn_blocking(move || decode_with_symphonia_sync(buffer)) - .await - .map_err(|e| format!("Decode task panicked: {}", e))??; - - tracing::debug!( - "Decoded {} samples, {} channels, {} Hz, {} bits", - samples.len(), - channels, - sample_rate, - bits_per_sample - ); - - // 4. Convertir les samples i32 en bytes PCM little-endian - let pcm_bytes = samples_to_pcm_bytes(&samples, bits_per_sample); - - // 5. Encoder en FLAC avec pmoflac (streaming) - let pcm_format = PcmFormat { - sample_rate, - channels: channels as u8, - bits_per_sample: bits_per_sample as u8, - }; - - let encoder_options = EncoderOptions { - compression_level: 5, - verify: false, - total_samples: Some((samples.len() / channels) as u64), - block_size: None, - }; - - // Utiliser tokio::io::duplex pour éviter le problème de lifetime - use std::io::Cursor; - let cursor = Cursor::new(pcm_bytes); - let mut encoded_stream = encode_flac_stream(cursor, pcm_format, encoder_options) - .await - .map_err(|e| format!("FLAC encode error: {}", e))?; - - // 6. Écrire le FLAC encodé progressivement - use tokio::io::AsyncReadExt; - let mut total_written = 0u64; - let mut write_buffer = vec![0u8; 64 * 1024]; - - loop { - let n = encoded_stream - .read(&mut write_buffer) - .await - .map_err(|e| format!("Failed to read encoded FLAC: {}", e))?; - - if n == 0 { - break; - } - - file.write_all(&write_buffer[..n]) - .await - .map_err(|e| format!("Failed to write to file: {}", e))?; - - total_written += n as u64; - progress(total_written); - } - - file.flush() - .await - .map_err(|e| format!("Failed to flush file: {}", e))?; - - encoded_stream - .wait() - .await - .map_err(|e| format!("Encoder error: {}", e))?; - - tracing::debug!("FLAC conversion complete: {} bytes", total_written); - - Ok(()) -} - -/// Décode un fichier audio avec Symphonia -/// -/// Retourne (samples, channels, sample_rate, bits_per_sample) -fn decode_with_symphonia_sync(buffer: Vec) -> Result<(Vec, usize, u32, u32), String> { - use std::io::Cursor; - use symphonia::core::audio::SampleBuffer; - use symphonia::core::codecs::{DecoderOptions, CODEC_TYPE_NULL}; - use symphonia::core::errors::Error as SymphoniaError; - use symphonia::core::formats::FormatOptions; - use symphonia::core::io::MediaSourceStream; - use symphonia::core::meta::MetadataOptions; - use symphonia::core::probe::Hint; - - let cursor = Cursor::new(buffer); - let mss = MediaSourceStream::new(Box::new(cursor), Default::default()); - - let hint = Hint::new(); - let probed = symphonia::default::get_probe() - .format( - &hint, - mss, - &FormatOptions::default(), - &MetadataOptions::default(), - ) - .map_err(|e| { - format!( - "Unable to detect audio format: {}. \ - Supported formats: MP3, WAV, OGG, FLAC, AAC, ALAC.", - e - ) - })?; - - let mut format = probed.format; - - let track = format - .tracks() - .iter() - .find(|t| t.codec_params.codec != CODEC_TYPE_NULL) - .ok_or_else(|| { - "No audio track found in the file. The file may be corrupted.".to_string() - })?; - - let codec_name = format!("{:?}", track.codec_params.codec); - tracing::debug!("Detected codec: {}", codec_name); - - let mut decoder = symphonia::default::get_codecs() - .make(&track.codec_params, &DecoderOptions::default()) - .map_err(|e| format!("Codec '{}' is not supported: {}", codec_name, e))?; - - let channels = track - .codec_params - .channels - .ok_or_else(|| "Audio file missing channel information.".to_string())? - .count(); - - let sample_rate = track - .codec_params - .sample_rate - .ok_or_else(|| "Audio file missing sample rate information.".to_string())?; - - let bits_per_sample = track.codec_params.bits_per_sample.unwrap_or(16); - - let mut samples_i32 = Vec::new(); - let track_id = track.id; - - // Décoder tous les packets - loop { - let packet = match format.next_packet() { - Ok(packet) => packet, - Err(SymphoniaError::ResetRequired) => { - decoder.reset(); - continue; - } - Err(SymphoniaError::IoError(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => { - break; - } - Err(e) => { - return Err(format!( - "Failed to read audio data: {}. The file may be corrupted.", - e - )); - } - }; - - if packet.track_id() != track_id { - continue; - } - - match decoder.decode(&packet) { - Ok(decoded) => { - let spec = *decoded.spec(); - let duration = decoded.capacity() as u64; - - let mut sample_buf = SampleBuffer::::new(duration, spec); - sample_buf.copy_interleaved_ref(decoded); - samples_i32.extend_from_slice(sample_buf.samples()); - } - Err(SymphoniaError::DecodeError(e)) => { - tracing::warn!("Skipping corrupted audio packet: {}", e); - continue; - } - Err(e) => { - return Err(format!( - "Failed to decode audio: {}. The file may be corrupted.", - e - )); - } - } - } - - if samples_i32.is_empty() { - return Err("No audio samples could be decoded. The file may be corrupted.".to_string()); - } - - // Normaliser les samples selon le bits_per_sample - let (normalized_samples, target_bits): (Vec, u32) = match bits_per_sample { - 0..=16 => { - tracing::debug!("Normalizing to 16-bit"); - let samples = samples_i32.iter().map(|&s| (s >> 16) as i32).collect(); - (samples, 16) - } - 17..=24 => { - tracing::debug!("Normalizing to 24-bit"); - let samples = samples_i32.iter().map(|&s| (s >> 8) as i32).collect(); - (samples, 24) - } - _ => { - tracing::debug!("Keeping 32-bit"); - (samples_i32, 32) - } - }; - - Ok((normalized_samples, channels, sample_rate, target_bits)) -} - -/// Convertit des samples i32 en bytes PCM little-endian -fn samples_to_pcm_bytes(samples: &[i32], bits_per_sample: u32) -> Vec { - let bytes_per_sample = (bits_per_sample / 8) as usize; - let mut bytes = Vec::with_capacity(samples.len() * bytes_per_sample); - - for &sample in samples { - match bits_per_sample { - 16 => { - let s = sample as i16; - bytes.extend_from_slice(&s.to_le_bytes()); - } - 24 => { - let s_bytes = sample.to_le_bytes(); - bytes.extend_from_slice(&s_bytes[0..3]); - } - 32 => { - bytes.extend_from_slice(&sample.to_le_bytes()); - } - _ => { - // Fallback pour bits non standard - bytes.extend_from_slice(&sample.to_le_bytes()); - } - } - } - - bytes -} - -/// Adaptateur qui convertit un Stream de Bytes en AsyncRead +/// Adapter exposing a byte stream as `AsyncRead`. struct StreamToAsyncRead { - stream: futures_util::stream::BoxStream<'static, Result>, - current_chunk: Option, - chunk_offset: usize, + stream: futures_util::stream::BoxStream<'static, Result>, + current_chunk: Option, + offset: usize, } impl StreamToAsyncRead { fn new( - stream: std::pin::Pin< - Box> + Send>, - >, + stream: std::pin::Pin> + Send>>, ) -> Self { use futures_util::StreamExt; Self { stream: stream.boxed(), current_chunk: None, - chunk_offset: 0, + offset: 0, } } } @@ -463,41 +100,33 @@ impl tokio::io::AsyncRead for StreamToAsyncRead { use std::task::Poll; loop { - // Si on a un chunk courant, lire dedans if let Some(chunk) = &self.current_chunk { - if self.chunk_offset < chunk.len() { - let available = chunk.len() - self.chunk_offset; - let to_read = std::cmp::min(available, buf.remaining()); - buf.put_slice(&chunk[self.chunk_offset..self.chunk_offset + to_read]); - self.chunk_offset += to_read; + if self.offset < chunk.len() { + let available = chunk.len() - self.offset; + let to_copy = available.min(buf.remaining()); + buf.put_slice(&chunk[self.offset..self.offset + to_copy]); + self.offset += to_copy; return Poll::Ready(Ok(())); - } else { - // Chunk épuisé, passer au suivant - self.current_chunk = None; - self.chunk_offset = 0; } + self.current_chunk = None; + self.offset = 0; } - // Pas de chunk courant, en récupérer un nouveau match self.stream.poll_next_unpin(cx) { Poll::Ready(Some(Ok(chunk))) => { if chunk.is_empty() { continue; } self.current_chunk = Some(chunk); - self.chunk_offset = 0; } Poll::Ready(Some(Err(e))) => { return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::Other, e))); } - Poll::Ready(None) => { - // Stream terminé - return Poll::Ready(Ok(())); - } - Poll::Pending => { - return Poll::Pending; - } + Poll::Ready(None) => return Poll::Ready(Ok(())), + Poll::Pending => return Poll::Pending, } } } } + +impl Unpin for StreamToAsyncRead {}