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

Feature/filter resolve probe (#720)

Browse Source 
* Filter for resolve
* Filter for probe
This commit is contained in:
euzu
2026-04-17 11:57:43 +02:00
committed by GitHub
parent 6841d791bc
commit 9eef3b24e0
49 changed files with 912 additions and 779 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/library.rs
+10 -10
View File
@@ -552,9 +552,9 @@ mod tests {
let seasons = series.details.as_ref().and_then(|details| details.seasons.as_ref()).expect("seasons missing");
let season = seasons.first().expect("season missing");
assert_eq!(season.cover.as_deref().map(|v| v.as_ref()), Some("/api/v1/library/thumbnail/series-thumb-uuid"));
assert_eq!(season.cover_tmdb.as_deref().map(|v| v.as_ref()), Some("/api/v1/library/thumbnail/series-thumb-uuid"));
assert_eq!(season.cover_big.as_deref().map(|v| v.as_ref()), Some("/api/v1/library/thumbnail/series-thumb-uuid"));
assert_eq!(season.cover.as_deref(), Some("/api/v1/library/thumbnail/series-thumb-uuid"));
assert_eq!(season.cover_tmdb.as_deref(), Some("/api/v1/library/thumbnail/series-thumb-uuid"));
assert_eq!(season.cover_big.as_deref(), Some("/api/v1/library/thumbnail/series-thumb-uuid"));
assert!(season.overview.is_none());
}
@@ -612,11 +612,11 @@ mod tests {
};
let details = video.details.as_ref().expect("details missing");
assert_eq!(details.video.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"hevc\"}"));
assert_eq!(details.audio.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"aac\"}"));
assert_eq!(details.duration_secs.as_deref().map(|value| value.as_ref()), Some("5430"));
assert_eq!(details.duration.as_deref().map(|value| value.as_ref()), Some("01:30:30"));
assert_eq!(details.runtime.as_deref().map(|value| value.as_ref()), Some("5430"));
assert_eq!(details.video.as_deref(), Some("{\"codec_name\":\"hevc\"}"));
assert_eq!(details.audio.as_deref(), Some("{\"codec_name\":\"aac\"}"));
assert_eq!(details.duration_secs.as_deref(), Some("5430"));
assert_eq!(details.duration.as_deref(), Some("01:30:30"));
assert_eq!(details.runtime.as_deref(), Some("5430"));
assert_eq!(details.bitrate, 8_200_000);
}
@@ -651,8 +651,8 @@ mod tests {
let episodes = series.details.as_ref().and_then(|details| details.episodes.as_ref()).expect("episodes missing");
let episode = episodes.first().expect("episode missing");
assert_eq!(episode.video.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"h264\"}"));
assert_eq!(episode.audio.as_deref().map(|value| value.as_ref()), Some("{\"codec_name\":\"ac3\"}"));
assert_eq!(episode.video.as_deref(), Some("{\"codec_name\":\"h264\"}"));
assert_eq!(episode.audio.as_deref(), Some("{\"codec_name\":\"ac3\"}"));
assert_eq!(episode.duration_secs, 1_541);
assert_eq!(episode.duration.as_ref(), "00:25:41");
assert_eq!(episode.bitrate, 3_100_000);
backend/src/processing/processor/playlist.rs
+77 -17
View File
@@ -1,3 +1,4 @@
use super::filtered_playlist_source::FilteredPlaylistSource;
use crate::{
api::{
model::{
@@ -29,7 +30,6 @@ use crate::{
debug_if_enabled, epg, log_memory_snapshot, m3u, trace_if_enabled, xtream, StepMeasure, StepMeasureCallback,
},
};
use super::filtered_playlist_source::FilteredPlaylistSource;
use futures::{FutureExt, StreamExt};
use indexmap::IndexMap;
use log::{debug, error, info, log_enabled, warn, Level};
@@ -231,10 +231,10 @@ fn map_playlist(source: &mut dyn PlaylistSource, target: &ConfigTarget) -> Optio
let mapping_binding = target.mapping.load();
let mappings = mapping_binding.as_ref()?;
let valid_mappings = mappings.iter().filter(|m| m.mapper.as_ref().is_some_and(|v| !v.is_empty()));
let iter: Box<dyn Iterator<Item = PlaylistItem>> = Box::new(source.into_items());
let iter: Box<dyn Iterator<Item=PlaylistItem>> = Box::new(source.into_items());
let mapped_iter = valid_mappings.fold(iter, |iter, mapping| {
Box::new(iter.flat_map(move |chan| map_channel_and_flatten(chan, mapping)))
as Box<dyn Iterator<Item = PlaylistItem>>
as Box<dyn Iterator<Item=PlaylistItem>>
});
let mut next_groups: IndexMap<CategoryKey, PlaylistGroup> = IndexMap::new();
let mut grp_id: u32 = 0;
@@ -464,7 +464,7 @@ async fn playlist_download_from_input(
input,
Some(xtream_clusters_to_download.as_slice()),
)
.await;
.await;
xtream_error_count = xtream_errors.len();
playlist.extend(xtream_playlist);
all_errors.extend(xtream_errors);
@@ -490,7 +490,7 @@ async fn playlist_download_from_input(
input,
Some(xtream_clusters_to_download.as_slice()),
)
.await;
.await;
let xtream_error_count = e.len();
(p, e, persisted, 0, xtream_error_count)
}
@@ -658,7 +658,7 @@ async fn process_source(
&mut errors,
consume_input_source,
)
.await
.await
{
Ok(()) => {
target_stats.push(TargetStats::success(&target.name));
@@ -1133,7 +1133,7 @@ async fn process_playlist_for_target(
target,
ctx.playlist_state.as_ref(),
)
.await;
.await;
step.stop("Persisting playlists");
log_memory_snapshot(format!("target '{}' after_persist", target.name).as_str());
result
@@ -1256,17 +1256,42 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
let mut queued_live_count = 0usize;
let mut queued_stream_count = 0usize;
let probe_filter = fpl.input.options.as_ref().and_then(|o| o.probe_filter.as_ref());
for item in fpl.items() {
if !is_probe_supported_item_type(item.header.item_type) {
continue;
}
match item.header.item_type {
PlaylistItemType::Live => {
if !probe_live_enabled {
continue;
}
}
PlaylistItemType::Video | PlaylistItemType::LocalVideo => {
if !probe_vod_enabled {
continue;
}
}
PlaylistItemType::Series | PlaylistItemType::LocalSeries => {
if !probe_series_enabled {
continue;
}
}
_ => continue,
}
// If input has a probe filter and this item doesn't match, skip probing
if let Some(p_filter) = probe_filter {
let provider = ValueProvider { pli: &item, match_as_ascii: false };
if !p_filter.filter(&provider) {
continue;
}
}
match item.header.item_type {
PlaylistItemType::Live => {
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) {
@@ -1301,18 +1326,12 @@ async fn playlist_probe(ctx: &PlaylistProcessingContext, target: &ConfigTarget,
// generic probe path to keep behaviour consistent with non-xtream outputs.
}
PlaylistItemType::Video | PlaylistItemType::LocalVideo => {
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;
@@ -1436,8 +1455,8 @@ async fn process_watch(
.filter(|pl| watches.iter().any(|r| r.is_match(&pl.title)))
.map(|pl| process_group_watch(app_config, client, &target.name, pl)),
)
.for_each_concurrent(16, |f| f)
.await;
.for_each_concurrent(16, |f| f)
.await;
true
} else {
@@ -1522,7 +1541,7 @@ pub async fn exec_processing(
max_update_duration,
std::panic::AssertUnwindSafe(process_sources(&ctx)).catch_unwind(),
)
.await;
.await;
let (stats, errors) = match process_result {
Ok(Ok((stats, errors))) => (stats, errors),
Ok(Err(_)) => {
@@ -1593,6 +1612,8 @@ pub async fn exec_processing(
mod tests {
use super::*;
use crate::model::StagedInput;
use shared::foundation::{get_filter, ValueProvider};
use shared::model::{PlaylistItem, PlaylistItemHeader, PlaylistItemType};
use shared::utils::Internable;
fn item_with_props(props: StreamProperties) -> PlaylistItem {
@@ -1770,4 +1791,43 @@ mod tests {
assert_eq!(filtered_groups.len(), 1);
assert_eq!(filtered_groups[0].xtream_cluster, XtreamCluster::Video);
}
fn make_test_item(name: &str, item_type: PlaylistItemType) -> PlaylistItem {
let header = PlaylistItemHeader {
name: name.into(),
group: "Test Group".intern(),
item_type,
..Default::default()
};
PlaylistItem { header }
}
#[test]
fn test_filter_evalutes_correctly() {
let filter = get_filter(r#"name ~ "Allowed""#, None).unwrap();
let allowed_item = make_test_item("Allowed Channel", PlaylistItemType::Live);
let denied_item = make_test_item("Denied Channel", PlaylistItemType::Live);
let allowed_provider = ValueProvider { pli: &allowed_item, match_as_ascii: false };
let denied_provider = ValueProvider { pli: &denied_item, match_as_ascii: false };
assert!(filter.filter(&allowed_provider));
assert!(!filter.filter(&denied_provider));
}
#[test]
fn test_filter_with_type_comparison() {
let filter = get_filter("type = vod", None).unwrap();
let vod_item = make_test_item("Test Movie", PlaylistItemType::Video);
let live_item = make_test_item("Test Channel", PlaylistItemType::Live);
let vod_provider = ValueProvider { pli: &vod_item, match_as_ascii: false };
let live_provider = ValueProvider { pli: &live_item, match_as_ascii: false };
assert!(filter.filter(&vod_provider));
assert!(!filter.filter(&live_provider));
}
}
backend/src/processing/processor/sort.rs
+3 -3
View File
@@ -795,14 +795,14 @@ mod tests {
header: PlaylistItemHeader { title: Arc::from("HD"), source_ordinal: 2, ..Default::default() },
}],
);
let group_uhd = make_group(
let group_ultra_hd = make_group(
2,
"uhd",
vec![PlaylistItem {
header: PlaylistItemHeader { title: Arc::from("UHD"), source_ordinal: 3, ..Default::default() },
}],
);
let group_fhd = make_group(
let group_full_hd = make_group(
3,
"fhd",
vec![PlaylistItem {
@@ -822,7 +822,7 @@ mod tests {
filter: Filter::default(),
};
let mut groups = vec![group_hd, group_uhd, group_fhd];
let mut groups = vec![group_hd, group_ultra_hd, group_full_hd];
sort_groups(&mut groups, &[group_sort], false);
let sorted = groups.iter().map(|group| group.channels[0].header.title.clone()).collect::<Vec<_>>();
backend/src/processing/processor/stream_probe.rs
+1 -1
View File
@@ -403,7 +403,7 @@ mod tests {
panic!("expected video properties");
};
let details = video.details.as_ref().unwrap_or_else(|| unreachable!());
assert_eq!(details.duration_secs.as_deref().map(|value| value.as_ref()), Some("1541"));
assert_eq!(details.duration_secs.as_deref().map(std::convert::AsRef::as_ref), Some("1541"));
assert_eq!(details.bitrate, 3_100_000);
}
}
backend/src/processing/processor/xtream_series.rs
+22
View File
@@ -34,6 +34,7 @@ use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use shared::foundation::ValueProvider;
use shared::utils::default_probe_user_priority;
create_resolve_options_function_for_xtream_target!(series);
@@ -193,12 +194,22 @@ fn queue_background_series_info(
let input = fpl.input;
let input_name_arc = input.name.clone();
// Extract filter before iterating to avoid borrow conflict
let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());
for pli in fpl.items_mut() {
if !filter(pli) {
continue;
}
// If input has a resolve filter and this item doesn't match, skip processing
if let Some(r_filter) = resolve_filter {
let provider = ValueProvider { pli, match_as_ascii: false };
if !r_filter.filter(&provider) {
continue;
}
}
let provider_id = if let Ok(uid) = pli.header.id.parse::<u32>() {
ProviderIdType::Id(uid)
} else {
@@ -277,11 +288,22 @@ async fn process_immediate_series_info(
SeriesProbeSettings::from_metadata_update(config.metadata_update.as_ref())
};
// Extract filter before iterating to avoid borrow conflict
let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());
for pli in fpl.items_mut() {
if !filter(pli) {
continue;
}
// If input has a resolve filter and this item doesn't match, skip processing
if let Some(r_filter) = resolve_filter {
let provider = ValueProvider { pli, match_as_ascii: false };
if !r_filter.filter(&provider) {
continue;
}
}
let provider_id = if let Ok(uid) = pli.header.id.parse::<u32>() {
ProviderIdType::Id(uid)
} else {
backend/src/processing/processor/xtream_vod.rs
+22
View File
@@ -31,6 +31,7 @@ use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use shared::foundation::ValueProvider;
use shared::utils::default_probe_user_priority;
create_resolve_options_function_for_xtream_target!(vod);
@@ -164,11 +165,22 @@ async fn process_immediate_vod_info(
let mut processed_count = 0;
let mut last_log_time = Instant::now();
// Extract filter before iterating to avoid borrow conflict
let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());
for pli in fpl.items_mut() {
if !filter(pli) {
continue;
}
// If input has a filter and this item doesn't match, skip processing
if let Some(r_filter) = resolve_filter {
let provider = ValueProvider { pli, match_as_ascii: false };
if !r_filter.filter(&provider) {
continue;
}
}
let provider_id = if let Ok(uid) = pli.header.id.parse::<u32>() {
ProviderIdType::Id(uid)
} else {
@@ -429,11 +441,21 @@ fn queue_background_vod_info(
};
let input = fpl.input;
// Extract filter before iterating to avoid borrow conflict
let resolve_filter = fpl.input.options.as_ref().and_then(|o| o.resolve_filter.as_ref());
for pli in fpl.items_mut() {
if !filter(pli) {
continue;
}
// If input has a resolve filter and this item doesn't match, skip processing
if let Some(r_filter) = resolve_filter {
let provider = ValueProvider { pli, match_as_ascii: false };
if !r_filter.filter(&provider) {
continue;
}
}
let provider_id = if let Ok(uid) = pli.header.id.parse::<u32>() {
ProviderIdType::Id(uid)
} else {