From 3c4dca72c2e7e2ed8f86adf42f864a6316e39061 Mon Sep 17 00:00:00 2001 From: euzu Date: Mon, 9 Dec 2024 16:43:19 +0100 Subject: [PATCH] refactored api functions --- src/api/api_utils.rs | 5 + src/api/m3u_api.rs | 96 ++++++++------- src/api/xtream_api.rs | 155 ++++++++++++------------ src/repository/m3u_playlist_iterator.rs | 6 +- src/repository/playlist_repository.rs | 3 +- 5 files changed, 144 insertions(+), 121 deletions(-) diff --git a/src/api/api_utils.rs b/src/api/api_utils.rs index 6e03ba219..98ea0b8ec 100644 --- a/src/api/api_utils.rs +++ b/src/api/api_utils.rs @@ -20,6 +20,7 @@ use std::time::Duration; use tokio::sync::broadcast; use tokio_stream::wrappers::BroadcastStream; use url::Url; +use crate::model::playlist::PlaylistItemType; pub async fn serve_file(file_path: &Path, req: &HttpRequest, mime_type: mime::Mime) -> HttpResponse { if file_path.exists() { @@ -204,3 +205,7 @@ async fn shared_stream_response(app_state: &AppState, stream_url: &str) -> Optio } None } + +pub fn is_stream_share_enabled(item_type: PlaylistItemType, target: &ConfigTarget) -> bool { + item_type == PlaylistItemType::Live && target.options.as_ref().map_or(false, |opt| opt.share_live_streams) +} \ No newline at end of file diff --git a/src/api/m3u_api.rs b/src/api/m3u_api.rs index 31aec9f65..cd30ab261 100644 --- a/src/api/m3u_api.rs +++ b/src/api/m3u_api.rs @@ -1,13 +1,14 @@ -use actix_web::{HttpRequest, HttpResponse, web}; -use log::{debug, error}; -use futures::{stream}; +use actix_web::{web, HttpRequest, HttpResponse}; use bytes::Bytes; +use futures::stream; +use log::{debug, error}; -use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, stream_response}; +use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, is_stream_share_enabled, stream_response}; use crate::api::model::app_state::AppState; use crate::api::model::request::UserApiRequest; use crate::model::api_proxy::ProxyType; use crate::model::config::TargetType; +use crate::repository::m3u_playlist_iterator::M3U_STREAM_PATH; use crate::repository::m3u_repository::{m3u_get_file_paths, m3u_get_item_for_stream_id, m3u_load_rewrite_playlist}; use crate::repository::storage::get_target_storage_path; use crate::utils::request_utils::mask_sensitive_info; @@ -36,8 +37,8 @@ async fn m3u_api( } } -async fn m3u_api_get( api_req: web::Query, - app_state: web::Data, +async fn m3u_api_get(api_req: web::Query, + app_state: web::Data, ) -> HttpResponse { m3u_api(&api_req.into_inner(), &app_state).await } @@ -55,44 +56,55 @@ async fn m3u_api_stream( app_state: web::Data, ) -> HttpResponse { let (username, password, stream_id) = path.into_inner(); - if let Ok(m3u_stream_id) = stream_id.parse::() { - if let Some((user, target)) = get_user_target_by_credentials(&username, &password, &api_req, &app_state) { - if target.has_output(&TargetType::M3u) { - match get_target_storage_path(&app_state.config, target.name.as_str()) { - Some(target_path) => { - let (m3u_path, idx_path) = m3u_get_file_paths(&target_path); - match m3u_get_item_for_stream_id(&app_state.config, m3u_stream_id, &m3u_path, &idx_path).await { - Ok(m3u_item) => { - if user.proxy == ProxyType::Redirect { - let stream_url = m3u_item.url; - debug!("Redirecting stream request to {}", mask_sensitive_info(&stream_url)); - return HttpResponse::Found().insert_header(("Location", stream_url.to_string())).finish(); - } - let share_live_streams = target.options.as_ref().map_or(false, |opt| opt.share_live_streams); - return stream_response(&app_state, m3u_item.url.as_str(), &req, None, share_live_streams).await; - } - Err(err) => { - error!("Failed to get m3u url: {}", mask_sensitive_info(err.to_string().as_str())); - } - } - } - None => { - error!("Failed to get target path for {}", target.name); - } - } - } - } + let Ok(m3u_stream_id) = stream_id.parse::() else { return HttpResponse::BadRequest().finish() }; + let Some((user, target)) = get_user_target_by_credentials(&username, &password, &api_req, &app_state) else { return HttpResponse::BadRequest().finish() }; + + if !target.has_output(&TargetType::M3u) { + return HttpResponse::BadRequest().finish(); } - HttpResponse::BadRequest().finish() + + let Some(target_path) = get_target_storage_path(&app_state.config, target.name.as_str()) else { + error!("Failed to get target path for {}", target.name); + return HttpResponse::BadRequest().finish(); + }; + + let (m3u_path, idx_path) = m3u_get_file_paths(&target_path); + let m3u_item = match m3u_get_item_for_stream_id(&app_state.config, m3u_stream_id, &m3u_path, &idx_path).await { + Ok(item) => item, + Err(err) => { + error!("Failed to get m3u url: {}", mask_sensitive_info(err.to_string().as_str())); + return HttpResponse::BadRequest().finish(); + } + }; + + if user.proxy == ProxyType::Redirect { + let stream_url = m3u_item.url; + debug!("Redirecting stream request to {}", mask_sensitive_info(&stream_url)); + return HttpResponse::Found().insert_header(("Location", stream_url.to_string())).finish(); + } + + let share_live_streams = is_stream_share_enabled(m3u_item.item_type, target); + stream_response(&app_state, m3u_item.url.as_str(), &req, None, share_live_streams).await +} + +macro_rules! register_m3u_stream_routes { + ($cfg:expr, [$($path:expr),*]) => {{ + $( + $cfg.service(web::resource(format!("/{M3U_STREAM_PATH}/{}/{{username}}/{{password}}/{{stream_id}}", $path)).route(web::get().to(m3u_api_stream))); + )* + }}; +} + +macro_rules! register_m3u_api_routes { + ($cfg:expr, [$($path:expr),*]) => {{ + $( + $cfg.service(web::resource(format!("/{}", $path)).route(web::get().to(m3u_api_get)).route(web::post().to(m3u_api_post))); + )* + }}; } pub fn m3u_api_register(cfg: &mut web::ServiceConfig) { - cfg.service(web::resource("/get.php").route(web::get().to(m3u_api_get)).route(web::post().to(m3u_api_post))) - .service(web::resource("/get.php").route(web::post().to(m3u_api_get)).route(web::post().to(m3u_api_post))) - .service(web::resource("/apiget").route(web::get().to(m3u_api_get)).route(web::post().to(m3u_api_post))) - .service(web::resource("/m3u").route(web::get().to(m3u_api_get)).route(web::post().to(m3u_api_post))) - .service(web::resource("/m3u-stream/live/{username}/{password}/{stream_id}").route(web::get().to(m3u_api_stream))) - .service(web::resource("/m3u-stream/movie/{username}/{password}/{stream_id}").route(web::get().to(m3u_api_stream))) - .service(web::resource("/m3u-stream/series/{username}/{password}/{stream_id}").route(web::get().to(m3u_api_stream))) - .service(web::resource("/m3u-stream/{username}/{password}/{stream_id}").route(web::get().to(m3u_api_stream))); + register_m3u_api_routes!(cfg, ["get.php", "apiget", "m3u"]); + register_m3u_stream_routes!(cfg, ["live", "movie", "series"]); + cfg.service(web::resource(format!("/{M3U_STREAM_PATH}/{}", "{username}/{password}/{stream_id}")).route(web::get().to(m3u_api_stream))); } \ No newline at end of file diff --git a/src/api/xtream_api.rs b/src/api/xtream_api.rs index f6b5c76af..eff4253b9 100644 --- a/src/api/xtream_api.rs +++ b/src/api/xtream_api.rs @@ -13,7 +13,7 @@ use futures::Stream; use log::{debug, error, log_enabled, warn, Level}; use serde_json::{Map, Value}; -use crate::api::api_utils::{get_user_server_info, get_user_target, get_user_target_by_credentials, serve_file, stream_response}; +use crate::api::api_utils::{get_user_server_info, get_user_target, get_user_target_by_credentials, is_stream_share_enabled, serve_file, stream_response}; use crate::api::model::app_state::AppState; use crate::api::model::request::UserApiRequest; use crate::api::model::xtream::XtreamAuthorizationResponse; @@ -235,75 +235,54 @@ async fn xtream_player_api_stream( if log_enabled!(Level::Debug) { debug!("Streaming stream request from {}", mask_sensitive_info(&stream_url)); } - let share_live_streams = target.options.as_ref().map_or(false, |opt| opt.share_live_streams); + let share_live_streams = is_stream_share_enabled(pli.item_type, target); stream_response(app_state, &stream_url, req, Some(input), share_live_streams).await } - -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(&req, &api_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::Live, &username, &password, &stream_id, "")).await +macro_rules! create_xtream_player_api_stream { + ($fn_name:ident, $context:expr) => { + async fn $fn_name( + 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(&req, &api_req, &app_state, XtreamApiStreamRequest::from($context, &username, &password, &stream_id, "")).await + } + } } -async fn xtream_player_api_live_stream_alt( - 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(&req, &api_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::LiveAlt, &username, &password, &stream_id, "")).await -} +create_xtream_player_api_stream!(xtream_player_api_live_stream, XtreamApiStreamContext::Live); +create_xtream_player_api_stream!(xtream_player_api_live_stream_alt, XtreamApiStreamContext::LiveAlt); +create_xtream_player_api_stream!(xtream_player_api_series_stream, XtreamApiStreamContext::Series); +create_xtream_player_api_stream!(xtream_player_api_movie_stream, XtreamApiStreamContext::Movie); -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(&req, &api_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::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(&req, &api_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::Movie, &username, &password, &stream_id, "")).await +fn get_non_empty<'a>(first: &'a str, second: &'a str, third: &'a str) -> &'a str { + if !first.is_empty() { + first + } else if !second.is_empty() { + second + } else { + third + } } async fn xtream_player_api_timeshift_stream( req: HttpRequest, - api_req: web::Query, + api_query_req: web::Query, + api_form_req: web::Form, path: web::Path<(String, String, String, String, String)>, app_state: web::Data, ) -> HttpResponse { - let (username, password, duration, start, stream_id) = path.into_inner(); + let (path_username, path_password, path_duration, path_start, path_stream_id) = path.into_inner(); + let username = get_non_empty(&path_username, &api_query_req.username, &api_form_req.username); + let password = get_non_empty(&path_password, &api_query_req.password, &api_form_req.password); + let stream_id = get_non_empty(&path_stream_id, &api_query_req.stream, &api_form_req.stream); + let duration = get_non_empty(&path_duration, &api_query_req.duration, &api_form_req.duration); + let start = get_non_empty(&path_start, &api_query_req.start, &api_form_req.start); let action_path = format!("{duration}/{start}"); - xtream_player_api_stream(&req, &api_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::Timeshift, &username, &password, &stream_id, &action_path)).await -} - - -async fn xtream_player_api_streaming_timeshift( - req: HttpRequest, - api_req: web::Query, - app_state: web::Data, -) -> HttpResponse { - let username = &api_req.username; - let password = &api_req.password; - let stream_id = &api_req.stream; - let duration = &api_req.duration; - let start = &api_req.start; - let action_path = format!("{duration}/{start}"); - xtream_player_api_stream(&req, &api_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::Timeshift, username, password, stream_id, &action_path)).await + xtream_player_api_stream(&req, &api_query_req, &app_state, XtreamApiStreamRequest::from(XtreamApiStreamContext::Timeshift, username, password, stream_id, &action_path)).await } fn get_xtream_vod_info(target: &ConfigTarget, pli: &XtreamPlaylistItem, content: &str) -> Result { @@ -579,16 +558,16 @@ async fn xtream_player_api( fn xtream_create_content_stream(xtream_iter: impl Iterator) -> impl Stream> { let mut first_item = true; stream::once(async { Ok::(Bytes::from("[")) }).chain( - stream::iter(xtream_iter.map(move |line| { - let line = if first_item { - first_item = false; - line - } else { - format!(",{line}") - }; - Ok::(Bytes::from(line)) - })) - .chain(stream::once(async { Ok::(Bytes::from("]")) }))) + stream::iter(xtream_iter.map(move |line| { + let line = if first_item { + first_item = false; + line + } else { + format!(",{line}") + }; + Ok::(Bytes::from(line)) + })) + .chain(stream::once(async { Ok::(Bytes::from("]")) }))) } @@ -606,17 +585,41 @@ async fn xtream_player_api_post(req: HttpRequest, xtream_player_api(&req, api_req.into_inner(), &app_state).await } +macro_rules! register_xtream_api { + ($cfg:expr, [$($path:expr),*]) => {{ + $( + $cfg.service(web::resource($path).route(web::get().to(xtream_player_api_get)).route(web::post().to(xtream_player_api_post))); + )* + }}; +} + +macro_rules! register_xtream_api_stream { + ($cfg:expr, [$(($path:expr, $fn_name:ident)),*]) => {{ + $( + $cfg.service(web::resource(format!("{}/{{username}}/{{password}}/{{stream_id}}", $path)).route(web::get().to($fn_name))); + )* + }}; +} + +macro_rules! register_xtream_api_timeshift { + ($cfg:expr, [$($path:expr),*]) => {{ + $( + $cfg.service(web::resource($path).route(web::get().to(xtream_player_api_timeshift_stream)).route(web::post().to(xtream_player_api_timeshift_stream))); + )* + }}; +} + pub fn xtream_api_register(cfg: &mut web::ServiceConfig) { - cfg.service(web::resource("/player_api.php").route(web::get().to(xtream_player_api_get)).route(web::post().to(xtream_player_api_post))) - .service(web::resource("/panel_api.php").route(web::get().to(xtream_player_api_get)).route(web::post().to(xtream_player_api_post))) - .service(web::resource("/xtream").route(web::get().to(xtream_player_api_get)).route(web::post().to(xtream_player_api_post))) - .service(web::resource("/{username}/{password}/{stream_id}").route(web::get().to(xtream_player_api_live_stream_alt))) - .service(web::resource("/live/{username}/{password}/{stream_id}").route(web::get().to(xtream_player_api_live_stream))) - .service(web::resource("/movie/{username}/{password}/{stream_id}").route(web::get().to(xtream_player_api_movie_stream))) - .service(web::resource("/series/{username}/{password}/{stream_id}").route(web::get().to(xtream_player_api_series_stream))) - .service(web::resource("/timeshift/{username}/{password}/{duration}/{start}/{stream_id}").route(web::get().to(xtream_player_api_timeshift_stream))) - .service(web::resource("/timeshift.php").route(web::get().to(xtream_player_api_streaming_timeshift))) - .service(web::resource("/streaming/timeshift.php").route(web::get().to(xtream_player_api_streaming_timeshift))); + register_xtream_api!(cfg, ["/player_api.php", "/panel_api.php", "/xtream"]); + register_xtream_api_stream!(cfg, [ + ("", xtream_player_api_live_stream_alt), + ("/live", xtream_player_api_live_stream), + ("/movie", xtream_player_api_movie_stream), + ("/series", xtream_player_api_series_stream)]); + register_xtream_api_timeshift!(cfg, [ + "/timeshift/{username}/{password}/{duration}/{start}/{stream_id}", + "/timeshift.php", + "/streaming/timeshift.php"]); /* TODO cfg.service(web::resource("/hlsr/{token}/{username}/{password}/{channel}/{hash}/{chunk}").route(web::get().to(xtream_player_api_hlsr_stream))); cfg.service(web::resource("/hls/{token}/{chunk}").route(web::get().to(xtream_player_api_hls_stream))); diff --git a/src/repository/m3u_playlist_iterator.rs b/src/repository/m3u_playlist_iterator.rs index 8f4762a57..4d7452b2f 100644 --- a/src/repository/m3u_playlist_iterator.rs +++ b/src/repository/m3u_playlist_iterator.rs @@ -8,6 +8,8 @@ use crate::repository::m3u_repository::m3u_get_file_paths; use crate::repository::storage::ensure_target_storage_path; use crate::utils::file_lock_manager::FileReadGuard; +pub const M3U_STREAM_PATH: &str = "m3u-stream"; + pub struct M3uPlaylistIterator { reader: IndexedDocumentReader, base_url: String, @@ -76,14 +78,14 @@ impl M3uPlaylistIterator { | PlaylistItemType::SeriesInfo | PlaylistItemType::SeriesEpisode => "series", }; - format!("{}/m3u-stream/{stream_type}/{}/{}/{}", + format!("{}/{M3U_STREAM_PATH}/{stream_type}/{}/{}/{}", &self.base_url, &self.username, &self.password, m3u_pli.virtual_id ) } else { - format!("{}/m3u-stream/{}/{}/{}", + format!("{}/{M3U_STREAM_PATH}/{}/{}/{}", &self.base_url, &self.username, &self.password, m3u_pli.virtual_id ) } diff --git a/src/repository/playlist_repository.rs b/src/repository/playlist_repository.rs index 0d0560fd2..ab614f650 100644 --- a/src/repository/playlist_repository.rs +++ b/src/repository/playlist_repository.rs @@ -11,7 +11,7 @@ use crate::repository::target_id_mapping::TargetIdMapping; use crate::repository::xtream_repository::xtream_write_playlist; pub async fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, - target: &ConfigTarget, cfg: &Config) -> Result<(), Vec> { + target: &ConfigTarget, cfg: &Config) -> Result<(), Vec> { let mut errors = vec![]; let target_path = match ensure_target_storage_path(cfg, &target.name) { Ok(path) => path, @@ -19,6 +19,7 @@ pub async fn persist_playlist(playlist: &mut [PlaylistGroup], epg: Option<&Epg>, }; let target_id_mapping_file = get_target_id_mapping_file(&target_path); + let _file_lock = match cfg.file_locks.write_lock(&target_id_mapping_file).await { Ok(lock) => lock, Err(err) => {