Merge pull request #496 from euzu/feature/fix_mapping_missing_groups

Fix missing categories

Cargo.lock

@@ -1156,7 +1156,7 @@ dependencies = [
 
 [[package]]
 name = "frontend"
-version = "3.2.34"
+version = "3.2.35"
 dependencies = [
  "anyhow",
  "base64",
@@ -3992,7 +3992,7 @@ dependencies = [
 
 [[package]]
 name = "shared"
-version = "3.2.34"
+version = "3.2.35"
 dependencies = [
  "base64",
  "bitflags 2.10.0",
@@ -4566,7 +4566,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b"
 
 [[package]]
 name = "tuliprox"
-version = "3.2.34"
+version = "3.2.35"
 dependencies = [
  "arc-swap",
  "async-compression",

README.md

@@ -158,6 +158,7 @@ messaging:
     token: <api_token>
     user: <api_username>
     url: `optional`, default is `https://api.pushover.net/1/messages.json`
+```
 
 ### 1.4.1 Messaging Templating
 For `discord` and `rest` messaging, you can use [Handlebars](https://handlebarsjs.com/) templates to format the message body.

backend/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "tuliprox"
-version = "3.2.34"
+version = "3.2.35"
 edition = "2021"
 rust-version = "1.87.0"
 

backend/src/api/config_file.rs

@@ -23,9 +23,11 @@ impl ConfigFile {
         let paths = <Arc<ArcSwap<ConfigPaths>> as Access<ConfigPaths>>::load(&app_state.app_config.paths);
         if let Some(mapping_file_path) = paths.mapping_file_path.as_ref() {
             match utils::read_mappings(mapping_file_path, true) {
-                Ok(Some(mappings_cfg)) => {
+                Ok(Some((mapping_files, mappings_cfg))) => {
                     app_state.app_config.set_mappings(mapping_file_path, &mappings_cfg);
-                    info!("Loaded mapping file {mapping_file_path}");
+                    for mapping_file in mapping_files {
+                        info!("Loaded mapping file {}", mapping_file.display());
+                    }
                 }
                 Ok(None) => {
                     info!("No mapping file loaded {mapping_file_path}");

backend/src/main.rs

@@ -10,14 +10,14 @@ mod modules;
 include_modules!();
 
 use crate::auth::generate_password;
+use crate::library::LibraryProcessor;
 use crate::model::{
     AppConfig, Config, Healthcheck, HealthcheckConfig, ProcessTargets, SourcesConfig,
 };
 use crate::processing::processor::playlist;
-use crate::utils::{db_viewer, init_logger};
 use crate::utils::request::create_client;
 use crate::utils::{config_file_reader, resolve_env_var};
-use crate::library::LibraryProcessor;
+use crate::utils::{db_viewer, init_logger};
 use arc_swap::access::Access;
 use arc_swap::ArcSwap;
 use chrono::{DateTime, Utc};
@@ -178,13 +178,13 @@ fn print_info(app_config: &AppConfig) {
     info!("Config file: {:?}", &paths.config_file_path);
     info!("Source file: {:?}", &paths.sources_file_path);
     info!("Api Proxy File: {:?}", &paths.api_proxy_file_path);
-    info!(
-        "Mapping file: {:?}",
-        &paths
-            .mapping_file_path
-            .as_ref()
-            .map_or_else(|| "not used", |v| v.as_str())
-    );
+    info!("Mapping path: {:?}", &paths.mapping_file_path.as_ref().map_or_else(|| "not used", |v| v.as_str()));
+
+    if let Some(mapping_paths) = paths.mapping_files_used.as_ref() {
+        for mapping_path in mapping_paths {
+            info!("Mapping file loaded: {mapping_path}");
+        }
+    }
 
     if let Some(cache) = config.reverse_proxy.as_ref().and_then(|r| r.cache.as_ref()) {
         if cache.enabled {
@@ -208,6 +208,7 @@ fn get_file_paths(args: &Args) -> ConfigPaths {
         config_file_path: config_file,
         sources_file_path: sources_file,
         mapping_file_path: mappings_file, // need to be set after config read
+        mapping_files_used: None,
         api_proxy_file_path: api_proxy_file,
         custom_stream_response_path: None,
     }

backend/src/processing/processor/playlist.rs

@@ -567,7 +567,7 @@ fn execute_pipe<'a>(target: &ConfigTarget, pipe: &ProcessingPipe, fpl: &FetchedP
     };
     if target.options.as_ref().is_some_and(|opt| opt.remove_duplicates) {
         for group in &mut new_fpl.playlist_groups {
-            // `HashSet::insert`  returns true for first insert, otherweise false
+            // `HashSet::insert`  returns true for first insert, otherwise false
             group.channels.retain(|item| duplicates.insert(duplicate_hash(item)));
         }
     }
@@ -614,7 +614,6 @@ async fn process_playlist_for_target(app_config: &Arc<AppConfig>,
                                      event_manager: Option<Arc<EventManager>>,
                                      playlist_state: Option<&Arc<PlaylistStorageState>>,
 ) -> Result<(), Vec<TuliproxError>> {
-    let pipe = get_processing_pipe(target);
     debug_if_enabled!("Processing order is {}", &target.processing_order);
 
     let mut duplicates: HashSet<UUIDType> = HashSet::new();
@@ -623,6 +622,7 @@ async fn process_playlist_for_target(app_config: &Arc<AppConfig>,
     debug!("Executing processing pipes");
     let broadcast_step = create_broadcast_callback(event_manager.as_ref());
 
+    let pipe = get_processing_pipe(target);
     let mut step = StepMeasure::new(&target.name, broadcast_step);
     for provider_fpl in playlists.iter_mut() {
         step.broadcast("Executing transformations on '{}' playlist", &target.name);

backend/src/repository/xtream_repository.rs

@@ -21,6 +21,7 @@ use shared::error::{create_tuliprox_error, create_tuliprox_error_result, info_er
 use shared::model::xtream_const::XTREAM_CLUSTER;
 use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemType, SeriesStreamProperties, StreamProperties, VideoStreamProperties, XtreamCluster, XtreamPlaylistItem};
 use shared::utils::get_u32_from_serde_value;
+use std::borrow::Cow;
 use std::collections::HashMap;
 use std::fs::File;
 use std::io::{Error, ErrorKind};
@@ -107,24 +108,6 @@ async fn write_playlists_to_file(
     }
     Ok(())
 }
-//
-// async fn write_playlists_to_file_2(
-//     app_config: &Arc<AppConfig>,
-//     storage_path: &Path,
-//     cluster: XtreamCluster,
-//     playlist: Vec<XtreamPlaylistItem>,
-// ) -> Result<(), TuliproxError> {
-//     if !playlist.is_empty() {
-//         let xtream_path = xtream_get_file_path(storage_path, cluster);
-//         let _file_lock = app_config.file_locks.write_lock(&xtream_path).await;
-//         let mut tree = BPlusTree::new();
-//         for item in playlist {
-//             tree.insert(item.virtual_id, item);
-//         }
-//         tree.store(&xtream_path).map_err(|err| cant_write_result!(&xtream_path, err))?;
-//     }
-//     Ok(())
-// }
 
 pub async fn write_playlist_item_update(
     app_config: &Arc<AppConfig>,
@@ -185,10 +168,12 @@ fn get_map_item_as_str(map: &serde_json::Map<String, Value>, key: &str) -> Optio
     None
 }
 
-async fn load_old_category_ids(path: &Path) -> (u32, HashMap<String, u32>) {
+type CategoryKey = (XtreamCluster, Cow<'static, str>);
+
+async fn load_old_category_ids(path: &Path) -> (u32, HashMap<CategoryKey, u32>) {
     let old_path = path.to_path_buf();
     tokio::task::spawn_blocking(move || {
-        let mut result: HashMap<String, u32> = HashMap::new();
+        let mut result: HashMap<CategoryKey, u32> = HashMap::new();
         let mut max_id: u32 = 0;
         for (cluster, cat) in [
             (XtreamCluster::Live, storage_const::COL_CAT_LIVE),
@@ -197,7 +182,7 @@ async fn load_old_category_ids(path: &Path) -> (u32, HashMap<String, u32>) {
         {
             let col_path = get_collection_path(&old_path, cat);
             if col_path.exists() {
-                if let Ok(file) = File::open(col_path) {
+                if let Ok(file) = File::open(&col_path) {
                     let reader = file_reader(file);
                     match serde_json::from_reader(reader) {
                         Ok(value) => {
@@ -206,7 +191,7 @@ async fn load_old_category_ids(path: &Path) -> (u32, HashMap<String, u32>) {
                                     if let Some(category_id) = entry.get(crate::model::XC_TAG_CATEGORY_ID).and_then(get_u32_from_serde_value) {
                                         if let Value::Object(item) = entry {
                                             if let Some(category_name) = get_map_item_as_str(&item, crate::model::XC_TAG_CATEGORY_NAME) {
-                                                result.insert(format!("{cluster}{category_name}"), category_id);
+                                                result.insert((cluster, Cow::Owned(category_name)), category_id);
                                                 max_id = max_id.max(category_id);
                                             }
                                         }
@@ -214,7 +199,9 @@ async fn load_old_category_ids(path: &Path) -> (u32, HashMap<String, u32>) {
                                 }
                             }
                         }
-                        Err(_err) => {}
+                        Err(err) => {
+                            log::warn!("Failed to parse category file {}: {err}", col_path.display());
+                        }
                     }
                 }
             }
@@ -263,38 +250,29 @@ pub async fn xtream_write_playlist(
     let mut series_col = Vec::with_capacity(50_000);
     let mut vod_col = Vec::with_capacity(50_000);
 
-    // preserve category_ids
-    let (max_cat_id, existing_cat_ids) = load_old_category_ids(&path).await;
-    let mut cat_id_counter = max_cat_id;
-    for plg in playlist.iter_mut() {
-        if !plg.channels.is_empty() {
-            let cat_key = format!("{}{}", plg.xtream_cluster, &plg.title);
-            let cat_id = existing_cat_ids.get(&cat_key).unwrap_or_else(|| {
-                cat_id_counter += 1;
-                &cat_id_counter
-            });
-            plg.id = *cat_id;
-
-            match &plg.xtream_cluster {
+    let categories = create_categories(playlist, &path).await;
+    {
+        for (xtream_cluster, category) in categories {
+            match xtream_cluster {
                 XtreamCluster::Live => &mut cat_live_col,
                 XtreamCluster::Series => &mut cat_series_col,
                 XtreamCluster::Video => &mut cat_vod_col,
-            }.push(json!(CategoryEntry {
-                category_id: *cat_id,
-                category_name: plg.title.clone(),
-                parent_id: 0
-            }));
+            }.push(category);
+        }
+    }
 
-            for pli in &mut plg.channels {
-                let header = &mut pli.header;
-                header.category_id = *cat_id;
-                let col = match header.xtream_cluster {
-                    XtreamCluster::Live => &mut live_col,
-                    XtreamCluster::Series => &mut series_col,
-                    XtreamCluster::Video => &mut vod_col,
-                };
-                col.push(pli);
-            }
+    for plg in playlist.iter_mut() {
+        if plg.channels.is_empty() {
+            continue;
+        }
+
+        for pli in &plg.channels {
+            let col = match pli.header.xtream_cluster {
+                XtreamCluster::Live => &mut live_col,
+                XtreamCluster::Series => &mut series_col,
+                XtreamCluster::Video => &mut vod_col,
+            };
+            col.push(pli);
         }
     }
 
@@ -344,6 +322,60 @@ pub async fn xtream_write_playlist(
     Ok(())
 }
 
+async fn create_categories(playlist: &mut [PlaylistGroup], path: &Path) -> Vec<(XtreamCluster, CategoryEntry)> {
+    // preserve category_ids
+    let (max_cat_id, existing_cat_ids) = load_old_category_ids(path).await;
+    let mut cat_id_counter = max_cat_id;
+
+    let mut new_categories: IndexMap<(XtreamCluster, Cow<'_, str>), CategoryEntry> = IndexMap::new();
+
+    let mut last_cluster: Option<XtreamCluster> = None;
+    let mut last_group = String::new();
+    let mut last_category_id: u32 = 0;
+
+    for plg in playlist.iter_mut() {
+        if plg.channels.is_empty() {
+            continue;
+        }
+
+        for channel in &mut plg.channels {
+            let cluster = channel.header.xtream_cluster;
+            let group = channel.header.group.as_str();
+
+            // Fast path
+            if last_cluster == Some(cluster) && last_group == group {
+                channel.header.category_id = last_category_id;
+                continue;
+            }
+
+            let key = (cluster, Cow::Borrowed(group));
+
+            let entry = new_categories.entry(key.clone()).or_insert_with(|| {
+                let cat_id = existing_cat_ids.get(&key).copied().unwrap_or_else(|| {
+                    cat_id_counter += 1;
+                    cat_id_counter
+                });
+
+                CategoryEntry {
+                    category_id: cat_id,
+                    category_name: group.to_string(),
+                    parent_id: 0,
+                }
+            });
+
+            last_cluster = Some(cluster);
+            last_group.clear();
+            last_group.push_str(group);
+            last_category_id = entry.category_id;
+
+            channel.header.category_id = last_category_id;
+        }
+    }
+    new_categories.into_iter()
+        .map(|((cluster, _group), value)| (cluster, value))
+        .collect::<Vec<(XtreamCluster, CategoryEntry)>>()
+}
+
 pub fn xtream_get_collection_path(
     cfg: &Config,
     target_name: &str,
@@ -810,40 +842,40 @@ pub async fn load_input_xtream_playlist(app_config: &Arc<AppConfig>, storage_pat
                 XtreamCluster::Series => storage_const::COL_CAT_SERIES,
             };
             let cat_path = get_collection_path(storage_path, cat_col_name);
-            
+
             if cat_path.exists() {
-               if let Ok(content) = tokio::fs::read_to_string(&cat_path).await {
-                   if let Ok(cats) = serde_json::from_str::<Vec<CategoryEntry>>(&content) {
-                       for cat in cats {
-                           groups.insert(cat.category_id, PlaylistGroup {
-                               id: cat.category_id,
-                               title: cat.category_name,
-                               channels: Vec::new(),
-                               xtream_cluster: cluster,
-                           });
-                       }
-                   }
-               }
+                if let Ok(content) = tokio::fs::read_to_string(&cat_path).await {
+                    if let Ok(cats) = serde_json::from_str::<Vec<CategoryEntry>>(&content) {
+                        for cat in cats {
+                            groups.insert(cat.category_id, PlaylistGroup {
+                                id: cat.category_id,
+                                title: cat.category_name,
+                                channels: Vec::new(),
+                                xtream_cluster: cluster,
+                            });
+                        }
+                    }
+                }
             }
 
             // Load Items
             let _file_lock = app_config.file_locks.read_lock(&xtream_path).await;
             if let Ok(mut query) = BPlusTreeQuery::<u32, XtreamPlaylistItem>::try_new(&xtream_path) {
                 for (_, ref item) in query.iter() {
-                     let cat_id = item.category_id;
-                     groups.entry(cat_id)
+                    let cat_id = item.category_id;
+                    groups.entry(cat_id)
                         .or_insert_with(|| PlaylistGroup {
-                             id: cat_id,
-                             title: String::from("Unknown"),
-                             channels: Vec::new(),
-                             xtream_cluster: cluster,
+                            id: cat_id,
+                            title: String::from("Unknown"),
+                            channels: Vec::new(),
+                            xtream_cluster: cluster,
                         })
                         .channels.push(PlaylistItem::from(item));
                 }
             }
         }
     }
-    
+
     Ok(groups.into_values().collect())
 }
 

backend/src/utils/file/config_reader.rs

@@ -117,7 +117,7 @@ pub fn read_app_config_dto(paths: &ConfigPaths,
     let config = read_config_file(config_file, resolve_env, include_computed)?;
     let sources = read_sources_file(sources_file, resolve_env, include_computed, config.get_hdhr_device_overview().as_ref())?;
     let mappings = if let Some(mappings_file) = paths.mapping_file_path.as_ref() {
-        read_mappings_file(mappings_file, resolve_env).unwrap_or(None)
+        read_mappings_file(mappings_file, resolve_env).unwrap_or(None).map(|(_, mappings)| mappings)
     } else {
         None
     };
@@ -244,7 +244,18 @@ pub async fn read_initial_app_config(paths: &mut ConfigPaths,
 
     if let Some(mappings_file) = &paths.mapping_file_path {
         match utils::read_mappings(mappings_file.as_str(), resolve_env) {
-            Ok(Some(mappings)) => app_config.set_mappings(mappings_file, &mappings),
+            Ok(Some((mapping_paths, mappings))) => {
+                app_config.set_mappings(mappings_file, &mappings);
+                paths.mapping_files_used = {
+                    let vec: Vec<String> = mapping_paths
+                        .iter()
+                        .map(|p| p.display().to_string())
+                        .collect();
+
+                    if vec.is_empty() { None } else { Some(vec) }
+                };
+                app_config.paths.store(Arc::new(paths.clone()));
+            },
             Ok(None) => info!("Mapping file: not used"),
             Err(err) => exit!("{err}"),
         }

backend/src/utils/file/mapping_reader.rs

@@ -26,12 +26,12 @@ fn read_mapping(mapping_file: &Path, resolve_var: bool, prepare_mappings: bool)
     Ok(None)
 }
 
-fn read_mappings_from_file(mappings_file: &Path, resolve_env: bool) -> Result<Option<MappingsDto>, TuliproxError> {
+fn read_mappings_from_file(mappings_file: &Path, resolve_env: bool) -> Result<Option<(Vec<PathBuf>, MappingsDto)>, TuliproxError> {
     match read_mapping(mappings_file, resolve_env, true) {
         Ok(mappings) => {
             match mappings {
                 None => Ok(None),
-                Some(mappings_cfg) => Ok(Some(mappings_cfg))
+                Some(mappings_cfg) => Ok(Some((vec![mappings_file.to_path_buf()], mappings_cfg))),
             }
         }
         Err(err) => Err(err),
@@ -81,7 +81,7 @@ fn merge_mapping_definitions(mappings: Vec<MappingsDto>) -> Result<Option<Mappin
     Ok(Some(result))
 }
 
-fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result<Option<MappingsDto>, TuliproxError> {
+fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result<Option<(Vec<PathBuf>, MappingsDto)>, TuliproxError> {
     let mut files = vec![];
     let mut visit = |entry: &std::fs::DirEntry, metadata: &std::fs::Metadata| {
         if metadata.is_file() {
@@ -96,9 +96,13 @@ fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result<Option
     files.sort_by(|a,b| a.file_name().cmp(&b.file_name()));
 
     let mut mappings = vec![];
+    let mut loaded_mapping_files = vec![];
     for file_path in files {
         match read_mapping(&file_path, resolve_env, false) {
-            Ok(Some(mapping)) => mappings.push(mapping),
+            Ok(Some(mapping)) => {
+                loaded_mapping_files.push(file_path);
+                mappings.push(mapping);
+            },
             Ok(None) => {}
             Err(err) => return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Failed to read mapping file {file_path:?}: {err:?}"),
         }
@@ -107,10 +111,14 @@ fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result<Option
     if mappings.is_empty() {
         return Ok(None);
     }
-    merge_mapping_definitions(mappings)
+    match merge_mapping_definitions(mappings) {
+        Ok(Some(merged_mappings)) => Ok(Some((loaded_mapping_files, merged_mappings))),
+        Ok(None) => Ok(None),
+        Err(err) => Err(err),
+    }
 }
 
-pub fn read_mappings_file(mappings_file: &str, resolve_env: bool) -> Result<Option<MappingsDto>, TuliproxError> {
+pub fn read_mappings_file(mappings_file: &str, resolve_env: bool) -> Result<Option<(Vec<PathBuf>, MappingsDto)>, TuliproxError> {
     let path = PathBuf::from(mappings_file);
     match std::fs::metadata(&path) {
         Ok(metadata) => {
@@ -128,9 +136,9 @@ pub fn read_mappings_file(mappings_file: &str, resolve_env: bool) -> Result<Opti
     }
 }
 
-pub fn read_mappings(mappings_file: &str, resolve_env: bool) -> Result<Option<Mappings>, TuliproxError> {
+pub fn read_mappings(mappings_file: &str, resolve_env: bool) -> Result<Option<(Vec<PathBuf>, Mappings)>, TuliproxError> {
     match read_mappings_file(mappings_file, resolve_env)? {
-        Some(dto) => Ok(Some(Mappings::from(&dto))),
+        Some((paths, dto)) => Ok(Some((paths, Mappings::from(&dto)))),
         None => Ok(None),
     }
 }

frontend/Cargo.toml

@@ -1,11 +1,11 @@
 [package]
 name = "frontend"
-version = "3.2.34"
+version = "3.2.35"
 edition = "2021"
 rust-version = "1.87.0"
 
 [dependencies]
-shared = { version = "3.2.34", path = "../shared" }
+shared = { version = "3.2.35", path = "../shared" }
 chrono = "0"
 yew = "0.21"
 yew-router = "0.18"

shared/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "shared"
-version = "3.2.34"
+version = "3.2.35"
 edition = "2021"
 rust-version = "1.87.0"
 

shared/src/model/config/paths.rs

@@ -1,4 +1,4 @@
-use crate::utils::is_blank_optional_string;
+use crate::utils::{is_blank_optional_string, is_empty_optional_vec};
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct ConfigPaths {
@@ -7,7 +7,10 @@ pub struct ConfigPaths {
     pub sources_file_path: String,
     #[serde(default, skip_serializing_if = "is_blank_optional_string")]
     pub mapping_file_path: Option<String>,
+    #[serde(default, skip_serializing_if = "is_empty_optional_vec")]
+    pub mapping_files_used: Option<Vec<String>>,
     pub api_proxy_file_path: String,
     #[serde(default, skip_serializing_if = "is_blank_optional_string")]
     pub custom_stream_response_path: Option<String>,
+
 }

shared/src/utils/default_utils.rs

@@ -9,6 +9,10 @@ pub fn is_blank_optional_string(s: &Option<String>) -> bool {
     s.as_ref().is_none_or(|s| s.chars().all(|c| c.is_whitespace()))
 }
 
+pub fn is_empty_optional_vec<T>(s: &Option<Vec<T>>) -> bool {
+    s.as_ref().is_none_or(|v| v.is_empty())
+}
+
 pub fn default_as_default() -> String { "default".into() }
 // Default delay values for resolving VOD or Series requests,
 // used to prevent frequent requests that could trigger a provider ban.