euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-20 14:52:22 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

feat: pipe remote ffprobe probes through reqwest (#730)

Browse Source 
pipe remote ffprobe probes through reqwest
This commit is contained in:
knylbyte
2026-04-24 09:59:06 +02:00
committed by GitHub
parent 5d793e0ac0
commit b51fc105a4
7 changed files with 836 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/playlist.rs
+2 -1
View File
@@ -27,7 +27,8 @@ use crate::{
PlaylistSource,
},
utils::{
debug_if_enabled, epg, log_memory_snapshot, m3u, trace_if_enabled, xtream, StepMeasure, StepMeasureCallback,
debug_if_enabled, epg, log_memory_snapshot, m3u, trace_if_enabled, xtream,
StepMeasure, StepMeasureCallback,
},
};
use futures::{FutureExt, StreamExt};
backend/src/processing/processor/stream_probe.rs
+37 -12
View File
@@ -4,8 +4,9 @@ use crate::model::{AppConfig};
use crate::processing::processor::{select_cancel_token, ProbeHandleGuard};
use crate::repository::{get_input_m3u_playlist_file_path, get_input_storage_path, get_input_local_library_playlist_file_path, xtream_get_file_path, BPlusTreeUpdate};
use crate::utils::debug_if_enabled;
use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeStreamStats, ProbeUrlOutcome};
use log::{info, warn};
use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeStreamStats, ProbeUrlOutcome};
use shared::utils::sanitize_sensitive_info;
use log::{debug, info, warn};
use shared::error::TuliproxError;
use shared::model::{EpisodeStreamProperties, InputType, PlaylistItemType, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, LiveStreamProperties, M3uPlaylistItem, XtreamCluster, XtreamPlaylistItem};
use std::sync::Arc;
@@ -34,6 +35,7 @@ fn requires_provider_connection_for_generic_probe(input_type: InputType) -> bool
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub async fn update_generic_stream_metadata(
app_config: &Arc<AppConfig>,
client: &reqwest::Client,
input: &ConfigInput,
unique_id: &str,
stream_url: &str,
@@ -106,6 +108,14 @@ pub async fn update_generic_stream_metadata(
}
let probe_url = input.resolve_url(stream_url)?.into_owned();
if !is_supported_probe_url(&probe_url) {
let safe_probe_url = sanitize_sensitive_info(&probe_url).into_owned();
debug!("Skipping unsupported generic stream probe for {unique_id}: {safe_probe_url}");
return Ok(GenericProbeOutcome::Noop);
}
let is_remote_probe = reqwest::Url::parse(&probe_url)
.ok()
.is_some_and(|url| matches!(url.scheme(), "http" | "https"));
let config = app_config.config.load();
let metadata_update = config.metadata_update.clone().unwrap_or_default();
let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60);
@@ -128,16 +138,31 @@ pub async fn update_generic_stream_metadata(
acquired_handle.as_ref().and_then(ProbeHandleGuard::handle),
active_handle,
);
let probe_data = FfmpegExecutor::new().probe_url_with_cancel(
&probe_url,
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
config.proxy.as_ref(),
cancel_token,
)
.await;
let probe_data = if is_remote_probe {
FfmpegExecutor::new()
.probe_remote_url_with_cancel(
client,
&probe_url,
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
cancel_token,
)
.await
} else {
FfmpegExecutor::new()
.probe_url_with_cancel(
&probe_url,
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
config.proxy.as_ref(),
cancel_token,
)
.await
};
if let Some(handle) = acquired_handle {
handle.release().await;
backend/src/processing/processor/xtream.rs
+36 -12
View File
@@ -5,15 +5,16 @@ use crate::model::{AppConfig, ConfigInput, ConfigInputFlags};
use shared::model::{LiveStreamProperties, StreamProperties, XtreamCluster, XtreamPlaylistItem};
use crate::repository::{get_input_storage_path, persist_input_live_info, BPlusTreeQuery, xtream_get_file_path};
use crate::utils::{debug_if_enabled};
use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome};
use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome};
use log::{debug, warn};
use crate::processing::parser::xtream::create_xtream_url;
use crate::api::model::{ActiveProviderManager, ProviderHandle, ProviderIdType};
/// Updates metadata for a single Live stream (primarily probing)
#[allow(clippy::too_many_lines)]
#[allow(clippy::too_many_arguments, clippy::too_many_lines)]
pub async fn update_live_stream_metadata(
app_config: &Arc<AppConfig>,
client: &reqwest::Client,
input: &ConfigInput,
id: ProviderIdType,
save: bool,
@@ -117,6 +118,14 @@ pub async fn update_live_stream_metadata(
use_prefix, no_ext
);
let probe_url_cow = input.resolve_url(&stream_url)?;
if !is_supported_probe_url(probe_url_cow.as_ref()) {
debug!(
"Skipping unsupported live probe for input {}: {}",
input.name,
probe_url_cow.as_ref()
);
return Ok(None);
}
let config = app_config.config.load();
let metadata_update = config.metadata_update.clone().unwrap_or_default();
let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60);
@@ -134,16 +143,31 @@ pub async fn update_live_stream_metadata(
let mut success = false;
let mut not_found = false;
match FfmpegExecutor::new().probe_url(
probe_url_cow.as_ref(),
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
config.proxy.as_ref(),
)
.await
{
let is_remote_probe = reqwest::Url::parse(probe_url_cow.as_ref())
.ok()
.is_some_and(|u| matches!(u.scheme(), "http" | "https"));
let probe_result = if is_remote_probe {
FfmpegExecutor::new().probe_remote_url(
client,
probe_url_cow.as_ref(),
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
)
.await
} else {
FfmpegExecutor::new().probe_url(
probe_url_cow.as_ref(),
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
config.proxy.as_ref(),
)
.await
};
match probe_result {
ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, _stats) => {
// 3. Update properties on success
if let Some(v) = raw_video {
backend/src/processing/processor/xtream_series.rs
+68 -12
View File
@@ -19,7 +19,7 @@ use crate::repository::{
get_input_storage_path, persist_input_series_info_batch, MemoryPlaylistSource, PlaylistSource,
};
use crate::repository::{xtream_get_file_path, BPlusTreeQuery};
use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome};
use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome};
use crate::utils::{debug_if_enabled, xtream};
use log::{debug, error, info, log_enabled, trace, warn, Level};
use parking_lot::Mutex;
@@ -928,6 +928,46 @@ pub async fn update_series_metadata(
if active_handle.is_some() || temp_handle.is_some() {
let episode_url =
create_xtream_series_episode_url(input_url, input_username, input_password, ep);
let probe_url = match input.resolve_url(&episode_url) {
Ok(url) => url,
Err(err) => {
if matches!(err, shared::error::TuliproxError::ConfigInput(_)) {
debug!(
"Provider config resolution failed for series episode '{}' (S{}E{}): {err}",
ep.title, ep.season, ep.episode_num
);
if probe_failure.is_none() {
probe_failure = Some(ProbeFailureKind::Other);
}
} else {
warn!(
"Failed to resolve probe URL for series episode '{}' (S{}E{}): {err}",
ep.title, ep.season, ep.episode_num
);
probe_pending = true;
if probe_failure.is_none() {
probe_failure = Some(ProbeFailureKind::Other);
}
}
if let Some(guard) = temp_handle {
guard.release().await;
}
continue;
}
};
if !is_supported_probe_url(probe_url.as_ref()) {
debug!(
"Skipping unsupported series probe for episode '{}' (S{}E{}): {}",
ep.title,
ep.season,
ep.episode_num,
probe_url.as_ref()
);
if let Some(guard) = temp_handle {
guard.release().await;
}
continue;
}
// Specific logging for the user to follow
let missing_reason = if missing_video && missing_audio {
@@ -946,17 +986,33 @@ pub async fn update_series_metadata(
temp_handle.as_ref().and_then(ProbeHandleGuard::handle),
active_handle,
);
match FfmpegExecutor::new().probe_url_with_cancel(
&episode_url,
user_agent.as_deref(),
probe_settings.analyze_duration_micros,
probe_settings.probe_size_bytes,
probe_settings.timeout_secs,
config.proxy.as_ref(),
cancel_token,
)
.await
{
let is_remote_probe = reqwest::Url::parse(probe_url.as_ref())
.ok()
.is_some_and(|u| matches!(u.scheme(), "http" | "https"));
let probe_result = if is_remote_probe {
FfmpegExecutor::new().probe_remote_url_with_cancel(
client,
probe_url.as_ref(),
user_agent.as_deref(),
probe_settings.analyze_duration_micros,
probe_settings.probe_size_bytes,
probe_settings.timeout_secs,
cancel_token,
)
.await
} else {
FfmpegExecutor::new().probe_url_with_cancel(
probe_url.as_ref(),
user_agent.as_deref(),
probe_settings.analyze_duration_micros,
probe_settings.probe_size_bytes,
probe_settings.timeout_secs,
config.proxy.as_ref(),
cancel_token,
)
.await
};
match probe_result {
ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, _stats) => {
if let Some(v) = raw_video {
ep.video = Some(v.to_string().into());
backend/src/processing/processor/xtream_vod.rs
+107 -69
View File
@@ -17,7 +17,7 @@ use crate::ptt::ptt_parse_title;
use crate::repository::persist_input_vod_info;
use crate::repository::persist_input_vod_info_batch;
use crate::repository::{xtream_get_file_path, BPlusTreeQuery};
use crate::utils::ffmpeg::{FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome};
use crate::utils::ffmpeg::{is_supported_probe_url, FfmpegExecutor, ProbeFailureKind, ProbeUrlOutcome};
use crate::utils::{debug_if_enabled, trace_if_enabled, xtream};
use log::{debug, error, info, log_enabled, trace, warn, Level};
use parking_lot::Mutex;
@@ -837,86 +837,124 @@ pub async fn update_vod_metadata(
true,
true,
);
let probe_url = input.resolve_url(&stream_url).map_err(|err| {
if matches!(err, shared::error::TuliproxError::ConfigInput(_)) {
shared::error::TuliproxError::ConfigInput(format!(
"Provider config resolution failed for VOD probe URL '{stream_url}': {err}"
))
} else {
shared::error::TuliproxError::Config(format!(
"Failed to resolve probe URL for VOD {display_id}: {err}"
))
}
})?;
let config = app_config.config.load();
let metadata_update = config.metadata_update.clone().unwrap_or_default();
let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60);
let user_agent = config.default_user_agent.clone();
let analyze_duration = metadata_update.ffprobe.analyze_duration_micros;
let probe_size = metadata_update.ffprobe.probe_size_bytes;
if is_supported_probe_url(probe_url.as_ref()) {
let probe_priority = config.metadata_update.as_ref().map_or(default_probe_user_priority(), |cfg| cfg.probe.user_priority);
let config = app_config.config.load();
let metadata_update = config.metadata_update.clone().unwrap_or_default();
let ffprobe_timeout = metadata_update.ffprobe.timeout.unwrap_or(60);
let user_agent = config.default_user_agent.clone();
let analyze_duration = metadata_update.ffprobe.analyze_duration_micros;
let probe_size = metadata_update.ffprobe.probe_size_bytes;
// Acquire Connection logic
let temp_handle = if active_handle.is_some() {
None
} else {
active_provider
.acquire_connection_for_probe(&input.name, probe_priority)
.await
.map(|handle| ProbeHandleGuard::new(active_provider, handle))
};
let probe_priority = config.metadata_update.as_ref().map_or(default_probe_user_priority(), |cfg| cfg.probe.user_priority);
if active_handle.is_some() || temp_handle.is_some() {
debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id);
let cancel_token = select_cancel_token(
temp_handle.as_ref().and_then(ProbeHandleGuard::handle),
active_handle,
);
match FfmpegExecutor::new().probe_url_with_cancel(
&stream_url,
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
config.proxy.as_ref(),
cancel_token,
)
.await
{
ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, stats) => {
if let Some(details) = properties.details.as_mut() {
if let Some(v) = raw_video {
details.video = Some(v.to_string().into());
properties_updated = true;
// Acquire Connection logic
let temp_handle = if active_handle.is_some() {
None
} else {
active_provider
.acquire_connection_for_probe(&input.name, probe_priority)
.await
.map(|handle| ProbeHandleGuard::new(active_provider, handle))
};
if active_handle.is_some() || temp_handle.is_some() {
debug_if_enabled!("Probing VOD '{}' (ID: {})", display_title, display_id);
let cancel_token = select_cancel_token(
temp_handle.as_ref().and_then(ProbeHandleGuard::handle),
active_handle,
);
let is_remote_probe = reqwest::Url::parse(probe_url.as_ref())
.ok()
.is_some_and(|u| matches!(u.scheme(), "http" | "https"));
let probe_result = if is_remote_probe {
FfmpegExecutor::new().probe_remote_url_with_cancel(
client,
probe_url.as_ref(),
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
cancel_token,
)
.await
} else {
FfmpegExecutor::new().probe_url_with_cancel(
probe_url.as_ref(),
user_agent.as_deref(),
analyze_duration,
probe_size,
ffprobe_timeout,
config.proxy.as_ref(),
cancel_token,
)
.await
};
match probe_result {
ProbeUrlOutcome::Success(_quality, raw_video, raw_audio, stats) => {
if let Some(details) = properties.details.as_mut() {
if let Some(v) = raw_video {
details.video = Some(v.to_string().into());
properties_updated = true;
}
if let Some(a) = raw_audio {
details.audio = Some(a.to_string().into());
properties_updated = true;
}
if let Some(duration_secs) = stats.duration_secs {
details.duration_secs = Some(duration_secs.to_string().into());
properties_updated = true;
}
if let Some(bitrate) = stats.bitrate {
details.bitrate = bitrate;
properties_updated = true;
}
}
if let Some(a) = raw_audio {
details.audio = Some(a.to_string().into());
properties_updated = true;
}
ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => {
probe_failure = Some(ProbeFailureKind::NotFound);
}
ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => {
probe_pending = true;
if probe_failure.is_none() {
probe_failure = Some(ProbeFailureKind::Other);
}
if let Some(duration_secs) = stats.duration_secs {
details.duration_secs = Some(duration_secs.to_string().into());
properties_updated = true;
}
if let Some(bitrate) = stats.bitrate {
details.bitrate = bitrate;
properties_updated = true;
}
ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => {
probe_pending = true;
if probe_failure.is_none() {
probe_failure = Some(ProbeFailureKind::Cancelled);
}
}
}
ProbeUrlOutcome::Failed(ProbeFailureKind::NotFound) => {
probe_failure = Some(ProbeFailureKind::NotFound);
}
ProbeUrlOutcome::Failed(ProbeFailureKind::Other) => {
probe_pending = true;
if probe_failure.is_none() {
probe_failure = Some(ProbeFailureKind::Other);
}
}
ProbeUrlOutcome::Failed(ProbeFailureKind::Cancelled) => {
probe_pending = true;
if probe_failure.is_none() {
probe_failure = Some(ProbeFailureKind::Cancelled);
}
}
} else {
probe_pending = true;
warn!("Skipping probe for VOD {display_id} due to connection limits");
}
if let Some(guard) = temp_handle {
guard.release().await;
}
} else {
probe_pending = true;
warn!("Skipping probe for VOD {display_id} due to connection limits");
}
if let Some(guard) = temp_handle {
guard.release().await;
debug_if_enabled!(
"Skipping unsupported VOD probe for '{}' (ID: {}): {}",
display_title,
display_id,
probe_url.as_ref()
);
}
}
}