diff --git a/CHANGELOG.md b/CHANGELOG.md index 645420e19..f061458d0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -125,6 +125,9 @@ - **Panel API Integration**: Optional integration to renew expired input accounts or provision new accounts to ensure a minimum valid input accounts. - **Playlist Caching**: Added `cache_duration` to inputs, allowing configurable provider playlist cache times during subsequent updates (e.g., `60s`, `5m` `12h`, `1d`). +- **Staged Cluster Source Routing**: Added per-cluster staged routing for Xtream inputs. + You can now decide cluster-wise whether `live` / `vod` / `series` is loaded from staged input, main input, or skipped. + Skip flags (`xtream_skip_live|vod|series`) remain highest priority and always force skip. - **Database Viewer**: New CLI flags `--dbx` and `--dbm` to inspect internal database content. - **Home Directory Override**: Added `--home` (`-H`) CLI argument to set the base directory for config, data, backup, and downloads. - **Added `disk_based_processing`**: (boolean, default `false`) to `config.yml`. When enabled, input playlists are processed from disk instead of @@ -173,7 +176,14 @@ active URL of the specified provider. - Added `resolve_tmdb`: Triggers TMDB lookup if ID is missing. - Added `probe_stream`: Triggers ffprobe if technical info is missing. - Added `probe_delay`: Delay between probe tasks (default `50` seconds). - - Added `staged.enabled`: Disables/enables the staged input + - Added `staged.enabled`: Disables/enables the staged input. + - Added `staged.live_source`: Selects source for Live cluster (`staged` | `input` | `skip`). + - Added `staged.vod_source`: Selects source for VOD cluster (`staged` | `input` | `skip`). + - Added `staged.series_source`: Selects source for Series cluster (`staged` | `input` | `skip`). + - Added staged validation rules: + - Cluster source rules apply only when `staged.enabled=true`. + - For Xtream main inputs with staged enabled, at least one cluster source must be `staged`. + - For staged type `m3u`, `vod_source=staged` and `series_source=staged` are rejected. - **source.yml (target output)**: - Added `probe_live`: Enables background probing for Live TV streams (default disabled). - Added `probe_live_interval_hours`: Sets the frequency for re-probing Live TV streams. diff --git a/README.md b/README.md index 8dc58da7e..c3d870557 100644 --- a/README.md +++ b/README.md @@ -1409,7 +1409,51 @@ the list. All streaming, proxying, and metadata still come from the provider’s - `headers` is optional - `method` can be `GET` or `POST` - `username` only mandatory for type `xtream` -- `pasword`only mandatory for type `xtream` +- `password` only mandatory for type `xtream` +- `live_source` (optional): `staged` | `input` | `skip` +- `vod_source` (optional): `staged` | `input` | `skip` +- `series_source` (optional): `staged` | `input` | `skip` + +Staged cluster source behavior: + +- Cluster source rules are applied only when `staged.enabled: true`. +- `input.options.xtream_skip_live|vod|series` has highest priority. If one of those is `true`, the cluster is skipped regardless of staged source selection. +- For Xtream main inputs (`type: xtream`) with staged enabled: + - At least one of `live_source` / `vod_source` / `series_source` must resolve to `staged`. + - If all effective values are `input`/`skip`, config validation fails. +- For staged input `type: m3u`: + - `live_source: staged` is valid. + - `vod_source: staged` and `series_source: staged` are invalid (M3U staged source cannot provide Xtream VOD/Series clusters). +- If main input type is `m3u`, staged cluster source fields are ignored. + +Defaults: + +- Staged type `xtream`: `live_source=staged`, `vod_source=staged`, `series_source=staged` +- Staged type `m3u`: `live_source=staged`, `vod_source=input`, `series_source=input` + +Example (Xtream main + staged Xtream per cluster): + +```yaml +inputs: + - name: provider_main + type: xtream + url: http://provider-a.example:8080 + username: main_user + password: main_pass + options: + xtream_skip_live: false + xtream_skip_vod: false + xtream_skip_series: false + staged: + enabled: true + type: xtream + url: http://provider-b.example:8080 + username: staged_user + password: staged_pass + live_source: staged + vod_source: input + series_source: skip +``` `persist` should be different for `m3u` and `xtream` types. For `m3u` use full filename like `./playlist_{}.m3u`. For `xtream` use a prefix like `./playlist_` diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index 9c0db035c..cdca35020 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -515,8 +515,8 @@ async fn epg_api_resource( pub fn xmltv_api_register() -> axum::Router> { axum::Router::new() .route("/xmltv.php", axum::routing::get(xmltv_api)) - .route("/update/epg.php", axum::routing::get(xmltv_api)) .route("/epg", axum::routing::get(xmltv_api)) + .route("/update/epg.php", axum::routing::get(xmltv_api)) .route( &format!("/{}/{{username}}/{{password}}/{{resource}}", storage_const::EPG_RESOURCE_PATH), axum::routing::get(epg_api_resource), diff --git a/backend/src/model/config/base.rs b/backend/src/model/config/base.rs index 46eda40c7..3e425f2dd 100644 --- a/backend/src/model/config/base.rs +++ b/backend/src/model/config/base.rs @@ -226,12 +226,19 @@ impl Config { } fn prepare_api_web_root(&mut self, home_path: &str) { - if self.api.web_root.is_empty() { + let web_root = self.api.web_root.trim(); + if web_root.is_empty() { self.api.web_root = utils::get_default_web_root_path_for_home(Path::new(home_path)) .display() .to_string(); } else { - self.api.web_root = utils::make_absolute_path(&self.api.web_root, &self.storage_dir); + let web_root_path = PathBuf::from(web_root); + let resolved = if web_root_path.is_absolute() { + web_root_path + } else { + PathBuf::from(home_path).join(web_root_path).clean() + }; + self.api.web_root = resolved.to_string_lossy().to_string(); } } @@ -308,3 +315,60 @@ impl From<&ConfigDto> for Config { } } } + +#[cfg(test)] +mod tests { + use super::Config; + use tempfile::tempdir; + + #[test] + fn prepare_resolves_relative_web_root_from_home_path() { + let temp_dir = tempdir().expect("failed to create temp directory"); + let home_path = temp_dir.path().join("home"); + let config_path = temp_dir.path().join("config"); + + let mut config = Config { + storage_dir: "data".to_string(), + ..Default::default() + }; + config.api.web_root = "./web".to_string(); + + config + .prepare( + config_path.to_string_lossy().as_ref(), + home_path.to_string_lossy().as_ref(), + ) + .expect("prepare should succeed"); + + assert_eq!( + config.api.web_root, + home_path.join("web").to_string_lossy() + ); + } + + #[test] + fn prepare_keeps_absolute_web_root_unchanged() { + let temp_dir = tempdir().expect("failed to create temp directory"); + let home_path = temp_dir.path().join("home"); + let config_path = temp_dir.path().join("config"); + let absolute_web_root = temp_dir.path().join("custom_web_root"); + + let mut config = Config { + storage_dir: "data".to_string(), + ..Default::default() + }; + config.api.web_root = absolute_web_root.to_string_lossy().to_string(); + + config + .prepare( + config_path.to_string_lossy().as_ref(), + home_path.to_string_lossy().as_ref(), + ) + .expect("prepare should succeed"); + + assert_eq!( + config.api.web_root, + absolute_web_root.to_string_lossy() + ); + } +} diff --git a/backend/src/model/config/input.rs b/backend/src/model/config/input.rs index 76a9eb736..1ca228b78 100644 --- a/backend/src/model/config/input.rs +++ b/backend/src/model/config/input.rs @@ -4,7 +4,7 @@ use chrono::Utc; use log::warn; use shared::{apply_flags, create_bitset}; use shared::error::TuliproxError; -use shared::model::{ConfigInputAliasDto, ConfigInputDto, ConfigInputOptionsDto, InputFetchMethod, InputType, StagedInputDto}; +use shared::model::{ClusterSource, ConfigInputAliasDto, ConfigInputDto, ConfigInputOptionsDto, InputFetchMethod, InputType, StagedInputDto, XtreamCluster}; use shared::utils::{get_credentials_from_url, parse_provider_scheme_url_parts, sanitize_sensitive_info, Internable, PROVIDER_SCHEME_PREFIX}; use shared::{check_input_connections, info_err_res, write_if_some}; use shared::{check_input_credentials, concat_string, info_err}; @@ -155,7 +155,7 @@ impl InputUserInfo { } } -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct StagedInput { pub enabled: bool, pub name: Arc, @@ -167,11 +167,55 @@ pub struct StagedInput { pub headers: HashMap, /// Provider configuration for failover support when using `provider://` scheme. pub provider_config: Option>, + pub live_source: ClusterSource, + pub vod_source: ClusterSource, + pub series_source: ClusterSource, + pub cluster_sources_configured: bool, +} + +impl Default for StagedInput { + fn default() -> Self { + Self { + enabled: false, + name: Arc::default(), + url: String::new(), + username: None, + password: None, + method: InputFetchMethod::default(), + input_type: InputType::default(), + headers: HashMap::default(), + provider_config: None, + live_source: ClusterSource::Staged, + vod_source: ClusterSource::Input, + series_source: ClusterSource::Input, + cluster_sources_configured: false, + } + } +} + +impl StagedInput { + /// Returns the `ClusterSource` for the given `XtreamCluster`. + pub fn get_cluster_source(&self, cluster: XtreamCluster) -> ClusterSource { + match cluster { + XtreamCluster::Live => self.live_source, + XtreamCluster::Video => self.vod_source, + XtreamCluster::Series => self.series_source, + } + } } macros::from_impl!(StagedInput); impl From<&StagedInputDto> for StagedInput { fn from(dto: &StagedInputDto) -> Self { + let resolve = |opt: Option, default: ClusterSource| -> ClusterSource { + opt.unwrap_or(default) + }; + let (live_default, vod_default, series_default) = if dto.input_type.is_m3u() { + (ClusterSource::Staged, ClusterSource::Input, ClusterSource::Input) + } else { + (ClusterSource::Staged, ClusterSource::Staged, ClusterSource::Staged) + }; + Self { enabled: dto.enabled, name: dto.name.clone(), @@ -182,6 +226,10 @@ impl From<&StagedInputDto> for StagedInput { method: dto.method, headers: dto.headers.clone(), provider_config: None, // Resolved later in ConfigInput::prepare() + live_source: resolve(dto.live_source, live_default), + vod_source: resolve(dto.vod_source, vod_default), + series_source: resolve(dto.series_source, series_default), + cluster_sources_configured: dto.live_source.is_some() || dto.vod_source.is_some() || dto.series_source.is_some(), } } } @@ -242,6 +290,126 @@ pub struct ConfigInput { } impl ConfigInput { + fn resolve_provider_config(url: &str, provider_configs: &[Arc]) -> Result, TuliproxError> { + let (host, _path) = parse_provider_scheme_url_parts(url).map_err(|err| { + info_err!( + "Malformed provider URL {}: {}", + sanitize_sensitive_info(url), + sanitize_sensitive_info(err.to_string().as_str()) + ) + })?; + + provider_configs + .iter() + .find(|p| p.name.as_ref() == host) + .cloned() + .ok_or_else(|| info_err!("Failed to resolve provider config for {}", sanitize_sensitive_info(url))) + } + + fn prepare_staged_input( + &mut self, + provider_configs: &[Arc], + used_provider_configs: &mut Vec>, + skip_live: bool, + skip_vod: bool, + skip_series: bool, + ) -> Result<(), TuliproxError> { + if let Some(staged_input) = &mut self.staged { + if staged_input.enabled { + if staged_input.url.starts_with(PROVIDER_SCHEME_PREFIX) { + let provider_cfg = Self::resolve_provider_config(&staged_input.url, provider_configs)?; + staged_input.provider_config = Some(provider_cfg.clone()); + used_provider_configs.push(provider_cfg); + } + + check_input_credentials!(staged_input, staged_input.input_type, false, true); + if !matches!(staged_input.input_type, InputType::M3u | InputType::Xtream) { + return info_err_res!( + "Staged input can only be from type m3u or xtream (input: {}, staged: {})", + self.name, + staged_input.name + ); + } + + if self.input_type.is_xtream() { + let live_uses_staged = matches!(staged_input.live_source, ClusterSource::Staged) && !skip_live; + let vod_uses_staged = matches!(staged_input.vod_source, ClusterSource::Staged) && !skip_vod; + let series_uses_staged = matches!(staged_input.series_source, ClusterSource::Staged) && !skip_series; + + if !live_uses_staged && !vod_uses_staged && !series_uses_staged { + return info_err_res!( + "Staged input is enabled but no cluster source uses 'staged'; set at least one of live_source/vod_source/series_source to 'staged' (input: {}, staged: {})", + self.name, + staged_input.name + ); + } + + if staged_input.input_type.is_m3u() && (vod_uses_staged || series_uses_staged) { + return info_err_res!( + "Staged M3U input cannot provide VOD or Series clusters; use 'input' or 'skip' (input: {}, staged: {})", + self.name, + staged_input.name + ); + } + } + + if self.input_type.is_m3u() { + if staged_input.cluster_sources_configured { + warn!( + "Input '{}': cluster source fields (live_source/vod_source/series_source) are ignored for M3U main inputs", + self.name + ); + } + staged_input.live_source = ClusterSource::Staged; + staged_input.vod_source = ClusterSource::Staged; + staged_input.series_source = ClusterSource::Staged; + staged_input.cluster_sources_configured = false; + } + } + } + + Ok(()) + } + + fn prepare_aliases( + &mut self, + provider_configs: &[Arc], + used_provider_configs: &mut Vec>, + ) -> Result<(), TuliproxError> { + if let Some(aliases) = &mut self.aliases { + for alias in aliases { + if is_input_expired(alias.exp_date) { + warn!( + "Account {} expired for provider: {}", + alias.username.as_ref().map_or("?", |s| s.as_str()), + alias.name + ); + alias.enabled = false; + } + + if alias.url.starts_with(PROVIDER_SCHEME_PREFIX) { + let provider_cfg = Self::resolve_provider_config(&alias.url, provider_configs)?; + if !used_provider_configs.iter().any(|p| p.name == provider_cfg.name) { + used_provider_configs.push(provider_cfg); + } + } + } + } + + Ok(()) + } + + fn apply_expiration(&mut self) { + if is_input_expired(self.exp_date) { + warn!( + "Account {} expired for provider: {}", + self.username.as_ref().map_or("?", |s| s.as_str()), + self.name + ); + self.enabled = false; + } + } + #[inline] pub fn get_download_input_type(&self) -> InputType { self.staged @@ -297,67 +465,30 @@ impl ConfigInput { self.options = Some(ConfigInputOptions::defaults().clone()); } + // For batch definitions, validate root URL/credentials before alias promotion in prepare_batch(). + if self.enabled && matches!(self.input_type, InputType::M3uBatch | InputType::XtreamBatch) { + check_input_credentials!(self, self.input_type, false, false); + check_input_connections!(self, self.input_type, false); + } + let mut used_provider_configs: Vec> = vec![]; let batch_file_path = self.prepare_batch(); self.name = self.name.trim().intern(); - let resolve_provider_config = |url: &str| -> Result, TuliproxError> { - let (host, _path) = parse_provider_scheme_url_parts(url).map_err(|err| { - info_err!( - "Malformed provider URL {}: {}", - sanitize_sensitive_info(url), - sanitize_sensitive_info(err.to_string().as_str()) - ) - })?; - - provider_configs - .iter() - .find(|p| p.name.as_ref() == host) - .cloned() - .ok_or_else(|| info_err!("Failed to resolve provider config for {}", sanitize_sensitive_info(url))) - }; - if self.url.starts_with(PROVIDER_SCHEME_PREFIX) { - let provider_cfg = resolve_provider_config(&self.url)?; + let provider_cfg = Self::resolve_provider_config(&self.url, provider_configs)?; used_provider_configs.push(provider_cfg); } if self.enabled { check_input_credentials!(self, self.input_type, false, false); check_input_connections!(self, self.input_type, false); - if let Some(staged_input) = &mut self.staged { - if staged_input.url.starts_with(PROVIDER_SCHEME_PREFIX) { - let provider_cfg = resolve_provider_config(&staged_input.url)?; - staged_input.provider_config = Some(provider_cfg.clone()); - used_provider_configs.push(provider_cfg); - } - - check_input_credentials!(staged_input, staged_input.input_type, false, true); - if !matches!(staged_input.input_type, InputType::M3u | InputType::Xtream) { - return info_err_res!("Staged input can only be from type m3u or xtream"); - } - } - - if is_input_expired(self.exp_date) { - warn!("Account {} expired for provider: {}", self.username.as_ref().map_or("?", |s| s.as_str()), self.name); - self.enabled = false; - } - - if let Some(aliases) = &mut self.aliases { - for alias in aliases { - if is_input_expired(alias.exp_date) { - warn!("Account {} expired for provider: {}", alias.username.as_ref().map_or("?", |s| s.as_str()), alias.name); - alias.enabled = false; - } - - if alias.url.starts_with(PROVIDER_SCHEME_PREFIX) { - let provider_cfg = resolve_provider_config(&alias.url)?; - if !used_provider_configs.iter().any(|p| p.name == provider_cfg.name) { - used_provider_configs.push(provider_cfg); - } - } - } - } + let skip_live = self.has_flag(ConfigInputFlags::XtreamSkipLive); + let skip_vod = self.has_flag(ConfigInputFlags::XtreamSkipVod); + let skip_series = self.has_flag(ConfigInputFlags::XtreamSkipSeries); + self.prepare_staged_input(provider_configs, &mut used_provider_configs, skip_live, skip_vod, skip_series)?; + self.apply_expiration(); + self.prepare_aliases(provider_configs, &mut used_provider_configs)?; if !used_provider_configs.is_empty() { self.provider_configs = Some(used_provider_configs); @@ -706,6 +837,7 @@ mod tests { url: "http://example.com/playlist.m3u".to_string(), enabled: true, staged: Some(StagedInput { + enabled: true, name: "staged".into(), input_type: InputType::M3u, url: "provider:///bad".to_string(), @@ -718,6 +850,28 @@ mod tests { assert!(err.to_string().contains("Malformed provider URL")); } + #[test] + fn test_prepare_ignores_disabled_staged_provider_url() { + let mut input = ConfigInput { + name: "test_input".into(), + input_type: InputType::M3u, + url: "http://example.com/playlist.m3u".to_string(), + enabled: true, + staged: Some(StagedInput { + enabled: false, + name: "staged".into(), + input_type: InputType::M3u, + url: "provider:///bad".to_string(), + ..Default::default() + }), + ..Default::default() + }; + + input + .prepare(&[]) + .expect("disabled staged input should not enforce provider URL validation"); + } + #[test] fn test_prepare_fails_on_malformed_provider_url_in_alias() { let mut input = ConfigInput { @@ -772,4 +926,255 @@ mod tests { assert_eq!(input.get_download_input_type(), InputType::Xtream); } + + #[test] + fn test_staged_from_dto_defaults_m3u() { + let dto = StagedInputDto { + input_type: InputType::M3u, + ..StagedInputDto::default() + }; + let staged = StagedInput::from(&dto); + assert_eq!(staged.live_source, ClusterSource::Staged); + assert_eq!(staged.vod_source, ClusterSource::Input); + assert_eq!(staged.series_source, ClusterSource::Input); + } + + #[test] + fn test_staged_from_dto_defaults_xtream() { + let dto = StagedInputDto { + input_type: InputType::Xtream, + ..StagedInputDto::default() + }; + let staged = StagedInput::from(&dto); + assert_eq!(staged.live_source, ClusterSource::Staged); + assert_eq!(staged.vod_source, ClusterSource::Staged); + assert_eq!(staged.series_source, ClusterSource::Staged); + } + + #[test] + fn test_staged_from_dto_explicit_overrides() { + let dto = StagedInputDto { + input_type: InputType::Xtream, + live_source: Some(ClusterSource::Input), + vod_source: Some(ClusterSource::Skip), + series_source: Some(ClusterSource::Staged), + ..StagedInputDto::default() + }; + let staged = StagedInput::from(&dto); + assert_eq!(staged.live_source, ClusterSource::Input); + assert_eq!(staged.vod_source, ClusterSource::Skip); + assert_eq!(staged.series_source, ClusterSource::Staged); + } + + #[test] + fn test_get_cluster_source() { + let staged = StagedInput { + live_source: ClusterSource::Staged, + vod_source: ClusterSource::Input, + series_source: ClusterSource::Skip, + ..Default::default() + }; + assert_eq!(staged.get_cluster_source(XtreamCluster::Live), ClusterSource::Staged); + assert_eq!(staged.get_cluster_source(XtreamCluster::Video), ClusterSource::Input); + assert_eq!(staged.get_cluster_source(XtreamCluster::Series), ClusterSource::Skip); + } + + #[test] + fn test_staged_from_dto_tracks_cluster_sources_configured_flag() { + let dto = StagedInputDto { + input_type: InputType::Xtream, + live_source: Some(ClusterSource::Input), + ..StagedInputDto::default() + }; + let staged = StagedInput::from(&dto); + assert!(staged.cluster_sources_configured); + } + + #[test] + fn test_prepare_m3u_main_input_ignores_cluster_sources() { + let mut input = ConfigInput { + name: "m3u_main".into(), + input_type: InputType::M3u, + url: "http://main.example/playlist.m3u".to_string(), + enabled: true, + staged: Some(StagedInput { + enabled: true, + input_type: InputType::Xtream, + url: "http://staged.example".to_string(), + username: Some("u".to_string()), + password: Some("p".to_string()), + live_source: ClusterSource::Input, + vod_source: ClusterSource::Skip, + series_source: ClusterSource::Input, + cluster_sources_configured: true, + ..StagedInput::default() + }), + ..Default::default() + }; + + input.prepare(&[]).expect("prepare should succeed"); + + let staged = input.staged.expect("staged should exist"); + assert_eq!(staged.live_source, ClusterSource::Staged); + assert_eq!(staged.vod_source, ClusterSource::Staged); + assert_eq!(staged.series_source, ClusterSource::Staged); + assert!(!staged.cluster_sources_configured); + } + + #[test] + fn test_prepare_enabled_staged_requires_at_least_one_staged_cluster_source() { + let mut input = ConfigInput { + name: "xtream_main".into(), + input_type: InputType::Xtream, + url: "http://main.example".to_string(), + username: Some("main_user".to_string()), + password: Some("main_pass".to_string()), + enabled: true, + staged: Some(StagedInput { + enabled: true, + input_type: InputType::Xtream, + url: "http://staged.example".to_string(), + username: Some("staged_user".to_string()), + password: Some("staged_pass".to_string()), + live_source: ClusterSource::Input, + vod_source: ClusterSource::Skip, + series_source: ClusterSource::Input, + ..StagedInput::default() + }), + ..Default::default() + }; + + let err = input + .prepare(&[]) + .expect_err("expected validation error for staged source selection"); + assert!(err.to_string().contains("no cluster source uses 'staged'")); + } + + #[test] + fn test_prepare_staged_m3u_rejects_vod_series_staged() { + let mut input = ConfigInput { + name: "xtream_main".into(), + input_type: InputType::Xtream, + url: "http://main.example".to_string(), + username: Some("main_user".to_string()), + password: Some("main_pass".to_string()), + enabled: true, + staged: Some(StagedInput { + enabled: true, + input_type: InputType::M3u, + url: "http://staged.example/playlist.m3u".to_string(), + vod_source: ClusterSource::Staged, + ..StagedInput::default() + }), + ..Default::default() + }; + + let err = input + .prepare(&[]) + .expect_err("expected staged M3U validation error for vod_source=staged"); + assert!(err.to_string().contains("Staged M3U input cannot provide VOD or Series")); + } + + #[test] + fn test_prepare_staged_m3u_vod_staged_allowed_when_vod_skipped() { + let mut input = ConfigInput { + name: "xtream_main".into(), + input_type: InputType::Xtream, + url: "http://main.example".to_string(), + username: Some("main_user".to_string()), + password: Some("main_pass".to_string()), + enabled: true, + options: Some(ConfigInputOptions::from(&ConfigInputOptionsDto { + xtream_skip_vod: true, + ..ConfigInputOptionsDto::default() + })), + staged: Some(StagedInput { + enabled: true, + input_type: InputType::M3u, + url: "http://staged.example/playlist.m3u".to_string(), + vod_source: ClusterSource::Staged, + ..StagedInput::default() + }), + ..Default::default() + }; + + input + .prepare(&[]) + .expect("staged M3U vod_source=staged is valid when VOD is skipped"); + } + + #[test] + fn test_prepare_xtream_batch_requires_root_url_even_with_aliases() { + let mut input = ConfigInput { + name: "xtream_batch_missing_root_url".into(), + input_type: InputType::XtreamBatch, + url: "".to_string(), + enabled: true, + aliases: Some(vec![ConfigInputAlias { + id: 1, + name: "alias".into(), + url: "http://alias.example".to_string(), + username: Some("alias_user".to_string()), + password: Some("alias_pass".to_string()), + priority: 0, + max_connections: 0, + exp_date: None, + enabled: true, + }]), + ..Default::default() + }; + + let err = input + .prepare(&[]) + .expect_err("prepare must require root URL even when aliases are attached directly"); + assert!(err.to_string().contains("url for input is mandatory"), "Error: {err}"); + assert!(err.to_string().contains("xtream_batch_missing_root_url"), "Error: {err}"); + } + + #[test] + fn test_prepare_xtream_batch_requires_root_credentials_for_non_batch_url() { + let mut input = ConfigInput { + name: "xtream_batch_missing_root_creds".into(), + input_type: InputType::XtreamBatch, + url: "http://root.example".to_string(), + enabled: true, + aliases: Some(vec![ConfigInputAlias { + id: 1, + name: "alias".into(), + url: "http://alias.example".to_string(), + username: Some("alias_user".to_string()), + password: Some("alias_pass".to_string()), + priority: 0, + max_connections: 0, + exp_date: None, + enabled: true, + }]), + ..Default::default() + }; + + let err = input + .prepare(&[]) + .expect_err("prepare must require root credentials for non-batch xtream-batch URL"); + assert!(err.to_string().contains("xtream-batch without batch:// URL"), "Error: {err}"); + assert!(err.to_string().contains("xtream_batch_missing_root_creds"), "Error: {err}"); + } + + #[test] + fn test_prepare_xtream_batch_rejects_root_credentials_for_batch_url() { + let mut input = ConfigInput { + name: "xtream_batch_root_creds_not_allowed".into(), + input_type: InputType::XtreamBatch, + url: "batch:///tmp/aliases.csv".to_string(), + username: Some("root_user".to_string()), + password: Some("root_pass".to_string()), + enabled: true, + ..Default::default() + }; + + let err = input + .prepare(&[]) + .expect_err("prepare must reject root credentials for batch:// xtream-batch URL"); + assert!(err.to_string().contains("with batch:// URL should not define username or password"), "Error: {err}"); + assert!(err.to_string().contains("xtream_batch_root_creds_not_allowed"), "Error: {err}"); + } } diff --git a/backend/src/processing/parser/xtream.rs b/backend/src/processing/parser/xtream.rs index ef994d4a7..d33f02f99 100644 --- a/backend/src/processing/parser/xtream.rs +++ b/backend/src/processing/parser/xtream.rs @@ -74,8 +74,7 @@ pub fn parse_xtream_series_info(parent_uuid: &UUIDType, series_info: &SeriesStre ); if let Some(episodes) = series_info.details.as_ref().and_then(|d| d.episodes.as_ref()) { - let base_source_ordinal = if parent_source_ordinal == 0 { 1 } else { parent_source_ordinal }; - let result: Vec = episodes.iter().enumerate().map(|(index, episode)| { + let result: Vec = episodes.iter().map(|episode| { let episode_id = episode.id.to_string(); let episode_url = create_xtream_series_episode_url(url, username, password, episode); @@ -102,7 +101,8 @@ pub fn parse_xtream_series_info(parent_uuid: &UUIDType, series_info: &SeriesStre additional_properties: Some(StreamProperties::Episode(Box::new(episode_info))), category_id: 0, input_name: input.name.intern(), - source_ordinal: base_source_ordinal.saturating_add(u32::try_from(index).unwrap_or(u32::MAX)), + // Keep episode ordering tied to its parent SeriesInfo to avoid cross-series ordinal overlap. + source_ordinal: parent_source_ordinal, ..Default::default() } } @@ -452,7 +452,7 @@ mod tests { } #[test] - fn test_parse_xtream_series_info_sets_episode_source_ordinals_from_parent() { + fn test_parse_xtream_series_info_episodes_inherit_parent_source_ordinal() { let input = ConfigInput { name: "input".intern(), url: "http://provider.example".to_string(), @@ -492,6 +492,45 @@ mod tests { .expect("episodes should be parsed"); assert_eq!(episodes[0].header.source_ordinal, 42); - assert_eq!(episodes[1].header.source_ordinal, 43); + assert_eq!(episodes[1].header.source_ordinal, 42); + } + + #[test] + fn test_parse_xtream_series_info_keeps_zero_parent_source_ordinal() { + let input = ConfigInput { + name: "input".intern(), + url: "http://provider.example".to_string(), + username: Some("user".to_string()), + password: Some("pass".to_string()), + ..ConfigInput::default() + }; + + let episode: SeriesStreamDetailEpisodeProperties = serde_json::from_str( + r#"{"id":101,"episode_num":1,"season":1,"title":"S01E01","container_extension":"mp4"}"#, + ) + .unwrap(); + + let series_props = SeriesStreamProperties { + details: Some(SeriesStreamDetailProperties { + year: None, + seasons: None, + episodes: Some(vec![episode]), + }), + ..SeriesStreamProperties::default() + }; + let parent_uuid = UUIDType::from_valid_uuid("parent_uuid"); + + let episodes = parse_xtream_series_info( + &parent_uuid, + &series_props, + "Series Group", + &"Series Name".intern(), + &input, + None, + 0, + ) + .expect("episodes should be parsed"); + + assert_eq!(episodes[0].header.source_ordinal, 0); } } diff --git a/backend/src/processing/processor/filtered_playlist_source.rs b/backend/src/processing/processor/filtered_playlist_source.rs new file mode 100644 index 000000000..2dcfb2b5e --- /dev/null +++ b/backend/src/processing/processor/filtered_playlist_source.rs @@ -0,0 +1,173 @@ +use crate::repository::{MemoryPlaylistSource, PlaylistSource}; +use futures::future::BoxFuture; +use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemType, UUIDType, XtreamCluster}; +use std::collections::HashSet; +use std::sync::Arc; + +pub(crate) struct FilteredPlaylistSource { + inner: Box, + skip_set: Arc>, +} + +impl FilteredPlaylistSource { + pub(crate) fn new(inner: Box, skip_set: HashSet) -> Self { + Self { + inner, + skip_set: Arc::new(skip_set), + } + } + + fn filter_group(&self, mut group: PlaylistGroup) -> Option { + if self.skip_set.contains(&group.xtream_cluster) { + return None; + } + group.channels.retain(|item| !self.skip_set.contains(&item.header.xtream_cluster)); + if group.channels.is_empty() { + None + } else { + Some(group) + } + } +} + +impl PlaylistSource for FilteredPlaylistSource { + fn is_memory(&self) -> bool { + self.inner.is_memory() + } + + fn get_channel_count(&mut self) -> usize { + let skip_set = Arc::clone(&self.skip_set); + self.inner + .items() + .filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster)) + .count() + } + + fn get_group_count(&mut self) -> usize { + let skip_set = Arc::clone(&self.skip_set); + let mut groups = HashSet::<(XtreamCluster, Arc)>::new(); + for item in self + .inner + .items() + .filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster)) + { + let pli = item.as_ref(); + groups.insert((pli.header.xtream_cluster, Arc::clone(&pli.header.group))); + } + groups.len() + } + + fn is_empty(&mut self) -> bool { + self.get_channel_count() == 0 + } + + #[allow(clippy::wrong_self_convention)] + fn into_items(&mut self) -> Box + Send + '_> { + let skip_set = Arc::clone(&self.skip_set); + Box::new( + self.inner + .into_items() + .filter(move |item| !skip_set.contains(&item.header.xtream_cluster)), + ) + } + + fn items_mut(&mut self) -> Box + Send + '_> { + let skip_set = Arc::clone(&self.skip_set); + Box::new(self.inner.items_mut().filter_map(move |item| { + if skip_set.contains(&item.header.xtream_cluster) { + None + } else { + Some(item) + } + })) + } + + fn items<'a>(&'a mut self) -> Box> + Send + 'a> { + let skip_set = Arc::clone(&self.skip_set); + Box::new( + self.inner + .items() + .filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster)), + ) + } + + fn update_playlist<'a>(&'a mut self, plg: &'a PlaylistGroup) -> BoxFuture<'a, ()> { + if self.skip_set.contains(&plg.xtream_cluster) { + return Box::pin(async move {}); + } + self.inner.update_playlist(plg) + } + + fn get_missing_vod_info_count(&mut self) -> usize { + let skip_set = Arc::clone(&self.skip_set); + self.inner + .items() + .filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster)) + .filter(|item| { + let pli = item.as_ref(); + pli.header.xtream_cluster == XtreamCluster::Video + && pli.header.item_type == PlaylistItemType::Video + && !pli.has_details() + }) + .count() + } + + fn get_missing_series_info_count(&mut self) -> usize { + let skip_set = Arc::clone(&self.skip_set); + self.inner + .items() + .filter(move |item| !skip_set.contains(&item.as_ref().header.xtream_cluster)) + .filter(|item| { + let pli = item.as_ref(); + pli.header.xtream_cluster == XtreamCluster::Series + && pli.header.item_type == PlaylistItemType::SeriesInfo + && pli.header.id.parse::().is_ok_and(|id| id > 0) + && !pli.has_details() + }) + .count() + } + + fn deduplicate(&mut self, duplicates: &mut HashSet) { + if self.inner.is_memory() { + let filtered_groups = self + .inner + .take_groups() + .into_iter() + .filter_map(|group| self.filter_group(group)) + .collect::>(); + let mut memory = MemoryPlaylistSource::new(filtered_groups).boxed(); + memory.deduplicate(duplicates); + self.inner = memory; + return; + } + + self.inner.deduplicate(duplicates); + } + + fn take_groups(&mut self) -> Vec { + self.inner + .take_groups() + .into_iter() + .filter_map(|group| self.filter_group(group)) + .collect() + } + + fn clone_box(&self) -> Box { + Box::new(Self { + inner: self.inner.clone_box(), + skip_set: Arc::clone(&self.skip_set), + }) + } + + fn release_resources(&mut self, cluster: XtreamCluster) { + self.inner.release_resources(cluster); + } + + fn obtain_resources(&mut self) -> BoxFuture<'_, ()> { + self.inner.obtain_resources() + } + + fn sort_by_provider_ordinal(&mut self) { + self.inner.sort_by_provider_ordinal(); + } +} diff --git a/backend/src/processing/processor/mod.rs b/backend/src/processing/processor/mod.rs index 344242733..c16739491 100644 --- a/backend/src/processing/processor/mod.rs +++ b/backend/src/processing/processor/mod.rs @@ -8,6 +8,7 @@ mod sort; mod trakt; mod library; mod stream_probe; +mod filtered_playlist_source; pub use self::playlist::*; pub use self::epg::*; pub use self::xtream::*; diff --git a/backend/src/processing/processor/playlist.rs b/backend/src/processing/processor/playlist.rs index 5dcd0248d..800da6d4b 100644 --- a/backend/src/processing/processor/playlist.rs +++ b/backend/src/processing/processor/playlist.rs @@ -29,6 +29,7 @@ use crate::{ debug_if_enabled, epg, log_memory_snapshot, m3u, trace_if_enabled, xtream, StepMeasure, StepMeasureCallback, }, }; +use super::filtered_playlist_source::FilteredPlaylistSource; use futures::{FutureExt, StreamExt}; use indexmap::IndexMap; use log::{debug, error, info, log_enabled, warn, Level}; @@ -37,9 +38,9 @@ use shared::{ error::{get_errors_notify_message, notify_err, TuliproxError}, foundation::{get_field_value, set_field_value, Filter, ValueAccessor, ValueProvider}, model::{ - xtream_const::XTREAM_CLUSTER, CounterModifier, FieldGetAccessor, FieldSetAccessor, InputStats, InputType, - ItemField, PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistStats, ProcessingOrder, SourceStats, - StreamProperties, TargetStats, UUIDType, XtreamCluster, + ClusterSource, CounterModifier, FieldGetAccessor, FieldSetAccessor, InputStats, InputType, ItemField, + PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistStats, ProcessingOrder, SourceStats, StreamProperties, + TargetStats, UUIDType, XtreamCluster, }, utils::{ create_alias_uuid, default_as_default, default_probe_delay_secs, default_probe_live_interval, interner_gc, @@ -320,6 +321,60 @@ impl PlaylistDownloadResult { } } +/// Returns true when the input uses a hybrid download strategy: +/// M3U staged providing some clusters and Xtream main providing others. +fn is_hybrid_m3u_xtream(input: &ConfigInput) -> bool { + if !input.input_type.is_xtream() { + return false; + } + input.staged.as_ref().is_some_and(|s| s.enabled && s.input_type.is_m3u()) +} + +fn collect_effective_skip_clusters(input: &ConfigInput) -> Vec { + if !input.input_type.is_xtream() { + return vec![]; + } + + let mut skip_cluster = xtream::get_skip_cluster(input); + if let Some(staged) = input.staged.as_ref().filter(|staged| staged.enabled) { + for cluster in [XtreamCluster::Live, XtreamCluster::Video, XtreamCluster::Series] { + if staged.get_cluster_source(cluster) == ClusterSource::Skip && !skip_cluster.contains(&cluster) { + skip_cluster.push(cluster); + } + } + } + skip_cluster +} + +fn hybrid_has_m3u_staged_cluster(input: &ConfigInput, skip_cluster: &[XtreamCluster]) -> bool { + if !is_hybrid_m3u_xtream(input) { + return false; + } + + input.staged.as_ref().is_some_and(|staged| { + [XtreamCluster::Live, XtreamCluster::Video, XtreamCluster::Series] + .iter() + .any(|cluster| { + !skip_cluster.contains(cluster) + && staged.get_cluster_source(*cluster) == ClusterSource::Staged + }) + }) +} + +fn filter_skipped_clusters_from_source( + source: Box, + input: &ConfigInput, +) -> Box { + let skip_clusters = collect_effective_skip_clusters(input); + if skip_clusters.is_empty() { + return source; + } + + let skip_set: HashSet = skip_clusters.into_iter().collect(); + Box::new(FilteredPlaylistSource::new(source, skip_set)) +} + +#[allow(clippy::too_many_lines)] async fn playlist_download_from_input( client: &reqwest::Client, app_config: &Arc, @@ -327,7 +382,6 @@ async fn playlist_download_from_input( ) -> PlaylistDownloadResult { let config = &*app_config.config.load(); let storage_dir = &config.storage_dir; - let download_input_type = input.get_download_input_type(); // Check Status let storage_path = input_cache::resolve_input_storage_path(storage_dir, &input.name).await; @@ -339,81 +393,145 @@ async fn playlist_download_from_input( let _ = std::fs::create_dir_all(&storage_path); } - let (clusters_to_download, fully_cached) = match download_input_type { - InputType::Xtream => { - let mut to_download = vec![]; - for c in XTREAM_CLUSTER { - if !input_cache::is_cache_valid(&status, &c.to_string(), cache_duration) { - to_download.push(c); - } - } - if to_download.is_empty() { - (None, true) // Everything cached - } else { - (Some(to_download), false) + let hybrid = is_hybrid_m3u_xtream(input); + let download_input_type = input.get_download_input_type(); + // Use per-cluster cache for effective Xtream downloads and hybrid M3U+Xtream inputs. + let use_per_cluster_cache = hybrid || download_input_type.is_xtream(); + + let mut xtream_clusters_to_download = Vec::new(); + let mut needs_m3u_download = false; + let fully_cached = if use_per_cluster_cache { + let skip_cluster = collect_effective_skip_clusters(input); + let (staged_candidates, main_candidates) = + xtream::partition_clusters_by_source(input, None, &skip_cluster); + + let mut xtream_cache_candidates = staged_candidates; + xtream_cache_candidates.extend(main_candidates); + + for cluster in xtream_cache_candidates { + if !input_cache::is_cache_valid(&status, &cluster.to_string(), cache_duration) { + xtream_clusters_to_download.push(cluster); } } - _ => { - // M3U / Library - if input_cache::is_cache_valid(&status, "default", cache_duration) { - (None, true) - } else { - (None, false) // Download all + + if hybrid { + if hybrid_has_m3u_staged_cluster(input, &skip_cluster) { + needs_m3u_download = !input_cache::is_cache_valid(&status, "default", cache_duration); } + xtream_clusters_to_download.is_empty() && !needs_m3u_download + } else { + xtream_clusters_to_download.is_empty() } + } else { + input_cache::is_cache_valid(&status, "default", cache_duration) }; if fully_cached { return PlaylistDownloadResult::new(vec![], vec![], true, false); } - let (playlist, errors, persisted) = match download_input_type { - InputType::M3u => { - let (p, e) = m3u::download_m3u_playlist(app_config, client, config, input).await; - (p, e, false) + let (playlist, errors, persisted, m3u_error_count, xtream_error_count) = if hybrid { + let mut playlist = Vec::new(); + let mut all_errors = Vec::new(); + let mut m3u_error_count = 0usize; + let mut xtream_error_count = 0usize; + let mut m3u_added_groups = false; + + if needs_m3u_download { + let staged = input.staged.as_ref().expect("hybrid requires staged input"); + let staged_source: crate::model::InputSource = staged.into(); + let (m3u_playlist, m3u_errors) = + m3u::download_m3u_playlist_from_source(app_config, client, config, input, Some(staged_source)).await; + m3u_error_count = m3u_errors.len(); + m3u_added_groups = !m3u_playlist.is_empty(); + playlist.extend(m3u_playlist); + all_errors.extend(m3u_errors); } - InputType::Xtream => { - xtream::download_xtream_playlist(app_config, client, input, clusters_to_download.as_deref()).await + + let mut xtream_persisted = false; + if !xtream_clusters_to_download.is_empty() { + let (xtream_playlist, xtream_errors, persisted) = xtream::download_xtream_playlist( + app_config, + client, + input, + Some(xtream_clusters_to_download.as_slice()), + ) + .await; + xtream_error_count = xtream_errors.len(); + playlist.extend(xtream_playlist); + all_errors.extend(xtream_errors); + xtream_persisted = persisted; + + if m3u_added_groups { + // Keep merged hybrid output in memory when staged M3U contributed groups. + xtream_persisted = false; + } } - InputType::M3uBatch | InputType::XtreamBatch => (vec![], vec![], false), - InputType::Library => { - let (p, e) = library::download_library_playlist(client, app_config, input).await; - (p, e, false) + + (playlist, all_errors, xtream_persisted, m3u_error_count, xtream_error_count) + } else { + match download_input_type { + InputType::M3u => { + let (p, e) = m3u::download_m3u_playlist(app_config, client, config, input).await; + (p, e, false, 0, 0) + } + InputType::Xtream => { + let (p, e, persisted) = xtream::download_xtream_playlist( + app_config, + client, + input, + Some(xtream_clusters_to_download.as_slice()), + ) + .await; + let xtream_error_count = e.len(); + (p, e, persisted, 0, xtream_error_count) + } + InputType::M3uBatch | InputType::XtreamBatch => (vec![], vec![], false, 0, 0), + InputType::Library => { + let (p, e) = library::download_library_playlist(client, app_config, input).await; + (p, e, false, 0, 0) + } } }; // Update Status - if errors.is_empty() { - if let InputType::Xtream = download_input_type { - if let Some(clusters) = clusters_to_download { - for c in clusters { - input_cache::update_cluster_status(&mut status, &c.to_string(), ClusterState::Ok); - } - } else { - // All clusters logic if None passed (implies all were invalid or first run) - for c in XTREAM_CLUSTER { - input_cache::update_cluster_status(&mut status, &c.to_string(), ClusterState::Ok); - } + let mut save_status = false; + if hybrid { + if needs_m3u_download { + input_cache::update_cluster_status( + &mut status, + "default", + if m3u_error_count == 0 { ClusterState::Ok } else { ClusterState::Failed }, + ); + save_status = true; + } + + if !xtream_clusters_to_download.is_empty() { + let state = if xtream_error_count == 0 { ClusterState::Ok } else { ClusterState::Failed }; + for cluster in &xtream_clusters_to_download { + input_cache::update_cluster_status(&mut status, &cluster.to_string(), state.clone()); } + save_status = true; + } + } else if errors.is_empty() { + if use_per_cluster_cache { + for cluster in &xtream_clusters_to_download { + input_cache::update_cluster_status(&mut status, &cluster.to_string(), ClusterState::Ok); + } + save_status = !xtream_clusters_to_download.is_empty(); } else { input_cache::update_cluster_status(&mut status, "default", ClusterState::Ok); + save_status = true; } + } else if use_per_cluster_cache { + for cluster in &xtream_clusters_to_download { + input_cache::update_cluster_status(&mut status, &cluster.to_string(), ClusterState::Failed); + } + save_status = !xtream_clusters_to_download.is_empty(); + } + + if save_status { input_cache::save_input_status(&storage_path, &status); - } else { - // Mark failed? - // We could mark specific clusters as failed if we knew which one failed. - // For simplicity, if error, we don't update the timestamp (so it stays expired/invalid). - // Or we mark as Failed. - if let InputType::Xtream = download_input_type { - if let Some(clusters) = clusters_to_download { - for c in clusters { - // Optimistic: Only mark failed if we are sure? - // Currently just don't update the status to OK. - input_cache::update_cluster_status(&mut status, &c.to_string(), ClusterState::Failed); - } - } - input_cache::save_input_status(&storage_path, &status); - } } PlaylistDownloadResult::new(playlist, errors, false, persisted) @@ -602,6 +720,8 @@ async fn download_input( (MemoryPlaylistSource::new(pl).boxed(), err) }; + let playlist = filter_skipped_clusters_from_source(playlist, input); + if mark_as_processed { // Mark after persist/load so other workers only see this input as ready when data is usable. ctx.mark_input_downloaded(input.name.clone()).await; @@ -1275,6 +1395,8 @@ pub async fn exec_processing( None }; + info!("🌷 Update process started."); + log_memory_snapshot("exec_processing start"); // Initialize Context @@ -1367,6 +1489,7 @@ pub async fn exec_processing( #[cfg(test)] mod tests { use super::*; + use crate::model::StagedInput; use shared::utils::Internable; fn item_with_props(props: StreamProperties) -> PlaylistItem { @@ -1424,4 +1547,124 @@ mod tests { let item = item_with_props(StreamProperties::Series(Box::new(series))); assert!(!has_probe_details(&item)); } + + #[test] + fn hybrid_detection_requires_xtream_main_and_m3u_staged() { + let hybrid_input = ConfigInput { + name: "hybrid".intern(), + input_type: InputType::Xtream, + staged: Some(StagedInput { enabled: true, input_type: InputType::M3u, ..StagedInput::default() }), + ..ConfigInput::default() + }; + assert!(is_hybrid_m3u_xtream(&hybrid_input)); + + let non_hybrid_main_m3u = ConfigInput { + name: "main_m3u".intern(), + input_type: InputType::M3u, + staged: Some(StagedInput { enabled: true, input_type: InputType::Xtream, ..StagedInput::default() }), + ..ConfigInput::default() + }; + assert!(!is_hybrid_m3u_xtream(&non_hybrid_main_m3u)); + } + + #[test] + fn collect_effective_skip_clusters_includes_staged_skip_only_when_enabled() { + let input_with_enabled_staged = ConfigInput { + name: "enabled".intern(), + input_type: InputType::Xtream, + staged: Some(StagedInput { + enabled: true, + live_source: ClusterSource::Skip, + vod_source: ClusterSource::Input, + series_source: ClusterSource::Staged, + ..StagedInput::default() + }), + ..ConfigInput::default() + }; + let skip_enabled = collect_effective_skip_clusters(&input_with_enabled_staged); + assert!(skip_enabled.contains(&XtreamCluster::Live)); + assert!(!skip_enabled.contains(&XtreamCluster::Video)); + assert!(!skip_enabled.contains(&XtreamCluster::Series)); + + let input_with_disabled_staged = ConfigInput { + name: "disabled".intern(), + input_type: InputType::Xtream, + staged: Some(StagedInput { + enabled: false, + live_source: ClusterSource::Skip, + vod_source: ClusterSource::Skip, + series_source: ClusterSource::Skip, + ..StagedInput::default() + }), + ..ConfigInput::default() + }; + assert!(collect_effective_skip_clusters(&input_with_disabled_staged).is_empty()); + } + + #[test] + fn hybrid_m3u_detects_any_staged_cluster_selection() { + let hybrid = ConfigInput { + name: "hybrid".intern(), + input_type: InputType::Xtream, + staged: Some(StagedInput { + enabled: true, + input_type: InputType::M3u, + live_source: ClusterSource::Input, + vod_source: ClusterSource::Staged, + series_source: ClusterSource::Skip, + ..StagedInput::default() + }), + ..ConfigInput::default() + }; + assert!(hybrid_has_m3u_staged_cluster(&hybrid, &[])); + assert!(!hybrid_has_m3u_staged_cluster(&hybrid, &[XtreamCluster::Video])); + } + + #[test] + fn filter_skipped_clusters_removes_cached_groups() { + let live_item = PlaylistItem { + header: shared::model::PlaylistItemHeader { + xtream_cluster: XtreamCluster::Live, + ..Default::default() + }, + }; + let vod_item = PlaylistItem { + header: shared::model::PlaylistItemHeader { + xtream_cluster: XtreamCluster::Video, + ..Default::default() + }, + }; + + let groups = vec![ + PlaylistGroup { + id: 1, + title: "Live".intern(), + channels: vec![live_item], + xtream_cluster: XtreamCluster::Live, + }, + PlaylistGroup { + id: 2, + title: "Vod".intern(), + channels: vec![vod_item], + xtream_cluster: XtreamCluster::Video, + }, + ]; + + let source = MemoryPlaylistSource::new(groups).boxed(); + let input = ConfigInput { + name: "skip_live".intern(), + input_type: InputType::Xtream, + staged: Some(StagedInput { + enabled: true, + live_source: ClusterSource::Skip, + ..StagedInput::default() + }), + ..ConfigInput::default() + }; + + let mut filtered = filter_skipped_clusters_from_source(source, &input); + let filtered_groups = filtered.take_groups(); + assert_eq!(filtered_groups.len(), 1); + assert_eq!(filtered_groups[0].xtream_cluster, XtreamCluster::Video); + } } diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 05111e823..f2198c814 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -45,6 +45,14 @@ macro_rules! cant_write_result { }; } +// fn rebuild_source_ordinal_index_if_present(xtream_path: &Path) -> Result<(), std::io::Error> { +// let index_path = get_file_path_for_db_index(xtream_path); +// if !index_path.exists() { +// return Ok(()); +// } +// BPlusTree::::store_index(xtream_path, |pli| pli.source_ordinal) +// } + #[inline] pub fn get_collection_path(path: &Path, collection: &str) -> PathBuf { path.join(format!("{collection}.json")) @@ -156,6 +164,7 @@ pub async fn write_playlist_item_update( let _guard = file_lock; let mut tree = BPlusTreeUpdate::::try_new_with_backoff(&xtream_path_clone)?; tree.upsert_batch_encoded(prepared_items)?; + // rebuild_source_ordinal_index_if_present(&xtream_path_clone)?; Ok(()) }) .await @@ -201,6 +210,7 @@ pub async fn write_playlist_batch_item_upsert( let _guard = file_lock; let mut tree = BPlusTreeUpdate::::try_new_with_backoff(&xtream_path_clone)?; tree.upsert_batch_encoded(prepared_items)?; + //rebuild_source_ordinal_index_if_present(&xtream_path_clone)?; Ok(()) }) .await @@ -1323,6 +1333,8 @@ async fn persist_input_info(app_config: &Arc, storage_path: &Path, cl Ok(Some(mut pli)) => { 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_owned}: {err}")))?; + //rebuild_source_ordinal_index_if_present(&xtream_path_clone) + // .map_err(|err| Error::other(format!("failed to rebuild sorted index for input {input_name_owned}: {err}")))?; } Ok(None) => { error!("Could not find input entry for provider_id: {provider_id} and input: {input_name_owned}"); @@ -1377,6 +1389,8 @@ pub async fn persist_input_info_batch(app_config: &Arc, storage_path: .map(|(id, pli)| (id, pli)) .collect(); tree.update_batch(&refs).map_err(|err| Error::other(format!("failed to write batch {cluster} info for input {input_name_owned}: {err}")))?; + //rebuild_source_ordinal_index_if_present(&xtream_path_clone) + // .map_err(|err| Error::other(format!("failed to rebuild sorted index for input {input_name_owned}: {err}")))?; } Ok(()) }).await.map_err(|err| Error::other(format!("failed to join blocking input info batch persist for {input_name}: {err}")))??; diff --git a/backend/src/utils/file/config_reader.rs b/backend/src/utils/file/config_reader.rs index 48879ecb8..85c79c64f 100644 --- a/backend/src/utils/file/config_reader.rs +++ b/backend/src/utils/file/config_reader.rs @@ -425,6 +425,8 @@ pub async fn prepare_sources_batch( current_index = std::cmp::max(current_index, max_id_in_source); for input in &mut sources.inputs { + // Normalize batch type from URL before deciding whether aliases must be loaded from CSV. + input.prepare_type()?; match get_batch_aliases(input.input_type, input.url.as_str()).await { Ok(Some((_, aliases))) => { if let Some(idx) = input.prepare_batch(aliases, current_index)? { @@ -437,7 +439,6 @@ pub async fn prepare_sources_batch( return Err(err); } } - input.prepare_type()?; // we need to prepare epg after alias, because epg `auto` depends on the first input url. input.prepare_epg(include_computed)?; } @@ -1025,8 +1026,9 @@ async fn persist_single_template(prefix: &str, kind: Option<&MsgKind>, template: #[cfg(test)] mod tests { + use super::prepare_sources_batch; use crate::utils::{file::config_reader::get_batch_aliases, resolve_env_var}; - use shared::model::InputType; + use shared::{model::{ConfigInputDto, InputType, SourcesConfigDto}, utils::Internable}; #[test] #[allow(clippy::manual_unwrap_or_default)] @@ -1049,4 +1051,25 @@ mod tests { .contains("does not support provider:// URLs")); } + #[tokio::test] + async fn prepare_sources_batch_skips_csv_for_non_batch_url_even_with_batch_type() { + let mut sources = SourcesConfigDto { + inputs: vec![ConfigInputDto { + name: "input_http_aliases".intern(), + input_type: InputType::XtreamBatch, + url: "http://localhost:3001".to_string(), + username: Some("user".to_string()), + password: Some("pass".to_string()), + ..ConfigInputDto::default() + }], + ..SourcesConfigDto::default() + }; + + prepare_sources_batch(&mut sources, false) + .await + .expect("non-batch URL must not be treated as CSV batch source"); + + assert_eq!(sources.inputs[0].input_type, InputType::Xtream); + } + } diff --git a/backend/src/utils/network/m3u.rs b/backend/src/utils/network/m3u.rs index 2eba66ac6..e5cef8d52 100644 --- a/backend/src/utils/network/m3u.rs +++ b/backend/src/utils/network/m3u.rs @@ -11,14 +11,24 @@ pub async fn download_m3u_playlist( client: &reqwest::Client, cfg: &Arc, input: &ConfigInput, +) -> (Vec, Vec) { + download_m3u_playlist_from_source(app_config, client, cfg, input, None).await +} + +pub async fn download_m3u_playlist_from_source( + app_config: &Arc, + client: &reqwest::Client, + cfg: &Arc, + input: &ConfigInput, + explicit_source: Option, ) -> (Vec, Vec) { let storage_dir = &cfg.storage_dir; - let input_source: InputSource = { + let input_source: InputSource = explicit_source.unwrap_or_else(|| { match input.staged.as_ref() { None => input.into(), Some(staged) => if staged.enabled { staged.into() } else { input.into() }, } - }; + }); let persist_file_path = prepare_file_path(input.persist.as_deref(), storage_dir, ""); match request::get_input_text_content_as_stream( app_config, diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index 819a33b36..83b373d0e 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -3,6 +3,7 @@ use crate::messaging::send_message; use crate::model::{is_input_expired, xtream_mapping_option_from_target_options, AppConfig, ConfigInput, ConfigInputFlags, ConfigTarget, MessageContent, XtreamLoginInfo, XtreamTargetOutput}; use crate::model::{InputSource, ProxyUserCredentials}; +use shared::model::ClusterSource; use crate::processing::parser::xtream; use crate::processing::parser::xtream::parse_xtream_series_info; use crate::repository::VirtualIdRecord; @@ -253,7 +254,7 @@ fn xtream_resolve_stream_info(app_state: &Arc, user: &ProxyUserCredent None } -fn get_skip_cluster(input: &ConfigInput) -> Vec { +pub(crate) fn get_skip_cluster(input: &ConfigInput) -> Vec { let mut skip_cluster = vec![]; if input.has_flag(ConfigInputFlags::XtreamSkipLive) { skip_cluster.push(XtreamCluster::Live); @@ -345,15 +346,51 @@ pub async fn notify_account_expire(exp_date: Option, app_config: &Arc, client: &reqwest::Client, input: &ConfigInput, clusters: Option<&[XtreamCluster]>) - -> (Vec, Vec, bool) { - let input_source: InputSource = { - match input.staged.as_ref() { - None => input.into(), - Some(staged) => if staged.enabled { staged.into() } else { input.into() }, - } - }; +/// Partitions the requested clusters into staged and main groups based on +/// skip flags and per-cluster source configuration. +pub(crate) fn partition_clusters_by_source( + input: &ConfigInput, + requested: Option<&[XtreamCluster]>, + skip_cluster: &[XtreamCluster], +) -> (Vec, Vec) { + let mut staged_clusters = Vec::new(); + let mut main_clusters = Vec::new(); + let all_clusters = [XtreamCluster::Live, XtreamCluster::Video, XtreamCluster::Series]; + for cluster in &all_clusters { + let is_requested = requested.is_none_or(|c| c.contains(cluster)); + if !is_requested || skip_cluster.contains(cluster) { + continue; + } + + if let Some(staged) = input.staged.as_ref().filter(|s| s.enabled) { + match staged.get_cluster_source(*cluster) { + ClusterSource::Staged => { + // M3U staged inputs are handled by the M3U download path, never by Xtream API calls. + if staged.input_type.is_xtream() { + staged_clusters.push(*cluster); + } + } + ClusterSource::Input => main_clusters.push(*cluster), + ClusterSource::Skip => {} // excluded + } + } else { + main_clusters.push(*cluster); + } + } + + (staged_clusters, main_clusters) +} + +/// Downloads xtream clusters from a single source (either main input or staged input). +async fn download_xtream_from_source( + app_config: &Arc, + client: &reqwest::Client, + input: &ConfigInput, + input_source: &InputSource, + source_input_type: InputType, + clusters: &[XtreamCluster], +) -> (Vec, Vec, bool) { let (username, password) = ( input_source.username.as_deref().unwrap_or(""), input_source.password.as_deref().unwrap_or(""), @@ -361,8 +398,6 @@ pub async fn download_xtream_playlist(app_config: &Arc, client: &reqw let is_provider_url = input_source.url.starts_with(PROVIDER_SCHEME_PREFIX); let base_input_url = if is_provider_url { - // Keep provider:// unresolved; send_with_retry_and_provider resolves per attempt - // so failover can switch provider hosts. input_source.url.clone() } else { match input.resolve_url(&input_source.url) { @@ -371,83 +406,106 @@ pub async fn download_xtream_playlist(app_config: &Arc, client: &reqw } }; - // Build a canonical Xtream player_api URL for both plain URLs and provider:// URLs. - // For provider URLs, this keeps the scheme unresolved while still adding the required path/query. let base_url = get_xtream_stream_url_base(&base_input_url, username, password); let input_source_login = input_source.with_url(base_url.clone()); - check_alias_user_state(app_config, client, input).await; - if let Err(err) = xtream_login(app_config, client, &input_source_login, username).await { error!("Could not log in with xtream user {username} for provider {}. {err}", input.name); return (Vec::with_capacity(0), vec![err], false); } let mut playlist_groups: Vec = Vec::with_capacity(128); - let skip_cluster = get_skip_cluster(input); let cfg = app_config.config.load(); let storage_dir = &cfg.storage_dir; - // Staged input type is the effective type during download. - let effective_type = input.get_download_input_type(); let use_disk_based_processing = - cfg.disk_based_processing && matches!(effective_type, InputType::Xtream | InputType::XtreamBatch); + cfg.disk_based_processing && matches!(source_input_type, InputType::Xtream | InputType::XtreamBatch); let mut errors = vec![]; for (xtream_cluster, category, stream) in &ACTIONS { - let is_requested = clusters.is_none_or(|c| c.contains(xtream_cluster)); - if is_requested && !skip_cluster.contains(xtream_cluster) { - let input_source_category = input_source.with_url(concat_string!(&base_url, "&action=", category)); - let input_source_stream = input_source.with_url(concat_string!(&base_url, "&action=", stream)); - let category_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), - storage_dir, concat_string!(category, "_").as_str()); - let stream_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), - storage_dir, concat_string!(stream, "_").as_str()); + if !clusters.contains(xtream_cluster) { + continue; + } + let input_source_category = input_source.with_url(concat_string!(&base_url, "&action=", category)); + let input_source_stream = input_source.with_url(concat_string!(&base_url, "&action=", stream)); + let category_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), + storage_dir, concat_string!(category, "_").as_str()); + let stream_file_path = crate::utils::prepare_file_path(input.persist.as_deref(), + storage_dir, concat_string!(stream, "_").as_str()); - match futures::join!( - request::get_input_json_content_as_stream(app_config, client, &input_source_category, category_file_path), - request::get_input_json_content_as_stream(app_config, client, &input_source_stream, stream_file_path) - ) { - (Ok(category_content), Ok(stream_content)) => { - if use_disk_based_processing { - // trace!("Using disk input playlist optimization for cluster {}", xtream_cluster); - if let Err(err) = persist_input_xtream_playlist_cluster_to_disk(app_config, input, *xtream_cluster, category_content, stream_content).await { - error!("persist_input_xtream_playlist_cluster_to_disk failed: {err}"); - errors.push(err); - //} else { - // trace!("persist_input_xtream_playlist_cluster_to_disk succeeded for cluster {}", xtream_cluster); - } - } else { - // trace!("Using in-memory playlist parsing for cluster {}", xtream_cluster); - match xtream::parse_xtream(input, - *xtream_cluster, - category_content, - stream_content).await { - Ok(sub_playlist_parsed) => { - 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)); - } + match futures::join!( + request::get_input_json_content_as_stream(app_config, client, &input_source_category, category_file_path), + request::get_input_json_content_as_stream(app_config, client, &input_source_stream, stream_file_path) + ) { + (Ok(category_content), Ok(stream_content)) => { + if use_disk_based_processing { + if let Err(err) = persist_input_xtream_playlist_cluster_to_disk(app_config, input, *xtream_cluster, category_content, stream_content).await { + error!("persist_input_xtream_playlist_cluster_to_disk failed: {err}"); + errors.push(err); + } + } else { + match xtream::parse_xtream(input, + *xtream_cluster, + category_content, + stream_content).await { + Ok(sub_playlist_parsed) => { + 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)); } - Err(err) => errors.push(err) } + Err(err) => errors.push(err) } } - (Err(err1), Err(err2)) => { - errors.extend([err1, err2]); - } - (_, Err(err)) | (Err(err), _) => errors.push(err), } + (Err(err1), Err(err2)) => { + errors.extend([err1, err2]); + } + (_, Err(err)) | (Err(err), _) => errors.push(err), } } - for (grp_id, plg) in (1_u32..).zip(playlist_groups.iter_mut()) { + (playlist_groups, errors, use_disk_based_processing) +} + +pub async fn download_xtream_playlist(app_config: &Arc, client: &reqwest::Client, input: &ConfigInput, clusters: Option<&[XtreamCluster]>) + -> (Vec, Vec, bool) { + let skip_cluster = get_skip_cluster(input); + let (staged_clusters, main_clusters) = partition_clusters_by_source(input, clusters, &skip_cluster); + + let mut all_groups = Vec::with_capacity(128); + let mut all_errors = Vec::new(); + let mut any_disk = false; + + if !staged_clusters.is_empty() { + if let Some(staged) = input.staged.as_ref().filter(|s| s.enabled) { + let source: InputSource = staged.into(); + let (g, e, d) = + download_xtream_from_source(app_config, client, input, &source, staged.input_type, &staged_clusters) + .await; + all_groups.extend(g); + all_errors.extend(e); + any_disk |= d; + } + } + + if !main_clusters.is_empty() { + check_alias_user_state(app_config, client, input).await; + let source: InputSource = input.into(); + let (g, e, d) = + download_xtream_from_source(app_config, client, input, &source, input.input_type, &main_clusters).await; + all_groups.extend(g); + all_errors.extend(e); + any_disk |= d; + } + + for (grp_id, plg) in (1_u32..).zip(all_groups.iter_mut()) { plg.id = grp_id; } - (playlist_groups, errors, use_disk_based_processing) + (all_groups, all_errors, any_disk) } async fn check_alias_user_state(app_config: &Arc, client: &reqwest::Client, input: &ConfigInput) { @@ -524,3 +582,70 @@ pub fn create_vod_info_from_item(target: &ConfigTarget, user: &ProxyUserCredenti serde_json::to_string(&doc).unwrap_or(String::new()) } + +#[cfg(test)] +mod tests { + use super::partition_clusters_by_source; + use crate::model::{ConfigInput, ConfigInputFlags, ConfigInputFlagsSet, ConfigInputOptions, StagedInput}; + use shared::model::{ClusterSource, InputType, XtreamCluster}; + use shared::utils::Internable; + + fn test_input_with_staged(staged: StagedInput) -> ConfigInput { + ConfigInput { + name: "test".intern(), + input_type: InputType::Xtream, + staged: Some(staged), + ..ConfigInput::default() + } + } + + fn options_with_flags(flags: &[ConfigInputFlags]) -> ConfigInputOptions { + let mut set = ConfigInputFlagsSet::new(); + for flag in flags { + set.set(*flag); + } + ConfigInputOptions { flags: set, ..ConfigInputOptions::defaults().clone() } + } + + #[test] + fn partition_respects_skip_flags_before_cluster_source() { + let staged = StagedInput { + enabled: true, + input_type: InputType::Xtream, + live_source: ClusterSource::Staged, + vod_source: ClusterSource::Input, + series_source: ClusterSource::Skip, + ..StagedInput::default() + }; + let mut input = test_input_with_staged(staged); + input.options = Some(options_with_flags(&[ConfigInputFlags::XtreamSkipLive])); + + let skip_cluster = super::get_skip_cluster(&input); + let (staged_clusters, main_clusters) = partition_clusters_by_source(&input, None, &skip_cluster); + + assert!(staged_clusters.is_empty()); + assert_eq!(main_clusters, vec![XtreamCluster::Video]); + } + + #[test] + fn partition_excludes_staged_clusters_for_m3u_staged_inputs() { + let staged = StagedInput { + enabled: true, + input_type: InputType::M3u, + live_source: ClusterSource::Staged, + vod_source: ClusterSource::Input, + series_source: ClusterSource::Input, + ..StagedInput::default() + }; + let input = test_input_with_staged(staged); + + let (staged_clusters, main_clusters) = partition_clusters_by_source( + &input, + Some(&[XtreamCluster::Live, XtreamCluster::Video]), + &[], + ); + + assert!(staged_clusters.is_empty()); + assert_eq!(main_clusters, vec![XtreamCluster::Video]); + } +} diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 8dc0267f6..de1c8d54e 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -374,7 +374,11 @@ "FETCH_METHOD": "HTTP method used to retrieve the input (e.g., GET, POST).", "HEADERS": "Custom HTTP headers sent when retrieving the input playlist.", "INPUT_TYPE": "Defines the streaming protocol or source format (e.g. xtream, m3u).", + "LIVE_SOURCE": "Source selector for the Live cluster when staged input is enabled (`default`, `staged`, `input`, `skip`).", "PROVIDERS": "Provider profiles define reusable upstream URLs and DNS behavior that this input can reference." + , + "SERIES_SOURCE": "Source selector for the Series cluster when staged input is enabled (`default`, `staged`, `input`, `skip`).", + "VOD_SOURCE": "Source selector for the VOD cluster when staged input is enabled (`default`, `staged`, `input`, `skip`)." }, "IP_CHECK_CONFIG": { "PATTERN_IPV4": "Optional regex to extract IPv4 address from the response (default matches standard decimal format).", @@ -1060,6 +1064,9 @@ "SPECIFIC_TEMPLATES": "Specific Templates", "SSDP_DISCOVERY": "SSDP Discovery", "STAGED": "Staged", + "LIVE_SOURCE": "Live Source", + "VOD_SOURCE": "VOD Source", + "SERIES_SOURCE": "Series Source", "START": "Start", "STAR_ON_GITHUB": "If you like this project, star ⭐ us on GitHub", "STATS": "Stats", diff --git a/frontend/src/app/components/playlist/input/staged_input_view.rs b/frontend/src/app/components/playlist/input/staged_input_view.rs index 87451bd43..4e4056be6 100644 --- a/frontend/src/app/components/playlist/input/staged_input_view.rs +++ b/frontend/src/app/components/playlist/input/staged_input_view.rs @@ -3,7 +3,7 @@ use crate::{ html_if, i18n::use_translation, }; -use shared::model::{InputType, StagedInputDto}; +use shared::model::{ClusterSource, InputType, StagedInputDto}; use yew::prelude::*; #[derive(Properties, Clone, PartialEq, Debug)] @@ -44,6 +44,29 @@ pub fn StagedInputView(props: &StagedInputViewProps) -> Html { + { + { + let fmt = |cs: &Option| -> String { + cs.as_ref().map(ToString::to_string).unwrap_or_else(|| ClusterSource::default().to_string()) + }; + html! { + <> +
+ + { fmt(&input.live_source) } +
+
+ + { fmt(&input.vod_source) } +
+
+ + { fmt(&input.series_source) } +
+ + } + } + } } } diff --git a/frontend/src/app/components/source_editor/editor_view.rs b/frontend/src/app/components/source_editor/editor_view.rs index 82fb1c637..e26ebdeb2 100644 --- a/frontend/src/app/components/source_editor/editor_view.rs +++ b/frontend/src/app/components/source_editor/editor_view.rs @@ -12,9 +12,12 @@ use crate::{ model::DialogResult, services::DialogService, }; -use shared::model::{ - ConfigInputDto, ConfigSourceDto, ConfigTargetDto, HdHomeRunTargetOutputDto, InputType, M3uTargetOutputDto, - SourcesConfigDto, StrmTargetOutputDto, TargetOutputDto, XtreamTargetOutputDto, +use shared::{ + model::{ + ConfigInputDto, ConfigSourceDto, ConfigTargetDto, HdHomeRunTargetOutputDto, InputType, M3uTargetOutputDto, + SourcesConfigDto, StrmTargetOutputDto, TargetOutputDto, XtreamTargetOutputDto, + }, + utils::BATCH_SCHEME_PREFIX, }; use std::{ cell::RefCell, @@ -257,14 +260,14 @@ fn create_output_instance(output: &TargetOutputDto) -> (BlockInstance, BlockType } } -fn normalize_input_type_by_aliases(input: &mut ConfigInputDto, block_type: BlockType) { - let has_aliases = input.aliases.as_ref().is_some_and(|aliases| !aliases.is_empty()); +fn normalize_input_type_by_url(input: &mut ConfigInputDto, block_type: BlockType) { + let is_batch_url = input.url.trim().starts_with(BATCH_SCHEME_PREFIX); match block_type { BlockType::InputXtream => { - input.input_type = if has_aliases { InputType::XtreamBatch } else { InputType::Xtream }; + input.input_type = if is_batch_url { InputType::XtreamBatch } else { InputType::Xtream }; } BlockType::InputM3u => { - input.input_type = if has_aliases { InputType::M3uBatch } else { InputType::M3u }; + input.input_type = if is_batch_url { InputType::M3uBatch } else { InputType::M3u }; } BlockType::InputLibrary => { input.input_type = InputType::Library; @@ -273,6 +276,16 @@ fn normalize_input_type_by_aliases(input: &mut ConfigInputDto, block_type: Block } } +fn input_dedupe_key(input_config: &ConfigInputDto) -> Option { + if input_config.id > 0 { + Some(format!("id:{}", input_config.id)) + } else if !input_config.name.is_empty() { + Some(format!("name:{}", input_config.name)) + } else { + None + } +} + fn editor_state_to_sources_config(base_sources: &SourcesConfigDto, editor_state: &EditorState) -> SourcesConfigDto { let mut sources_config = base_sources.clone(); let mut gen_sources: Vec = Vec::new(); @@ -283,7 +296,7 @@ fn editor_state_to_sources_config(base_sources: &SourcesConfigDto, editor_state: for block in input_blocks { if let BlockInstance::Input(input) = &block.instance { let mut normalized_input = input.as_ref().clone(); - normalize_input_type_by_aliases(&mut normalized_input, block.block_type); + normalize_input_type_by_url(&mut normalized_input, block.block_type); gen_inputs.push(normalized_input); } } @@ -396,12 +409,13 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { { let playlists = playlist_ctx.clone(); + let config_ctx = config_ctx.clone(); let editor_state_ref = editor_state_ref.clone(); let force_update = force_update.clone(); let initialized_from_playlist = initialized_from_playlist.clone(); use_effect_with( - (playlists.sources.clone(), is_local_mode, *initialized_from_playlist), - move |(sources, local_mode, initialized)| { + (playlists.sources.clone(), config_ctx.config.clone(), is_local_mode, *initialized_from_playlist), + move |(sources, app_config, local_mode, initialized)| { if !(*local_mode && *initialized) { if let Some(entries) = sources.as_ref() { let mut current_id = 1; @@ -413,13 +427,7 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { for input_row in inputs { match input_row.as_ref() { InputRow::Input(input_config) => { - let dedupe_key = if input_config.id > 0 { - Some(format!("id:{}", input_config.id)) - } else if !input_config.name.is_empty() { - Some(format!("name:{}", input_config.name)) - } else { - None - }; + let dedupe_key = input_dedupe_key(input_config); let block_id = if let Some(key) = dedupe_key { if let Some(&existing_id) = added_inputs.get(&key) { @@ -472,6 +480,30 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { } } } + + // Also include standalone inputs from sources.yml that are currently not wired to any source. + if let Some(cfg) = app_config.as_ref() { + for input_config in &cfg.sources.inputs { + let dedupe_key = input_dedupe_key(input_config); + if let Some(ref key) = dedupe_key { + if added_inputs.contains_key(key) { + continue; + } + } + + let block_id = current_id; + current_id += 1; + if let Some(key) = dedupe_key { + added_inputs.insert(key, block_id); + } + gen_blocks.push(create_block( + block_id, + BlockType::from(input_config.input_type), + BlockInstance::Input(Rc::new(input_config.clone())), + )); + } + } + if !gen_blocks.is_empty() { layout(&mut gen_blocks, &gen_connections); } @@ -563,6 +595,14 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { let base_sources = config_ctx.config.as_ref().map(|c| c.sources.clone()).unwrap_or_default(); let editor_state = editor_state_ref.borrow(); let sources_config = editor_state_to_sources_config(&base_sources, &editor_state); + let mut sources_to_validate = sources_config.clone(); + let hdhr_overview = config_ctx.config.as_ref().and_then(|cfg| cfg.config.get_hdhr_device_overview()); + let global_templates = + config_ctx.config.as_ref().and_then(|cfg| cfg.templates.as_ref().map(|defs| defs.templates.as_slice())); + if let Err(err) = sources_to_validate.prepare(false, hdhr_overview.as_ref(), global_templates) { + services.toastr.error(err.to_string()); + return; + } let services = services.clone(); let translate = translate.clone(); @@ -1288,7 +1328,7 @@ pub fn SourceEditor(props: &SourceEditorProps) -> Html { block.instance = match instance { BlockInstance::Input(input_cfg) => { let mut normalized_input = input_cfg.as_ref().clone(); - normalize_input_type_by_aliases(&mut normalized_input, block.block_type); + normalize_input_type_by_url(&mut normalized_input, block.block_type); BlockInstance::Input(Rc::new(normalized_input)) } other => other, diff --git a/frontend/src/app/components/source_editor/input_form.rs b/frontend/src/app/components/source_editor/input_form.rs index c6e126567..568065d18 100644 --- a/frontend/src/app/components/source_editor/input_form.rs +++ b/frontend/src/app/components/source_editor/input_form.rs @@ -16,8 +16,8 @@ use shared::{ error::TuliproxError, info_err_res, model::{ - ConfigInputAliasDto, ConfigInputDto, ConfigInputOptionsDto, ConfigProviderDto, EpgConfigDto, EpgSourceDto, - InputFetchMethod, InputType, StagedInputDto, + ClusterSource, ConfigInputAliasDto, ConfigInputDto, ConfigInputOptionsDto, ConfigProviderDto, EpgConfigDto, + EpgSourceDto, InputFetchMethod, InputType, StagedInputDto, }, }; use std::{ @@ -68,6 +68,9 @@ const LABEL_CACHE_DURATION: &str = "LABEL.CACHE_DURATION"; const LABEL_MAIN: &str = "LABEL.MAIN_CONFIG"; const LABEL_OPTIONS: &str = "LABEL.OPTIONS"; const LABEL_STAGED: &str = "LABEL.STAGED"; +const LABEL_LIVE_SOURCE: &str = "LABEL.LIVE_SOURCE"; +const LABEL_VOD_SOURCE: &str = "LABEL.VOD_SOURCE"; +const LABEL_SERIES_SOURCE: &str = "LABEL.SERIES_SOURCE"; const LABEL_ADVANCED: &str = "LABEL.ADVANCED"; const LABEL_ALIAS: &str = "LABEL.ALIAS"; const LABEL_PROVIDER: &str = "LABEL.PROVIDER"; @@ -157,6 +160,9 @@ generate_form_reducer!( Password => password: Option, Method => method: InputFetchMethod, InputType => input_type: InputType, + LiveSource => live_source: Option, + VodSource => vod_source: Option, + SeriesSource => series_source: Option, // Headers => headers: HashMap, } ); @@ -178,6 +184,23 @@ generate_form_reducer!( CacheDuration => cache_duration: Option, } ); +fn cluster_source_options(current: Option) -> Rc> { + let options: [(Option, &str); 3] = [ + (Some(ClusterSource::Staged), "staged"), + (Some(ClusterSource::Input), "input"), + (Some(ClusterSource::Skip), "skip"), + ]; + Rc::new( + options + .iter() + .map(|(val, label)| DropDownOption { + id: label.to_string(), + label: html! { label.to_string() }, + selected: *val == current, + }) + .collect(), + ) +} fn apply_parsed_input_type(staged_input_state: &UseReducerHandle, selected: Option<&str>) { let input_type = @@ -611,6 +634,13 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { let staged_method_selection = Rc::new(vec![staged_input_state.form.method.to_string()]); let staged_input_state_1 = staged_input_state.clone(); let staged_input_state_2 = staged_input_state.clone(); + let staged_input_state_live = staged_input_state.clone(); + let staged_input_state_vod = staged_input_state.clone(); + let staged_input_state_series = staged_input_state.clone(); + let show_cluster_sources = input_form_state.form.input_type.is_xtream(); + let live_source_options = cluster_source_options(staged_input_state.form.live_source); + let vod_source_options = cluster_source_options(staged_input_state.form.vod_source); + let series_source_options = cluster_source_options(staged_input_state.form.series_source); html! { { edit_field_bool!(staged_input_state, translate.t(LABEL_ENABLED), enabled, StagedInputFormAction::Enabled) } @@ -658,6 +688,60 @@ pub fn ConfigInputView(props: &ConfigInputViewProps) -> Html { /> }})} + { + html_if!(show_cluster_sources, { +
+ { config_field_child!(translate.t(LABEL_LIVE_SOURCE), "INPUT_FORM.LIVE_SOURCE", { + html! { + + } + })} + { config_field_child!(translate.t(LABEL_SERIES_SOURCE), "INPUT_FORM.SERIES_SOURCE", { + html! { +