euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-20 23:02:34 +02:00
Code Issues Packages Projects Releases 5 Wiki Activity

Improved HTTP request retry mechanism with standardized backoff handling for better reliability

Browse Source 
UI Improvements
Enhanced playlist explorer styling
This commit is contained in:
euzu
2026-01-23 16:06:20 +01:00
parent c5239b5614
commit 3752125112
36 changed files with 990 additions and 834 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/parser/xmltv.rs
+52 -75
View File
@@ -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<HashMap<String, Arc<str>>>,
}
impl EpgInterner {
fn new() -> Self {
Self { map: RefCell::new(HashMap::new()) }
}
fn intern(&self, s: &str) -> Arc<str> {
let mut map = self.map.borrow_mut();
if let Some(arc) = map.get(s) {
return arc.clone();
}
let arc: Arc<str> = 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<str>, 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<String>) {
fn find_best_fuzzy_match(id_cache: &mut EpgIdCache, tag: &XmlTag) -> (bool, Option<Arc<str>>) {
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<str>, Arc<str>)> = 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<Epg> {
async fn process_epg_file(id_cache: &mut EpgIdCache, epg_source: &PersistedEpgSource) -> Option<Epg> {
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<Arc<XmlTag>> = vec![];
let mut tv_attributes: Option<HashMap<Arc<str>, String>> = None;
let mut tv_attributes: Option<HashMap<Arc<str>, Arc<str>>> = 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<Vec<Epg>> {
pub async fn filter(&self, id_cache: &mut EpgIdCache) -> Option<Vec<Epg>> {
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<F>(callback: &mut F, stack: &mut Vec<XmlTag>, e: &BytesStart, interner: &EpgInterner)
fn handle_tag_start<F>(callback: &mut F, stack: &mut Vec<XmlTag>, 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<F>(callback: &mut F, stack: &mut Vec<XmlTag>)
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<XmlTag> = vec![];
let mut xml_reader = quick_xml::reader::Reader::from_reader(async_file_reader(content));
let mut buf = Vec::<u8>::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<Arc<str>, String> {
fn collect_tag_attributes(e: &BytesStart, is_channel: bool, is_program: bool) -> HashMap<Arc<str>, Arc<str>> {
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::<HashMap<Arc<str>, String>>();
}).collect::<HashMap<Arc<str>, Arc<str>>>();
attributes
}
@@ -486,11 +461,13 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option<Epg> {
None
} else {
let epg_children: Mutex<Vec<Arc<XmlTag>>> = Mutex::new(Vec::new());
let epg_attributes: Option<HashMap<Arc<str>, String>> = tv_guides.first().and_then(|t| t.attributes.clone());
let epg_attributes: Option<HashMap<Arc<str>, Arc<str>>> = tv_guides.first().and_then(|t| t.attributes.clone());
let count = tv_guides.iter().map(|tvg| tvg.children.len()).sum();
let channel_mapping: DashMap<Arc<str>, 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<Epg> {
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<Epg> {
});
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<Epg> {
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.