diff --git a/CHANGELOG.md b/CHANGELOG.md index d4bfc4767..e29dde97e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # Changelog # 3.3.0 (2025-11-xx) -- !BREAKING CHANGE! Due to some structure changes the old data format is invalid. You need to clean your data folder and update the playlists +- !BREAKING CHANGE! Due to some heavy refactoring, the old data format is invalid. You need to clean your data folder and update the playlists - !BREAKING CHANGE! config.yml threads attribute is now renamed to process_parallel and is a boolean (true or false). - !BREAKING CHANGE! config.yml adds a reverse proxy config field rewrite_secret to keep resource URLs valid after restart. - !BREAKING CHANGE! removed `forced_retry_interval_secs`. diff --git a/backend/src/api/endpoints/api_playlist_utils.rs b/backend/src/api/endpoints/api_playlist_utils.rs index 5185a69c0..05a2624e7 100644 --- a/backend/src/api/endpoints/api_playlist_utils.rs +++ b/backend/src/api/endpoints/api_playlist_utils.rs @@ -85,7 +85,7 @@ fn group_playlist_groups_by_cluster(playlist: Vec) -> (Vec, target: &Arc, virtual_id: u32) -> Option { if target.has_output(TargetType::Xtream) { - if let Ok((pli, _)) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, app_state, target, None).await { + if let Ok(pli) = xtream_repository::xtream_get_item_for_stream_id(virtual_id, app_state, target, None).await { return Some(pli.to_stream_channel(target.id)); } } diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index a5c51a8de..04e093709 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -32,8 +32,8 @@ use futures::Stream; use log::{debug, error, warn}; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; -use shared::error::{create_tuliprox_error_result}; use shared::error::info_err; +use shared::error::create_tuliprox_error_result; use shared::error::{str_to_io_error, TuliproxError, TuliproxErrorKind}; use shared::model::{create_stream_channel_with_type, PlaylistEntry, PlaylistItemType, ProxyType, TargetType, UserConnectionPermission, XtreamCluster}; @@ -246,7 +246,7 @@ async fn xtream_player_api_stream( let (action_stream_id, stream_ext) = separate_number_and_remainder(stream_req.stream_id); let req_virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); - let (pli, mapping) = try_result_not_found!( + let pli = try_result_not_found!( xtream_repository::xtream_get_item_for_stream_id(req_virtual_id, app_state, &target, None).await, true, format!("Failed to read xtream item for stream id {req_virtual_id}") @@ -346,7 +346,7 @@ async fn xtream_player_api_stream( let redirect_params = RedirectParams { item: &pli, - provider_id: Some(mapping.provider_id), + provider_id: pli.get_provider_id(), cluster, target_type: TargetType::Xtream, target: &target, @@ -435,7 +435,7 @@ async fn xtream_player_api_stream_with_token( } let (action_stream_id, stream_ext) = separate_number_and_remainder(stream_req.stream_id); let req_virtual_id: u32 = try_result_bad_request!(action_stream_id.trim().parse()); - let (pli, _mapping) = try_result_bad_request!( + let pli = try_result_bad_request!( xtream_repository::xtream_get_item_for_stream_id( req_virtual_id, app_state, @@ -593,7 +593,7 @@ async fn xtream_player_api_resource( } let req_virtual_id: u32 = try_result_bad_request!(resource_req.stream_id.trim().parse()); let resource = resource_req.action_path.trim(); - let (pli, _) = try_result_bad_request!( + let pli = try_result_bad_request!( xtream_repository::xtream_get_item_for_stream_id( req_virtual_id, app_state, @@ -824,7 +824,7 @@ async fn xtream_get_stream_info_response( Err(_) => return axum::http::StatusCode::BAD_REQUEST.into_response(), }; - if let Ok((pli, virtual_record)) = xtream_repository::xtream_get_item_for_stream_id( + if let Ok(pli) = xtream_repository::xtream_get_item_for_stream_id( virtual_id, app_state, target, @@ -843,9 +843,7 @@ async fn xtream_get_stream_info_response( if pli.provider_id > 0 { let input_name = &pli.input_name; if let Some(input) = app_state.app_config.get_input_by_name(input_name.as_str()) { - if let Some(info_url) = - xtream::get_xtream_player_api_info_url(&input, cluster, pli.provider_id) - { + if let Some(info_url) = xtream::get_xtream_player_api_info_url(&input, cluster, pli.provider_id) { // Redirect is only possible for live streams, vod and series info needs to be modified if user.proxy == ProxyType::Redirect && cluster == XtreamCluster::Live { return redirect(&info_url).into_response(); @@ -858,10 +856,8 @@ async fn xtream_get_stream_info_response( &pli, info_url.as_str(), cluster, - ) - .await + ).await { - return try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header( @@ -877,7 +873,7 @@ async fn xtream_get_stream_info_response( return match cluster { XtreamCluster::Video => { let content = - create_vod_info_from_item(target, user, &pli, virtual_record.last_updated); + create_vod_info_from_item(target, user, &pli); try_unwrap_body!(axum::response::Response::builder() .status(axum::http::StatusCode::OK) .header( @@ -914,7 +910,7 @@ async fn xtream_get_short_epg( Err(_) => return axum::http::StatusCode::BAD_REQUEST.into_response(), }; - if let Ok((pli, _)) = xtream_repository::xtream_get_item_for_stream_id( + if let Ok(pli) = xtream_repository::xtream_get_item_for_stream_id( virtual_id, app_state, target, @@ -1085,7 +1081,7 @@ async fn xtream_get_catchup_response( end: &str, ) -> impl IntoResponse + Send { let req_virtual_id: u32 = try_result_bad_request!(FromStr::from_str(stream_id)); - let (pli, _) = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id( + let pli = try_result_bad_request!(xtream_repository::xtream_get_item_for_stream_id( req_virtual_id, app_state, target, diff --git a/backend/src/api/main_api.rs b/backend/src/api/main_api.rs index a8f4cf6e5..522abf46a 100644 --- a/backend/src/api/main_api.rs +++ b/backend/src/api/main_api.rs @@ -81,7 +81,7 @@ async fn create_shared_data( Arc::new(ArcSwapOption::from(Some(Arc::new(db)))) } Err(err) => { - error!("Failed to load GeoIp db: {err}"); + info!("No GeoIp db found: {err}"); Arc::new(ArcSwapOption::from(None)) } } diff --git a/backend/src/api/model/playlist_mem_cache.rs b/backend/src/api/model/playlist_mem_cache.rs index 897bb2157..eff694b8f 100644 --- a/backend/src/api/model/playlist_mem_cache.rs +++ b/backend/src/api/model/playlist_mem_cache.rs @@ -75,7 +75,7 @@ impl PlaylistStorageState { XtreamCluster::Live => &mut xtream.live, XtreamCluster::Video => &mut xtream.vod, XtreamCluster::Series => &mut xtream.series, - }.insert(pli.header.virtual_id, pli.to_xtream()); + }.insert(pli.header.virtual_id, XtreamPlaylistItem::from(&pli)); } } } diff --git a/backend/src/main.rs b/backend/src/main.rs index 7b1d0d5a3..2f410e230 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -14,7 +14,7 @@ use crate::model::{ AppConfig, Config, Healthcheck, HealthcheckConfig, ProcessTargets, SourcesConfig, }; use crate::processing::processor::playlist; -use crate::utils::init_logger; +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; @@ -85,6 +85,13 @@ struct Args { /// Force rescan of all local Library files #[arg(long = "force-library-rescan", default_value_t = false, default_missing_value = "true")] force_library_rescan: bool, + + #[arg(long = "dbt", default_missing_value = "None")] + db_content_type: Option, + + #[arg(long = "dbf", default_missing_value = "None")] + db_file_name: Option, + } const VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -102,6 +109,11 @@ const BUILD_TIMESTAMP: &str = env!("VERGEN_BUILD_TIMESTAMP"); async fn main() { let args = Args::parse(); + match (args.db_content_type.as_ref(), args.db_file_name.as_ref()) { + (Some(dbt), Some(dbf)) => db_viewer(dbf, dbt), + (Some(_) | None, None) | (None, Some(_))=> info!("You need to define database content type and filename!"), + } + if args.genpwd { match generate_password() { Ok(pwd) => println!("{pwd}"), diff --git a/backend/src/model/config/api_proxy.rs b/backend/src/model/config/api_proxy.rs index 672b499dc..d885266bc 100644 --- a/backend/src/model/config/api_proxy.rs +++ b/backend/src/model/config/api_proxy.rs @@ -128,7 +128,7 @@ impl ApiProxyConfig { } } else { let user_db_path = get_api_user_db_path(cfg); - if user_db_path.exists() { + if let Ok(true) = tokio::fs::try_exists(&user_db_path).await { // we cant have user defined in db file. // we need to load them and save them into the config file if let Ok(stored_users) = load_api_user(cfg).await { diff --git a/backend/src/processing/playlist_watch.rs b/backend/src/processing/playlist_watch.rs index 60a8fe1b2..dcd24e999 100644 --- a/backend/src/processing/playlist_watch.rs +++ b/backend/src/processing/playlist_watch.rs @@ -20,7 +20,7 @@ pub async fn process_group_watch(client: &reqwest::Client, cfg: &Config, target_ Some(path) => { let save_path = path.as_path(); let mut changed = false; - if path.exists() { + if let Ok(true) = tokio::fs::try_exists(&path).await { if let Some(loaded_tree) = load_watch_tree(&path).await { // Find elements in set2 but not in set1 let added_difference: BTreeSet = new_tree.difference(&loaded_tree).cloned().collect(); diff --git a/backend/src/processing/processor/library.rs b/backend/src/processing/processor/library.rs index 0fc226cc2..d948fff71 100644 --- a/backend/src/processing/processor/library.rs +++ b/backend/src/processing/processor/library.rs @@ -164,7 +164,7 @@ pub fn metadata_cache_entry_to_xtream_movie_info( name: movie.title.clone(), category_id: 0, stream_id: 0, - stream_icon: String::new(), + stream_icon: movie.poster.as_deref().or(movie.fanart.as_deref()).unwrap_or("").to_owned(), direct_source: String::new(), custom_sid: None, added: entry.file_modified.to_string(), diff --git a/backend/src/processing/processor/xtream_series.rs b/backend/src/processing/processor/xtream_series.rs index 39933bfd8..b35d4c749 100644 --- a/backend/src/processing/processor/xtream_series.rs +++ b/backend/src/processing/processor/xtream_series.rs @@ -11,7 +11,7 @@ use shared::model::{PlaylistGroup, PlaylistItemType, XtreamCluster}; use std::sync::Arc; use std::time::Instant; use crate::repository::storage::get_input_storage_path; -use crate::repository::xtream_repository::persists_input_series_info; +use crate::repository::xtream_repository::{persists_input_series_info}; create_resolve_options_function_for_xtream_target!(series); @@ -64,9 +64,8 @@ async fn playlist_resolve_series_info(app_config: &Arc, client: &reqw if let Some(episodes) = parse_xtream_series_info(&pli.get_uuid(), &series_stream_props, &group, &series_name, input) { group_series.extend(episodes.into_iter()); } - // Update in-memory playlist items with the newly fetched vod info. - // This makes the data available for subsequent processing steps like STRM export. + // This makes the data available for later processing steps like STRM export. pli.header.additional_properties = Some(StreamProperties::Series(Box::new(series_stream_props))); } } diff --git a/backend/src/processing/processor/xtream_vod.rs b/backend/src/processing/processor/xtream_vod.rs index 7b987a0ef..e56c39a59 100644 --- a/backend/src/processing/processor/xtream_vod.rs +++ b/backend/src/processing/processor/xtream_vod.rs @@ -3,13 +3,13 @@ use crate::model::{AppConfig, ConfigTarget}; use crate::processing::processor::xtream::playlist_resolve_download_playlist_item; use crate::processing::processor::create_resolve_options_function_for_xtream_target; use crate::repository::storage::get_input_storage_path; -use crate::repository::xtream_repository::persists_input_vod_info; use log::{error, info, log_enabled, Level}; use shared::error::TuliproxError; use shared::model::{InputType, PlaylistEntry, StreamProperties, VideoStreamProperties, XtreamVideoInfo}; use shared::model::{PlaylistItemType, XtreamCluster}; use std::sync::Arc; use std::time::Instant; +use crate::repository::xtream_repository::{persist_input_vod_info}; create_resolve_options_function_for_xtream_target!(vod); @@ -52,7 +52,7 @@ pub async fn playlist_resolve_vod(app_config: &Arc, client: &reqwest: if let Some(content) = playlist_resolve_download_playlist_item(client, pli, fpl.input, errors, resolve_delay, XtreamCluster::Video).await { if let Ok(info) = serde_json::from_str::(&content) { let video_stream_props = VideoStreamProperties::from_info(&info, pli); - if let Err(err) = persists_input_vod_info(app_config, &storage_path, pli.header.xtream_cluster, &input.name, provider_id, &video_stream_props).await { + if let Err(err) = persist_input_vod_info(app_config, &storage_path, pli.header.xtream_cluster, &input.name, provider_id, &video_stream_props).await { error!("Failed to persist VOD info for provider_id {provider_id}: {err}"); } // This makes the data available for subsequent processing steps like STRM export. diff --git a/backend/src/repository/bplustree.rs b/backend/src/repository/bplustree.rs index 3ec088e5b..fdbf2ff1c 100644 --- a/backend/src/repository/bplustree.rs +++ b/backend/src/repository/bplustree.rs @@ -1610,6 +1610,7 @@ mod tests { tree.insert(i, Record { id: i, data: format!("{content} {i}") }); } tree.store(&filepath)?; + drop(tree); // Update via BPlusTreeUpdate let mut tree_update = BPlusTreeUpdate::::try_new(&filepath)?; @@ -1803,6 +1804,7 @@ mod tests { tree.insert(i, Record { id: i, data: format!("Value {i}") }); } tree.store(&filepath)?; + drop(tree); let mut query = BPlusTreeQuery::::try_new(&filepath)?; diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index e5fb1ae10..fb06a5f3a 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -9,7 +9,7 @@ use crate::utils; use shared::error::{create_tuliprox_error, info_err}; use shared::error::{str_to_io_error, TuliproxError, TuliproxErrorKind}; use shared::model::PlaylistItemType; -use shared::model::{M3uPlaylistItem, PlaylistGroup, PlaylistItem}; +use shared::model::{M3uPlaylistItem, PlaylistGroup}; use std::io::Error; use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -92,7 +92,7 @@ pub async fn m3u_write_playlist( .iter() .flat_map(|pg| &pg.channels) .filter(|&pli| !matches!(pli.header.item_type, PlaylistItemType::SeriesInfo | PlaylistItemType::LocalSeriesInfo)) - .map(PlaylistItem::to_m3u) + .map(M3uPlaylistItem::from) .collect::>(), ); @@ -155,7 +155,7 @@ pub async fn m3u_get_item_for_stream_id(stream_id: u32, app_state: &AppState, ta pub async fn iter_raw_m3u_playlist(config: &AppConfig, target: &ConfigTarget) -> Option<(utils::FileReadGuard, impl Iterator)> { let target_path = get_target_storage_path(&config.config.load(), target.name.as_str())?; let m3u_path = m3u_get_file_path(&target_path); - if !m3u_path.exists() { + if let Ok(false) = tokio::fs::try_exists(&m3u_path).await { return None; } let file_lock = config.file_locks.read_lock(&m3u_path).await; diff --git a/backend/src/repository/playlist_scratch.rs b/backend/src/repository/playlist_scratch.rs index c401a905c..788301467 100644 --- a/backend/src/repository/playlist_scratch.rs +++ b/backend/src/repository/playlist_scratch.rs @@ -59,13 +59,13 @@ impl PlaylistScratch { } } - pub fn get(&self, cluster: XtreamCluster) -> &C { - match cluster { - XtreamCluster::Live => &self.live, - XtreamCluster::Video => &self.vods, - XtreamCluster::Series => &self.series, - } - } + // pub fn get(&self, cluster: XtreamCluster) -> &C { + // match cluster { + // XtreamCluster::Live => &self.live, + // XtreamCluster::Video => &self.vods, + // XtreamCluster::Series => &self.series, + // } + // } pub fn set(&mut self, cluster: XtreamCluster, content: C) { match cluster { diff --git a/backend/src/repository/storage_const.rs b/backend/src/repository/storage_const.rs index 0b90c4c5c..46ac5cab9 100644 --- a/backend/src/repository/storage_const.rs +++ b/backend/src/repository/storage_const.rs @@ -3,8 +3,6 @@ pub(in crate::repository) const FILE_SUFFIX_DB: &str = "db"; pub(in crate::repository) const FILE_ID_MAPPING: &str = "id_mapping.db"; pub(in crate::repository) const FILE_STRM: &str = "strm"; pub(in crate::repository) const FILE_M3U: &str = "m3u"; - -pub const FILE_SUFFIX_WAL: &str = "wal"; pub const M3U_STREAM_PATH: &str = "m3u-stream"; pub const M3U_RESOURCE_PATH: &str = "resource/m3u"; pub const EPG_RESOURCE_PATH: &str = "resource/epg"; @@ -17,13 +15,5 @@ pub(in crate::repository) const USER_LIVE_BOUQUET: &str = "live_bouquet.json"; pub(in crate::repository) const USER_VOD_BOUQUET: &str = "vod_bouquet.json"; pub(in crate::repository) const USER_SERIES_BOUQUET: &str = "series_bouquet.json"; pub(in crate::repository) const API_USER_DB_FILE: &str = "api_user.db"; - - -pub(in crate::repository) const FILE_SERIES_INFO: &str = "series_info"; -pub(in crate::repository) const FILE_VOD_INFO: &str = "vod_info"; -pub(in crate::repository) const FILE_VOD_INFO_RECORD: &str = "vod_info_record"; -pub(in crate::repository) const FILE_SERIES_INFO_RECORD: &str = "series_info_record"; -pub(in crate::repository) const FILE_SERIES_EPISODE_RECORD: &str = "series_episode_record"; -// pub(in crate::repository) const FILE_SERIES: &str = "series"; pub(in crate::repository) const PATH_XTREAM: &str = "xtream"; diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 07483f8d6..61011972c 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -1,33 +1,31 @@ use crate::api::model::AppState; use crate::model::PlaylistXtreamCategory; use crate::model::{AppConfig, ProxyUserCredentials}; -use crate::model::{Config, ConfigInput, ConfigTarget}; +use crate::model::{Config, ConfigTarget}; use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery, BPlusTreeUpdate}; use crate::repository::playlist_scratch::PlaylistScratch; -use crate::repository::storage::{get_input_storage_path, get_target_id_mapping_file, get_target_storage_path}; +use crate::repository::storage::{get_target_id_mapping_file, get_target_storage_path}; use crate::repository::storage_const; use crate::repository::target_id_mapping::VirtualIdRecord; use crate::repository::xtream_playlist_iterator::XtreamPlaylistJsonIterator; +use crate::utils::file_reader; use crate::utils::json_write_documents_to_file; use crate::utils::FileReadGuard; -use crate::utils::{async_file_reader, async_open_readonly_file, file_reader}; use bytes::Bytes; use futures::{stream, Stream, StreamExt}; use indexmap::IndexMap; use log::error; use serde::Serialize; use serde_json::{json, Value}; -use shared::error::{create_tuliprox_error, create_tuliprox_error_result, info_err, notify_err, str_to_io_error, to_io_error, TuliproxError, TuliproxErrorKind}; +use shared::error::{create_tuliprox_error, create_tuliprox_error_result, info_err, notify_err, str_to_io_error, TuliproxError, TuliproxErrorKind}; 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::collections::HashMap; -use std::fs; use std::fs::File; use std::io::{Error, ErrorKind}; use std::path::{Path, PathBuf}; use std::sync::Arc; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio::task; macro_rules! cant_write_result { @@ -41,15 +39,6 @@ macro_rules! cant_write_result { }; } -macro_rules! try_option_ok { - ($option:expr) => { - match $option { - Some(value) => value, - None => return Ok(()), - } - }; -} - #[inline] fn get_collection_path(path: &Path, collection: &str) -> PathBuf { path.join(format!("{collection}.json")) @@ -87,33 +76,17 @@ fn ensure_xtream_storage_path(cfg: &Config, target_name: &str) -> Result Option { - if cluster == XtreamCluster::Series { - return Some(storage_path.join(format!("{}.{}", storage_const::FILE_SERIES_INFO, storage_const::FILE_SUFFIX_DB))); - } else if cluster == XtreamCluster::Video { - return Some(storage_path.join(format!("{}.{}", storage_const::FILE_VOD_INFO, storage_const::FILE_SUFFIX_DB))); - } - None +#[derive(Debug, Copy, Clone)] +enum StorageKey { + VirtualId, + ProviderId, } -pub fn xtream_get_record_file_path(storage_path: &Path, item_type: PlaylistItemType) -> Option { - match item_type { - PlaylistItemType::Video - | PlaylistItemType::LocalVideo => Some(storage_path.join(format!("{}.{}", storage_const::FILE_VOD_INFO_RECORD, storage_const::FILE_SUFFIX_DB))), - PlaylistItemType::SeriesInfo - | PlaylistItemType::LocalSeriesInfo => Some(storage_path.join(format!("{}.{}", storage_const::FILE_SERIES_INFO_RECORD, storage_const::FILE_SUFFIX_DB))), - PlaylistItemType::Series - | PlaylistItemType::LocalSeries => Some(storage_path.join(format!("{}.{}", storage_const::FILE_SERIES_EPISODE_RECORD, storage_const::FILE_SUFFIX_DB))), - _ => None, - } -} async fn write_playlists_to_file( app_config: &Arc, storage_path: &Path, - collections: Vec<(XtreamCluster, &[&mut PlaylistItem])>, + storage_key: StorageKey, + collections: Vec<(XtreamCluster, Vec)>, ) -> Result<(), TuliproxError> { for (cluster, playlist) in collections { if playlist.is_empty() { @@ -124,31 +97,34 @@ async fn write_playlists_to_file( let _file_lock = app_config.file_locks.write_lock(&xtream_path).await; let mut tree = BPlusTree::new(); for item in playlist { - tree.insert(item.header.virtual_id, item.to_xtream()); + tree.insert(match storage_key { + StorageKey::VirtualId => item.virtual_id, + StorageKey::ProviderId => item.provider_id, + }, item); } tree.store(&xtream_path).map_err(|err| cant_write_result!(&xtream_path, err))?; } } Ok(()) } - -async fn write_playlists_to_file_2( - app_config: &Arc, - storage_path: &Path, - cluster: XtreamCluster, - playlist: &[PlaylistItem], -) -> 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.header.virtual_id, item.to_xtream()); - } - tree.store(&xtream_path).map_err(|err| cant_write_result!(&xtream_path, err))?; - } - 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_to_file( app_config: &Arc, @@ -233,28 +209,6 @@ pub fn xtream_get_file_path(storage_path: &Path, cluster: XtreamCluster) -> Path xtream_get_file_path_for_name(storage_path, &cluster.as_str().to_lowercase()) } -// pub fn xtream_get_file_paths_for_series(storage_path: &Path) -> (PathBuf, PathBuf) { -// xtream_get_file_paths_for_name(storage_path, storage_const::FILE_SERIES) -// } - -async fn xtream_garbage_collect(app_cfg: &Arc, target_name: &str) -> std::io::Result<()> { - // Garbage collect series - let storage_path = { - let cfg = app_cfg.config.load(); - try_option_ok!(xtream_get_storage_path(&cfg, target_name)) - }; - let info_path = try_option_ok!(xtream_get_info_file_path( - &storage_path, - XtreamCluster::Series - )); - if info_path.exists() { - let _file_lock = app_cfg.file_locks.write_lock(&info_path).await; - let mut tree = BPlusTreeUpdate::::try_new(&info_path)?; - tree.compact(&info_path)?; - } - Ok(()) -} - #[derive(Serialize)] struct CategoryEntry { category_id: u32, @@ -336,25 +290,17 @@ pub async fn xtream_write_playlist( }).await.map_err(|e| notify_err!(format!("Task panicked: {}", e)))?; errors.extend(write_errors); - match write_playlists_to_file( + if let Err(err) = write_playlists_to_file( app_cfg, &path, + StorageKey::VirtualId, vec![ - (XtreamCluster::Live, &live_col), - (XtreamCluster::Video, &vod_col), - (XtreamCluster::Series, &series_col), + (XtreamCluster::Live, live_col.iter().map(|item| XtreamPlaylistItem::from(&**item)).collect::>()), + (XtreamCluster::Video, vod_col.iter().map(|item| XtreamPlaylistItem::from(&**item)).collect::>()), + (XtreamCluster::Series, series_col.iter().map(|item| XtreamPlaylistItem::from(&**item)).collect::>()), ], ).await { - Ok(()) => { - if let Err(err) = xtream_garbage_collect(app_cfg, &target.name).await { - if err.kind() != ErrorKind::NotFound { - errors.push(format!("Garbage collection failed:{err}")); - } - } - } - Err(err) => { - errors.push(format!("Persisting collection failed:{err}")); - } + errors.push(format!("Persisting collection failed:{err}")); } if !errors.is_empty() { @@ -493,11 +439,11 @@ pub async fn xtream_get_item_for_stream_id( app_state: &Arc, target: &ConfigTarget, xtream_cluster: Option, -) -> Result<(XtreamPlaylistItem, VirtualIdRecord), Error> { +) -> Result { if target.use_memory_cache { - if let Ok(Some((playlist_item, virtual_record))) = + if let Ok(Some((playlist_item, _virtual_record))) = xtream_get_item_for_stream_id_from_memory(virtual_id, app_state, target, xtream_cluster).await { - return Ok((playlist_item, virtual_record)); + return Ok(playlist_item); } // fall through to disk lookup on cache miss } @@ -507,42 +453,46 @@ pub async fn xtream_get_item_for_stream_id( let target_path = get_target_storage_path(&config, target.name.as_str()).ok_or_else(|| str_to_io_error(&format!("Could not find path for target {}", &target.name)))?; let storage_path = xtream_get_storage_path(&config, target.name.as_str()).ok_or_else(|| str_to_io_error(&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 = app_config.file_locks.read_lock(&target_id_mapping_file).await; + let result = if let Some(cluster) = xtream_cluster { + xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, cluster).await + } else { + let target_id_mapping_file = get_target_id_mapping_file(&target_path); + let _file_lock = app_config.file_locks.read_lock(&target_id_mapping_file).await; - let mut target_id_mapping = BPlusTreeQuery::::try_new(&target_id_mapping_file).map_err(|err| str_to_io_error(&format!("Could not load id mapping for target {} err:{err}", target.name)))?; - let mapping = target_id_mapping.query(&virtual_id).ok_or_else(|| str_to_io_error(&format!("Could not find mapping for target {} and id {}", target.name, virtual_id)))?; - let result = match mapping.item_type { - PlaylistItemType::SeriesInfo - | PlaylistItemType::LocalSeriesInfo => { - xtream_read_series_item_for_stream_id(app_config, virtual_id, &storage_path).await - } - PlaylistItemType::Series - | PlaylistItemType::LocalSeries => { - if let Ok(mut item) = xtream_read_series_item_for_stream_id(app_config, mapping.parent_virtual_id, &storage_path).await { + let mut target_id_mapping = BPlusTreeQuery::::try_new(&target_id_mapping_file).map_err(|err| str_to_io_error(&format!("Could not load id mapping for target {} err:{err}", target.name)))?; + let mapping = target_id_mapping.query(&virtual_id).ok_or_else(|| str_to_io_error(&format!("Could not find mapping for target {} and id {}", target.name, virtual_id)))?; + match mapping.item_type { + PlaylistItemType::SeriesInfo + | PlaylistItemType::LocalSeriesInfo => { + xtream_read_series_item_for_stream_id(app_config, virtual_id, &storage_path).await + } + PlaylistItemType::Series + | PlaylistItemType::LocalSeries => { + if let Ok(mut item) = xtream_read_series_item_for_stream_id(app_config, mapping.parent_virtual_id, &storage_path).await { + item.provider_id = mapping.provider_id; + item.item_type = PlaylistItemType::Series; + item.virtual_id = mapping.virtual_id; + Ok(item) + } else { + xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, XtreamCluster::Series).await + } + } + PlaylistItemType::Catchup => { + let cluster = try_cluster!(xtream_cluster, mapping.item_type, virtual_id)?; + let mut item = xtream_read_item_for_stream_id(app_config, mapping.parent_virtual_id, &storage_path, cluster).await?; item.provider_id = mapping.provider_id; - item.item_type = PlaylistItemType::Series; + item.item_type = PlaylistItemType::Catchup; item.virtual_id = mapping.virtual_id; Ok(item) - } else { - xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, XtreamCluster::Series).await } - } - PlaylistItemType::Catchup => { - let cluster = try_cluster!(xtream_cluster, mapping.item_type, virtual_id)?; - let mut item = xtream_read_item_for_stream_id(app_config, mapping.parent_virtual_id, &storage_path, cluster).await?; - item.provider_id = mapping.provider_id; - item.item_type = PlaylistItemType::Catchup; - item.virtual_id = mapping.virtual_id; - Ok(item) - } - _ => { - let cluster = try_cluster!(xtream_cluster, mapping.item_type, virtual_id)?; - xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, cluster).await + _ => { + let cluster = try_cluster!(xtream_cluster, mapping.item_type, virtual_id)?; + xtream_read_item_for_stream_id(app_config, virtual_id, &storage_path, cluster).await + } } }; - result.map(|xpli| (xpli, mapping)) + result } } @@ -556,159 +506,6 @@ pub async fn xtream_load_rewrite_playlist( XtreamPlaylistJsonIterator::new(cluster, config, target, category_id, user).await } -pub async fn xtream_write_series_info( - app_config: &AppConfig, - target_name: &str, - series_info_id: u32, - content: &StreamProperties, -) -> Result<(), Error> { - if let StreamProperties::Series(_series) = content { - let config = app_config.config.load(); - let target_path = try_option_ok!(get_target_storage_path(&config, target_name)); - let storage_path = try_option_ok!(xtream_get_storage_path(&config, target_name)); - let info_path = try_option_ok!(xtream_get_info_file_path( - &storage_path, - XtreamCluster::Series - )); - - { - let _file_lock = app_config.file_locks.write_lock(&info_path).await; - let mut tree = BPlusTreeUpdate::try_new(&info_path).map_err(|_| str_to_io_error("failed to open series info for update"))?; - tree.update(&series_info_id, content.clone()).map_err(|_| str_to_io_error("failed to update series info"))?; - } - { - let target_id_mapping_file = get_target_id_mapping_file(&target_path); - let _file_lock = app_config.file_locks.write_lock(&target_id_mapping_file).await; - if let Ok(mut target_id_mapping) = BPlusTreeUpdate::::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); - } - } - } - - Ok(()) - } else { - Err(std::io::Error::new(ErrorKind::InvalidData, "No series info data found!".to_string())) - } -} - -pub async fn xtream_write_vod_info( - app_config: &AppConfig, - target_name: &str, - virtual_id: u32, - content: &StreamProperties, -) -> Result<(), Error> { - if let StreamProperties::Video(video) = content { - let config = app_config.config.load(); - let storage_path = try_option_ok!(xtream_get_storage_path(&config, target_name)); - let info_path = try_option_ok!(xtream_get_info_file_path(&storage_path, XtreamCluster::Video)); - { - let _file_lock = app_config.file_locks.write_lock(&info_path).await; - let mut tree = BPlusTreeUpdate::try_new(&info_path).map_err(|_| str_to_io_error("failed to open vod info for update"))?; - tree.update(&virtual_id, StreamProperties::Video(video.clone())).map_err(|_| str_to_io_error("failed to update vod info"))?; - } - Ok(()) - } else { - Err(std::io::Error::new(ErrorKind::InvalidData, "No video info data found!".to_string())) - } -} - -pub async fn xtream_get_input_info( - cfg: &AppConfig, - input: &ConfigInput, - provider_id: u32, - cluster: XtreamCluster, -) -> Option { - if let Ok(Some(info_path)) = get_input_storage_path(&input.name, &cfg.config.load().working_dir).map(|storage_path| xtream_get_info_file_path(&storage_path, cluster)) - { - let _file_lock = cfg.file_locks.read_lock(&info_path).await; - let mut query = BPlusTreeQuery::::try_new(&info_path).ok()?; - return query.query(&provider_id); - } - None -} - -pub async fn xtream_update_input_info_file( - cfg: &AppConfig, - input: &ConfigInput, - wal_path: &Path, - cluster: XtreamCluster, -) -> Result<(), TuliproxError> { - let config = cfg.config.load(); - if let Ok(Some(info_path)) = get_input_storage_path(&input.name, &config.working_dir).map(|storage_path| xtream_get_info_file_path(&storage_path, cluster)) { - let _file_lock = cfg.file_locks.write_lock(&info_path).await; - let mut reader = async_file_reader(async_open_readonly_file(wal_path).await.map_err(|err| notify_err!(format!("Could not read {cluster} info {err}")))?); - - let mut tree = if info_path.exists() { - BPlusTreeUpdate::::try_new(&info_path).map_err(|err| notify_err!(format!("Could not open {cluster} info for update {err}")))? - } else { - return Err(notify_err!(format!("BPlusTree file not found for {cluster} info"))); - }; - - let mut provider_id_bytes = [0u8; 4]; - let mut length_bytes = [0u8; 4]; - loop { - if reader.read_exact(&mut provider_id_bytes).await.is_err() { - break; // End of file - } - let provider_id = u32::from_le_bytes(provider_id_bytes); - reader.read_exact(&mut length_bytes).await.map_err(|err| notify_err!(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).await.map_err(|err| notify_err!(format!("Could not read temporary {cluster} info {err}")))?; - if let Ok(content) = String::from_utf8(buffer) { - let _ = tree.update(&provider_id, content); - } - } - drop(reader); - if let Err(err) = tree.compact(&info_path) { - error!("Failed to compact {cluster} info: {err}"); - } - if let Err(err) = fs::remove_file(wal_path) { - error!("Failed to delete WAL file for {cluster} {err}"); - } - Ok(()) - } else { - Err(notify_err!(format ! ("Could not determine storage path for input {}", & input.name))) - } -} - -pub async fn xtream_update_input_series_record_from_wal_file( - cfg: &AppConfig, - input: &ConfigInput, - wal_path: &Path, -) -> Result<(), TuliproxError> { - let config = cfg.config.load(); - let record_path = get_input_storage_path(&input.name, &config.working_dir).map(|storage_path| xtream_get_record_file_path(&storage_path, PlaylistItemType::SeriesInfo)) - .map_err(|err| notify_err!(format!("Error accessing storage path: {err}"))) - .and_then(|opt| opt.ok_or_else(|| notify_err!(format!("Error accessing storage path for input: {}", &input.name))))?; - { - let _file_lock = cfg.file_locks.write_lock(&record_path).await; - let mut reader = async_file_reader(async_open_readonly_file(wal_path).await.map_err(|err| notify_err!(format!("Could not read series wal info {err}")))?); - 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).await.is_err() { - break; // End of file - } - let provider_id = u32::from_le_bytes(provider_id_bytes); - if reader.read_exact(&mut ts_bytes).await.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| notify_err!(format!("Could not store series record info {err}")))?; - drop(reader); - if let Err(err) = tokio::fs::remove_file(wal_path).await { - error!("Failed to delete record WAL file for series {err}"); - } - Ok(()) - } -} - pub async fn iter_raw_xtream_playlist(app_config: &AppConfig, target: &ConfigTarget, cluster: XtreamCluster) -> Option<(FileReadGuard, impl Iterator)> { let config = app_config.config.load(); if let Some(storage_path) = xtream_get_storage_path(&config, target.name.as_str()) { @@ -770,34 +567,6 @@ pub(crate) async fn xtream_get_playlist_categories(config: &Config, target_name: None } -pub async fn write_series_info_to_wal_file(provider_id: u32, ts: u64, content: &str, - content_write: &mut tokio::io::BufWriter<&mut tokio::fs::File>, - record_writer: &mut tokio::io::BufWriter<&mut tokio::fs::File>) -> std::io::Result<()> { - let encoded_content = encode_info_content_for_wal_file(provider_id, content)?; - let encoded_record = encode_series_info_record_for_wal_file(provider_id, ts); - content_write.write_all(&encoded_content).await?; - record_writer.write_all(&encoded_record).await?; - Ok(()) -} - -fn encode_info_content_for_wal_file(provider_id: u32, content: &str) -> std::io::Result> { - let length = u32::try_from(content.len()).map_err(to_io_error)?; - let mut buffer = Vec::with_capacity(8 + content.len()); - - buffer.extend_from_slice(&provider_id.to_le_bytes()); - buffer.extend_from_slice(&length.to_le_bytes()); - buffer.extend_from_slice(content.as_bytes()); - - Ok(buffer) -} - -fn encode_series_info_record_for_wal_file(provider_id: u32, ts: u64) -> Vec { - let mut buffer = Vec::with_capacity(12); - buffer.extend_from_slice(&provider_id.to_le_bytes()); - buffer.extend_from_slice(&ts.to_le_bytes()); - buffer -} - #[allow(clippy::too_many_lines)] pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_path: &Path, playlist: Vec) -> (Vec, Option) { @@ -805,19 +574,17 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ let mut fetched_categories = PlaylistScratch::>::new(1_000); let mut fetched_scratch = PlaylistScratch::>::new(50_000); - let mut stored_scratch = PlaylistScratch::>::new(50_000); + let mut stored_scratch = PlaylistScratch::>::new(50_000); // load for cluster in XTREAM_CLUSTER { let xtream_path = xtream_get_file_path(storage_path, cluster); - if xtream_path.exists() { + if let Ok(true) = tokio::fs::try_exists(&xtream_path).await { let file_lock = app_config.file_locks.read_lock(&xtream_path).await; let stored_entries = stored_scratch.get_mut(cluster); - if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { + if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { for (_, doc) in query.iter() { - if let Ok(provider_id) = doc.header.id.parse::() { - stored_entries.insert(provider_id, doc); - } + stored_entries.insert(doc.provider_id, doc); } } drop(file_lock); @@ -841,7 +608,7 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ if let Ok(provider_id) = pli.header.id.parse::() { if let Some(stored_pli) = stored_col.get_mut(&provider_id) { - if let (Some(new_stream_props), Some(old_stream_props)) = (&mut pli.header.additional_properties, stored_pli.header.additional_properties.take()) { + if let (Some(new_stream_props), Some(old_stream_props)) = (&mut pli.header.additional_properties, stored_pli.additional_properties.take()) { if !needs_update_info_details(new_stream_props, &old_stream_props) { match (new_stream_props, old_stream_props) { (StreamProperties::Video(value_1), StreamProperties::Video(value_2)) => { @@ -865,7 +632,7 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ let mut processed_scratch = PlaylistScratch::>::new(0); for xc in XTREAM_CLUSTER { processed_scratch.set(xc, if !stored_scratch.is_empty(xc) && fetched_scratch.is_empty(xc) { - stored_scratch.take(xc).into_values().collect() + stored_scratch.take(xc).iter().map(|(_, item)| PlaylistItem::from(item)).collect::>() } else { fetched_scratch.take(xc) }); @@ -886,10 +653,7 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ let data = fetched_categories.get_mut(cluster); let lock = app_cfg.file_locks.write_lock(&col_path); if let Err(err) = json_write_documents_to_file(&col_path, data) { - write_errors.push(format!( - "Persisting collection failed: {}: {err}", - col_path.display() - )); + write_errors.push(format!("Persisting collection failed: {}: {err}", col_path.display())); } drop(lock); } @@ -897,36 +661,18 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ }).await.map_err(|e| notify_err!(format!("Task panicked: {}", e))); for cluster in XTREAM_CLUSTER { - let data = processed_scratch.get(cluster); - match write_playlists_to_file_2( + let col = processed_scratch.take(cluster); + + // persist playlist + if let Err(err) = write_playlists_to_file( app_config, storage_path, - cluster, - data, + StorageKey::ProviderId, + vec![(cluster, col.iter().map(Into::into).collect::>())], ).await { - Ok(()) => { - - // TODO GARBAGE collect - - // if let Err(err) = xtream_garbage_collect(app_config, &target.name).await { - // if err.kind() != ErrorKind::NotFound { - // errors.push(format!("Garbage collection failed:{err}")); - // } - // } - } - Err(err) => { - errors.push(format!("Persisting collection failed:{err}")); - } + errors.push(format!("Persisting collection failed:{err}")); } - } - match write_errors { - Ok(write_err) => errors.extend(write_err), - Err(err) => errors.push(err.to_string()), - } - - for xc in XTREAM_CLUSTER { - let col = processed_scratch.take(xc); for item in col { groups .entry(item.header.category_id) @@ -941,6 +687,11 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ } } + match write_errors { + Ok(write_err) => errors.extend(write_err), + Err(err) => errors.push(err.to_string()), + } + let result = groups.into_iter().map(|(_, group)| group).collect(); let err = if errors.is_empty() { @@ -967,30 +718,32 @@ fn needs_update_info_details( } } -async fn persist_input_info(app_config: &Arc, storage_path: &Path, cluster: XtreamCluster, - input_name: &str, provider_id: u32, props: T) -> Result<(), Error> -where - T: serde::Serialize + for<'de> serde::Deserialize<'de> + Clone, -{ +async fn persist_input_info(app_config: &Arc, storage_path: &Path, cluster: XtreamCluster, + input_name: &str, provider_id: u32, props: StreamProperties) -> Result<(), Error> { let xtream_path = xtream_get_file_path(storage_path, cluster); if xtream_path.exists() { { let _file_lock = app_config.file_locks.write_lock(&xtream_path).await; - let mut tree = BPlusTreeUpdate::try_new(&xtream_path).map_err(|err| Error::other(format!("failed to open BPlusTree for input {input_name}: {err}")))?; - tree.update(&provider_id, props).map_err(|err| Error::other(format!("failed to write {cluster} info for input {input_name}: {err}")))?; + let mut tree: BPlusTreeUpdate = BPlusTreeUpdate::try_new(&xtream_path).map_err(|err| Error::other(format!("failed to open BPlusTree for input {input_name}: {err}")))?; + if let Some(mut pli) = tree.query(&provider_id) { + pli.additional_properties = Some(props); + tree.update(&provider_id, pli).map_err(|err| Error::other(format!("failed to write {cluster} info for input {input_name}: {err}")))?; + } else { + error!("Could not find input entry for provider_id: {provider_id} and input: {input_name}"); + } } } Ok(()) } -pub async fn persists_input_vod_info(app_config: &Arc, storage_path: &Path, - cluster: XtreamCluster, input_name: &str, provider_id: u32, - props: &VideoStreamProperties) -> Result<(), Error> { - persist_input_info::(app_config, storage_path, cluster, input_name, provider_id, props.clone()).await +pub async fn persist_input_vod_info(app_config: &Arc, storage_path: &Path, + cluster: XtreamCluster, input_name: &str, provider_id: u32, + props: &VideoStreamProperties) -> Result<(), Error> { + persist_input_info(app_config, storage_path, cluster, input_name, provider_id, StreamProperties::Video(Box::new(props.clone()))).await } pub async fn persists_input_series_info(app_config: &Arc, storage_path: &Path, cluster: XtreamCluster, input_name: &str, provider_id: u32, props: &SeriesStreamProperties) -> Result<(), Error> { - persist_input_info::(app_config, storage_path, cluster, input_name, provider_id, props.clone()).await + persist_input_info(app_config, storage_path, cluster, input_name, provider_id, StreamProperties::Series(Box::new(props.clone()))).await } \ No newline at end of file diff --git a/backend/src/utils/db_viewer.rs b/backend/src/utils/db_viewer.rs new file mode 100644 index 000000000..df5b68042 --- /dev/null +++ b/backend/src/utils/db_viewer.rs @@ -0,0 +1,59 @@ +use std::io::Write; +use std::path::PathBuf; +use serde::{Deserialize, Serialize}; +use shared::model::{M3uPlaylistItem, XtreamPlaylistItem}; +use crate::repository::bplustree::{BPlusTreeDiskIterator, BPlusTreeQuery}; + +pub fn db_viewer(filename: &str, content_type: &str) { + let path = match PathBuf::from(filename).canonicalize() { + Ok(p) => p, + Err(err) => { + println!("File does not exist! {err}"); + let _ = std::io::stdout().flush(); + std::process::exit(1); + } + }; + + if !path.exists() { + println!("File does not exist! {}", path.display()); + let _ = std::io::stdout().flush(); + std::process::exit(1); + } + match content_type { + "xtream" => { + if let Ok(mut query) = BPlusTreeQuery::::try_new(&path) { + let iterator = query.iter(); + print_json_from_iter(iterator); + } + }, + "m3u" => { + if let Ok(mut query) = BPlusTreeQuery::::try_new(&path) { + let iterator = query.iter(); + print_json_from_iter(iterator); + } + } + _ => println!("Allowed content types are: [m3u, xtream]"), + } + let _ = std::io::stdout().flush(); + std::process::exit(1); +} + +fn print_json_from_iter

