diff --git a/Cargo.lock b/Cargo.lock index 6f63f80e9..c9aee2d00 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1156,7 +1156,7 @@ dependencies = [ [[package]] name = "frontend" -version = "3.2.48" +version = "3.2.49" dependencies = [ "anyhow", "base64", @@ -1168,7 +1168,7 @@ dependencies = [ "futures-signals", "gloo-render 0.2.0", "gloo-storage 0.3.0", - "gloo-timers 0.3.0", + "gloo-timers 0.2.6", "gloo-utils 0.2.0", "implicit-clone", "indexmap", @@ -1692,6 +1692,8 @@ version = "0.2.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b995a66bb87bebce9a0f4a95aed01daca4872c050bfcb21653361c03bc35e5c" dependencies = [ + "futures-channel", + "futures-core", "js-sys", "wasm-bindgen", ] @@ -4013,7 +4015,7 @@ dependencies = [ [[package]] name = "shared" -version = "3.2.48" +version = "3.2.49" dependencies = [ "base64", "bitflags 2.10.0", @@ -4590,7 +4592,7 @@ checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" [[package]] name = "tuliprox" -version = "3.2.48" +version = "3.2.49" dependencies = [ "arc-swap", "async-compression", diff --git a/backend/Cargo.toml b/backend/Cargo.toml index b6f3cf516..75f39ab7e 100644 --- a/backend/Cargo.toml +++ b/backend/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "tuliprox" -version = "3.2.48" +version = "3.2.49" edition = "2021" rust-version = "1.89.0" diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index cf3d3933e..db0935f96 100644 --- a/backend/src/api/api_utils.rs +++ b/backend/src/api/api_utils.rs @@ -1343,7 +1343,7 @@ async fn fetch_resource_with_retry( let disabled_headers = app_state.get_disabled_headers(); - let response = match send_with_retry( + let Ok(response) = send_with_retry( &app_state.app_config, url, || { @@ -1357,12 +1357,7 @@ async fn fetch_resource_with_retry( default_user_agent.as_deref(), ) }, - ) - .await - { - Ok(response) => response, - Err(_) => return None, - }; + ).await else { return None }; let status = response.status(); diff --git a/backend/src/api/endpoints/v1_api_playlist.rs b/backend/src/api/endpoints/v1_api_playlist.rs index 0b97b1af2..9dc6688e9 100644 --- a/backend/src/api/endpoints/v1_api_playlist.rs +++ b/backend/src/api/endpoints/v1_api_playlist.rs @@ -3,16 +3,17 @@ use crate::api::endpoints::api_playlist_utils::{get_playlist_for_custom_provider use crate::api::endpoints::extract_accept_header::ExtractAcceptHeader; use crate::api::model::AppState; use crate::auth::create_access_token; -use crate::model::{parse_xmltv_for_web_ui_from_file, parse_xmltv_for_web_ui_from_url, ConfigInput, ConfigInputOptions}; +use crate::model::{parse_xmltv_for_web_ui_from_url, ConfigInput, ConfigInputOptions}; use crate::processing::processor::playlist; use axum::response::IntoResponse; use axum::{Router}; -use log::{error, info}; +use log::{debug, error}; use serde_json::json; use shared::model::{InputType, PlaylistEpgRequest, PlaylistRequest, ProxyType, TargetType, WebplayerUrlRequest, XtreamCluster}; use shared::utils::{sanitize_sensitive_info, Internable}; use std::sync::Arc; use url::Url; +use crate::api::endpoints::xmltv_api::serve_epg; use crate::api::endpoints::xtream_api::xtream_get_stream_info_response; fn create_config_input_for_m3u(url: &str) -> ConfigInput { @@ -169,13 +170,13 @@ async fn playlist_series_info( if let Some(input) = app_state.app_config.get_input_by_id(input_id) { if matches!(input.input_type, InputType::Xtream | InputType::XtreamBatch) { // TODO: Implement series info retrieval for input-based requests - info!("TODO: Implement series info retrieval for input-based requests"); + debug!("TODO: Implement series info retrieval for input-based requests"); } } } PlaylistRequest::CustomXtream(_xtream) => { // TODO: Implement series info retrieval for custom Xtream requests - info!("TODO: Implement series info retrieval for custom Xtream requests"); + debug!("TODO: Implement series info retrieval for custom Xtream requests"); }, PlaylistRequest::CustomM3u(_) => {} } @@ -204,9 +205,15 @@ async fn playlist_epg( if let Some(target) = app_state.app_config.get_target_by_id(target_id) { let config = &app_state.app_config.config.load(); if let Some(epg_path) = crate::api::endpoints::xmltv_api::get_epg_path_for_target(config, &target) { - if let Ok(epg) = parse_xmltv_for_web_ui_from_file(&epg_path).await { - return json_or_bin_response(accept.as_deref(), &epg).into_response(); - } + let user = create_api_proxy_user(&app_state); + // TODO we should serve EpgChannel instead of xml + return serve_epg( + &app_state, + &epg_path, + &user, + &target, + None, + ).await } } } diff --git a/backend/src/api/endpoints/xmltv_api.rs b/backend/src/api/endpoints/xmltv_api.rs index 89957bd02..ad5f4ec8e 100644 --- a/backend/src/api/endpoints/xmltv_api.rs +++ b/backend/src/api/endpoints/xmltv_api.rs @@ -1,23 +1,21 @@ -use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, internal_server_error, resource_response, serve_file, try_unwrap_body}; +use crate::api::api_utils::{get_user_target, get_user_target_by_credentials, internal_server_error, resource_response, try_unwrap_body}; use crate::api::model::AppState; use crate::api::model::UserApiRequest; -use crate::model::Config; +use crate::model::{Config, EPG_ATTRIB_ID, EPG_TAG_CHANNEL}; use crate::model::{ConfigTarget, ProxyUserCredentials, TargetOutput}; -use crate::repository::get_target_storage_path; -use crate::repository::m3u_get_epg_file_path; +use crate::repository::m3u_get_epg_file_path_for_target; use crate::repository::storage_const; use crate::repository::XML_PREAMBLE; -use crate::repository::{xtream_get_epg_file_path, xtream_get_storage_path}; -use crate::utils::{async_file_reader, deobscure_text}; -use crate::utils::{format_xtream_time, EpgConsumer, EpgProcessor}; +use crate::repository::{get_target_storage_path, BPlusTreeQuery}; +use crate::repository::{xtream_get_epg_file_path_for_target, xtream_get_storage_path}; use crate::utils; +use crate::utils::{deobscure_text, file_exists_async, format_xmltv_time_utc, get_epg_processing_options, obscure_text, EpgProcessingOptions}; use axum::response::IntoResponse; -use chrono::{Offset, TimeZone, Utc}; -use chrono_tz::Tz; +use chrono::DateTime; use log::{error, trace}; -use quick_xml::events::Event; -use shared::error::{info_err, TuliproxError}; -use shared::model::{parse_xmltv_time, PlaylistItemType, ShortEpgDto, ShortEpgResultDto}; +use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event}; +use shared::concat_string; +use shared::model::{EpgChannel, EpgProgramme, ShortEpgDto, ShortEpgResultDto}; use std::path::{Path, PathBuf}; use std::sync::Arc; use tokio::io::AsyncWriteExt; @@ -53,7 +51,7 @@ pub(in crate::api) fn get_epg_path_for_target(config: &Config, target: &ConfigTa if let Some(storage_path) = xtream_get_storage_path(config, &target.name) { return get_epg_path_for_target_of_type( &target.name, - xtream_get_epg_file_path(&storage_path), + xtream_get_epg_file_path_for_target(&storage_path), ); } } @@ -61,7 +59,7 @@ pub(in crate::api) fn get_epg_path_for_target(config: &Config, target: &ConfigTa if let Some(target_path) = get_target_storage_path(config, &target.name) { return get_epg_path_for_target_of_type( &target.name, - m3u_get_epg_file_path(&target_path), + m3u_get_epg_file_path_for_target(&target_path), ); } } @@ -71,280 +69,320 @@ pub(in crate::api) fn get_epg_path_for_target(config: &Config, target: &ConfigTa None } -/// Parses user-defined EPG timeshift configuration. -/// Supports either a numeric offset (e.g. "+2:30", "-1:15") -/// or a timezone name (e.g. "`Europe/Berlin`", "`UTC`", "`America/New_York`"). -/// -/// Returns the total offset in minutes (i32). -fn parse_timeshift(time_shift: Option<&str>) -> Option { - time_shift.and_then(|offset| { - // Try to parse as timezone name first - if let Ok(tz) = offset.parse::() { - // Determine the current UTC offset of that timezone (including DST) - let now = Utc::now(); - let local_time = tz.from_utc_datetime(&now.naive_utc()); - let offset_minutes = local_time.offset().fix().local_minus_utc() / 60; - return Some(offset_minutes); - } - - // If not a timezone, try to parse as numeric offset - let sign_factor = if offset.starts_with('-') { -1 } else { 1 }; - let offset = offset.trim_start_matches(&['-', '+'][..]); - - let parts: Vec<&str> = offset.split(':').collect(); - let hours: i32 = parts.first().and_then(|h| h.parse().ok()).unwrap_or(0); - let minutes: i32 = parts.get(1).and_then(|m| m.parse().ok()).unwrap_or(0); - - let total_minutes = hours * 60 + minutes; - (total_minutes > 0).then_some(sign_factor * total_minutes) - }) -} - pub async fn serve_epg( app_state: &Arc, epg_path: &Path, user: &ProxyUserCredentials, target: &Arc, - filter: Option>, + limit: Option, ) -> axum::response::Response { - if let Ok(exists) = tokio::fs::try_exists(epg_path).await { - if exists { - let rewrite_resources = app_state.app_config.is_reverse_proxy_resource_rewrite_enabled(); - let encrypt_secret = app_state.app_config.get_reverse_proxy_rewrite_secret().unwrap_or_else(|| app_state.app_config.encrypt_secret); + if file_exists_async(epg_path).await { + serve_epg_with_rewrites(app_state, epg_path, user, target, limit).await + } else { + get_empty_epg_response() + } +} - // If redirect is true → rewrite_urls = false → keep original - // If redirect is false and rewrite_resources is true → rewrite_urls = true → rewriting allowed - // If redirect is false and rewrite_resources is false → rewrite_urls = false → no rewriting - let redirect = user.proxy.is_redirect(PlaylistItemType::Live) || target.is_force_redirect(PlaylistItemType::Live); - let rewrite_urls = !redirect && rewrite_resources; - - // Use 0 for timeshift if None - let timeshift = parse_timeshift(user.epg_timeshift.as_deref()).unwrap_or(0); - - return if timeshift != 0 || rewrite_urls || filter.is_some() { - let server_info = app_state.app_config.get_user_server_info(user); - let base_url = format!("{}/{}/{}/{}/", server_info.get_base_url(), - storage_const::EPG_RESOURCE_PATH, &user.username, &user.password); - // Apply timeshift and/or rewrite URLs and/or filter - serve_epg_with_rewrites(epg_path, timeshift, rewrite_urls, &encrypt_secret, &base_url, filter).await - } else { - // Neither timeshift nor rewrite needed, serve original file - serve_file(epg_path, mime::TEXT_XML.to_string()).await.into_response() - }; +macro_rules! continue_on_err { + ($expr:expr) => { + if let Err(_err) = $expr { + continue; } - } - get_empty_epg_response() -} - -struct XmlEpgConsumer { - writer: quick_xml::writer::Writer, -} - -impl EpgConsumer for XmlEpgConsumer { - async fn handle_event(&mut self, event: &Event<'_>, _decoder: quick_xml::Decoder) -> Result<(), TuliproxError> { - self.writer.write_event_async(event.clone()) - .await - .map_err(|e| info_err!("Failed to write EPG event: {}", e)) - } + }; } #[allow(clippy::too_many_lines)] async fn serve_epg_with_rewrites( + app_state: &Arc, epg_path: &Path, - offset_minutes: i32, - rewrite_urls: bool, - secret: &[u8; 16], - base_url: &str, - filter: Option>, + user: &ProxyUserCredentials, + target: &Arc, + limit: Option, ) -> axum::response::Response { - match tokio::fs::try_exists(epg_path).await { - Ok(exists) => { - if !exists { - return axum::http::StatusCode::NOT_FOUND.into_response(); - } - } - Err(err) => { - error!("Failed to open egp file {}, {err:?}", epg_path.display()); - return axum::http::StatusCode::NOT_FOUND.into_response(); - } + if !file_exists_async(epg_path).await { + return get_empty_epg_response(); } - let encrypt_secret = *secret; - let rewrite_base_url = base_url.to_owned(); - match tokio::fs::File::open(epg_path).await { - Ok(file) => { - let reader = async_file_reader(file); - let (tx, rx) = tokio::io::duplex(8192); - tokio::spawn(async move { - let mut encoder = async_compression::tokio::write::GzipEncoder::new(tx); + let mut query = match BPlusTreeQuery::, EpgChannel>::try_new(&epg_path) { + Ok(query) => query, + Err(err) => { + error!("Failed to open BPlusTreeQuery {epg_path:?} - {err}"); + return get_empty_epg_response(); + } + }; - // Work-Around BytesText DocType escape, see below - if let Err(err) = encoder.write_all(XML_PREAMBLE.as_ref()).await { - error!("EPG: Failed to write xml header {err}"); + let epg_processing_options = get_epg_processing_options(app_state, user, target); + + let base_url = if epg_processing_options.offset_minutes != 0 || epg_processing_options.rewrite_urls { + let server_info = app_state.app_config.get_user_server_info(user); + Some(concat_string!(&server_info.get_base_url(), "/", storage_const::EPG_RESOURCE_PATH, "/", &user.username, "/", &user.password)) + } else { + None + }; + + let limit = limit.unwrap_or_default(); + + let (tx, rx) = tokio::io::duplex(8192); + tokio::spawn(async move { + let mut encoder = async_compression::tokio::write::GzipEncoder::new(tx); + // Work-Around BytesText DocType escape, see below + if let Err(err) = encoder.write_all(XML_PREAMBLE.as_ref()).await { + error!("EPG: Failed to write xml header {err}"); + }; + if let Err(err) = encoder.write_all(r#""#.as_bytes()).await { + error!("EPG: Failed to write xml tv header {err}"); + }; + + let mut writer = quick_xml::writer::Writer::new(encoder); + for (_, channel) in query.iter() { + let programmes = if limit > 0 { + channel.get_programme_with_limit(limit) + } else { + channel.programmes.iter().collect::>() + }; + + if !programmes.is_empty() { + let mut elem = BytesStart::new(EPG_TAG_CHANNEL); + elem.push_attribute((EPG_ATTRIB_ID, channel.id.as_ref())); + continue_on_err!(writer.write_event_async(Event::Start(elem)).await); + + let elem = BytesStart::new("display-name"); + continue_on_err!(writer.write_event_async(Event::Start(elem)).await); + let title: &str = channel.title.as_deref().unwrap_or(""); + continue_on_err!(writer.write_event_async(Event::Text(BytesText::new(title))).await); + + let elem = BytesEnd::new("display-name"); + continue_on_err!(writer.write_event_async(Event::End(elem)).await); + + if let Some(icon_url) = &channel.icon { + let icon = if epg_processing_options.rewrite_urls && base_url.is_some() { + if let Ok(encrypted) = obscure_text(&epg_processing_options.encrypt_secret, icon_url) { + concat_string!(base_url.as_ref().unwrap(), &encrypted) + } else { + icon_url.to_string() + } + } else { + icon_url.to_string() + }; + + let mut elem = BytesStart::new("icon"); + elem.push_attribute(("src", icon.as_ref())); + if let Err(_) = writer.write_event_async(Event::Empty(elem)).await { + // ignore + } } - let mut consumer = XmlEpgConsumer { - writer: quick_xml::writer::Writer::new(encoder), - }; + let elem = BytesEnd::new(EPG_TAG_CHANNEL); + continue_on_err!(writer.write_event_async(Event::End(elem)).await); - let mut processor = EpgProcessor::new( - reader, - offset_minutes, - rewrite_urls, - rewrite_base_url, - encrypt_secret, - filter, - ); + for programme in programmes { + let mut elem = BytesStart::new("programme"); + elem.push_attribute(("start", format_xmltv_time_utc(programme.start).as_str())); + elem.push_attribute(("stop", format_xmltv_time_utc(programme.stop).as_str())); + elem.push_attribute(("channel", &programme.channel[..])); + continue_on_err!(writer.write_event_async(Event::Start(elem)).await); - if let Err(e) = processor.process(&mut consumer).await { - error!("EPG: Failed to process EPG: {e}"); + if let Some(title) = &programme.title { + let elem = BytesStart::new("title"); + continue_on_err!(writer.write_event_async(Event::Start(elem)).await); + continue_on_err!(writer.write_event_async(Event::Text(BytesText::new(title))).await); + continue_on_err!(writer.write_event_async(Event::End(BytesEnd::new("title"))).await); + } + + if let Some(desc) = &programme.desc { + let elem = BytesStart::new("desc"); + continue_on_err!(writer.write_event_async(Event::Start(elem)).await); + continue_on_err!(writer.write_event_async(Event::Text(BytesText::new(desc))).await); + continue_on_err!(writer.write_event_async(Event::End(BytesEnd::new("desc"))).await); + } + continue_on_err!(writer.write_event_async(Event::End(BytesEnd::new("programme"))).await); } + } + } - let mut encoder = consumer.writer.into_inner(); - if let Err(e) = encoder.shutdown().await { - error!("Failed to shutdown epg gzip encoder: {e}"); - } - }); + let mut encoder = writer.into_inner(); - let body_stream = ReaderStream::new(rx); - try_unwrap_body!(axum::response::Response::builder() - .header( - axum::http::header::CONTENT_TYPE, - mime::TEXT_XML.to_string() - ) + if let Err(err) = encoder.write_all(r#""#.as_bytes()).await { + error!("EPG: Failed to write xml tv close {err}"); + }; + + if let Err(e) = encoder.shutdown().await { + error!("Failed to shutdown epg gzip encoder: {e}"); + } + }); + + let body_stream = ReaderStream::new(rx); + try_unwrap_body!(axum::response::Response::builder() + .header(axum::http::header::CONTENT_TYPE, mime::TEXT_XML.to_string()) .header(axum::http::header::CONTENT_ENCODING, "gzip") // Set Content-Encoding header .body(axum::body::Body::from_stream(body_stream))) + +} + +async fn get_epg_channel(app_state: &Arc, channel_id: &Arc, epg_path: &Path) -> Option { + let _file_lock = app_state.app_config.file_locks.read_lock(epg_path).await; + match BPlusTreeQuery::, EpgChannel>::try_new(&epg_path) { + Ok(mut query) => { + match query.query(channel_id) { + Ok(Some(item)) => return Some(item), + Ok(None) => {} + Err(err) => { + error!("Failed to quer db file {}: {err}", epg_path.display()); + } + } } - Err(_) => internal_server_error!(), + Err(err) => { error!("Failed to read db file {}: {err}", epg_path.display()); } + }; + None +} + + +fn format_xmltv_time(ts: i64) -> String { + if let Some(dt) = DateTime::from_timestamp(ts, 0) { + dt.naive_utc().format("%Y-%m-%d %H:%M:%S").to_string() + } else { + String::new() + } +} + +fn from_programme(stream_id: &Arc, programme: &EpgProgramme, epg_processing_options: &EpgProcessingOptions) -> ShortEpgDto { + let user_start = programme.start + i64::from(epg_processing_options.offset_minutes) * 60; + let user_end = programme.stop + i64::from(epg_processing_options.offset_minutes) * 60; + + ShortEpgDto { + id: Arc::clone(stream_id), + epg_id: Arc::clone(&programme.channel), + title: programme.title.as_ref().map_or_else(String::new, |v| v.to_string()), + lang: "".to_string(), + start: format_xmltv_time(programme.start), + end: format_xmltv_time(programme.stop), + description: programme.desc.as_ref().map_or_else(String::new, |v| v.to_string()), + channel_id: Arc::clone(&programme.channel), + start_timestamp: user_start.to_string(), + stop_timestamp: user_end.to_string(), + stream_id: Arc::clone(stream_id), } } pub async fn serve_short_epg( - _app_state: &Arc, + app_state: &Arc, epg_path: &Path, - _user: &ProxyUserCredentials, - _target: &Arc, - channel_id: std::sync::Arc, - stream_id: &str, + user: &ProxyUserCredentials, + target: &Arc, + channel_id: &Arc, + stream_id: Arc, + limit: u32, ) -> axum::response::Response { - match tokio::fs::File::open(epg_path).await { - Ok(file) => { - let reader = async_file_reader(file); - let mut consumer = DtoEpgConsumer { - items: Vec::new(), - current_item: None, - current_tag: String::new(), - stream_id: stream_id.to_string(), - }; - - let mut processor = EpgProcessor::new( - reader, - 0, - false, - String::new(), - [0u8; 16], - Some(channel_id), - ); - - if let Err(e) = processor.process(&mut consumer).await { - error!("EPG: Failed to process EPG for short epg: {e}"); - } - - let short_epg = ShortEpgResultDto::new(consumer.items); - - match serde_json::to_string(&short_epg) { - Ok(json) => ( - axum::http::StatusCode::OK, - [(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())], - json - ).into_response(), - Err(_) => internal_server_error!(), + let short_epg = { + if file_exists_async(epg_path).await { + if let Some(epg_channel) = get_epg_channel(app_state, channel_id, epg_path).await { + let epg_processing_options = get_epg_processing_options(app_state, user, target); + ShortEpgResultDto { + epg_listings: if limit > 0 { + epg_channel.get_programme_with_limit(limit).iter().map(|p| from_programme(&stream_id, p, &epg_processing_options)).collect() + } else { + epg_channel.programmes.iter().map(|p| from_programme(&stream_id, p, &epg_processing_options)).collect() + }, + error: None, + } + } else { + ShortEpgResultDto::default() } + } else { + ShortEpgResultDto::default() } + }; + + match serde_json::to_string(&short_epg) { + Ok(json) => ( + axum::http::StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, mime::APPLICATION_JSON.to_string())], + json + ).into_response(), Err(_) => internal_server_error!(), } } -struct DtoEpgConsumer { - items: Vec, - current_item: Option, - current_tag: String, - stream_id: String, -} - - -impl EpgConsumer for DtoEpgConsumer { - async fn handle_event(&mut self, event: &Event<'_>, decoder: quick_xml::Decoder) -> Result<(), TuliproxError> { - match event { - Event::Start(e) => { - let tag = e.name(); - match tag.as_ref() { - b"programme" => { - let mut item = ShortEpgDto::default(); - item.stream_id.clone_from(&self.stream_id); - item.id.clone_from(&self.stream_id); - for attr in e.attributes().flatten() { - match attr.key.as_ref() { - b"start" => { - if let Ok(v) = attr.decode_and_unescape_value(decoder) { - if let Some(ts) = parse_xmltv_time(&v) { - item.start_timestamp = ts.to_string(); - item.start = format_xtream_time(ts); - } - } - } - b"stop" => { - if let Ok(v) = attr.decode_and_unescape_value(decoder) { - if let Some(ts) = parse_xmltv_time(&v) { - item.stop_timestamp = ts.to_string(); - item.end = format_xtream_time(ts); - } - } - } - b"channel" => { - if let Ok(v) = attr.decode_and_unescape_value(decoder) { - item.epg_id = v.to_string(); - item.channel_id = v.to_string(); - } - } - _ => {} - } - } - self.current_item = Some(item); - } - _ => self.current_tag = String::from_utf8_lossy(tag.as_ref()).to_string(), - } - } - Event::Text(e) => { - if let Some(item) = &mut self.current_item { - if let Ok(text) = decoder.decode(e.as_ref()) { - let unescaped = match quick_xml::escape::unescape(&text) { - Ok(u) => u, - Err(_) => text, - }; - match self.current_tag.as_str() { - "title" => item.title.push_str(&unescaped), - "desc" => item.description.push_str(&unescaped), - _ => {} - } - } - } - } - Event::End(e) => { - if e.name().as_ref() == b"programme" { - if let Some(item) = self.current_item.take() { - self.items.push(item); - } - } - if matches!(e.name().as_ref(), b"programme" | b"title" | b"desc") { - self.current_tag.clear(); - } - } - _ => {} - } - Ok(()) - } -} +// struct DtoEpgConsumer { +// items: Vec, +// current_item: Option, +// current_tag: String, +// stream_id: String, +// } +// +// +// impl EpgConsumer for DtoEpgConsumer { +// async fn handle_event(&mut self, event: &Event<'_>, decoder: quick_xml::Decoder) -> Result<(), TuliproxError> { +// match event { +// Event::Start(e) => { +// let tag = e.name(); +// match tag.as_ref() { +// b"programme" => { +// let mut item = ShortEpgDto::default(); +// item.stream_id.clone_from(&self.stream_id); +// item.id.clone_from(&self.stream_id); +// for attr in e.attributes().flatten() { +// match attr.key.as_ref() { +// b"start" => { +// if let Ok(v) = attr.decode_and_unescape_value(decoder) { +// if let Some(ts) = parse_xmltv_time(&v) { +// item.start_timestamp = ts.to_string(); +// item.start = format_xtream_time(ts); +// } +// } +// } +// b"stop" => { +// if let Ok(v) = attr.decode_and_unescape_value(decoder) { +// if let Some(ts) = parse_xmltv_time(&v) { +// item.stop_timestamp = ts.to_string(); +// item.end = format_xtream_time(ts); +// } +// } +// } +// b"channel" => { +// if let Ok(v) = attr.decode_and_unescape_value(decoder) { +// item.epg_id = v.to_string(); +// item.channel_id = v.to_string(); +// } +// } +// _ => {} +// } +// } +// self.current_item = Some(item); +// } +// _ => self.current_tag = String::from_utf8_lossy(tag.as_ref()).to_string(), +// } +// } +// Event::Text(e) => { +// if let Some(item) = &mut self.current_item { +// if let Ok(text) = decoder.decode(e.as_ref()) { +// let unescaped = match quick_xml::escape::unescape(&text) { +// Ok(u) => u, +// Err(_) => text, +// }; +// match self.current_tag.as_str() { +// "title" => item.title.push_str(&unescaped), +// "desc" => item.description.push_str(&unescaped), +// _ => {} +// } +// } +// } +// } +// Event::End(e) => { +// if e.name().as_ref() == b"programme" { +// if let Some(item) = self.current_item.take() { +// self.items.push(item); +// } +// } +// if matches!(e.name().as_ref(), b"programme" | b"title" | b"desc") { +// self.current_tag.clear(); +// } +// } +// _ => {} +// } +// Ok(()) +// } +// } /// Handles XMLTV EPG API requests, serving the appropriate EPG file with optional time-shifting based on user configuration. /// @@ -425,43 +463,3 @@ pub fn xmltv_api_register() -> axum::Router> { axum::routing::get(epg_api_resource), ) } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_parse_timeshift() { - assert_eq!(parse_timeshift(Some(&String::from("2"))), Some(120)); - assert_eq!(parse_timeshift(Some(&String::from("-1:30"))), Some(-90)); - assert_eq!(parse_timeshift(Some(&String::from("+0:15"))), Some(15)); - assert_eq!(parse_timeshift(Some(&String::from("1:45"))), Some(105)); - assert_eq!(parse_timeshift(Some(&String::from(":45"))), Some(45)); - assert_eq!(parse_timeshift(Some(&String::from("-:45"))), Some(-45)); - assert_eq!(parse_timeshift(Some(&String::from("0:30"))), Some(30)); - assert_eq!(parse_timeshift(Some(&String::from(":3"))), Some(3)); - assert_eq!(parse_timeshift(Some(&String::from("2:"))), Some(120)); - assert_eq!(parse_timeshift(Some(&String::from("+2:00"))), Some(120)); - assert_eq!(parse_timeshift(Some(&String::from("-0:10"))), Some(-10)); - assert_eq!(parse_timeshift(Some(&String::from("invalid"))), None); - assert_eq!(parse_timeshift(Some(&String::from("+abc"))), None); - assert_eq!(parse_timeshift(Some(&String::new())), None); - assert_eq!(parse_timeshift(None), None); - } - - #[test] - fn test_parse_timezone() { - // This will depend on current DST; we just check it’s within a valid range - let berlin = parse_timeshift(Some(&"Europe/Berlin".to_string())).unwrap(); - assert!(berlin == 60 || berlin == 120, "Berlin offset should be 60 or 120, got {berlin}"); - - let new_york = parse_timeshift(Some(&"America/New_York".to_string())).unwrap(); - assert!(new_york == -300 || new_york == -240, "New York offset should be -300 or -240, got {new_york}"); - - let tokyo = parse_timeshift(Some(&"Asia/Tokyo".to_string())).unwrap(); - assert_eq!(tokyo, 540); // always UTC+9 - - let utc = parse_timeshift(Some(&"UTC".to_string())).unwrap(); - assert_eq!(utc, 0); - } -} diff --git a/backend/src/api/endpoints/xtream_api.rs b/backend/src/api/endpoints/xtream_api.rs index 7689c91bb..9a3d8076c 100644 --- a/backend/src/api/endpoints/xtream_api.rs +++ b/backend/src/api/endpoints/xtream_api.rs @@ -17,7 +17,7 @@ use crate::repository::storage_const; use crate::repository::VirtualIdRecord; use crate::repository::{get_target_id_mapping, user_get_bouquet_filter, xtream_get_collection_path, xtream_get_item_for_stream_id, xtream_load_rewrite_playlist}; use crate::utils::xtream::create_vod_info_from_item; -use crate::utils::{debug_if_enabled, trace_if_enabled}; +use crate::utils::{debug_if_enabled, file_exists_async, trace_if_enabled}; use crate::utils::{request, xtream}; use axum::http::HeaderMap; use axum::response::IntoResponse; @@ -31,8 +31,7 @@ use shared::concat_string; use shared::error::{info_err, info_err_res, TuliproxError}; use shared::model::{create_stream_channel_with_type, PlaylistEntry, PlaylistItemType, ProxyType, TargetType, UserConnectionPermission, XtreamCluster, XtreamPlaylistItem}; -use shared::utils::{deserialize_as_string, extract_extension_from_url, generate_playlist_uuid, - sanitize_sensitive_info, trim_slash, HLS_EXT}; +use shared::utils::{deserialize_as_string, extract_extension_from_url, generate_playlist_uuid, sanitize_sensitive_info, trim_slash, Internable, HLS_EXT}; use std::fmt::{Display, Formatter}; use std::str::FromStr; use std::sync::Arc; @@ -887,7 +886,7 @@ async fn xtream_get_short_epg( user: &ProxyUserCredentials, target: &Arc, stream_id: &str, - limit: &str, + limit: u32, ) -> impl IntoResponse + Send { let target_name = &target.name; if target.has_output(TargetType::Xtream) { @@ -904,10 +903,8 @@ async fn xtream_get_short_epg( ).await { let config = &app_state.app_config.config.load(); if let (Some(epg_path), Some(channel_id)) = (get_epg_path_for_target(config, target), &pli.epg_channel_id) { - if let Ok(exists) = tokio::fs::try_exists(&epg_path).await { - if exists { - return serve_short_epg(app_state, epg_path.as_path(), user, target, Arc::clone(channel_id), stream_id).await; - } + if file_exists_async(&epg_path).await { + return serve_short_epg(app_state, epg_path.as_path(), user, target, channel_id, stream_id.intern(), limit).await; } } @@ -923,7 +920,7 @@ async fn xtream_get_short_epg( crate::model::XC_TAG_STREAM_ID, pli.provider_id ); - if !(limit.is_empty() || limit.eq("0")) { + if limit > 0 { info_url = format!("{info_url}&limit={limit}"); } if user.proxy.is_redirect(pli.item_type) @@ -932,7 +929,6 @@ async fn xtream_get_short_epg( return redirect(&info_url).into_response(); } - // TODO serve epg from own db let input_source = InputSource::from(&*input).with_url(info_url); return match request::download_text_content( &app_state.app_config, @@ -1213,7 +1209,7 @@ async fn xtream_player_api( &user, &target, api_req.stream_id.trim(), - api_req.limit.trim(), + api_req.limit, ) .await .into_response(); diff --git a/backend/src/api/model/request.rs b/backend/src/api/model/request.rs index 64edb5619..8b6a3e14b 100644 --- a/backend/src/api/model/request.rs +++ b/backend/src/api/model/request.rs @@ -17,7 +17,7 @@ pub struct UserApiRequest { #[serde(default)] pub category_id: String, #[serde(default)] - pub limit: String, + pub limit: u32, #[serde(default)] pub start: String, #[serde(default)] diff --git a/backend/src/main.rs b/backend/src/main.rs index b8cef8ab2..f88fa4b3e 100644 --- a/backend/src/main.rs +++ b/backend/src/main.rs @@ -92,6 +92,8 @@ struct Args { #[arg(long = "dbm")] db_m3u_file_name: Option, + #[arg(long = "dbe")] + db_epg_file_name: Option, } const VERSION: &str = env!("CARGO_PKG_VERSION"); @@ -109,7 +111,7 @@ const BUILD_TIMESTAMP: &str = env!("VERGEN_BUILD_TIMESTAMP"); async fn main() { let args = Args::parse(); - db_viewer(args.db_xtream_file_name.as_deref(), args.db_m3u_file_name.as_deref()); + db_viewer(args.db_xtream_file_name.as_deref(), args.db_m3u_file_name.as_deref(), args.db_epg_file_name.as_deref()); if args.genpwd { match generate_password() { diff --git a/backend/src/model/config/api_proxy.rs b/backend/src/model/config/api_proxy.rs index cd2495c18..64674dacb 100644 --- a/backend/src/model/config/api_proxy.rs +++ b/backend/src/model/config/api_proxy.rs @@ -8,6 +8,7 @@ use arc_swap::access::Access; use arc_swap::ArcSwap; use shared::model::{ApiProxyConfigDto, ApiProxyServerInfoDto, ConfigPaths, TargetUserDto}; use crate::{utils}; +use crate::utils::file_exists_async; const API_USER: &str = "api"; const TEST_USER: &str = "test"; @@ -128,7 +129,7 @@ impl ApiProxyConfig { } } else { let user_db_path = get_api_user_db_path(cfg); - if let Ok(true) = tokio::fs::try_exists(&user_db_path).await { + if file_exists_async(&user_db_path).await { // we can't have user defined in db file. // we need to load them and save them into the config file if let Ok(stored_users) = load_api_user(cfg).await { diff --git a/backend/src/model/xmltv.rs b/backend/src/model/xmltv.rs index 182eeef43..f9867b005 100644 --- a/backend/src/model/xmltv.rs +++ b/backend/src/model/xmltv.rs @@ -2,7 +2,7 @@ use crate::model::xmltv::XmlTagIcon::Undefined; use chrono::{Datelike, TimeZone, Utc}; use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event}; use shared::error::{TuliproxError, TuliproxErrorKind}; -use shared::model::{parse_xmltv_time, EpgChannel, EpgProgramme, EpgTv, InputFetchMethod}; +use shared::model::{EpgChannel, EpgProgramme, EpgTv, InputFetchMethod}; use std::cmp::{max, min}; use std::collections::HashMap; use std::path::{Path, PathBuf}; @@ -10,10 +10,11 @@ use std::sync::Arc; use futures::TryFutureExt; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; use url::Url; +use shared::concat_string; use shared::utils::{sanitize_sensitive_info, Internable}; use crate::api::model::AppState; use crate::model::{InputSource}; -use crate::utils::async_file_reader; +use crate::utils::{async_file_reader, parse_xmltv_time}; use crate::utils::request::{get_remote_content_as_stream}; pub const EPG_TAG_TV: &str = "tv"; @@ -31,22 +32,22 @@ pub const EPG_TAG_ICON: &str = "icon"; pub enum XmlTagIcon { #[default] Undefined, - Src(String), + Src(Arc), Exists, } #[derive(Debug, Clone)] pub struct XmlTag { pub name: Arc, - pub value: Option, - pub attributes: Option, String>>, + pub value: Option>, + pub attributes: Option, Arc>>, pub children: Option>>, pub icon: XmlTagIcon, - pub normalized_epg_ids: Option>, + pub normalized_epg_ids: Option>>, } impl XmlTag { - pub(crate) fn new(name: Arc, attribs: Option, String>>) -> Self { + pub(crate) fn new(name: Arc, attribs: Option, Arc>>) -> Self { Self { name, value: None, @@ -57,7 +58,7 @@ impl XmlTag { } } - pub fn get_attribute_value(&self, attr_name: &str) -> Option<&String> { + pub fn get_attribute_value(&self, attr_name: &Arc) -> Option<&Arc> { self.attributes.as_ref().and_then(|attr| attr.get(attr_name)) } } @@ -67,7 +68,7 @@ impl XmlTag { pub struct Epg { pub priority: i16, pub logo_override: bool, - pub attributes: Option, String>>, + pub attributes: Option, Arc>>, pub children: Vec>, } @@ -75,13 +76,13 @@ impl Epg { pub async fn write_to_async( &self, writer: &mut quick_xml::writer::Writer, - rename_map: Option<&HashMap<&str, &str>>, + rename_map: Option<&HashMap, Arc>>, ) -> Result<(), quick_xml::Error> { // Start tv-element let mut elem = BytesStart::new("tv"); if let Some(attrs) = &self.attributes { for (k, v) in attrs { - elem.push_attribute((k.as_ref(), v.as_str())); + elem.push_attribute((k.as_ref(), v.as_ref())); } } writer.write_event_async(Event::Start(elem)).await?; @@ -96,7 +97,8 @@ impl Epg { .collect(); let mut write_counter = 0usize; - let mut current_channel_id: Option = None; + let mut current_channel_id: Option> = None; + let epg_id_key = EPG_ATTRIB_ID.intern(); while let Some((tag, ended)) = stack.pop() { if ended { @@ -112,12 +114,12 @@ impl Epg { let mut elem = BytesStart::new(tag.name.as_ref()); if let Some(attrs) = &tag.attributes { for (k, v) in attrs { - elem.push_attribute((k.as_ref(), v.as_str())); + elem.push_attribute((k.as_ref(), v.as_ref())); } } if tag.name.as_ref() == EPG_TAG_CHANNEL { - current_channel_id = tag.get_attribute_value(EPG_ATTRIB_ID).cloned(); + current_channel_id = tag.get_attribute_value(&epg_id_key).cloned(); } writer.write_event_async(Event::Start(elem)).await?; @@ -125,10 +127,10 @@ impl Epg { // write text let value_to_write = if tag.name.as_ref() == EPG_TAG_DISPLAY_NAME { current_channel_id.as_ref() - .and_then(|cid| rename_map.and_then(|m| m.get(cid.as_str())).copied()) - .or(tag.value.as_deref()) + .and_then(|cid| rename_map.and_then(|m| m.get(cid)) + .or(tag.value.as_ref())) } else { - tag.value.as_deref() + tag.value.as_ref() }; if let Some(text) = value_to_write { @@ -191,7 +193,7 @@ fn filter_channels_and_programmes( channels: &mut Vec, programmes: &mut Vec, ) { - let mut prog_map: HashMap> = HashMap::new(); + let mut prog_map: HashMap, Vec> = HashMap::new(); for prog in programmes.drain(..) { prog_map.entry(prog.channel.clone()).or_default().push(prog); } @@ -253,20 +255,20 @@ pub async fn parse_xmltv_for_web_ui_from_url(app_state: &Arc, url: &st } } -fn concat_text(t1: &String, t2: &str) -> String { - if t1.is_empty() { - t2.to_string() - } else if t1.ends_with('\\') { - let mut t = t1.clone(); - t.pop(); - format!("{t}'{t2}") - } else { - format!("{t1}{t2}") +fn concat_text(t1: Option<&Arc>, t2: &str) -> Arc { + match t1 { + None => t2.intern(), + Some(s) if s.ends_with('\\') => { + let mut t = s.to_string(); + t.pop(); + concat_string!(&t, "'", t2).intern() + } + Some(s) => concat_string!(s, t2).intern(), } } -pub fn get_attr_value(attr: &quick_xml::events::attributes::Attribute) -> Option { - attr.unescape_value().ok().map(|v| v.to_string()) +pub fn get_attr_value(attr: &quick_xml::events::attributes::Attribute) -> Option> { + attr.unescape_value().ok().map(|v| v.intern()) } // This function filters a timeslot starting from yesterday. @@ -360,13 +362,15 @@ async fn parse_xmltv_for_web_ui(reader: R) -> Resul if !text.is_empty() { if let Some(channel) = &mut current_channel { if current_tag == EPG_TAG_DISPLAY_NAME { - channel.title = concat_text(&channel.title, text); + channel.title = Some(concat_text(channel.title.as_ref(), text)); } } if let Some(program) = &mut current_programme { if current_tag == "title" { - program.title = concat_text(&program.title, text); + program.title = Some(concat_text(program.title.as_ref(), text)); + } else if current_tag == "desc" { + program.desc = Some(concat_text(program.desc.as_ref(), text)); } } } diff --git a/backend/src/processing/parser/xmltv.rs b/backend/src/processing/parser/xmltv.rs index a7a230b55..a32153beb 100644 --- a/backend/src/processing/parser/xmltv.rs +++ b/backend/src/processing/parser/xmltv.rs @@ -6,36 +6,16 @@ use crate::utils::compressed_file_reader_async::CompressedFileReaderAsync; use dashmap::DashMap; use quick_xml::events::{BytesStart, BytesText, Event}; use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; +use shared::concat_string; use shared::model::EpgNamePrefix; use shared::utils::{deunicode_string, Internable, CONSTANTS}; use std::borrow::Cow; -use std::cell::RefCell; use std::cmp::min; use std::collections::HashMap; use std::mem; use std::sync::{Arc, Mutex}; use tokio::io::AsyncRead; -struct EpgInterner { - map: RefCell>>, -} - -impl EpgInterner { - fn new() -> Self { - Self { map: RefCell::new(HashMap::new()) } - } - - fn intern(&self, s: &str) -> Arc { - let mut map = self.map.borrow_mut(); - if let Some(arc) = map.get(s) { - return arc.clone(); - } - let arc: Arc = Arc::from(s); - map.insert(s.to_string(), arc.clone()); - arc - } -} - /// Splits a string at the first delimiter if the prefix matches a known country code. /// /// Returns a tuple containing the country code prefix (if found) and the remainder of the string, both trimmed. If no valid prefix is found, returns `None` and the original input. @@ -131,16 +111,17 @@ impl TVGuide { fn prepare_tag(id_cache: &mut EpgIdCache, tag: &mut XmlTag, smart_match: bool) { { let maybe_epg_id = { - tag.get_attribute_value(EPG_ATTRIB_ID).cloned() + tag.get_attribute_value(&EPG_ATTRIB_ID.intern()).cloned() }; if let Some(epg_id) = maybe_epg_id { tag.normalized_epg_ids .get_or_insert_with(Vec::new) - .push(normalize_channel_name(&epg_id, &id_cache.smart_match_config)); + .push(normalize_channel_name(&epg_id, &id_cache.smart_match_config).intern()); } } if let Some(children) = &tag.children { + let src = "src".intern(); for child in children { match child.name.as_ref() { EPG_TAG_DISPLAY_NAME => { @@ -148,12 +129,12 @@ impl TVGuide { if let Some(name) = &child.value { tag.normalized_epg_ids .get_or_insert_with(Vec::new) - .push(normalize_channel_name(name, &id_cache.smart_match_config)); + .push(normalize_channel_name(name, &id_cache.smart_match_config).intern()); } } } EPG_TAG_ICON => { - if let Some(src) = child.get_attribute_value("src") { + if let Some(src) = child.get_attribute_value(&src) { if !src.is_empty() { tag.icon = XmlTagIcon::Src(src.clone()); // We cannot easily modify the child icon since it's inside Arc, @@ -167,7 +148,7 @@ impl TVGuide { } } - fn try_fuzzy_matching(id_cache: &mut EpgIdCache, epg_id: &str, tag: &XmlTag, fuzzy_matching: bool) -> bool { + fn try_fuzzy_matching(id_cache: &mut EpgIdCache, epg_id: &Arc, tag: &XmlTag, fuzzy_matching: bool) -> bool { let mut matched = tag .normalized_epg_ids .as_ref() @@ -176,10 +157,9 @@ impl TVGuide { let (fuzzy_matched, matched_normalized_name) = Self::find_best_fuzzy_match(id_cache, tag); if fuzzy_matched { 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)); + entry.replace(epg_id.clone()); + id_cache.channel_epg_id.insert(epg_id.clone()); matched = true; }); } @@ -206,7 +186,7 @@ impl TVGuide { /// println!("Best match: {:?}", matched); /// } /// ``` - fn find_best_fuzzy_match(id_cache: &mut EpgIdCache, tag: &XmlTag) -> (bool, Option) { + fn find_best_fuzzy_match(id_cache: &mut EpgIdCache, tag: &XmlTag) -> (bool, Option>) { let match_threshold = id_cache.smart_match_config.match_threshold; let best_match_threshold = id_cache.smart_match_config.best_match_threshold; @@ -215,16 +195,16 @@ impl TVGuide { }; // 1) Precalculation: (tag_normalized, tag_code) - let pre: Vec<(&str, String)> = normalized_epg_ids + let pre: Vec<(Arc, Arc)> = normalized_epg_ids .iter() - .map(|tn| (tn.as_str(), id_cache.phonetic(tn))) + .map(|tn| (tn.clone(), id_cache.phonetic(tn))) .collect(); // 2) Early exit if match >= best_match_threshold for (tag_normalized, tag_code) in &pre { if let Some(candidates) = id_cache.phonetics.get(tag_code) { if let Some(good_enough) = candidates.par_iter().find_any(|norm_key| { - let jw = strsim::jaro_winkler(norm_key.as_str(), tag_normalized); + let jw = strsim::jaro_winkler(norm_key, tag_normalized); #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] let score = min(100, (jw * 100.0).round() as u16); score >= best_match_threshold @@ -242,10 +222,10 @@ impl TVGuide { candidates .par_iter() .map(|norm_key| { - let jw = strsim::jaro_winkler(norm_key.as_str(), tag_normalized); + let jw = strsim::jaro_winkler(norm_key, tag_normalized); #[allow(clippy::cast_possible_truncation, clippy::cast_sign_loss)] let score = min(100, (jw * 100.0).round() as u16); - (score, norm_key.as_str()) + (score, norm_key) }) .reduce_with(|a, b| if a.0 >= b.0 { a } else { b }) }) @@ -255,7 +235,7 @@ impl TVGuide { if let Some((score, best_key)) = best { if score >= match_threshold { - return (true, Some(best_key.to_string())); + return (true, Some(Arc::clone(best_key))); } } @@ -276,17 +256,19 @@ impl TVGuide { /// assert!(!epg.children.is_empty()); /// } /// ``` - async fn process_epg_file(id_cache: &mut EpgIdCache<'_>, epg_source: &PersistedEpgSource) -> Option { + async fn process_epg_file(id_cache: &mut EpgIdCache, epg_source: &PersistedEpgSource) -> Option { + let epg_attrib_id = EPG_ATTRIB_ID.intern(); + let epg_attrib_channel = EPG_ATTRIB_CHANNEL.intern(); match CompressedFileReaderAsync::new(&epg_source.file_path).await { Ok(mut reader) => { let mut children: Vec> = vec![]; - let mut tv_attributes: Option, String>> = None; + let mut tv_attributes: Option, Arc>> = None; let smart_match = id_cache.smart_match_config.enabled; let fuzzy_matching = smart_match && id_cache.smart_match_config.fuzzy_matching; let mut filter_tags = |mut tag: XmlTag| { match tag.name.as_ref() { EPG_TAG_CHANNEL => { - let tag_epg_id = tag.get_attribute_value(EPG_ATTRIB_ID).map_or_else(String::new, std::string::ToString::to_string); + let tag_epg_id = tag.get_attribute_value(&epg_attrib_id).map_or_else(|| "".intern(), Internable::intern); if !tag_epg_id.is_empty() && !id_cache.processed.contains(&tag_epg_id) { Self::prepare_tag(id_cache, &mut tag, smart_match); if smart_match { @@ -294,22 +276,16 @@ impl TVGuide { children.push(Arc::new(tag)); id_cache.processed.insert(tag_epg_id); } - } else { - let borrowed_tag_epg_id = Cow::Borrowed(tag_epg_id.as_str()); - if id_cache.channel_epg_id.contains(&borrowed_tag_epg_id) { - children.push(Arc::new(tag)); - id_cache.processed.insert(tag_epg_id); - } + } else if id_cache.channel_epg_id.contains(&tag_epg_id) { + children.push(Arc::new(tag)); + id_cache.processed.insert(tag_epg_id); } } } EPG_TAG_PROGRAMME => { - if let Some(epg_id) = tag.get_attribute_value(EPG_ATTRIB_CHANNEL) { - if id_cache.processed.contains(epg_id) { - let borrowed_epg_id = Cow::Borrowed(epg_id.as_str()); - if id_cache.channel_epg_id.contains(&borrowed_epg_id) { - children.push(Arc::new(tag)); - } + if let Some(epg_id) = tag.get_attribute_value(&epg_attrib_channel) { + if id_cache.processed.contains(epg_id) && id_cache.channel_epg_id.contains(epg_id) { + children.push(Arc::new(tag)); } } } @@ -340,7 +316,7 @@ impl TVGuide { } } - pub async fn filter(&self, id_cache: &mut EpgIdCache<'_>) -> Option> { + pub async fn filter(&self, id_cache: &mut EpgIdCache) -> Option> { if id_cache.channel_epg_id.is_empty() && id_cache.normalized.is_empty() { return None; } @@ -355,16 +331,15 @@ impl TVGuide { } } - -fn handle_tag_start(callback: &mut F, stack: &mut Vec, e: &BytesStart, interner: &EpgInterner) +fn handle_tag_start(callback: &mut F, stack: &mut Vec, e: &BytesStart) where F: FnMut(XmlTag), { let binding = e.name(); let name_raw = String::from_utf8_lossy(binding.as_ref()); - let name = interner.intern(name_raw.as_ref()); + let name = name_raw.intern(); let (is_tv_tag, is_channel, is_program) = get_tag_types(&name); - let attributes = collect_tag_attributes(e, is_channel, is_program, interner); + let attributes = collect_tag_attributes(e, is_channel, is_program); let attribs = if attributes.is_empty() { None } else { Some(attributes) }; let tag = XmlTag::new(name, attribs); @@ -375,7 +350,6 @@ where } } - fn handle_tag_end(callback: &mut F, stack: &mut Vec) where F: FnMut(XmlTag), @@ -383,13 +357,13 @@ where if !stack.is_empty() { if let Some(tag) = stack.pop() { if tag.name.as_ref() == EPG_TAG_CHANNEL { - if let Some(chan_id) = tag.get_attribute_value(EPG_ATTRIB_ID) { + if let Some(chan_id) = tag.get_attribute_value(&EPG_ATTRIB_ID.intern()) { if !chan_id.is_empty() { callback(tag); } } } else if tag.name.as_ref() == EPG_TAG_PROGRAMME { - if let Some(chan_id) = tag.get_attribute_value(EPG_ATTRIB_CHANNEL) { + if let Some(chan_id) = tag.get_attribute_value(&EPG_ATTRIB_CHANNEL.intern()) { if !chan_id.is_empty() { callback(tag); } @@ -419,8 +393,10 @@ fn handle_text_tag(stack: &mut [XmlTag], e: &BytesText) { Cow::Borrowed(t) }; - let old = tag.value.get_or_insert_with(String::new); - old.push_str(&t_fixed); + tag.value = Some(match tag.value.take() { + None => t_fixed.intern(), + Some(old) => concat_string!(old.as_ref(), t_fixed.as_ref()).intern(), + }); } } } @@ -434,13 +410,12 @@ where let mut stack: Vec = vec![]; let mut xml_reader = quick_xml::reader::Reader::from_reader(async_file_reader(content)); let mut buf = Vec::::new(); - let interner = EpgInterner::new(); loop { match xml_reader.read_event_into_async(&mut buf).await { Ok(Event::Eof) => break, - Ok(Event::Start(e)) => handle_tag_start(callback, &mut stack, &e, &interner), + Ok(Event::Start(e)) => handle_tag_start(callback, &mut stack, &e), Ok(Event::Empty(e)) => { - handle_tag_start(callback, &mut stack, &e, &interner); + handle_tag_start(callback, &mut stack, &e); handle_tag_end(callback, &mut stack); } Ok(Event::End(_e)) => handle_tag_end(callback, &mut stack), @@ -460,24 +435,24 @@ fn get_tag_types(name: &str) -> (bool, bool, bool) { (is_tv_tag, is_channel, is_program) } -fn collect_tag_attributes(e: &BytesStart, is_channel: bool, is_program: bool, interner: &EpgInterner) -> HashMap, String> { +fn collect_tag_attributes(e: &BytesStart, is_channel: bool, is_program: bool) -> HashMap, Arc> { let attributes = e.attributes().filter_map(Result::ok) .filter_map(|a| { let key_binding = a.key; let key_raw = String::from_utf8_lossy(key_binding.as_ref()); - let key = interner.intern(key_raw.as_ref()); + let key = key_raw.intern(); if let Ok(value) = a.unescape_value().as_ref() { if value.is_empty() { None } else if (is_channel && key.as_ref() == EPG_ATTRIB_ID) || (is_program && key.as_ref() == EPG_ATTRIB_CHANNEL) { - Some((key, value.to_lowercase())) + Some((key, value.to_lowercase().intern())) } else { - Some((key, value.to_string())) + Some((key, value.intern())) } } else { None } - }).collect::, String>>(); + }).collect::, Arc>>(); attributes } @@ -486,11 +461,13 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option { None } else { let epg_children: Mutex>> = Mutex::new(Vec::new()); - let epg_attributes: Option, String>> = tv_guides.first().and_then(|t| t.attributes.clone()); + let epg_attributes: Option, Arc>> = tv_guides.first().and_then(|t| t.attributes.clone()); let count = tv_guides.iter().map(|tvg| tvg.children.len()).sum(); let channel_mapping: DashMap, i16> = DashMap::with_capacity(count); let mut sorted_guides = tv_guides.to_vec(); + let epg_attrib_id = EPG_ATTRIB_ID.intern(); + let epg_attrib_channel = EPG_ATTRIB_CHANNEL.intern(); // sort by priority sorted_guides.sort_by(|a, b| a.priority.cmp(&b.priority)); // if executed parallel it does not matter how we sort. @@ -498,7 +475,7 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option { let mut children = vec![]; guide.children.iter().for_each(|c| { if c.name.as_ref() == EPG_TAG_CHANNEL { - if let Some(chan_id) = c.get_attribute_value(EPG_ATTRIB_ID) { + if let Some(chan_id) = c.get_attribute_value(&epg_attrib_id) { let chan_id = chan_id.intern(); let should_add = { // if not stored @@ -522,7 +499,7 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option { }); guide.children.iter().for_each(|c| { if c.name.as_ref() == EPG_TAG_PROGRAMME { - if let Some(chan_id) = c.get_attribute_value(EPG_ATTRIB_CHANNEL) { + if let Some(chan_id) = c.get_attribute_value(&epg_attrib_channel) { let chan_id = chan_id.intern(); if let Some(stored_priority) = channel_mapping.get(&chan_id) { if *stored_priority == guide.priority { @@ -556,7 +533,6 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option { mod tests { use crate::model::{EpgSmartMatchConfig, PersistedEpgSource, TVGuide}; use crate::processing::parser::xmltv::normalize_channel_name; - use std::borrow::Cow; use std::collections::HashSet; use std::io; use std::path::PathBuf; @@ -587,10 +563,10 @@ mod tests { let tv_guide = TVGuide::new(vec![PersistedEpgSource { file_path, priority: 0, logo_override: false }]); let mut id_cache = EpgIdCache::new(None); - id_cache.channel_epg_id.insert(Cow::Owned("342".to_string())); + id_cache.channel_epg_id.insert(342u32.intern()); //id_cache.collect_epg_id(fp); - let channel_ids = HashSet::from(["342".to_string()]); + let channel_ids = HashSet::from([342u32.intern()]); match tv_guide.filter(&mut id_cache).await { None => assert!(false, "No epg filtered"), Some(epgs) => { @@ -632,6 +608,7 @@ mod tests { use crate::processing::processor::epg::EpgIdCache; use rphonetic::{Encoder, Metaphone}; use shared::model::{EpgNamePrefix, EpgSmartMatchConfigDto}; + use shared::utils::Internable; #[test] /// Demonstrates phonetic encoding (Metaphone) of normalized channel names with various prefixes and suffixes. diff --git a/backend/src/processing/playlist_watch.rs b/backend/src/processing/playlist_watch.rs index 1a5f19e29..ad11a11d2 100644 --- a/backend/src/processing/playlist_watch.rs +++ b/backend/src/processing/playlist_watch.rs @@ -6,7 +6,7 @@ use shared::model::{MsgKind, PlaylistGroup}; use crate::messaging::{send_message}; use crate::model::Config; use crate::utils; -use crate::utils::{binary_deserialize, binary_serialize}; +use crate::utils::{binary_deserialize, binary_serialize, file_exists_async}; pub async fn process_group_watch(client: &reqwest::Client, cfg: &Config, target_name: &str, pl: &PlaylistGroup) { let mut new_tree = BTreeSet::new(); @@ -21,7 +21,7 @@ pub async fn process_group_watch(client: &reqwest::Client, cfg: &Config, target_ Some(path) => { let save_path = path.as_path(); let mut changed = false; - if let Ok(true) = tokio::fs::try_exists(&path).await { + if file_exists_async(&path).await { if let Some(loaded_tree) = load_watch_tree(&path).await { // Find elements in set2 but not in set1 let added_difference: BTreeSet> = new_tree.difference(&loaded_tree).cloned().collect(); diff --git a/backend/src/processing/processor/epg.rs b/backend/src/processing/processor/epg.rs index bcec64e30..6c2a33d12 100644 --- a/backend/src/processing/processor/epg.rs +++ b/backend/src/processing/processor/epg.rs @@ -4,23 +4,23 @@ use crate::model::FetchedPlaylist; use crate::processing::parser::xmltv::normalize_channel_name; use log::{debug, trace, warn}; use rphonetic::{DoubleMetaphone, Encoder}; -use std::borrow::Cow; use std::collections::{HashMap, HashSet}; use shared::model::{EpgSmartMatchConfigDto, PlaylistItem, XtreamCluster}; use std::sync::Arc; +use shared::utils::Internable; -pub struct EpgIdCache<'a> { - pub channel_epg_id: HashSet>, - pub normalized: HashMap>, - pub phonetics: HashMap>, - pub processed: HashSet, +pub struct EpgIdCache { + pub channel_epg_id: HashSet>, + pub normalized: HashMap, Option>>, + pub phonetics: HashMap, HashSet>>, + pub processed: HashSet>, pub smart_match_config: EpgSmartMatchConfig, pub metaphone: DoubleMetaphone, pub smart_match_enabled: bool, // smart match is enabled, normalizing names pub fuzzy_match_enabled: bool, // fuzzy matching enabled } -impl EpgIdCache<'_> { +impl EpgIdCache { /// Creates a new `EpgIdCache` with configuration for smart and fuzzy matching. /// /// Initializes all internal caches and sets matching options based on the provided EPG configuration. If no configuration is given, defaults are used. @@ -65,7 +65,7 @@ impl EpgIdCache<'_> { /// cache.normalize_and_store("Discovery Channel", Some(&"discovery.epg".to_string())); /// assert!(cache.normalized.contains_key(&cache.normalize("Discovery Channel"))); /// ``` - fn normalize_and_store(&mut self, name: &str, epg_id: Option<&str>) { + fn normalize_and_store(&mut self, name: &str, epg_id: Option<&Arc>) { self.insert_normalized(name); if let Some(chan_epg_id) = epg_id { @@ -73,7 +73,7 @@ impl EpgIdCache<'_> { } } fn insert_normalized(&mut self, key: &str) { - let normalized = self.normalize(key); + let normalized = self.normalize(key).intern(); let phonetic = self.phonetic(&normalized); self.normalized.insert(normalized.clone(), None); @@ -96,12 +96,12 @@ impl EpgIdCache<'_> { normalize_channel_name(name, &self.smart_match_config) } - pub(crate) fn phonetic(&self, name: &str) -> String { + pub(crate) fn phonetic(&self, name: &Arc) -> Arc { let result = self.metaphone.encode(name); if result.is_empty() { - name.to_owned() + name.clone() } else { - result + result.intern() } } @@ -111,13 +111,13 @@ impl EpgIdCache<'_> { // Helper closure to process a single item // We use a closure here to capture `self` and avoid code duplication - let mut process_item = |name: &str, epg_channel_id: Option<&str>| { + let mut process_item = |name: &str, epg_channel_id: Option<&Arc>| { let mut missing_epg_id = true; // insert epg_id to known channel epg_ids if let Some(id) = epg_channel_id { if !id.is_empty() { missing_epg_id = false; - self.channel_epg_id.insert(Cow::Owned(id.to_string())); + self.channel_epg_id.insert(Arc::clone(id)); } } @@ -132,16 +132,16 @@ impl EpgIdCache<'_> { for channel in fp.items() { if channel.header.xtream_cluster == XtreamCluster::Live && channel.header.item_type.is_live() { - process_item(&channel.header.name, channel.header.epg_channel_id.as_deref()); + process_item(&channel.header.name, channel.header.epg_channel_id.as_ref()); } } } - pub fn match_with_normalized(&mut self, epg_id: &str, normalized_epg_ids: &[String]) -> bool { + pub fn match_with_normalized(&mut self, epg_id: &Arc, normalized_epg_ids: &[Arc]) -> bool { for key in normalized_epg_ids { if let Some(entry) = self.normalized.get_mut(key) { - entry.replace(epg_id.to_string()); - self.channel_epg_id.insert(epg_id.to_string().into()); + entry.replace(Arc::clone(epg_id)); + self.channel_epg_id.insert(Arc::clone(epg_id)); return true; } } @@ -161,17 +161,18 @@ impl EpgIdCache<'_> { /// let mut id_cache = EpgIdCache::new(None); /// assign_channel_epg(&mut new_epg, &mut playlist, &mut id_cache); /// ``` -async fn assign_channel_epg(new_epg: &mut Vec, fp: &mut FetchedPlaylist<'_>, id_cache: &mut EpgIdCache<'_>) { +async fn assign_channel_epg(new_epg: &mut Vec, fp: &mut FetchedPlaylist<'_>, id_cache: &mut EpgIdCache) { //id_cache.normalized.retain(|_, v| v.is_some()); if let Some(tv_guide) = &fp.epg { let mut processed_epgs = vec![]; if let Some(epg_sources) = tv_guide.filter(id_cache).await { let mut icon_assigned = HashSet::new(); + let epg_attrib_id = EPG_ATTRIB_ID.intern(); for epg_source in epg_sources { // icon tags - let icon_tags: HashMap<&str, &Arc> = epg_source.children.iter() + let icon_tags: HashMap<&Arc, &Arc> = epg_source.children.iter() .filter(|tag| tag.icon != XmlTagIcon::Undefined) - .filter_map(|tag| tag.get_attribute_value(EPG_ATTRIB_ID).map(|id| (id.as_str(), tag))) + .filter_map(|tag| tag.get_attribute_value(&epg_attrib_id).map(|id| (id, tag))) .collect(); let assign_values = |chan: &mut PlaylistItem| { @@ -184,7 +185,7 @@ async fn assign_channel_epg(new_epg: &mut Vec, fp: &mut FetchedPlaylist<'_> }; if not_found_in_epg { let try_match = |key: &str| { - let normalized = id_cache.normalize(key); + let normalized = id_cache.normalize(key).intern(); id_cache.normalized.get(&normalized).and_then(|epg_id| { epg_id.as_ref().map(|id| { trace!("Matched channel {} to epg {id:?}", chan.header.name); @@ -195,22 +196,22 @@ async fn assign_channel_epg(new_epg: &mut Vec, fp: &mut FetchedPlaylist<'_> if let Some(new_id) = try_match(&chan.header.name) .or_else(|| chan.header.epg_channel_id.as_deref().and_then(try_match)) { - chan.header.epg_channel_id = Some(new_id.into()); + chan.header.epg_channel_id = Some(new_id); } } } 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(&**epg_channel_id) { + 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.clone()); if epg_source.logo_override || chan.header.logo.is_empty() { trace!("Matched channel {} to epg icon {icon}", chan.header.name); - chan.header.logo = icon.clone().into(); + chan.header.logo = icon.clone(); } if epg_source.logo_override || chan.header.logo_small.is_empty() { - chan.header.logo_small = icon.clone().into(); + chan.header.logo_small = icon.clone(); } } } diff --git a/backend/src/repository/alias_repository.rs b/backend/src/repository/alias_repository.rs index c04bb2872..f0a793ed1 100644 --- a/backend/src/repository/alias_repository.rs +++ b/backend/src/repository/alias_repository.rs @@ -541,7 +541,7 @@ pub async fn csv_patch_batch_sort_by_exp_date( #[cfg(test)] mod tests { use crate::repository::csv_read_inputs_from_reader; - use crate::utils::{file_reader, resolve_env_var}; + use crate::utils::{file_reader}; use shared::model::InputType; use std::io::Cursor; diff --git a/backend/src/repository/epg_repository.rs b/backend/src/repository/epg_repository.rs index 4f13cee24..82912e83d 100644 --- a/backend/src/repository/epg_repository.rs +++ b/backend/src/repository/epg_repository.rs @@ -1,14 +1,14 @@ -use crate::model::Epg; -use crate::model::{Config, ConfigTarget, TargetOutput}; -use crate::repository::m3u_get_epg_file_path; -use shared::model::PlaylistGroup; -use crate::repository::{xtream_get_epg_file_path, xtream_get_storage_path}; -use crate::utils::{async_file_writer, debug_if_enabled}; +use crate::model::{Config, ConfigTarget, TargetOutput, XmlTagIcon}; +use crate::model::{Epg, EPG_ATTRIB_CHANNEL, EPG_ATTRIB_ID, EPG_TAG_CHANNEL, EPG_TAG_DISPLAY_NAME, EPG_TAG_ICON, EPG_TAG_PROGRAMME}; +use crate::repository::{m3u_get_epg_file_path_for_target, BPlusTree}; +use crate::repository::{xtream_get_epg_file_path_for_target, xtream_get_storage_path}; +use crate::utils::{debug_if_enabled, parse_xmltv_time}; use shared::error::{notify_err, TuliproxError}; +use shared::model::{EpgChannel, EpgProgramme, PlaylistGroup}; +use shared::utils::Internable; use std::collections::HashMap; use std::path::Path; -use tokio::io::AsyncWriteExt; - +use std::sync::Arc; pub const XML_PREAMBLE: &str = r#" @@ -24,21 +24,25 @@ pub const XML_PREAMBLE: &str = r#" // // DOCTYPE via events (DO NOT USE): // writer.write_event_async(quick_xml::events::Event::DocType(quick_xml::events::BytesText::new(r#"tv SYSTEM "xmltv.dtd""#))) // .await.map_err(|e| notify_err!("failed to write doctype: {}", e))?; -pub async fn epg_write_file(target: &ConfigTarget, epg: &Epg, path: &Path, playlist: Option<&[PlaylistGroup]>) -> Result<(), TuliproxError> { - let file = tokio::fs::File::create(path).await - .map_err(|e| notify_err!("failed to create epg file {}: {}", path.display(), e))?; +pub fn epg_write_file(target: &ConfigTarget, epg: &Epg, path: &Path, playlist: Option<&[PlaylistGroup]>) -> Result<(), TuliproxError> { + let tag_channel = EPG_TAG_CHANNEL.intern(); + let tag_programme = EPG_TAG_PROGRAMME.intern(); + let tag_display_name = EPG_TAG_DISPLAY_NAME.intern(); + let tag_icon = EPG_TAG_ICON.intern(); + let tag_title = "title".intern(); + let tag_desc = "desc".intern(); + let epg_id_attrib = EPG_ATTRIB_ID.intern(); + let channel_id_attrib = EPG_ATTRIB_CHANNEL.intern(); + let start_attrib = "start".intern(); + let stop_attrib = "stop".intern(); - // Use a larger buffer for sequential writes to reduce syscalls - let mut buf_writer = async_file_writer(file); - - // Header/DOCTYPE workaround: write both lines in a single call for fewer syscalls - buf_writer - .write_all(XML_PREAMBLE.as_bytes()) - .await - .map_err(|e| notify_err!("failed to write XML preamble {}: {}", path.display(), e))?; + if epg.children.is_empty() { + return Ok(()); + } + // If the epg titles differ from playlist, then we should use the ones from playlist // Build a temporary rename map with zero allocations (uses references) - let mut rename_map: HashMap<&str, &str> = HashMap::new(); + let mut rename_map: HashMap<&Arc, &Arc> = HashMap::new(); if let Some(pl) = playlist { for group in pl { for channel in &group.channels { @@ -51,43 +55,100 @@ pub async fn epg_write_file(target: &ConfigTarget, epg: &Epg, path: &Path, playl } } - // EPG content streamed via quick_xml writer (compact output) - let mut writer = quick_xml::writer::Writer::new(buf_writer); - epg - .write_to_async(&mut writer, if rename_map.is_empty() { None } else { Some(&rename_map) }) - .await - .map_err(|e| notify_err!("failed to write epg {}: {}", path.display(), e))?; + let mut channels: HashMap, EpgChannel> = + epg.children + .iter() + .filter(|tag| tag.name == tag_channel) + .filter_map(|tag| { + let channel_id = tag.get_attribute_value(&epg_id_attrib)?; + let mut title = rename_map.get(channel_id).map(|v| Arc::clone(v)); + let mut icon = match tag.icon { + XmlTagIcon::Src(ref url) => Some(Arc::clone(url)), + XmlTagIcon::Undefined | XmlTagIcon::Exists => None, + }; + if let Some(children) = tag.children.as_ref() { + for child in children { + if child.name == tag_display_name { + if title.is_none() { + title.clone_from(&child.value); + } + } else if icon.is_none() && child.name == tag_icon { + icon.clone_from(&child.value); + } + } + } + let channel = EpgChannel { + id: Arc::clone(channel_id), + title, + icon, + programmes: vec![], + }; - // Ensure buffers are flushed to the OS and capture any I/O error - let mut buf_writer = writer.into_inner(); - buf_writer - .flush() - .await - .map_err(|e| notify_err!("failed to flush epg {}: {}", path.display(), e))?; + Some((Arc::clone(channel_id), channel)) + }) + .collect(); + drop(rename_map); - buf_writer.shutdown().await.map_err(|e| notify_err!("failed to write epg {}: {}", path.display(), e))?; + epg.children.iter().filter(|tag| tag.name == tag_programme).for_each(|tag| { + if let Some(attribs) = tag.attributes.as_ref() { + let opt_channel_id = attribs.get(&channel_id_attrib); + let opt_start = attribs.get(&start_attrib); + let opt_stop = attribs.get(&stop_attrib); + if let (Some(channel_id), Some(start), Some(stop)) = (opt_channel_id, opt_start, opt_stop) { + if let (Some(start_time), Some(stop_time)) = (parse_xmltv_time(start), parse_xmltv_time(stop)) { + if let Some(channel) = channels.get_mut(channel_id) { + let mut title = None; + let mut desc = None; + if let Some(children) = tag.children.as_ref() { + for child in children { + if child.name == tag_title { + title.clone_from(&child.value); + } else if child.name == tag_desc { + desc.clone_from(&child.value); + } + } + channel.programmes.push(EpgProgramme { + start: start_time, + stop: stop_time, + channel: Arc::clone(channel_id), title, desc + }); + } + } + } + } + } + }); + + let mut tree = BPlusTree::, EpgChannel>::new(); + for (key, mut channel) in channels { + channel.programmes.sort_by_key(|p| p.start); + tree.insert(key, channel); + } + tree.store(path).map_err(|err| notify_err!("Failed to write epg for target {}: {} - {err}", target.name, path.display()))?; debug_if_enabled!("Epg for target {} written to {}", target.name, path.display()); Ok(()) } -pub async fn epg_write(cfg: &Config, target: &ConfigTarget, target_path: &Path, epg: Option<&Epg>, output: &TargetOutput, playlist: Option<&[PlaylistGroup]>) -> Result<(), TuliproxError> { +pub async fn epg_write_for_target(cfg: &Config, target: &ConfigTarget, target_path: &Path, + epg: Option<&Epg>, output: &TargetOutput, + playlist: Option<&[PlaylistGroup]>) -> Result<(), TuliproxError> { if let Some(epg_data) = epg { match output { TargetOutput::Xtream(_) => { match xtream_get_storage_path(cfg, &target.name) { Some(path) => { - let epg_path = xtream_get_epg_file_path(&path); + let epg_path = xtream_get_epg_file_path_for_target(&path); debug_if_enabled!("writing xtream epg to {}", epg_path.display()); - epg_write_file(target, epg_data, &epg_path, playlist).await?; + epg_write_file(target, epg_data, &epg_path, playlist)?; } None => return Err(notify_err!("failed to write epg for target: {}, storage path not found", target.name)), } } TargetOutput::M3u(_) => { - let path = m3u_get_epg_file_path(target_path); + let path = m3u_get_epg_file_path_for_target(target_path); debug_if_enabled!("writing m3u epg to {}", path.display()); - epg_write_file(target, epg_data, &path, playlist).await?; + epg_write_file(target, epg_data, &path, playlist)?; } TargetOutput::Strm(_) | TargetOutput::HdHomeRun(_) => {} } diff --git a/backend/src/repository/library_repository.rs b/backend/src/repository/library_repository.rs index 1e1ba1161..c029fb37f 100644 --- a/backend/src/repository/library_repository.rs +++ b/backend/src/repository/library_repository.rs @@ -7,6 +7,7 @@ use std::sync::Arc; use indexmap::IndexMap; use shared::model::UUIDType; use crate::repository::xtream_repository::CategoryKey; +use crate::utils::file_exists_async; pub async fn persist_input_library_playlist(app_config: &Arc, library_path: &Path, playlist: &[PlaylistGroup]) -> Result<(), TuliproxError> { if playlist.is_empty() { @@ -30,7 +31,7 @@ pub async fn persist_input_library_playlist(app_config: &Arc, library pub async fn load_input_local_library_playlist(app_config: &Arc, lib_path: &Path) -> Result, TuliproxError> { let mut groups: IndexMap = IndexMap::new(); - if tokio::fs::try_exists(lib_path).await.unwrap_or(false) { + if file_exists_async(lib_path).await { // Load Items let _file_lock = app_config.file_locks.read_lock(lib_path).await; if let Ok(mut query) = BPlusTreeQuery::::try_new(lib_path) { diff --git a/backend/src/repository/m3u_repository.rs b/backend/src/repository/m3u_repository.rs index b868dce50..3adb0ba9e 100644 --- a/backend/src/repository/m3u_repository.rs +++ b/backend/src/repository/m3u_repository.rs @@ -8,7 +8,7 @@ use crate::repository::storage::{get_input_storage_path, get_target_storage_path use crate::repository::storage_const; use crate::repository::xtream_repository::CategoryKey; use crate::utils; -use crate::utils::{async_file_writer, FileReadGuard, IO_BUFFER_SIZE}; +use crate::utils::{async_file_writer, file_exists_async, FileReadGuard, IO_BUFFER_SIZE}; use indexmap::IndexMap; use log::error; use shared::concat_string; @@ -33,9 +33,9 @@ pub fn m3u_get_file_path_for_db(target_path: &Path) -> PathBuf { target_path.join(PathBuf::from(concat_string!(storage_const::FILE_M3U, ".", storage_const::FILE_SUFFIX_DB))) } -pub fn m3u_get_epg_file_path(target_path: &Path) -> PathBuf { +pub fn m3u_get_epg_file_path_for_target(target_path: &Path) -> PathBuf { let path = target_path.join(PathBuf::from(concat_string!(storage_const::FILE_M3U, ".", storage_const::FILE_SUFFIX_DB))); - utils::add_prefix_to_filename(&path, "epg_", Some("xml")) + utils::add_prefix_to_filename(&path, "epg_", Some(storage_const::FILE_SUFFIX_DB)) } macro_rules! await_playlist_write { @@ -181,7 +181,7 @@ where SortKey: for<'de> Deserialize<'de> + Send + 'static, { let file_lock = app_config.file_locks.read_lock(m3u_path).await; - if !tokio::fs::try_exists(m3u_path).await.unwrap_or(false) { + if !file_exists_async(m3u_path).await { return None; } @@ -233,7 +233,7 @@ pub async fn persist_input_m3u_playlist(app_config: &Arc, m3u_path: & pub async fn load_input_m3u_playlist(app_config: &Arc, m3u_path: &Path) -> Result, TuliproxError> { let mut groups: IndexMap = IndexMap::new(); - if tokio::fs::try_exists(m3u_path).await.unwrap_or(false) { + if file_exists_async(m3u_path).await { // Load Items let _file_lock = app_config.file_locks.read_lock(m3u_path).await; if let Ok(mut query) = BPlusTreeQuery::, M3uPlaylistItem>::try_new(m3u_path) { diff --git a/backend/src/repository/playlist_repository.rs b/backend/src/repository/playlist_repository.rs index 765ad1f39..5caa26b37 100644 --- a/backend/src/repository/playlist_repository.rs +++ b/backend/src/repository/playlist_repository.rs @@ -3,7 +3,7 @@ use crate::model::Epg; use crate::model::{AppConfig, ConfigInput, ConfigTarget, TargetOutput}; use crate::processing::processor::playlist::{apply_filter_to_playlist, PlaylistProcessingContext}; use crate::repository::{BPlusTree, BPlusTreeQuery}; -use crate::repository::epg_write; +use crate::repository::epg_write_for_target; use crate::repository::{load_input_m3u_playlist, m3u_get_file_path_for_db, m3u_write_playlist, persist_input_m3u_playlist}; use crate::repository::{ensure_target_storage_path, get_input_storage_path, get_target_id_mapping_file, get_target_storage_path}; use crate::repository::FILE_SUFFIX_DB; @@ -110,7 +110,7 @@ pub async fn persist_playlist(app_config: &Arc, playlist: &mut [Playl Ok(()) => { if !pl.is_empty() { let epg_pl: &[PlaylistGroup] = pl; - if let Err(err) = epg_write(config, target, &target_path, epg, output, Some(epg_pl)).await { + if let Err(err) = epg_write_for_target(config, target, &target_path, epg, output, Some(epg_pl)).await { errors.push(err); } } diff --git a/backend/src/repository/target_id_mapping.rs b/backend/src/repository/target_id_mapping.rs index 7c5fd26f6..63e7cf6af 100644 --- a/backend/src/repository/target_id_mapping.rs +++ b/backend/src/repository/target_id_mapping.rs @@ -248,7 +248,7 @@ mod tests { // Create mapping in disk-only mode let uuid1 = UUIDType::default(); { - let mut mapping = TargetIdMapping::new(&path)?; + let mut mapping = TargetIdMapping::new(&path, false)?; let vid1 = mapping.get_and_update_virtual_id(&uuid1, 100, PlaylistItemType::Live, 0); assert_eq!(vid1, 1); mapping.persist().map_err(|_| info_err!("Failed to persist mapping"))?; @@ -256,7 +256,7 @@ mod tests { // Reopen and verify persistence { - let mut mapping = TargetIdMapping::new(&path)?; + let mut mapping = TargetIdMapping::new(&path, false)?; let vid1_again = mapping.get_and_update_virtual_id(&uuid1, 100, PlaylistItemType::Live, 0); assert_eq!(vid1_again, 1); // Should get same virtual_id } @@ -271,7 +271,7 @@ mod tests { let uuid1 = UUIDType::default(); { - let mut mapping = TargetIdMapping::new(&path)?; + let mut mapping = TargetIdMapping::new(&path, false)?; let vid1 = mapping.get_and_update_virtual_id(&uuid1, 100, PlaylistItemType::Video, 0); assert_eq!(vid1, 1); mapping.persist().map_err(|err| info_err!("{err}"))?; @@ -279,7 +279,7 @@ mod tests { // Reopen with memory cache and verify { - let mut mapping = TargetIdMapping::new(&path)?; + let mut mapping = TargetIdMapping::new(&path, false)?; let vid1_again = mapping.get_and_update_virtual_id(&uuid1, 100, PlaylistItemType::Video, 0); assert_eq!(vid1_again, 1); } diff --git a/backend/src/repository/user_repository.rs b/backend/src/repository/user_repository.rs index 36b0a1e70..928637cba 100644 --- a/backend/src/repository/user_repository.rs +++ b/backend/src/repository/user_repository.rs @@ -5,7 +5,7 @@ use crate::repository::BPlusTree; use crate::repository::storage_const; use crate::repository::xtream_get_playlist_categories; use crate::utils; -use crate::utils::json_write_documents_to_file; +use crate::utils::{file_exists_async, json_write_documents_to_file}; use chrono::Local; use log::error; use shared::model::{PlaylistBouquetDto, PlaylistClusterBouquetDto, ProxyType, ProxyUserStatus, TargetType, XtreamCluster}; @@ -260,7 +260,7 @@ async fn save_xtream_user_bouquet_for_target(config: &Config, target_name: &str, if let Some(xtream_categories) = xtream_get_playlist_categories(config, target_name, cluster).await { let filtered: Vec = xtream_categories.iter().filter(|p| bouquet_categories.contains(&p.name)).cloned().collect(); if filtered.is_empty() { - if tokio::fs::try_exists(&bouquet_path).await.is_ok_and(|r| r) { + if file_exists_async(&bouquet_path).await { tokio::fs::remove_file(bouquet_path).await?; } } else { diff --git a/backend/src/repository/xtream_repository.rs b/backend/src/repository/xtream_repository.rs index 3d53344d3..14ae0d002 100644 --- a/backend/src/repository/xtream_repository.rs +++ b/backend/src/repository/xtream_repository.rs @@ -8,7 +8,7 @@ use crate::repository::storage::{get_input_storage_path, get_target_id_mapping_f use crate::repository::storage_const; use crate::repository::target_id_mapping::VirtualIdRecord; use crate::repository::xtream_playlist_iterator::XtreamPlaylistJsonIterator; -use crate::utils::file_reader; +use crate::utils::{file_exists_async, file_reader}; use crate::utils::json_write_documents_to_file; use crate::utils::FileReadGuard; use bytes::Bytes; @@ -26,7 +26,7 @@ use std::fs::File; use std::io::{Error, ErrorKind}; use std::path::{Path, PathBuf}; use std::sync::Arc; -use shared::notify_err_res; +use shared::{concat_string, notify_err_res}; macro_rules! cant_write_result { ($path:expr, $err:expr) => { @@ -218,12 +218,12 @@ pub fn xtream_get_storage_path(cfg: &Config, target_name: &str) -> Option PathBuf { - path.join(storage_const::FILE_EPG) +pub fn xtream_get_epg_file_path_for_target(path: &Path) -> PathBuf { + path.join(concat_string!("epg.", storage_const::FILE_SUFFIX_DB)) } fn xtream_get_file_path_for_name(storage_path: &Path, name: &str) -> PathBuf { - storage_path.join(format!("{name}.{}", storage_const::FILE_SUFFIX_DB)) + storage_path.join(concat_string!(name, ".", storage_const::FILE_SUFFIX_DB)) } pub fn xtream_get_file_path(storage_path: &Path, cluster: XtreamCluster) -> PathBuf { @@ -599,7 +599,7 @@ where SortKey: for<'de> Deserialize<'de> + Send + 'static, { let file_lock = app_config.file_locks.read_lock(xtream_path).await; - if !tokio::fs::try_exists(xtream_path).await.unwrap_or(false) { + if !file_exists_async(xtream_path).await { return None; } @@ -674,7 +674,7 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ // load for cluster in XTREAM_CLUSTER { let xtream_path = xtream_get_file_path(storage_path, cluster); - if tokio::fs::try_exists(&xtream_path).await.unwrap_or(false) { + if file_exists_async(&xtream_path).await { let file_lock = app_config.file_locks.read_lock(&xtream_path).await; let stored_entries = stored_scratch.get_mut(cluster); if let Ok(mut query) = BPlusTreeQuery::::try_new(&xtream_path) { @@ -745,7 +745,7 @@ pub async fn persist_input_xtream_playlist(app_config: &Arc, storage_ }; let data = fetched_categories.get_mut(cluster); // if there is no data save only if no file exists! Prevent data loss from failed download attempt - if !data.is_empty() || !tokio::fs::try_exists(&col_path).await.unwrap_or(false) { + if !data.is_empty() || !file_exists_async(&col_path).await { let lock = app_cfg.file_locks.write_lock(&col_path).await; if let Err(err) = json_write_documents_to_file(&col_path, data).await { errors.push(format!("Persisting collection failed: {}: {err}", col_path.display())); diff --git a/backend/src/utils/db_viewer.rs b/backend/src/utils/db_viewer.rs index 96bb8af36..91e5d24ec 100644 --- a/backend/src/utils/db_viewer.rs +++ b/backend/src/utils/db_viewer.rs @@ -2,27 +2,36 @@ use crate::repository::{BPlusTreeDiskIterator, BPlusTreeQuery}; use env_logger::{Builder, Target}; use log::{error, LevelFilter}; use serde::{Deserialize, Serialize}; -use shared::model::{M3uPlaylistItem, XtreamPlaylistItem}; +use shared::model::{EpgChannel, M3uPlaylistItem, XtreamPlaylistItem}; use std::io::Write; -use std::path::{PathBuf}; +use std::path::PathBuf; +use std::sync::Arc; #[derive(Debug, Copy, Clone)] enum DbType { Xtream, M3u, + Epg, } -pub fn db_viewer(xtream_filename: Option<&str>, m3u_filename: Option<&str>) { +pub fn db_viewer(xtream_filename: Option<&str>, m3u_filename: Option<&str>, epg_filename: Option<&str>) { let mut any_processed = false; if let Some(filename) = xtream_filename { any_processed = true; - if !dump_db(filename, DbType::Xtream) { - exit_app(1); - } + if !dump_db(filename, DbType::Xtream) { + exit_app(1); + } } if let Some(filename) = m3u_filename { any_processed = true; - if ! dump_db(filename, DbType::M3u) { + if !dump_db(filename, DbType::M3u) { + exit_app(1); + } + } + + if let Some(filename) = epg_filename { + any_processed = true; + if !dump_db(filename, DbType::Epg) { exit_app(1); } } @@ -53,6 +62,12 @@ fn dump_db(filename: &str, db_type: DbType) -> bool { return print_json_from_iter(iterator); } } + DbType::Epg => { + if let Ok(mut query) = BPlusTreeQuery::, EpgChannel>::try_new(&path) { + let iterator = query.iter(); + return print_json_from_iter(iterator); + } + } } } Err(err) => { @@ -63,8 +78,9 @@ fn dump_db(filename: &str, db_type: DbType) -> bool { false } -fn print_json_from_iter

(iterator: BPlusTreeDiskIterator) -> bool +fn print_json_from_iter(iterator: BPlusTreeDiskIterator) -> bool where + K: Ord + Serialize + for<'de> Deserialize<'de> + Clone, P: Serialize + for<'de> Deserialize<'de> + Clone, { let mut error_count = 0; @@ -83,7 +99,7 @@ where Err(err) => { error!("Failed: {err}"); error_count += 1; - }, + } } } println!("]"); diff --git a/backend/src/utils/epg_parser.rs b/backend/src/utils/epg_parser.rs index 2daea07a6..537828a8f 100644 --- a/backend/src/utils/epg_parser.rs +++ b/backend/src/utils/epg_parser.rs @@ -1,223 +1,281 @@ -use quick_xml::events::{Event, BytesStart}; -use quick_xml::reader::Reader; -use tokio::io::{AsyncRead, AsyncBufRead, BufReader}; -use shared::error::{TuliproxError, info_err}; -use chrono::{DateTime, Duration, FixedOffset, NaiveDateTime, TimeZone}; -use crate::utils::{obscure_text}; -use crate::model::{EPG_TAG_ICON, EPG_TAG_PROGRAMME}; -use log::error; +use std::sync::Arc; +use chrono::{DateTime, Offset, TimeZone, Utc}; +use chrono_tz::Tz; +use crate::model::{ConfigTarget, ProxyUserCredentials}; +use shared::model::PlaylistItemType; +use crate::api::model::AppState; -pub trait EpgConsumer: Send { - fn handle_event(&mut self, event: &Event<'_>, decoder: quick_xml::Decoder) -> impl std::future::Future> + Send; -} -pub struct EpgProcessor { - reader: Reader, - offset_minutes: i32, - rewrite_urls: bool, - rewrite_base_url: String, - encrypt_secret: [u8; 16], - filter_channel_id: Option>, -} - -impl EpgProcessor> { - pub fn new( - reader: R, - offset_minutes: i32, - rewrite_urls: bool, - rewrite_base_url: String, - encrypt_secret: [u8; 16], - filter_channel_id: Option>, - ) -> Self { - Self { - reader: Reader::from_reader(BufReader::new(reader)), - offset_minutes, - rewrite_urls, - rewrite_base_url, - encrypt_secret, - filter_channel_id, +/// Parses user-defined EPG timeshift configuration. +/// Supports either a numeric offset (e.g. "+2:30", "-1:15") +/// or a timezone name (e.g. "`Europe/Berlin`", "`UTC`", "`America/New_York`"). +/// +/// Returns the total offset in minutes (i32). +fn parse_timeshift(time_shift: Option<&str>) -> Option { + time_shift.and_then(|offset| { + // Try to parse as timezone name first + if let Ok(tz) = offset.parse::() { + // Determine the current UTC offset of that timezone (including DST) + let now = Utc::now(); + let local_time = tz.from_utc_datetime(&now.naive_utc()); + let offset_minutes = local_time.offset().fix().local_minus_utc() / 60; + return Some(offset_minutes); } + + // If not a timezone, try to parse as numeric offset + let sign_factor = if offset.starts_with('-') { -1 } else { 1 }; + let offset = offset.trim_start_matches(&['-', '+'][..]); + + let parts: Vec<&str> = offset.split(':').collect(); + let hours: i32 = parts.first().and_then(|h| h.parse().ok()).unwrap_or(0); + let minutes: i32 = parts.get(1).and_then(|m| m.parse().ok()).unwrap_or(0); + + let total_minutes = hours * 60 + minutes; + (total_minutes > 0).then_some(sign_factor * total_minutes) + }) +} + + +#[derive(Debug, Clone)] +pub struct EpgProcessingOptions { + pub rewrite_urls: bool, + pub offset_minutes: i32, + pub encrypt_secret: [u8; 16], +} + +pub fn get_epg_processing_options(app_state: &Arc, user: &ProxyUserCredentials, target: &Arc) -> EpgProcessingOptions { + let rewrite_resources = app_state.app_config.is_reverse_proxy_resource_rewrite_enabled(); + let encrypt_secret = app_state.app_config.get_reverse_proxy_rewrite_secret().unwrap_or_else(|| app_state.app_config.encrypt_secret); + + // If redirect is true → rewrite_urls = false → keep original + // If redirect is false and rewrite_resources is true → rewrite_urls = true → rewriting allowed + // If redirect is false and rewrite_resources is false → rewrite_urls = false → no rewriting + let redirect = user.proxy.is_redirect(PlaylistItemType::Live) || target.is_force_redirect(PlaylistItemType::Live); + let rewrite_urls = !redirect && rewrite_resources; + + // Use 0 for timeshift if None + let timeshift = parse_timeshift(user.epg_timeshift.as_deref()).unwrap_or(0); + EpgProcessingOptions { + rewrite_urls, + offset_minutes: timeshift, + encrypt_secret, } } - -#[allow(clippy::too_many_lines)] -impl EpgProcessor { - pub async fn process(&mut self, consumer: &mut C) -> Result<(), TuliproxError> { - let mut buf = Vec::with_capacity(4096); - let duration = Duration::minutes(i64::from(self.offset_minutes)); - let mut skip_depth = None; - - loop { - buf.clear(); - let event = match self.reader.read_event_into_async(&mut buf).await { - Ok(e) => e, - Err(e) => { - error!("Error reading epg XML event: {e}"); - return Err(info_err!("Error reading epg XML event: {}", e)); - } - }; - - if let Some(flt) = &self.filter_channel_id { - match &event { - Event::Start(e) => { - if skip_depth.is_none() { - let should_skip = match e.name().as_ref() { - b"channel" => { - e.attributes() - .filter_map(Result::ok) - .find(|a| a.key.as_ref() == b"id") - .and_then(|a| a.unescape_value().ok()) - .is_some_and(|v| flt.as_ref() != v.as_ref()) - } - b"programme" => { - e.attributes() - .filter_map(Result::ok) - .find(|a| a.key.as_ref() == b"channel") - .and_then(|a| a.unescape_value().ok()) - .is_some_and(|v| flt.as_ref() != v.as_ref()) - } - _ => false, - }; - - if should_skip { - skip_depth = Some(1); - continue; - } - } else { - skip_depth = skip_depth.map(|d| d + 1); - continue; - } - } - Event::End(_) => { - if let Some(depth) = skip_depth { - if depth == 1 { - skip_depth = None; - } else { - skip_depth = Some(depth - 1); - } - continue; - } - } - Event::Empty(_) => { - if skip_depth.is_some() { - continue; - } - } - _ => {} - } - - if skip_depth.is_some() { - continue; - } - } - - match event { - Event::Start(ref e) if self.offset_minutes != 0 && e.name().as_ref() == b"programme" => { - let mut elem = BytesStart::new(EPG_TAG_PROGRAMME); - for attr in e.attributes() { - match attr { - Ok(attr) if attr.key.as_ref() == b"start" => { - if let Ok(start_value) = attr.decode_and_unescape_value(self.reader.decoder()) { - elem.push_attribute(("start", time_correct(&start_value, &duration).as_str())); - } else { - elem.push_attribute(attr); - } - } - Ok(attr) if attr.key.as_ref() == b"stop" => { - if let Ok(stop_value) = attr.decode_and_unescape_value(self.reader.decoder()) { - elem.push_attribute(("stop", time_correct(&stop_value, &duration).as_str())); - } else { - elem.push_attribute(attr); - } - } - Ok(attr) => { - elem.push_attribute(attr); - } - Err(e) => { - error!("Error parsing epg attribute: {e}"); - } - } - } - consumer.handle_event(&Event::Start(elem), self.reader.decoder()).await?; - } - ref event @ (Event::Empty(ref e) | Event::Start(ref e)) if self.rewrite_urls && e.name().as_ref() == b"icon" => { - let mut elem = BytesStart::new(EPG_TAG_ICON); - for attr in e.attributes() { - match attr { - Ok(attr) if attr.key.as_ref() == b"src" => { - if let Some(icon) = get_attr_value_unescaped(&attr, self.reader.decoder()) { - if icon.is_empty() { - elem.push_attribute(attr); - } else { - let rewritten_url = if let Ok(encrypted) = obscure_text(&self.encrypt_secret, &icon) { - format!("{}{}", self.rewrite_base_url, encrypted) - } else { - icon - }; - elem.push_attribute(("src", rewritten_url.as_str())); - } - } else { - elem.push_attribute(attr); - } - } - Ok(attr) => { - elem.push_attribute(attr); - } - Err(e) => { - error!("Error parsing epg attribute: {e}"); - } - } - } - - let out_event = match event { - Event::Empty(_) => Event::Empty(elem), - Event::Start(_) => Event::Start(elem), - _ => unreachable!(), - }; - consumer.handle_event(&out_event, self.reader.decoder()).await?; - } - Event::Decl(_) | Event::DocType(_) => {}, - Event::Eof => break, - ref event => { - consumer.handle_event(event, self.reader.decoder()).await?; - } - } - } - Ok(()) - } -} +// +// pub trait EpgConsumer: Send { +// fn handle_event(&mut self, event: &Event<'_>, decoder: quick_xml::Decoder) -> impl std::future::Future> + Send; +// } +// +// pub struct EpgProcessor { +// reader: Reader, +// epg_processing_options: EpgProcessingOptions, +// rewrite_base_url: String, +// filter_channel_id: Option>, +// limit: u32 +// } +// +// impl EpgProcessor> { +// pub fn new( +// reader: R, +// epg_processing_options: EpgProcessingOptions, +// rewrite_base_url: String, +// filter_channel_id: Option>, +// limit: u32 +// ) -> Self { +// Self { +// reader: Reader::from_reader(BufReader::new(reader)), +// epg_processing_options, +// rewrite_base_url, +// filter_channel_id, +// limit +// } +// } +// } +// +// #[allow(clippy::too_many_lines)] +// impl EpgProcessor { +// pub async fn process(&mut self, consumer: &mut C) -> Result<(), TuliproxError> { +// let mut buf = Vec::with_capacity(4096); +// let duration = Duration::minutes(i64::from(self.epg_processing_options.offset_minutes)); +// let mut skip_depth = None; +// +// loop { +// buf.clear(); +// let event = match self.reader.read_event_into_async(&mut buf).await { +// Ok(e) => e, +// Err(e) => { +// error!("Error reading epg XML event: {e}"); +// return Err(info_err!("Error reading epg XML event: {}", e)); +// } +// }; +// +// if let Some(flt) = &self.filter_channel_id { +// match &event { +// Event::Start(e) => { +// if skip_depth.is_none() { +// let should_skip = match e.name().as_ref() { +// b"channel" => { +// e.attributes() +// .filter_map(Result::ok) +// .find(|a| a.key.as_ref() == b"id") +// .and_then(|a| a.unescape_value().ok()) +// .is_some_and(|v| flt.as_ref() != v.as_ref()) +// } +// b"programme" => { +// e.attributes() +// .filter_map(Result::ok) +// .find(|a| a.key.as_ref() == b"channel") +// .and_then(|a| a.unescape_value().ok()) +// .is_some_and(|v| flt.as_ref() != v.as_ref()) +// } +// _ => false, +// }; +// +// if should_skip { +// skip_depth = Some(1); +// continue; +// } +// } else { +// skip_depth = skip_depth.map(|d| d + 1); +// continue; +// } +// } +// Event::End(_) => { +// if let Some(depth) = skip_depth { +// if depth == 1 { +// skip_depth = None; +// } else { +// skip_depth = Some(depth - 1); +// } +// continue; +// } +// } +// Event::Empty(_) => { +// if skip_depth.is_some() { +// continue; +// } +// } +// _ => {} +// } +// +// if skip_depth.is_some() { +// continue; +// } +// } +// +// match event { +// Event::Start(ref e) if self.epg_processing_options.offset_minutes != 0 && e.name().as_ref() == b"programme" => { +// let mut elem = BytesStart::new(EPG_TAG_PROGRAMME); +// for attr in e.attributes() { +// match attr { +// Ok(attr) if attr.key.as_ref() == b"start" => { +// if let Ok(start_value) = attr.decode_and_unescape_value(self.reader.decoder()) { +// elem.push_attribute(("start", time_correct(&start_value, &duration).as_str())); +// } else { +// elem.push_attribute(attr); +// } +// } +// Ok(attr) if attr.key.as_ref() == b"stop" => { +// if let Ok(stop_value) = attr.decode_and_unescape_value(self.reader.decoder()) { +// elem.push_attribute(("stop", time_correct(&stop_value, &duration).as_str())); +// } else { +// elem.push_attribute(attr); +// } +// } +// Ok(attr) => { +// elem.push_attribute(attr); +// } +// Err(e) => { +// error!("Error parsing epg attribute: {e}"); +// } +// } +// } +// consumer.handle_event(&Event::Start(elem), self.reader.decoder()).await?; +// } +// ref event @ (Event::Empty(ref e) | Event::Start(ref e)) if self.epg_processing_options.rewrite_urls && e.name().as_ref() == b"icon" => { +// let mut elem = BytesStart::new(EPG_TAG_ICON); +// for attr in e.attributes() { +// match attr { +// Ok(attr) if attr.key.as_ref() == b"src" => { +// if let Some(icon) = get_attr_value_unescaped(&attr, self.reader.decoder()) { +// if icon.is_empty() { +// elem.push_attribute(attr); +// } else { +// let rewritten_url = if let Ok(encrypted) = obscure_text(&self.epg_processing_options.encrypt_secret, &icon) { +// format!("{}{}", self.rewrite_base_url, encrypted) +// } else { +// icon +// }; +// elem.push_attribute(("src", rewritten_url.as_str())); +// } +// } else { +// elem.push_attribute(attr); +// } +// } +// Ok(attr) => { +// elem.push_attribute(attr); +// } +// Err(e) => { +// error!("Error parsing epg attribute: {e}"); +// } +// } +// } +// +// let out_event = match event { +// Event::Empty(_) => Event::Empty(elem), +// Event::Start(_) => Event::Start(elem), +// _ => unreachable!(), +// }; +// consumer.handle_event(&out_event, self.reader.decoder()).await?; +// } +// Event::Decl(_) | Event::DocType(_) => {}, +// Event::Eof => break, +// ref event => { +// consumer.handle_event(event, self.reader.decoder()).await?; +// } +// } +// } +// Ok(()) +// } +// } +// +// /// # Panics +// /// unwrap for `FixedOffset` should not panic! +// pub fn time_correct(original: &str, shift: &Duration) -> String { +// let (datetime_part, tz_part) = if let Some((dt, tz)) = original.trim().rsplit_once(' ') { +// (dt, tz) +// } else { +// (original.trim(), "+0000") +// }; +// +// let Ok(naive_dt) = NaiveDateTime::parse_from_str(datetime_part, "%Y%m%d%H%M%S") else { return original.to_string() }; +// +// let tz_offset_minutes = if tz_part.len() == 5 { +// let bytes = tz_part.as_bytes(); +// let sign = if bytes.first() == Some(&b'-') { -1 } else { 1 }; +// let hours: i32 = tz_part.get(1..3).and_then(|s| s.parse().ok()).unwrap_or(0); +// let mins: i32 = tz_part.get(3..5).and_then(|s| s.parse().ok()).unwrap_or(0); +// sign * (hours * 60 + mins) +// } else { +// 0 +// }; +// +// let tz = FixedOffset::east_opt(tz_offset_minutes * 60).unwrap_or(FixedOffset::east_opt(0).unwrap()); // should not panic +// +// let dt: DateTime = tz +// .from_local_datetime(&naive_dt) +// .single() +// .unwrap_or_else(|| tz.from_utc_datetime(&naive_dt)); +// +// let shifted_dt = dt + *shift; +// +// format!("{} {}", shifted_dt.format("%Y%m%d%H%M%S"), format_offset(tz_offset_minutes)) +// } /// # Panics /// unwrap for `FixedOffset` should not panic! -pub fn time_correct(original: &str, shift: &Duration) -> String { - let (datetime_part, tz_part) = if let Some((dt, tz)) = original.trim().rsplit_once(' ') { - (dt, tz) - } else { - (original.trim(), "+0000") - }; - - let Ok(naive_dt) = NaiveDateTime::parse_from_str(datetime_part, "%Y%m%d%H%M%S") else { return original.to_string() }; - - let tz_offset_minutes = if tz_part.len() == 5 { - let bytes = tz_part.as_bytes(); - let sign = if bytes.first() == Some(&b'-') { -1 } else { 1 }; - let hours: i32 = tz_part.get(1..3).and_then(|s| s.parse().ok()).unwrap_or(0); - let mins: i32 = tz_part.get(3..5).and_then(|s| s.parse().ok()).unwrap_or(0); - sign * (hours * 60 + mins) - } else { - 0 - }; - - let tz = FixedOffset::east_opt(tz_offset_minutes * 60).unwrap_or(FixedOffset::east_opt(0).unwrap()); // should not panic - - let dt: DateTime = tz - .from_local_datetime(&naive_dt) - .single() - .unwrap_or_else(|| tz.from_utc_datetime(&naive_dt)); - - let shifted_dt = dt + *shift; - - format!("{} {}", shifted_dt.format("%Y%m%d%H%M%S"), format_offset(tz_offset_minutes)) +pub fn apply_offset(ts_utc: i64, offset_minutes: i32) -> i64 { + ts_utc + i64::from(offset_minutes) * 60 } pub fn format_offset(offset_minutes: i32) -> String { @@ -228,15 +286,53 @@ pub fn format_offset(offset_minutes: i32) -> String { format!("{sign}{hours:02}{mins:02}") } -#[inline] -fn get_attr_value_unescaped(attr: &quick_xml::events::attributes::Attribute, decoder: quick_xml::Decoder) -> Option { - attr.decode_and_unescape_value(decoder).ok().map(|v| v.to_string()) +pub fn parse_xmltv_time(t: &str) -> Option { + DateTime::parse_from_str(t, "%Y%m%d%H%M%S %z") + .ok() + .map(|dt| dt.with_timezone(&Utc).timestamp()) } -pub fn format_xtream_time(ts: i64) -> String { - if let Some(dt) = DateTime::from_timestamp(ts, 0) { - dt.naive_utc().format("%Y-%m-%d %H:%M:%S").to_string() - } else { - String::new() +pub fn format_xmltv_time_utc(ts: i64) -> String { + let dt = Utc.timestamp_opt(ts, 0).unwrap(); + dt.format("%Y%m%d%H%M%S %z").to_string() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_parse_timeshift() { + assert_eq!(parse_timeshift(Some(&String::from("2"))), Some(120)); + assert_eq!(parse_timeshift(Some(&String::from("-1:30"))), Some(-90)); + assert_eq!(parse_timeshift(Some(&String::from("+0:15"))), Some(15)); + assert_eq!(parse_timeshift(Some(&String::from("1:45"))), Some(105)); + assert_eq!(parse_timeshift(Some(&String::from(":45"))), Some(45)); + assert_eq!(parse_timeshift(Some(&String::from("-:45"))), Some(-45)); + assert_eq!(parse_timeshift(Some(&String::from("0:30"))), Some(30)); + assert_eq!(parse_timeshift(Some(&String::from(":3"))), Some(3)); + assert_eq!(parse_timeshift(Some(&String::from("2:"))), Some(120)); + assert_eq!(parse_timeshift(Some(&String::from("+2:00"))), Some(120)); + assert_eq!(parse_timeshift(Some(&String::from("-0:10"))), Some(-10)); + assert_eq!(parse_timeshift(Some(&String::from("invalid"))), None); + assert_eq!(parse_timeshift(Some(&String::from("+abc"))), None); + assert_eq!(parse_timeshift(Some(&String::new())), None); + assert_eq!(parse_timeshift(None), None); + } + + #[test] + fn test_parse_timezone() { + // This will depend on current DST; we just check it’s within a valid range + let berlin = parse_timeshift(Some(&"Europe/Berlin".to_string())).unwrap(); + assert!(berlin == 60 || berlin == 120, "Berlin offset should be 60 or 120, got {berlin}"); + + let new_york = parse_timeshift(Some(&"America/New_York".to_string())).unwrap(); + assert!(new_york == -300 || new_york == -240, "New York offset should be -300 or -240, got {new_york}"); + + let tokyo = parse_timeshift(Some(&"Asia/Tokyo".to_string())).unwrap(); + assert_eq!(tokyo, 540); // always UTC+9 + + let utc = parse_timeshift(Some(&"UTC".to_string())).unwrap(); + assert_eq!(utc, 0); } } diff --git a/backend/src/utils/file/file_utils.rs b/backend/src/utils/file/file_utils.rs index d8c1c0a75..9284e226e 100644 --- a/backend/src/utils/file/file_utils.rs +++ b/backend/src/utils/file/file_utils.rs @@ -140,6 +140,9 @@ pub fn resolve_directory_path(input: &str) -> String { ) } +pub async fn file_exists_async(path: &Path) -> bool { + tokio::fs::try_exists(path).await.unwrap_or(false) +} #[inline] pub fn open_file(file_name: &Path) -> Result { diff --git a/backend/src/utils/network/xtream.rs b/backend/src/utils/network/xtream.rs index 999237d52..a91f4b004 100644 --- a/backend/src/utils/network/xtream.rs +++ b/backend/src/utils/network/xtream.rs @@ -10,7 +10,7 @@ use crate::repository::{ensure_input_storage_path, get_input_storage_path, get_t use crate::repository::VirtualIdRecord; use crate::repository::{get_live_cat_collection_path, get_series_cat_collection_path, get_vod_cat_collection_path, xtream_get_file_path, CategoryEntry}; use crate::repository::{persist_input_vod_info, persists_input_series_info, write_playlist_batch_item_upsert, write_playlist_item_update}; -use crate::utils::request; +use crate::utils::{file_exists_async, request}; use chrono::{DateTime, Utc}; use log::{error, info, warn}; use shared::error::{TuliproxError}; @@ -619,10 +619,10 @@ async fn process_xtream_cluster_to_disk( // Cleanup - temporary files are usually replaced/moved by swap, but defensive removal of leftovers // We strictly check for existence first to avoid errors if rename_or_copy acted as a move. - if tokio::fs::try_exists(&tmp_xtream_path).await.unwrap_or(false) { + if file_exists_async(&tmp_xtream_path).await { let _ = tokio::fs::remove_file(tmp_xtream_path).await; } - if tokio::fs::try_exists(&tmp_col_path).await.unwrap_or(false) { + if file_exists_async(&tmp_col_path).await { let _ = tokio::fs::remove_file(tmp_col_path).await; } diff --git a/frontend/Cargo.toml b/frontend/Cargo.toml index e239d7f17..bf8307f9f 100644 --- a/frontend/Cargo.toml +++ b/frontend/Cargo.toml @@ -1,11 +1,11 @@ [package] name = "frontend" -version = "3.2.48" +version = "3.2.49" edition = "2021" rust-version = "1.89.0" [dependencies] -shared = { version = "3.2.48", path = "../shared" } +shared = { version = "3.2.49", path = "../shared" } chrono = "0" yew = "0.21" yew-router = "0.18" diff --git a/frontend/scss/_theme.scss b/frontend/scss/_theme.scss index 0cb7a8d37..ac4727c4c 100644 --- a/frontend/scss/_theme.scss +++ b/frontend/scss/_theme.scss @@ -91,6 +91,7 @@ --card-color: var(--modest-text-color); --sub-card-background-color: rgba(18, 22, 33, 0.8); + --text-button-background-color: rgba(50, 80, 140, 0.3); --text-button-color: #ffffff; --text-button-border-color: rgba(80, 120, 200, 0.3); @@ -203,9 +204,12 @@ --hint-color: #888888; --label-color: #74c6ff; + --playlist-color: #aeb0b2; + --playlist-rating-color: #ffffff; + --playlist-rating-background-color: #121621; + --sidebar-toggle-color: #ffffff; --sidebar-toggle-hover-color: #47495d; - --playlist-color: #aeb0b2; --hover-border-color: rgba(100, 140, 255, 0.4); --spinner-color: rgba(255, 255, 255, 0.4); --spinner-highlight-color: #5599f0; diff --git a/frontend/scss/app/components/playlist/_playlist_explorer.scss b/frontend/scss/app/components/playlist/_playlist_explorer.scss index ee49c4630..d7ecd9c53 100644 --- a/frontend/scss/app/components/playlist/_playlist_explorer.scss +++ b/frontend/scss/app/components/playlist/_playlist_explorer.scss @@ -223,8 +223,8 @@ $series-backdrop-height: 300px; align-items: center; &__rating { - color: var(--card-color) !important; - background-color: rgb(from var(--card-background-color) r g b / 1) !important; + color: var(--playlist-rating-color) !important; + background-color: var(--playlist-rating-background-color) !important; font-weight: bold; } diff --git a/frontend/src/app/components/playlist/epg_view.rs b/frontend/src/app/components/playlist/epg_view.rs index 3ece0812a..3467c0983 100644 --- a/frontend/src/app/components/playlist/epg_view.rs +++ b/frontend/src/app/components/playlist/epg_view.rs @@ -186,11 +186,11 @@ pub fn EpgView() -> Html {

{ for tv.channels.iter().enumerate().skip(start_index).take(end_index - start_index).map(|(_i, ch)| { html! { -
{ if let Some(icon) = &ch.icon { - html! { {ch.title.clone()} } + html! { {ch.title.as_ref().map(ToString::to_string).unwrap_or_default()} } } else { html!{} } }
@@ -243,7 +243,7 @@ pub fn EpgView() -> Html { let program_style = format!("left:{left}px; width:{width}px; min-width:{width}px; max-width:{width}px"); html! { -
+
{ &pstart } {"-"} { &pend }
{ &p.title }
diff --git a/frontend/src/app/components/playlist/playlist_explorer.rs b/frontend/src/app/components/playlist/playlist_explorer.rs index a1abaae72..94418d6ab 100644 --- a/frontend/src/app/components/playlist/playlist_explorer.rs +++ b/frontend/src/app/components/playlist/playlist_explorer.rs @@ -404,7 +404,7 @@ pub fn PlaylistExplorer() -> Html { - {render_channel_logo(&*chan.logo)} + {render_channel_logo(&chan.logo)} {chan.title.clone()} } @@ -415,7 +415,7 @@ pub fn PlaylistExplorer() -> Html { let chan_clone = Rc::clone(chan); html! { - {render_channel_logo(&*chan.logo)} + {render_channel_logo(&chan.logo)} { html_if!(chan.rating > 0.001, { @@ -441,7 +441,7 @@ pub fn PlaylistExplorer() -> Html { }; html! { - {render_channel_logo(&*chan.logo)} + {render_channel_logo(&chan.logo)} { html_if!(chan.rating > 0.001, { @@ -467,7 +467,7 @@ pub fn PlaylistExplorer() -> Html { let rating = chan.rating.unwrap_or_default(); html! { - {render_channel_logo(&*chan.movie_image)} + {render_channel_logo(&chan.movie_image)} { html_if!(rating > 0.001, { diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 460e67851..6e170b6b6 100644 --- a/shared/Cargo.toml +++ b/shared/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "shared" -version = "3.2.48" +version = "3.2.49" edition = "2021" rust-version = "1.89.0" diff --git a/shared/src/model/epg.rs b/shared/src/model/epg.rs index 14e439fca..e95ab9e90 100644 --- a/shared/src/model/epg.rs +++ b/shared/src/model/epg.rs @@ -1,17 +1,7 @@ -use chrono::{DateTime, Utc}; +use std::sync::Arc; +use chrono::Utc; use serde::{Deserialize, Serialize}; -pub fn parse_xmltv_time(t: &str) -> Option { - let fmt = "%Y%m%d%H%M%S %z"; - if let Ok(result) = DateTime::parse_from_str(t, fmt) { - Some(result - .with_timezone(&Utc) - .timestamp()) - } else { - None - } -} - #[derive(Debug, Serialize, Deserialize, Clone)] pub struct EpgTv { pub start: i64, @@ -29,38 +19,57 @@ impl PartialEq for EpgTv { #[derive(Debug, Serialize, Deserialize, Clone)] pub struct EpgChannel { - pub id: String, - pub title: String, - pub icon: Option, + pub id: Arc, + pub title: Option>, + pub icon: Option>, pub programmes: Vec, } impl EpgChannel { - pub fn new(id: String) -> Self { + pub fn new(id: Arc) -> Self { Self { id, - title: String::new(), + title: None, icon: None, programmes: Vec::new(), } } + + pub fn get_programme_with_limit(&self, limit: u32) -> Vec<&EpgProgramme> { + let now = Utc::now().timestamp(); + + // find index for first relevant entry + let start_idx = self.programmes.iter() + .position(|p| (p.start <= now && now <= p.stop) || (p.start >= now)) + .unwrap_or(self.programmes.len()); // nothing found, empty response + + // slice from start_idx, max. limit + self.programmes + .iter() + .skip(start_idx) + .take(limit as usize) + .collect() + } } #[derive(Debug, Serialize, Deserialize, Clone)] pub struct EpgProgramme { pub start: i64, pub stop: i64, - pub channel: String, - pub title: String, + pub title: Option>, + pub desc: Option>, + #[serde(skip)] + pub channel: Arc, } impl EpgProgramme { - pub fn new(start: i64, stop: i64, channel: String) -> Self { + pub fn new(start: i64, stop: i64, channel: Arc) -> Self { Self { start, stop, channel, - title: String::new(), + title: None, + desc: None } } } diff --git a/shared/src/model/info_doc_utils.rs b/shared/src/model/info_doc_utils.rs index caa00ba24..0757b62ac 100644 --- a/shared/src/model/info_doc_utils.rs +++ b/shared/src/model/info_doc_utils.rs @@ -35,7 +35,7 @@ impl InfoDocUtils { "0".to_string() } else { let s = format!("{:.2}", n); - s.strip_suffix(".00").unwrap_or(&s).to_string() + s.trim_end_matches('0').trim_end_matches('.').to_string() } } diff --git a/shared/src/model/playlist_info_document.rs b/shared/src/model/playlist_info_document.rs index 29af94dd3..49ad6864f 100644 --- a/shared/src/model/playlist_info_document.rs +++ b/shared/src/model/playlist_info_document.rs @@ -215,9 +215,9 @@ pub struct XtreamSeriesEpisodeInfoData { pub tmdb_id: u32, #[serde(with = "arc_str_serde")] pub movie_image: Arc, + pub duration_secs: u32, #[serde(with = "arc_str_serde")] pub duration: Arc, - pub duration_secs: u32, pub video: Value, pub audio: Value, pub bitrate: u32, diff --git a/shared/src/model/short_epg.rs b/shared/src/model/short_epg.rs index 96ad2859d..70faee3db 100644 --- a/shared/src/model/short_epg.rs +++ b/shared/src/model/short_epg.rs @@ -1,9 +1,11 @@ -use crate::utils::{serialize_as_base64_padded, serialize_option_string_as_null_if_empty}; - +use std::sync::Arc; +use crate::utils::{serialize_as_base64_padded, serialize_option_string_as_null_if_empty, arc_str_serde}; #[derive(Debug, serde::Serialize, serde::Deserialize, Default)] pub struct ShortEpgDto { - pub id: String, - pub epg_id: String, + #[serde(with = "arc_str_serde")] + pub id: Arc, + #[serde(with = "arc_str_serde")] + pub epg_id: Arc, #[serde(serialize_with = "serialize_as_base64_padded")] pub title: String, pub lang: String, @@ -11,15 +13,17 @@ pub struct ShortEpgDto { pub end: String, // Format "2026-01-14 00:45:00" #[serde(serialize_with = "serialize_as_base64_padded")] pub description: String, - pub channel_id : String, + #[serde(with = "arc_str_serde")] + pub channel_id: Arc, pub start_timestamp: String, // Format "1768431000" pub stop_timestamp: String, // Format "1768434300" - pub stream_id: String, + #[serde(with = "arc_str_serde")] + pub stream_id: Arc, } #[derive(Debug, serde::Serialize, serde::Deserialize, Default)] pub struct ShortEpgResultDto { - pub epg_listings : Vec, + pub epg_listings: Vec, #[serde(default, serialize_with = "serialize_option_string_as_null_if_empty")] pub error: Option, } diff --git a/shared/src/model/ui_playlist_item.rs b/shared/src/model/ui_playlist_item.rs index 8c8f4e824..18398c530 100644 --- a/shared/src/model/ui_playlist_item.rs +++ b/shared/src/model/ui_playlist_item.rs @@ -39,51 +39,30 @@ fn pick_logo(logo: &Arc, logo_small: &Arc, props: Option<&StreamProper return Arc::clone(logo_small); } - if let Some(props) = props { - match props { - StreamProperties::Video(v) => { - if !v.stream_icon.is_empty() { - return Arc::clone(&v.stream_icon); - } - if let Some(details) = v.details.as_ref() { - if let Some(movie_image) = details.movie_image.as_ref() { - if !movie_image.is_empty() { - return Arc::clone(movie_image); - } - } - if let Some(cover_big) = details.cover_big.as_ref() { - if !cover_big.is_empty() { - return Arc::clone(cover_big); - } - } - if let Some(backdrop) = details.backdrop_path.as_ref() { - if let Some(first) = backdrop.first() { - if !first.is_empty() { - return Arc::clone(first); - } - } - } - } - } - StreamProperties::Series(s) => { - if !s.cover.is_empty() { - return Arc::clone(&s.cover); - } - if let Some(backdrop) = s.backdrop_path.as_ref() { - if let Some(first) = backdrop.first() { - if !first.is_empty() { - return Arc::clone(first); - } - } - } - } - StreamProperties::Live(_) => {} - StreamProperties::Episode(_) => {} + props.and_then(|p| match p { + StreamProperties::Video(v) => { + non_empty(&v.stream_icon) + .or_else(|| v.details.as_ref().and_then(|d| { + non_empty_opt(d.movie_image.as_ref()) + .or_else(|| non_empty_opt(d.cover_big.as_ref())) + .or_else(|| d.backdrop_path.as_ref().and_then(|b| non_empty(b.first()?))) + })) } - } - "".intern() + StreamProperties::Series(s) => { + non_empty(&s.cover) + .or_else(|| s.backdrop_path.as_ref().and_then(|b| non_empty(b.first()?))) + } + _ => None, + }).unwrap_or_else(|| "".intern()) } +fn non_empty(s: &Arc) -> Option> { + (!s.is_empty()).then(|| Arc::clone(s)) +} + +fn non_empty_opt(s: Option<&Arc>) -> Option> { + s.and_then(non_empty) +} /// Helper to get rating #[inline]