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

Fix/peristed m3u file invalid provider url (#595)

Browse Source 
Fixed m3u playlist with  provider_config persistence as text file
Added probe_delay input option (default 2s); stream probing renamed “Stream Probing” and now covers live, VOD and series as a low‑priority background task.
This commit is contained in:
euzu
2026-02-18 08:18:28 +01:00
committed by GitHub
parent d2506dabba
commit 815487e4f5
35 changed files with 1696 additions and 760 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/playlist.rs
+18 -10
View File
@@ -48,13 +48,11 @@ use shared::model::{
};
use shared::model::{InputStats, PlaylistStats, SourceStats, TargetStats, UUIDType};
use shared::utils::{
create_alias_uuid, default_as_default, default_probe_live_interval, default_resolve_delay_secs, interner_gc,
create_alias_uuid, default_as_default, default_probe_delay_secs, default_probe_live_interval, interner_gc,
Internable,
};
use std::time::Instant;
const DEFAULT_PROBE_STREAM_DELAY: u16 = 50;
fn is_valid(pli: &PlaylistItem, filter: &Filter, match_as_ascii: bool) -> bool {
let provider = ValueProvider { pli, match_as_ascii };
filter.filter(&provider)
@@ -955,11 +953,11 @@ fn get_live_probe_interval_settings(
return None;
}
target.get_xtream_output().map(|_| {
let (resolve_delay, input_probe_live_interval_hours) = input_options.map_or(
(default_resolve_delay_secs(), default_probe_live_interval()),
|options| (options.resolve_delay, options.probe_live_interval_hours),
let (probe_delay, input_probe_live_interval_hours) = input_options.map_or(
(default_probe_delay_secs(), default_probe_live_interval()),
|options| (options.probe_delay, options.probe_live_interval_hours),
);
(resolve_delay, u64::from(input_probe_live_interval_hours) * 3600)
(probe_delay, u64::from(input_probe_live_interval_hours) * 3600)
})
}
@@ -992,6 +990,7 @@ fn provider_id_from_item(item: &PlaylistItem) -> Option<ProviderIdType> {
}
}
#[allow(clippy::too_many_lines)]
async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget, fpl: &mut FetchedPlaylist<'_>) {
let Some(mgr) = ctx.metadata_manager.as_ref() else {
return;
@@ -1012,6 +1011,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
let input_name = fpl.input.name.clone();
let input_type = fpl.input.input_type;
let xtream_probe_handled = input_type.is_xtream() && target.get_xtream_output().is_some();
let live_probe_settings = if probe_live_enabled {
get_live_probe_interval_settings(target, input_type, Some(opts)).map(|(delay, interval_secs)| {
let interval_signed = i64::try_from(interval_secs).unwrap_or(i64::MAX);
@@ -1038,14 +1038,14 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
continue;
}
if let Some((resolve_delay, interval_secs, cutoff_ts)) = live_probe_settings {
if let Some((probe_delay, interval_secs, cutoff_ts)) = live_probe_settings {
if needs_live_probe(&item, cutoff_ts) {
if let Some(provider_id) = provider_id_from_item(&item) {
if queued_live_keys.insert(provider_id.clone()) {
let task = UpdateTask::ProbeLive {
id: provider_id,
reason: ResolveReasonSet::from_variants(&[ResolveReason::Probe]),
delay: resolve_delay,
delay: probe_delay,
interval: interval_secs,
};
mgr.queue_task_background(input_name.clone(), task);
@@ -1062,11 +1062,19 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
if !probe_vod_enabled {
continue;
}
// Xtream outputs handle VOD probe as part of the resolve pipeline (after resolve).
if xtream_probe_handled {
continue;
}
}
PlaylistItemType::Series | PlaylistItemType::LocalSeries => {
if !probe_series_enabled {
continue;
}
// Xtream outputs handle Series probe as part of the resolve pipeline (after resolve).
if xtream_probe_handled {
continue;
}
}
_ => continue,
}
@@ -1094,7 +1102,7 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
url: item.header.url.to_string(),
item_type: item.header.item_type,
reason: ResolveReasonSet::from_variants(&[ResolveReason::MissingDetails]),
delay: DEFAULT_PROBE_STREAM_DELAY,
delay: opts.probe_delay,
};
mgr.queue_task_background(input_name.clone(), task);
queued_stream_count += 1;
backend/src/processing/processor/xtream.rs
+43 -11
View File
@@ -1,3 +1,4 @@
use parking_lot::Mutex;
use std::sync::Arc;
use shared::error::{TuliproxError};
use crate::model::{AppConfig, ConfigInput, ConfigInputFlags};
@@ -9,12 +10,13 @@ use crate::processing::parser::xtream::create_xtream_url;
use crate::api::model::ProviderIdType;
/// Updates metadata for a single Live stream (primarily probing)
#[allow(clippy::too_many_lines)]
pub async fn update_live_stream_metadata(
app_config: &Arc<AppConfig>,
input: &ConfigInput,
id: ProviderIdType,
save: bool,
db_query: Option<&mut BPlusTreeQuery<u32, XtreamPlaylistItem>>,
db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
) -> Result<Option<LiveStreamProperties>, TuliproxError> {
let working_dir = &app_config.config.load().working_dir;
let storage_path = get_input_storage_path(&input.name, working_dir).await
@@ -27,24 +29,54 @@ pub async fn update_live_stream_metadata(
let stream_id_opt = if let ProviderIdType::Id(vid) = id { Some(vid) } else { None };
if let Some(stream_id) = stream_id_opt {
// Use provided query or open new one
if let Some(query) = db_query {
if let Ok(Some(item)) = query.query_zero_copy(&stream_id) {
let query = Arc::clone(&query);
let item = match tokio::task::spawn_blocking(move || {
let mut guard = query.lock();
guard.query_zero_copy(&stream_id).ok().flatten()
})
.await
{
Ok(item) => item,
Err(err) => {
warn!("Failed to query Live metadata from disk for {stream_id}: {err}");
None
}
};
if let Some(item) = item {
existing_item = Some(item.clone());
if let Some(StreamProperties::Live(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
props = Some(*p.clone());
}
}
} else {
let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Live);
if xtream_path.exists() {
let _file_lock = app_config.file_locks.read_lock(&xtream_path).await;
if let Ok(mut query) = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path) {
if let Ok(Some(item)) = query.query_zero_copy(&stream_id) {
existing_item = Some(item.clone());
if let Some(StreamProperties::Live(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
}
let file_lock = app_config.file_locks.read_lock(&xtream_path).await;
let xtream_path = xtream_path.clone();
let item = match tokio::task::spawn_blocking(move || {
let _guard = file_lock;
let mut query = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)?;
query.query_zero_copy(&stream_id)
})
.await
{
Ok(Ok(item)) => item,
Ok(Err(err)) => {
warn!("Failed to query Live metadata from disk for {stream_id}: {err}");
None
}
Err(err) => {
warn!("Failed to query Live metadata from disk for {stream_id}: {err}");
None
}
};
if let Some(item) = item {
existing_item = Some(item.clone());
if let Some(StreamProperties::Live(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
}
}
}
backend/src/processing/processor/xtream_series.rs
+94 -33
View File
@@ -21,6 +21,7 @@ use shared::error::TuliproxError;
use shared::model::{MediaQuality, PlaylistEntry, PlaylistItem, SeriesStreamProperties, StreamProperties, XtreamPlaylistItem, XtreamSeriesInfo};
use shared::model::{PlaylistGroup, PlaylistItemType, XtreamCluster};
use std::collections::HashMap;
use parking_lot::Mutex;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -247,7 +248,7 @@ async fn process_immediate_series_info(
// Keep an optional read query open and reopen lazily only when needed.
let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Series);
let mut db_query_holder = None;
let mut db_query_holder: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>> = None;
let mut _db_lock_holder = None;
let mut groups_to_add = Vec::new();
@@ -272,14 +273,31 @@ async fn process_immediate_series_info(
if !reasons.is_empty() {
if let Some(active_provider) = ctx.provider_manager.as_ref() {
if db_query_holder.is_none() && xtream_path.exists() {
let lock = ctx.config.file_locks.read_lock(&xtream_path).await;
if let Ok(query) = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path) {
db_query_holder = Some(query);
_db_lock_holder = Some(lock);
let file_lock = ctx.config.file_locks.read_lock(&xtream_path).await;
let xtream_path = xtream_path.clone();
let query = match tokio::task::spawn_blocking(move || {
let guard = file_lock;
let query = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)?;
Ok::<_, std::io::Error>((query, guard))
}).await {
Ok(Ok((query, guard))) => Some((query, guard)),
Ok(Err(err)) => {
error!("Failed to open BPlusTreeQuery for Series: {err}");
None
}
Err(err) => {
error!("Failed to open BPlusTreeQuery for Series: {err}");
None
}
};
if let Some((query, guard)) = query {
db_query_holder = Some(Arc::new(Mutex::new(query)));
_db_lock_holder = Some(guard);
}
}
// Pass the optional reference to the query
let db_query_ref = db_query_holder.as_mut();
let db_query_ref = db_query_holder.as_ref().map(Arc::clone);
match update_series_info_immediate(ctx, active_provider, input, pli, provider_id.clone(), &reasons, db_query_ref).await {
Ok(Some(updated_props)) => {
@@ -401,7 +419,7 @@ async fn update_series_info_immediate(
pli: &PlaylistItem,
id: ProviderIdType,
reasons: &ResolveReasonSet,
db_query: Option<&mut BPlusTreeQuery<u32, XtreamPlaylistItem>>,
db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
) -> Result<Option<SeriesStreamProperties>, TuliproxError> {
let fetch_info = reasons.contains(ResolveReason::Info);
@@ -431,7 +449,7 @@ fn check_resolve_reasons(
check_resolve_tmdb(resolve_options, resolve_tmdb_enabled, pli, needs_info, &mut reasons);
if resolve_options.has_flag(ResolveOptionsFlags::Probe) {
check_needs_probe(pli, needs_info, &mut reasons);
check_needs_probe(pli, &mut reasons);
}
reasons
}
@@ -467,23 +485,24 @@ fn check_resolve_tmdb(
}
}
fn check_needs_probe(pli: &mut PlaylistItem, needs_info: bool, reasons: &mut ResolveReasonSet) {
if !needs_info {
return;
}
if let Some(StreamProperties::Series(series_stream_props)) = pli.header.additional_properties.as_ref() {
if let Some(details) = series_stream_props.details.as_ref() {
if let Some(episodes) = details.episodes.as_ref() {
for ep in episodes {
fn check_needs_probe(pli: &PlaylistItem, reasons: &mut ResolveReasonSet) {
let needs_probe = match pli.header.additional_properties.as_ref() {
Some(StreamProperties::Series(series_stream_props)) => series_stream_props
.details
.as_ref()
.and_then(|d| d.episodes.as_ref())
.is_some_and(|episodes| {
episodes.iter().any(|ep| {
let missing_video = !MediaQuality::is_valid_media_info(ep.video.as_deref());
let missing_audio = !MediaQuality::is_valid_media_info(ep.audio.as_deref());
if missing_video || missing_audio {
reasons.set(ResolveReason::Probe);
break;
}
}
}
}
missing_video || missing_audio
})
}),
_ => false,
};
if needs_probe {
reasons.set(ResolveReason::Probe);
}
}
@@ -506,7 +525,7 @@ pub async fn update_series_metadata(
save: bool,
fetch_info: bool,
do_probe: bool,
db_query: Option<&mut BPlusTreeQuery<u32, XtreamPlaylistItem>>,
db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
) -> Result<Option<SeriesStreamProperties>, TuliproxError> {
let working_dir = &app_config.config.load().working_dir;
let storage_path = get_input_storage_path(&input.name, working_dir).await
@@ -523,9 +542,26 @@ pub async fn update_series_metadata(
let series_id_opt = if let ProviderIdType::Id(vid) = id { Some(vid) } else { None };
if let Some(series_id) = series_id_opt {
// Use provided query or open new one
if let Some(query) = db_query {
if let Ok(Some(item)) = query.query_zero_copy(&series_id) {
let query = Arc::clone(&query);
let item = match tokio::task::spawn_blocking(move || {
let mut guard = query.lock();
guard.query_zero_copy(&series_id)
})
.await
{
Ok(Ok(item)) => item,
Ok(Err(err)) => {
error!("Failed to query Series metadata from disk for {series_id}: {err}");
None
}
Err(err) => {
error!("Failed to query Series metadata from disk for {series_id}: {err}");
None
}
};
if let Some(item) = item {
existing_item = Some(item.clone());
if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
@@ -534,13 +570,30 @@ pub async fn update_series_metadata(
} else {
let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Series);
if xtream_path.exists() {
let _file_lock = app_config.file_locks.read_lock(&xtream_path).await;
if let Ok(mut query) = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path) {
if let Ok(Some(item)) = query.query_zero_copy(&series_id) {
existing_item = Some(item.clone());
if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
}
let file_lock = app_config.file_locks.read_lock(&xtream_path).await;
let xtream_path = xtream_path.clone();
let item = match tokio::task::spawn_blocking(move || {
let _guard = file_lock;
let mut query = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)?;
query.query_zero_copy(&series_id)
})
.await
{
Ok(Ok(item)) => item,
Ok(Err(err)) => {
error!("Failed to query Series metadata from disk for {series_id}: {err}");
None
}
Err(err) => {
error!("Failed to query Series metadata from disk for {series_id}: {err}");
None
}
};
if let Some(item) = item {
existing_item = Some(item.clone());
if let Some(StreamProperties::Series(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
}
}
}
@@ -676,12 +729,14 @@ pub async fn update_series_metadata(
let input_password = input.password.as_deref().unwrap_or("");
let mut probed_count = 0;
let mut missing_any = false;
for ep in episodes {
let missing_video = !MediaQuality::is_valid_media_info(ep.video.as_deref());
let missing_audio = !MediaQuality::is_valid_media_info(ep.audio.as_deref());
if missing_video || missing_audio {
missing_any = true;
// Acquire Connection logic
let temp_handle = if active_handle.is_some() {
@@ -725,8 +780,14 @@ pub async fn update_series_metadata(
}
if probed_count > 0 {
info!("Probed {probed_count} episodes for Series ID {display_id}");
} else if !missing_any {
debug!("Series probe skipped for ID {display_id} (all episodes already have A/V details)");
}
} else {
debug!("Series probe skipped for ID {display_id} (no episode details available)");
}
} else {
debug!("Series probe skipped for ID {display_id} (no series details available)");
}
}
backend/src/processing/processor/xtream_vod.rs
+64 -21
View File
@@ -18,6 +18,7 @@ use serde_json::Value;
use shared::error::TuliproxError;
use shared::model::{MediaQuality, PlaylistEntry, PlaylistItem, PlaylistItemType, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster, XtreamPlaylistItem, XtreamVideoInfo};
use std::collections::HashMap;
use parking_lot::Mutex;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -180,7 +181,7 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F
// Keep an optional read query open and reopen lazily only when needed.
let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Video);
let mut db_query_holder = None;
let mut db_query_holder: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>> = None;
let mut _db_lock_holder = None;
let mut batch: Vec<(ProviderIdType, VideoStreamProperties)> = Vec::with_capacity(BATCH_SIZE);
@@ -205,13 +206,28 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F
if let Some(prov_mgr) = ctx.provider_manager.as_ref() {
if db_query_holder.is_none() && xtream_path.exists() {
let lock = ctx.config.file_locks.read_lock(&xtream_path).await;
if let Ok(query) = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path) {
db_query_holder = Some(query);
let xtream_path = xtream_path.clone();
let query = match tokio::task::spawn_blocking(move || {
BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)
}).await {
Ok(Ok(query)) => Some(query),
Ok(Err(err)) => {
error!("Failed to open BPlusTreeQuery for VOD: {err}");
None
}
Err(err) => {
error!("Failed to open BPlusTreeQuery for VOD: {err}");
None
}
};
if let Some(query) = query {
db_query_holder = Some(Arc::new(Mutex::new(query)));
_db_lock_holder = Some(lock);
}
}
// Pass the optional reference to the query
let db_query_ref = db_query_holder.as_mut();
let db_query_ref = db_query_holder.as_ref().map(Arc::clone);
match update_vod_info_immediate(ctx, prov_mgr, input, pli, provider_id.clone(), &reasons, db_query_ref).await {
Ok(Some(updated_props)) => {
@@ -311,17 +327,14 @@ fn check_resolve_reasons(
// TMDB check
check_resolve_tmdb(resolve_options, resolve_tmdb_enabled, pli, needs_info, &mut reasons);
// Probe check
// Probe check (independent of resolve info fetch; we still probe when A/V details are missing)
if do_probe {
check_needs_probe(pli, needs_info, &mut reasons);
check_needs_probe(pli, &mut reasons);
}
reasons
}
fn check_needs_probe(pli: &PlaylistItem, needs_info: bool, reasons: &mut ResolveReasonSet) {
if !needs_info {
return;
}
fn check_needs_probe(pli: &PlaylistItem, reasons: &mut ResolveReasonSet) {
let needs_probe = match pli.header.additional_properties.as_ref() {
Some(StreamProperties::Video(props)) => {
let details = props.details.as_ref();
@@ -410,7 +423,7 @@ async fn update_vod_info_immediate(
pli: &PlaylistItem,
id: ProviderIdType,
reasons: &ResolveReasonSet,
db_query: Option<&mut BPlusTreeQuery<u32, XtreamPlaylistItem>>,
db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
) -> Result<Option<VideoStreamProperties>, TuliproxError> {
let fetch_info = reasons.contains(ResolveReason::Info);
@@ -449,7 +462,7 @@ pub async fn update_vod_metadata(
save: bool,
fetch_info: bool,
do_probe: bool,
db_query: Option<&mut BPlusTreeQuery<u32, XtreamPlaylistItem>>,
db_query: Option<Arc<Mutex<BPlusTreeQuery<u32, XtreamPlaylistItem>>>>,
) -> Result<Option<VideoStreamProperties>, TuliproxError> {
let working_dir = &app_config.config.load().working_dir;
let storage_path = get_input_storage_path(&input.name, working_dir).await
@@ -467,9 +480,22 @@ pub async fn update_vod_metadata(
let stream_id_opt = if let ProviderIdType::Id(vid) = id { Some(vid) } else { None };
if let Some(stream_id) = stream_id_opt {
// Use provided query or open new one
if let Some(query) = db_query {
if let Ok(Some(item)) = query.query_zero_copy(&stream_id) {
let query = Arc::clone(&query);
let item = match tokio::task::spawn_blocking(move || {
let mut guard = query.lock();
guard.query_zero_copy(&stream_id).ok().flatten()
})
.await
{
Ok(item) => item,
Err(err) => {
error!("Failed to query VOD metadata from disk for {stream_id}: {err}");
None
}
};
if let Some(item) = item {
existing_item = Some(item.clone());
if let Some(StreamProperties::Video(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
@@ -478,13 +504,30 @@ pub async fn update_vod_metadata(
} else {
let xtream_path = xtream_get_file_path(&storage_path, XtreamCluster::Video);
if xtream_path.exists() {
let _file_lock = app_config.file_locks.read_lock(&xtream_path).await;
if let Ok(mut query) = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path) {
if let Ok(Some(item)) = query.query_zero_copy(&stream_id) {
existing_item = Some(item.clone());
if let Some(StreamProperties::Video(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
}
let file_lock = app_config.file_locks.read_lock(&xtream_path).await;
let xtream_path = xtream_path.clone();
let item = match tokio::task::spawn_blocking(move || {
let _guard = file_lock;
let mut query = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)?;
query.query_zero_copy(&stream_id)
})
.await
{
Ok(Ok(item)) => item,
Ok(Err(err)) => {
error!("Failed to query VOD metadata from disk for {stream_id}: {err}");
None
}
Err(err) => {
error!("Failed to query VOD metadata from disk for {stream_id}: {err}");
None
}
};
if let Some(item) = item {
existing_item = Some(item.clone());
if let Some(StreamProperties::Video(p)) = item.additional_properties.as_ref() {
props = Some(*p.clone());
}
}
}