resolve series wip

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 [<get_resolve_ $cluster _options>](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<V, F>(cfg: &Config, errors: &mut Vec<M3uFilterError>, fpl: &FetchedPlaylist<'_>,
+                                                                 cluster: XtreamCluster, extract_ts: F) -> HashMap<u32, u64>
+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::<u32, V>::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
+}