From 980410fa5bf6e1fd5270b07c5abd586b9c61ea05 Mon Sep 17 00:00:00 2001 From: euzu Date: Sat, 2 Nov 2024 16:44:38 +0100 Subject: [PATCH] epg processing on disk. The raw epg file is written directly to disc and processed with sax parser. --- src/model/xmltv.rs | 51 +++------ src/processing/playlist_processor.rs | 15 +-- src/processing/playlist_watch.rs | 7 +- src/processing/xmltv_parser.rs | 151 ++++++++++++++++++++++----- src/repository/storage.rs | 11 +- src/repository/xtream_repository.rs | 2 +- src/utils/compressed_file_reader.rs | 65 ++++++++++++ src/utils/compression_utils.rs | 12 +++ src/utils/download.rs | 18 ++-- src/utils/file_utils.rs | 25 +++-- src/utils/mod.rs | 4 +- src/utils/request_utils.rs | 138 ++++++++++++++++++++++-- 12 files changed, 393 insertions(+), 106 deletions(-) create mode 100644 src/utils/compressed_file_reader.rs create mode 100644 src/utils/compression_utils.rs diff --git a/src/model/xmltv.rs b/src/model/xmltv.rs index 8c017599c..e3da54b96 100644 --- a/src/model/xmltv.rs +++ b/src/model/xmltv.rs @@ -1,7 +1,15 @@ -use std::collections::{HashMap, HashSet}; +use std::collections::{HashMap}; +use std::path::PathBuf; use std::rc::Rc; -use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event}; + use quick_xml::{Error, Writer}; +use quick_xml::events::{BytesEnd, BytesStart, BytesText, Event}; + +pub(crate) const EPG_TAG_TV: &str = "tv"; +pub(crate) const EPG_TAG_PROGRAMME: &str = "programme"; +pub(crate) const EPG_TAG_CHANNEL: &str = "channel"; +pub(crate) const EPG_ATTRIB_ID: &str = "id"; +pub(crate) const EPG_ATTRIB_CHANNEL: &str = "channel"; // https://github.com/XMLTV/xmltv/blob/master/xmltv.dtd @@ -43,7 +51,7 @@ impl XmlTag { #[derive(Debug, Clone)] pub(crate) struct Epg { pub attributes: Option>>, - pub children: Vec>, + pub children: Vec, } impl Epg { @@ -62,40 +70,5 @@ impl Epg { #[derive(Debug, Clone)] pub(crate) struct TVGuide { - pub epg: XmlTag, + pub file: PathBuf, } - -impl TVGuide { - pub(crate) fn filter(&self, channel_ids: &HashSet>) -> Option { - if !channel_ids.is_empty() { - if let Some(epg_children) = &self.epg.children { - let children: Vec> = epg_children.iter().filter(|c| { - match c.name.as_str() { - "channel" => { - match c.get_attribute_value("id") { - None => false, - Some(val) => channel_ids.contains(val) - } - } - "programme" => { - match c.get_attribute_value("channel") { - None => false, - Some(val) => channel_ids.contains(val) - } - } - _ => false, - } - }).cloned().collect(); - - if !children.is_empty() { - return Some(Epg { - attributes: self.epg.attributes.clone(), - children, - }); - } - } - } - None - } -} - diff --git a/src/processing/playlist_processor.rs b/src/processing/playlist_processor.rs index 1442c2c03..db2e72dce 100644 --- a/src/processing/playlist_processor.rs +++ b/src/processing/playlist_processor.rs @@ -310,7 +310,8 @@ async fn process_source(cfg: Arc, source_idx: usize, user_targets: Arc

