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

Patch of the web logger

Browse Source
This commit is contained in:
Eric Coissac
2025-10-20 16:23:50 +02:00
parent f5f6bcf3ab
commit f510e59b1a
12 changed files with 432 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
pmoparadise/src/client.rs
View File

@@ -6,6 +6,44 @@ use reqwest::Client;
use std::time::Duration;
use url::Url;
fn normalize_base_url(base: &str) -> String {
let mut normalized = base.trim().to_string();
if normalized.is_empty() {
return "https://img.radioparadise.com/".to_string();
}
if normalized.starts_with("//") {
normalized = format!("https:{}", normalized);
} else if !(normalized.starts_with("http://") || normalized.starts_with("https://")) {
normalized = format!("https://{}", normalized.trim_start_matches('/'));
}
if !normalized.ends_with('/') {
normalized.push('/');
}
normalized
}
fn resolve_cover_with_base(base: &str, cover_path: &str) -> Result<Url> {
let cover_path = cover_path.trim();
if cover_path.starts_with("http://") || cover_path.starts_with("https://") {
return Ok(Url::parse(cover_path)?);
}
if cover_path.starts_with("//") {
let url = format!("https:{}", cover_path);
return Ok(Url::parse(&url)?);
}
let base = normalize_base_url(base);
let base_url = Url::parse(&base)?;
Ok(base_url.join(cover_path)?)
}
/// Default Radio Paradise API base URL
pub const DEFAULT_API_BASE: &str = "https://api.radioparadise.com/api";
@@ -13,10 +51,13 @@ pub const DEFAULT_API_BASE: &str = "https://api.radioparadise.com/api";
pub const DEFAULT_BLOCK_BASE: &str = "https://apps.radioparadise.com/blocks/chan/0";
/// Default image base URL
pub const DEFAULT_IMAGE_BASE: &str = "https://img.radioparadise.com/covers/l/";
pub const DEFAULT_IMAGE_BASE: &str = "https://img.radioparadise.com/";
/// Default timeout for HTTP requests
pub const DEFAULT_TIMEOUT_SECS: u64 = 30;
/// Default timeout for metadata HTTP requests
pub const DEFAULT_REQUEST_TIMEOUT_SECS: u64 = 30;
/// Default timeout for large block downloads/streams
pub const DEFAULT_BLOCK_TIMEOUT_SECS: u64 = 180;
/// Default User-Agent
pub const DEFAULT_USER_AGENT: &str = "pmoparadise/0.1.0";
@@ -49,7 +90,8 @@ pub struct RadioParadiseClient {
image_base: String,
bitrate: Bitrate,
channel: u8,
pub(crate) timeout: Duration,
pub(crate) request_timeout: Duration,
pub(crate) block_timeout: Duration,
next_block_url: Option<String>,
}
@@ -74,10 +116,11 @@ impl RadioParadiseClient {
client,
api_base: DEFAULT_API_BASE.to_string(),
block_base: DEFAULT_BLOCK_BASE.to_string(),
image_base: DEFAULT_IMAGE_BASE.to_string(),
image_base: normalize_base_url(DEFAULT_IMAGE_BASE),
bitrate: Bitrate::default(),
channel: 0,
timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS),
request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS),
block_timeout: Duration::from_secs(DEFAULT_BLOCK_TIMEOUT_SECS),
next_block_url: None,
}
}
@@ -162,7 +205,12 @@ impl RadioParadiseClient {
#[cfg(feature = "logging")]
tracing::debug!("Fetching block: {}", url);
let response = self.client.get(url).timeout(self.timeout).send().await?;
let response = self
.client
.get(url)
.timeout(self.request_timeout)
.send()
.await?;
if !response.status().is_success() {
return Err(Error::other(format!(
@@ -174,7 +222,9 @@ impl RadioParadiseClient {
let mut block: Block = response.json().await?;
// Set image_base if not provided
if block.image_base.is_none() {
if let Some(ref mut base) = block.image_base {
*base = normalize_base_url(base);
} else {
block.image_base = Some(self.image_base.clone());
}
@@ -219,8 +269,7 @@ impl RadioParadiseClient {
/// # }
/// ```
pub fn cover_url(&self, cover_path: &str) -> Result<Url> {
let url_str = format!("{}{}", self.image_base, cover_path);
Ok(Url::parse(&url_str)?)
resolve_cover_with_base(&self.image_base, cover_path)
}
/// Prefetch metadata for the next block
@@ -270,7 +319,8 @@ pub struct ClientBuilder {
image_base: String,
bitrate: Bitrate,
channel: u8,
timeout: Duration,
request_timeout: Duration,
block_timeout: Duration,
user_agent: String,
proxy: Option<String>,
}
@@ -284,7 +334,8 @@ impl Default for ClientBuilder {
image_base: DEFAULT_IMAGE_BASE.to_string(),
bitrate: Bitrate::default(),
channel: 0,
timeout: Duration::from_secs(DEFAULT_TIMEOUT_SECS),
request_timeout: Duration::from_secs(DEFAULT_REQUEST_TIMEOUT_SECS),
block_timeout: Duration::from_secs(DEFAULT_BLOCK_TIMEOUT_SECS),
user_agent: DEFAULT_USER_AGENT.to_string(),
proxy: None,
}
@@ -343,7 +394,13 @@ impl ClientBuilder {
/// Set the request timeout
pub fn timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
self.request_timeout = timeout;
self
}
/// Set the timeout specifically for block downloads/streams
pub fn block_timeout(mut self, timeout: Duration) -> Self {
self.block_timeout = timeout;
self
}
@@ -366,7 +423,7 @@ impl ClientBuilder {
} else {
let mut builder = Client::builder()
.user_agent(&self.user_agent)
.timeout(self.timeout);
.timeout(self.request_timeout);
if let Some(proxy_url) = &self.proxy {
let proxy = reqwest::Proxy::all(proxy_url)
@@ -382,15 +439,17 @@ impl ClientBuilder {
} else {
self.block_base.clone()
};
let image_base = normalize_base_url(&self.image_base);
Ok(RadioParadiseClient {
client,
api_base: self.api_base,
block_base,
image_base: self.image_base,
image_base,
bitrate: self.bitrate,
channel: self.channel,
timeout: self.timeout,
request_timeout: self.request_timeout,
block_timeout: self.block_timeout,
next_block_url: None,
})
}

7
pmoparadise/src/models.rs
View File

@@ -3,6 +3,7 @@
use serde::{Deserialize, Deserializer, Serialize};
use serde_json::Number;
use std::collections::HashMap;
use url::Url;
/// Deserialize a string or number into a u64
fn deserialize_string_or_u64<'de, D>(deserializer: D) -> Result<u64, D::Error>
@@ -306,9 +307,9 @@ impl Block {
/// Get the full URL for a cover image
pub fn cover_url(&self, cover_path: &str) -> Option<String> {
self.image_base
.as_ref()
.map(|base| format!("{}{}", base, cover_path))
let base = self.image_base.as_ref()?;
let base_url = Url::parse(base).ok()?;
base_url.join(cover_path).ok().map(|url| url.to_string())
}
/// Find which song is playing at a given timestamp (ms from block start)

203
pmoparadise/src/source.rs
View File

@@ -98,6 +98,39 @@ fn parse_track_identifier(track_id: &str) -> Option<(u8, u64, usize)> {
}
}
fn resolve_cover_url(
image_base: Option<&str>,
client: &RadioParadiseClient,
cover: &str,
) -> anyhow::Result<Url> {
if cover.starts_with("http://") || cover.starts_with("https://") {
return Url::parse(cover).map_err(|e| anyhow!("Invalid cover URL '{}': {}", cover, e));
}
if cover.starts_with("//") {
let url = format!("https:{}", cover);
return Url::parse(&url).map_err(|e| anyhow!("Invalid cover URL '{}': {}", cover, e));
}
if let Some(base) = image_base {
match Url::parse(base).and_then(|base_url| base_url.join(cover)) {
Ok(url) => return Ok(url),
Err(err) => {
tracing::debug!(
"Failed to join cover '{}' with image base '{}': {}",
cover,
base,
err
);
}
}
}
client
.cover_url(cover)
.map_err(|e| anyhow!("Invalid cover URL '{}': {}", cover, e))
}
/// Radio Paradise music source with full MusicSource trait implementation
///
/// This struct combines a [`RadioParadiseClient`] for API access with a FIFO playlist
@@ -348,6 +381,70 @@ impl RadioParadiseSource {
Ok(())
}
async fn prepare_initial_track(
&self,
channel: Arc<ChannelState>,
block: Arc<Block>,
) -> Result<()> {
let ordered_songs = block.songs_ordered();
let (song_index, song) = match ordered_songs.first() {
Some(entry) => entry,
None => return Ok(()),
};
let track_id = track_identifier(channel.descriptor.id, block.event, *song_index);
if channel.playlist.has_track(&track_id).await {
return Ok(());
}
let placeholder_uri = format!("{}#{}", block.url, *song_index);
channel
.cache_manager
.update_metadata(
track_id.clone(),
pmosource::TrackMetadata {
original_uri: placeholder_uri.clone(),
cached_audio_pk: None,
cached_cover_pk: None,
},
)
.await;
let mut track = Track::new(
track_id.clone(),
song.title.clone(),
placeholder_uri.clone(),
);
if !song.artist.is_empty() {
track = track.with_artist(song.artist.clone());
}
if let Some(ref album) = song.album {
if !album.is_empty() {
track = track.with_album(album.clone());
}
}
let duration_ms = song_duration_ms(&block, &ordered_songs, 0);
if duration_ms > 0 {
track = track.with_duration((duration_ms / 1000) as u32);
}
if let Some(ref cover) = song.cover {
if let Ok(url) = resolve_cover_url(block.image_base.as_deref(), &channel.client, cover)
{
track = track.with_image(url.to_string());
}
}
channel.playlist.append_track(track).await;
Ok(())
}
async fn populate_channel_locked(&self, channel: Arc<ChannelState>) -> Result<()> {
tracing::info!(
"📻 Fetching Radio Paradise block for channel {}",
@@ -361,7 +458,25 @@ impl RadioParadiseSource {
.map_err(|e| MusicSourceError::SourceUnavailable(e.to_string()))?;
let block = Arc::new(now_playing.block);
self.ingest_block(channel, block).await
self.prepare_initial_track(channel.clone(), block.clone())
.await?;
let source_clone = self.clone();
tokio::spawn(async move {
if let Err(e) = source_clone
.ingest_block(channel.clone(), block.clone())
.await
{
tracing::error!(
"Failed to ingest block {} on channel {}: {}",
block.event,
channel.descriptor.name,
e
);
}
});
Ok(())
}
async fn ingest_block(&self, channel: Arc<ChannelState>, block: Arc<Block>) -> Result<()> {
@@ -396,14 +511,25 @@ impl RadioParadiseSource {
for (position, (song_index, song)) in ordered_songs.iter().enumerate() {
let track_id = track_identifier(channel.descriptor.id, block.event, *song_index);
let placeholder_uri = format!("{}#{}", block.url, *song_index);
if channel
.cache_manager
.get_metadata(&track_id)
.await
.is_some()
{
continue;
let existing_metadata = channel.cache_manager.get_metadata(&track_id).await;
if let Some(ref metadata) = existing_metadata {
if metadata.cached_audio_pk.is_some() {
continue;
}
} else {
channel
.cache_manager
.update_metadata(
track_id.clone(),
pmosource::TrackMetadata {
original_uri: placeholder_uri.clone(),
cached_audio_pk: None,
cached_cover_pk: None,
},
)
.await;
}
let duration_ms = song_duration_ms(&block, &ordered_songs, position);
@@ -451,23 +577,34 @@ impl RadioParadiseSource {
.cache_audio_from_reader(&audio_source_uri, reader, Some(data_len))
.await?;
let cached_cover_pk = if let Some(ref image_base) = block.image_base {
if let Some(ref cover) = song.cover {
let image_url = format!("{}{}", image_base, cover);
match channel.cache_manager.cache_cover(&image_url).await {
Ok(pk) => Some(pk),
let resolved_cover_url =
song.cover.as_ref().and_then(|cover| {
match resolve_cover_url(block.image_base.as_deref(), &channel.client, cover) {
Ok(url) => Some(url.to_string()),
Err(e) => {
tracing::warn!(
"Failed to cache cover {} on channel {}: {}",
image_url,
"Failed to resolve cover '{}' for channel {}: {}",
cover,
channel.descriptor.name,
e
);
None
}
}
} else {
None
});
let cached_cover_pk = if let Some(ref cover_url) = resolved_cover_url {
match channel.cache_manager.cache_cover(cover_url).await {
Ok(pk) => Some(pk),
Err(e) => {
tracing::warn!(
"Failed to cache cover {} on channel {}: {}",
cover_url,
channel.descriptor.name,
e
);
None
}
}
} else {
None
@@ -479,7 +616,15 @@ impl RadioParadiseSource {
.update_metadata(
track_id.clone(),
pmosource::TrackMetadata {
original_uri: block.url.clone(),
original_uri: existing_metadata
.and_then(|m| {
if m.original_uri.is_empty() {
None
} else {
Some(m.original_uri)
}
})
.unwrap_or_else(|| placeholder_uri.clone()),
cached_audio_pk: Some(audio_pk.clone()),
cached_cover_pk: metadata_cover_pk,
},
@@ -506,13 +651,25 @@ impl RadioParadiseSource {
if let Ok(url) = channel.cache_manager.cover_url(cover_pk, None) {
track = track.with_image(url);
}
} else if let Some(ref cover) = song.cover {
if let Some(ref image_base) = block.image_base {
track = track.with_image(format!("{}{}", image_base, cover));
}
} else if let Some(ref cover_url) = resolved_cover_url {
track = track.with_image(cover_url.clone());
}
channel.playlist.append_track(track).await;
let updated = channel
.playlist
.update_track(&track_id, |existing| {
existing.title = track.title.clone();
existing.artist = track.artist.clone();
existing.album = track.album.clone();
existing.duration = track.duration;
existing.uri = track.uri.clone();
existing.image = track.image.clone();
})
.await;
if !updated {
channel.playlist.append_track(track).await;
}
let channel_for_wait = channel.clone();
let track_id_for_wait = track_id.clone();

4
pmoparadise/src/stream.rs
View File

@@ -73,7 +73,7 @@ impl RadioParadiseClient {
let response = self
.client
.get(block_url.clone())
.timeout(self.timeout)
.timeout(self.block_timeout)
.send()
.await?;
@@ -151,7 +151,7 @@ impl RadioParadiseClient {
let response = self
.client
.get(block_url.clone())
.timeout(self.timeout)
.timeout(self.block_timeout)
.send()
.await?;