euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-29 02:02:23 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

Env variable resolving is now done for complete config files.

Browse Source 
Batch input from files added
This commit is contained in:
euzu
2025-03-22 14:24:42 +01:00
parent 392f4cfca0
commit db241fbd07
11 changed files with 373 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+38
View File
@@ -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.
README.md
+39
View File
@@ -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
src/api/endpoints/v1_api.rs
+2 -2
View File
@@ -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)
src/main.rs
+4 -3
View File
@@ -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<String>, 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<String>, 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::<Healthcheck>() {
src/model/api_proxy.rs
+6 -13
View File
@@ -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<String>) {
fn prepare_target_user(&mut self, errors: &mut Vec<String>) {
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(())
src/model/config.rs
+34 -47
View File
@@ -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<u16, M3uFilterError> {
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<String>, default_subdir: &str, working_dir: &str, resolve_var: bool) {
fn prepare_directories(&mut self) {
fn set_directory(path: &mut Option<String>, 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<Arc<Vec<u8>>> {
fn load_and_set_file(path: Option<&String>, working_dir: &str) -> Option<Arc<Vec<u8>>> {
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);
}
}
src/utils/file/config_reader.rs
+189 -89
View File
@@ -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<String>, cfg: &mut Config) -> Result<Option<String>, M3uFilterError> {
enum EitherReader<L, R> {
Left(L),
Right(R),
}
// `Read`-Trait für Either implementieren
impl<L: Read, R: Read> Read for EitherReader<L, R> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
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<String>, cfg: &mut Config, resolve_env: bool) -> Result<Option<String>, 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<String>, cfg: &
pub fn read_config(config_path: &str, config_file: &str, sources_file: &str) -> Result<Config, M3uFilterError> {
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<Option<Mappings>, M3uFilterError> {
pub fn read_mapping(mapping_file: &str, resolve_var: bool) -> Result<Option<Mappings>, M3uFilterError> {
let mapping_file = std::path::PathBuf::from(mapping_file);
if let Ok(file) = file_utils::open_file(&mapping_file) {
let mapping: Result<Mappings, _> = serde_yaml::from_reader(file);
let mapping: Result<Mappings, _> = 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<Option<Mappings>, M3uFilterErr
Ok(None)
}
pub fn read_api_proxy(config: &Config, api_proxy_file: &str, resolve_var: bool) -> Option<ApiProxyConfig> {
pub fn read_api_proxy(config: &Config, api_proxy_file: &str, resolve_env: bool) -> Option<ApiProxyConfig> {
file_utils::open_file(&std::path::PathBuf::from(api_proxy_file)).map_or(None, |file| {
let mapping: Result<ApiProxyConfig, _> = serde_yaml::from_reader(file);
let mapping: Result<ApiProxyConfig, _> = 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<Config, M3uFilterError> {
//
// }
//
//
// pub fn read_json_inputs(input_type: InputType, file_path: &str) -> Result<Config, M3uFilterError> {
//
// }
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::<u16>().unwrap_or(1);
config_input.max_connections = max_connections;
},
}
FIELD_PRIO => {
let priority = value.parse::<i16>().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<Vec<ConfigInputAlias>, io::Error> {
pub fn csv_read_inputs_from_reader(batch_input_type: InputType, reader: impl BufRead) -> Result<Vec<ConfigInputAlias>, 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<Vec<ConfigInputAlias>, 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]
src/utils/file/env_resolving_reader.rs
+52
View File
@@ -0,0 +1,52 @@
use std::io::{self, BufRead, BufReader, Read, Cursor};
use crate::utils::file::config_reader::resolve_env_var;
pub struct EnvResolvingReader<R: Read> {
inner: BufReader<R>,
buffer: Cursor<Vec<u8>>,
}
impl<R: Read> EnvResolvingReader<R> {
pub(crate) fn new(reader: BufReader<R>) -> 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<R: Read> Read for EnvResolvingReader<R> {
fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
let bytes_read = self.buffer.read(buf)?;
if bytes_read == 0 {
self.fill_buffer()?;
self.buffer.read(buf)
} else {
Ok(bytes_read)
}
}
}
impl<R: Read> BufRead for EnvResolvingReader<R> {
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);
}
}
src/utils/file/file_utils.rs
+3 -6
View File
@@ -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<Vec<u8>> {
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()
}
src/utils/file/mod.rs
+2 -1
View File
@@ -1,4 +1,5 @@
pub mod file_utils;
pub mod multi_file_reader;
pub mod file_lock_manager;
pub mod config_reader;
pub mod config_reader;
pub mod env_resolving_reader;
src/utils/file/multi_file_reader.rs
+4 -3
View File
@@ -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<File>,
current_reader: Option<BufReader<File>>,
current_reader: Option<EnvResolvingReader<File>>,
}
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';