diff --git a/src/processing/mod.rs b/src/processing/mod.rs index a0db237a1..7ac87204b 100644 --- a/src/processing/mod.rs +++ b/src/processing/mod.rs @@ -6,4 +6,4 @@ mod playlist_watch; mod xtream_processor; mod affix_processor; mod xtream_processor_vod; -mod xtream_processor_series; \ No newline at end of file +mod xtream_processor_series; diff --git a/src/processing/playlist_processor.rs b/src/processing/playlist_processor.rs index 3736eeadf..5abd5b427 100644 --- a/src/processing/playlist_processor.rs +++ b/src/processing/playlist_processor.rs @@ -469,7 +469,7 @@ async fn process_playlist(playlists: &mut [FetchedPlaylist<'_>], let mut processed_fetched_playlists: Vec = vec![]; for provider_fpl in playlists.iter_mut() { let mut processed_fpl = execute_pipe(target, &pipe, provider_fpl); - playlist_resolve_series(target, errors, &pipe, provider_fpl, &mut processed_fpl).await; + playlist_resolve_series(cfg, target, errors, &pipe, provider_fpl, &mut processed_fpl).await; playlist_resolve_vod(cfg, target, errors, &processed_fpl).await; // stats let input_stats = stats.get_mut(&processed_fpl.input.id); diff --git a/src/processing/xtream_processor.rs b/src/processing/xtream_processor.rs index dcbcfe72c..18d46ef0b 100644 --- a/src/processing/xtream_processor.rs +++ b/src/processing/xtream_processor.rs @@ -1,48 +1,44 @@ use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::model::config::{Config, ConfigInput}; -use crate::model::playlist::{PlaylistEntry, PlaylistItem, XtreamCluster}; +use crate::model::playlist::{FetchedPlaylist, PlaylistEntry, PlaylistItem, XtreamCluster}; use crate::repository::storage::get_input_storage_path; use crate::utils::download; use serde_json::Value; -use std::collections::{HashMap}; +use std::collections::HashMap; use std::fs::{File, OpenOptions}; use std::io::{BufWriter, Error, ErrorKind, Write}; +use std::path::PathBuf; +use serde::{Deserialize, Serialize}; -const FILE_SERIES_INFO:&str = "xtream_series_info"; -const FILE_VOD_INFO:&str = "xtream_vod_info"; -const FILE_SUFFIX_WAL:&str = "wal"; +const FILE_SERIES_INFO: &str = "xtream_series_info"; +const FILE_VOD_INFO: &str = "xtream_vod_info"; +const FILE_SUFFIX_WAL: &str = "wal"; #[macro_export] macro_rules! handle_error { - ($stmt:expr) => { - if let Err(err) = $stmt { - errors.push(err); - } - }; - ($stmt:expr, $map_err:expr) => { if let Err(err) = $stmt { - errors.push($map_err(err)); + $map_err(err); } }; } -pub(in crate::processing) use handle_error; #[macro_export] macro_rules! handle_error_and_return { ($stmt:expr, $map_err:expr) => { if let Err(err) = $stmt { - errors.push($map_err(err)); - return; + $map_err(err); + return Default::default(); } }; } -pub(in crate::processing) use handle_error_and_return; +use crate::repository::bplustree::BPlusTree; +use crate::repository::xtream_repository::xtream_get_record_file_path; #[macro_export] -macro_rules! create_resolve_options_function_for_xtream_input { +macro_rules! create_resolve_options_function_for_xtream_target { ($cluster:ident) => { - paste::paste! { // Paste Makro benötigt! + paste::paste! { fn [](target: &ConfigTarget, fpl: &FetchedPlaylist) -> (bool, u16) { let (resolve, resolve_delay) = target.options.as_ref().map_or((false, 0), |opt| { @@ -101,22 +97,21 @@ pub(in crate::processing) fn write_info_content_to_wal_file(writer: &mut BufWrit Ok(()) } -pub(in crate::processing) fn create_resolve_info_wal_files(cfg: &Config, input: &ConfigInput, cluster: XtreamCluster) -> Option<(File, File)> { +pub(in crate::processing) fn create_resolve_info_wal_files(cfg: &Config, input: &ConfigInput, cluster: XtreamCluster) -> Option<(File, File, PathBuf, PathBuf)> { match get_input_storage_path(input, &cfg.working_dir) { Ok(storage_path) => { if let Some(file_prefix) = match cluster { XtreamCluster::Live => None, - XtreamCluster::Video => Some(FILE_SERIES_INFO), - XtreamCluster::Series => Some(FILE_VOD_INFO) + XtreamCluster::Video => Some(FILE_VOD_INFO), + XtreamCluster::Series => Some(FILE_SERIES_INFO) } { let content_path = storage_path.join(format!("{file_prefix}_content.{FILE_SUFFIX_WAL}")); let info_path = storage_path.join(format!("{file_prefix}_record.{FILE_SUFFIX_WAL}")); - let content_file = OpenOptions::new().append(true).open(content_path).ok()?; - let info_file = OpenOptions::new().append(true).open(info_path).ok()?; - return Some((content_file, info_file)); + let content_file = OpenOptions::new().append(true).open(&content_path).ok()?; + let info_file = OpenOptions::new().append(true).open(&info_path).ok()?; + return Some((content_file, info_file, content_path, info_path)); } None - } Err(_) => None } @@ -149,3 +144,36 @@ pub(in crate::processing) fn should_update_info(pli: &PlaylistItem, processed_pr false } } + +pub(in crate::processing) async fn read_processed_info_ids(cfg: &Config, errors: &mut Vec, fpl: &FetchedPlaylist<'_>, + cluster: XtreamCluster, extract_ts: F) -> HashMap +where + F: Fn(&V) -> u64, + V: Serialize + for<'de> Deserialize<'de> + Clone, +{ + let mut processed_info_ids = HashMap::new(); + + let file_path = match get_input_storage_path(fpl.input, &cfg.working_dir) + .map(|storage_path| xtream_get_record_file_path(&storage_path, cluster)).and_then(|opt| opt.ok_or_else(|| Error::new(ErrorKind::Other, "Not supported".to_string()))) + { + Ok(file_path) => file_path, + Err(err) => { + let fpl_name = fpl.input.name.as_ref().map_or("?", String::as_str); + errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create storage path for input {fpl_name}: {err}"))); + return processed_info_ids; + } + }; + + match cfg.file_locks.read_lock(&file_path).await { + Ok(file_lock) => { + if let Ok(info_records) = BPlusTree::::load(&file_path) { + info_records.iter().for_each(|(provider_id, record)| { + processed_info_ids.insert(*provider_id, extract_ts(record)); + }); + } + drop(file_lock); + } + Err(err) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}"))), + } + processed_info_ids +} diff --git a/src/processing/xtream_processor_series.rs b/src/processing/xtream_processor_series.rs index 0e22915e3..6d2180b21 100644 --- a/src/processing/xtream_processor_series.rs +++ b/src/processing/xtream_processor_series.rs @@ -1,23 +1,23 @@ -use std::collections::HashMap; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::model::config::{Config, ConfigTarget, InputType}; -use crate::model::playlist::{FetchedPlaylist, PlaylistItem, PlaylistItemType, XtreamCluster}; +use crate::model::playlist::{FetchedPlaylist, PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster}; +use crate::processing::playlist_processor::ProcessingPipe; +use crate::processing::xtream_processor::{create_resolve_info_wal_files, get_u32_from_serde_value, get_u64_from_serde_value, playlist_resolve_process_playlist_item, read_processed_info_ids, should_update_info, write_info_content_to_wal_file}; +use crate::repository::xtream_repository::{xtream_get_info_file_paths, xtream_update_input_info_file, xtream_update_input_series_record_from_wal_file}; +use crate::{create_resolve_options_function_for_xtream_target, handle_error, handle_error_and_return}; use serde_json::{Map, Value}; +use std::collections::HashMap; use std::fs::File; use std::io::{BufWriter, Write}; -use crate::{create_resolve_options_function_for_xtream_input, handle_error, handle_error_and_return}; -use crate::processing::playlist_processor::ProcessingPipe; -use crate::processing::xtream_processor::{create_resolve_info_wal_files, get_u32_from_serde_value, - get_u64_from_serde_value, playlist_resolve_process_playlist_item, - should_update_info, write_info_content_to_wal_file}; +use log::error; use crate::repository::bplustree::BPlusTree; +use crate::repository::IndexedDocumentIndex; use crate::repository::storage::get_input_storage_path; -use crate::repository::xtream_repository::{xtream_get_record_file_path, xtream_update_input_info_file, xtream_update_input_record_from_wal_file}; const TAG_SERIES_INFO_SERIES_ID: &str = "series_id"; const TAG_SERIES_INFO_LAST_MODIFIED: &str = "last_modified"; -create_resolve_options_function_for_xtream_input!(series); +create_resolve_options_function_for_xtream_target!(series); fn write_series_info_tmdb_to_wal_file(writer: &mut BufWriter<&File>, provider_id: u32, tmdb_id: u32) -> std::io::Result<()> { writer.write_all(&provider_id.to_le_bytes())?; @@ -26,77 +26,58 @@ fn write_series_info_tmdb_to_wal_file(writer: &mut BufWriter<&File>, provider_id } async fn read_processed_series_info_ids(cfg: &Config, errors: &mut Vec, fpl: &FetchedPlaylist<'_>, - cluster: XtreamCluster) -> HashMap { - let mut processed_info_ids = HashMap::new(); - { - match get_input_storage_path(fpl.input, &cfg.working_dir) - .map(|storage_path| xtream_get_record_file_path(&storage_path, cluster)).await { - Ok(file_path) => { - match cfg.file_locks.read_lock(&file_path).await { - Ok(file_lock) => { - if let Ok(info_records) = BPlusTree::::load(&file_path) { - info_records.traverse(|keys, timestamps| { - for (provider_id, ts) in keys.iter().zip(timestamps.iter()) { - processed_info_ids.insert(*provider_id, *ts); - } - }); - } - drop(file_lock); - } - Err(err) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}"))), - } - } - Ok(None) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create storage path for input {}", &fpl.input.name.as_ref().map_or("?", |v| v)))), - Err(err) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create storage path for input {err}"))), - } - } - processed_info_ids + cluster: XtreamCluster) -> HashMap { + read_processed_info_ids(cfg, errors, fpl, cluster, |ts: &u64| *ts).await } fn extract_info_record_from_series_info(content: &str) -> Option<(u32, u64)> { let doc = serde_json::from_str::>(content).ok()?; let provider_id = get_u32_from_serde_value(doc.get(TAG_SERIES_INFO_SERIES_ID)?)?; - let last_modified = get_u64_from_serde_value(doc.get(TAG_SERIES_INFO_LAST_MODIFIED)?) - .unwrap_or(0); + let last_modified = get_u64_from_serde_value(doc.get(TAG_SERIES_INFO_LAST_MODIFIED)?).unwrap_or(0); Some((provider_id, last_modified)) } +fn write_series_info_record_to_wal_file( + writer: &mut BufWriter<&File>, + provider_id: u32, + ts: u64, +) -> std::io::Result<()> { + writer.write_all(&provider_id.to_le_bytes())?; + writer.write_all(&ts.to_le_bytes())?; + Ok(()) +} + fn should_update_series_info(pli: &PlaylistItem, processed_provider_ids: &HashMap) -> bool { should_update_info(pli, processed_provider_ids, "last_modified") } -pub async fn playlist_resolve_series(cfg: &Config, target: &ConfigTarget, errors: &mut Vec, - pipe: &ProcessingPipe, - provider_fpl: &mut FetchedPlaylist<'_>, - processed_fpl: &mut FetchedPlaylist<'_> -) { - let (resolve_series, resolve_delay) = get_resolve_series_options(target, processed_fpl); - if !resolve_series { return; } - +async fn playlist_resolve_series_info(cfg: &Config, errors: &mut Vec, + processed_fpl: &mut FetchedPlaylist<'_>, resolve_delay: u16) -> HashMap{ // 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((mut wal_file_content, mut wal_file_record)) = create_resolve_info_wal_files(cfg, processed_fpl.input, XtreamCluster::Series) - else { return; }; + let Some((mut wal_content_file, mut wal_record_file, wal_content_path, wal_record_path)) = create_resolve_info_wal_files(cfg, processed_fpl.input, XtreamCluster::Series) + else { return HashMap::new(); }; let mut processed_info_ids = read_processed_series_info_ids(cfg, errors, processed_fpl, XtreamCluster::Series).await; - let mut content_writer = BufWriter::new(&wal_file_content); - let mut record_writer = BufWriter::new(&wal_file_record); + 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() .filter(|&plg| plg.xtream_cluster == XtreamCluster::Series) .flat_map(|plg| &plg.channels) .filter(|&pli| pli.header.borrow().item_type == PlaylistItemType::SeriesInfo) - .filter(|&pli| should_update_series_info(pli, &processed_info_ids)) { - // content contains all episodes - if let Some(content) = playlist_resolve_process_playlist_item(pli, processed_fpl.input, errors, resolve_delay, XtreamCluster::Series).await { - if let Some((provider_id, ts)) = extract_info_record_from_series_info(&content) { - handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &content), |err| M3uFilterError::new( M3uFilterErrorKind::Notify, format!("Failed to resolve series, could not write to content wal file {err}"))); - processed_info_ids.insert(provider_id, ts); - handle_error_and_return!(write_info_record_to_wal_file(&mut record_writer, provider_id, &info_record), |err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve series wal, could not write to record wal file {err}"))); - content_updated = true; + let should_update = should_update_series_info(pli, &processed_info_ids); + if should_update { + if let Some(content) = playlist_resolve_process_playlist_item(pli, processed_fpl.input, errors, resolve_delay, XtreamCluster::Series).await { + if let Some((provider_id, ts)) = extract_info_record_from_series_info(&content) { + handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &content), |err| M3uFilterError::new( M3uFilterErrorKind::Notify, format!("Failed to resolve series, could not write to content wal file {err}"))); + processed_info_ids.insert(provider_id, ts); + handle_error_and_return!(write_series_info_record_to_wal_file(&mut record_writer, provider_id, ts), |err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve series wal, could not write to record wal file {err}"))); + content_updated = true; + } } } } @@ -105,12 +86,57 @@ pub async fn playlist_resolve_series(cfg: &Config, target: &ConfigTarget, errors drop(content_writer); handle_error!(record_writer.flush(), |err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve vod tmdb, could not write to wal file {err}"))); drop(record_writer); - handle_error!(xtream_update_input_info_file(cfg, processed_fpl.input, &mut wal_file_content, XtreamCluster::Series).await); - handle_error!(xtream_update_input_record_from_wal_file(cfg, processed_fpl.input, &mut wal_file_record, XtreamCluster::Series).await); + handle_error!(xtream_update_input_info_file(cfg, processed_fpl.input, &mut wal_content_file, &wal_content_path, XtreamCluster::Series).await, |err| errors.push(err)); + handle_error!(xtream_update_input_series_record_from_wal_file(cfg, processed_fpl.input, &mut wal_record_file, &wal_record_path).await, |err| errors.push(err)); + } + + processed_info_ids +} + +pub async fn playlist_resolve_series(cfg: &Config, target: &ConfigTarget, + errors: &mut Vec, + pipe: &ProcessingPipe, + provider_fpl: &mut FetchedPlaylist<'_>, + processed_fpl: &mut FetchedPlaylist<'_>, +) { + let (resolve_series, resolve_delay) = get_resolve_series_options(target, processed_fpl); + if !resolve_series { return; } + + let processed_ids = playlist_resolve_series_info(cfg, errors, processed_fpl, resolve_delay).await; + if processed_ids.is_empty() { return; } + + if let Ok(Some((info_path, idx_path))) = get_input_storage_path(provider_fpl.input, &cfg.working_dir).map(|storage_path| xtream_get_info_file_paths(&storage_path, XtreamCluster::Series)) { + match cfg.file_locks.read_lock(&info_path).await { + Ok(_file_lock) => { + let index = IndexedDocumentIndex::::load(&idx_path).unwrap_or_else(|err| { + error!("Failed to load index {idx_path:?}: {err}"); + IndexedDocumentIndex::::new() + }); + let mut result: Vec = vec![]; + for plg in &mut provider_fpl.playlistgroups.iter() + .filter(|&plg| plg.xtream_cluster == XtreamCluster::Series) + { + let mut group_series: Vec = vec![]; + for pli in plg.channels.iter().filter(|&pli| pli.header.borrow().item_type == PlaylistItemType::SeriesInfo) { + if let Some(provider_id) = pli.header.borrow_mut().get_provider_id() { + if processed_ids.contains_key(&provider_id) { + if let Some(offset) = index.query(&provider_id) { + IndexedDocumentReader:: + } + } + } + } + } + }, + Err(err) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, "Could not lock input info file for series".to_string())) + } + } else { + errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, "Failed to open input info file for series".to_string())); } - Now update provider_fpl + // Now update provider_fpl + } diff --git a/src/processing/xtream_processor_vod.rs b/src/processing/xtream_processor_vod.rs index d415c3049..bd2b163af 100644 --- a/src/processing/xtream_processor_vod.rs +++ b/src/processing/xtream_processor_vod.rs @@ -1,18 +1,13 @@ -use std::collections::HashMap; -use std::fs::File; -use crate::{create_resolve_options_function_for_xtream_input, handle_error, handle_error_and_return}; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::model::config::{Config, ConfigTarget, InputType}; use crate::model::playlist::{FetchedPlaylist, PlaylistItem, XtreamCluster}; -use crate::processing::xtream_processor::{create_resolve_info_wal_files, get_u32_from_serde_value, - get_u64_from_serde_value, playlist_resolve_process_playlist_item, - should_update_info, write_info_content_to_wal_file}; -use crate::repository::xtream_repository::{xtream_get_record_file_path, xtream_update_input_info_file, - xtream_update_input_record_from_wal_file}; +use crate::processing::xtream_processor::{create_resolve_info_wal_files, get_u32_from_serde_value, get_u64_from_serde_value, playlist_resolve_process_playlist_item, read_processed_info_ids, should_update_info, write_info_content_to_wal_file}; +use crate::repository::xtream_repository::{xtream_update_input_info_file, xtream_update_input_vod_record_from_wal_file, InputVodInfoRecord}; +use crate::{create_resolve_options_function_for_xtream_target, handle_error, handle_error_and_return}; use serde_json::{Map, Value}; +use std::collections::HashMap; +use std::fs::File; use std::io::{BufWriter, Write}; -use crate::repository::bplustree::BPlusTree; -use crate::repository::storage::get_input_storage_path; const TAG_VOD_INFO_INFO: &str = "info"; const TAG_VOD_INFO_MOVIE_DATA: &str = "movie_data"; @@ -20,39 +15,11 @@ const TAG_VOD_INFO_TMDB_ID: &str = "tmdb_id"; const TAG_VOD_INFO_STREAM_ID: &str = "stream_id"; const TAG_VOD_INFO_ADDED: &str = "added"; -struct InputVodInfoRecord { - pub(crate) tmdb_id: u32, - pub(crate) ts: u64, -} - -create_resolve_options_function_for_xtream_input!(video); +create_resolve_options_function_for_xtream_target!(video); async fn read_processed_vod_info_ids(cfg: &Config, errors: &mut Vec, fpl: &FetchedPlaylist<'_>, cluster: XtreamCluster) -> HashMap { - let mut processed_info_ids = HashMap::new(); - { - match get_input_storage_path(fpl.input, &cfg.working_dir) - .map(|storage_path| xtream_get_record_file_path(&storage_path, cluster)).await { - Ok(file_path) => { - match cfg.file_locks.read_lock(&file_path).await { - Ok(file_lock) => { - if let Ok(info_records) = BPlusTree::::load(&file_path) { - info_records.traverse(|keys, records| { - for (provider_id, record) in keys.iter().zip(records.iter()) { - processed_info_ids.insert(*provider_id, *record.ts); - } - }); - } - drop(file_lock); - } - Err(err) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}"))), - } - } - Ok(None) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create storage path for input {}", &fpl.input.name.as_ref().map_or("?", |v| v)))), - Err(err) => errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create storage path for input {err}"))), - } - } - processed_info_ids + read_processed_info_ids(cfg, errors, fpl, cluster, |record: &InputVodInfoRecord| record.ts).await } fn extract_info_record_from_vod_info(content: &str) -> Option<(u32, InputVodInfoRecord)> { @@ -68,8 +35,7 @@ fn extract_info_record_from_vod_info(content: &str) -> Option<(u32, InputVodInfo .and_then(get_u64_from_serde_value) .unwrap_or(0); - let info = doc.get(TAG_VOD_INFO_INFO)?.as_object(); - let tmdb_id = info + let tmdb_id = doc.get(TAG_VOD_INFO_INFO)?.as_object() .and_then(|info| info.get(TAG_VOD_INFO_TMDB_ID)) .and_then(get_u32_from_serde_value) .unwrap_or(0); @@ -80,7 +46,6 @@ fn extract_info_record_from_vod_info(content: &str) -> Option<(u32, InputVodInfo })) } - fn write_vod_info_record_to_wal_file( writer: &mut BufWriter<&File>, provider_id: u32, @@ -92,10 +57,8 @@ fn write_vod_info_record_to_wal_file( Ok(()) } - - fn should_update_vod_info(pli: &PlaylistItem, processed_provider_ids: &HashMap) -> bool { - should_update_info(pli, processed_provider_ids, "added") + should_update_info(pli, processed_provider_ids, TAG_VOD_INFO_ADDED) } pub async fn playlist_resolve_vod(cfg: &Config, target: &ConfigTarget, errors: &mut Vec, fpl: &FetchedPlaylist<'_>) { @@ -105,35 +68,37 @@ pub async fn playlist_resolve_vod(cfg: &Config, target: &ConfigTarget, errors: & // 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((mut wal_file_content, mut wal_file_record)) = create_resolve_info_wal_files(cfg, fpl.input, XtreamCluster::Video) + let Some((mut wal_content_file, mut wal_record_file, wal_content_path, wal_record_path)) = create_resolve_info_wal_files(cfg, fpl.input, XtreamCluster::Video) else { return; }; let mut processed_info_ids = read_processed_vod_info_ids(cfg, errors, fpl, XtreamCluster::Video).await; - let mut content_writer = BufWriter::new(&wal_file_content); - let mut record_writer = BufWriter::new(&wal_file_record); + 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 fpl.playlistgroups.iter() .flat_map(|plg| &plg.channels) - .filter(|&chan| chan.header.borrow().xtream_cluster == XtreamCluster::Video) - .filter(|&chan| should_update_vod_info(chan, &processed_info_ids)) { + .filter(|&pli| pli.header.borrow().xtream_cluster == XtreamCluster::Video) { + let should_update = should_update_vod_info(pli, &processed_info_ids); + if should_update { if let Some(content) = playlist_resolve_process_playlist_item(pli, fpl.input, errors, resolve_delay, XtreamCluster::Video).await { if let Some((provider_id, info_record)) = extract_info_record_from_vod_info(&content) { let ts = info_record.ts; - handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &content), |err| M3uFilterError::new( M3uFilterErrorKind::Notify, format!("Failed to resolve vod, could not write to content wal file {err}"))); + handle_error_and_return!(write_info_content_to_wal_file(&mut content_writer, provider_id, &content), |err| errors.push(M3uFilterError::new( M3uFilterErrorKind::Notify, format!("Failed to resolve vod, could not write to content wal file {err}")))); processed_info_ids.insert(provider_id, ts); handle_error_and_return!(write_vod_info_record_to_wal_file(&mut record_writer, provider_id, - &info_record), |err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve vod wal, could not write to record wal file {err}"))); + &info_record), |err| errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve vod wal, could not write to record wal file {err}")))); content_updated = true; } } + } } if content_updated { - handle_error!(content_writer.flush(), |err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve vod, could not write to wal file {err}"))); + handle_error!(content_writer.flush(), |err| errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve vod, could not write to wal file {err}")))); drop(content_writer); - handle_error!(record_writer.flush(), |err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve vod tmdb, could not write to wal file {err}"))); + handle_error!(record_writer.flush(), |err| errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Failed to resolve vod tmdb, could not write to wal file {err}")))); drop(record_writer); - handle_error!(xtream_update_input_info_file(cfg, fpl.input, &mut wal_file_content, XtreamCluster::Video).await); - handle_error!(xtream_update_input_record_from_wal_file(cfg, fpl.input, &mut wal_file_record, XtreamCluster::Video).await); + handle_error!(xtream_update_input_info_file(cfg, fpl.input, &mut wal_content_file, &wal_content_path, XtreamCluster::Video).await, |err| errors.push(err)); + handle_error!(xtream_update_input_vod_record_from_wal_file(cfg, fpl.input, &mut wal_record_file, &wal_record_path).await, |err| errors.push(err)); } } diff --git a/src/repository/bplustree.rs b/src/repository/bplustree.rs index 0cdea080d..0bbbbc8fe 100644 --- a/src/repository/bplustree.rs +++ b/src/repository/bplustree.rs @@ -591,8 +591,84 @@ where } } +pub struct BPlusTreeIterator<'a, K, V> { + stack: Vec<&'a BPlusTreeNode>, + current_keys: Option<&'a [K]>, + current_values: Option<&'a [V]>, + index: usize, +} + +impl<'a, K, V> BPlusTreeIterator<'a, K, V> +where + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, + V: Serialize + for<'de> Deserialize<'de> + Clone, +{ + pub fn new(tree: &'a BPlusTree) -> Self { + let mut stack = Vec::new(); + stack.push(&tree.root); + Self { + stack, + current_keys: None, + current_values: None, + index: 0, + } + } +} + +impl<'a, K, V> Iterator for BPlusTreeIterator<'a, K, V> +where + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, + V: Serialize + for<'de> Deserialize<'de> + Clone, +{ + type Item = (&'a K, &'a V); + + fn next(&mut self) -> Option { + // Iterate over the current node + if let Some(keys) = self.current_keys { + if let Some(values) = self.current_values { + if self.index < keys.len() { + let key = &keys[self.index]; + let value = &values[self.index]; + self.index += 1; + return Some((key, value)); + } + } + } + + // Move to the next node + while let Some(node) = self.stack.pop() { + if !node.is_leaf { + // Push children in reverse order to maintain traversal order + for child in node.children.iter().rev() { + self.stack.push(child); + } + } + + if node.is_leaf { + self.current_keys = Some(&node.keys); + self.current_values = Some(&node.values); + self.index = 0; + return self.next(); // Process the new leaf node + } + } + + None // No more elements + } +} + +impl BPlusTree +where + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, + V: Serialize + for<'de> Deserialize<'de> + Clone, +{ + pub fn iter(&self) -> BPlusTreeIterator { + BPlusTreeIterator::new(self) + } +} + #[cfg(test)] mod tests { + use std::collections::HashSet; use std::io; use std::path::PathBuf; @@ -678,28 +754,6 @@ mod tests { Ok(()) } - // #[test] - // fn traverse_test() -> io::Result<()> { - // let mut tree = BPlusTree::::new(); - // for i in 0u32..=500 { - // tree.insert(i, Record { - // id: i, - // data: format!("Entry {i}"), - // }); - // } - // let filepath = PathBuf::from("/tmp/tree.bin"); - // // Serialize the tree to a file - // tree.store(&filepath)?; - // - // let mut tree_query: BPlusTreeQuery = BPlusTreeQuery::try_new(&filepath)?; - // - // // Traverse the tree - // tree_query.traverse(|keys, values| { - // // TODO real test - // println!("Node: {:?} {:?}", keys, values); - // }); - // Ok(()) - // } #[test] fn insert_dulplicate_test() -> io::Result<()> { @@ -725,4 +779,30 @@ mod tests { Ok(()) } -} \ No newline at end of file + + #[test] + fn iterator_test() -> io::Result<()> { + let mut tree = BPlusTree::::new(); + let mut entry_set = HashSet::new(); + for i in 0u32..=500 { + tree.insert(i, Record { + id: i, + data: format!("Entry {i}"), + }); + entry_set.insert(i); + } + let filepath = PathBuf::from("/tmp/tree.bin"); + // Serialize the tree to a file + tree.store(&filepath)?; + + let tree: BPlusTree = BPlusTree::load(&filepath)?; + + // Traverse the tree + for (key, value) in tree.iter() { + assert!(format!("Entry {}", key).eq(&value.data), "Wrong entry"); + entry_set.remove(key); + } + assert!(entry_set.is_empty()); + Ok(()) + } +} diff --git a/src/repository/indexed_document.rs b/src/repository/indexed_document.rs index 41aa8d159..21b7fde4f 100644 --- a/src/repository/indexed_document.rs +++ b/src/repository/indexed_document.rs @@ -334,6 +334,17 @@ where } Err(Error::new(ErrorKind::Other, format!("Failed to read item for id {:?} - {}", doc_id, main_path.to_str().unwrap()))) } + + pub(in crate::repository) fn read_indexed_item_at_offset(main_file: &mut File, offset: u64, doc_id: &K) -> Result { + main_file.seek(SeekFrom::Start(offset))?; + let buf_size = IndexedDocument::read_content_size(main_file)?; + let mut buffer: Vec = vec![0; buf_size]; + main_file.read_exact(&mut buffer)?; + if let Ok(item) = bincode::deserialize::(&buffer) { + return Ok(item); + } + Err(Error::new(ErrorKind::Other, format!("Failed to read item for id {:?}", doc_id))) + } } impl Iterator for IndexedDocumentReader diff --git a/src/repository/kodi_repository.rs b/src/repository/kodi_repository.rs index ff640f010..4abf2f94e 100644 --- a/src/repository/kodi_repository.rs +++ b/src/repository/kodi_repository.rs @@ -1,7 +1,7 @@ use crate::create_m3u_filter_error_result; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::model::config::{Config, ConfigTarget}; -use crate::model::playlist::PlaylistGroup; +use crate::model::playlist::{PlaylistGroup, XtreamCluster}; use crate::repository::bplustree::BPlusTree; use crate::repository::storage::get_input_storage_path; use crate::utils::file_lock_manager::FileReadGuard; @@ -12,6 +12,7 @@ use std::collections::HashMap; use std::fs::File; use std::io::Write; use std::sync::LazyLock; +use crate::repository::xtream_repository::xtream_get_record_file_path; struct KodiStyle { year: regex::Regex, @@ -92,7 +93,7 @@ static KODY_STYLE: LazyLock = LazyLock::new(|| KodiStyle { type InputTmdbIndexMap = HashMap)>>; async fn get_tmdb_id(cfg: &Config, provider_id: Option, input_id: u16, - input_indexes: &mut InputTmdbIndexMap) -> Option { + input_indexes: &mut InputTmdbIndexMap, cluster: XtreamCluster) -> Option { match provider_id { None => None, Some(pid) => { @@ -106,7 +107,7 @@ async fn get_tmdb_id(cfg: &Config, provider_id: Option, input_id: u16, } std::collections::hash_map::Entry::Vacant(entry) => { if let Some(input) = cfg.get_input_by_id(input_id) { - if let Ok(tmdb_path) = get_input_storage_path(input, &cfg.working_dir) + if let Ok(Some(tmdb_path)) = get_input_storage_path(input, &cfg.working_dir) .map(|storage_path| xtream_get_record_file_path(&storage_path, cluster)) { if let Ok(file_lock) = cfg.file_locks.read_lock(&tmdb_path).await { if let Ok(tree) = BPlusTree::::load(&tmdb_path) { @@ -158,7 +159,7 @@ pub async fn kodi_write_strm_playlist(target: &ConfigTarget, cfg: &Config, new_p kodi_file_name = kodi_style_filename; kodi_file_dir_name.iter().for_each(|p| dir_path = dir_path.join(p)); - let tmdb_id = get_tmdb_id(cfg, provider_id, input_id, &mut input_tmdb_indexes).await; + let tmdb_id = get_tmdb_id(cfg, provider_id, input_id, &mut input_tmdb_indexes, header.xtream_cluster).await; additional_info = match tmdb_id { None => { String::new() } Some(id) => { format!(" {{tmdb={id}}}") } diff --git a/src/repository/m3u_playlist_iterator.rs b/src/repository/m3u_playlist_iterator.rs index 3b7958635..3c7e0a480 100644 --- a/src/repository/m3u_playlist_iterator.rs +++ b/src/repository/m3u_playlist_iterator.rs @@ -40,12 +40,7 @@ impl M3uPlaylistIterator { })?; let reader = - IndexedDocumentReader::::new(&m3u_path, &idx_path).map_err(|err| { - M3uFilterError::new( - M3uFilterErrorKind::Info, - format!("Could not deserialize file {m3u_path:?} - {err}"), - ) - })?; + IndexedDocumentReader::::new(&m3u_path, &idx_path).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Info,format!("Could not deserialize file {m3u_path:?} - {err}")))?; let target_options = target.options.as_ref(); let include_type_in_url = target_options.is_some_and( |opts| opts.m3u_include_type_in_url); diff --git a/src/repository/mod.rs b/src/repository/mod.rs index e4caf9039..d78c54e5d 100644 --- a/src/repository/mod.rs +++ b/src/repository/mod.rs @@ -2,7 +2,7 @@ pub mod storage; pub mod target_id_mapping; pub mod bplustree; mod indexed_document; -pub use indexed_document::{IndexedDocumentIndex}; +pub use indexed_document::IndexedDocumentIndex; pub mod playlist_repository; pub mod m3u_repository; pub mod xtream_repository; diff --git a/src/repository/storage.rs b/src/repository/storage.rs index a96224254..0285e43d2 100644 --- a/src/repository/storage.rs +++ b/src/repository/storage.rs @@ -52,7 +52,6 @@ pub fn get_target_storage_path(cfg: &Config, target_name: &str) -> Option std::io::Result { let name = format!("input_{}", input.name.clone().unwrap_or_else(|| format!("{}", input.id))); let path = Path::new(working_dir).join(name); - // Create the directory and return the path or propagate the error std::fs::create_dir_all(&path).map(|()| path) } diff --git a/src/repository/xtream_repository.rs b/src/repository/xtream_repository.rs index 2d272f5b2..7d2a1de4b 100644 --- a/src/repository/xtream_repository.rs +++ b/src/repository/xtream_repository.rs @@ -10,19 +10,11 @@ use serde_json::{json, Map, Value}; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::model::config::{Config, ConfigInput, ConfigTarget}; -use crate::model::playlist::{ - PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster, XtreamPlaylistItem, -}; +use crate::model::playlist::{PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster, XtreamPlaylistItem}; use crate::model::xtream::XtreamMappingOptions; -use crate::repository::bplustree::BPlusTree; -use crate::repository::indexed_document::{ - IndexedDocumentGarbageCollector, IndexedDocumentQuery, IndexedDocumentReader, - IndexedDocumentUpdate, IndexedDocumentWriter, -}; -use crate::repository::storage::{ - get_input_storage_path, get_target_id_mapping_file, get_target_storage_path, hash_string, - FILE_SUFFIX_DB, FILE_SUFFIX_INDEX, -}; +use crate::repository::bplustree::{BPlusTree}; +use crate::repository::indexed_document::{IndexedDocumentGarbageCollector, IndexedDocumentQuery, IndexedDocumentReader,IndexedDocumentUpdate, IndexedDocumentWriter}; +use crate::repository::storage::{get_input_storage_path, get_target_id_mapping_file, get_target_storage_path, hash_string,FILE_SUFFIX_DB, FILE_SUFFIX_INDEX}; use crate::repository::target_id_mapping::{TargetIdMapping, VirtualIdRecord}; use crate::repository::xtream_playlist_iterator::XtreamPlaylistIterator; use crate::utils::json_utils::{json_iter_array, json_write_documents_to_file}; @@ -65,6 +57,13 @@ macro_rules! try_option_ok { }; } + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +pub struct InputVodInfoRecord { + pub(crate) tmdb_id: u32, + pub(crate) ts: u64, +} + fn get_collection_path(path: &Path, collection: &str) -> PathBuf { path.join(format!("{collection}.json")) } @@ -80,8 +79,7 @@ fn ensure_xtream_storage_path(cfg: &Config, target_name: &str) -> Result { for item in playlist { @@ -136,9 +130,7 @@ async fn write_playlists_to_file( Err(err) => return Err(cant_write_result!(&xtream_path, err)), } } - writer - .store() - .map_err(|err| cant_write_result!(&xtream_path, err))?; + writer.store().map_err(|err| cant_write_result!(&xtream_path, err))?; } Err(err) => return Err(cant_write_result!(&xtream_path, err)), } @@ -159,19 +151,15 @@ fn get_map_item_as_str(map: &serde_json::Map, key: &str) -> Optio fn load_old_category_ids(path: &Path) -> (u32, HashMap) { let mut result: HashMap = HashMap::new(); let mut max_id: u32 = 0; - for col_path in [ - get_collection_path(path, COL_CAT_LIVE), - get_collection_path(path, COL_CAT_VOD), - get_collection_path(path, COL_CAT_SERIES), - ] { + for cat in [COL_CAT_LIVE, COL_CAT_VOD, COL_CAT_SERIES] { + let col_path = get_collection_path(path, cat); if col_path.exists() { if let Ok(file) = File::open(col_path) { let reader = BufReader::new(file); for entry in json_iter_array::>(reader).flatten() { if let Some(item) = entry.as_object() { if let Some(category_id) = get_map_item_as_str(item, TAG_CATEGORY_ID) { - if let Some(category_name) = - get_map_item_as_str(item, TAG_CATEGORY_NAME) + if let Some(category_name) = get_map_item_as_str(item, TAG_CATEGORY_NAME) { if let Ok(cat_id) = category_id.parse::() { result.insert(category_name, cat_id); @@ -188,8 +176,7 @@ fn load_old_category_ids(path: &Path) -> (u32, HashMap) { } pub fn xtream_get_storage_path(cfg: &Config, target_name: &str) -> Option { - get_target_storage_path(cfg, target_name) - .map(|target_path| target_path.join(PathBuf::from(PATH_XTREAM))) + get_target_storage_path(cfg, target_name).map(|target_path| target_path.join(PathBuf::from(PATH_XTREAM))) } pub fn xtream_get_epg_file_path(path: &Path) -> PathBuf { @@ -253,8 +240,7 @@ pub async fn xtream_write_playlist( XtreamCluster::Live => &mut cat_live_col, XtreamCluster::Series => &mut cat_series_col, XtreamCluster::Video => &mut cat_vod_col, - } - .push(json!({ + }.push(json!({ TAG_CATEGORY_ID: format!("{}", &cat_id), TAG_CATEGORY_NAME: plg.title.clone(), TAG_PARENT_ID: 0 @@ -302,11 +288,7 @@ pub async fn xtream_write_playlist( match json_write_documents_to_file(&col_path, data) { Ok(()) => {} Err(err) => { - errors.push(format!( - "Persisting collection failed: {}: {}", - &col_path.to_str().unwrap(), - err - )); + errors.push(format!("Persisting collection failed: {}: {}", &col_path.to_str().unwrap(), err)); } } } @@ -319,9 +301,7 @@ pub async fn xtream_write_playlist( (XtreamCluster::Video, &mut vod_col), (XtreamCluster::Series, &mut series_col), ], - ) - .await - { + ).await { Ok(()) => { if let Err(err) = xtream_garbage_collect(cfg, &target.name).await { if err.kind() != ErrorKind::NotFound { @@ -387,11 +367,7 @@ async fn xtream_read_series_item_for_stream_id( let (xtream_path, idx_path) = xtream_get_file_paths_for_series(storage_path); { let _file_lock = cfg.file_locks.read_lock(&xtream_path).await?; - IndexedDocumentReader::::read_indexed_item( - &xtream_path, - &idx_path, - &stream_id, - ) + IndexedDocumentReader::::read_indexed_item(&xtream_path, &idx_path, &stream_id) } } @@ -399,15 +375,7 @@ macro_rules! try_cluster { ($xtream_cluster:expr, $item_type:expr, $virtual_id:expr) => { $xtream_cluster .or_else(|| XtreamCluster::try_from($item_type).ok()) - .ok_or_else(|| { - Error::new( - ErrorKind::Other, - format!( - "Could not determine cluster for xtream item with stream-id {}", - $virtual_id - ), - ) - }) + .ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not determine cluster for xtream item with stream-id {}",$virtual_id))) }; } @@ -417,82 +385,26 @@ pub async fn xtream_get_item_for_stream_id( target: &ConfigTarget, xtream_cluster: Option, ) -> Result { - let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| { - Error::new( - ErrorKind::Other, - format!("Could not find path for target {}", &target.name), - ) - })?; - let storage_path = xtream_get_storage_path(config, target.name.as_str()).ok_or_else(|| { - Error::new( - ErrorKind::Other, - format!( - "Could not find path for target {} xtream output", - &target.name - ), - ) - })?; + let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find path for target {}", &target.name)))?; + let storage_path = xtream_get_storage_path(config, target.name.as_str()).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find path for target {} xtream output", &target.name)))?; { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config - .file_locks - .read_lock(&target_id_mapping_file) - .await - .map_err(|err| { - Error::new( - ErrorKind::Other, - format!( - "Could not get lock for id mapping for target {} err:{err}", - target.name - ), - ) - })?; - - let mut target_id_mapping = - IndexedDocumentQuery::::try_new(&target_id_mapping_file) - .map_err(|err| { - Error::new( - ErrorKind::Other, - format!( - "Could not load id mapping for target {} err:{err}", - target.name - ), - ) - })?; - - let mapping = target_id_mapping.query(&virtual_id).ok_or_else(|| { - Error::new( - ErrorKind::Other, - format!( - "Could not find mapping for target {} and id {}", - target.name, virtual_id - ), - ) - })?; + let _file_lock = config.file_locks.read_lock(&target_id_mapping_file).await.map_err(|err| Error::new(ErrorKind::Other, format!("Could not get lock for id mapping for target {} err:{err}", target.name)))?; + let mut target_id_mapping = IndexedDocumentQuery::::try_new(&target_id_mapping_file).map_err(|err| Error::new(ErrorKind::Other, format!("Could not load id mapping for target {} err:{err}", target.name)))?; + let mapping = target_id_mapping.query(&virtual_id).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find mapping for target {} and id {}", target.name, virtual_id)))?; match mapping.item_type { PlaylistItemType::SeriesInfo => { xtream_read_series_item_for_stream_id(config, virtual_id, &storage_path).await } PlaylistItemType::SeriesEpisode => { - let mut item = xtream_read_series_item_for_stream_id( - config, - mapping.parent_virtual_id, - &storage_path, - ) - .await?; + let mut item = xtream_read_series_item_for_stream_id(config, mapping.parent_virtual_id, &storage_path).await?; item.provider_id = mapping.provider_id; Ok(item) } PlaylistItemType::Catchup => { let cluster = try_cluster!(xtream_cluster, mapping.item_type, virtual_id)?; - let mut item = xtream_read_item_for_stream_id( - config, - mapping.parent_virtual_id, - &storage_path, - cluster, - ) - .await?; + let mut item = xtream_read_item_for_stream_id(config, mapping.parent_virtual_id, &storage_path, cluster).await?; item.provider_id = mapping.provider_id; Ok(item) } @@ -509,10 +421,8 @@ pub async fn xtream_load_rewrite_playlist( config: &Config, target: &ConfigTarget, category_id: u32, -) -> Result>, M3uFilterError> { - Ok(Box::new( - XtreamPlaylistIterator::new(cluster, config, target, category_id).await?, - )) +) -> Result>, M3uFilterError> { + Ok(Box::new(XtreamPlaylistIterator::new(cluster, config, target, category_id).await?)) } pub async fn xtream_write_series_info( @@ -531,24 +441,13 @@ pub async fn xtream_write_series_info( { let _file_lock = config.file_locks.write_lock(&info_path).await?; let mut writer = IndexedDocumentWriter::new_append(info_path, idx_path)?; - writer.write_doc(series_info_id, content).map_err(|_| { - Error::new( - ErrorKind::Other, - format!("failed to write xtream series info for target {target_name}"), - ) - })?; - + writer.write_doc(series_info_id, content).map_err(|_| Error::new(ErrorKind::Other, format!("failed to write xtream series info for target {target_name}")))?; writer.store()?; } { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config - .file_locks - .write_lock(&target_id_mapping_file) - .await?; - if let Ok(mut target_id_mapping) = - IndexedDocumentUpdate::::try_new(&target_id_mapping_file) - { + let _file_lock = config.file_locks.write_lock(&target_id_mapping_file).await?; + if let Ok(mut target_id_mapping) = IndexedDocumentUpdate::::try_new(&target_id_mapping_file) { if let Some(record) = target_id_mapping.query(&series_info_id) { let new_record = record.copy_update_timestamp(); let _ = target_id_mapping.update(&series_info_id, new_record); @@ -566,97 +465,60 @@ pub async fn xtream_write_vod_info( content: &str, ) -> Result<(), Error> { let storage_path = try_option_ok!(xtream_get_storage_path(config, target_name)); - let (info_path, idx_path) = try_option_ok!(xtream_get_info_file_paths( - &storage_path, - XtreamCluster::Video - )); + let (info_path, idx_path) = try_option_ok!(xtream_get_info_file_paths(&storage_path, XtreamCluster::Video)); { let _file_lock = config.file_locks.write_lock(&info_path).await?; let mut writer = IndexedDocumentWriter::new_append(info_path, idx_path)?; - writer.write_doc(virtual_id, content).map_err(|_| { - Error::new( - ErrorKind::Other, - format!("failed to write xtream vod info for target {target_name}"), - ) - })?; - + writer.write_doc(virtual_id, content).map_err(|_| Error::new(ErrorKind::Other, format!("failed to write xtream vod info for target {target_name}")))?; writer.store()?; } Ok(()) } +async fn xtream_get_series_info_mapping( + config: &Config, + target_name: &str, + series_id: u32, +) -> Option { + xtream_get_info_mapping(config, target_name, series_id).await.filter(|id_record| !id_record.is_expired()) +} + +async fn xtream_get_info_mapping(config: &Config, target_name: &str, info_id: u32) -> Option { + let target_path = get_target_storage_path(config, target_name)?; + + let target_id_mapping_file = get_target_id_mapping_file(&target_path); + let _file_lock = config.file_locks.read_lock(&target_id_mapping_file).await.map_err(|err| { + error!("Could not lock id mapping for target {target_name}: {}", err); + Error::new(ErrorKind::Other, format!("ID mapping load error for target {target_name}")) + }).ok()?; + IndexedDocumentQuery::::try_new(&target_id_mapping_file).map_err(|err| { + error!("Could not load id mapping for target {target_name}: {}", err); + Error::new(ErrorKind::Other, format!("ID mapping load error for target {target_name}")) + }).ok().map(|mut tree| tree.query(&info_id))? +} + // Reads the series info entry if exists pub async fn xtream_load_series_info( config: &Config, target_name: &str, series_id: u32, ) -> Option { - let target_path = get_target_storage_path(config, target_name)?; + xtream_get_series_info_mapping(config, target_name, series_id).await?; + let storage_path = xtream_get_storage_path(config, target_name)?; - { - let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config - .file_locks - .read_lock(&target_id_mapping_file) - .await - .map_err(|err| { - error!( - "Could not lock id mapping for target {target_name}: {}", - err - ); - Error::new( - ErrorKind::Other, - format!("ID mapping load error for target {target_name}"), - ) - }) - .ok()?; - let mut target_id_mapping = - IndexedDocumentQuery::::try_new(&target_id_mapping_file) - .map_err(|err| { - error!( - "Could not load id mapping for target {target_name}: {}", - err - ); - Error::new( - ErrorKind::Other, - format!("ID mapping load error for target {target_name}"), - ) - }) - .ok()?; - - if let Some(id_record) = target_id_mapping.query(&series_id) { - if id_record.is_expired() { - return None; - } - } - } - let (info_path, idx_path) = xtream_get_info_file_paths(&storage_path, XtreamCluster::Series)?; if info_path.exists() && idx_path.exists() { { - let _file_lock = config - .file_locks - .read_lock(&info_path) - .await - .map_err(|err| { - error!("Could not lock document {:?}: {}", info_path, err); - Error::new( - ErrorKind::Other, - format!("Document Reader error for target {target_name}"), - ) - }) - .ok()?; - return match IndexedDocumentReader::::read_indexed_item( - &info_path, &idx_path, &series_id, - ) { + let _file_lock = config.file_locks.read_lock(&info_path).await.map_err(|err| { + error!("Could not lock document {:?}: {}", info_path, err); + Error::new(ErrorKind::Other, format!("Document Reader error for target {target_name}")) + }).ok()?; + return match IndexedDocumentReader::::read_indexed_item(&info_path, &idx_path, &series_id) { Ok(content) => Some(content), Err(err) => { - error!( - "Failed to read series info for id {series_id} for {target_name}: {}", - err - ); + error!("Failed to read series info for id {series_id} for {target_name}: {}", err); None } }; @@ -665,76 +527,35 @@ pub async fn xtream_load_series_info( None } -pub async fn xtream_get_vod_info_mapping( +async fn xtream_get_vod_info_mapping( config: &Config, target_name: &str, vod_id: u32, ) -> Option { - let target_path = get_target_storage_path(config, target_name)?; - let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config - .file_locks - .read_lock(&target_id_mapping_file) - .await - .map_err(|err| { - error!( - "Could not lock id mapping for target {target_name}: {}", - err - ); - Error::new( - ErrorKind::Other, - format!("ID mapping load error for target {target_name}"), - ) - }) - .ok()?; - let mut target_id_mapping = - IndexedDocumentQuery::::try_new(&target_id_mapping_file) - .map_err(|err| { - error!( - "Could not load id mapping for target {target_name}: {}", - err - ); - Error::new( - ErrorKind::Other, - format!("ID mapping load error for target {target_name}"), - ) - }) - .ok()?; - - // if let Some(id_record) = target_id_mapping.query(&vod_id) { - // if id_record.is_expired() { - // return None; - // } - // } - target_id_mapping.query(&vod_id) + xtream_get_info_mapping(config, target_name, vod_id).await + //.filter(|id_record| !id_record.is_expired()) } -// Reads the series info entry if exists +// Reads the vod info entry if exists pub async fn xtream_load_vod_info( config: &Config, target_name: &str, vod_id: u32, ) -> Option { - let storage_path = xtream_get_storage_path(config, target_name)?; + // Check if the entry exists; if not, we don't need to look further. xtream_get_vod_info_mapping(config, target_name, vod_id).await.as_ref()?; + // Entry exists, read db entry + let target_storage_path = xtream_get_storage_path(config, target_name)?; - let (info_path, idx_path) = xtream_get_info_file_paths(&storage_path, XtreamCluster::Video)?; + let (info_path, idx_path) = xtream_get_info_file_paths(&target_storage_path, XtreamCluster::Video)?; if info_path.exists() && idx_path.exists() { { - let _file_lock = config - .file_locks - .read_lock(&info_path) - .await - .map_err(|err| { - error!("Could not lock document {:?}: {}", info_path, err); - Error::new( - ErrorKind::Other, - format!("Document Reader error for target {target_name}"), - ) - }) - .ok()?; + let _file_lock = config.file_locks.read_lock(&info_path).await.map_err(|err| { + error!("Could not lock document {:?}: {}", info_path, err); + Error::new(ErrorKind::Other, format!("Document Reader error for target {target_name}")) + }).ok()?; return match IndexedDocumentReader::::read_indexed_item( &info_path, &idx_path, &vod_id, ) { @@ -755,12 +576,10 @@ pub async fn write_and_get_xtream_vod_info

( target: &ConfigTarget, pli: &P, content: &str, -) -> Result -where +) -> Result where P: PlaylistEntry, { - let mut doc = serde_json::from_str::>(content) - .map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; + let mut doc = serde_json::from_str::>(content).map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; // wen need to update the movie data with virtual ids. if let Some(Value::Object(movie_data)) = doc.get_mut(TAG_MOVIE_DATA) { @@ -784,11 +603,8 @@ where ); } } - let result = serde_json::to_string(&doc) - .map_err(|_| Error::new(ErrorKind::Other, "Failed to serialize vod info"))?; - xtream_write_vod_info(config, target.name.as_str(), pli.get_virtual_id(), &result) - .await - .ok(); + let result = serde_json::to_string(&doc).map_err(|_| Error::new(ErrorKind::Other, "Failed to serialize vod info"))?; + xtream_write_vod_info(config, target.name.as_str(), pli.get_virtual_id(), &result).await.ok(); Ok(result) } @@ -798,45 +614,25 @@ pub async fn write_and_get_xtream_series_info

( target: &ConfigTarget, pli_series_info: &P, content: &str, -) -> Result -where +) -> Result where P: PlaylistEntry, { - let mut doc = serde_json::from_str::(content) - .map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; + let mut doc = serde_json::from_str::(content).map_err(|_| Error::new(ErrorKind::Other, "Failed to parse JSON content"))?; - let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| { - Error::new( - ErrorKind::Other, - format!("Could not find path for target {}", target.name), - ) - })?; - - let episodes = doc - .get_mut("episodes") - .and_then(Value::as_object_mut) - .ok_or_else(|| Error::new(ErrorKind::Other, "No episodes found in content"))?; + let target_path = get_target_storage_path(config, target.name.as_str()).ok_or_else(|| Error::new(ErrorKind::Other, format!("Could not find path for target {}", target.name)))?; + let episodes = doc.get_mut("episodes").and_then(Value::as_object_mut).ok_or_else(|| Error::new(ErrorKind::Other, "No episodes found in content"))?; let virtual_id = pli_series_info.get_virtual_id(); { let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = config - .file_locks - .write_lock(&target_id_mapping_file) - .await - .map_err(|err| { - Error::new(ErrorKind::Other, format!("Could not load id mapping for target {} err:{err}", target.name )) - })?; + let _file_lock = config.file_locks.write_lock(&target_id_mapping_file).await.map_err(|err| Error::new(ErrorKind::Other, format!("Could not load id mapping for target {} err:{err}", target.name)))?; let mut target_id_mapping = TargetIdMapping::new(&target_id_mapping_file); let options = XtreamMappingOptions::from_target_options(target.options.as_ref()); let provider_url = pli_series_info.get_provider_url(); for episode_list in episodes.values_mut().filter_map(Value::as_array_mut) { for episode in episode_list.iter_mut().filter_map(Value::as_object_mut) { - if let Some(provider_id) = episode - .get("id") - .and_then(Value::as_str) - .and_then(|id| id.parse::().ok()) + if let Some(provider_id) = episode.get("id").and_then(Value::as_str).and_then(|id| id.parse::().ok()) { let uuid = hash_string(&format!("{provider_url}/{provider_id}")); let episode_virtual_id = target_id_mapping.insert_entry( @@ -858,22 +654,19 @@ where drop(target_id_mapping); } - let result = serde_json::to_string(&doc) - .map_err(|_| Error::new(ErrorKind::Other, "Failed to serialize updated series info"))?; - xtream_write_series_info(config, target.name.as_str(), virtual_id, &result) - .await - .ok(); + let result = serde_json::to_string(&doc).map_err(|_| Error::new(ErrorKind::Other, "Failed to serialize updated series info"))?; + xtream_write_series_info(config, target.name.as_str(), virtual_id, &result).await.ok(); Ok(result) } -pub async fn xtream_get_input_vod_info( +pub async fn xtream_get_input_info( cfg: &Config, input: &ConfigInput, provider_id: u32, + cluster: XtreamCluster, ) -> Option { - if 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::Video)) + if 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, cluster)) { if let Ok(_file_lock) = cfg.file_locks.read_lock(&info_path).await { if let Ok(content) = IndexedDocumentReader::::read_indexed_item( @@ -892,131 +685,130 @@ pub async fn xtream_update_input_info_file( cfg: &Config, input: &ConfigInput, wal_file: &mut File, + wal_path: &Path, cluster: XtreamCluster, ) -> Result<(), M3uFilterError> { - match get_input_storage_path(input, &cfg.working_dir) - .map(|storage_path| xtream_get_info_file_paths(&storage_path, cluster)) - { + match get_input_storage_path(input, &cfg.working_dir).map(|storage_path| xtream_get_info_file_paths(&storage_path, cluster)) { Ok(Some((info_path, idx_path))) => { match cfg.file_locks.write_lock(&info_path).await { Ok(_file_lock) => { - wal_file.seek(SeekFrom::Start(0)).map_err(|err| { - M3uFilterError::new( - M3uFilterErrorKind::Notify, - format!("Could not read {cluster} info {err}"), - ) - })?; + wal_file.seek(SeekFrom::Start(0)).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify,format!("Could not read {cluster} info {err}")))?; let mut reader = BufReader::new(wal_file); match IndexedDocumentWriter::::new_append(info_path, idx_path) { - Ok(mut writer) => { - let mut provider_id_bytes = [0u8; 4]; - let mut length_bytes = [0u8; 4]; - loop { - if reader.read_exact(&mut provider_id_bytes).is_err() { - break; // End of file - } - let provider_id = u32::from_le_bytes(provider_id_bytes); - reader.read_exact(&mut length_bytes).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not read temporary {cluster} info {err}")))?; - let length = u32::from_le_bytes(length_bytes) as usize; - let mut buffer = vec![0u8; length]; - reader.read_exact(&mut buffer).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not read temporary {cluster} info {err}")))?; - if let Ok(content) = String::from_utf8(buffer) { - let _ = writer.write_doc(provider_id, &content); - } + Ok(mut writer) => { + let mut provider_id_bytes = [0u8; 4]; + let mut length_bytes = [0u8; 4]; + loop { + if reader.read_exact(&mut provider_id_bytes).is_err() { + break; // End of file } - writer.store().map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not store {cluster} info {err}")))?; - if let Err(err) = fs::remove_file(wal_file) { - error!("Failed to delete WAL file for {cluster}"); + let provider_id = u32::from_le_bytes(provider_id_bytes); + reader.read_exact(&mut length_bytes).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not read temporary {cluster} info {err}")))?; + let length = u32::from_le_bytes(length_bytes) as usize; + let mut buffer = vec![0u8; length]; + reader.read_exact(&mut buffer).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not read temporary {cluster} info {err}")))?; + if let Ok(content) = String::from_utf8(buffer) { + let _ = writer.write_doc(provider_id, &content); } - Ok(()) } - Err(err) => Err(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create create indexed document writer for {cluster} info {err}"))), + writer.store().map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not store {cluster} info {err}")))?; + drop(reader); + if let Err(err) = fs::remove_file(wal_path) { + error!("Failed to delete WAL file for {cluster} {err}"); + } + Ok(()) } + Err(err) => Err(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create create indexed document writer for {cluster} info {err}"))), + } } - Err(err) => Err(M3uFilterError::new( - M3uFilterErrorKind::Info, - format!("{err}"), - )), + Err(err) => Err(M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}"))), } } - Ok(None) => Err(M3uFilterError::new( - M3uFilterErrorKind::Notify, - format!( - "Could not create storage path for input {}", - &input.name.as_ref().map_or("?", |v| v) - ), - )), - Err(err) => Err(M3uFilterError::new( - M3uFilterErrorKind::Notify, - format!("Could not create storage path for input {err}"), + Ok(None) => Err(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create storage path for input {}", &input.name.as_ref().map_or("?", |v| v)), )), + Err(err) => Err(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not create storage path for input {err}"))), } } -pub async fn xtream_update_input_record_from_wal_file( +pub async fn xtream_update_input_vod_record_from_wal_file( cfg: &Config, input: &ConfigInput, wal_file: &mut File, - cluster: XtreamCluster, + wal_path: &Path, ) -> Result<(), M3uFilterError> { - match get_input_storage_path(input, &cfg.working_dir) - .map(|storage_path| xtream_get_vod_record_file_path(&storage_path)) - { - Ok(record_path) => { - match cfg.file_locks.write_lock(&record_path).await { - Ok(_file_lock) => { - wal_file.seek(SeekFrom::Start(0)).map_err(|err| { - M3uFilterError::new( - M3uFilterErrorKind::Notify, - format!("Could not read vod wal info {err}"), - ) - })?; - let mut reader = BufReader::new(wal_file); - let mut provider_id_bytes = [0u8; 4]; - let mut tmdb_id_bytes = [0u8; 4]; - let mut ts_bytes = [0u8; 8]; - let mut tree_record_index: BPlusTree = - BPlusTree::load(&record_path).unwrap_or_else(|_| BPlusTree::new()); - loop { - if reader.read_exact(&mut provider_id_bytes).is_err() { - break; // End of file - } - let provider_id = u32::from_le_bytes(provider_id_bytes); - if reader.read_exact(&mut tmdb_id_bytes).is_err() { - break; // End of file - } - let tmdb_id = u32::from_le_bytes(tmdb_id_bytes); - if reader.read_exact(&mut ts_bytes).is_err() { - break; // End of file - } - let ts = u64::from_le_bytes(tmdb_id_bytes); - tree_record_index.insert( - provider_id, - InputInfoRecord { - provider_id, - tmdb_id, - ts, - }, - ); - } - tree_record_index.store(&record_path).map_err(|err| { - M3uFilterError::new( - M3uFilterErrorKind::Notify, - format!("Could not store vod record info {err}"), - ) - })?; - Ok(()) - } + let record_path = get_input_storage_path(input, &cfg.working_dir).map(|storage_path| xtream_get_record_file_path(&storage_path, XtreamCluster::Video)) + .map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Error accessing storage path: {err}"))) + .and_then(|opt| opt.ok_or_else(|| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Error accessing storage path for input: {}", input.name.clone().unwrap_or_else(|| input.id.to_string())))))?; - Err(err) => Err(M3uFilterError::new( - M3uFilterErrorKind::Info, - format!("{err}"), - )), + match cfg.file_locks.write_lock(&record_path).await { + Ok(_file_lock) => { + wal_file.seek(SeekFrom::Start(0)).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not read vod wal info {err}")))?; + let mut reader = BufReader::new(wal_file); + let mut provider_id_bytes = [0u8; 4]; + let mut tmdb_id_bytes = [0u8; 4]; + let mut ts_bytes = [0u8; 8]; + let mut tree_record_index: BPlusTree = BPlusTree::load(&record_path).unwrap_or_else(|_| BPlusTree::new()); + loop { + if reader.read_exact(&mut provider_id_bytes).is_err() { + break; // End of file + } + let provider_id = u32::from_le_bytes(provider_id_bytes); + if reader.read_exact(&mut tmdb_id_bytes).is_err() { + break; // End of file + } + let tmdb_id = u32::from_le_bytes(tmdb_id_bytes); + if reader.read_exact(&mut ts_bytes).is_err() { + break; // End of file + } + let ts = u64::from_le_bytes(ts_bytes); + tree_record_index.insert(provider_id, InputVodInfoRecord { tmdb_id, ts }); } + tree_record_index.store(&record_path).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not store vod record info {err}")))?; + drop(reader); + if let Err(err) = fs::remove_file(wal_path) { + error!("Failed to delete record WAL file for vod {err}"); + } + Ok(()) } - Err(err) => Err(M3uFilterError::new( - M3uFilterErrorKind::Notify, - format!("Could not create storage path for input {err}"), - )), + Err(err) => Err(M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}"))), + } +} + +pub async fn xtream_update_input_series_record_from_wal_file( + cfg: &Config, + input: &ConfigInput, + wal_file: &mut File, + wal_path: &Path, +) -> Result<(), M3uFilterError> { + let record_path = get_input_storage_path(input, &cfg.working_dir).map(|storage_path| xtream_get_record_file_path(&storage_path, XtreamCluster::Series)) + .map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Error accessing storage path: {err}"))) + .and_then(|opt| opt.ok_or_else(|| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Error accessing storage path for input: {}", input.name.clone().unwrap_or_else(|| input.id.to_string())))))?; + match cfg.file_locks.write_lock(&record_path).await { + Ok(_file_lock) => { + wal_file.seek(SeekFrom::Start(0)).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not read vod wal info {err}")))?; + let mut reader = BufReader::new(wal_file); + let mut provider_id_bytes = [0u8; 4]; + let mut ts_bytes = [0u8; 8]; + let mut tree_record_index: BPlusTree = BPlusTree::load(&record_path).unwrap_or_else(|_| BPlusTree::new()); + loop { + if reader.read_exact(&mut provider_id_bytes).is_err() { + break; // End of file + } + let provider_id = u32::from_le_bytes(provider_id_bytes); + if reader.read_exact(&mut ts_bytes).is_err() { + break; // End of file + } + let ts = u64::from_le_bytes(ts_bytes); + tree_record_index.insert(provider_id, ts); + } + tree_record_index.store(&record_path).map_err(|err| M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Could not store series record info {err}")))?; + drop(reader); + if let Err(err) = fs::remove_file(wal_path) { + error!("Failed to delete recrd WAL file for series {err}"); + } + Ok(()) + } + + Err(err) => Err(M3uFilterError::new(M3uFilterErrorKind::Info, format!("{err}"))), } } diff --git a/src/utils/download.rs b/src/utils/download.rs index 0e494b982..40d8a3d0e 100644 --- a/src/utils/download.rs +++ b/src/utils/download.rs @@ -4,7 +4,7 @@ use crate::model::playlist::{FetchedPlaylist, PlaylistEntry, PlaylistGroup, Play use crate::model::xmltv::TVGuide; use crate::processing::xtream_parser::parse_xtream_series_info; use crate::processing::{m3u_parser, xtream_parser}; -use crate::repository::xtream_repository::xtream_get_input_vod_info; +use crate::repository::xtream_repository::{xtream_get_input_info}; use crate::repository::xtream_repository; use crate::utils::{file_utils, request_utils}; use log::{debug, info}; @@ -122,16 +122,29 @@ where { if cluster == XtreamCluster::Series { if let Some(content) = xtream_repository::xtream_load_series_info(config, target.name.as_str(), pli.get_virtual_id()).await { + // Deliver existing target content return Ok(content); } + + // Check if the content has been resolved + let resolve_series = target.options.as_ref().is_some_and(|opt| opt.xtream_resolve_series); + if resolve_series { + if let Some(provider_id) = pli.get_provider_id() { + if let Some(content) = xtream_get_input_info(config, input, provider_id, XtreamCluster::Series).await { + return xtream_repository::write_and_get_xtream_series_info(config, target, pli, &content).await; + } + } + } } else if cluster == XtreamCluster::Video { if let Some(content) = xtream_repository::xtream_load_vod_info(config, target.name.as_str(), pli.get_virtual_id()).await { + // Deliver existing target content return Ok(content); } - let resolve_movies = target.options.as_ref().is_some_and(|opt| opt.xtream_resolve_video); - if resolve_movies { + // Check if the content has been resolved + let resolve_vod = target.options.as_ref().is_some_and(|opt| opt.xtream_resolve_video); + if resolve_vod { if let Some(provider_id) = pli.get_provider_id() { - if let Some(content) = xtream_get_input_vod_info(config, input, provider_id).await { + if let Some(content) = xtream_get_input_info(config, input, provider_id, XtreamCluster::Video).await { return xtream_repository::write_and_get_xtream_vod_info(config, target, pli, &content).await; } } @@ -164,7 +177,7 @@ fn get_skip_cluster(input: &ConfigInput) -> Vec { } } if skip_cluster.len() == 3 { - let name = input.name.as_ref().map_or_else(|| input.id.to_string(), std::string::ToString::to_string); + let name = input.name.as_ref().map_or_else(|| input.id.to_string(), ToString::to_string); info!("You have skipped all sections from xtream input {name}"); } skip_cluster