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

Resolve Series name processing added

Browse Source
This commit is contained in:
euzu
2024-12-30 10:27:25 +01:00
parent ac9867dbb5
commit a027e808b1
5 changed files with 71 additions and 75 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/processing/xtream_processor_series.rs
+31 -51
View File
@@ -11,6 +11,8 @@ use crate::{create_resolve_options_function_for_xtream_target, handle_error, han
use std::collections::HashMap;
use std::fs::File;
use std::io::{BufWriter, Write};
use std::time::Instant;
use log::info;
use crate::model::xtream::{XtreamSeriesEpisode, XtreamSeriesInfoEpisode};
const TAG_SERIES_INFO_LAST_MODIFIED: &str = "last_modified";
@@ -51,26 +53,32 @@ fn should_update_series_info(pli: &PlaylistItem, processed_provider_ids: &HashMa
}
async fn playlist_resolve_series_info(cfg: &Config, errors: &mut Vec<M3uFilterError>,
processed_fpl: &mut FetchedPlaylist<'_>, resolve_delay: u16) -> bool {
let mut processed_info_ids = read_processed_series_info_ids(cfg, errors, processed_fpl).await;
fpl: &mut FetchedPlaylist<'_>, resolve_delay: u16) -> bool {
let mut processed_info_ids = read_processed_series_info_ids(cfg, errors, fpl).await;
// we cant write to the indexed-document directly because of the write lock and time-consuming operation.
// All readers would be waiting for the lock and the app would be unresponsive.
// We collect the content into a wal file and write it once we collected everything.
let Some((wal_content_file, wal_record_file, wal_content_path, wal_record_path)) = create_resolve_info_wal_files(cfg, processed_fpl.input, XtreamCluster::Series)
let Some((wal_content_file, wal_record_file, wal_content_path, wal_record_path)) = create_resolve_info_wal_files(cfg, fpl.input, XtreamCluster::Series)
else { return !processed_info_ids.is_empty(); };
let mut content_writer = BufWriter::new(&wal_content_file);
let mut record_writer = BufWriter::new(&wal_record_file);
let mut content_updated = false;
for pli in processed_fpl.playlistgroups.iter()
let series_info_iter = fpl.playlistgroups.iter()
.filter(|&plg| plg.xtream_cluster == XtreamCluster::Series)
.flat_map(|plg| &plg.channels)
.filter(|&pli| pli.header.borrow().item_type == PlaylistItemType::SeriesInfo)
{
.filter(|&pli| pli.header.borrow().item_type == PlaylistItemType::SeriesInfo);
let series_info_count = series_info_iter.clone().count();
info!("Found {series_info_count} series info to resolve");
let start_time = Instant::now();
let mut processed_series_info = 0;
for pli in series_info_iter {
let (should_update, provider_id, ts) = should_update_series_info(pli, &processed_info_ids);
if should_update {
if let Some(content) = playlist_resolve_download_playlist_item(pli, processed_fpl.input, errors, resolve_delay, XtreamCluster::Series).await {
if let Some(content) = playlist_resolve_download_playlist_item(pli, fpl.input, errors, resolve_delay, XtreamCluster::Series).await {
handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &content),
|err| errors.push(notify_err!(format!("Failed to resolve series, could not write to content wal file {err}"))));
processed_info_ids.insert(provider_id, ts);
@@ -79,7 +87,13 @@ async fn playlist_resolve_series_info(cfg: &Config, errors: &mut Vec<M3uFilterEr
content_updated = true;
}
}
processed_series_info += 1;
let elapsed = start_time.elapsed().as_secs();
if elapsed > 0 && elapsed % 5 == 0 {
info!("resolved {processed_series_info}/{series_info_count} series info");
}
}
info!("resolved {processed_series_info}/{series_info_count} series info");
// content_wal contains the provider_id and series_info with episode listing
// record_wal contains provider_id and timestamp
if content_updated {
@@ -91,9 +105,9 @@ async fn playlist_resolve_series_info(cfg: &Config, errors: &mut Vec<M3uFilterEr
|err| errors.push(notify_err!(format!("Failed to resolve vod tmdb, could not write to wal file {err}"))));
drop(record_writer);
drop(wal_record_file);
handle_error!(xtream_update_input_info_file(cfg, processed_fpl.input, &wal_content_path, XtreamCluster::Series).await,
handle_error!(xtream_update_input_info_file(cfg, fpl.input, &wal_content_path, XtreamCluster::Series).await,
|err| errors.push(err));
handle_error!(xtream_update_input_series_record_from_wal_file(cfg, processed_fpl.input, &wal_record_path).await,
handle_error!(xtream_update_input_series_record_from_wal_file(cfg, fpl.input, &wal_record_path).await,
|err| errors.push(err));
}
@@ -104,11 +118,11 @@ async fn playlist_resolve_series_info(cfg: &Config, errors: &mut Vec<M3uFilterEr
}
async fn process_series_info(
cfg: &Config,
provider_fpl: &mut FetchedPlaylist<'_>,
fpl: &mut FetchedPlaylist<'_>,
errors: &mut Vec<M3uFilterError>,
) -> Vec<PlaylistGroup> {
let mut result: Vec<PlaylistGroup> = vec![];
let input = provider_fpl.input;
let input = fpl.input;
let Ok(Some((info_path, idx_path))) = get_input_storage_path(input, &cfg.working_dir)
.map(|storage_path| xtream_get_info_file_paths(&storage_path, XtreamCluster::Series))
@@ -131,7 +145,7 @@ async fn process_series_info(
};
let mut wal_writer = BufWriter::new(&wal_file);
for plg in provider_fpl
for plg in fpl
.playlistgroups
.iter_mut()
.filter(|plg| plg.xtream_cluster == XtreamCluster::Series)
@@ -147,7 +161,11 @@ async fn process_series_info(
let Ok(content) = info_reader.get(&provider_id) else { continue; };
match serde_json::from_str::<serde_json::Value>(&content) {
Ok(series_content) => {
match parse_xtream_series_info(&series_content, pli.header.borrow().group.as_str(), input) {
let (group, series_name) = {
let header = pli.header.borrow();
(header.group.clone(), if header.name.is_empty() {header.title.clone()} else { header.name.clone()})
};
match parse_xtream_series_info(&series_content, &group, &series_name, input) {
Ok(Some(series)) => {
for (episode, pli_episode) in &series {
let Some(provider_id) = &pli_episode.header.borrow_mut().get_provider_id() else { continue; };
@@ -213,41 +231,3 @@ pub async fn playlist_resolve_series(cfg: &Config, target: &ConfigTarget,
processed_fpl.update_playlist(plg);
}
}
//
// pub(in crate::processing) async fn playlist_resolve_series(target: &ConfigTarget, errors: &mut Vec<M3uFilterError>,
// pipe: &ProcessingPipe,
// provider_fpl: &mut FetchedPlaylist<'_>,
// processed_fpl: &mut FetchedPlaylist<'_>) {
// let (resolve_series, resolve_series_delay) =
// if let Some(options) = &target.options {
// (options.xtream_resolve_series && provider_fpl.input.input_type == InputType::Xtream && target.has_output(&TargetType::M3u),
// options.xtream_resolve_series_delay)
// } else {
// (false, 0)
// };
// if resolve_series {
// // collect all series in the processed lists
// let to_process_uuids: HashSet<Rc<UUIDType>> = processed_fpl.playlistgroups.iter()
// .filter(|plg| plg.xtream_cluster == XtreamCluster::Series)
// .flat_map(|plg| &plg.channels)
// .filter(|pli| pli.header.borrow().item_type == PlaylistItemType::SeriesInfo)
// .map(|pli| Rc::clone(&pli.header.borrow().uuid)).collect();
// let mut series_playlist = download::get_xtream_playlist_series(provider_fpl, to_process_uuids, errors, resolve_series_delay).await;
// // original content saved into original list
// for plg in &series_playlist {
// provider_fpl.update_playlist(plg);
// }
// // run processing pipe over new items
// for f in pipe {
// let r = f(&mut series_playlist, target);
// if let Some(v) = r {
// series_playlist = v;
// }
// }
// // assign new items to the new playlist
// for plg in &series_playlist {
// processed_fpl.update_playlist(plg);
// }
// }
// }