(iterator: BPlusTreeDiskIterator) +where P: Serialize + for<'de> Deserialize<'de> + Clone +{ + println!("["); + let mut first = true; + for (_, entry) in iterator { + match serde_json::to_string(&entry) { + Ok(text) => { + if !first { + println!(","); + } + println!("{text}"); + first = false; + }, + Err(err) => eprintln!("Failed: {err}"), + } + } + println!("]"); +} diff --git a/backend/src/utils/file/file_utils.rs b/backend/src/utils/file/file_utils.rs index 9dc19881d..56f8d6511 100644 --- a/backend/src/utils/file/file_utils.rs +++ b/backend/src/utils/file/file_utils.rs @@ -351,9 +351,25 @@ pub fn cleanup_unlisted_files_with_suffix( for entry in (fs::read_dir(dir)?).flatten() { let path = entry.path(); - if path.is_file() && !keep_set.contains(&path) && path.file_name() - .and_then(|n| n.to_str()) - .is_some_and(|name| name.ends_with(suffix)) && fs::remove_file(&path).is_ok() { + if !path.is_file() || keep_set.contains(&path) { + continue; + } + + let delete = { + let zero_size = entry + .metadata() + .map(|m| m.len() == 0) + .unwrap_or(false); + + let suffix_match = path + .file_name() + .and_then(|n| n.to_str()) + .is_some_and(|name| name.ends_with(suffix)); + + zero_size || suffix_match + }; + + if delete && fs::remove_file(&path).is_ok() { debug!("Deleted {:?}", path.display()); } } diff --git a/backend/src/utils/mod.rs b/backend/src/utils/mod.rs index 4494602c6..a4c664bcc 100644 --- a/backend/src/utils/mod.rs +++ b/backend/src/utils/mod.rs @@ -10,12 +10,14 @@ mod json_utils; mod binary_utils; mod telegram; mod geoip; +mod db_viewer; pub use self::binary_utils::*; pub use self::logging::*; pub use self::trakt::*; pub use self::telegram::*; pub use self::geoip::*; +pub use self::db_viewer::*; #[macro_export] macro_rules! debug_if_enabled { diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index aa7c87605..68073abf5 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -1,4 +1,3 @@ -use std::collections::HashMap; use crate::api::model::AppState; use crate::messaging::send_message; use crate::model::{is_input_expired, xtream_mapping_option_from_target_options, Config, ConfigInput, ConfigTarget, XtreamLoginInfo}; @@ -8,13 +7,14 @@ use crate::processing::parser::xtream::parse_xtream_series_info; use crate::repository::playlist_repository::{get_target_id_mapping, rewrite_provider_series_info_episode_virtual_id, ProviderEpisodeKey}; use crate::repository::storage::{get_input_storage_path, get_target_storage_path}; use crate::repository::target_id_mapping::VirtualIdRecord; -use crate::repository::xtream_repository::{persists_input_series_info, persists_input_vod_info, write_playlist_item_to_file}; +use crate::repository::xtream_repository::{persist_input_vod_info, persists_input_series_info, write_playlist_item_to_file}; use crate::utils::request; use chrono::{DateTime, Utc}; use log::{error, info, warn}; use shared::error::{str_to_io_error, to_io_error, TuliproxError}; use shared::model::{MsgKind, PlaylistEntry, PlaylistGroup, ProxyUserStatus, SeriesStreamProperties, StreamProperties, VideoStreamProperties, XtreamCluster, XtreamPlaylistItem, XtreamSeriesInfo, XtreamVideoInfo}; use shared::utils::{extract_extension_from_url, get_i64_from_serde_value, get_string_from_serde_value, sanitize_sensitive_info}; +use std::collections::HashMap; use std::io::Error; use std::str::FromStr; use std::sync::Arc; @@ -85,13 +85,18 @@ pub async fn get_xtream_stream_info(client: &reqwest::Client, let working_dir = &app_config.config.load().working_dir; if let Ok(storage_path) = get_input_storage_path(&input.name, working_dir) { if let Ok(info) = serde_json::from_str::(&content) { + // parse downloaded info into StreamProperties let video_stream_props = VideoStreamProperties::from_info(&info, pli); - if let Err(err) = persists_input_vod_info(&app_state.app_config, &storage_path, cluster, &input.name, provider_id, &video_stream_props).await { + + // persist input info + if let Err(err) = persist_input_vod_info(&app_state.app_config, &storage_path, cluster, &input.name, provider_id, &video_stream_props).await { error!("Failed to persist video stream for input {}: {err}", &input.name); } + // update target playlist let mut vod_pli = pli.clone(); vod_pli.additional_properties = Some(StreamProperties::Video(Box::new(video_stream_props))); + if let Err(err) = write_playlist_item_to_file(app_config, &target.name, &vod_pli).await { error!("Failed to persist video stream: {err}"); } @@ -109,10 +114,13 @@ pub async fn get_xtream_stream_info(client: &reqwest::Client, if let Ok(storage_path) = get_input_storage_path(&input.name, working_dir) { if let Ok(info) = serde_json::from_str::(&content) { + // parse series info let series_stream_props = SeriesStreamProperties::from_info(&info, pli); + // update input db if let Err(err) = persists_input_series_info(app_config, &storage_path, cluster, &input.name, provider_id, &series_stream_props).await { error!("Failed to persist series info for input {}: {err}", &input.name); } + if let Some(mut episodes) = parse_xtream_series_info(&pli.get_uuid(), &series_stream_props, &group, &series_name, input) { let config = &app_state.app_config.config.load(); match get_target_storage_path(config, target.name.as_str()) { @@ -208,7 +216,8 @@ async fn xtream_login(cfg: &Config, client: &reqwest::Client, input: &InputSourc let content = if let Ok(content) = request::get_input_json_content(client, None, input, None).await { content } else { - let input_source_account_info = input.with_url(format!("{}&action={}", &input.url, crate::model::XC_ACTION_GET_ACCOUNT_INFO)); + let input_source_account_info = + input.with_url(format!("{}&action={}", &input.url, crate::model::XC_ACTION_GET_ACCOUNT_INFO)); match request::get_input_json_content(client, None, &input_source_account_info, None).await { Ok(content) => content, Err(err) => { @@ -230,7 +239,8 @@ async fn xtream_login(cfg: &Config, client: &reqwest::Client, input: &InputSourc login_info.status = Some(cur_status); if !matches!(cur_status, ProxyUserStatus::Active | ProxyUserStatus::Trial) { warn!("User status for user {username} is {cur_status:?}"); - send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User status for user {username} is {cur_status:?}")).await; + send_message(client, MsgKind::Info, cfg.messaging.as_ref(), + &format!("User status for user {username} is {cur_status:?}")).await; } } } @@ -251,7 +261,8 @@ async fn xtream_login(cfg: &Config, client: &reqwest::Client, input: &InputSourc } } -pub async fn notify_account_expire(exp_date: Option, cfg: &Config, client: &reqwest::Client, username: &str, input_name: &str) { +pub async fn notify_account_expire(exp_date: Option, cfg: &Config, client: &reqwest::Client, + username: &str, input_name: &str) { if let Some(expiration_timestamp) = exp_date { let now_secs = Utc::now().timestamp(); // UTC-Time if expiration_timestamp > now_secs { @@ -261,12 +272,14 @@ pub async fn notify_account_expire(exp_date: Option, cfg: &Config, client: if let Some(datetime) = DateTime::::from_timestamp(expiration_timestamp, 0) { let formatted = datetime.format("%Y-%m-%d %H:%M:%S").to_string(); warn!("User account for user {username} expires {formatted}"); - send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} expires {formatted}")).await; + send_message(client, MsgKind::Info, cfg.messaging.as_ref(), + &format!("User account for user {username} expires {formatted}")).await; } } } else { warn!("User account for user {username} is expired"); - send_message(client, MsgKind::Info, cfg.messaging.as_ref(), &format!("User account for user {username} for provider {input_name} is expired")).await; + send_message(client, MsgKind::Info, cfg.messaging.as_ref(), + &format!("User account for user {username} for provider {input_name} is expired")).await; } } } @@ -303,8 +316,10 @@ pub async fn download_xtream_playlist(cfg: &Arc, client: &reqwest::Clien if !skip_cluster.contains(xtream_cluster) { let input_source_category = input_source.with_url(format!("{base_url}&action={category}")); let input_source_stream = input_source.with_url(format!("{base_url}&action={stream}")); - let category_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), working_dir, format!("{category}_").as_str()); - let stream_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), working_dir, format!("{stream}_").as_str()); + let category_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), + working_dir, format!("{category}_").as_str()); + let stream_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), + working_dir, format!("{stream}_").as_str()); match futures::join!( request::get_input_json_content_as_stream(client, None, &input_source_category, category_file_path), @@ -319,7 +334,8 @@ pub async fn download_xtream_playlist(cfg: &Arc, client: &reqwest::Clien if let Some(mut xtream_sub_playlist) = sub_playlist_parsed { playlist_groups.append(&mut xtream_sub_playlist); } else { - error!("Could not parse playlist {xtream_cluster} for input {}: {}", input_source.name, sanitize_sensitive_info(&input_source.url)); + error!("Could not parse playlist {xtream_cluster} for input {}: {}", + input_source.name, sanitize_sensitive_info(&input_source.url)); } } Err(err) => errors.push(err) @@ -343,7 +359,8 @@ async fn check_alias_user_state(cfg: &Arc, client: &reqwest::Client, inp if let Some(aliases) = input.aliases.as_ref() { for alias in aliases { if is_input_expired(alias.exp_date) { - notify_account_expire(alias.exp_date, cfg, client, alias.username.as_ref().map_or("", |s| s.as_str()), &alias.name).await; + notify_account_expire(alias.exp_date, cfg, client, alias.username.as_ref() + .map_or("", |s| s.as_str()), &alias.name).await; } } } @@ -388,18 +405,18 @@ async fn check_alias_user_state(cfg: &Arc, client: &reqwest::Client, inp // }); } -pub fn create_vod_info_from_item(target: &ConfigTarget, user: &ProxyUserCredentials, pli: &XtreamPlaylistItem, last_updated: i64) -> String { +pub fn create_vod_info_from_item(target: &ConfigTarget, user: &ProxyUserCredentials, pli: &XtreamPlaylistItem) -> String { let category_id = pli.category_id; let stream_id = if user.proxy.is_redirect(pli.item_type) || target.is_force_redirect(pli.item_type) { pli.provider_id } else { pli.virtual_id }; + let added = pli.additional_properties.as_ref().and_then(StreamProperties::get_last_modified).unwrap_or(0); let name = &pli.name; let extension = pli .get_container_extension() .as_deref() .filter(|ce| !ce.is_empty()) .or_else(|| extract_extension_from_url(&pli.url)) - .map_or_else(String::new, std::string::ToString::to_string); + .map_or_else(String::new, ToString::to_string); - let added = last_updated / 1000; format!(r#"{{ "info": {{}}, "movie_data": {{ diff --git a/shared/src/model/playlist.rs b/shared/src/model/playlist.rs index dd1b034c4..620914dcb 100644 --- a/shared/src/model/playlist.rs +++ b/shared/src/model/playlist.rs @@ -874,7 +874,6 @@ impl XtreamPlaylistItem { ("stream_icon".to_string(), Value::String(stream_icon)), ("stream_type".to_string(), Value::String(live.stream_type.as_ref().map_or_else(|| "live".to_string(), Clone::clone))), ("epg_channel_id".to_string(), InfoDocUtils::build_string(self.epg_channel_id.as_deref())), - ("added".to_string(), InfoDocUtils::empty_string()), ("is_adult".to_string(), Value::Number(serde_json::Number::from(live.is_adult))), ("category_id".to_string(), Value::String(self.category_id.to_string())), ("category_ids".to_string(), Value::Array(Vec::from([Value::Number(serde_json::Number::from(self.category_id))]))), @@ -1143,83 +1142,6 @@ pub struct PlaylistItem { generate_field_accessor_impl_for_xtream_playlist_item!(name, logo, logo_small, group, title, parent_code, rec, url;); impl PlaylistItem { - pub fn to_m3u(&self) -> M3uPlaylistItem { - let header = &self.header; - M3uPlaylistItem { - virtual_id: header.virtual_id, - provider_id: header.id.to_string(), - name: if header.item_type == PlaylistItemType::Series { &header.title } else { &header.name }.to_string(), - chno: header.chno, - logo: header.logo.to_string(), - logo_small: header.logo_small.to_string(), - group: header.group.to_string(), - title: header.title.to_string(), - parent_code: header.parent_code.to_string(), - audio_track: header.audio_track.to_string(), - time_shift: header.time_shift.to_string(), - rec: header.rec.to_string(), - url: header.url.to_string(), - epg_channel_id: header.epg_channel_id.clone(), - input_name: header.input_name.to_string(), - item_type: header.item_type, - t_stream_url: header.url.to_string(), - t_resource_url: None, - } - } - - pub fn to_xtream(&self) -> XtreamPlaylistItem { - let header = &self.header; - let provider_id = header.id.parse::().unwrap_or_default(); - let additional_properties = Self::get_additional_properties(header); - - XtreamPlaylistItem { - virtual_id: header.virtual_id, - provider_id, - name: if header.item_type == PlaylistItemType::Series { &header.title } else { &header.name }.to_string(), - logo: header.logo.to_string(), - logo_small: header.logo_small.to_string(), - group: header.group.to_string(), - title: header.title.to_string(), - parent_code: header.parent_code.to_string(), - rec: header.rec.to_string(), - url: header.url.to_string(), - epg_channel_id: header.epg_channel_id.clone(), - xtream_cluster: header.xtream_cluster, - additional_properties, - item_type: header.item_type, - category_id: header.category_id, - input_name: header.input_name.to_string(), - channel_no: header.chno, - } - } - - pub fn to_common(&self) -> CommonPlaylistItem { - let header = &self.header; - - let additional_properties = Self::get_additional_properties(header); - - CommonPlaylistItem { - virtual_id: header.virtual_id, - provider_id: header.id.clone(), - name: if header.item_type == PlaylistItemType::Series { &header.title } else { &header.name }.clone(), - logo: header.logo.clone(), - logo_small: header.logo_small.clone(), - group: header.group.clone(), - title: header.title.clone(), - parent_code: header.parent_code.clone(), - audio_track: header.audio_track.clone(), - time_shift: header.time_shift.clone(), - rec: header.rec.clone(), - url: header.url.clone(), - epg_channel_id: header.epg_channel_id.clone(), - xtream_cluster: Some(header.xtream_cluster), - additional_properties, - item_type: header.item_type, - category_id: Some(header.category_id), - input_name: header.input_name.clone(), - chno: header.chno, - } - } fn get_additional_properties(header: &PlaylistItemHeader) -> Option { match &header.additional_properties { @@ -1301,6 +1223,123 @@ impl PlaylistItem { } } +impl From<&PlaylistItem> for XtreamPlaylistItem { + fn from(item: &PlaylistItem) -> Self { + let header = &item.header; + let provider_id = header.id.parse::().unwrap_or_default(); + let additional_properties = PlaylistItem::get_additional_properties(header); + + XtreamPlaylistItem { + virtual_id: header.virtual_id, + provider_id, + name: if header.item_type == PlaylistItemType::Series { &header.title } else { &header.name }.to_string(), + logo: header.logo.to_string(), + logo_small: header.logo_small.to_string(), + group: header.group.to_string(), + title: header.title.to_string(), + parent_code: header.parent_code.to_string(), + rec: header.rec.to_string(), + url: header.url.to_string(), + epg_channel_id: header.epg_channel_id.clone(), + xtream_cluster: header.xtream_cluster, + additional_properties, + item_type: header.item_type, + category_id: header.category_id, + input_name: header.input_name.to_string(), + channel_no: header.chno, + } + } +} + +impl From<&PlaylistItem> for M3uPlaylistItem { + fn from(item: &PlaylistItem) -> Self { + let header = &item.header; + M3uPlaylistItem { + virtual_id: header.virtual_id, + provider_id: header.id.to_string(), + name: if header.item_type == PlaylistItemType::Series { &header.title } else { &header.name }.to_string(), + chno: header.chno, + logo: header.logo.to_string(), + logo_small: header.logo_small.to_string(), + group: header.group.to_string(), + title: header.title.to_string(), + parent_code: header.parent_code.to_string(), + audio_track: header.audio_track.to_string(), + time_shift: header.time_shift.to_string(), + rec: header.rec.to_string(), + url: header.url.to_string(), + epg_channel_id: header.epg_channel_id.clone(), + input_name: header.input_name.to_string(), + item_type: header.item_type, + t_stream_url: header.url.to_string(), + t_resource_url: None, + } + } +} + +impl From<&PlaylistItem> for CommonPlaylistItem { + fn from(item: &PlaylistItem) -> Self { + let header = &item.header; + + let additional_properties = PlaylistItem::get_additional_properties(header); + + CommonPlaylistItem { + virtual_id: header.virtual_id, + provider_id: header.id.clone(), + name: if header.item_type == PlaylistItemType::Series { &header.title } else { &header.name }.clone(), + logo: header.logo.clone(), + logo_small: header.logo_small.clone(), + group: header.group.clone(), + title: header.title.clone(), + parent_code: header.parent_code.clone(), + audio_track: header.audio_track.clone(), + time_shift: header.time_shift.clone(), + rec: header.rec.clone(), + url: header.url.clone(), + epg_channel_id: header.epg_channel_id.clone(), + xtream_cluster: Some(header.xtream_cluster), + additional_properties, + item_type: header.item_type, + category_id: Some(header.category_id), + input_name: header.input_name.clone(), + chno: header.chno, + } + } +} + +impl From<&XtreamPlaylistItem> for PlaylistItem { + fn from(item: &XtreamPlaylistItem) -> Self { + let header = PlaylistItemHeader { + uuid: item.get_uuid(), + virtual_id: item.virtual_id, + id: item.provider_id.to_string(), // u32 → String + name: item.name.clone(), + title: item.title.clone(), + logo: item.logo.clone(), + logo_small: item.logo_small.clone(), + group: item.group.clone(), + parent_code: item.parent_code.clone(), + rec: item.rec.clone(), + url: item.url.clone(), + epg_channel_id: item.epg_channel_id.clone(), + xtream_cluster: item.xtream_cluster, + item_type: item.item_type, + category_id: item.category_id, + input_name: item.input_name.clone(), + chno: item.channel_no, + audio_track: String::new(), + time_shift: String::new(), + additional_properties: item.additional_properties.clone(), + }; + + PlaylistItem { + header + } + } +} + + + impl PlaylistEntry for PlaylistItem { #[inline] fn get_virtual_id(&self) -> VirtualId { diff --git a/shared/src/model/xtream_const.rs b/shared/src/model/xtream_const.rs index e92280ab2..ec114f1ea 100644 --- a/shared/src/model/xtream_const.rs +++ b/shared/src/model/xtream_const.rs @@ -23,28 +23,7 @@ pub const XC_TAG_STREAM_ID: &str = "stream_id"; pub const XC_TAG_EPG_LISTINGS: &str = "epg_listings"; pub const XC_PROP_BACKDROP_PATH: &str = "backdrop_path"; pub const XC_PROP_COVER: &str = "cover"; -pub const XC_TAG_CATEGORY_IDS: &str = "category_ids"; pub const XC_TAG_CATEGORY_NAME: &str = "category_name"; -pub const XC_TAG_DIRECT_SOURCE: &str = "direct_source"; -pub const XC_TAG_PARENT_ID: &str = "parent_id"; -pub const XC_TAG_MOVIE_DATA: &str = "movie_data"; -pub const XC_TAG_INFO_DATA: &str = "info"; -pub const XC_TAG_SEASONS_DATA: &str = "seasons"; -pub const XC_TAG_EPISODES: &str = "episodes"; -pub const XC_TAG_VOD_INFO_INFO: &str = "info"; -pub const XC_TAG_VOD_INFO_MOVIE_DATA: &str = "movie_data"; -pub const XC_TAG_VOD_INFO_TMDB_ID: &str = "tmdb_id"; -pub const XC_TAG_VOD_INFO_TMDB: &str = "tmdb"; -pub const XC_TAG_VOD_INFO_STREAM_ID: &str = "stream_id"; -pub const XC_TAG_VOD_INFO_ADDED: &str = "added"; -pub const XC_TAG_VOD_INFO_RELEASEDATE: &str = "release_date"; - -pub const XC_FILE_SERIES_INFO: &str = "xtream_series_info"; -pub const XC_FILE_VOD_INFO: &str = "xtream_vod_info"; -pub const XC_FILE_SERIES_EPISODE_RECORD: &str = "series_episode_record"; -pub const XC_TAG_SERIES_INFO_LAST_MODIFIED: &str = "last_modified"; - -pub const LIVE_STREAM_FIELDS: [&str;0] = []; pub const VIDEO_STREAM_FIELDS: [&str;14] = [ "release_date", "cast",