From fac52fe69f8e06891d90a03ae2640a0f6edccd50 Mon Sep 17 00:00:00 2001 From: euzu Date: Fri, 1 Dec 2023 21:38:23 +0100 Subject: [PATCH] refactored xtream api code --- CHANGELOG.md | 2 +- src/api/api_utils.rs | 19 ++-- src/api/download_api.rs | 53 ++++----- src/api/scheduler.rs | 24 ++-- src/api/xtream_player_api.rs | 167 +++++++++++++++++++++------- src/config_reader.rs | 2 +- src/download.rs | 4 +- src/main.rs | 2 +- src/processing/playlist_watch.rs | 4 +- src/repository/xtream_repository.rs | 152 ++++++++++--------------- src/utils.rs | 26 +++-- 11 files changed, 262 insertions(+), 193 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 186ab0629..47e08f207 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,7 +2,7 @@ # v1.1.4(2023-11-??) * Added regexp search in Web-UI * Added config Web-UI -* Added xtream vod_info and series_info +* Added xtream vod_info and series_info, stream seek. * Added input options with attribute xtream_info_cache to cache get_vod_info and get_series_info on disc # v1.1.3(2023-11-08) diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index 18997a40b..69cb688b4 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -7,21 +7,20 @@ use crate::model::config::ConfigTarget; pub(crate) async fn serve_file(file_path: &Path, req: &HttpRequest) -> HttpResponse { if file_path.exists() { - let file = actix_files::NamedFile::open_async(file_path).await.unwrap() - .set_content_type(mime::APPLICATION_JSON) - .disable_content_disposition(); - let mut result = file.into_response(req); - let headers = result.headers_mut(); - headers.insert(CACHE_CONTROL, HeaderValue::from_bytes("no-cache".as_bytes()).unwrap()); - result - } else { - HttpResponse::NoContent().finish() + if let Ok(file) = actix_files::NamedFile::open_async(file_path).await { + let mut result = file.set_content_type(mime::APPLICATION_JSON) + .disable_content_disposition().into_response(req); + let headers = result.headers_mut(); + headers.insert(CACHE_CONTROL, HeaderValue::from_bytes("no-cache".as_bytes()).unwrap()); + return result; + } } + HttpResponse::NoContent().finish() } pub(crate) fn get_user_target_by_credentials<'a>(username: &str, password: &str, api_req: &'a web::Query, app_state: &'a web::Data) -> Option<(UserCredentials, &'a ConfigTarget)> { if !username.is_empty() && !password.is_empty() { - app_state.config.get_target_for_user(username, password) + app_state.config.get_target_for_user(username, password) } else { let token = api_req.token.as_str().trim(); if !token.is_empty() { diff --git a/src/api/download_api.rs b/src/api/download_api.rs index 77b424cbf..63b077851 100644 --- a/src/api/download_api.rs +++ b/src/api/download_api.rs @@ -17,38 +17,41 @@ async fn download_file(active: Arc>>, client: &reqwe Ok(response) => { match fs::create_dir_all(&file_download.file_dir) { Ok(_) => { - let file_path_str = file_download.file_path.to_str().unwrap(); - info!("Downloading {}", file_path_str); - match File::create(&file_download.file_path) { - Ok(mut file) => { - let mut downloaded: u64 = 0; - let mut stream = response.bytes_stream().map_err(|err| io::Error::new(ErrorKind::Other, err)); - loop { - match stream.try_next().await { - Ok(item) => { - match item { - Some(chunk) => { - match file.write_all(&chunk) { - Ok(_) => { - downloaded += chunk.len() as u64; - active.write().unwrap().as_mut().unwrap().size = downloaded; + if let Some(file_path_str) = file_download.file_path.to_str() { + info!("Downloading {}", file_path_str); + match File::create(&file_download.file_path) { + Ok(mut file) => { + let mut downloaded: u64 = 0; + let mut stream = response.bytes_stream().map_err(|err| io::Error::new(ErrorKind::Other, err)); + loop { + match stream.try_next().await { + Ok(item) => { + match item { + Some(chunk) => { + match file.write_all(&chunk) { + Ok(_) => { + downloaded += chunk.len() as u64; + active.write().unwrap().as_mut().unwrap().size = downloaded; + } + Err(err) => return Err(format!("Error while writing to file: {} {}", file_path_str, err)) } - Err(err) => return Err(format!("Error while writing to file: {} {}", file_path_str, err)) + } + None => { + let megabytes = bytes_to_megabytes(downloaded); + info!("Downloaded {}, filesize: {}MB", file_path_str, megabytes); + active.write().unwrap().as_mut().unwrap().size = downloaded; + return Ok(()); } } - None => { - let megabytes = bytes_to_megabytes(downloaded); - info!("Downloaded {}, filesize: {}MB", file_path_str, megabytes); - active.write().unwrap().as_mut().unwrap().size = downloaded; - return Ok(()); - } } + Err(err) => return Err(format!("Error while writing to file: {} {}", file_path_str, err)) } - Err(err) => return Err(format!("Error while writing to file: {} {}", file_path_str, err)) } } + Err(err) => Err(format!("Error while writing to file: {} {}", file_path_str, err)) } - Err(err) => Err(format!("Error while writing to file: {} {}", file_path_str, err)) + } else { + Err("Error file-download file-path unknown".to_string()) } } Err(err) => Err(format!("Error while creating directory for file: {} {}", &file_download.file_dir.to_str().unwrap_or("?"), err)) @@ -64,7 +67,7 @@ fn run_download_queue(download_cfg: &VideoDownloadConfig, download_queue: Arc { diff --git a/src/api/scheduler.rs b/src/api/scheduler.rs index 88c2dc9d8..4c0d0c6e1 100644 --- a/src/api/scheduler.rs +++ b/src/api/scheduler.rs @@ -3,21 +3,27 @@ use std::time::Duration; use actix_web::web::Data; use chrono::Local; use cron::Schedule; +use log::error; use crate::api::api_model::AppState; +use crate::exit; use crate::processing::playlist_processor::exec_processing; pub(crate) async fn start_scheduler(expression: &str, data: Data) -> ! { - let schedule = Schedule::from_str(expression).unwrap(); - let offset = *Local::now().offset(); - loop { - let mut upcoming = schedule.upcoming(offset).take(1); - actix_rt::time::sleep(Duration::from_millis(500)).await; - let local = &Local::now(); + match Schedule::from_str(expression) { + Ok(schedule) => { + let offset = *Local::now().offset(); + loop { + let mut upcoming = schedule.upcoming(offset).take(1); + actix_rt::time::sleep(Duration::from_millis(500)).await; + let local = &Local::now(); - if let Some(datetime) = upcoming.next() { - if datetime.timestamp() <= local.timestamp() { - exec_processing(data.config.clone(), data.targets.clone()).await; + if let Some(datetime) = upcoming.next() { + if datetime.timestamp() <= local.timestamp() { + exec_processing(data.config.clone(), data.targets.clone()).await; + } + } } } + Err(err) => exit!("Failed to start scheduler: {}", err) } } \ No newline at end of file diff --git a/src/api/xtream_player_api.rs b/src/api/xtream_player_api.rs index 49f495157..8e29cd33f 100644 --- a/src/api/xtream_player_api.rs +++ b/src/api/xtream_player_api.rs @@ -1,21 +1,50 @@ // https://github.com/tellytv/go.xtream-codes/blob/master/structs.go +use std::collections::HashMap; use std::io::{Error}; use std::str::FromStr; use actix_web::{HttpRequest, HttpResponse, web, Resource}; use chrono::{Duration, Local}; -use log::{debug}; +use log::{debug, error}; +use url::{Url}; use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, serve_file}; use crate::api::api_model::{AppState, UserApiRequest, XtreamAuthorizationResponse, XtreamServerInfo, XtreamUserInfo}; use crate::model::api_proxy::{UserCredentials}; -use crate::model::config::{Config, InputType}; +use crate::model::config::{Config, ConfigInput, InputType}; use crate::model::model_config::{TargetType}; use crate::model::model_m3u::XtreamCluster; use crate::repository::xtream_repository::{COL_CAT_LIVE, COL_CAT_SERIES, COL_CAT_VOD, COL_LIVE, COL_SERIES, COL_VOD, - xtream_get_all, - xtream_get_short_epg, xtream_get_stream_info}; -use crate::utils::get_client_request; + xtream_get_all, xtream_get_stored_stream_info, xtream_persist_stream_info}; +use crate::utils::{get_client_request}; + + +fn get_xtream_player_api_action_url(input: &ConfigInput, action: &str) -> Option { + match input.input_type { + InputType::M3u => None, + InputType::Xtream => Some( + format!("{}/player_api.php?username={}&password={}&action={}", + input.url.as_str(), + input.username.as_ref().unwrap_or(&"".to_string()).as_str(), + input.password.as_ref().unwrap_or(&"".to_string()).as_str(), + action + )) + } +} + +fn get_xtream_player_api_stream_url(input: &ConfigInput, context: &str, action_path: &str) -> Option { + match input.input_type { + InputType::M3u => None, + InputType::Xtream => Some(format!("{}/{}/{}/{}/{}", + input.url.as_str(), + context, + input.username.as_ref().unwrap_or(&"".to_string()).as_str(), + input.password.as_ref().unwrap_or(&"".to_string()).as_str(), + action_path + )) + } +} + fn get_user_info(user: &UserCredentials, cfg: &Config) -> XtreamAuthorizationResponse { let server = cfg._api_proxy.read().unwrap().as_ref().unwrap().server.clone(); @@ -48,6 +77,7 @@ fn get_user_info(user: &UserCredentials, cfg: &Config) -> XtreamAuthorizationRes } async fn xtream_player_api_stream( + req: &HttpRequest, api_req: &web::Query, _app_state: &web::Data, context: &str, @@ -58,33 +88,26 @@ async fn xtream_player_api_stream( if let Some((_user, target)) = get_user_target_by_credentials(username, password, api_req, _app_state) { let target_name = &target.name; if target.has_output(&TargetType::Xtream) { - if let Some(input) = match _app_state.config.get_input_for_target(target_name, &InputType::Xtream) { + if let Some(target_input) = match _app_state.config.get_input_for_target(target_name, &InputType::Xtream) { None => _app_state.config.get_input_for_target(target_name, &InputType::M3u), Some(inp) => Some(inp) } { - let (uname, passwd, base_url) = match &input.input_type { - InputType::M3u => (username, password, "/"), - InputType::Xtream => ( - input.username.as_ref().unwrap().as_str(), - input.password.as_ref().unwrap().as_str(), - input.url.as_str() - ) - }; - - let stream_url = format!("{}/{}/{}/{}/{}", base_url, context, uname, passwd, action_path); - debug!("Try to open stream {}", &stream_url); - let url = reqwest::Url::parse(&stream_url).unwrap(); - let client = get_client_request(input, url); - if let Ok(response) = client.send().await { - if response.status().is_success() { - let mut response_builder = HttpResponse::Ok(); - for (key, value) in response.headers().iter() { - //error!("{:?} => {:?}", &key, &value); - response_builder.insert_header((key.as_str(), value)); + if let Some(stream_url) = get_xtream_player_api_stream_url(target_input, context, action_path) { + let req_headers: HashMap<&str, &[u8]> = req.headers().iter().map(|(k, v)| (k.as_str(), v.as_bytes())).collect(); + debug!("Try to open stream {}", &stream_url); + if let Ok(url) = Url::parse(&stream_url) { + let client = get_client_request(target_input, url, Some(&req_headers)); + if let Ok(response) = client.send().await { + if response.status().is_success() { + let mut response_builder = HttpResponse::Ok(); + response.headers().iter().for_each(|(k, v)| { + response_builder.insert_header((k, v)); + }); + return response_builder.body(actix_web::body::BodyStream::new(response.bytes_stream())); + } else { + debug!("Failed to open stream got status {} for {}", response.status(), &stream_url) + } } - return response_builder.body(actix_web::body::BodyStream::new(response.bytes_stream())); - } else { - debug!("Failed to open stream got status {} for {}", response.status(), &stream_url) } } } @@ -94,47 +117,50 @@ async fn xtream_player_api_stream( } async fn xtream_player_api_live_stream( + req: HttpRequest, api_req: web::Query, path: web::Path<(String, String, String)>, _app_state: web::Data, ) -> HttpResponse { let (username, password, stream_id) = path.into_inner(); - xtream_player_api_stream(&api_req, &_app_state, "live", &username, &password, &stream_id).await + xtream_player_api_stream(&req, &api_req, &_app_state, "live", &username, &password, &stream_id).await } async fn xtream_player_api_series_stream( + req: HttpRequest, api_req: web::Query, path: web::Path<(String, String, String)>, _app_state: web::Data, ) -> HttpResponse { let (username, password, stream_id) = path.into_inner(); - xtream_player_api_stream(&api_req, &_app_state, "series", &username, &password, &stream_id).await + xtream_player_api_stream(&req, &api_req, &_app_state, "series", &username, &password, &stream_id).await } async fn xtream_player_api_movie_stream( + req: HttpRequest, api_req: web::Query, path: web::Path<(String, String, String)>, _app_state: web::Data, ) -> HttpResponse { let (username, password, stream_id) = path.into_inner(); - xtream_player_api_stream(&api_req, &_app_state, "movie", &username, &password, &stream_id).await + xtream_player_api_stream(&req, &api_req, &_app_state, "movie", &username, &password, &stream_id).await } async fn xtream_player_api_timeshift_stream( + req: HttpRequest, api_req: web::Query, path: web::Path<(String, String, String, String, String)>, _app_state: web::Data, ) -> HttpResponse { let (username, password, duration, start, stream_id) = path.into_inner(); let action_path = format!("{}/{}/{}", duration, start, stream_id); - xtream_player_api_stream(&api_req, &_app_state, "timeshift", &username, &password, &action_path).await + xtream_player_api_stream(&req, &api_req, &_app_state, "timeshift", &username, &password, &action_path).await } -async fn xtream_get_stream_info_response(app_state: &AppState, target_name: &str, stream_id: &str, cluster: XtreamCluster, - user: &UserCredentials) -> HttpResponse { +async fn xtream_get_stream_info_response(app_state: &AppState, target_name: &str, stream_id: &str, cluster: &XtreamCluster) -> HttpResponse { match FromStr::from_str(stream_id) { Ok(xtream_stream_id) => { - match xtream_get_stream_info(app_state, target_name, xtream_stream_id, cluster, user).await { + match xtream_get_stream_info(app_state, target_name, xtream_stream_id, cluster).await { Ok(content) => HttpResponse::Ok().content_type(mime::APPLICATION_JSON).body(content), Err(_) => HttpResponse::NoContent().finish() } @@ -143,9 +169,72 @@ async fn xtream_get_stream_info_response(app_state: &AppState, target_name: &str } } +async fn xtream_get_stream_info(app_state: &AppState, target_name: &str, stream_id: i32, + cluster: &XtreamCluster) -> Result { + if let Some(target_input) = app_state.config.get_input_for_target(target_name, &InputType::Xtream) { + if let Ok(content) = xtream_get_stored_stream_info(app_state, target_name, stream_id, cluster, target_input).await { + return Ok(content); + } + + let (action, stream_id_field) = match cluster { + XtreamCluster::Live => ("get_live_info", "live_id"), + XtreamCluster::Video => ("get_vod_info", "vod_id"), + XtreamCluster::Series => ("get_series_info", "series_id"), + }; + + if let Some(action_url) = get_xtream_player_api_action_url(target_input, action) { + let info_url = format!("{}&{}={}", action_url, stream_id_field, stream_id); + if let Ok(url) = Url::parse(&info_url) { + let client = get_client_request(target_input, url, None); + if let Ok(response) = client.send().await { + debug!("{}", response.status()); + if response.status().is_success() { + match response.text().await { + Ok(content) => { + xtream_persist_stream_info(app_state, target_name, stream_id, cluster, + target_input, content.as_str()).await; + return Ok(content); + } + Err(err) => { error!("Failed to download info {}", err.to_string()); } + } + } + } + } + } + } + Err(Error::new(std::io::ErrorKind::Other, format!("Cant find stream with id: {}/{}/{}", target_name, &cluster, stream_id))) +} + + +async fn xtream_get_short_epg(app_state: &AppState, target_name: &str, stream_id: &str, limit: &str) -> Result { + if let Some(target_input) = app_state.config.get_input_for_target(target_name, &InputType::Xtream) { + if let Some(action_url) = get_xtream_player_api_action_url(target_input, "get_short_epg") { + let mut info_url = format!("{}&stream_id={}", action_url, stream_id); + if !(limit.is_empty() || limit.eq("0")) { + info_url = format!("{}&limit={}", info_url, limit); + } + if let Ok(url) = Url::parse(&info_url) { + let client = get_client_request(target_input, url, None); + if let Ok(response) = client.send().await { + if response.status().is_success() { + match response.text().await { + Ok(content) => { + return Ok(content); + } + Err(err) => { error!("Failed to download epg {}", err.to_string()); } + } + } + } + } + } + } + Err(Error::new(std::io::ErrorKind::Other, format!("Cant find short epg with id: {}/{}", target_name, stream_id))) +} + + async fn xtream_player_api( - api_req: web::Query, req: HttpRequest, + api_req: web::Query, _app_state: web::Data, ) -> HttpResponse { match get_user_target(&api_req, &_app_state) { @@ -161,15 +250,15 @@ async fn xtream_player_api( "get_series_info" => { xtream_get_stream_info_response(&_app_state, target_name, api_req.series_id.trim(), - XtreamCluster::Series, &user).await + &XtreamCluster::Series).await } "get_vod_info" => { xtream_get_stream_info_response(&_app_state, target_name, api_req.vod_id.trim(), - XtreamCluster::Video, &user).await + &XtreamCluster::Video).await } "get_short_epg" => { - match xtream_get_short_epg(&_app_state, target_name, api_req.stream_id.trim(), api_req.limit.trim(), &user).await { + match xtream_get_short_epg(&_app_state, target_name, api_req.stream_id.trim(), api_req.limit.trim()).await { Ok(content) => HttpResponse::Ok().content_type(mime::APPLICATION_JSON).body(content), Err(_) => HttpResponse::NoContent().finish() } diff --git a/src/config_reader.rs b/src/config_reader.rs index 849736646..0c31116fa 100644 --- a/src/config_reader.rs +++ b/src/config_reader.rs @@ -72,7 +72,7 @@ pub(crate) fn read_mapping(mapping_file: &str) -> Result, M3uFi } } _ => { - warn!("cant read mapping file: {}", mapping_file.to_str().unwrap()); + warn!("cant read mapping file: {}", mapping_file.to_str().unwrap_or("?")); Ok(None) } } diff --git a/src/download.rs b/src/download.rs index f515a4838..56cb310c5 100644 --- a/src/download.rs +++ b/src/download.rs @@ -46,8 +46,8 @@ const ACTIONS: [(XtreamCluster, &str, &str); 3] = [ pub(crate) async fn get_xtream_playlist(input: &ConfigInput, working_dir: &String) -> (Vec, Vec) { let mut playlist: Vec = Vec::new(); - let username = input.username.as_ref().unwrap().clone(); - let password = input.password.as_ref().unwrap().clone(); + let username = input.username.as_ref().map_or("", |v| v); + let password = input.password.as_ref().map_or("", |v| v); let base_url = format!("{}/player_api.php?username={}&password={}", input.url, username, password); let stream_base_url = format!("{}/{}/{}", input.url, username, password); diff --git a/src/main.rs b/src/main.rs index 2fde3ed52..0ad4132f5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -101,7 +101,7 @@ fn start_in_server_mode(cfg: Arc, targets: Arc) { } fn init_logger(log_level: &str) { - let mut log_builder = Builder::new(); + let mut log_builder = Builder::from_default_env(); // Set the log level based on the parsed value match log_level.to_lowercase().as_str() { "trace" => log_builder.filter_level(LevelFilter::Trace), diff --git a/src/processing/playlist_watch.rs b/src/processing/playlist_watch.rs index c43e9cb64..fdd63ff8c 100644 --- a/src/processing/playlist_watch.rs +++ b/src/processing/playlist_watch.rs @@ -34,7 +34,7 @@ pub(crate) fn process_group_watch(cfg: &Config, target_name: &str, pl: &Playlist } } None => { - error!("failed to load watch_file {}", &path.to_str().unwrap()); + error!("failed to load watch_file {}", &path.to_str().unwrap_or_default()); } } } @@ -42,7 +42,7 @@ pub(crate) fn process_group_watch(cfg: &Config, target_name: &str, pl: &Playlist match save_watch_tree(&save_path, new_tree) { Ok(_) => {} Err(err) => { - error!("failed to write watch_file {}: {}", &save_path.to_str().unwrap(), err) + error!("failed to write watch_file {}: {}", &save_path.to_str().unwrap_or_default(), err) } } } diff --git a/src/repository/xtream_repository.rs b/src/repository/xtream_repository.rs index f8458a47d..fd16b8b0b 100644 --- a/src/repository/xtream_repository.rs +++ b/src/repository/xtream_repository.rs @@ -5,16 +5,14 @@ use std::fs::{File, OpenOptions}; use std::io::{BufReader, BufWriter, Error, Read, Seek, SeekFrom, Write}; use std::iter::FromIterator; use std::path::{Path, PathBuf}; -use log::error; +use log::{error}; use serde::Serialize; use serde_json::{json, Map, Value}; -use crate::model::config::{Config, ConfigTarget, InputType}; +use crate::model::config::{Config, ConfigInput, ConfigTarget}; use crate::model::model_m3u::{PlaylistGroup, PlaylistItemHeader, XtreamCluster}; use crate::{create_m3u_filter_error_result, utils}; use crate::api::api_model::AppState; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; -use crate::model::api_proxy::UserCredentials; -use crate::utils::{get_client_request}; type IndexTree = BTreeMap; @@ -87,14 +85,13 @@ fn get_info_collection_and_idx_path(path: &Path, cluster: &XtreamCluster) -> (Pa (get_info_collection_path(path, collection), get_info_idx_path(path, collection)) } -fn write_xtream_info(app_state: &AppState, target_name: &str, stream_id: i32, cluster: &XtreamCluster, content: &String, index_tree: &mut IndexTree) -> Result<(), Error> { +fn write_xtream_info(app_state: &AppState, target_name: &str, stream_id: i32, cluster: &XtreamCluster, + content: &str, index_tree: &mut IndexTree) -> Result<(), Error> { if let Some(path) = get_xtream_storage_path(&app_state.config, target_name) { let (col_path, idx_path) = get_info_collection_and_idx_path(&path, cluster); let mut comp: Vec = Vec::new(); lzma_rs::lzma_compress(&mut BufReader::new(content.as_bytes()), &mut comp)?; let size = comp.len(); - let lock = app_state.shared_locks.get_lock(target_name); - let shared_lock = lock.write().unwrap(); match OpenOptions::new() .create(true) .write(true) @@ -106,9 +103,10 @@ fn write_xtream_info(app_state: &AppState, target_name: &str, stream_id: i32, cl file.flush()?; index_tree.insert(stream_id, (offset as u32, size as u16)); write_index(&idx_path, index_tree)?; - drop(shared_lock); } - Err(err) => return Err(err) + Err(err) => { + return Err(err) + } } } Ok(()) @@ -338,94 +336,60 @@ fn seek_read( Ok(buf) } -pub(crate) async fn xtream_get_stream_info(app_state: &AppState, target_name: &str, stream_id: i32, - cluster: XtreamCluster, - user: &UserCredentials) -> Result { - if let Some(target_input) = app_state.config.get_input_for_target(target_name, &InputType::Xtream) { - let cache_info = target_input.options.as_ref() - .map(|o| o.xtream_info_cache).unwrap_or(false); - let mut index_tree: Option = None; - if cache_info { - if let Some(path) = get_xtream_storage_path(&app_state.config, target_name) { - let (col_path, idx_path) = get_info_collection_and_idx_path(&path, &cluster); - let lock = app_state.shared_locks.get_lock(target_name); - let shared_lock = lock.read().unwrap(); - if idx_path.exists() && col_path.exists() { - index_tree = load_index(&idx_path); - if let Some(idx_map) = &index_tree { - if let Some((offset, size)) = idx_map.get(&stream_id) { - let mut reader = BufReader::new(File::open(&col_path).unwrap()); - if let Ok(bytes) = seek_read(&mut reader, *offset, *size) { - let mut decomp: Vec = Vec::new(); - let _ = lzma_rs::lzma_decompress(&mut bytes.as_slice(), &mut decomp); - drop(shared_lock); - return Ok(String::from_utf8(decomp).unwrap()); - } + +pub(crate) async fn xtream_get_stored_stream_info( + app_state: &AppState, target_name: &str, stream_id: i32, + cluster: &XtreamCluster, target_input: &ConfigInput) -> Result { + let cache_info = target_input.options.as_ref() + .map(|o| o.xtream_info_cache).unwrap_or(false); + if cache_info { + if let Some(path) = get_xtream_storage_path(&app_state.config, target_name) { + let (col_path, idx_path) = get_info_collection_and_idx_path(&path, &cluster); + let lock = app_state.shared_locks.get_lock(target_name); + let shared_lock = lock.read().unwrap(); + if idx_path.exists() && col_path.exists() { + let index_tree = load_index(&idx_path); + if let Some(idx_map) = &index_tree { + if let Some((offset, size)) = idx_map.get(&stream_id) { + let mut reader = BufReader::new(File::open(&col_path).unwrap()); + if let Ok(bytes) = seek_read(&mut reader, *offset, *size) { + let mut decomp: Vec = Vec::new(); + let _ = lzma_rs::lzma_decompress(&mut bytes.as_slice(), &mut decomp); + drop(shared_lock); + return Ok(String::from_utf8(decomp).unwrap()); } } } + } + drop(shared_lock); + } + } + Err(()) +} + +pub(crate) async fn xtream_persist_stream_info( + app_state: &AppState, target_name: &str, stream_id: i32, + cluster: &XtreamCluster, target_input: &ConfigInput, content: &str) { + let cache_info = target_input.options.as_ref() + .map(|o| o.xtream_info_cache).unwrap_or(false); + if cache_info { + if let Some(path) = get_xtream_storage_path(&app_state.config, target_name) { + let lock = app_state.shared_locks.get_lock(target_name); + let shared_lock = lock.write().unwrap(); + let mut index_tree = { + let (col_path, idx_path) = get_info_collection_and_idx_path(&path, cluster); + if idx_path.exists() && col_path.exists() { + load_index(&idx_path).unwrap_or_default() + } else { + IndexTree::new() + } + }; + match write_xtream_info(app_state, target_name, stream_id, cluster, content, + &mut index_tree) { + Ok(_) => {} + Err(err) => { error!("{}", err.to_string()); } + } drop(shared_lock); - } } - - let (action, stream_id_field) = match cluster { - XtreamCluster::Live => ("get_live_info", "live_id"), - XtreamCluster::Video => ("get_vod_info", "vod_id"), - XtreamCluster::Series => ("get_series_info", "series_id"), - }; - - let info_url = format!("{}/player_api.php?username={}&password={}&action={}&{}={}", - target_input.url, &user.username, &user.password, action, - stream_id_field, stream_id); - let url = reqwest::Url::parse(&info_url).unwrap(); - let client = get_client_request(target_input, url); - if let Ok(response) = client.send().await { - if response.status().is_success() { - match response.text().await { - Ok(content) => { - if cache_info { - if index_tree.is_none() { - index_tree = Some(IndexTree::new()); - } - match write_xtream_info(app_state, target_name, stream_id, &cluster, &content, - index_tree.as_mut().unwrap()) { - Ok(_) => {} - Err(err) => { error!("{}", err.to_string()); } - } - } - return Ok(content); - } - Err(err) => { error!("Failed to download info {}", err.to_string()); } - } - } - } - //} else if let Some(target_input) = app_state.config.get_input_for_target(target_name, &InputType::M3u) { - // TODO return dummy data ? } - Err(Error::new(std::io::ErrorKind::Other, format!("Cant find stream with id: {}/{}/{}", target_name, &cluster, stream_id))) -} - -pub(crate) async fn xtream_get_short_epg(app_state: &AppState, target_name: &str, stream_id: &str, limit: &str, user: &UserCredentials) -> Result { - if let Some(target_input) = app_state.config.get_input_for_target(target_name, &InputType::Xtream) { - let mut info_url = format!("{}/player_api.php?username={}&password={}&action=get_short_epg&stream_id={}", - target_input.url, &user.username, &user.password, stream_id); - if !(limit.is_empty() || limit.eq("0")) { - info_url = format!("{}&limit={}", info_url, limit); - } - let url = reqwest::Url::parse(&info_url).unwrap(); - let client = get_client_request(target_input, url); - if let Ok(response) = client.send().await { - if response.status().is_success() { - match response.text().await { - Ok(content) => { - return Ok(content); - } - Err(err) => { error!("Failed to download epg {}", err.to_string()); } - } - } - } - //} else if let Some(target_input) = app_state.config.get_input_for_target(target_name, &InputType::M3u) { - // TODO return dummy data ? - } - Err(Error::new(std::io::ErrorKind::Other, format!("Cant find short epg with id: {}/{}", target_name, stream_id))) -} +} \ No newline at end of file diff --git a/src/utils.rs b/src/utils.rs index d82e117eb..bb647c836 100644 --- a/src/utils.rs +++ b/src/utils.rs @@ -4,7 +4,6 @@ use std::io::{Read, Write}; use std::path::{Path, PathBuf}; use log::{debug, error}; use path_absolutize::*; -use reqwest::header; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use crate::create_m3u_filter_error_result; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; @@ -196,10 +195,10 @@ pub(crate) fn get_file_path(wd: &String, path: Option) -> Option reqwest::RequestBuilder { +pub(crate) fn get_client_request(input: &ConfigInput, url: url::Url, custom_headers: Option<&HashMap<&str, &[u8]>>) -> reqwest::RequestBuilder { let mut request = reqwest::Client::new().get(url); if input.headers.is_empty() { - let headers = get_request_headers(&input.headers); + let headers = get_request_headers(&input.headers, custom_headers); request = request.headers(headers); } request @@ -215,20 +214,29 @@ pub(crate) fn get_client_request(input: &ConfigInput, url: url::Url) -> reqwest: // } -pub(crate) fn get_request_headers(defined_headers: &HashMap) -> HeaderMap { - let mut headers = header::HeaderMap::new(); +pub(crate) fn get_request_headers(defined_headers: &HashMap, custom_headers: Option<&HashMap<&str, &[u8]>>) -> HeaderMap { + debug!("Custom header: {:?}", custom_headers); + let mut headers = HeaderMap::new(); for (key, value) in defined_headers { headers.insert( HeaderName::from_bytes(key.as_bytes()).unwrap(), - HeaderValue::from_bytes(value.as_bytes()).unwrap(), - ); + HeaderValue::from_bytes(value.as_bytes()).unwrap()); + } + if let Some(custom) = custom_headers { + for (key, value) in custom { + if !("host".eq(*key) || headers.contains_key(*key)) { + headers.insert( + HeaderName::from_bytes(key.as_bytes()).unwrap(), + HeaderValue::from_bytes(value).unwrap()); + } + } } debug!("Request with headers{:?}", &headers); headers } async fn download_json_content(input: &ConfigInput, url: url::Url, persist_filepath: Option) -> Result { - let request = get_client_request(input, url); + let request = get_client_request(input, url, None); match request.send().await { Ok(response) => { if response.status().is_success() { @@ -260,7 +268,7 @@ pub(crate) async fn get_input_json_content(input: &ConfigInput, url_str: &String } async fn download_text_content(input: &ConfigInput, url: url::Url, persist_filepath: Option) -> Result { - let request = get_client_request(input, url); + let request = get_client_request(input, url, None); match request.send().await { Ok(response) => { if response.status().is_success() {