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/web UI save overrides templates (#604)

Browse Source 
- WebUI save overrides templates
- Deadlock during playlist update and metadata background task
- Global templates like mappings
This commit is contained in:
euzu
2026-02-23 17:00:11 +01:00
committed by GitHub
parent b445232bf6
commit d0ce95eb3c
52 changed files with 2513 additions and 1127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/playlist.rs
+36 -22
View File
@@ -34,7 +34,7 @@ use crate::repository::{load_input_playlist, persist_input_playlist, persist_pla
use crate::repository::{CategoryKey, MemoryPlaylistSource, PlaylistSource};
use crate::utils::StepMeasure;
use crate::utils::{debug_if_enabled, trace_if_enabled};
use futures::StreamExt;
use futures::{FutureExt, StreamExt};
use indexmap::IndexMap;
use log::{debug, error, info, log_enabled, warn, Level};
use shared::concat_string;
@@ -200,7 +200,7 @@ fn map_channel(mut channel: PlaylistItem, mapping: &Mapping) -> (PlaylistItem, V
virtual_items: vec![],
match_as_ascii: mapping.match_as_ascii,
};
script.eval(&mut accessor, templates);
script.eval(&mut accessor, templates.map(Vec::as_slice));
virtual_items.extend(accessor.virtual_items.into_iter().map(|(_, pli)| pli));
}
}
@@ -224,10 +224,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;
@@ -515,7 +515,7 @@ async fn process_source(
&mut errors,
consume_input_source,
)
.await
.await
{
Ok(()) => {
target_stats.push(TargetStats::success(&target.name));
@@ -560,11 +560,7 @@ async fn download_input(
let need_download = !ctx.is_input_downloaded(&input.name).await;
// Keep this lock for the whole critical section (download + persist/load + mark processed)
// so parallel sources sharing the same input cannot observe a half-written state.
let _input_lock = if need_download {
Some(ctx.get_input_lock(&input.name).await)
} else {
None
};
let _input_lock = if need_download { Some(ctx.get_input_lock(&input.name).await) } else { None };
let mut mark_as_processed = false;
let playlist_download_result = if need_download {
@@ -912,7 +908,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
@@ -961,10 +957,10 @@ fn get_live_probe_interval_settings(
return None;
}
target.get_xtream_output().map(|_| {
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),
);
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)
});
(probe_delay, u64::from(input_probe_live_interval_hours) * 3600)
})
}
@@ -1195,8 +1191,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 {
@@ -1218,14 +1214,13 @@ pub async fn exec_processing(
pre_processed_inputs: Option<HashSet<Arc<str>>>,
acquired_permit: Option<crate::api::model::UpdateGuardPermit>,
) {
let _guard = if let Some(permit) = acquired_permit {
let playlist_guard = if let Some(permit) = acquired_permit {
Some(permit)
} else if let Some(guard) = &update_guard {
let lock_result: Option<crate::api::model::UpdateGuardPermit> = guard.try_playlist();
if let Some(permit) = lock_result {
if let Some(permit) = guard.acquire_playlist_lock().await {
Some(permit)
} else {
warn!("Playlist update already in progress; update skipped.");
warn!("Playlist update lock is closed; update skipped.");
if let Some(events) = event_manager.as_deref() {
events.send_event(EventMessage::PlaylistUpdate(shared::model::PlaylistUpdateState::Failure));
}
@@ -1235,6 +1230,13 @@ pub async fn exec_processing(
None
};
// Pause background metadata/probe tasks for the full update lifecycle.
let _background_pause_guard = if let Some(manager) = metadata_manager.as_ref() {
Some(manager.acquire_update_pause_guard().await)
} else {
None
};
log_memory_snapshot("exec_processing start");
// Initialize Context
@@ -1253,8 +1255,20 @@ pub async fn exec_processing(
};
let start_time = Instant::now();
let (stats, errors) = process_sources(&ctx).await;
let process_result = std::panic::AssertUnwindSafe(process_sources(&ctx)).catch_unwind().await;
let Ok((stats, errors)) = process_result else {
error!("Playlist processing panicked");
if let Some(events) = event_manager.as_deref() {
events.send_event(EventMessage::PlaylistUpdate(shared::model::PlaylistUpdateState::Failure));
}
return;
};
log_memory_snapshot("exec_processing after_process_sources");
// Keep the update lock only for the critical processing section.
drop(playlist_guard);
debug!("Released playlist update lock; dispatching notifications and events");
// log errors
for err in &errors {
error!("{}", err.message);
backend/src/processing/processor/sort.rs
+88 -154
View File
@@ -97,14 +97,8 @@ fn evaluate_sequence(plan: &SequencePlan, value: &str) -> SequenceMatch {
fn compare_sequence_match(a: &SequenceMatch, b: &SequenceMatch, order: SortOrder) -> Ordering {
match (a, b) {
(
SequenceMatch::Matched {
sequence_idx: idx_a,
captures: captures_a,
},
SequenceMatch::Matched {
sequence_idx: idx_b,
captures: captures_b,
},
SequenceMatch::Matched { sequence_idx: idx_a, captures: captures_a },
SequenceMatch::Matched { sequence_idx: idx_b, captures: captures_b },
) => {
// Sequence index is an explicit priority list and is never reversed by DESC.
let idx_ord = idx_a.cmp(idx_b);
@@ -163,21 +157,13 @@ fn build_rule_cache_entry(
match_as_ascii: bool,
) -> RuleCacheEntry {
let Some(item) = item else {
return RuleCacheEntry {
filter_pass: false,
value: None,
sequence_match: None,
};
return RuleCacheEntry { filter_pass: false, value: None, sequence_match: None };
};
let provider = ValueProvider { pli: item, match_as_ascii };
let filter_pass = rule.rule.filter.filter(&provider);
if !filter_pass {
return RuleCacheEntry {
filter_pass: false,
value: None,
sequence_match: None,
};
return RuleCacheEntry { filter_pass: false, value: None, sequence_match: None };
}
let value = provider.get(rule.rule.field.as_str());
@@ -186,18 +172,11 @@ fn build_rule_cache_entry(
_ => None,
};
RuleCacheEntry {
filter_pass: true,
value,
sequence_match,
}
RuleCacheEntry { filter_pass: true, value, sequence_match }
}
fn build_group_rule_cache(rule: &PreparedRule, groups: &[PlaylistGroup], match_as_ascii: bool) -> Vec<RuleCacheEntry> {
groups
.iter()
.map(|group| build_rule_cache_entry(rule, group.channels.first(), match_as_ascii))
.collect()
groups.iter().map(|group| build_rule_cache_entry(rule, group.channels.first(), match_as_ascii)).collect()
}
fn build_channel_rule_cache(
@@ -205,23 +184,42 @@ fn build_channel_rule_cache(
channels: &[shared::model::PlaylistItem],
match_as_ascii: bool,
) -> Vec<RuleCacheEntry> {
channels
.iter()
.map(|channel| build_rule_cache_entry(rule, Some(channel), match_as_ascii))
.collect()
channels.iter().map(|channel| build_rule_cache_entry(rule, Some(channel), match_as_ascii)).collect()
}
fn reorder_by_indices<T>(items: &mut Vec<T>, indices: &[usize]) {
debug_assert_eq!(items.len(), indices.len());
if items.len() != indices.len() {
log::error!(
"Invalid sort permutation: index length {} does not match item length {}",
indices.len(),
items.len()
);
return;
}
let mut seen = vec![false; items.len()];
for &idx in indices {
if idx >= items.len() {
log::error!("Invalid sort permutation: index {idx} out of bounds for {} items", items.len());
return;
}
if std::mem::replace(&mut seen[idx], true) {
log::error!("Invalid sort permutation: duplicate index {idx}");
return;
}
}
let mut original: Vec<Option<T>> = std::mem::take(items).into_iter().map(Some).collect();
items.reserve(indices.len());
for &idx in indices {
let item = original[idx]
.take()
.expect("index permutation must contain each source position exactly once");
items.push(item);
if let Some(item) = original[idx].take() {
items.push(item);
} else {
log::error!("Invalid sort permutation: missing index {idx} after validation");
items.extend(original.into_iter().flatten());
return;
}
}
}
@@ -304,11 +302,7 @@ pub(in crate::processing::processor) fn sort_playlist(
true
}
fn sort_groups(
groups: &mut Vec<PlaylistGroup>,
rules: &[ConfigSortRule],
match_as_ascii: bool,
) {
fn sort_groups(groups: &mut Vec<PlaylistGroup>, rules: &[ConfigSortRule], match_as_ascii: bool) {
let group_rules: Vec<_> = rules
.iter()
.filter(|r| matches!(r.target, SortTarget::Group))
@@ -321,10 +315,8 @@ fn sort_groups(
return;
}
let rule_caches: Vec<Vec<RuleCacheEntry>> = group_rules
.iter()
.map(|rule| build_group_rule_cache(rule, groups.as_slice(), match_as_ascii))
.collect();
let rule_caches: Vec<Vec<RuleCacheEntry>> =
group_rules.iter().map(|rule| build_group_rule_cache(rule, groups.as_slice(), match_as_ascii)).collect();
let mut group_indices: Vec<usize> = (0..groups.len()).collect();
group_indices.sort_by(|left_idx, right_idx| {
@@ -333,25 +325,15 @@ fn sort_groups(
return ord;
}
let order1 = groups[*left_idx]
.channels
.first()
.map_or(u32::MAX, |c| c.header.source_ordinal);
let order2 = groups[*right_idx]
.channels
.first()
.map_or(u32::MAX, |c| c.header.source_ordinal);
let order1 = groups[*left_idx].channels.first().map_or(u32::MAX, |c| c.header.source_ordinal);
let order2 = groups[*right_idx].channels.first().map_or(u32::MAX, |c| c.header.source_ordinal);
order1.cmp(&order2)
});
reorder_by_indices(groups, &group_indices);
}
fn sort_channels_in_groups(
groups: &mut [PlaylistGroup],
rules: &[ConfigSortRule],
match_as_ascii: bool,
) {
fn sort_channels_in_groups(groups: &mut [PlaylistGroup], rules: &[ConfigSortRule], match_as_ascii: bool) {
let channel_rules: Vec<_> = rules
.iter()
.filter(|r| matches!(r.target, SortTarget::Channel))
@@ -361,18 +343,14 @@ fn sort_channels_in_groups(
if channel_rules.is_empty() {
for group in groups {
group
.channels
.sort_by_key(|a| a.header.source_ordinal);
group.channels.sort_by_key(|a| a.header.source_ordinal);
}
return;
}
for group in groups {
let rule_caches: Vec<Vec<RuleCacheEntry>> = channel_rules
.iter()
.map(|rule| build_channel_rule_cache(rule, &group.channels, match_as_ascii))
.collect();
let rule_caches: Vec<Vec<RuleCacheEntry>> =
channel_rules.iter().map(|rule| build_channel_rule_cache(rule, &group.channels, match_as_ascii)).collect();
let mut channel_indices: Vec<usize> = (0..group.channels.len()).collect();
channel_indices.sort_by(|left_idx, right_idx| {
@@ -381,10 +359,7 @@ fn sort_channels_in_groups(
return ord;
}
group.channels[*left_idx]
.header
.source_ordinal
.cmp(&group.channels[*right_idx].header.source_ordinal)
group.channels[*left_idx].header.source_ordinal.cmp(&group.channels[*right_idx].header.source_ordinal)
});
reorder_by_indices(&mut group.channels, &channel_indices);
@@ -398,17 +373,14 @@ mod tests {
};
use crate::model::ConfigSortRule;
use shared::foundation::Filter;
use shared::model::{ItemField, PlaylistGroup, PlaylistItem, PlaylistItemHeader, SortOrder, SortTarget, XtreamCluster};
use shared::model::{
ItemField, PlaylistGroup, PlaylistItem, PlaylistItemHeader, SortOrder, SortTarget, XtreamCluster,
};
use std::cmp::Ordering;
use std::sync::Arc;
fn make_group(id: u32, title: &str, channels: Vec<PlaylistItem>) -> PlaylistGroup {
PlaylistGroup {
id,
title: title.into(),
channels,
xtream_cluster: XtreamCluster::Live,
}
PlaylistGroup { id, title: title.into(), channels, xtream_cluster: XtreamCluster::Live }
}
#[test]
@@ -427,16 +399,16 @@ mod tests {
("T", "SD"),
("A", "FHD"),
]
.into_iter()
.enumerate()
.map(|(i, (name, quality))| PlaylistItem {
header: PlaylistItemHeader {
title: format!("Chanel {name} [{quality}]").into(),
source_ordinal: u32::try_from(i).unwrap(),
..Default::default()
},
})
.collect();
.into_iter()
.enumerate()
.map(|(i, (name, quality))| PlaylistItem {
header: PlaylistItemHeader {
title: format!("Chanel {name} [{quality}]").into(),
source_ordinal: u32::try_from(i).unwrap(),
..Default::default()
},
})
.collect();
let channel_sort = ConfigSortRule {
target: SortTarget::Channel,
@@ -466,7 +438,10 @@ mod tests {
"Chanel K [HD]",
"Chanel Z [HD]",
"Chanel T [SD]",
].into_iter().map(Into::into).collect::<Vec<Arc<str>>>();
]
.into_iter()
.map(Into::into)
.collect::<Vec<Arc<str>>>();
let sorted = groups
.into_iter()
@@ -508,16 +483,16 @@ mod tests {
"US| West e",
"US| West f",
]
.into_iter()
.enumerate()
.map(|(i, name)| PlaylistItem {
header: PlaylistItemHeader {
title: name.to_string().into(),
source_ordinal: u32::try_from(i).unwrap(),
..Default::default()
},
})
.collect();
.into_iter()
.enumerate()
.map(|(i, name)| PlaylistItem {
header: PlaylistItemHeader {
title: name.to_string().into(),
source_ordinal: u32::try_from(i).unwrap(),
..Default::default()
},
})
.collect();
let channel_sort = ConfigSortRule {
target: SortTarget::Channel,
@@ -564,7 +539,10 @@ mod tests {
"US| East d",
"US| West e",
"US| West f",
].into_iter().map(Into::into).collect::<Vec<Arc<str>>>();
]
.into_iter()
.map(Into::into)
.collect::<Vec<Arc<str>>>();
let sorted = groups
.into_iter()
@@ -598,12 +576,7 @@ mod tests {
];
channels.sort_by(|a, b| {
let ord = playlist_comparator(
Some(&sequence),
SortOrder::Desc,
&a.header.title,
&b.header.title,
);
let ord = playlist_comparator(Some(&sequence), SortOrder::Desc, &a.header.title, &b.header.title);
if ord == Ordering::Equal {
a.header.source_ordinal.cmp(&b.header.source_ordinal)
@@ -612,15 +585,9 @@ mod tests {
}
});
let sorted = channels
.into_iter()
.map(|pli| pli.header.title)
.collect::<Vec<_>>();
let sorted = channels.into_iter().map(|pli| pli.header.title).collect::<Vec<_>>();
let expected = vec!["A-7", "A-1", "B-9", "B-2"]
.into_iter()
.map(Into::into)
.collect::<Vec<Arc<str>>>();
let expected = vec!["A-7", "A-1", "B-9", "B-2"].into_iter().map(Into::into).collect::<Vec<Arc<str>>>();
assert_eq!(expected, sorted);
}
@@ -636,21 +603,9 @@ mod tests {
};
let prepared = PreparedRule::new(&rule);
let pass_with_value = RuleCacheEntry {
filter_pass: true,
value: Some(Arc::from("A-2")),
sequence_match: None,
};
let pass_without_value = RuleCacheEntry {
filter_pass: true,
value: None,
sequence_match: None,
};
let fail_filter = RuleCacheEntry {
filter_pass: false,
value: None,
sequence_match: None,
};
let pass_with_value = RuleCacheEntry { filter_pass: true, value: Some(Arc::from("A-2")), sequence_match: None };
let pass_without_value = RuleCacheEntry { filter_pass: true, value: None, sequence_match: None };
let fail_filter = RuleCacheEntry { filter_pass: false, value: None, sequence_match: None };
assert_eq!(compare_rule_entries(&prepared, &pass_with_value, &fail_filter), Ordering::Less);
assert_eq!(compare_rule_entries(&prepared, &fail_filter, &pass_with_value), Ordering::Greater);
@@ -690,16 +645,9 @@ mod tests {
let mut groups = vec![make_group(1, "G1", channels)];
sort_channels_in_groups(groups.as_mut_slice(), &[first_rule, second_rule], false);
let sorted = groups[0]
.channels
.iter()
.map(|pli| pli.header.title.clone())
.collect::<Vec<_>>();
let sorted = groups[0].channels.iter().map(|pli| pli.header.title.clone()).collect::<Vec<_>>();
let expected = vec!["A-2-x", "A-1-y"]
.into_iter()
.map(Into::into)
.collect::<Vec<Arc<str>>>();
let expected = vec!["A-2-x", "A-1-y"].into_iter().map(Into::into).collect::<Vec<Arc<str>>>();
assert_eq!(expected, sorted);
}
@@ -709,22 +657,14 @@ mod tests {
1,
"A",
vec![PlaylistItem {
header: PlaylistItemHeader {
title: Arc::from("A-2-x"),
source_ordinal: 0,
..Default::default()
},
header: PlaylistItemHeader { title: Arc::from("A-2-x"), source_ordinal: 0, ..Default::default() },
}],
);
let group_b = make_group(
2,
"B",
vec![PlaylistItem {
header: PlaylistItemHeader {
title: Arc::from("A-1-y"),
source_ordinal: 1,
..Default::default()
},
header: PlaylistItemHeader { title: Arc::from("A-1-y"), source_ordinal: 1, ..Default::default() },
}],
);
@@ -740,14 +680,8 @@ mod tests {
let mut groups = vec![group_a, group_b];
sort_groups(&mut groups, &[group_rule], false);
let sorted = groups
.iter()
.map(|group| group.channels[0].header.title.clone())
.collect::<Vec<_>>();
let expected = vec!["A-1-y", "A-2-x"]
.into_iter()
.map(Into::into)
.collect::<Vec<Arc<str>>>();
let sorted = groups.iter().map(|group| group.channels[0].header.title.clone()).collect::<Vec<_>>();
let expected = vec!["A-1-y", "A-2-x"].into_iter().map(Into::into).collect::<Vec<Arc<str>>>();
assert_eq!(expected, sorted);
}
}
backend/src/processing/processor/xtream_series.rs
+109 -89
View File
@@ -7,7 +7,9 @@ use crate::model::{ConfigInput, ConfigInputFlags, InputSource};
use crate::processing::parser::xtream::create_xtream_series_episode_url;
use crate::processing::parser::xtream::parse_xtream_series_info;
use crate::processing::processor::playlist::{PlaylistProcessingContext, ProcessingPipe};
use crate::processing::processor::{create_resolve_options_function_for_xtream_target, ResolveOptions, ResolveOptionsFlags};
use crate::processing::processor::{
create_resolve_options_function_for_xtream_target, ResolveOptions, ResolveOptionsFlags,
};
use crate::ptt::ptt_parse_title;
use crate::repository::persists_input_series_info;
use crate::repository::{
@@ -16,12 +18,15 @@ use crate::repository::{
use crate::repository::{xtream_get_file_path, BPlusTreeQuery};
use crate::utils::{debug_if_enabled, xtream};
use log::{debug, error, info, log_enabled, trace, warn, Level};
use parking_lot::Mutex;
use serde_json::Value;
use shared::error::TuliproxError;
use shared::model::{MediaQuality, PlaylistEntry, PlaylistItem, SeriesStreamProperties, StreamProperties, XtreamPlaylistItem, XtreamSeriesInfo};
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};
@@ -50,15 +55,7 @@ pub async fn playlist_resolve_series(
}
provider_fpl.source.release_resources(XtreamCluster::Series);
playlist_resolve_series_info(
ctx,
errors,
processed_fpl,
resolve_options,
pipe,
target,
)
.await;
playlist_resolve_series_info(ctx, errors, processed_fpl, resolve_options, pipe, target).await;
if provider_fpl.is_memory() {
sync_resolved_series_properties(provider_fpl, processed_fpl);
@@ -68,16 +65,16 @@ pub async fn playlist_resolve_series(
}
#[allow(clippy::too_many_lines, clippy::too_many_arguments)]
async fn playlist_resolve_series_info(ctx: &PlaylistProcessingContext,
_errors: &mut Vec<TuliproxError>,
fpl: &mut FetchedPlaylist<'_>,
resolve_options: ResolveOptions,
pipe: &ProcessingPipe,
target: &ConfigTarget) {
async fn playlist_resolve_series_info(
ctx: &PlaylistProcessingContext,
_errors: &mut Vec<TuliproxError>,
fpl: &mut FetchedPlaylist<'_>,
resolve_options: ResolveOptions,
pipe: &ProcessingPipe,
target: &ConfigTarget,
) {
let filter = |pli: &PlaylistItem| {
if pli.header.xtream_cluster != XtreamCluster::Series
|| pli.header.item_type != PlaylistItemType::SeriesInfo
{
if pli.header.xtream_cluster != XtreamCluster::Series || pli.header.item_type != PlaylistItemType::SeriesInfo {
return false;
}
true
@@ -91,24 +88,9 @@ async fn playlist_resolve_series_info(ctx: &PlaylistProcessingContext,
let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb);
let groups_to_add = if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() {
queue_background_series_info(
ctx,
fpl,
filter,
&resolve_options,
skip_resolve,
resolve_tmdb_enabled,
)
queue_background_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled)
} else {
process_immediate_series_info(
ctx,
fpl,
filter,
&resolve_options,
skip_resolve,
resolve_tmdb_enabled,
)
.await
process_immediate_series_info(ctx, fpl, filter, &resolve_options, skip_resolve, resolve_tmdb_enabled).await
};
// Apply pipe transformations to new groups
@@ -128,16 +110,11 @@ async fn playlist_resolve_series_info(ctx: &PlaylistProcessingContext,
}
}
fn sync_resolved_series_properties(
provider_fpl: &mut FetchedPlaylist<'_>,
processed_fpl: &mut FetchedPlaylist<'_>,
) {
fn sync_resolved_series_properties(provider_fpl: &mut FetchedPlaylist<'_>, processed_fpl: &mut FetchedPlaylist<'_>) {
let mut resolved_series_by_provider_id: HashMap<u32, SeriesStreamProperties> = HashMap::new();
for pli in processed_fpl.items() {
if pli.header.xtream_cluster != XtreamCluster::Series
|| pli.header.item_type != PlaylistItemType::SeriesInfo
{
if pli.header.xtream_cluster != XtreamCluster::Series || pli.header.item_type != PlaylistItemType::SeriesInfo {
continue;
}
@@ -149,9 +126,7 @@ fn sync_resolved_series_properties(
}
if let Some(StreamProperties::Series(properties)) = pli.header.additional_properties.as_ref() {
resolved_series_by_provider_id
.entry(provider_id)
.or_insert_with(|| properties.as_ref().clone());
resolved_series_by_provider_id.entry(provider_id).or_insert_with(|| properties.as_ref().clone());
}
}
@@ -174,8 +149,7 @@ fn sync_resolved_series_properties(
}
if let Some(resolved) = resolved_series_by_provider_id.get(&provider_id) {
source_pli.header.additional_properties =
Some(StreamProperties::Series(Box::new(resolved.clone())));
source_pli.header.additional_properties = Some(StreamProperties::Series(Box::new(resolved.clone())));
}
}
}
@@ -279,7 +253,9 @@ async fn process_immediate_series_info(
let guard = file_lock;
let query = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)?;
Ok::<_, std::io::Error>((query, guard))
}).await {
})
.await
{
Ok(Ok((query, guard))) => Some((query, guard)),
Ok(Err(err)) => {
error!("Failed to open BPlusTreeQuery for Series: {err}");
@@ -299,9 +275,20 @@ async fn process_immediate_series_info(
// Pass the optional reference to the query
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 {
match update_series_info_immediate(
ctx,
active_provider,
input,
pli,
provider_id.clone(),
&reasons,
db_query_ref,
)
.await
{
Ok(Some(updated_props)) => {
pli.header.additional_properties = Some(StreamProperties::Series(Box::new(updated_props.clone())));
pli.header.additional_properties =
Some(StreamProperties::Series(Box::new(updated_props.clone())));
batch.push((provider_id.clone(), updated_props));
if batch.len() >= BATCH_SIZE {
@@ -309,7 +296,8 @@ async fn process_immediate_series_info(
db_query_holder = None;
_db_lock_holder = None;
let updates: Vec<(u32, SeriesStreamProperties)> = batch.iter()
let updates: Vec<(u32, SeriesStreamProperties)> = batch
.iter()
.filter_map(|(id, props)| {
if let ProviderIdType::Id(vid) = id {
Some((*vid, props.clone()))
@@ -328,7 +316,9 @@ async fn process_immediate_series_info(
XtreamCluster::Series,
&input.name,
updates,
).await {
)
.await
{
Ok(()) => batch.clear(),
Err(err) => {
error!(
@@ -368,18 +358,16 @@ async fn process_immediate_series_info(
// Release lock before final persist
_db_lock_holder = None;
let updates: Vec<(u32, SeriesStreamProperties)> = batch.into_iter()
let updates: Vec<(u32, SeriesStreamProperties)> = batch
.into_iter()
.filter_map(|(id, props)| if let ProviderIdType::Id(vid) = id { Some((vid, props)) } else { None })
.collect();
if !updates.is_empty() {
if let Err(err) = persist_input_series_info_batch(
&ctx.config,
&storage_path,
XtreamCluster::Series,
&input.name,
updates,
).await {
if let Err(err) =
persist_input_series_info_batch(&ctx.config, &storage_path, XtreamCluster::Series, &input.name, updates)
.await
{
error!("Failed to persist final batch series info: {err}");
}
}
@@ -400,7 +388,14 @@ fn expand_series_item(pli: &PlaylistItem, input: &ConfigInput) -> Option<Playlis
(header.group.clone(), header.get_name())
};
if let Some(episodes) = parse_xtream_series_info(&pli.get_uuid(), properties, &group, &series_name, input, global_release_date.as_ref()) {
if let Some(episodes) = parse_xtream_series_info(
&pli.get_uuid(),
properties,
&group,
&series_name,
input,
global_release_date.as_ref(),
) {
return Some(PlaylistGroup {
id: pli.header.category_id,
title: pli.header.group.clone(),
@@ -435,7 +430,8 @@ async fn update_series_info_immediate(
fetch_info,
reasons.contains(ResolveReason::Probe),
db_query,
).await
)
.await
}
fn check_resolve_reasons(
@@ -469,17 +465,19 @@ fn check_resolve_tmdb(
needs_info: bool,
reasons: &mut ResolveReasonSet,
) {
if resolve_tmdb_enabled
&& resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing)
&& needs_info {
if resolve_tmdb_enabled && resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing) && needs_info {
if let Some(StreamProperties::Series(series_stream_props)) = pli.header.additional_properties.as_ref() {
let has_tmdb = series_stream_props.tmdb.is_some();
let has_date = series_stream_props.release_date.is_some();
let title_present = !series_stream_props.name.is_empty() || !pli.header.title.is_empty();
if title_present && (!has_tmdb || !has_date) {
if !has_tmdb { reasons.set(ResolveReason::Tmdb); }
if !has_date { reasons.set(ResolveReason::Date); }
if !has_tmdb {
reasons.set(ResolveReason::Tmdb);
}
if !has_date {
reasons.set(ResolveReason::Date);
}
}
}
}
@@ -487,17 +485,15 @@ fn check_resolve_tmdb(
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| {
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());
missing_video || missing_audio
})
}),
})
}
_ => false,
};
@@ -528,7 +524,8 @@ pub async fn update_series_metadata(
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
let storage_path = get_input_storage_path(&input.name, working_dir)
.await
.map_err(|e| shared::error::info_err!("Storage path error: {e}"))?;
if input.has_flag(ConfigInputFlags::XtreamSkipSeries) {
@@ -652,7 +649,9 @@ pub async fn update_series_metadata(
}
}
let mut properties = props.unwrap();
let mut properties = props.ok_or_else(|| {
shared::error::info_err!("No Series properties available after fallback creation for {display_id}")
})?;
let resolve_tmdb_enabled = input.has_flag(ConfigInputFlags::ResolveTmdb);
@@ -660,7 +659,8 @@ pub async fn update_series_metadata(
if fetch_info
&& resolve_tmdb_enabled
&& (properties.tmdb.is_none() || properties.release_date.is_none())
&& !properties.name.is_empty() {
&& !properties.name.is_empty()
{
let config = app_config.config.load();
let library_config = config.library.as_ref();
let meta_resolver = MetadataResolver::new(library_config, client.clone());
@@ -689,12 +689,13 @@ pub async fn update_series_metadata(
}
// 3. API Name (fallback)
if (meta.is_none() || (meta.as_ref().is_some_and(|m| m.tmdb_id().is_none())))
&& !properties.name.is_empty() {
if (meta.is_none() || (meta.as_ref().is_some_and(|m| m.tmdb_id().is_none()))) && !properties.name.is_empty() {
let title_already_tried = if let Some(t) = title_candidate { t == properties.name.as_ref() } else { false };
if !tried_title || !title_already_tried {
debug!("Fallback to API Name '{}'...", properties.name);
meta = meta_resolver.resolve_from_title(&properties.name, properties.tmdb, false, resolve_tmdb_enabled).await;
meta = meta_resolver
.resolve_from_title(&properties.name, properties.tmdb, false, resolve_tmdb_enabled)
.await;
}
}
@@ -746,11 +747,21 @@ 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 episode_url =
create_xtream_series_episode_url(input_url, input_username, input_password, ep);
// Specific logging for the user to follow
let missing_reason = if missing_video && missing_audio { "video/audio" } else if missing_video { "video" } else { "audio" };
debug!("Probing Series Episode '{}' (S{}E{}) - Missing {}", ep.title, ep.season, ep.episode_num, missing_reason);
let missing_reason = if missing_video && missing_audio {
"video/audio"
} else if missing_video {
"video"
} else {
"audio"
};
debug!(
"Probing Series Episode '{}' (S{}E{}) - Missing {}",
ep.title, ep.season, ep.episode_num, missing_reason
);
if let Some((_quality, raw_video, raw_audio)) = crate::utils::ffmpeg::probe_url(
&episode_url,
@@ -758,7 +769,9 @@ pub async fn update_series_metadata(
analyze_duration,
probe_size,
ffprobe_timeout,
).await {
)
.await
{
if let Some(v) = raw_video {
ep.video = Some(v.to_string().into());
properties_updated = true;
@@ -795,9 +808,16 @@ pub async fn update_series_metadata(
if properties_updated || fetched_new {
if save {
if let Some(series_id) = series_id_opt {
persists_input_series_info(app_config, &storage_path, XtreamCluster::Series, &input.name, series_id, &properties)
.await
.map_err(|e| shared::error::info_err!("Persist error: {e}"))?;
persists_input_series_info(
app_config,
&storage_path,
XtreamCluster::Series,
&input.name,
series_id,
&properties,
)
.await
.map_err(|e| shared::error::info_err!("Persist error: {e}"))?;
debug_if_enabled!("Successfully updated Series metadata for ID {}", series_id);
}
backend/src/processing/processor/xtream_vod.rs
+108 -96
View File
@@ -1,24 +1,29 @@
use crate::api::model::{ActiveProviderManager, ProviderHandle};
use crate::api::model::{ResolveReason, ResolveReasonSet, UpdateTask, ProviderIdType};
use crate::api::model::{ProviderIdType, ResolveReason, ResolveReasonSet, UpdateTask};
use crate::library::MetadataResolver;
use crate::model::{ConfigInput, ConfigInputFlags};
use crate::model::FetchedPlaylist;
use crate::model::InputSource;
use crate::model::{AppConfig, ConfigTarget};
use crate::model::{ConfigInput, ConfigInputFlags};
use crate::processing::processor::playlist::PlaylistProcessingContext;
use crate::processing::processor::{create_resolve_options_function_for_xtream_target, ResolveOptions, ResolveOptionsFlags};
use crate::processing::processor::{
create_resolve_options_function_for_xtream_target, ResolveOptions, ResolveOptionsFlags,
};
use crate::ptt::ptt_parse_title;
use crate::repository::get_input_storage_path;
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::{trace_if_enabled, debug_if_enabled, xtream};
use crate::utils::{debug_if_enabled, trace_if_enabled, xtream};
use log::{debug, error, info, log_enabled, trace, warn, Level};
use parking_lot::Mutex;
use serde_json::Value;
use shared::error::TuliproxError;
use shared::model::{MediaQuality, PlaylistEntry, PlaylistItem, PlaylistItemType, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster, XtreamPlaylistItem, XtreamVideoInfo};
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};
@@ -47,19 +52,14 @@ pub async fn playlist_resolve_vod(
// Determine if we need to do anything
if !resolve_options.has_flag(ResolveOptionsFlags::Resolve)
&& !do_probe
&& !resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing) {
&& !resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing)
{
return;
}
provider_fpl.source.release_resources(XtreamCluster::Video);
playlist_resolve_vod_info(
ctx,
errors,
fpl,
resolve_options,
do_probe,
).await;
playlist_resolve_vod_info(ctx, errors, fpl, resolve_options, do_probe).await;
if provider_fpl.is_memory() {
sync_resolved_vod_properties(provider_fpl, fpl);
@@ -68,16 +68,11 @@ pub async fn playlist_resolve_vod(
provider_fpl.source.obtain_resources().await;
}
fn sync_resolved_vod_properties(
provider_fpl: &mut FetchedPlaylist<'_>,
processed_fpl: &mut FetchedPlaylist<'_>,
) {
fn sync_resolved_vod_properties(provider_fpl: &mut FetchedPlaylist<'_>, processed_fpl: &mut FetchedPlaylist<'_>) {
let mut resolved_vod_by_provider_id: HashMap<u32, VideoStreamProperties> = HashMap::new();
for pli in processed_fpl.items() {
if pli.header.xtream_cluster != XtreamCluster::Video
|| pli.header.item_type != PlaylistItemType::Video
{
if pli.header.xtream_cluster != XtreamCluster::Video || pli.header.item_type != PlaylistItemType::Video {
continue;
}
@@ -89,9 +84,7 @@ fn sync_resolved_vod_properties(
}
if let Some(StreamProperties::Video(properties)) = pli.header.additional_properties.as_ref() {
resolved_vod_by_provider_id
.entry(provider_id)
.or_insert_with(|| properties.as_ref().clone());
resolved_vod_by_provider_id.entry(provider_id).or_insert_with(|| properties.as_ref().clone());
}
}
@@ -114,8 +107,7 @@ fn sync_resolved_vod_properties(
}
if let Some(resolved) = resolved_vod_by_provider_id.get(&provider_id) {
source_pli.header.additional_properties =
Some(StreamProperties::Video(Box::new(resolved.clone())));
source_pli.header.additional_properties = Some(StreamProperties::Video(Box::new(resolved.clone())));
}
}
}
@@ -128,11 +120,8 @@ async fn playlist_resolve_vod_info(
resolve_options: ResolveOptions,
do_probe: bool,
) {
let filter = |pli: &PlaylistItem| {
if pli.header.xtream_cluster != XtreamCluster::Video
|| pli.header.item_type != PlaylistItemType::Video
{
if pli.header.xtream_cluster != XtreamCluster::Video || pli.header.item_type != PlaylistItemType::Video {
return false;
}
true
@@ -141,33 +130,21 @@ async fn playlist_resolve_vod_info(
let resolve_tmdb_enabled = fpl.input.has_flag(ConfigInputFlags::ResolveTmdb);
if resolve_options.has_flag(ResolveOptionsFlags::Background) && ctx.metadata_manager.is_some() {
queue_background_vod_info(
ctx,
fpl,
filter,
&resolve_options,
do_probe,
resolve_tmdb_enabled,
);
queue_background_vod_info(ctx, fpl, filter, &resolve_options, do_probe, resolve_tmdb_enabled);
} else {
process_immediate_vod_info(
ctx,
fpl,
filter,
resolve_options,
do_probe,
resolve_tmdb_enabled,
)
.await;
process_immediate_vod_info(ctx, fpl, filter, resolve_options, do_probe, resolve_tmdb_enabled).await;
}
}
#[allow(clippy::too_many_lines)]
async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut FetchedPlaylist<'_>,
filter: impl Fn(&PlaylistItem) -> bool,
resolve_options: ResolveOptions,
do_probe: bool,
resolve_tmdb_enabled: bool, ) {
async fn process_immediate_vod_info(
ctx: &PlaylistProcessingContext,
fpl: &mut FetchedPlaylist<'_>,
filter: impl Fn(&PlaylistItem) -> bool,
resolve_options: ResolveOptions,
do_probe: bool,
resolve_tmdb_enabled: bool,
) {
let input = fpl.input;
let working_dir = &ctx.config.config.load().working_dir;
@@ -209,7 +186,9 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F
let xtream_path = xtream_path.clone();
let query = match tokio::task::spawn_blocking(move || {
BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path)
}).await {
})
.await
{
Ok(Ok(query)) => Some(query),
Ok(Err(err)) => {
error!("Failed to open BPlusTreeQuery for VOD: {err}");
@@ -229,10 +208,13 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F
// Pass the optional reference to the query
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 {
match update_vod_info_immediate(ctx, prov_mgr, input, pli, provider_id.clone(), &reasons, db_query_ref)
.await
{
Ok(Some(updated_props)) => {
// Update the current item in fpl
pli.header.additional_properties = Some(StreamProperties::Video(Box::new(updated_props.clone())));
pli.header.additional_properties =
Some(StreamProperties::Video(Box::new(updated_props.clone())));
batch.push((provider_id.clone(), updated_props));
if batch.len() >= BATCH_SIZE {
// Release lock before persisting to avoid deadlock (persist needs write lock)
@@ -240,7 +222,8 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F
_db_lock_holder = None;
// Filter for u32 IDs for persistence
let updates: Vec<(u32, VideoStreamProperties)> = batch.iter()
let updates: Vec<(u32, VideoStreamProperties)> = batch
.iter()
.filter_map(|(id, props)| {
if let ProviderIdType::Id(vid) = id {
Some((*vid, props.clone()))
@@ -259,7 +242,9 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F
XtreamCluster::Video,
&input.name,
updates,
).await {
)
.await
{
Ok(()) => batch.clear(),
Err(err) => {
error!(
@@ -298,12 +283,16 @@ async fn process_immediate_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut F
// Release lock before final persist
_db_lock_holder = None;
let updates: Vec<(u32, VideoStreamProperties)> = batch.into_iter()
let updates: Vec<(u32, VideoStreamProperties)> = batch
.into_iter()
.filter_map(|(id, props)| if let ProviderIdType::Id(vid) = id { Some((vid, props)) } else { None })
.collect();
if !updates.is_empty() {
if let Err(err) = persist_input_vod_info_batch(&ctx.config, &storage_path, XtreamCluster::Video, &input.name, updates).await {
if let Err(err) =
persist_input_vod_info_batch(&ctx.config, &storage_path, XtreamCluster::Video, &input.name, updates)
.await
{
error!("Failed to persist final batch VOD info: {err}");
}
}
@@ -342,9 +331,7 @@ fn check_needs_probe(pli: &PlaylistItem, reasons: &mut ResolveReasonSet) {
let missing_audio = !MediaQuality::is_valid_media_info(details.and_then(|d| d.audio.as_deref()));
missing_video || missing_audio
}
None => {
true
}
None => true,
_ => false,
};
if needs_probe {
@@ -368,25 +355,30 @@ fn check_resolve_tmdb(
needs_info: bool,
reasons: &mut ResolveReasonSet,
) {
if resolve_tmdb_enabled
&& resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing)
&& needs_info {
if resolve_tmdb_enabled && resolve_options.has_flag(ResolveOptionsFlags::TmdbMissing) && needs_info {
if let Some(StreamProperties::Video(video_stream_props)) = pli.header.additional_properties.as_ref() {
let has_tmdb = video_stream_props.tmdb.is_some();
let has_date = video_stream_props.details.as_ref().and_then(|d| d.release_date.as_ref()).is_some();
if !has_tmdb || !has_date {
if !has_tmdb { reasons.set(ResolveReason::Tmdb); }
if !has_date { reasons.set(ResolveReason::Date); }
if !has_tmdb {
reasons.set(ResolveReason::Tmdb);
}
if !has_date {
reasons.set(ResolveReason::Date);
}
}
}
}
}
fn queue_background_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut FetchedPlaylist<'_>,
filter: impl Fn(&PlaylistItem) -> bool,
resolve_options: &ResolveOptions,
do_probe: bool,
resolve_tmdb_enabled: bool, ) {
fn queue_background_vod_info(
ctx: &PlaylistProcessingContext,
fpl: &mut FetchedPlaylist<'_>,
filter: impl Fn(&PlaylistItem) -> bool,
resolve_options: &ResolveOptions,
do_probe: bool,
resolve_tmdb_enabled: bool,
) {
let Some(mgr) = ctx.metadata_manager.as_ref() else {
return;
};
@@ -406,11 +398,8 @@ fn queue_background_vod_info(ctx: &PlaylistProcessingContext, fpl: &mut FetchedP
let reasons = check_resolve_reasons(resolve_options, do_probe, resolve_tmdb_enabled, pli);
if !reasons.is_empty() {
let task = UpdateTask::ResolveVod {
id: provider_id,
reason: reasons,
delay: resolve_options.resolve_delay,
};
let task =
UpdateTask::ResolveVod { id: provider_id, reason: reasons, delay: resolve_options.resolve_delay };
mgr.queue_task_background(input.name.clone(), task);
}
}
@@ -439,10 +428,10 @@ async fn update_vod_info_immediate(
fetch_info,
reasons.contains(ResolveReason::Probe),
db_query,
).await
)
.await
}
/// Updates metadata for a single VOD item (Info + Probe).
///
/// # Arguments
@@ -465,7 +454,8 @@ pub async fn update_vod_metadata(
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
let storage_path = get_input_storage_path(&input.name, working_dir)
.await
.map_err(|e| shared::error::info_err!("Storage path error: {e}"))?;
// Check if we should skip based on input options
@@ -603,8 +593,9 @@ pub async fn update_vod_metadata(
}
}
// Now it's safe to unwrap because we handled the None case above
let mut properties = props.unwrap();
let Some(mut properties) = props else {
return Err(shared::error::info_err!("No VOD properties available after fallback creation for {display_id}"));
};
let resolve_tmdb_enabled = input.has_flag(ConfigInputFlags::ResolveTmdb);
@@ -650,13 +641,16 @@ pub async fn update_vod_metadata(
}
// 3. API Name (fallback)
if (meta.is_none() || (meta.as_ref().is_some_and(|m| m.tmdb_id().is_none())))
&& !properties.name.is_empty() {
let title_already_tried = if let Some(t) = title_candidate { t == properties.name.as_ref() } else { false };
if (meta.is_none() || (meta.as_ref().is_some_and(|m| m.tmdb_id().is_none()))) && !properties.name.is_empty()
{
let title_already_tried =
if let Some(t) = title_candidate { t == properties.name.as_ref() } else { false };
if !tried_title || !title_already_tried {
trace!("Fallback to API Name '{}'...", properties.name);
meta = meta_resolver.resolve_from_title(&properties.name, properties.tmdb, true, resolve_tmdb_enabled).await;
meta = meta_resolver
.resolve_from_title(&properties.name, properties.tmdb, true, resolve_tmdb_enabled)
.await;
}
}
@@ -665,7 +659,8 @@ pub async fn update_vod_metadata(
if let Some(o_name) = properties.details.as_ref().and_then(|d| d.o_name.as_deref()) {
if !o_name.is_empty() && o_name != properties.name.as_ref() {
trace!("Fallback to API Original Name '{o_name}'...");
meta = meta_resolver.resolve_from_title(o_name, properties.tmdb, true, resolve_tmdb_enabled).await;
meta =
meta_resolver.resolve_from_title(o_name, properties.tmdb, true, resolve_tmdb_enabled).await;
}
}
}
@@ -697,18 +692,26 @@ pub async fn update_vod_metadata(
properties.details = Some(VideoStreamDetailProperties::default());
}
let details = properties.details.as_ref().unwrap();
let missing_video = !MediaQuality::is_valid_media_info(details.video.as_deref());
let missing_audio = !MediaQuality::is_valid_media_info(details.audio.as_deref());
let (missing_video, missing_audio) = match properties.details.as_ref() {
Some(details) => (
!MediaQuality::is_valid_media_info(details.video.as_deref()),
!MediaQuality::is_valid_media_info(details.audio.as_deref()),
),
None => (true, true),
};
if missing_video || missing_audio {
let input_url = input.url.as_str();
let username = input.username.as_deref().unwrap_or("");
let password = input.password.as_deref().unwrap_or("");
let stream_url = crate::processing::parser::xtream::create_xtream_url(
XtreamCluster::Video, input_url, username, password,
XtreamCluster::Video,
input_url,
username,
password,
&StreamProperties::Video(Box::new(properties.clone())),
true, true,
true,
true,
);
let config = app_config.config.load();
@@ -732,7 +735,9 @@ pub async fn update_vod_metadata(
analyze_duration,
probe_size,
ffprobe_timeout,
).await {
)
.await
{
if let Some(details) = properties.details.as_mut() {
if let Some(v) = raw_video {
details.video = Some(v.to_string().into());
@@ -757,9 +762,16 @@ pub async fn update_vod_metadata(
if properties_updated || fetched_new {
if save {
if let Some(stream_id) = stream_id_opt {
persist_input_vod_info(app_config, &storage_path, XtreamCluster::Video, &input.name, stream_id, &properties)
.await
.map_err(|e| shared::error::info_err!("Persist error: {e}"))?;
persist_input_vod_info(
app_config,
&storage_path,
XtreamCluster::Video,
&input.name,
stream_id,
&properties,
)
.await
.map_err(|e| shared::error::info_err!("Persist error: {e}"))?;
}
debug_if_enabled!("Successfully updated VOD metadata for '{}' (ID: {})", display_title, display_id);