diff --git a/Cargo.lock b/Cargo.lock index ad274c2d7..76e33db17 100644 --- a/Cargo.lock +++ b/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", diff --git a/README.md b/README.md index 315f2788b..baed2387d 100644 --- a/README.md +++ b/README.md @@ -158,6 +158,7 @@ messaging: token: user: 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. diff --git a/backend/Cargo.toml b/backend/Cargo.toml index 59010ad01..c1cfd4f71 100644 --- a/backend/Cargo.toml +++ b/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" diff --git a/backend/src/api/config_file.rs b/backend/src/api/config_file.rs index 173a30af8..6c25730eb 100644 --- a/backend/src/api/config_file.rs +++ b/backend/src/api/config_file.rs @@ -23,9 +23,11 @@ impl ConfigFile { let paths = > as Access>::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}"); diff --git a/backend/src/main.rs b/backend/src/main.rs index 2910634d3..87182603f 100644 --- a/backend/src/main.rs +++ b/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, } diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 9592e5eb6..4fe2768c3 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/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, event_manager: Option>, playlist_state: Option<&Arc>, ) -> Result<(), Vec> { - let pipe = get_processing_pipe(target); debug_if_enabled!("Processing order is {}", &target.processing_order); let mut duplicates: HashSet = HashSet::new(); @@ -623,6 +622,7 @@ async fn process_playlist_for_target(app_config: &Arc, 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); diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 2a229c7c0..3114f2a4b 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/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, -// storage_path: &Path, -// cluster: XtreamCluster, -// playlist: Vec, -// ) -> 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, @@ -185,10 +168,12 @@ fn get_map_item_as_str(map: &serde_json::Map, key: &str) -> Optio None } -async fn load_old_category_ids(path: &Path) -> (u32, HashMap) { +type CategoryKey = (XtreamCluster, Cow<'static, str>); + +async fn load_old_category_ids(path: &Path) -> (u32, HashMap) { let old_path = path.to_path_buf(); tokio::task::spawn_blocking(move || { - let mut result: HashMap = HashMap::new(); + let mut result: HashMap = 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) { { 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) { 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) { } } } - 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 = 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::>() +} + 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, 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::>(&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::>(&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::::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()) } diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index 2ea49897f..afff6c676 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/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 = 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}"), } diff --git a/backend/src/utils/file/mapping_reader.rs b/backend/src/utils/file/mapping_reader.rs index fb55259d1..d3eccbf54 100644 --- a/backend/src/utils/file/mapping_reader.rs +++ b/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, TuliproxError> { +fn read_mappings_from_file(mappings_file: &Path, resolve_env: bool) -> Result, 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) -> Result Result, TuliproxError> { +fn read_mappings_from_directory(path: &Path, resolve_env: bool) -> Result, 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