diff --git a/CHANGELOG.md b/CHANGELOG.md index 29f83786d..82fa62f42 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,6 @@ # Changelog # 2.2.3 (2023-04-xx) +- variable resolving for config files now for all settings - hls reverse proxy implemented - dash redirect implemented (reverse proxy not supported) - !BREAKING CHANGE! `channel_unavailable_file` is now under `custom_stream_response`, @@ -27,6 +28,43 @@ custom_stream_response: targets: - name: test ``` +Input aliases can be defined as batches in csv files with `;` separator. +There are 2 batch input types `xtream_batch` and `m3u_batch`. +`XtreamBatch`: + +```yaml +- sources: +- inputs: + - type: xtream_batch + url: 'file:///home/m3u-filter/config/my_provider_batch.csv' + targets: + - name: test +``` + +```csv +#name;username;password;url;max_connections;priority +my_provider_1;user1;password1;http://my_provider_1.com:80;1;0 +my_provider_2;user2;password2;http://my_provider_2.com:8080;1;0 +``` + +`M3uBatch`: +```yaml +- sources: +- inputs: + - type: m3u_batch + url: 'file:///home/m3u-filter/config/my_provider_batch.csv' + targets: + - name: test +``` + +```csv +#name;url;max_connections;priority +http://my_provider_1.com:80/get_php?username=user1&password=password1;1;0 +http://my_provider_2.com:8080/get_php?username=user2&password=password2;1;0 +``` +The Fields `max_connections` and `priority`are optional. +`max_connections` will be set default to `1`. This is different from yaml config where the default is `0=unlimited` + - added two options to reverse proxy config `forced_retry_interval_secs` and `connect_timeout_secs` `forced_retry_interval_secs` forces every x seconds a reconnect to the provider, `connect_timeout_secs` tries only x seconds for connection, if not successfully starts a retry. diff --git a/README.md b/README.md index e6332a5d7..12aef6a02 100644 --- a/README.md +++ b/README.md @@ -505,6 +505,45 @@ Input alias definition for same provider with same content but different credent - name: test ``` +Input aliases can be defined as batches in csv files with `;` separator. +There are 2 batch input types `xtream_batch` and `m3u_batch`. + +##### `XtreamBatch` + +```yaml +- sources: +- inputs: + - type: xtream_batch + url: 'file:///home/m3u-filter/config/my_provider_batch.csv' + targets: + - name: test +``` + +```csv +#name;username;password;url;max_connections;priority +my_provider_1;user1;password1;http://my_provider_1.com:80;1;0 +my_provider_2;user2;password2;http://my_provider_2.com:8080;1;0 +``` + +##### `M3uBatch` +```yaml +- sources: +- inputs: + - type: m3u_batch + url: 'file:///home/m3u-filter/config/my_provider_batch.csv' + targets: + - name: test +``` + +```csv +#name;url;max_connections;priority +http://my_provider_1.com:80/get_php?username=user1&password=password1;1;0 +http://my_provider_2.com:8080/get_php?username=user2&password=password2;1;0 +``` + +The Fields `max_connections` and `priority`are optional. +`max_connections` will be set default to `1`. This is different from yaml config where the default is `0=unlimited` + ### 2.2.2 `targets` Has the following top level entries: - `enabled` _optional_ default is `true`, if you disable the processing is skipped diff --git a/src/api/endpoints/v1_api.rs b/src/api/endpoints/v1_api.rs index b3b8fc6c8..518f758bd 100644 --- a/src/api/endpoints/v1_api.rs +++ b/src/api/endpoints/v1_api.rs @@ -76,7 +76,7 @@ async fn save_config_api_proxy_user( let mut lock = app_state.config.t_api_proxy.write().await; if let Some(api_proxy) = lock.as_mut() { api_proxy.user = users; - api_proxy.user.iter_mut().flat_map(|t| &mut t.credentials).for_each(|c| c.prepare(true)); + api_proxy.user.iter_mut().flat_map(|t| &mut t.credentials).for_each(|c| c.prepare()); if api_proxy.use_user_db { if let Err(err) = store_api_user(&app_state.config, &api_proxy.user) { return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response(); @@ -350,7 +350,7 @@ async fn config( app_state.config.t_config_file_path.as_str(), app_state.config.t_sources_file_path.as_str()) { Ok(mut cfg) => { - let _ = cfg.prepare(true); + let _ = cfg.prepare(); map_config(&cfg) } Err(_) => map_config(&app_state.config) diff --git a/src/main.rs b/src/main.rs index cffe64444..c5b276235 100644 --- a/src/main.rs +++ b/src/main.rs @@ -23,6 +23,7 @@ use crate::utils::network::request::set_sanitize_sensitive_info; use clap::Parser; use env_logger::Builder; use log::{error, info, LevelFilter}; +use crate::utils::file::config_reader::config_file_reader; const LOG_ERROR_LEVEL_MOD: &[&str] = &[ "reqwest::async_impl::client", @@ -137,7 +138,7 @@ fn main() { } } - match config_reader::read_mappings(args.mapping_file, &mut cfg) { + match config_reader::read_mappings(args.mapping_file, &mut cfg, true) { Ok(Some(mapping_file)) => { info!("Mapping file: {mapping_file:?}"); } @@ -226,7 +227,7 @@ fn init_logger(user_log_level: Option<&String>, env_log_level: Option, c .or(env_log_level) // env .or_else(|| { // config File::open(config_file).ok() - .and_then(|file| serde_yaml::from_reader::<_, LogLevelConfig>(file).ok()) + .and_then(|file| serde_yaml::from_reader::<_, LogLevelConfig>(config_file_reader(file, true)).ok()) .and_then(|cfg| cfg.log.and_then(|l| l.log_level)) }) .unwrap_or_else(|| "info".to_string()); // Default @@ -252,7 +253,7 @@ fn init_logger(user_log_level: Option<&String>, env_log_level: Option, c fn healthcheck(config_file: &str) { let path = std::path::PathBuf::from(config_file); let file = File::open(path).expect("Failed to open config file"); - let config: HealthcheckConfig = serde_yaml::from_reader(file).expect("Failed to parse config file"); + let config: HealthcheckConfig = serde_yaml::from_reader(config_file_reader(file, true)).expect("Failed to parse config file"); if let Ok(response) = reqwest::blocking::get(format!("http://localhost:{}/healthcheck", config.api.port)) { if let Ok(check) = response.json::() { diff --git a/src/model/api_proxy.rs b/src/model/api_proxy.rs index ec826c4d3..09deb0af7 100644 --- a/src/model/api_proxy.rs +++ b/src/model/api_proxy.rs @@ -128,15 +128,8 @@ pub struct ProxyUserCredentials { } impl ProxyUserCredentials { - pub fn prepare(&mut self, resolve_var: bool) { - if resolve_var { - self.username = config_reader::resolve_env_var(&self.username); - self.password = config_reader::resolve_env_var(&self.password); - if let Some(tkn) = &self.token { - self.token = Some(config_reader::resolve_env_var(tkn)); - } - self.trim(); - } + pub fn prepare(&mut self) { + self.trim(); } pub fn matches_token(&self, token: &str) -> bool { @@ -385,12 +378,12 @@ impl ApiProxyConfig { } } - fn prepare_target_user(&mut self, resolve_var: bool, errors: &mut Vec) { + fn prepare_target_user(&mut self, errors: &mut Vec) { let mut usernames = HashSet::new(); let mut tokens = HashSet::new(); for target_user in &mut self.user { for user in &mut target_user.credentials { - user.prepare(resolve_var); + user.prepare(); if usernames.contains(&user.username) { errors.push(format!("Non unique username found {}", &user.username)); } else { @@ -420,14 +413,14 @@ impl ApiProxyConfig { } } - pub fn prepare(&mut self, cfg: &Config, resolve_var: bool) -> Result<(), M3uFilterError> { + pub fn prepare(&mut self, cfg: &Config) -> Result<(), M3uFilterError> { let mut errors = Vec::new(); if self.server.is_empty() { errors.push("No server info defined".to_string()); } else { self.prepare_server_config(&mut errors); } - self.prepare_target_user(resolve_var, &mut errors); + self.prepare_target_user(&mut errors); self.migrate_api_user(cfg, &mut errors); if errors.is_empty() { Ok(()) diff --git a/src/model/config.rs b/src/model/config.rs index d58c03ec4..d72bd561b 100644 --- a/src/model/config.rs +++ b/src/model/config.rs @@ -23,7 +23,6 @@ use crate::model::api_proxy::{ApiProxyConfig, ApiProxyServerInfo, ProxyUserCrede use crate::model::mapping::Mapping; use crate::model::mapping::Mappings; use crate::utils::default_utils::{default_as_default, default_as_true, default_as_two_u16}; -use crate::utils::file::config_reader; use crate::utils::file::file_lock_manager::FileLockManager; use crate::utils::file::file_utils; use crate::utils::file::file_utils::file_reader; @@ -52,7 +51,7 @@ macro_rules! valid_property { pub use valid_property; use crate::m3u_filter_error::{create_m3u_filter_error_result, handle_m3u_filter_error_result, handle_m3u_filter_error_result_list}; use crate::model::hdhomerun_config::HdHomeRunConfig; -use crate::utils::file::config_reader::{csv_read_inputs, resolve_env_var}; +use crate::utils::file::config_reader::{csv_read_inputs}; use crate::utils::network::request::{get_credentials_from_url, get_credentials_from_url_str}; use crate::utils::string_utils::get_trimmed_string; @@ -423,11 +422,11 @@ impl ConfigTarget { } TargetOutput::M3u(m3u_output) => { m3u_cnt += 1; - m3u_output.filename = m3u_output.filename.as_ref().map(|s| resolve_env_var(s.trim())); + m3u_output.filename = m3u_output.filename.as_ref().map(|s| s.trim().to_string()); } TargetOutput::Strm(strm_output) => { strm_cnt += 1; - strm_output.directory = resolve_env_var(strm_output.directory.trim()); + strm_output.directory = strm_output.directory.trim().to_string(); if strm_output.directory.trim().is_empty() { return create_m3u_filter_error_result!(M3uFilterErrorKind::Info, "directory is required for strm type: {}", self.name); } @@ -783,7 +782,6 @@ impl ConfigInput { pub fn prepare(&mut self, index: u16) -> Result { self.id = index; self.check_url()?; - self.url = resolve_env_var(&self.url); self.prepare_batch()?; self.name = self.name.trim().to_string(); @@ -823,6 +821,7 @@ impl ConfigInput { match csv_read_inputs(input_type.clone(), &self.url) { Ok(mut batch_aliases) => { if !batch_aliases.is_empty() { + batch_aliases.reverse(); if let Some(mut first) = batch_aliases.pop() { self.username = first.username.take(); self.password = first.password.take(); @@ -832,6 +831,7 @@ impl ConfigInput { } } if !batch_aliases.is_empty() { + batch_aliases.reverse(); if let Some(aliases) = self.aliases.as_mut() { aliases.extend(batch_aliases); } else { @@ -1053,14 +1053,7 @@ pub struct WebAuthConfig { } impl WebAuthConfig { - pub fn prepare(&mut self, config_path: &str, resolve_var: bool) -> Result<(), M3uFilterError> { - if resolve_var { - self.issuer = config_reader::resolve_env_var(&self.issuer); - self.secret = config_reader::resolve_env_var(&self.secret); - if let Some(file) = &self.userfile { - self.userfile = Some(config_reader::resolve_env_var(file)); - } - } + pub fn prepare(&mut self, config_path: &str) -> Result<(), M3uFilterError> { let userfile_name = self.userfile.as_ref().map_or_else(|| file_utils::get_default_user_file_path(config_path), std::borrow::ToOwned::to_owned); self.userfile = Some(userfile_name.clone()); @@ -1128,13 +1121,13 @@ pub struct CacheConfig { } impl CacheConfig { - fn prepare(&mut self, working_dir: &str, resolve_var: bool) { + fn prepare(&mut self, working_dir: &str) { if self.enabled { let work_path = PathBuf::from(working_dir); if self.dir.is_none() { self.dir = Some(work_path.join("cache").to_string_lossy().to_string()); } else { - let mut cache_dir = if resolve_var { config_reader::resolve_env_var(self.dir.as_ref().unwrap()) } else { self.dir.as_ref().unwrap().to_string() }; + let mut cache_dir = self.dir.as_ref().unwrap().to_string(); if PathBuf::from(&cache_dir).is_relative() { cache_dir = work_path.join(&cache_dir).clean().to_string_lossy().to_string(); } @@ -1203,7 +1196,7 @@ pub struct ReverseProxyConfig { } impl ReverseProxyConfig { - fn prepare(&mut self, working_dir: &str, resolve_var: bool) { + fn prepare(&mut self, working_dir: &str) { if let Some(stream) = self.stream.as_mut() { stream.prepare(); } @@ -1212,7 +1205,7 @@ impl ReverseProxyConfig { warn!("The cache is disabled because resource rewrite is disabled"); cache.enabled = false; } - cache.prepare(working_dir, resolve_var); + cache.prepare(working_dir); } } } @@ -1517,44 +1510,38 @@ impl Config { Ok(()) } - pub fn prepare(&mut self, resolve_var: bool) -> Result<(), M3uFilterError> { - let work_dir = if resolve_var { &config_reader::resolve_env_var(&self.working_dir) } else { &self.working_dir }; + pub fn prepare(&mut self) -> Result<(), M3uFilterError> { + let work_dir = &self.working_dir; self.working_dir = file_utils::get_working_path(work_dir); - self.prepare_custom_stream_response(resolve_var); - self.prepare_directories(resolve_var); + self.prepare_custom_stream_response(); + self.prepare_directories(); if let Some(reverse_proxy) = self.reverse_proxy.as_mut() { - reverse_proxy.prepare(&self.working_dir, resolve_var); + reverse_proxy.prepare(&self.working_dir); } self.prepare_hdhomerun()?; self.api.prepare(); - self.prepare_api_web_root(resolve_var); + self.prepare_api_web_root(); self.prepare_templates()?; - self.check_unique_input_names()?; + self.prepare_sources()?; let target_names = self.check_unique_target_names()?; self.check_scheduled_targets(&target_names)?; - self.prepare_sources()?; + self.check_unique_input_names()?; self.prepare_video_config()?; - self.prepare_web_auth(resolve_var)?; + self.prepare_web_auth()?; Ok(()) } - fn prepare_directories(&mut self, resolve_var: bool) { - fn set_directory(path: &mut Option, default_subdir: &str, working_dir: &str, resolve_var: bool) { + fn prepare_directories(&mut self) { + fn set_directory(path: &mut Option, default_subdir: &str, working_dir: &str) { *path = Some(match path.as_ref() { - Some(existing) => { - if resolve_var { - resolve_env_var(existing) - } else { - existing.to_owned() - } - } + Some(existing) => existing.to_owned(), None => PathBuf::from(working_dir).join(default_subdir).clean().to_string_lossy().to_string(), }); } - set_directory(&mut self.backup_dir, "backup", &self.working_dir, resolve_var); - set_directory(&mut self.user_config_dir, "user_config", &self.working_dir, resolve_var); + set_directory(&mut self.backup_dir, "backup", &self.working_dir); + set_directory(&mut self.user_config_dir, "user_config", &self.working_dir); } fn prepare_hdhomerun(&mut self) -> Result<(), M3uFilterError> { @@ -1599,14 +1586,14 @@ impl Config { Ok(()) } - fn prepare_web_auth(&mut self, resolve_var: bool) -> Result<(), M3uFilterError> { + fn prepare_web_auth(&mut self) -> Result<(), M3uFilterError> { if !self.web_ui_enabled { self.web_auth = None; } if let Some(web_auth) = &mut self.web_auth { if web_auth.enabled { - web_auth.prepare(&self.t_config_path, resolve_var)?; + web_auth.prepare(&self.t_config_path)?; } else { self.web_auth = None; } @@ -1633,11 +1620,11 @@ impl Config { Ok(()) } - fn prepare_custom_stream_response(&mut self, resolve_var: bool) { + fn prepare_custom_stream_response(&mut self) { if let Some(custom_stream_response) = self.custom_stream_response.as_ref() { - fn load_and_set_file(path: Option<&String>, working_dir: &str, resolve_var: bool) -> Option>> { + fn load_and_set_file(path: Option<&String>, working_dir: &str) -> Option>> { path.as_ref() - .map(|file| file_utils::make_absolute_path(file, working_dir, resolve_var)) + .map(|file| file_utils::make_absolute_path(file, working_dir)) .and_then(|absolute_path| match file_utils::read_file_as_bytes(&PathBuf::from(&absolute_path)) { Ok(data) => Some(Arc::new(data)), Err(err) => { @@ -1647,15 +1634,15 @@ impl Config { }) } - self.t_channel_unavailable_video = load_and_set_file(custom_stream_response.channel_unavailable.as_ref(), &self.working_dir, resolve_var); - self.t_user_connections_exhausted_video = load_and_set_file(custom_stream_response.user_connections_exhausted.as_ref(), &self.working_dir, resolve_var); - self.t_provider_connections_exhausted_video = load_and_set_file(custom_stream_response.provider_connections_exhausted.as_ref(), &self.working_dir, resolve_var); + self.t_channel_unavailable_video = load_and_set_file(custom_stream_response.channel_unavailable.as_ref(), &self.working_dir); + self.t_user_connections_exhausted_video = load_and_set_file(custom_stream_response.user_connections_exhausted.as_ref(), &self.working_dir); + self.t_provider_connections_exhausted_video = load_and_set_file(custom_stream_response.provider_connections_exhausted.as_ref(), &self.working_dir); } } - fn prepare_api_web_root(&mut self, resolve_var: bool) { + fn prepare_api_web_root(&mut self) { if !self.api.web_root.is_empty() { - self.api.web_root = file_utils::make_absolute_path(&self.api.web_root, &self.working_dir, resolve_var); + self.api.web_root = file_utils::make_absolute_path(&self.api.web_root, &self.working_dir); } } diff --git a/src/utils/file/config_reader.rs b/src/utils/file/config_reader.rs index d3167ad53..18e6d8803 100644 --- a/src/utils/file/config_reader.rs +++ b/src/utils/file/config_reader.rs @@ -2,7 +2,10 @@ use crate::m3u_filter_error::{create_m3u_filter_error, create_m3u_filter_error_r use crate::model::api_proxy::ApiProxyConfig; use crate::model::config::{Config, ConfigDto, ConfigInputAlias, InputType}; use crate::model::mapping::Mappings; +use crate::utils::file::env_resolving_reader::EnvResolvingReader; +use crate::utils::file::file_utils::file_reader; use crate::utils::file::{file_utils, multi_file_reader}; +use crate::utils::network::request::get_credentials_from_url; use crate::utils::sys_utils::exit; use chrono::Local; use log::{debug, error, info, warn}; @@ -10,16 +13,39 @@ use regex::Regex; use serde::Serialize; use std::env; use std::fs::File; -use std::io::{self, BufRead}; +use std::io::{self, BufRead, BufReader, Read}; use std::path::PathBuf; use std::sync::LazyLock; -use url::{Url}; -use crate::utils::network::request::get_credentials_from_url; +use url::Url; -pub fn read_mappings(args_mapping: Option, cfg: &mut Config) -> Result, M3uFilterError> { +enum EitherReader { + Left(L), + Right(R), +} + +// `Read`-Trait für Either implementieren +impl Read for EitherReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + match self { + EitherReader::Left(reader) => reader.read(buf), + EitherReader::Right(reader) => reader.read(buf), + } + } +} + +pub fn config_file_reader(file: File, resolve_env: bool) -> impl Read +{ + if resolve_env { + EitherReader::Left(EnvResolvingReader::new(file_reader(file))) + } else { + EitherReader::Right(BufReader::new(file)) + } +} + +pub fn read_mappings(args_mapping: Option, cfg: &mut Config, resolve_env: bool) -> Result, M3uFilterError> { let mappings_file: String = args_mapping.unwrap_or_else(|| file_utils::get_default_mappings_path(cfg.t_config_path.as_str())); - match read_mapping(mappings_file.as_str()) { + match read_mapping(mappings_file.as_str(), resolve_env) { Ok(mappings) => { match mappings { None => { @@ -55,13 +81,13 @@ pub async fn read_api_proxy_config(args_api_proxy_config: Option, cfg: & pub fn read_config(config_path: &str, config_file: &str, sources_file: &str) -> Result { let files = vec![std::path::PathBuf::from(config_file), std::path::PathBuf::from(sources_file)]; match multi_file_reader::MultiFileReader::new(&files) { - Ok(file) => { - match serde_yaml::from_reader::<_, Config>(file) { + Ok(reader) => { + match serde_yaml::from_reader::<_, Config>(reader) { Ok(mut result) => { result.t_config_path = config_path.to_string(); result.t_config_file_path = config_file.to_string(); result.t_sources_file_path = sources_file.to_string(); - match result.prepare(true) { + match result.prepare() { Err(err) => Err(err), _ => Ok(result), } @@ -75,10 +101,10 @@ pub fn read_config(config_path: &str, config_file: &str, sources_file: &str) -> } } -pub fn read_mapping(mapping_file: &str) -> Result, M3uFilterError> { +pub fn read_mapping(mapping_file: &str, resolve_var: bool) -> Result, M3uFilterError> { let mapping_file = std::path::PathBuf::from(mapping_file); if let Ok(file) = file_utils::open_file(&mapping_file) { - let mapping: Result = serde_yaml::from_reader(file); + let mapping: Result = serde_yaml::from_reader(config_file_reader(file, resolve_var)); match mapping { Ok(mut result) => { handle_m3u_filter_error_result!(M3uFilterErrorKind::Info, result.prepare()); @@ -93,12 +119,12 @@ pub fn read_mapping(mapping_file: &str) -> Result, M3uFilterErr Ok(None) } -pub fn read_api_proxy(config: &Config, api_proxy_file: &str, resolve_var: bool) -> Option { +pub fn read_api_proxy(config: &Config, api_proxy_file: &str, resolve_env: bool) -> Option { file_utils::open_file(&std::path::PathBuf::from(api_proxy_file)).map_or(None, |file| { - let mapping: Result = serde_yaml::from_reader(file); + let mapping: Result = serde_yaml::from_reader(config_file_reader(file, resolve_env)); match mapping { Ok(mut result) => { - match result.prepare(config, resolve_var) { + match result.prepare(config) { Err(err) => { exit!("cant read api-proxy-config file: {}", err); } @@ -155,16 +181,6 @@ pub fn resolve_env_var(value: &str) -> String { }).to_string() } -// -// pub fn read_yaml_inputs(input_type: InputType, file_path: &str) -> Result { -// -// } -// -// -// pub fn read_json_inputs(input_type: InputType, file_path: &str) -> Result { -// -// } - const CSV_SEPARATOR: char = ';'; const HEADER_PREFIX: char = '#'; const FIELD_MAX_CON: &str = "max_connections"; @@ -174,23 +190,37 @@ const FIELD_NAME: &str = "name"; const FIELD_USERNAME: &str = "username"; const FIELD_PASSWORD: &str = "password"; const FIELD_UNKNOWN: &str = "?"; +const DEFAULT_COLUMNS: &[&str] = &[FIELD_URL, FIELD_MAX_CON, FIELD_PRIO, FIELD_NAME, FIELD_USERNAME, FIELD_PASSWORD]; -fn csv_assign_mandatory_fields(alias: &mut ConfigInputAlias) { - if alias.name.is_empty() && !alias.url.is_empty() { - match Url::parse(alias.url.as_str()) { +fn csv_assign_mandatory_fields(alias: &mut ConfigInputAlias, input_type: &InputType) { + if !alias.url.is_empty() { + match Url::parse(alias.url.as_str()) { Ok(url) => { let (username, password) = get_credentials_from_url(&url); if username.is_none() || password.is_none() { // xtream url - alias.url = url.origin().ascii_serialization().to_string(); + if input_type == &InputType::XtreamBatch { + alias.url = url.origin().ascii_serialization().to_string(); + } else if input_type == &InputType::M3uBatch { + if alias.username.is_some() && alias.password.is_some() { + alias.url = format!("{}/get_php?username={}&password={}&type=m3u_plus", + url.origin().ascii_serialization().to_string(), + alias.username.as_deref().unwrap_or("").to_string(), + alias.password.as_deref().unwrap_or("").to_string() + ) + } + } } else { + if input_type == &InputType::XtreamBatch { + alias.url = url.origin().ascii_serialization().to_string(); + } // m3u url alias.username = username; alias.password = password; } if alias.name.is_empty() { - let username = alias.username.as_ref().map(|s|s.as_str()).unwrap_or_default(); + let username = alias.username.as_ref().map(|s| s.as_str()).unwrap_or_default(); let domain: Vec<&str> = url.domain().unwrap_or_default().split('.').collect(); if domain.len() > 1 { alias.name = format!("{}_{username}", domain[domain.len() - 2]); @@ -204,96 +234,166 @@ fn csv_assign_mandatory_fields(alias: &mut ConfigInputAlias) { } } -fn csv_assign_config_input_column(config_input: &mut ConfigInputAlias, input_type: &InputType, header: &str, value: &str) -> Result<(), io::Error> { +fn csv_assign_config_input_column(config_input: &mut ConfigInputAlias, header: &str, raw_value: &str) -> Result<(), io::Error> { + let value = raw_value.trim(); + if !value.is_empty() { match header { FIELD_URL => { let url = Url::parse(value.trim()).map_err(to_io_error)?; - match input_type { - InputType::Xtream | InputType::M3u => { - config_input.url = url.to_string(); - } - InputType::M3uBatch | InputType::XtreamBatch => { - } - }; - }, + config_input.url = url.to_string(); + } FIELD_MAX_CON => { let max_connections = value.parse::().unwrap_or(1); config_input.max_connections = max_connections; - }, + } FIELD_PRIO => { let priority = value.parse::().unwrap_or(0); config_input.priority = priority; - }, + } FIELD_NAME => { config_input.name = value.to_string(); - }, + } + FIELD_USERNAME => { + config_input.username = Some(value.to_string()); + } + FIELD_PASSWORD => { + config_input.password = Some(value.to_string()); + } _ => {} } + } Ok(()) } -pub fn csv_read_inputs(input_type: InputType, file_uri: &str) -> Result, io::Error> { +pub fn csv_read_inputs_from_reader(batch_input_type: InputType, reader: impl BufRead) -> Result, io::Error> { + let input_type = match batch_input_type { + InputType::M3uBatch | InputType::M3u => InputType::M3uBatch, + InputType::XtreamBatch | InputType::Xtream => InputType::XtreamBatch + }; + let mut result = vec![]; + let mut default_columns = vec![]; + default_columns.extend_from_slice(DEFAULT_COLUMNS); + for line in reader.lines().into_iter() { + let line = line?; + if line.is_empty() { + continue + } + if line.starts_with(HEADER_PREFIX) { + default_columns = line[1..].split(CSV_SEPARATOR).map(|s| { + match s { + FIELD_URL => FIELD_URL, + FIELD_MAX_CON => FIELD_MAX_CON, + FIELD_PRIO => FIELD_PRIO, + FIELD_NAME => FIELD_NAME, + FIELD_USERNAME => FIELD_USERNAME, + FIELD_PASSWORD => FIELD_PASSWORD, + _ => { + error!("Field {s} is unsupported for csv input"); + FIELD_UNKNOWN + } + } + }).collect(); + continue; + } + + let mut config_input = ConfigInputAlias { + id: 0, + name: "".to_string(), + url: "".to_string(), + username: None, + password: None, + priority: 0, + max_connections: 1, + }; + + let columns: Vec<&str> = line.split(CSV_SEPARATOR).collect(); + for (&header, &value) in default_columns.iter().zip(columns.iter()).into_iter() { + if let Err(err) = csv_assign_config_input_column(&mut config_input, header, value) { + error!("Could not parse input line: {} err: {err}", line); + continue; + } + } + csv_assign_mandatory_fields(&mut config_input, &input_type); + result.push(config_input); + } + Ok(result) +} + + +pub fn csv_read_inputs(batch_input_type: InputType, file_uri: &str) -> Result, io::Error> { let file_path = file_uri.strip_prefix("file://").unwrap_or(file_uri); let inputs_file = std::path::PathBuf::from(file_path); if let Ok(file) = file_utils::open_file(&inputs_file) { - let mut result = vec![]; - let mut default_columns = vec![FIELD_URL, FIELD_MAX_CON, FIELD_PRIO, FIELD_NAME]; - let reader = io::BufReader::new(file); - for line in reader.lines().into_iter() { - let line = line?; - if line.is_empty() || line.starts_with(HEADER_PREFIX) { - default_columns = line[1..].split(CSV_SEPARATOR).map(|s| { - println!("{s}"); - match s { - FIELD_URL => FIELD_URL, - FIELD_MAX_CON => FIELD_MAX_CON, - FIELD_PRIO => FIELD_PRIO, - FIELD_NAME => FIELD_NAME, - _ => { - error!("Field {s} is unsupported for csv input"); - FIELD_UNKNOWN - } - } - }).collect(); - println!("{default_columns:?}"); - continue; - } - - let mut config_input = ConfigInputAlias { - id: 0, - name: "".to_string(), - url: "".to_string(), - username: None, - password: None, - priority: 0, - max_connections: 1, - }; - - let columns: Vec<&str> = line.split(CSV_SEPARATOR).collect(); - for (&header, &value) in default_columns.iter().zip(columns.iter()).into_iter() { - if let Err(err) = csv_assign_config_input_column(&mut config_input, &input_type, header, value) { - error!("Could not parse input line: {} err: {err}", line); - continue; - } - } - csv_assign_mandatory_fields(&mut config_input); - result.push(config_input); - } - return Ok(result); + return csv_read_inputs_from_reader(batch_input_type, EnvResolvingReader::new(file_reader(file))); } Err(str_to_io_error(&format!("Could not open {file_path}"))) } #[cfg(test)] mod tests { + use std::io::{BufReader, Cursor}; use crate::model::config::InputType; - use crate::utils::file::config_reader::{csv_read_inputs, resolve_env_var}; + use crate::utils::file::config_reader::{csv_read_inputs_from_reader, resolve_env_var}; +const M3U_BATCH: &str = r#" +#url;name;max_connections;priority +http://hd.providerline.com:8080/get.php?username=user1&password=user1&type=m3u_plus;input_1 +http://hd.providerline.com/get.php?username=user2&password=user2&type=m3u_plus;input_2;1;2 +http://hd.providerline.com/get.php?username=user3&password=user3&type=m3u_plus;input_3;1;2 +http://hd.providerline.com/get.php?username=user4&password=user4&type=m3u_plus;input_4 +"#; + +const XTREAM_BATCH: &str = r#" +#name;username;password;url;max_connections +input_1;desanocra;eyCG8SN523KQ;http://provider_1.tv:80;1 +input_2;desanocra;eyCG8SN523KQ;http://provider_2.tv:8080;1 +"#; #[test] - fn test_read_inputs() { - let result = csv_read_inputs(InputType::Xtream, &resolve_env_var("${env:HOME}/projects/m3u-test/settings/inputs.txt")); + fn test_read_inputs_xtream_as_m3u() { + let reader = BufReader::new(Cursor::new(XTREAM_BATCH)); + let result = csv_read_inputs_from_reader(InputType::M3uBatch, reader); assert_eq!(result.is_ok(), true); - println!("result: {:?}", result.unwrap()); + let aliases = result.unwrap(); + assert_eq!(aliases.is_empty(), false); + for config in aliases { + assert_eq!(config.url.contains("username"), true); + } + } + + #[test] + fn test_read_inputs_m3u_as_m3u() { + let reader = BufReader::new(Cursor::new(M3U_BATCH)); + let result = csv_read_inputs_from_reader(InputType::M3uBatch, reader); + assert_eq!(result.is_ok(), true); + let aliases = result.unwrap(); + assert_eq!(aliases.is_empty(), false); + for config in aliases { + assert_eq!(config.url.contains("username"), true); + } + } + + #[test] + fn test_read_inputs_xtream_as_xtream() { + let reader = BufReader::new(Cursor::new(XTREAM_BATCH)); + let result = csv_read_inputs_from_reader(InputType::XtreamBatch, reader); + assert_eq!(result.is_ok(), true); + let aliases = result.unwrap(); + assert_eq!(aliases.is_empty(), false); + for config in aliases { + assert_eq!(config.url.contains("username"), false); + } + } + + #[test] + fn test_read_inputs_m3u_as_xtream() { + let reader = BufReader::new(Cursor::new(M3U_BATCH)); + let result = csv_read_inputs_from_reader(InputType::XtreamBatch, reader); + assert_eq!(result.is_ok(), true); + let aliases = result.unwrap(); + assert_eq!(aliases.is_empty(), false); + for config in aliases { + assert_eq!(config.url.contains("username"), false); + } } #[test] diff --git a/src/utils/file/env_resolving_reader.rs b/src/utils/file/env_resolving_reader.rs new file mode 100644 index 000000000..2242647e1 --- /dev/null +++ b/src/utils/file/env_resolving_reader.rs @@ -0,0 +1,52 @@ +use std::io::{self, BufRead, BufReader, Read, Cursor}; +use crate::utils::file::config_reader::resolve_env_var; + +pub struct EnvResolvingReader { + inner: BufReader, + buffer: Cursor>, +} + +impl EnvResolvingReader { + pub(crate) fn new(reader: BufReader) -> Self { + Self { + inner: reader, + buffer: Cursor::new(Vec::new()), + } + } + + fn fill_buffer(&mut self) -> io::Result<()> { + let mut line = String::new(); + self.buffer = Cursor::new(Vec::new()); + + if self.inner.read_line(&mut line)? > 0 { + let processed_line = resolve_env_var(&line); + self.buffer = Cursor::new(processed_line.into_bytes()); + } + + Ok(()) + } +} + +impl Read for EnvResolvingReader { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + let bytes_read = self.buffer.read(buf)?; + + if bytes_read == 0 { + self.fill_buffer()?; + self.buffer.read(buf) + } else { + Ok(bytes_read) + } + } +} + +impl BufRead for EnvResolvingReader { + fn fill_buf(&mut self) -> io::Result<&[u8]> { + self.fill_buffer()?; + self.buffer.fill_buf() + } + + fn consume(&mut self, amt: usize) { + self.buffer.consume(amt); + } +} \ No newline at end of file diff --git a/src/utils/file/file_utils.rs b/src/utils/file/file_utils.rs index f85e73110..970526f50 100644 --- a/src/utils/file/file_utils.rs +++ b/src/utils/file/file_utils.rs @@ -6,7 +6,6 @@ use std::path::{Path, PathBuf}; use crate::m3u_filter_error::str_to_io_error; use crate::utils::debug_if_enabled; -use crate::utils::file::{config_reader}; use log::{debug, error}; use path_clean::PathClean; @@ -262,10 +261,8 @@ pub fn read_file_as_bytes(path: &Path) -> std::io::Result> { Ok(buffer) } -pub fn make_absolute_path(path: &str, working_dir: &str, resolve_var: bool) -> String { - let resolved_path = if resolve_var { config_reader::resolve_env_var(path) } else { path.to_string() }; - - let rpb = std::path::PathBuf::from(&resolved_path); +pub fn make_absolute_path(path: &str, working_dir: &str) -> String { + let rpb = std::path::PathBuf::from(path); if rpb.is_relative() { let mut rpb2 = std::path::PathBuf::from(working_dir).join(&rpb); if !rpb2.exists() { @@ -281,5 +278,5 @@ pub fn make_absolute_path(path: &str, working_dir: &str, resolve_var: bool) -> S return String::from(rpb2.clean().to_str().unwrap_or_default()); } } - resolved_path + path.to_string() } \ No newline at end of file diff --git a/src/utils/file/mod.rs b/src/utils/file/mod.rs index c26f5ccfd..ffdbe06f8 100644 --- a/src/utils/file/mod.rs +++ b/src/utils/file/mod.rs @@ -1,4 +1,5 @@ pub mod file_utils; pub mod multi_file_reader; pub mod file_lock_manager; -pub mod config_reader; \ No newline at end of file +pub mod config_reader; +pub mod env_resolving_reader; \ No newline at end of file diff --git a/src/utils/file/multi_file_reader.rs b/src/utils/file/multi_file_reader.rs index da4a0bdac..705f5407e 100644 --- a/src/utils/file/multi_file_reader.rs +++ b/src/utils/file/multi_file_reader.rs @@ -1,11 +1,12 @@ use std::fs::File; -use std::io::{self, BufReader, ErrorKind, Read}; +use std::io::{self, ErrorKind, Read}; use std::path::{PathBuf}; +use crate::utils::file::env_resolving_reader::EnvResolvingReader; use crate::utils::file::file_utils::file_reader; pub struct MultiFileReader { files: Vec, - current_reader: Option>, + current_reader: Option>, } impl MultiFileReader { @@ -34,7 +35,7 @@ impl Read for MultiFileReader { if self.files.is_empty() { return Ok(0); } - self.current_reader = Some(file_reader(self.files.pop().unwrap())); + self.current_reader = Some(EnvResolvingReader::new(file_reader(self.files.pop().unwrap()))); // we put a newline if the config does not have one if !buf.is_empty() && buf[0] != b'\n' { buf[0] = b'\n';