euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-09-27 19:42:28 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

refactored unwrap calls to make it more stable

Browse Source
This commit is contained in:
euzu
2025-07-22 11:32:46 +02:00
parent 7ef30e1aae
commit 6c238c607f
44 changed files with 560 additions and 479 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/api/api_utils.rs
+26 -22
View File
@@ -37,6 +37,8 @@ use std::path::Path;
use std::sync::Arc;
use tokio::sync::Mutex;
use url::Url;
use crate::api::model::active_user_manager::UserSession;
use crate::api::model::provider_config::ProviderConfig;
#[macro_export]
macro_rules! try_option_bad_request {
@@ -76,10 +78,19 @@ macro_rules! try_result_bad_request {
};
}
#[macro_export]
macro_rules! try_unwrap_body {
($body:expr) => {
$body.map_or_else(
|_| axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response(),
|resp| resp.into_response(),
)
};
}
pub use try_option_bad_request;
pub use try_result_bad_request;
use crate::api::model::active_user_manager::UserSession;
use crate::api::model::provider_config::ProviderConfig;
pub use try_unwrap_body;
pub fn get_server_time() -> String {
chrono::offset::Local::now().with_timezone(&chrono::Local).format("%Y-%m-%d %H:%M:%S %Z").to_string()
@@ -103,13 +114,11 @@ pub async fn serve_file(file_path: &Path, mime_type: mime::Mime) -> impl IntoRes
let stream = tokio_util::io::ReaderStream::new(reader);
let body = axum::body::Body::from_stream(stream);
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime_type.to_string())
.header(axum::http::header::CACHE_CONTROL, axum::http::header::HeaderValue::from_static("no-cache"))
.body(body)
.unwrap()
.into_response()
.body(body))
}
Err(_) => axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response(),
};
@@ -295,8 +304,6 @@ async fn resolve_streaming_strategy(app_state: &AppState, stream_url: &str, addr
None => app_state.active_provider.acquire_connection(&input.name, addr).await
};
error!("{:?}", app_state.active_provider.active_connections().await);
let stream_response_params = match &**provider_connection_guard {
ProviderAllocation::Exhausted => {
debug!("Input {} is exhausted. No connections allowed.", input.name);
@@ -556,14 +563,14 @@ pub async fn force_provider_stream_response(addr: &str,
let body_stream = prepare_body_stream(app_state, item_type, stream);
debug_if_enabled!("Streaming provider forced stream request from {}", sanitize_sensitive_info(&user_session.stream_url));
return response.body(body_stream).unwrap().into_response();
return try_unwrap_body!(response.body(body_stream));
}
drop(stream_details.provider_connection_guard.take());
if let (Some(stream), _stream_info) =
create_channel_unavailable_stream(&app_state.app_config, &[], axum::http::StatusCode::BAD_GATEWAY)
{
debug!("Streaming custom stream");
axum::response::Response::builder().status(axum::http::StatusCode::OK).body(axum::body::Body::from_stream(stream)).unwrap().into_response()
try_unwrap_body!(axum::response::Response::builder().status(axum::http::StatusCode::OK).body(axum::body::Body::from_stream(stream)))
} else {
axum::http::StatusCode::BAD_REQUEST.into_response()
}
@@ -611,7 +618,7 @@ pub async fn stream_response(addr: &str,
for (key, value) in &header_map {
response = response.header(key, value);
}
response.body(axum::body::Body::from_stream(broadcast_stream)).unwrap().into_response()
try_unwrap_body!(response.body(axum::body::Body::from_stream(broadcast_stream)))
} else {
axum::http::StatusCode::BAD_REQUEST.into_response()
}
@@ -637,7 +644,7 @@ pub async fn stream_response(addr: &str,
}
let body_stream = prepare_body_stream(app_state, item_type, stream);
response.body(body_stream).unwrap().into_response()
try_unwrap_body!(response.body(body_stream))
};
return stream_resp.into_response();
@@ -666,7 +673,7 @@ fn shared_stream_response(app_state: &AppState, stream_url: &str, addr: &str, us
for (key, value) in &header_map {
response = response.header(key, value);
}
return Some(response.body(axum::body::Body::from_stream(stream)).unwrap());
return response.body(axum::body::Body::from_stream(stream)).ok();
}
}
None
@@ -743,10 +750,10 @@ pub async fn resource_response(app_state: &AppState, resource_url: &str, req_hea
let writer = BufWriter::new(file);
let add_cache_content = get_add_cache_content(resource_url, &app_state.cache);
let stream = PersistPipeStream::new(byte_stream, writer, add_cache_content);
return response_builder.body(axum::body::Body::from_stream(stream)).unwrap().into_response();
return try_unwrap_body!(response_builder.body(axum::body::Body::from_stream(stream)));
}
}
return response_builder.body(axum::body::Body::from_stream(byte_stream)).unwrap().into_response();
return try_unwrap_body!(response_builder.body(axum::body::Body::from_stream(byte_stream)));
}
debug_if_enabled!("Failed to open resource got status {} for {}", status, sanitize_sensitive_info(resource_url));
}
@@ -770,12 +777,10 @@ pub fn separate_number_and_remainder(input: &str) -> (String, Option<String>) {
/// # Panics
pub fn empty_json_list_response() -> impl IntoResponse + Send {
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header("Content-Type", mime::APPLICATION_JSON.to_string())
.body("[]".to_string())
.unwrap()
.into_response()
.body("[]".to_string()))
}
pub fn get_username_from_auth_header(
@@ -797,11 +802,10 @@ pub fn get_username_from_auth_header(
/// # Panics
pub fn redirect(url: &str) -> impl IntoResponse {
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::FOUND)
.header("Location", url)
.body(axum::body::Body::empty())
.unwrap()
.body(axum::body::Body::empty()))
}
pub async fn is_seek_request(
backend/src/api/endpoints/api_playlist_utils.rs
+39 -30
View File
@@ -1,13 +1,13 @@
use crate::model::{AppConfig, Config, ConfigInput, ConfigTarget};
use shared::model::{InputType, M3uPlaylistItem, PlaylistGroup, PlaylistItemType, TargetType, XtreamCluster};
use crate::repository::{m3u_repository, xtream_repository};
use crate::utils::{m3u, xtream};
use crate::utils;
use axum::response::IntoResponse;
use indexmap::IndexMap;
use serde::Serialize;
use serde_json::{json, Value};
use shared::model::{InputType, M3uPlaylistItem, PlaylistGroup, PlaylistItemType, TargetType, XtreamCluster};
use std::sync::Arc;
use indexmap::IndexMap;
use crate::utils;
#[derive(serde::Serialize, serde::Deserialize)]
struct PlaylistResponseGroup {
@@ -57,35 +57,36 @@ where
fn group_playlist_items_by_cluster(params: Option<(utils::FileReadGuard,
impl Iterator<Item=(M3uPlaylistItem, bool)>)>) ->
(Vec<M3uPlaylistItem>, Vec<M3uPlaylistItem>, Vec<M3uPlaylistItem>) {
if params.is_none() {
return (vec![], vec![], vec![]);
}
let mut live = Vec::new();
let mut video = Vec::new();
let mut series = Vec::new();
let (guard, iter) = params.unwrap();
for (item, _) in iter {
match item.item_type {
PlaylistItemType::Live
| PlaylistItemType::LiveUnknown
| PlaylistItemType::LiveHls
| PlaylistItemType::LiveDash => {
live.push(item);
}
PlaylistItemType::Catchup
| PlaylistItemType::Video => {
video.push(item);
}
PlaylistItemType::Series
| PlaylistItemType::SeriesInfo => {
series.push(item);
match params {
None => (vec![], vec![], vec![]),
Some((guard, iter)) => {
let mut live = Vec::new();
let mut video = Vec::new();
let mut series = Vec::new();
for (item, _) in iter {
match item.item_type {
PlaylistItemType::Live
| PlaylistItemType::LiveUnknown
| PlaylistItemType::LiveHls
| PlaylistItemType::LiveDash => {
live.push(item);
}
PlaylistItemType::Catchup
| PlaylistItemType::Video => {
video.push(item);
}
PlaylistItemType::Series
| PlaylistItemType::SeriesInfo => {
series.push(item);
}
}
}
drop(guard);
(live, video, series)
}
}
drop(guard);
(live, video, series)
}
fn group_playlist_groups_by_cluster(playlist: Vec<PlaylistGroup>, input_type: InputType) -> (Vec<PlaylistResponseGroup>, Vec<PlaylistResponseGroup>, Vec<PlaylistResponseGroup>) {
@@ -93,7 +94,15 @@ fn group_playlist_groups_by_cluster(playlist: Vec<PlaylistGroup>, input_type: In
let mut video = Vec::new();
let mut series = Vec::new();
for group in playlist {
let channels = group.channels.iter().map(|item| if input_type == InputType::M3u { serde_json::to_value(item.to_m3u()).unwrap() } else { serde_json::to_value(item.to_xtream()).unwrap() }).collect();
let channels = group.channels.iter()
.filter_map(|item| {
if input_type == InputType::M3u {
serde_json::to_value(item.to_m3u())
} else {
serde_json::to_value(item.to_xtream())
}.ok()
})
.collect::<serde_json::Value>();
let grp = PlaylistResponseGroup {
id: group.id,
title: group.title,
backend/src/api/endpoints/download_api.rs
+58 -47
View File
@@ -19,49 +19,56 @@ use shared::error::to_io_error;
use crate::utils::request::create_client;
async fn download_file(active: Arc<RwLock<Option<FileDownload>>>, client: &reqwest::Client) -> Result<(), String> {
let file_download = { active.read().await.as_ref().unwrap().clone() };
match client.get(file_download.url.clone()).send().await {
Ok(response) => {
match fs::create_dir_all(&file_download.file_dir) {
Ok(()) => {
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(to_io_error);
loop {
match stream.try_next().await {
Ok(item) => {
if let Some(chunk) = item {
match file.write_all(&chunk) {
Ok(()) => {
downloaded += chunk.len() as u64;
active.write().await.as_mut().unwrap().size = downloaded;
if let Some(file_download) = active.read().await.as_ref().as_ref() {
match client.get(file_download.url.clone()).send().await {
Ok(response) => {
match fs::create_dir_all(&file_download.file_dir) {
Ok(()) => {
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(to_io_error);
loop {
match stream.try_next().await {
Ok(item) => {
if let Some(chunk) = item {
match file.write_all(&chunk) {
Ok(()) => {
downloaded += chunk.len() as u64;
if let Some(lock) = active.write().await.as_mut() {
lock.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}"))
} else {
let megabytes = bytes_to_megabytes(downloaded);
info!("Downloaded {file_path_str}, filesize: {megabytes}MB");
if let Some(lock) = active.write().await.as_mut() {
lock.size = downloaded;
}
return Ok(());
}
} else {
let megabytes = bytes_to_megabytes(downloaded);
info!("Downloaded {file_path_str}, filesize: {megabytes}MB");
active.write().await.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())
}
} 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))
}
Err(err) => Err(format!("Error while creating directory for file: {} {}", &file_download.file_dir.to_str().unwrap_or("?"), err))
}
Err(err) => Err(format!("Error while opening url: {} {}", &file_download.url, err))
}
Err(err) => Err(format!("Error while opening url: {} {}", &file_download.url, err))
} else {
Err("No active file download".to_string())
}
}
@@ -121,25 +128,29 @@ pub async fn queue_download_file(
let app_config = &*app_state.app_config;
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&app_config.config);
if let Some(download_cfg) = config.video.as_ref().unwrap().download.as_ref() {
if download_cfg.directory.is_none() {
return (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download.directory configuration"}))).into_response();
}
match FileDownload::new(req.url.as_str(), req.filename.as_str(), download_cfg) {
Some(file_download) => {
app_state.downloads.queue.lock().await.push_back(file_download.clone());
if app_state.downloads.active.read().await.is_none() {
match run_download_queue(&app_state.app_config, download_cfg, &app_state.downloads).await {
Ok(()) => {}
Err(err) => return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err}))).into_response(),
}
}
axum::Json(download_info!(&file_download)).into_response()
if let Some(video_cfg) = config.video.as_ref() {
if let Some(download_cfg) = video_cfg.download.as_ref() {
if download_cfg.directory.is_empty() {
return (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download.directory configuration"}))).into_response();
}
None => (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid Arguments"}))).into_response(),
match FileDownload::new(req.url.as_str(), req.filename.as_str(), download_cfg) {
Some(file_download) => {
app_state.downloads.queue.lock().await.push_back(file_download.clone());
if app_state.downloads.active.read().await.is_none() {
match run_download_queue(&app_state.app_config, download_cfg, &app_state.downloads).await {
Ok(()) => {}
Err(err) => return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err}))).into_response(),
}
}
axum::Json(download_info!(&file_download)).into_response()
}
None => (axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Invalid Arguments"}))).into_response(),
}
} else {
(axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download configuration"}))).into_response()
}
} else {
(axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video.download configuration"}))).into_response()
(axum::http::StatusCode::BAD_REQUEST, axum::Json(json!({"error": "Server config missing video configuration"}))).into_response()
}
}
backend/src/api/endpoints/hdhomerun_api.rs
+7 -11
View File
@@ -13,6 +13,7 @@ use serde::{Deserialize, Serialize};
use serde_json::json;
use std::sync::Arc;
use shared::utils::get_string_from_serde_value;
use crate::api::api_utils::try_unwrap_body;
// https://info.hdhomerun.com/info/http_api
// const DISCOVERY_BYTES: &[u8] = &[0, 2, 0, 12, 1, 4, 255, 255, 255, 255, 2, 4, 255, 255, 255, 255, 115, 204, 125, 143];
@@ -184,12 +185,10 @@ fn create_device(app_state: &Arc<HdHomerunAppState>) -> Option<Device> {
async fn device_xml(axum::extract::State(app_state): axum::extract::State<Arc<HdHomerunAppState>>) -> impl IntoResponse {
if let Some(device) = create_device(&app_state) {
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, "application/xml")
.body(axum::body::Body::from(device.as_xml()))
.unwrap()
.into_response()
.body(axum::body::Body::from(device.as_xml())))
} else {
axum::http::StatusCode::INTERNAL_SERVER_ERROR.into_response()
}
@@ -231,11 +230,10 @@ async fn lineup(app_state: &Arc<HdHomerunAppState>, cfg: &Arc<AppConfig>, creden
let body_stream = stream::once(async { Ok(Bytes::from("[")) })
.chain(stream)
.chain(stream::once(async { Ok(Bytes::from("]")) }));
return axum::response::Response::builder()
return try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from_stream(body_stream))
.unwrap().into_response();
.body(axum::body::Body::from_stream(body_stream)));
} else if (use_all || use_xtream) && target.has_output(&TargetType::Xtream) {
let server_info = app_state.app_state.app_config.get_user_server_info(credentials);
let base_url = server_info.get_base_url();
@@ -262,12 +260,10 @@ async fn lineup(app_state: &Arc<HdHomerunAppState>, cfg: &Arc<AppConfig>, creden
.chain(comma_stream)
.chain(vod_stream_peek)
.chain(stream::once(async { Ok(Bytes::from("]")) }));
return axum::response::Response::builder()
return try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from_stream(body_stream))
.unwrap()
.into_response();
.body(axum::body::Body::from_stream(body_stream)));
}
axum::http::StatusCode::NOT_FOUND.into_response()
}
backend/src/api/endpoints/hls_api.rs
+24 -25
View File
@@ -1,19 +1,20 @@
use crate::api::api_utils::{force_provider_stream_response, get_stream_alternative_url, is_seek_request};
use crate::api::api_utils::{try_option_bad_request};
use crate::api::api_utils::try_option_bad_request;
use crate::api::model::active_user_manager::UserSession;
use crate::api::model::app_state::AppState;
use crate::api::model::streams::provider_stream::{create_custom_video_stream_response, CustomVideoStreamType};
use crate::model::{ProxyUserCredentials};
use crate::auth::Fingerprint;
use crate::model::ConfigInput;
use shared::model::{PlaylistItemType, UserConnectionPermission, XtreamCluster};
use crate::model::ProxyUserCredentials;
use crate::processing::parser::hls::{get_hls_session_token_and_url_from_token, rewrite_hls, RewriteHlsProps};
use shared::utils::{is_hls_url, replace_url_extension, sanitize_sensitive_info, HLS_EXT};
use crate::api::api_utils::try_unwrap_body;
use crate::utils::request;
use axum::response::IntoResponse;
use log::{debug, error};
use serde::Deserialize;
use shared::model::{PlaylistItemType, UserConnectionPermission, XtreamCluster};
use shared::utils::{is_hls_url, replace_url_extension, sanitize_sensitive_info, HLS_EXT};
use std::sync::Arc;
use crate::api::model::active_user_manager::UserSession;
use crate::auth::Fingerprint;
#[derive(Debug, Deserialize)]
struct HlsApiPathParams {
@@ -25,24 +26,22 @@ struct HlsApiPathParams {
}
fn hls_response(hls_content: String) -> impl IntoResponse + Send {
let builder = axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, "application/x-mpegurl");
builder.body(hls_content)
.unwrap()
.into_response()
.header(axum::http::header::CONTENT_TYPE, "application/x-mpegurl")
.body(hls_content))
}
#[allow(clippy::too_many_arguments)]
pub(in crate::api) async fn handle_hls_stream_request(
fingerprint: &str, addr: &str,
app_state: &Arc<AppState>,
user: &ProxyUserCredentials,
user_session: Option<&UserSession>,
hls_url: &str,
virtual_id: u32,
input: &ConfigInput,
connection_permission: UserConnectionPermission) -> impl IntoResponse + Send {
fingerprint: &str, addr: &str,
app_state: &Arc<AppState>,
user: &ProxyUserCredentials,
user_session: Option<&UserSession>,
hls_url: &str,
virtual_id: u32,
input: &ConfigInput,
connection_permission: UserConnectionPermission) -> impl IntoResponse + Send {
let url = replace_url_extension(hls_url, HLS_EXT);
let server_info = app_state.app_config.get_user_server_info(user);
@@ -52,18 +51,18 @@ pub(in crate::api) async fn handle_hls_stream_request(
Some(provider_cfg) => {
let stream_url = get_stream_alternative_url(&url, input, &provider_cfg);
(stream_url, Some(session.token.to_string()))
},
}
None => (url, None),
}
},
}
None => {
match app_state.active_provider.get_next_provider(&input.name).await {
Some(provider_cfg) => {
let stream_url = get_stream_alternative_url(&url, input, &provider_cfg);
let user_session_token = format!("{fingerprint}{virtual_id}");
let session_token= app_state.active_users.create_user_session(user, &user_session_token, virtual_id, &provider_cfg.name, &stream_url, addr, connection_permission);
let session_token = app_state.active_users.create_user_session(user, &user_session_token, virtual_id, &provider_cfg.name, &stream_url, addr, connection_permission);
(stream_url, Some(session_token))
},
}
None => (url, None),
}
}
@@ -110,7 +109,7 @@ async fn hls_api_stream(
let user_session_token = format!("{fingerprint}{virtual_id}");
let mut user_session = app_state.active_users.get_user_session(&user.username, &user_session_token);
if let Some(session) = &mut user_session {
if let Some(session) = &mut user_session {
if session.permission == UserConnectionPermission::Exhausted {
return create_custom_video_stream_response(&app_state.app_config, CustomVideoStreamType::UserConnectionsExhausted).into_response();
}
@@ -128,7 +127,7 @@ async fn hls_api_stream(
if session.virtual_id == virtual_id {
if is_seek_request(XtreamCluster::Live, &req_headers).await {
// partial request means we are in reverse proxy mode, seek happened
return force_provider_stream_response(&addr, &app_state, session, PlaylistItemType::LiveHls, &req_headers, &input, &user).await.into_response()
return force_provider_stream_response(&addr, &app_state, session, PlaylistItemType::LiveHls, &req_headers, &input, &user).await.into_response();
}
} else {
return axum::http::StatusCode::BAD_REQUEST.into_response();
backend/src/api/endpoints/m3u_api.rs
+2 -1
View File
@@ -14,6 +14,7 @@ use futures::stream;
use log::{debug, error};
use std::sync::Arc;
use crate::auth::Fingerprint;
use crate::api::api_utils::try_unwrap_body;
async fn m3u_api(
api_req: &UserApiRequest,
@@ -32,7 +33,7 @@ async fn m3u_api(
if api_req.content_type == "m3u_plus" {
builder = builder.header("Content-Disposition", "attachment; filename=\"playlist.m3u\"");
}
builder.body(axum::body::Body::from_stream(content_stream)).unwrap().into_response()
try_unwrap_body!(builder.body(axum::body::Body::from_stream(content_stream)))
}
Err(err) => {
error!("{}", sanitize_sensitive_info(err.to_string().as_str()));
backend/src/api/endpoints/user_api.rs
+9 -13
View File
@@ -15,6 +15,7 @@ use std::collections::HashSet;
use std::sync::Arc;
use axum::response::IntoResponse;
use crate::auth::AuthBearer;
use crate::api::api_utils::try_unwrap_body;
fn get_categories_from_xtream(categories: Option<Vec<PlaylistXtreamCategory>>) -> Vec<String> {
let mut groups: Vec<String> = Vec::new();
@@ -86,13 +87,10 @@ async fn playlist_categories(
.chain(m3u_stream)
.chain(stream::once(async { Ok::<Bytes, String>(Bytes::from("}")) }));
return axum::response::Response::builder()
return try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header("Content-Type", mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from_stream(json_stream))
.unwrap()
.into_response();
.body(axum::body::Body::from_stream(json_stream)));
}
}
axum::http::StatusCode::BAD_REQUEST.into_response()
@@ -134,20 +132,18 @@ async fn playlist_bouquet(
let config = &app_state.app_config.config.load();
let xtream = load_user_bouquet_as_json(config, &username, TargetType::Xtream).await;
let m3u = load_user_bouquet_as_json(config, &username, TargetType::M3u).await;
return axum::response::Response::builder()
return try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header("Content-Type", mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from(format!(r#"{{"xtream": {}, "m3u": {} }}"#, xtream.unwrap_or("null".to_string()), m3u.unwrap_or("null".to_string()))))
.unwrap()
.into_response();
.body(axum::body::Body::from(format!(r#"{{"xtream": {}, "m3u": {} }}"#,
xtream.unwrap_or("null".to_string()),
m3u.unwrap_or("null".to_string())))));
}
}
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header("Content-Type", mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from("{}"))
.unwrap()
.into_response()
.body(axum::body::Body::from("{}")))
}
pub fn user_api_register(app_state: Arc<AppState>) -> axum::Router<Arc<AppState>> {
backend/src/api/endpoints/v1_api.rs
+12 -11
View File
@@ -11,6 +11,7 @@ use crate::processing::processor::playlist;
use crate::repository::user_repository::store_api_user;
use crate::utils::ip_checker::get_ips;
use crate::{utils, VERSION};
use crate::api::api_utils::try_unwrap_body;
use axum::response::IntoResponse;
use log::error;
use serde_json::json;
@@ -79,9 +80,9 @@ async fn save_config_api_proxy_user(
}
} else {
let config = app_state.app_config.config.load();
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
let backup_dir = config.get_backup_dir();
let paths = app_state.app_config.paths.load();
if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(&*new_api_proxy), paths.api_proxy_file_path.as_str()) {
if let Some(err) = intern_save_config_api_proxy(backup_dir.as_ref(), &ApiProxyConfigDto::from(&*new_api_proxy), paths.api_proxy_file_path.as_str()) {
return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response();
}
}
@@ -97,8 +98,8 @@ async fn save_config_main(
let paths = app_state.app_config.paths.load();
let file_path = paths.config_file_path.as_str();
let config = app_state.app_config.config.load();
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
if let Some(err) = intern_save_config_main(file_path, backup_dir, &cfg) {
let backup_dir = config.get_backup_dir();
if let Some(err) = intern_save_config_main(file_path, backup_dir.as_ref(), &cfg) {
return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response();
}
axum::http::StatusCode::OK.into_response()
@@ -117,16 +118,16 @@ async fn save_config_api_proxy_config(
}
}
// TODO wenn hot reload an ist wird doppelt geladen
// TODO if hot reload is on, loaded twice
if let Some(old_api_proxy) = app_state.app_config.api_proxy.load().clone() {
let mut api_proxy = (*old_api_proxy).clone();
api_proxy.server = req_api_proxy.iter().map(Into::into).collect();
let new_api_proxy = Arc::new(api_proxy);
app_state.app_config.api_proxy.store(Some(Arc::clone(&new_api_proxy)));
let config = app_state.app_config.config.load();
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
let backup_dir = config.get_backup_dir();
let paths = app_state.app_config.paths.load();
if let Some(err) = intern_save_config_api_proxy(backup_dir, &ApiProxyConfigDto::from(new_api_proxy.as_ref()), paths.api_proxy_file_path.as_str()) {
if let Some(err) = intern_save_config_api_proxy(backup_dir.as_ref(), &ApiProxyConfigDto::from(new_api_proxy.as_ref()), paths.api_proxy_file_path.as_str()) {
return (axum::http::StatusCode::INTERNAL_SERVER_ERROR, axum::Json(json!({"error": err.to_string()}))).into_response();
}
}
@@ -318,8 +319,8 @@ pub async fn create_status_check(app_state: &Arc<AppState>) -> StatusCheck {
async fn status(axum::extract::State(app_state): axum::extract::State<Arc<AppState>>) -> axum::response::Response {
let status = create_status_check(&app_state).await;
match serde_json::to_string_pretty(&status) {
Ok(pretty_json) => axum::response::Response::builder().status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(pretty_json).unwrap().into_response(),
Ok(pretty_json) => try_unwrap_body!(axum::response::Response::builder().status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(pretty_json)),
Err(_) => axum::Json(status).into_response(),
}
}
@@ -331,8 +332,8 @@ async fn ipinfo(axum::extract::State(app_state): axum::extract::State<Arc<AppSta
ipv6,
};
return match serde_json::to_string(&ipcheck) {
Ok(json) => axum::response::Response::builder().status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(json).unwrap().into_response(),
Ok(json) => try_unwrap_body!(axum::response::Response::builder().status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string()).body(json)),
Err(_) => axum::Json(ipcheck).into_response(),
};
}
backend/src/api/endpoints/web_index.rs
+10 -12
View File
@@ -1,14 +1,15 @@
use crate::api::api_utils::serve_file;
use crate::api::api_utils::try_unwrap_body;
use crate::api::model::app_state::AppState;
use crate::auth::{AuthBearer, verify_password, create_jwt_admin, create_jwt_user, is_admin, verify_token};
use crate::auth::{create_jwt_admin, create_jwt_user, is_admin, verify_password, verify_token, AuthBearer};
use axum::response::IntoResponse;
use log::error;
use serde_json::json;
use std::path::{Path, PathBuf};
use std::sync::{Arc};
use tower::Service;
use shared::model::{TokenResponse, UserCredential};
use shared::utils::CONSTANTS;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use tower::Service;
fn no_web_auth_token() -> impl axum::response::IntoResponse + Send {
axum::Json(TokenResponse {
@@ -114,10 +115,9 @@ async fn index(
new_content.replace_range(pos..pos + 6, &base_href);
}
return axum::response::Response::builder()
return try_unwrap_body!(axum::response::Response::builder()
.header("Content-Type", mime::TEXT_HTML_UTF_8.as_ref())
.body(new_content.into())
.unwrap();
.body(new_content));
}
Err(err) => {
error!("Failed to read web ui index.hml: {err}");
@@ -151,10 +151,9 @@ async fn index_config(
}
}
if let Ok(json_content) = serde_json::to_string(&json_data) {
return axum::response::Response::builder()
return try_unwrap_body!(axum::response::Response::builder()
.header("Content-Type", mime::APPLICATION_JSON.as_ref())
.body(axum::body::Body::from(json_content))
.unwrap();
.body(axum::body::Body::from(json_content)));
}
}
}
@@ -208,8 +207,7 @@ pub fn index_register_with_path(web_dir_path: &Path, web_ui_path: &str) -> axum:
let new_req = axum::http::Request::builder()
.method(req.method())
.uri(new_uri)
.body(req.into_body())
.unwrap();
.body(req.into_body()).unwrap();
serve_dir.call(new_req)
}
backend/src/api/endpoints/websocket_api.rs
+21 -17
View File
@@ -1,14 +1,14 @@
use std::sync::Arc;
use axum::{
extract::ws::{WebSocketUpgrade, WebSocket, Message},
response::IntoResponse,
};
use axum::extract::ws::CloseFrame;
use log::{error, info};
use shared::model::{ProtocolHandler, ProtocolMessage, WsCloseCode, PROTOCOL_VERSION};
use crate::api::endpoints::v1_api::create_status_check;
use crate::api::model::app_state::AppState;
use crate::auth::verify_token;
use crate::auth::verify_token_admin;
use axum::extract::ws::CloseFrame;
use axum::{
extract::ws::{Message, WebSocket, WebSocketUpgrade},
response::IntoResponse,
};
use log::{error, info};
use shared::model::{ProtocolHandler, ProtocolMessage, WsCloseCode, PROTOCOL_VERSION};
use std::sync::Arc;
// WebSocket upgrade handler
async fn websocket_handler(
@@ -34,8 +34,17 @@ pub fn ws_api_register(web_auth_enabled: bool, web_ui_path: &str) -> axum::Route
}
}
#[inline]
fn verify_auth_admin_token(auth_token: &str, secret_key: Option<&Vec<u8>>) -> bool {
match secret_key.as_ref() {
None => false,
Some(key) => verify_token_admin(auth_token, key.as_slice())
}
}
// WebSocket communication logic
#[allow(clippy::too_many_lines)]
async fn handle_socket(mut socket: WebSocket, app_state: Arc<AppState>, auth: bool) {
let secret_key = if auth {
if let Some(web_auth_config) = &app_state.app_config.config.load().web_ui.as_ref().and_then(|c| c.auth.as_ref()) {
@@ -48,13 +57,8 @@ async fn handle_socket(mut socket: WebSocket, app_state: Arc<AppState>, auth: bo
None
};
let verify_auth_token = |auth_token: &str| {
secret_key.as_ref().map(|key| verify_token(auth_token, key.as_slice()))
};
let mut active_user_change_rx = app_state.active_users.get_active_user_change_channel();
let mut active_provider_change_rx = app_state.active_provider.get_active_provider_change_channel();
let mut active_user_change_rx = app_state.active_users.get_active_user_change_channel();
let mut active_provider_change_rx = app_state.active_provider.get_active_provider_change_channel();
let mut handler = ProtocolHandler::Version(PROTOCOL_VERSION);
@@ -94,7 +98,7 @@ async fn handle_socket(mut socket: WebSocket, app_state: Arc<AppState>, auth: bo
if let Message::Binary(bytes) = msg {
match ProtocolMessage::from_bytes(bytes) {
Ok(ProtocolMessage::StatusRequest(auth_token)) => {
if !auth || verify_auth_token(&auth_token).is_some() {
if !auth || verify_auth_admin_token(&auth_token, secret_key.as_ref()) {
let status = create_status_check(&app_state).await;
if let Ok(response) = ProtocolMessage::StatusResponse(status).to_bytes() {
if socket.send(Message::Binary(response)).await.is_err() {
backend/src/api/endpoints/xmltv_api.rs
+13 -13
View File
@@ -12,19 +12,19 @@ use std::sync::Arc;
use crate::api::api_utils::{get_user_target, serve_file};
use crate::api::model::app_state::AppState;
use crate::api::model::request::UserApiRequest;
use crate::model::Config;
use crate::model::{ConfigTarget, ProxyUserCredentials, TargetOutput};
use crate::model::{Config};
use crate::repository::m3u_repository::m3u_get_epg_file_path;
use crate::repository::storage::get_target_storage_path;
use crate::repository::xtream_repository::{xtream_get_epg_file_path, xtream_get_storage_path};
use crate::utils;
use crate::{utils};
use crate::api::api_utils::try_unwrap_body;
pub fn get_empty_epg_response() -> impl axum::response::IntoResponse + Send {
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK) // Entspricht `HttpResponse::Ok()`
.header(axum::http::header::CONTENT_TYPE, axum::http::HeaderValue::from_static("text/xml"))
.body(axum::body::Body::from(r#"<?xml version="1.0" encoding="utf-8" ?><!DOCTYPE tv SYSTEM "xmltv.dtd"><tv generator-info-name="Xtream Codes" generator-info-url=""></tv>"#)) // Setzt den Body der Antwort
.unwrap()
.body(axum::body::Body::from(r#"<?xml version="1.0" encoding="utf-8" ?><!DOCTYPE tv SYSTEM "xmltv.dtd"><tv generator-info-name="Xtream Codes" generator-info-url=""></tv>"#)))
}
fn time_correct(date_time: &str, correction: &TimeDelta) -> String {
@@ -159,14 +159,14 @@ fn serve_epg_with_timeshift(epg_file: File, offset_minutes: i32) -> impl axum::r
buf.clear();
}
let compressed_data = xml_writer.into_inner().finish().unwrap();
axum::response::Response::builder()
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_OCTET_STREAM.to_string())
.header(axum::http::header::CONTENT_ENCODING, "gzip") // Set Content-Encoding header
.body(axum::body::Body::from(compressed_data))
.unwrap()
.into_response()
match xml_writer.into_inner().finish() {
Ok(compressed_data) =>
try_unwrap_body!(axum::response::Response::builder()
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_OCTET_STREAM.to_string())
.header(axum::http::header::CONTENT_ENCODING, "gzip") // Set Content-Encoding header
.body(axum::body::Body::from(compressed_data))),
Err(err) => (axum::http::StatusCode::INTERNAL_SERVER_ERROR, err.to_string()).into_response()
}
}
/// Handles XMLTV EPG API requests, serving the appropriate EPG file with optional time-shifting based on user configuration.
backend/src/api/endpoints/xtream_api.rs
+16 -24
View File
@@ -38,6 +38,7 @@ use std::path::Path;
use std::str::FromStr;
use std::sync::Arc;
use shared::model::{PlaylistItemType, XtreamCluster, FieldGetAccessor, PlaylistEntry, TargetType, UserConnectionPermission, ProxyType, get_backdrop_path_value, XtreamPlaylistItem};
use crate::api::api_utils::try_unwrap_body;
#[derive(Serialize, Deserialize, Debug, Copy, Clone, Eq, PartialEq)]
pub enum ApiStreamContext {
@@ -626,12 +627,10 @@ async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserC
if user.proxy == ProxyType::Redirect && cluster == XtreamCluster::Live {
return redirect(&info_url).into_response();
} else if let Ok(content) = xtream::get_xtream_stream_info(Arc::clone(&app_state.http_client.load()), &app_state.app_config, user, &input, target, &pli, info_url.as_str(), cluster).await {
return axum::response::Response::builder()
return try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from(content))
.unwrap()
.into_response()
.body(axum::body::Body::from(content)))
}
}
}
@@ -640,27 +639,21 @@ async fn xtream_get_stream_info_response(app_state: &AppState, user: &ProxyUserC
return match cluster {
XtreamCluster::Video => {
let content = create_vod_info_from_item(target, user, &pli, virtual_record.last_updated);
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from(content))
.unwrap()
.into_response()
.body(axum::body::Body::from(content)))
}
XtreamCluster::Live | XtreamCluster::Series => axum::response::Response::builder()
XtreamCluster::Live | XtreamCluster::Series => try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from("{}".as_bytes()))
.unwrap()
.into_response(),
.body(axum::body::Body::from("{}".as_bytes()))),
};
}
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from("{}".as_bytes()))
.unwrap()
.into_response()
.body(axum::body::Body::from("{}".as_bytes())))
}
async fn xtream_get_short_epg(app_state: &AppState, user: &ProxyUserCredentials, target: &ConfigTarget, stream_id: &str, limit: &str) -> impl IntoResponse + Send {
@@ -721,9 +714,9 @@ async fn xtream_player_api_handle_content_action(config: &Config, target_name: &
}
return Some(serve_file(&file_path, mime::APPLICATION_JSON).await.into_response());
} else if let Some(payload) = content {
return Some(axum::response::Response::builder()
return Some(try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.body(payload).unwrap().into_response());
.body(payload)));
}
return Some(api_utils::empty_json_list_response().into_response());
}
@@ -758,10 +751,10 @@ async fn xtream_get_catchup_response(app_state: &AppState, target: &ConfigTarget
serde_json::to_string(&doc)
.map_or_else(
|_| axum::http::StatusCode::BAD_REQUEST.into_response(),
|result| axum::response::Response::builder()
|result| try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(result).unwrap().into_response())
.body(result)))
}
macro_rules! skip_json_response_if_flag_set {
@@ -848,8 +841,7 @@ async fn xtream_player_api(
skip_flag_optional!(skip_vod, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Video, &app_state.app_config, &target, category_id, &user).await),
crate::model::XC_ACTION_GET_SERIES =>
skip_flag_optional!(skip_series, xtream_repository::xtream_load_rewrite_playlist(XtreamCluster::Series, &app_state.app_config, &target, category_id, &user).await),
_ => Some(Err(info_err!(format!("Cant find content: {action} for target: {}", &target.name))
)),
_ => Some(Err(info_err!(format!("Cant find content: {action} for target: {}", &target.name)))),
};
match result {
@@ -858,10 +850,10 @@ async fn xtream_player_api(
Ok(xtream_iter) => {
// Convert the iterator into a stream of `Bytes`
let content_stream = xtream_create_content_stream(xtream_iter);
axum::response::Response::builder()
try_unwrap_body!(axum::response::Response::builder()
.status(axum::http::StatusCode::OK)
.header(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())
.body(axum::body::Body::from_stream(content_stream)).unwrap().into_response()
.body(axum::body::Body::from_stream(content_stream)))
}
Err(err) => {
error!("Failed response for xtream target: {} action: {} error: {}", &target.name, action, err);
backend/src/api/main_api.rs
+16 -16
View File
@@ -1,8 +1,11 @@
use crate::api::api_utils::{get_build_time, get_server_time};
use crate::api::config_watch::exec_config_watch;
use crate::api::endpoints::hdhomerun_api::hdhr_api_register;
use crate::api::endpoints::hls_api::hls_api_register;
use crate::api::endpoints::m3u_api::m3u_api_register;
use crate::api::endpoints::v1_api::v1_api_register;
use crate::api::endpoints::web_index::{index_register_with_path, index_register_without_path};
use crate::api::endpoints::websocket_api::ws_api_register;
use crate::api::endpoints::xmltv_api::xmltv_api_register;
use crate::api::endpoints::xtream_api::xtream_api_register;
use crate::api::model::active_provider_manager::ActiveProviderManager;
@@ -10,23 +13,20 @@ use crate::api::model::active_user_manager::ActiveUserManager;
use crate::api::model::app_state::{create_cache, create_http_client, AppState, CancelTokens, HdHomerunAppState};
use crate::api::model::download::DownloadQueue;
use crate::api::model::streams::shared_stream_manager::SharedStreamManager;
use crate::api::scheduler::{exec_scheduler};
use crate::api::scheduler::exec_scheduler;
use crate::api::serve::serve;
use crate::model::{AppConfig, Config, ProcessTargets, RateLimitConfig};
use crate::model::{Healthcheck};
use crate::model::Healthcheck;
use crate::processing::processor::playlist;
use crate::VERSION;
use arc_swap::{ArcSwap, ArcSwapOption};
use axum::Router;
use log::{error, info};
use std::io::ErrorKind;
use std::path::PathBuf;
use std::sync::Arc;
use arc_swap::{ArcSwap, ArcSwapOption};
use axum::Router;
use tokio_util::sync::CancellationToken;
use tower_governor::key_extractor::SmartIpKeyExtractor;
use crate::api::api_utils::{get_build_time, get_server_time};
use crate::api::config_watch::exec_config_watch;
use crate::api::endpoints::websocket_api::ws_api_register;
use crate::api::serve::serve;
use crate::VERSION;
fn get_web_dir_path(web_ui_enabled: bool, web_root: &str) -> Result<PathBuf, std::io::Error> {
let web_dir = web_root.to_string();
@@ -232,7 +232,7 @@ pub async fn start_server(app_config: Arc<AppConfig>, targets: Arc<ProcessTarget
router = router.layer(create_cors_layer())
.layer(create_compression_layer());
//.layer(tower_http::trace::TraceLayer::new_for_http()); // `Logger::default()`
//.layer(tower_http::trace::TraceLayer::new_for_http()); // `Logger::default()`
// router = router.layer(axum::middleware::from_fn(log_routes));
let router: axum::Router<()> = router.with_state(shared_data.clone());
@@ -246,15 +246,15 @@ pub async fn start_server(app_config: Arc<AppConfig>, targets: Arc<ProcessTarget
fn add_rate_limiter(router: Router<Arc<AppState>>, rate_limit_cfg: &RateLimitConfig) -> Router<Arc<AppState>> {
if rate_limit_cfg.enabled {
let governor_conf = Arc::new(tower_governor::governor::GovernorConfigBuilder::default()
let governor_conf = tower_governor::governor::GovernorConfigBuilder::default()
.key_extractor(SmartIpKeyExtractor)
.per_millisecond(rate_limit_cfg.period_millis)
.burst_size(rate_limit_cfg.burst_size)
.finish()
.unwrap());
router.layer(tower_governor::GovernorLayer {
config: governor_conf,
})
.finish();
if let Some(config) = governor_conf { router.layer(tower_governor::GovernorLayer { config: Arc::new(config) }) } else {
error!("Failed to initialize rate limiter");
router
}
} else {
router
}
backend/src/api/model/active_provider_manager.rs
+6 -5
View File
@@ -335,11 +335,12 @@ impl MultiProviderLineup {
let mut idx = index.load(Ordering::Relaxed) % provider_count;
let start = idx;
for _ in start..provider_count {
let p = pg.get(idx).unwrap();
let result = p.get_next(grace, grace_period_timeout_secs).await;
if result.is_some() {
index.store((idx + 1) % provider_count, Ordering::Relaxed);
return result;
if let Some(p) = pg.get(idx) {
let result = p.get_next(grace, grace_period_timeout_secs).await;
if result.is_some() {
index.store((idx + 1) % provider_count, Ordering::Relaxed);
return result;
}
}
idx = (idx + 1) % provider_count;
}
backend/src/api/model/download.rs
+2 -2
View File
@@ -52,10 +52,10 @@ fn get_download_directory(download_cfg: &VideoDownloadConfig, filestem: &str) ->
}
}
let dir_name = CONSTANTS.re_remove_filename_ending.replace(stem, "");
let file_dir: PathBuf = [download_cfg.directory.as_ref().unwrap(), dir_name.as_ref()].iter().collect();
let file_dir: PathBuf = [download_cfg.directory.as_str(), dir_name.as_ref()].iter().collect();
file_dir
} else {
PathBuf::from(download_cfg.directory.as_ref().unwrap())
PathBuf::from(download_cfg.directory.as_str())
}
}
backend/src/api/model/model_utils.rs
+4 -1
View File
@@ -7,7 +7,10 @@ use shared::utils::{filter_response_header};
pub fn get_response_headers(headers: &HeaderMap) -> Vec<(String, String)> {
let mut response_headers: Vec<(String, String)> = headers.iter()
.filter(|(key, _)| filter_response_header(key.as_str()))
.map(|(key, value)| (key.to_string(), value.to_str().unwrap().to_string())).collect();
.filter_map(|(key, value)| {
value.to_str().ok().map(|v| (key.to_string(), v.to_string()))
})
.collect();
response_headers.push((axum::http::header::CONNECTION.as_str().to_string(), "keep-alive".to_string()));
response_headers
}
backend/src/api/model/provider_config.rs
+1 -1
View File
@@ -111,7 +111,7 @@ impl ProviderConfig {
}
fn notify_connection_change(&self, new_connections: usize) {
self.connection_change_tx.send((self.name.clone(), new_connections)).unwrap();
let _ = self.connection_change_tx.send((self.name.clone(), new_connections));
}
#[inline]
backend/src/api/model/streams/active_client_stream.rs
+27 -18
View File
@@ -4,18 +4,18 @@ use crate::api::model::active_user_manager::{ActiveUserManager, UserConnectionGu
use crate::api::model::app_state::AppState;
use crate::api::model::stream::BoxedProviderStream;
use crate::api::model::stream_error::StreamError;
use crate::api::model::streams::timed_client_stream::TimedClientStream;
use crate::api::model::streams::transport_stream_buffer::TransportStreamBuffer;
use crate::model::{ProxyUserCredentials};
use crate::model::ProxyUserCredentials;
use bytes::Bytes;
use futures::Stream;
use futures::StreamExt;
use log::{error, info};
use shared::model::UserConnectionPermission;
use std::pin::Pin;
use std::sync::atomic::AtomicU8;
use std::sync::{Arc, Mutex};
use std::task::{Poll, Waker};
use crate::api::model::streams::timed_client_stream::TimedClientStream;
use futures::{StreamExt};
use shared::model::UserConnectionPermission;
const INNER_STREAM: u8 = 0_u8;
const GRACE_BLOCK_STREAM: u8 = 1_u8;
@@ -35,10 +35,10 @@ pub(in crate::api) struct ActiveClientStream {
impl ActiveClientStream {
pub(crate) fn new(mut stream_details: StreamDetails,
app_state: &AppState,
user: &ProxyUserCredentials,
connection_permission: UserConnectionPermission,
addr: &str) -> Self {
app_state: &AppState,
user: &ProxyUserCredentials,
connection_permission: UserConnectionPermission,
addr: &str) -> Self {
let active_user = app_state.active_users.clone();
let active_provider = app_state.active_provider.clone();
if connection_permission == UserConnectionPermission::Exhausted {
@@ -59,15 +59,24 @@ impl ActiveClientStream {
c.provider_connections_exhausted.clone()
));
let stream = stream_details.stream.take().unwrap();
let stream = match app_state.app_config.config.load().sleep_timer_mins {
None => stream,
Some(mins) => {
let secs = u32::try_from((u64::from(mins) * 60).min(u64::from(u32::MAX))).unwrap_or(0);
if secs > 0 {
TimedClientStream::new(stream, secs).boxed()
} else {
stream
let stream = match stream_details.stream.take() {
None => {
if let Some(guard) = stream_details.provider_connection_guard.as_ref() {
guard.release();
}
futures::stream::empty::<Result<Bytes, StreamError>>().boxed()
}
Some(stream) => {
match app_state.app_config.config.load().sleep_timer_mins {
None => stream,
Some(mins) => {
let secs = u32::try_from((u64::from(mins) * 60).min(u64::from(u32::MAX))).unwrap_or(0);
if secs > 0 {
TimedClientStream::new(stream, secs).boxed()
} else {
stream
}
}
}
}
};
@@ -193,7 +202,7 @@ impl Stream for ActiveClientStream {
};
if let Some(buffer) = buffer_opt {
return Poll::Ready(Some(Ok(buffer.next_chunk())));
return Poll::Ready(Some(Ok(buffer.next_chunk())));
}
Poll::Ready(None)
backend/src/api/model/streams/provider_stream.rs
+2 -1
View File
@@ -7,6 +7,7 @@ use reqwest::StatusCode;
use axum::response::IntoResponse;
use crate::api::model::stream::ProviderStreamResponse;
use crate::api::model::streams::transport_stream_buffer::TransportStreamBuffer;
use crate::api::api_utils::try_unwrap_body;
#[derive(Debug, Copy, Clone)]
pub enum CustomVideoStreamType {
@@ -65,7 +66,7 @@ pub fn create_custom_video_stream_response(config: &AppConfig, video_response: C
for (key, value) in headers {
builder = builder.header(key, value);
}
return builder.body(axum::body::Body::from_stream(stream)).unwrap().into_response();
return try_unwrap_body!(builder.body(axum::body::Body::from_stream(stream)));
}
axum::http::StatusCode::FORBIDDEN.into_response()
}
backend/src/model/config/api_proxy.rs
+4 -4
View File
@@ -108,9 +108,9 @@ impl ApiProxyConfig {
errors.push(err.to_string());
} else {
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&cfg.config);
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
let backup_dir = config.get_backup_dir();
self.user = vec![];
if let Err(err) = utils::save_api_proxy(api_proxy_file, backup_dir, &ApiProxyConfigDto::from(&*self)) {
if let Err(err) = utils::save_api_proxy(api_proxy_file, backup_dir.as_ref(), &ApiProxyConfigDto::from(&*self)) {
errors.push(format!("Error saving api proxy file: {err}"));
}
}
@@ -144,8 +144,8 @@ impl ApiProxyConfig {
}
let config = <Arc<ArcSwap<Config>> as Access<Config>>::load(&cfg.config);
let backup_dir = config.backup_dir.as_ref().unwrap().as_str();
if let Err(err) = save_api_proxy(api_proxy_file, backup_dir, &ApiProxyConfigDto::from(&*self)) {
let backup_dir = config.get_backup_dir();
if let Err(err) = save_api_proxy(api_proxy_file, backup_dir.as_ref(), &ApiProxyConfigDto::from(&*self)) {
errors.push(format!("Error saving api proxy file: {err}"));
} else {
backup_api_user_db_file(cfg, &user_db_path);
backend/src/model/config/app.rs
+2 -1
View File
@@ -369,7 +369,8 @@ impl AppConfig {
TuliproxError::new(TuliproxErrorKind::Info, "API proxy config not loaded".to_string())
}) {
let server_info_list = api_proxy.server.clone();
server_info_list.iter().find(|c| c.name.eq(server_info_name)).map_or_else(|| server_info_list.first().unwrap().clone(), Clone::clone)
server_info_list.iter().find(|c| c.name.eq(server_info_name))
.map_or_else(|| server_info_list.first().unwrap().clone(), Clone::clone)
} else {
panic!("ApiProxyServer info not found");
}
backend/src/model/config/base.rs
+8 -2
View File
@@ -1,3 +1,4 @@
use std::borrow::Cow;
use std::path::{Path, PathBuf};
use log::{error, info};
use path_clean::PathClean;
@@ -8,6 +9,7 @@ use crate::model::{macros, ConfigApi, ReverseProxyConfig, ScheduleConfig};
use crate::model::{HdHomeRunConfig, IpCheckConfig, LogConfig, MessagingConfig, ProxyConfig, VideoConfig, WebUiConfig};
use crate::{utils};
const DEFAULT_BACKUP_DIR: &str = "backup";
fn create_directories(cfg: &Config, temp_path: &Path) {
// Collect the paths into a vector.
@@ -15,7 +17,7 @@ fn create_directories(cfg: &Config, temp_path: &Path) {
Some(cfg.working_dir.clone()),
cfg.backup_dir.clone(),
cfg.user_config_dir.clone(),
cfg.video.as_ref().and_then(|v| v.download.as_ref()).and_then(|d| d.directory.clone()),
cfg.video.as_ref().and_then(|v| v.download.as_ref()).map(|d| d.directory.to_string()),
cfg.reverse_proxy.as_ref().and_then(|r| r.cache.as_ref().and_then(|c| if c.enabled { Some(c.dir.to_string()) } else { None }))
];
@@ -85,10 +87,14 @@ impl Config {
});
}
set_directory(&mut self.backup_dir, "backup", &self.working_dir);
set_directory(&mut self.backup_dir, DEFAULT_BACKUP_DIR, &self.working_dir);
set_directory(&mut self.user_config_dir, "user_config", &self.working_dir);
}
pub fn get_backup_dir(&self) -> Cow<str> {
self.backup_dir.as_ref().map_or_else(|| Cow::Borrowed(DEFAULT_BACKUP_DIR), |v| Cow::Borrowed(v))
}
fn prepare_api_web_root(&mut self) {
if !self.api.web_root.is_empty() {
self.api.web_root = utils::make_absolute_path(&self.api.web_root, &self.working_dir);
backend/src/model/config/video_download.rs
+4 -5
View File
@@ -6,20 +6,19 @@ use crate::model::macros;
#[derive(Debug, Clone)]
pub struct VideoDownloadConfig {
pub headers: HashMap<String, String>,
pub directory: Option<String>,
pub directory: String,
pub organize_into_directories: bool,
pub episode_pattern: Option<Regex>,
}
macros::from_impl!(VideoDownloadConfig);
impl From<&VideoDownloadConfigDto> for VideoDownloadConfig {
fn from(dto: &VideoDownloadConfigDto) -> Self {
Self {
headers: dto.headers.clone(),
directory: dto.directory.clone(),
directory: dto.directory.as_ref().map_or_else(|| "downloads".to_string(), ToString::to_string),
organize_into_directories: dto.organize_into_directories,
episode_pattern: dto.episode_pattern.as_ref().map(|s| Regex::new(s).unwrap()),
episode_pattern: dto.episode_pattern.as_ref().and_then(|s| Regex::new(s).ok()),
}
}
}
@@ -28,7 +27,7 @@ impl From<&VideoDownloadConfig> for VideoDownloadConfigDto {
fn from(instance: &VideoDownloadConfig) -> Self {
Self {
headers: instance.headers.clone(),
directory: instance.directory.clone(),
directory: Some(instance.directory.clone()),
organize_into_directories: instance.organize_into_directories,
episode_pattern: instance.episode_pattern.as_ref().map(std::string::ToString::to_string),
}
backend/src/model/xtream.rs
+3 -3
View File
@@ -112,13 +112,13 @@ macro_rules! add_opt_i64_property_if_exists {
macro_rules! add_opt_f64_property_if_exists {
($vec:expr, $prop:expr, $prop_name:expr) => {
$prop.as_ref().map(|v| $vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from(*v)).unwrap())));
$prop.as_ref().map(|v| $vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from(*v)).unwrap_or_else(|| serde_json::Number::from(0)))));
}
}
macro_rules! add_f64_property_if_exists {
($vec:expr, $prop:expr, $prop_name:expr) => {
$vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from($prop)).unwrap()));
$vec.insert(String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from($prop)).unwrap_or_else(|| serde_json::Number::from(0))));
}
}
@@ -427,7 +427,7 @@ fn append_prepared_series_properties(add_props: Option<&Map<String, Value>>, doc
match props.get("rating") {
Some(value) => {
document.insert("rating".to_string(), match value {
Value::Number(val) => Value::String(format!("{:.0}", val.as_f64().unwrap())),
Value::Number(val) => Value::String(format!("{:.0}", val.as_f64().unwrap_or(0f64))),
Value::String(val) => Value::String(val.to_string()),
_ => Value::String("0".to_string()),
});
backend/src/processing/parser/m3u.rs
+38 -28
View File
@@ -1,5 +1,5 @@
use crate::model::{Config, ConfigInput};
use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, XtreamCluster};
use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, XtreamCluster, DEFAULT_VIDEO_EXTENSIONS};
use shared::utils::extract_id_from_url;
use std::borrow::BorrowMut;
@@ -116,26 +116,25 @@ fn process_header(input_name: &str, video_suffixes: &[&str], content: &str, url:
let mut provider_id = None::<String>;
let mut c = skip_digit(&mut it);
loop {
if c.is_none() {
break;
}
let chr = c.unwrap();
if chr.is_whitespace() {
// skip
} else if chr == ',' {
plih.title = get_value(&mut stack, &mut it);
} else {
stack.push(chr);
let token = token_till(&mut stack, &mut it, '=', true);
if let Some(t) = token {
let value = token_value(&mut stack, &mut it);
let token = t.to_lowercase();
if token.as_str() == "xui-id" {
if !value.is_empty() {
provider_id = Some(value);
}
match c {
None=> break,
Some(chr) => {
if chr.is_whitespace() {
// skip
} else if chr == ',' {
plih.title = get_value(&mut stack, &mut it);
} else {
process_header_fields!(plih, token.as_str(),
stack.push(chr);
let token = token_till(&mut stack, &mut it, '=', true);
if let Some(t) = token {
let value = token_value(&mut stack, &mut it);
let token = t.to_lowercase();
if token.as_str() == "xui-id" {
if !value.is_empty() {
provider_id = Some(value);
}
} else {
process_header_fields!(plih, token.as_str(),
(id, "tvg-id"),
(group, "group-title"),
(name, "tvg-name"),
@@ -146,6 +145,8 @@ fn process_header(input_name: &str, video_suffixes: &[&str], content: &str, url:
(logo_small, "tvg-logo-small"),
(time_shift, "timeshift"),
(rec, "tvg-rec"); value);
}
}
}
}
}
@@ -188,7 +189,6 @@ fn process_header(input_name: &str, video_suffixes: &[&str], content: &str, url:
plih
}
pub fn consume_m3u<'a, I, F: FnMut(PlaylistItem)>(cfg: &Config, input: &ConfigInput, lines: I, mut visit: F)
where
I: Iterator<Item=&'a str>,
@@ -197,7 +197,12 @@ where
let mut group: Option<String> = None;
let input_name = input.name.as_str();
let video_suffixes = cfg.video.as_ref().unwrap().extensions.iter().map(String::as_str).collect::<Vec<&str>>();
let video_suffixes = match cfg.video.as_ref() {
Some(config) => {
config.extensions.iter().map(String::as_str).collect::<Vec<&str>>()
},
None => DEFAULT_VIDEO_EXTENSIONS.to_vec()
};
for line in lines {
if line.starts_with("#EXTINF") {
header = Some(String::from(line));
@@ -248,18 +253,23 @@ where
sort_order_idx += 1;
}
std::collections::hash_map::Entry::Occupied(o) => {
sort_order.get_mut(*o.get()).unwrap().push(item);
if let Some(order) = sort_order.get_mut(*o.get()) {
order.push(item);
}
}
}
});
let mut grp_id = 0;
let result: Vec<PlaylistGroup> = sort_order.into_iter().map(|channels| {
let result: Vec<PlaylistGroup> = sort_order.into_iter().filter_map(|channels| {
// create a group based on the first playlist item
let channel = channels.first();
let (cluster, group_title) = channel.map(|pli|
(pli.header.xtream_cluster, &pli.header.group)).unwrap();
grp_id += 1;
PlaylistGroup { id: grp_id, xtream_cluster: cluster, title: group_title.to_string(), channels }
if let Some((cluster, group_title)) = channel.map(|pli|
(pli.header.xtream_cluster, &pli.header.group)) {
grp_id += 1;
Some(PlaylistGroup { id: grp_id, xtream_cluster: cluster, title: group_title.to_string(), channels })
} else {
None
}
}).collect();
result
}
backend/src/processing/parser/xmltv.rs
+38 -25
View File
@@ -2,6 +2,7 @@ use crate::model::{Epg, TVGuide, XmlTag, XmlTagIcon, EPG_ATTRIB_CHANNEL, EPG_ATT
use crate::model::{EpgSmartMatchConfig, PersistedEpgSource};
use crate::processing::processor::epg::EpgIdCache;
use crate::utils::compressed_file_reader::CompressedFileReader;
use dashmap::DashMap;
use deunicode::deunicode;
use quick_xml::events::{BytesStart, BytesText, Event};
use quick_xml::Reader;
@@ -14,7 +15,6 @@ use std::collections::HashMap;
use std::mem;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use dashmap::DashMap;
/// Splits a string at the first delimiter if the prefix matches a known country code.
///
@@ -143,13 +143,14 @@ impl TVGuide {
if !matched && fuzzy_matching {
let (fuzzy_matched, matched_normalized_name) = Self::find_best_fuzzy_match(id_cache, tag);
if fuzzy_matched {
let key = matched_normalized_name.unwrap();
let id = epg_id.to_string();
id_cache.normalized.entry(key).and_modify(|entry| {
entry.replace(id.clone());
id_cache.channel_epg_id.insert(Cow::Owned(id));
matched = true;
});
if let Some(key) = matched_normalized_name {
let id = epg_id.to_string();
id_cache.normalized.entry(key).and_modify(|entry| {
entry.replace(id.clone());
id_cache.channel_epg_id.insert(Cow::Owned(id));
matched = true;
});
}
}
}
matched
@@ -190,9 +191,10 @@ impl TVGuide {
#[allow(clippy::cast_sign_loss)]
let mjw = min(100, (match_jw * 100.0).round() as u16);
if mjw >= match_threshold {
let mut lock = data.lock().unwrap();
if lock.0 < mjw {
*lock = (mjw, Some(Cow::Borrowed(norm_key)));
if let Ok(mut lock) = data.lock() {
if lock.0 < mjw {
*lock = (mjw, Some(Cow::Borrowed(norm_key)));
}
}
if mjw > best_match_threshold {
return true; // (true, matched_normalized_epg_id.map(|s| s.to_string()));
@@ -206,8 +208,9 @@ impl TVGuide {
// is there an early exit strategy ???
if early_exit_flag.load(Ordering::SeqCst) {
let result = data.lock().unwrap().1.take();
return (true, result.as_ref().map(std::string::ToString::to_string));
if let Ok(mut result) = data.lock() {
return (true, result.1.take().as_ref().map(ToString::to_string));
}
}
(false, None)
}
@@ -354,13 +357,14 @@ where
}
}
fn handle_text_tag(stack: &mut [XmlTag], e: &BytesText) {
if !stack.is_empty() {
if let Ok(text) = e.unescape() {
let t = text.trim();
if !t.is_empty() {
stack.last_mut().unwrap().value = Some(t.to_string());
if let Some(tag) = stack.last_mut() {
tag.value = Some(t.to_string());
}
}
}
}
@@ -403,14 +407,16 @@ fn collect_tag_attributes(e: &BytesStart, is_channel: bool, is_program: bool) ->
let attributes = e.attributes().filter_map(Result::ok)
.filter_map(|a| {
let key = String::from_utf8_lossy(a.key.as_ref()).to_string();
let mut value = String::from(a.unescape_value().unwrap().as_ref());
if (is_channel && key == EPG_ATTRIB_ID) || (is_program && key == EPG_ATTRIB_CHANNEL) {
value = value.to_lowercase().to_string();
}
if value.is_empty() {
None
if let Ok(value) = a.unescape_value().as_ref() {
if value.is_empty() {
None
} else if (is_channel && key == EPG_ATTRIB_ID) || (is_program && key == EPG_ATTRIB_CHANNEL) {
Some((key, value.to_lowercase().to_string()))
} else {
Some((key, value.to_string()))
}
} else {
Some((key, value))
None
}
}).collect::<HashMap<String, String>>();
attributes
@@ -466,13 +472,20 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option<Epg> {
}
});
epg_children.lock().unwrap().extend(children);
if let Ok(mut guard) = epg_children.lock() {
guard.extend(children);
}
});
let children = if let Ok(mut children) = epg_children.lock() {
mem::take(&mut *children)
} else {
vec![]
};
let epg = Epg {
logo_override: false,
priority: 0,
attributes: epg_attributes,
children: mem::take(&mut *epg_children.lock().unwrap()),
children,
};
Some(epg)
}
@@ -492,7 +505,7 @@ mod tests {
/// parse_normalize().unwrap();
/// ```
fn parse_normalize() {
let epg_normalize_dto = EpgSmartMatchConfigDto {..Default::default()};
let epg_normalize_dto = EpgSmartMatchConfigDto { ..Default::default() };
let epg_normalize = EpgSmartMatchConfig::from(epg_normalize_dto);
let normalized = normalize_channel_name("Love Nature", &epg_normalize);
assert_eq!(normalized, "lovenature".to_string());
backend/src/processing/processor/epg.rs
+4 -3
View File
@@ -146,8 +146,9 @@ fn assign_channel_epg(new_epg: &mut Vec<Epg>, fp: &mut FetchedPlaylist, id_cache
for epg_source in epg_sources {
// icon tags
let icon_tags: HashMap<&String, &XmlTag> = epg_source.children.iter()
.filter(|tag| tag.icon != XmlTagIcon::Undefined && tag.get_attribute_value(EPG_ATTRIB_ID).is_some())
.map(|t| (t.get_attribute_value(EPG_ATTRIB_ID).unwrap(), t)).collect();
.filter(|tag| tag.icon != XmlTagIcon::Undefined)
.filter_map(|tag| tag.get_attribute_value(EPG_ATTRIB_ID).map(|id| (id, tag)))
.collect();
let assign_values = |chan: &mut PlaylistItem| {
if id_cache.smart_match_enabled && chan.header.epg_channel_id.is_none() {
@@ -169,7 +170,7 @@ fn assign_channel_epg(new_epg: &mut Vec<Epg>, fp: &mut FetchedPlaylist, id_cache
if let Some(epg_channel_id) = chan.header.epg_channel_id.as_ref() {
if !icon_assigned.contains(epg_channel_id) &&
(epg_source.logo_override || chan.header.logo.is_empty() || chan.header.logo_small.is_empty()) {
if let Some(icon_tag) = icon_tags.get(chan.header.epg_channel_id.as_ref().unwrap()) {
if let Some(icon_tag) = icon_tags.get(epg_channel_id) {
if let XmlTagIcon::Src(icon) = &icon_tag.icon {
icon_assigned.insert(epg_channel_id.to_string());
if epg_source.logo_override || chan.header.logo.is_empty() {
backend/src/processing/processor/playlist.rs
+79 -68
View File
@@ -234,70 +234,71 @@ fn is_target_enabled(target: &ConfigTarget, user_targets: &ProcessTargets) -> bo
async fn process_source(client: Arc<reqwest::Client>, cfg: Arc<AppConfig>, source_idx: usize, user_targets: Arc<ProcessTargets>) -> (Vec<InputStats>, Vec<TargetStats>, Vec<TuliproxError>) {
let sources = cfg.sources.load();
let source = sources.get_source_at(source_idx).unwrap();
let mut errors = vec![];
let mut input_stats = HashMap::<String, InputStats>::new();
let mut target_stats = Vec::<TargetStats>::new();
let mut source_playlists = Vec::with_capacity(128);
// Download the sources
let mut source_downloaded = false;
for input in &source.inputs {
if is_input_enabled(input, &user_targets) {
let config = cfg.config.load();
let working_dir = &config.working_dir;
if let Some(source) = sources.get_source_at(source_idx) {
let mut source_playlists = Vec::with_capacity(128);
// Download the sources
let mut source_downloaded = false;
for input in &source.inputs {
if is_input_enabled(input, &user_targets) {
let config = cfg.config.load();
let working_dir = &config.working_dir;
source_downloaded = true;
let start_time = Instant::now();
let (mut playlistgroups, mut error_list) = match input.input_type {
InputType::M3u => m3u::get_m3u_playlist(Arc::clone(&client), &config, input, working_dir).await,
InputType::Xtream => xtream::get_xtream_playlist(&config, Arc::clone(&client), input, working_dir).await,
InputType::M3uBatch | InputType::XtreamBatch => (vec![], vec![])
};
let (tvguide, mut tvguide_errors) = if error_list.is_empty() {
epg::get_xmltv(Arc::clone(&client), input, working_dir).await
} else {
(None, vec![])
};
errors.append(&mut error_list);
errors.append(&mut tvguide_errors);
let group_count = playlistgroups.len();
let channel_count = playlistgroups.iter()
.map(|group| group.channels.len())
.sum();
let input_name = &input.name;
if playlistgroups.is_empty() {
info!("Source is empty {input_name}");
errors.push(notify_err!(format!("Source is empty {input_name}")));
} else {
playlistgroups.iter_mut().for_each(PlaylistGroup::on_load);
source_playlists.push(
FetchedPlaylist {
input,
playlistgroups,
epg: tvguide,
}
);
}
let elapsed = start_time.elapsed().as_secs();
input_stats.insert(input_name.to_string(), create_input_stat(group_count, channel_count, error_list.len(),
input.input_type, input_name, elapsed));
}
}
if source_downloaded {
if source_playlists.is_empty() {
debug!("Source at index {source_idx} is empty");
errors.push(notify_err!(format!("Source at {source_idx} is empty")));
} else {
debug_if_enabled!("Source has {} groups", source_playlists.iter().map(|fpl| fpl.playlistgroups.len()).sum::<usize>());
for target in &source.targets {
if is_target_enabled(target, &user_targets) {
match process_playlist_for_target(&cfg, Arc::clone(&client), &mut source_playlists, target, &mut input_stats, &mut errors).await {
Ok(()) => {
target_stats.push(TargetStats::success(&target.name));
source_downloaded = true;
let start_time = Instant::now();
let (mut playlistgroups, mut error_list) = match input.input_type {
InputType::M3u => m3u::get_m3u_playlist(Arc::clone(&client), &config, input, working_dir).await,
InputType::Xtream => xtream::get_xtream_playlist(&config, Arc::clone(&client), input, working_dir).await,
InputType::M3uBatch | InputType::XtreamBatch => (vec![], vec![])
};
let (tvguide, mut tvguide_errors) = if error_list.is_empty() {
epg::get_xmltv(Arc::clone(&client), input, working_dir).await
} else {
(None, vec![])
};
errors.append(&mut error_list);
errors.append(&mut tvguide_errors);
let group_count = playlistgroups.len();
let channel_count = playlistgroups.iter()
.map(|group| group.channels.len())
.sum();
let input_name = &input.name;
if playlistgroups.is_empty() {
info!("Source is empty {input_name}");
errors.push(notify_err!(format!("Source is empty {input_name}")));
} else {
playlistgroups.iter_mut().for_each(PlaylistGroup::on_load);
source_playlists.push(
FetchedPlaylist {
input,
playlistgroups,
epg: tvguide,
}
Err(mut err) => {
target_stats.push(TargetStats::failure(&target.name));
errors.append(&mut err);
);
}
let elapsed = start_time.elapsed().as_secs();
input_stats.insert(input_name.to_string(), create_input_stat(group_count, channel_count, error_list.len(),
input.input_type, input_name, elapsed));
}
}
if source_downloaded {
if source_playlists.is_empty() {
debug!("Source at index {source_idx} is empty");
errors.push(notify_err!(format!("Source at {source_idx} is empty")));
} else {
debug_if_enabled!("Source has {} groups", source_playlists.iter().map(|fpl| fpl.playlistgroups.len()).sum::<usize>());
for target in &source.targets {
if is_target_enabled(target, &user_targets) {
match process_playlist_for_target(&cfg, Arc::clone(&client), &mut source_playlists, target, &mut input_stats, &mut errors).await {
Ok(()) => {
target_stats.push(TargetStats::success(&target.name));
}
Err(mut err) => {
target_stats.push(TargetStats::failure(&target.name));
errors.append(&mut err);
}
}
}
}
@@ -351,13 +352,17 @@ async fn process_sources(client: Arc<reqwest::Client>, config: &Arc<AppConfig>,
let handles = &mut handle_list;
let process = move || {
// TODO better way ?
let rt = tokio::runtime::Runtime::new().unwrap();
rt.block_on(async {
let (input_stats, target_stats, mut res_errors) = process_source(Arc::clone(&http_client), cfg, index, usr_trgts).await;
shared_errors.lock().await.append(&mut res_errors);
let process_stats = SourceStats::new(input_stats, target_stats);
shared_stats.lock().await.push(process_stats);
});
match tokio::runtime::Runtime::new() {
Ok(rt) => {
rt.block_on(async {
let (input_stats, target_stats, mut res_errors) = process_source(Arc::clone(&http_client), cfg, index, usr_trgts).await;
shared_errors.lock().await.append(&mut res_errors);
let process_stats = SourceStats::new(input_stats, target_stats);
shared_stats.lock().await.push(process_stats);
});
},
Err(err) => error!("Could not create runtime !!! {err}"),
}
};
handles.push(thread::spawn(process));
if handles.len() >= thread_num as usize {
@@ -374,7 +379,11 @@ async fn process_sources(client: Arc<reqwest::Client>, config: &Arc<AppConfig>,
for handle in handle_list {
let _ = handle.join();
}
(Arc::try_unwrap(stats).unwrap().into_inner(), Arc::try_unwrap(errors).unwrap().into_inner())
if let (Ok(s), Ok(e)) = (Arc::try_unwrap(stats), Arc::try_unwrap(errors)) {
(s.into_inner(), e.into_inner())
} else {
(vec![], vec![])
}
}
pub type ProcessingPipe = Vec<fn(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Option<Vec<PlaylistGroup>>>;
@@ -430,7 +439,9 @@ fn flatten_groups(playlistgroups: Vec<PlaylistGroup>) -> Vec<PlaylistGroup> {
sort_order.push(group);
}
std::collections::hash_map::Entry::Occupied(o) => {
sort_order.get_mut(*o.get()).unwrap().channels.extend(group.channels);
if let Some(pl_group) = sort_order.get_mut(*o.get()) {
pl_group.channels.extend(group.channels);
}
}
}
}
backend/src/processing/processor/xtream_series.rs
+7 -4
View File
@@ -16,7 +16,7 @@ use std::fs::File;
use std::io::{BufWriter, Write};
use std::sync::Arc;
use std::time::Instant;
use log::{info, log_enabled, Level};
use log::{error, info, log_enabled, Level};
use shared::utils::bincode_serialize;
use crate::model::{XtreamSeriesEpisode, XtreamSeriesInfoEpisode};
use crate::utils;
@@ -36,9 +36,12 @@ fn write_series_episode_record_to_wal_file(
let series_episode = XtreamSeriesEpisode::from(episode);
if let Ok(content_bytes) = bincode_serialize(&series_episode) {
writer.write_all(&provider_id.to_le_bytes())?;
let len = u32::try_from(content_bytes.len()).unwrap();
writer.write_all(&len.to_le_bytes())?;
writer.write_all(&content_bytes)?;
if let Ok(len) = u32::try_from(content_bytes.len()) {
writer.write_all(&len.to_le_bytes())?;
writer.write_all(&content_bytes)?;
} else {
error!("Cant write to WAL file, content length exceeds u32");
}
}
Ok(())
}
backend/src/repository/bplustree.rs
+20 -14
View File
@@ -1,16 +1,16 @@
use crate::utils;
use log::error;
use ruzstd::decoding::StreamingDecoder;
use ruzstd::encoding::{compress_to_vec, CompressionLevel};
use serde::{Deserialize, Serialize};
use shared::error::{str_to_io_error, to_io_error};
use shared::utils::{bincode_deserialize, bincode_serialize};
use std::fs::File;
use std::io::{self, BufReader, Read, Seek, SeekFrom, Write};
use std::marker::PhantomData;
use std::mem::size_of;
use std::path::Path;
use shared::error::{str_to_io_error, to_io_error};
use log::error;
use ruzstd::decoding::StreamingDecoder;
use ruzstd::encoding::{compress_to_vec, CompressionLevel};
use serde::{Deserialize, Serialize};
use tempfile::NamedTempFile;
use shared::utils::{bincode_deserialize, bincode_serialize};
use crate::utils;
const BLOCK_SIZE: usize = 4096;
const BINCODE_OVERHEAD: usize = 8;
@@ -102,9 +102,10 @@ where
fn find_leaf_entry(node: &Self) -> Option<&K> {
if node.is_leaf {
node.keys.first()
} else {
let child = node.children.first().unwrap();
} else if let Some(child) = node.children.first() {
Self::find_leaf_entry(child)
} else {
None
}
}
@@ -185,7 +186,9 @@ where
let mut node = Self::new(false);
node.keys = self.keys.split_off(median + 1);
node.children = self.children.split_off(median + 1);
self.children.push(node.children.first().unwrap().clone());
if let Some(child) = node.children.first() {
self.children.push(child.clone());
}
node
}
}
@@ -301,7 +304,7 @@ where
// Deserialize values if leaf node
let values = if is_leaf {
let use_compression = u8::from_le_bytes(buffer[read_pos..=read_pos].try_into().unwrap()) == 1;
let use_compression = u8::from_le_bytes(buffer[read_pos..=read_pos].try_into().unwrap_or([0u8])) == 1;
read_pos += FLAG_SIZE;
let values_length = u32_from_bytes(&buffer[read_pos..read_pos + LEN_SIZE])? as usize;
read_pos += LEN_SIZE;
@@ -528,7 +531,11 @@ where
};
}
let child_idx = get_entry_index_upper_bound::<K>(&node.keys, key);
offset = *pointers.unwrap().get(child_idx).unwrap();
if let Some(child_offset) = pointers.unwrap().get(child_idx) {
offset = *child_offset;
} else {
return None;
}
}
Err(err) => {
error!("Failed to read id tree from file {err}");
@@ -769,9 +776,9 @@ mod tests {
use std::io;
use std::path::PathBuf;
use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery, BPlusTreeUpdate};
use serde::{Deserialize, Serialize};
use shared::utils::generate_random_string;
use crate::repository::bplustree::{BPlusTree, BPlusTreeQuery, BPlusTreeUpdate};
// Example usage with a simple struct
#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, Eq)]
@@ -878,7 +885,6 @@ mod tests {
assert!(format!("{content} {}", k + 1).eq(&v.data), "Wrong entry");
});
});
}
#[test]
backend/src/repository/indexed_document.rs
+3 -3
View File
@@ -208,7 +208,7 @@ where
}
}
Err(err) => {
return Err(str_to_io_error(&format!("failed to write document: {} - {}", self.main_path.to_str().unwrap(), err)));
return Err(str_to_io_error(&format!("failed to write document: {} - {}", self.main_path.display(), err)));
}
}
Ok(())
@@ -320,7 +320,7 @@ where
Err(e) => Err(e)
}
} else {
Err(Error::new(ErrorKind::NotFound, format!("File not found {}", main_path.to_str().unwrap())))
Err(Error::new(ErrorKind::NotFound, format!("File not found {}", main_path.display())))
}
}
@@ -404,7 +404,7 @@ impl IndexedDocumentDirectAccess {
return Ok(item);
}
}
Err(str_to_io_error(&format!("Failed to read item for id {:?} - {}", doc_id, main_path.to_str().unwrap())))
Err(str_to_io_error(&format!("Failed to read item for id {:?} - {}", doc_id, main_path.display())))
}
}
backend/src/repository/m3u_repository.rs
+2 -2
View File
@@ -16,7 +16,7 @@ use crate::utils;
macro_rules! cant_write_result {
($path:expr, $err:expr) => {
create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {}", $path.to_str().unwrap() ,$err)
create_tuliprox_error!(TuliproxErrorKind::Notify, "failed to write m3u playlist: {} - {}", $path.display() ,$err)
}
}
@@ -44,7 +44,7 @@ fn persist_m3u_playlist_as_text(cfg: &Config, target: &ConfigTarget, target_outp
}
}
Err(_) => {
error!("Can't write m3u plain playlist {}", &m3u_filename.to_str().unwrap());
error!("Can't write m3u plain playlist {}", &m3u_filename.display());
}
}
}
backend/src/repository/strm_repository.rs
+6 -4
View File
@@ -128,10 +128,12 @@ fn style_rename_year<'a>(
if let Ok(year) = year_match.as_str().parse::<u32>() {
if (1900..=cur_year).contains(&year) {
years.push(year);
let match_start = caps.get(0).unwrap().start();
let match_end = caps.get(0).unwrap().end();
new_name.push_str(&name[last_index..match_start]);
last_index = match_end;
if let Some(matched) = caps.get(0) {
let match_start = matched.start();
let match_end = matched.end();
new_name.push_str(&name[last_index..match_start]);
last_index = match_end;
}
}
}
}
backend/src/repository/xtream_repository.rs
+2 -2
View File
@@ -32,7 +32,7 @@ macro_rules! cant_write_result {
create_tuliprox_error!(
TuliproxErrorKind::Notify,
"failed to write xtream playlist: {} - {}",
$path.to_str().unwrap(),
$path.display(),
$err
)
};
@@ -64,7 +64,7 @@ fn ensure_xtream_storage_path(cfg: &Config, target_name: &str) -> Result<PathBuf
if std::fs::create_dir_all(&path).is_err() {
let msg = format!(
"Failed to save xtream data, can't create directory {}",
&path.to_str().unwrap()
&path.display()
);
return Err(notify_err!(msg));
}
shared/src/foundation/mapper.rs
+16 -13
View File
@@ -531,21 +531,24 @@ impl MapperScript {
let first = pairs.next().unwrap();
let key = match first.as_rule() {
Rule::map_key => {
let map_key = first.into_inner().next().unwrap();
match map_key.as_rule() {
Rule::field => {
MapKey::FieldAccess(map_key.as_str().trim().to_string())
}
Rule::var_access => {
let text = map_key.as_str();
if text.contains('.') {
let splitted: Vec<&str> = text.splitn(2, '.').collect();
MapKey::VarAccess(splitted[0].trim().to_string(), splitted[1].trim().to_string())
} else {
MapKey::Identifier(text.trim().to_string())
if let Some(map_key) = first.into_inner().next() {
match map_key.as_rule() {
Rule::field => {
MapKey::FieldAccess(map_key.as_str().trim().to_string())
}
Rule::var_access => {
let text = map_key.as_str();
if text.contains('.') {
let splitted: Vec<&str> = text.splitn(2, '.').collect();
MapKey::VarAccess(splitted[0].trim().to_string(), splitted[1].trim().to_string())
} else {
MapKey::Identifier(text.trim().to_string())
}
}
_ => return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Unexpected map case key: {:?}", map_key.as_rule()),
}
_ => return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Unexpected map case key: {:?}", map_key.as_rule()),
} else {
return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Missing map case key");
}
}
_ => return create_tuliprox_error_result!(TuliproxErrorKind::Info, "Unexpected map case key: {:?}", first.as_rule()),
shared/src/model/config/api_user.rs
+1 -2
View File
@@ -77,8 +77,7 @@ impl FromStr for ProxyType {
return Ok(Self::Reverse(None));
}
if s.starts_with(Self::REVERSE) {
let suffix = s.strip_prefix(Self::REVERSE).unwrap();
if let Some(suffix) = s.strip_prefix(Self::REVERSE) {
if let Ok(force_redirect) = ClusterFlags::try_from(suffix) {
if force_redirect.has_full_flags() {
return Ok(ProxyType::Reverse(None));
shared/src/model/config/base.rs
+2 -4
View File
@@ -1,7 +1,5 @@
use crate::error::{TuliproxError, TuliproxErrorKind};
use crate::model::{ConfigApiDto, HdHomeRunConfigDto, IpCheckConfigDto, LogConfigDto,
MessagingConfigDto, ProxyConfigDto, ReverseProxyConfigDto, ScheduleConfigDto,
VideoConfigDto, WebUiConfigDto};
use crate::model::{ConfigApiDto, HdHomeRunConfigDto, IpCheckConfigDto, LogConfigDto, MessagingConfigDto, ProxyConfigDto, ReverseProxyConfigDto, ScheduleConfigDto, VideoConfigDto, WebUiConfigDto, DEFAULT_VIDEO_EXTENSIONS};
use crate::utils::default_connect_timeout_secs;
pub const DEFAULT_USER_AGENT: &str = "VLC/3.0.16 LibVLC/3.0.16";
@@ -103,7 +101,7 @@ impl ConfigDto {
match &mut self.video {
None => {
self.video = Some(VideoConfigDto {
extensions: vec!["mkv".to_string(), "avi".to_string(), "mp4".to_string()],
extensions: DEFAULT_VIDEO_EXTENSIONS.iter().map(ToString::to_string).collect(),
download: None,
web_search: None,
});
shared/src/model/config/cache.rs
+8 -8
View File
@@ -20,14 +20,14 @@ impl CacheConfigDto {
pub(crate) fn prepare(&mut self, working_dir: &str) -> Result<(), TuliproxError> {
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 = 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();
}
self.dir = Some(cache_dir.to_string());
match self.dir.as_ref() {
None => self.dir = Some(work_path.join("cache").to_string_lossy().to_string()),
Some(work_dir) => {
let mut cache_dir = work_dir.to_string();
if PathBuf::from(&cache_dir).is_relative() {
cache_dir = work_path.join(&cache_dir).clean().to_string_lossy().to_string();
}
self.dir = Some(cache_dir.to_string()); }
}
if let Some(val) = self.size.as_ref() {
shared/src/model/config/video_download.rs
+3 -1
View File
@@ -4,6 +4,8 @@ use crate::create_tuliprox_error_result;
use crate::error::{TuliproxError, TuliproxErrorKind};
use crate::model::DEFAULT_USER_AGENT;
pub const DEFAULT_VIDEO_EXTENSIONS: [&str; 6] = ["mkv", "avi", "mp4", "mpeg", "divx", "mov"];
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, Default, PartialEq)]
#[serde(deny_unknown_fields)]
pub struct VideoDownloadConfigDto {
@@ -35,7 +37,7 @@ impl VideoConfigDto {
/// Will panic if default `RegEx` gets invalid
pub fn prepare(&mut self) -> Result<(), TuliproxError> {
if self.extensions.is_empty() {
self.extensions = ["mkv", "avi", "mp4", "mpeg", "divx", "mov"]
self.extensions = DEFAULT_VIDEO_EXTENSIONS
.iter()
.map(|&arg| arg.to_string())
.collect();
shared/src/utils/directed_graph.rs
+4 -3
View File
@@ -71,9 +71,10 @@ where
self.dfs_find_cycles(neighbor, visited, recursion_stack, cycles);
} else if recursion_stack.contains(neighbor) {
// Cycle detected; collect the cycle path
let cycle_start_index = recursion_stack.iter().position(|n| n == neighbor).unwrap();
let cycle = recursion_stack[cycle_start_index..].to_vec();
cycles.push(cycle);
if let Some(cycle_start_index) = recursion_stack.iter().position(|n| n == neighbor) {
let cycle = recursion_stack[cycle_start_index..].to_vec();
cycles.push(cycle);
}
}
}
}
webui/src/app/components/dashboard/stats_view.rs
+1 -1
View File
@@ -19,7 +19,7 @@ pub fn StatsView() -> Html {
match &status_ctx.status {
Some(stats) => {
if let Some(map) = &stats.active_provider_connections {
if map.len() > 0 {
if !map.is_empty() {
let cards = map.iter().map(|(provider, connections)| {
html! {
<Card>
webui/src/app/components/home.rs
+4 -6
View File
@@ -93,12 +93,10 @@ pub fn Home() -> Html {
} else {
treemap.insert(provider, connections);
}
} else {
if connections > 0 {
let mut treemap = BTreeMap::new();
treemap.insert(provider, connections);
server_status.active_provider_connections = Some(treemap);
}
} else if connections > 0 {
let mut treemap = BTreeMap::new();
treemap.insert(provider, connections);
server_status.active_provider_connections = Some(treemap);
}
let new_status = Rc::new(server_status);
*status_holder_signal.borrow_mut() = Some(Rc::clone(&new_status));
webui/src/services/requests.rs
+2 -1
View File
@@ -48,7 +48,8 @@ where
match response.status() {
200 => {
if std::any::TypeId::of::<T>() == std::any::TypeId::of::<()>() {
Ok(serde_json::from_str("null").unwrap()) // `T = ()` valid
// `T = ()` valid
serde_json::from_str("null").map_err(|_| Error::DeserializeError)
} else {
let data: Result<T, _> = response.json::<T>().await;
if let Ok(data) = data {
webui/src/services/websocket_service.rs
+4 -2
View File
@@ -19,12 +19,14 @@ pub enum WsMessage {
const WS_PATH: &str = "/ws";
type Subscriber = RefCell<HashMap<usize, Box<dyn Fn(WsMessage)>>>;
pub struct WebSocketService {
connected: Rc<AtomicBool>,
ws: Rc<RefCell<Option<WebSocket>>>,
status_service: Rc<StatusService>,
subscriber_id: Rc<AtomicUsize>,
subscribers: Rc<RefCell<HashMap<usize, Box<dyn Fn(WsMessage)>>>>,
subscribers: Rc<Subscriber>,
}
impl WebSocketService {
@@ -73,7 +75,7 @@ impl WebSocketService {
// onmessage
let onmessage_callback = Closure::<dyn FnMut(_)>::wrap(Box::new(move |event: MessageEvent| {
trace!("WebSocket received message: {:?}", event);
trace!("WebSocket received message: {event:?}");
if let Ok(buf) = event.data().dyn_into::<ArrayBuffer>() {
let array = Uint8Array::new(&buf);
let bytes = bytes::Bytes::from(array.to_vec());