name_val.as_str(), None => input.url.as_str(), @@ -354,7 +355,7 @@ async fn process_source(cfg: Arc, source_idx: usize, user_targets: Arc

InputStats { @@ -393,8 +394,8 @@ async fn process_sources(config: Arc, user_targets: Arc) let (mut res_stats, mut res_errors) = System::new().block_on(async { process_source(cfg, index, usr_trgts).await }); - shared_errors.lock().unwrap().extend(res_errors.drain(..)); - shared_stats.lock().unwrap().extend(res_stats.drain(..)); + shared_errors.lock().unwrap().append(&mut res_errors); + shared_stats.lock().unwrap().append(&mut res_stats); }; handles.push(thread::spawn(process)); if handles.len() >= thread_num as usize { @@ -402,8 +403,8 @@ async fn process_sources(config: Arc, user_targets: Arc) } } else { let (mut res_stats, mut res_errors) = process_source(cfg, index, usr_trgts).await; - shared_errors.lock().unwrap().extend(res_errors.drain(..)); - shared_stats.lock().unwrap().extend(res_stats.drain(..)); + shared_errors.lock().unwrap().append(&mut res_errors); + shared_stats.lock().unwrap().append(&mut res_stats); } } for handle in handle_list { @@ -492,6 +493,8 @@ async fn process_playlist<'a>(playlists: &mut [FetchedPlaylist<'a>], let mut new_playlist = vec![]; let mut new_epg = vec![]; + // each fetched playlist can have its own epgl url. + // we need to process each input epg. new_fetched_playlists.drain(..).for_each(|mut fp| { // collect all epg_channel ids let epg_channel_ids: HashSet<_> = fp.playlistgroups.iter().flat_map(|g| &g.channels) diff --git a/src/processing/playlist_watch.rs b/src/processing/playlist_watch.rs index 99e3c9afd..98081390f 100644 --- a/src/processing/playlist_watch.rs +++ b/src/processing/playlist_watch.rs @@ -1,11 +1,11 @@ use std::collections::BTreeSet; use std::path::{Path}; use log::{error, info}; -use regex::Regex; use crate::messaging::{MsgKind, send_message}; use crate::model::config::Config; use crate::model::playlist::PlaylistGroup; use crate::utils::file_utils; +use crate::utils::file_utils::sanitize_filename; pub(crate) fn process_group_watch(cfg: &Config, target_name: &str, pl: &PlaylistGroup) { let mut new_tree = BTreeSet::new(); @@ -15,9 +15,8 @@ pub(crate) fn process_group_watch(cfg: &Config, target_name: &str, pl: &Playlist new_tree.insert(title); }); - let filename_re = Regex::new(r"[^A-Za-z0-9_-]").unwrap(); - let file_name = format!("watch_{}_{}", target_name, &pl.title); - let watch_filename = format!("{}.bin", filename_re.replace_all(&file_name, "_")).to_string(); + let file_name = format!("watch_{target_name}_{}", &pl.title); + let watch_filename = sanitize_filename(&format!("{file_name}.bin")); match file_utils::get_file_path(&cfg.working_dir, Some(std::path::PathBuf::from(&watch_filename))) { Some(path) => { let save_path = path.clone(); diff --git a/src/processing/xmltv_parser.rs b/src/processing/xmltv_parser.rs index d4fcfe10b..15825ba31 100644 --- a/src/processing/xmltv_parser.rs +++ b/src/processing/xmltv_parser.rs @@ -1,22 +1,74 @@ -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use std::rc::Rc; + use quick_xml::events::Event; use quick_xml::Reader; -use crate::model::xmltv::{Epg, TVGuide, XmlTag}; -static EPG_PROGRAMME: &str = "programme"; -static EPG_CHANNEL: &str = "channel"; -static EPG_ID: &str = "id"; +use crate::model::xmltv::{Epg, EPG_ATTRIB_CHANNEL, EPG_ATTRIB_ID, EPG_TAG_TV, EPG_TAG_CHANNEL, EPG_TAG_PROGRAMME, TVGuide, XmlTag}; +use crate::utils::compressed_file_reader::CompressedFileReader; -pub(crate) fn parse_tvguide(content: &str) -> Option { +impl TVGuide { + pub(crate) fn filter(&self, channel_ids: &HashSet>) -> Option { + if channel_ids.is_empty() { + return None; + } + match CompressedFileReader::new(&self.file) { + Ok(mut reader) => { + let mut children: Vec = vec![]; + let mut tv_attributes: Option>> = None; + let mut filter_tags = |tag: XmlTag| { + if match tag.name.as_str() { + EPG_TAG_CHANNEL => { + match tag.get_attribute_value(EPG_ATTRIB_ID) { + None => false, + Some(val) => channel_ids.contains(val) + } + } + EPG_TAG_PROGRAMME => { + match tag.get_attribute_value(EPG_ATTRIB_CHANNEL) { + None => false, + Some(val) => channel_ids.contains(val) + } + }, + EPG_TAG_TV => { + tv_attributes.clone_from(&tag.attributes); + false + }, + _ => false, + } { + children.push(tag); + } + }; + + parse_tvguide(&mut reader, &mut filter_tags); + + if children.is_empty() { + return None; + } + Some(Epg { + attributes: tv_attributes, + children, + }) + } + Err(_) => None + } + } +} + +pub(crate) fn parse_tvguide(content: R, callback: &mut F) +where + R: std::io::BufRead, + F: FnMut(XmlTag), +{ let mut stack: Vec = vec![]; - let mut reader = Reader::from_str(content); + let mut reader = Reader::from_reader(content); let mut buf = Vec::::new(); loop { match reader.read_event_into(&mut buf) { Ok(Event::Eof) => break, Ok(Event::Start(e)) => { let name = String::from_utf8_lossy(e.name().as_ref()).to_string(); + let is_tv_tag = name == EPG_TAG_TV; let attributes = e.attributes().filter_map(Result::ok) .filter_map(|a| { let key = String::from_utf8_lossy(a.key.as_ref()).to_string(); @@ -34,22 +86,40 @@ pub(crate) fn parse_tvguide(content: &str) -> Option { children: None, }; - stack.push(tag); + if is_tv_tag { + callback(tag); + } else { + stack.push(tag); + } } Ok(Event::End(_e)) => { - if stack.len() > 1 { + if !stack.is_empty() { if let Some(tag) = stack.pop() { - if let Some(old_tag) = stack.pop().map(|mut r| { - let rc_tag = Rc::new(tag); - r.children = Some( - r.children.map_or(vec![rc_tag.clone()], - |mut c| { - c.push(rc_tag.clone()); - c - })); - r - }) { - stack.push(old_tag); + if tag.name == EPG_TAG_CHANNEL { + if let Some(chan_id) = tag.get_attribute_value(EPG_ATTRIB_ID) { + if !chan_id.is_empty() { + callback(tag); + } + } + } else if tag.name == EPG_TAG_PROGRAMME { + if let Some(chan_id) = tag.get_attribute_value(EPG_ATTRIB_CHANNEL) { + if !chan_id.is_empty() { + callback(tag); + } + } + } else if !stack.is_empty() { + if let Some(old_tag) = stack.pop().map(|mut r| { + let rc_tag = Rc::new(tag); + r.children = Some( + r.children.map_or(vec![rc_tag.clone()], + |mut c| { + c.push(rc_tag.clone()); + c + })); + r + }) { + stack.push(old_tag); + } } } } @@ -63,9 +133,6 @@ pub(crate) fn parse_tvguide(content: &str) -> Option { _ => {} } } - stack.pop().map(|epg| TVGuide { - epg, - }) } pub(crate) fn flatten_tvguide(tv_guides: &[Epg]) -> Option { @@ -82,8 +149,8 @@ pub(crate) fn flatten_tvguide(tv_guides: &[Epg]) -> Option { epg.attributes.clone_from(&guide.attributes); } guide.children.iter().for_each(|c| { - if c.name.as_str() == EPG_CHANNEL { - if let Some(chan_id) = c.get_attribute_value(EPG_ID) { + if c.name.as_str() == EPG_TAG_CHANNEL { + if let Some(chan_id) = c.get_attribute_value(EPG_ATTRIB_ID) { if !channel_ids.contains(&chan_id) { channel_ids.push(chan_id); epg.children.push(c.clone()); @@ -92,8 +159,8 @@ pub(crate) fn flatten_tvguide(tv_guides: &[Epg]) -> Option { } }); guide.children.iter().for_each(|c| { - if c.name.as_str() == EPG_PROGRAMME { - if let Some(chan_id) = c.get_attribute_value(EPG_CHANNEL) { + if c.name.as_str() == EPG_TAG_PROGRAMME { + if let Some(chan_id) = c.get_attribute_value(EPG_TAG_CHANNEL) { if channel_ids.contains(&chan_id) { epg.children.push(c.clone()); } @@ -103,4 +170,34 @@ pub(crate) fn flatten_tvguide(tv_guides: &[Epg]) -> Option { } Some(epg) } +} + + +#[cfg(test)] +mod tests { + use std::collections::HashSet; + use std::io; + use std::path::PathBuf; + use std::rc::Rc; + + use crate::model::xmltv::{TVGuide}; + + #[test] + fn parse_test() -> io::Result<()> { + let file_path = "/tmp/epg.xml.gz"; + + let tv_guide = TVGuide { file: PathBuf::from(file_path) }; + + let channel_ids = vec!["channel.1", "channel.2", "channel.3"]; + let channel_ids : HashSet> = channel_ids.into_iter().map(|s| Rc::new(s.to_string())).collect(); + + match tv_guide.filter(&channel_ids) { + None => assert!(false, "No epg filtered"), + Some(epg) => { + assert_eq!(epg.children.len(), channel_ids.len() * 2, "Epg size does not match") + } + } + + Ok(()) + } } \ No newline at end of file diff --git a/src/repository/storage.rs b/src/repository/storage.rs index d7aae3e83..723f0e94b 100644 --- a/src/repository/storage.rs +++ b/src/repository/storage.rs @@ -1,6 +1,7 @@ use std::path::{Path, PathBuf}; + use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; -use crate::model::config::Config; +use crate::model::config::{Config, ConfigInput}; use crate::utils::file_utils; pub(in crate::repository) const FILE_SUFFIX_DB: &str = "db"; @@ -33,3 +34,11 @@ pub(crate) fn ensure_target_storage_path(cfg: &Config, target_name: &str) -> Res pub(crate) fn get_target_storage_path(cfg: &Config, target_name: &str) -> Option { file_utils::get_file_path(&cfg.working_dir, Some(std::path::PathBuf::from(target_name.replace(' ', "_")))) } + +pub(crate) fn get_input_storage_path(input: &ConfigInput, working_dir: &str) -> std::io::Result { + let name = format!("input_{}", input.name.clone().unwrap_or_else(|| format!("{}", input.id))); + let path = Path::new(working_dir).join(name); + + // Create the directory and return the path or propagate the error + std::fs::create_dir_all(&path).map(|()| path) +} diff --git a/src/repository/xtream_repository.rs b/src/repository/xtream_repository.rs index 182bc5b02..956c4c2aa 100644 --- a/src/repository/xtream_repository.rs +++ b/src/repository/xtream_repository.rs @@ -23,7 +23,7 @@ pub(crate) static COL_CAT_SERIES: &str = "cat_series"; pub(crate) static COL_CAT_VOD: &str = "cat_vod"; const FILE_SERIES_EPISODES: &str = "series_episodes"; const FILE_SERIES: &str = "series"; -const FILE_EPG: &str = "epg.xml"; +pub(crate) const FILE_EPG: &str = "epg.xml"; const PATH_XTREAM: &str = "xtream"; const TAG_CATEGORY_ID: &str = "category_id"; const TAG_CATEGORY_NAME: &str = "category_name"; diff --git a/src/utils/compressed_file_reader.rs b/src/utils/compressed_file_reader.rs new file mode 100644 index 000000000..77659b410 --- /dev/null +++ b/src/utils/compressed_file_reader.rs @@ -0,0 +1,65 @@ +use std::fs::OpenOptions; +use std::io::{BufRead, BufReader, Read, Seek, SeekFrom}; +use std::path::Path; +use flate2::bufread::{GzDecoder, ZlibDecoder}; +use crate::utils::compression_utils::{is_deflate, is_gzip}; + + +pub(crate) struct CompressedFileReader { + reader: BufReader>, +} + +impl CompressedFileReader { + pub(crate) fn new(path: &Path) -> std::io::Result { + let file = OpenOptions::new() + .read(true) + .open(path)?; + + let mut buffered_file = BufReader::new(file); + let mut header = [0u8; 2]; + buffered_file.read_exact(&mut header)?; + buffered_file.seek(SeekFrom::Start(0))?; + + let reader: BufReader> = if is_gzip(&header) { + BufReader::new(Box::new(GzDecoder::new(buffered_file)) as Box) + } else if is_deflate(&header) { + BufReader::new(Box::new(ZlibDecoder::new(buffered_file)) as Box) + } else { + BufReader::new(Box::new(buffered_file) as Box) + }; + + Ok(Self { reader }) + } +} + +// Implement the Read trait for CompressedFileReader +impl Read for CompressedFileReader { + fn read(&mut self, buf: &mut [u8]) -> std::io::Result { + self.reader.read(buf) + } +} + +// Implement BufRead for CompressedFileReader +impl BufRead for CompressedFileReader { + fn fill_buf(&mut self) -> std::io::Result<&[u8]> { + self.reader.fill_buf() + } + + fn consume(&mut self, amt: usize) { + self.reader.consume(amt); + } +} + +impl Iterator for CompressedFileReader +{ + type Item = std::io::Result; + + fn next(&mut self) -> Option { + let mut line = String::new(); + match self.reader.read_line(&mut line) { + Ok(0) => None, // EOF + Ok(_) => Some(Ok(line.trim_end().to_string())), + Err(e) => Some(Err(e)), + } + } +} diff --git a/src/utils/compression_utils.rs b/src/utils/compression_utils.rs new file mode 100644 index 000000000..a01cd6200 --- /dev/null +++ b/src/utils/compression_utils.rs @@ -0,0 +1,12 @@ + +pub(crate) const ENCODING_GZIP: &str = "gzip"; +pub(crate) const ENCODING_DEFLATE: &str = "deflate"; + +pub(crate) fn is_gzip(bytes: &[u8]) -> bool { + // Gzip files start with the bytes 0x1F 0x8B + bytes.len() >= 2 && bytes[0] == 0x1F && bytes[1] == 0x8B +} + +pub(crate) fn is_deflate(bytes: &[u8]) -> bool { + bytes[0] == 0x78 && (bytes[1] == 0x01 || bytes[1] == 0x9C || bytes[1] == 0xDA) +} \ No newline at end of file diff --git a/src/utils/download.rs b/src/utils/download.rs index 7df8d46a7..26a4dc834 100644 --- a/src/utils/download.rs +++ b/src/utils/download.rs @@ -6,11 +6,11 @@ use crate::m3u_filter_error::M3uFilterError; use crate::model::config::{Config, ConfigInput}; use crate::model::playlist::{FetchedPlaylist, PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster}; use crate::model::xmltv::TVGuide; -use crate::processing::{m3u_parser, xmltv_parser, xtream_parser}; +use crate::processing::{m3u_parser, xtream_parser}; use crate::processing::xtream_parser::parse_xtream_series_info; use crate::utils::{file_utils, request_utils}; -fn prepare_file_path(persist: Option<&String>, working_dir: &String, action: &str) -> Option { +fn prepare_file_path(persist: Option<&String>, working_dir: &str, action: &str) -> Option { let persist_file: Option = persist.map(|persist_path| file_utils::prepare_persist_path(persist_path.as_str(), action)); if persist_file.is_some() { @@ -106,7 +106,7 @@ const ACTIONS: [(XtreamCluster, &str, &str); 3] = [ (XtreamCluster::Video, "get_vod_categories", "get_vod_streams"), (XtreamCluster::Series, "get_series_categories", "get_series")]; -pub(crate) async fn get_xtream_playlist(input: &ConfigInput, working_dir: &String) -> (Vec, Vec) { +pub(crate) async fn get_xtream_playlist(input: &ConfigInput, working_dir: &str) -> (Vec, Vec) { let mut playlist_groups: Vec = Vec::new(); let username = input.username.as_ref().map_or("", |v| v); let password = input.password.as_ref().map_or("", |v| v); @@ -154,15 +154,17 @@ pub(crate) async fn get_xtream_playlist(input: &ConfigInput, working_dir: &Strin (playlist_groups, errors) } -pub(crate) async fn get_xmltv(_cfg: &Config, input: &ConfigInput, working_dir: &String) -> (Option, Vec) { +pub(crate) async fn get_xmltv(_cfg: &Config, input: &ConfigInput, working_dir: &str) -> (Option, Vec) { match &input.epg_url { None => (None, vec![]), Some(url) => { debug!("Getting epg file path for url: {}", url); - let persist_file_path = prepare_file_path(input.persist.as_ref(), working_dir, "").map(|path| file_utils::add_prefix_to_filename(&path, "epg_", Some("xml"))); - match request_utils::get_input_text_content(input, working_dir, url, persist_file_path).await { - Ok(xml_content) => { - (xmltv_parser::parse_tvguide(xml_content.as_str()), vec![]) + let persist_file_path = prepare_file_path(input.persist.as_ref(), working_dir, "") + .map(|path| file_utils::add_prefix_to_filename(&path, "epg_", Some("xml"))); + + match request_utils::get_input_text_content_as_file(input, working_dir, url, persist_file_path).await { + Ok(file) => { + (Some(TVGuide { file }), vec![]) } Err(err) => (None, vec![err]) } diff --git a/src/utils/file_utils.rs b/src/utils/file_utils.rs index 55310dc31..ba8ba82ae 100644 --- a/src/utils/file_utils.rs +++ b/src/utils/file_utils.rs @@ -6,6 +6,13 @@ use std::path::{Path, PathBuf}; use log::{debug, error}; use path_absolutize::Absolutize; +const USER_FILE: &str = "user.txt"; +const CONFIG_PATH: &str = "config"; +const CONFIG_FILE: &str = "config.yml"; +const SOURCE_FILE: &str = "source.yml"; +const MAPPING_FILE: &str = "mapping.yml"; +const API_PROXY_FILE: &str = "api-proxy.yml"; + #[macro_export] macro_rules! exit { ($($arg:tt)*) => {{ @@ -48,13 +55,6 @@ pub(crate) fn get_default_file_path(config_path: &str, file: &str) -> String { }) } -static USER_FILE: &str = "user.txt"; -static CONFIG_PATH: &str = "config"; -static CONFIG_FILE: &str = "config.yml"; -static SOURCE_FILE: &str = "source.yml"; -static MAPPING_FILE: &str = "mapping.yml"; -static API_PROXY_FILE: &str = "api-proxy.yml"; - #[inline] pub(crate) fn get_default_user_file_path(config_path: &str) -> String { get_default_file_path(config_path, USER_FILE) @@ -143,7 +143,7 @@ pub(crate) fn prepare_persist_path(file_name: &str, date_prefix: &str) -> PathBu std::path::PathBuf::from(persist_filename) } -pub(crate) fn get_file_path(wd: &String, path: Option) -> Option { +pub(crate) fn get_file_path(wd: &str, path: Option) -> Option { match path { Some(p) => { if p.is_relative() { @@ -191,3 +191,12 @@ pub(crate) fn append_extension(path: &Path, ext: &str) -> PathBuf { let extension = path.extension().map(|ext| ext.to_str().unwrap_or("")); path.with_extension(format!("{}{ext}", &extension.unwrap_or_default())) } + + +#[inline] +pub(crate) fn sanitize_filename(file_name: &str) -> String { + file_name + .chars() + .map(|c| if c.is_alphanumeric() || c == '_' || c == '-' { c } else { '_' }) + .collect() +} \ No newline at end of file diff --git a/src/utils/mod.rs b/src/utils/mod.rs index be648a6f0..e584c2097 100644 --- a/src/utils/mod.rs +++ b/src/utils/mod.rs @@ -6,4 +6,6 @@ pub (crate) mod json_utils; pub (crate) mod config_reader; pub (crate) mod default_utils; pub (crate) mod multi_file_reader; -pub (crate) mod file_lock_manager; \ No newline at end of file +pub (crate) mod file_lock_manager; +pub(crate) mod compressed_file_reader; +mod compression_utils; \ No newline at end of file diff --git a/src/utils/request_utils.rs b/src/utils/request_utils.rs index ed3e9c25a..eb68604df 100644 --- a/src/utils/request_utils.rs +++ b/src/utils/request_utils.rs @@ -1,9 +1,11 @@ use std::collections::{HashMap, HashSet}; use std::fs; -use std::io::{Error, ErrorKind, Read}; -use std::path::PathBuf; +use std::fs::File; +use std::io::{BufWriter, Error, ErrorKind, Read, Write}; +use std::path::{Path, PathBuf}; use flate2::read::{GzDecoder, ZlibDecoder}; +use futures::StreamExt; use log::{debug, error, Level, log_enabled}; use reqwest::header::{HeaderMap, HeaderName, HeaderValue}; use reqwest::header::CONTENT_ENCODING; @@ -12,20 +14,66 @@ use url::Url; use crate::create_m3u_filter_error_result; use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind}; use crate::model::config::ConfigInput; +use crate::repository::storage::{get_input_storage_path}; +use crate::repository::xtream_repository::FILE_EPG; +use crate::utils::compression_utils::{ENCODING_GZIP, ENCODING_DEFLATE, is_gzip}; use crate::utils::file_utils::{get_file_path, persist_file}; -const ENCODING_GZIP: &str = "gzip"; -const ENCODING_DEFLATE: &str = "deflate"; - -fn is_gzip(bytes: &[u8]) -> bool { - // Gzip files start with the bytes 0x1F 0x8B - bytes.len() >= 2 && bytes[0] == 0x1F && bytes[1] == 0x8B -} - pub(crate) fn bytes_to_megabytes(bytes: u64) -> u64 { bytes / 1_048_576 } +pub(crate) async fn get_input_text_content_as_file(input: &ConfigInput, working_dir: &str, url_str: &str, persist_filepath: Option) -> Result { + if log_enabled!(Level::Debug) { + debug!("getting input text content working_dir: {}, url: {}", working_dir, url_str); + } + + if url_str.parse::().is_ok() { + match download_text_content_as_file(input, url_str, working_dir, persist_filepath).await { + Ok(content) => Ok(content), + Err(e) => { + error!("cant download input url: {} => {}", url_str, e); + create_m3u_filter_error_result!(M3uFilterErrorKind::Notify, "Failed to download") + } + } + } else { + let result = match get_file_path(working_dir, Some(PathBuf::from(url_str))) { + Some(filepath) => { + if filepath.exists() { + if let Some(persist_file_value) = persist_filepath { + let to_file = &persist_file_value; + match fs::copy(&filepath, to_file) { + Ok(_) => {} + Err(e) => { + error!("cant persist to: {} => {}", to_file.to_str().unwrap_or("?"), e); + return create_m3u_filter_error_result!(M3uFilterErrorKind::Notify, "Failed to persist: {} => {}", to_file.to_str().unwrap_or("?"), e); + } + } + }; + + if filepath.exists() { + Some(filepath) + } else { + return create_m3u_filter_error_result!(M3uFilterErrorKind::Notify, "Failed: file does not exists {filepath:?}"); + } + } else { + None + } + } + None => None + }; + + if let Some(path) = result { + Ok(path) + } else { + let msg = format!("cant read input url: {url_str:?}"); + error!("{}", msg); + create_m3u_filter_error_result!(M3uFilterErrorKind::Notify, "{}", msg) + } + } +} + + pub(crate) async fn get_input_text_content(input: &ConfigInput, working_dir: &String, url_str: &str, persist_filepath: Option) -> Result { if log_enabled!(Level::Debug) { debug!("getting input text content working_dir: {}, url: {}", working_dir, url_str); @@ -129,6 +177,38 @@ fn get_local_file_content(file_path: &PathBuf) -> Result { Err(Error::new(ErrorKind::InvalidData, format!("Cant find file {file_str}"))) } + +async fn get_remote_content_as_file(input: &ConfigInput, url: &Url, file_path: &Path) -> Result { + let request = get_client_request(Some(input), url, None); + match request.send().await { + Ok(response) => { + if response.status().is_success() { + // Open a file in write mode + let mut file = BufWriter::with_capacity(8192, File::create(file_path)?); + // Stream the response body in chunks + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream.next().await { + match chunk { + Ok(bytes) => { + file.write_all(&bytes)?; + } + Err(err) => { + return Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Failed to read chunk: {err}"))); + } + } + } + + file.flush()?; + debug!("File downloaded successfully to {file_path:?}"); + Ok(file_path.to_path_buf()) + } else { + Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Request failed with status {}", response.status()))) + } + } + Err(err) => Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Request failed: {err}"))), + } +} + async fn get_remote_content(input: &ConfigInput, url: &Url) -> Result { let request = get_client_request(Some(input), url, None); match request.send().await { @@ -158,7 +238,7 @@ async fn get_remote_content(input: &ConfigInput, url: &Url) -> Result {} Err(err) => return Err(std::io::Error::new(ErrorKind::Other, format!("failed to decode gzip content {err}"))) }; - } + }, ENCODING_DEFLATE => { let mut decoder = ZlibDecoder::new(&bytes[..]); match decoder.read_to_string(&mut decode_buffer) { @@ -189,6 +269,42 @@ async fn get_remote_content(input: &ConfigInput, url: &Url) -> Result) -> Result { + if let Ok(url) = url_str.parse::() { + + if url.scheme() == "file" { + if let Ok(file_path) = url.to_file_path() { + if file_path.exists() { + Ok(file_path) + } else { + Err(Error::new(ErrorKind::NotFound, format!("Unknown file {file_path:?}"))) + } + } else { + Err(Error::new(ErrorKind::Unsupported, format!("Unknown file {url}"))) + } + } else { + let file_path = match persist_filepath { + None => { + match get_input_storage_path(input, working_dir) { + Ok(download_path) => { + Ok(download_path.join(FILE_EPG)) + } + Err(err) => Err(err) + } + } + Some(path) => Ok(path), + }; + match file_path { + Ok(persist_path) => get_remote_content_as_file(input, &url, &persist_path).await, + Err(err) => Err(err) + } + } + } else { + Err(std::io::Error::new(ErrorKind::Unsupported, format!("Malformed URL {url_str}"))) + } +} + + pub(crate) async fn download_text_content(input: &ConfigInput, url_str: &str, persist_filepath: Option) -> Result { if let Ok(url) = url_str.parse::() { let result = if url.scheme() == "file" {