euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-26 00:32:30 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

implementing multi input

Browse Source
This commit is contained in:
euzu
2024-05-05 19:15:54 +02:00
parent 8049449de3
commit 4d89773b45
18 changed files with 601 additions and 521 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/processing/m3u_parser.rs
+16 -13
View File
@@ -1,7 +1,7 @@
use std::borrow::{BorrowMut};
use std::cell::RefCell;
use std::rc::Rc;
use crate::model::config::Config;
use crate::model::config::{Config, ConfigInput};
use crate::model::config::default_as_empty_rc_str;
use crate::model::playlist::{default_playlist_item_type, default_stream_cluster, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, XtreamCluster};
use crate::utils::string_utils;
@@ -53,7 +53,7 @@ fn skip_digit(it: &mut std::str::Chars) -> Option<char> {
}
}
fn create_empty_playlistitem_header(content: &String, url: String) -> PlaylistItemHeader {
fn create_empty_playlistitem_header(input_id: u16, url: String) -> PlaylistItemHeader {
PlaylistItemHeader {
id: default_as_empty_rc_str(),
stream_id: default_as_empty_rc_str(),
@@ -66,13 +66,14 @@ fn create_empty_playlistitem_header(content: &String, url: String) -> PlaylistIt
audio_track: default_as_empty_rc_str(),
time_shift: default_as_empty_rc_str(),
rec: default_as_empty_rc_str(),
source: Rc::new(content.to_owned()),
source: Rc::new(input_id.to_string()),//Rc::new(content.to_owned()),
url: Rc::new(url),
epg_channel_id: None,
item_type: default_playlist_item_type(),
xtream_cluster: default_stream_cluster(),
additional_properties: None,
series_fetched: false,
category_id: 0,
}
}
@@ -87,8 +88,8 @@ macro_rules! process_header_fields {
};
}
fn process_header(video_suffixes: &Vec<&str>, content: &String, url: String) -> PlaylistItemHeader {
let mut plih = create_empty_playlistitem_header(content, url.clone());
fn process_header(input: &ConfigInput, video_suffixes: &Vec<&str>, content: &String, url: String) -> PlaylistItemHeader {
let mut plih = create_empty_playlistitem_header(input.id, url.clone());
let mut it = content.chars();
let line_token = token_till(&mut it, ':');
if line_token == Some(String::from("#EXTINF")) {
@@ -154,7 +155,7 @@ fn extract_id_from_url(url: &str) -> Option<String> {
None
}
pub(crate) fn consume_m3u<F: FnMut(PlaylistItem)>(cfg: &Config, lines: impl Iterator<Item=String>, mut visit: F) {
pub(crate) fn consume_m3u<F: FnMut(PlaylistItem)>(cfg: &Config, input: &ConfigInput, lines: impl Iterator<Item=String>, mut visit: F) {
let mut header: Option<String> = None;
let mut group: Option<String> = None;
@@ -172,15 +173,17 @@ pub(crate) fn consume_m3u<F: FnMut(PlaylistItem)>(cfg: &Config, lines: impl Iter
continue;
}
if let Some(header_value) = header {
let item = PlaylistItem { header: RefCell::new(process_header(&video_suffixes, &header_value,line)) };
if item.header.borrow().group.is_empty() {
let item = PlaylistItem { header: RefCell::new(process_header(&input, &video_suffixes, &header_value, line)) };
let mut header = item.header.borrow_mut();
if header.group.is_empty() {
if let Some(group_value) = group {
item.header.borrow_mut().group = Rc::new(group_value);
header.group = Rc::new(group_value);
} else {
let current_title = item.header.borrow().title.to_owned();
item.header.borrow_mut().group = Rc::new(string_utils::get_title_group(current_title.as_str()));
let current_title = header.title.to_owned();
header.group = Rc::new(string_utils::get_title_group(current_title.as_str()));
}
}
drop(header);
visit(item);
}
header = None;
@@ -188,11 +191,11 @@ pub(crate) fn consume_m3u<F: FnMut(PlaylistItem)>(cfg: &Config, lines: impl Iter
}
}
pub(crate) fn parse_m3u(cfg: &Config, lines: &[String]) -> Vec<PlaylistGroup> {
pub(crate) fn parse_m3u(cfg: &Config, input: &ConfigInput, lines: &[String]) -> Vec<PlaylistGroup> {
let mut groups: std::collections::HashMap<Rc<String>, Vec<PlaylistItem>> = std::collections::HashMap::new();
let mut sort_order: Vec<Rc<String>> = vec![];
let mut playlist = Vec::new();
consume_m3u(cfg, lines.iter().cloned(), |item| playlist.push(item));
consume_m3u(cfg, input, lines.iter().cloned(), |item| playlist.push(item));
playlist.drain(..).for_each(|item| {
let key = Rc::clone(&item.header.borrow().group);
// let key2 = String::from(&item.header.group);
src/processing/playlist_processor.rs
+38 -82
View File
@@ -1,5 +1,6 @@
extern crate unidecode;
use core::cmp::Ordering;
use std::cell::RefCell;
use std::collections::{HashMap, HashSet};
use std::rc::Rc;
@@ -14,20 +15,17 @@ use crate::{Config, get_errors_notify_message, model::config};
use crate::filter::{get_field_value, MockValueProcessor, set_field_value, ValueProvider};
use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind};
use crate::messaging::{MsgKind, send_message};
use crate::model::config::{ConfigTarget, default_as_default, InputType, ProcessTargets};
use crate::model::config::{ItemField, ProcessingOrder, SortOrder::{Asc, Desc}, TargetType};
use crate::model::config::{ConfigSortChannel, ConfigSortGroup, ConfigTarget, default_as_default, InputType,
ItemField, ProcessingOrder, ProcessTargets, SortOrder::{Asc, Desc}};
use crate::model::mapping::{Mapping, MappingValueProcessor};
use crate::model::playlist::{FetchedPlaylist, PlaylistGroup, PlaylistItem};
use crate::model::stats::{InputStats, PlaylistStats};
use crate::model::xtream::MultiXtreamMapping;
use crate::processing::affix_processor::apply_affixes;
use crate::processing::playlist_watch::process_group_watch;
use crate::processing::xmltv_parser::flatten_tvguide;
use crate::processing::xtream_processor::playlist_resolve_series;
use crate::repository::playlist_repository::persist_playlist;
use crate::repository::xtream_repository::write_xtream_mapping;
use crate::utils::download;
use crate::processing::affix_processor::apply_affixes;
fn is_valid(pli: &PlaylistItem, target: &ConfigTarget) -> bool {
let provider = ValueProvider { pli: RefCell::new(pli) };
@@ -53,37 +51,41 @@ fn filter_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Opt
Some(new_playlist)
}
fn playlistgroup_comparator(a: &PlaylistGroup, b: &PlaylistGroup, group_sort: &ConfigSortGroup, match_as_ascii: bool) -> Ordering {
let value_a = if match_as_ascii { Rc::new(unidecode(&a.title)) } else { Rc::clone(&a.title) };
let value_b = if match_as_ascii { Rc::new(unidecode(&b.title)) } else { Rc::clone(&b.title) };
let ordering = value_a.partial_cmp(&value_b).unwrap();
match group_sort.order {
Asc => ordering,
Desc => ordering.reverse()
}
}
fn playlistitem_comparator(a: &PlaylistItem, b: &PlaylistItem, channel_sort: &ConfigSortChannel, match_as_ascii: bool) -> Ordering {
let raw_value_a = get_field_value(a, &channel_sort.field);
let raw_value_b = get_field_value(b, &channel_sort.field);
let value_a = if match_as_ascii { Rc::new(unidecode(&raw_value_a)) } else { raw_value_a };
let value_b = if match_as_ascii { Rc::new(unidecode(&raw_value_b)) } else { raw_value_b };
let ordering = value_a.partial_cmp(&value_b).unwrap();
match channel_sort.order {
Asc => ordering,
Desc => ordering.reverse()
}
}
fn sort_playlist(target: &ConfigTarget, new_playlist: &mut [PlaylistGroup]) {
if let Some(sort) = &target.sort {
let match_as_ascii = &sort.match_as_ascii;
let match_as_ascii = sort.match_as_ascii;
if let Some(group_sort) = &sort.groups {
new_playlist.sort_by(|a, b| {
let value_a = if *match_as_ascii { Rc::new(unidecode(&a.title)) } else { Rc::clone(&a.title) };
let value_b = if *match_as_ascii { Rc::new(unidecode(&b.title)) } else { Rc::clone(&b.title) };
let ordering = value_a.partial_cmp(&value_b).unwrap();
match group_sort.order {
Asc => ordering,
Desc => ordering.reverse()
}
});
new_playlist.sort_by(|a, b| playlistgroup_comparator(a, b, group_sort, match_as_ascii));
}
if let Some(channel_sorts) = &sort.channels {
channel_sorts.iter().for_each(|channel_sort| {
let regexp = channel_sort.re.as_ref().unwrap();
new_playlist.iter_mut().for_each(|group| {
let group_title = if *match_as_ascii { Rc::new(unidecode(&group.title)) } else { Rc::clone(&group.title) };
let group_title = if match_as_ascii { Rc::new(unidecode(&group.title)) } else { Rc::clone(&group.title) };
if regexp.is_match(group_title.as_str()) {
group.channels.sort_by(|a, b| {
let raw_value_a = get_field_value(a, &channel_sort.field);
let raw_value_b = get_field_value(b, &channel_sort.field);
let value_a = if *match_as_ascii { Rc::new(unidecode(&raw_value_a)) } else { raw_value_a };
let value_b = if *match_as_ascii { Rc::new(unidecode(&raw_value_b)) } else { raw_value_b };
let ordering = value_a.partial_cmp(&value_b).unwrap();
match channel_sort.order {
Asc => ordering,
Desc => ordering.reverse()
}
});
group.channels.sort_by(|chan1, chan2| playlistitem_comparator(chan1, chan2, channel_sort, match_as_ascii));
}
});
});
@@ -392,68 +394,22 @@ async fn process_playlist<'a>(playlists: &mut [FetchedPlaylist<'a>],
apply_affixes(&mut new_fetched_playlists);
if target.is_multi_input() && target.has_output(&TargetType::Xtream) {
let mut stream_id_mappings: Vec<MultiXtreamMapping> = Vec::new();
let mut counter: u32 = 0;
new_fetched_playlists.iter()
.flat_map(|pl| {
let input_id = &pl.input.id;
pl.playlist.iter().map(move |plg| (input_id, &plg.channels))
})
.flat_map(|(input_id, channels)| channels.iter().map(move |chan| (input_id, chan)))
.for_each(|(input_id, chan)| {
let mut header = chan.header.borrow_mut();
if header.stream_id.is_empty() {
header.stream_id = Rc::clone(&header.id);
}
match header.stream_id.parse::<u32>() {
Ok(stream_id) => {
let xtream_mapping = MultiXtreamMapping {
stream_id,
input_id: *input_id,
};
stream_id_mappings.push(xtream_mapping);
}
Err(_) => {
error!("Failed to parse stream_id: {}", &header.id)
}
}
counter += 1;
header.id = Rc::new(counter.to_string());
});
match write_xtream_mapping(&stream_id_mappings, cfg, &target.name) {
Ok(_) => {
debug!("wrote multi xtream input mapping for {}", &target.name);
}
Err(err) => {
return Err(vec![M3uFilterError::new(
M3uFilterErrorKind::Notify,
format!("Write multi xtream input mapping {} failed: {}", target.name, err))]);
}
}
}
let mut new_playlist = vec![];
let mut new_epg = vec![];
let mut tv_guides = vec![];
new_fetched_playlists.drain(..).for_each(|mut fp| {
let epg_channel_ids: HashSet<_> = fp.playlist.iter().flat_map(|g| &g.channels)
.filter_map(|c| c.header.borrow().epg_channel_id.clone()).collect();
fp.playlist.drain(..).for_each(|group| new_playlist.push(group));
if let Some(tv_guide) = fp.epg {
tv_guides.push(tv_guide);
let guide = tv_guides.last().unwrap();
if log_enabled!(Level::Debug) {
if !epg_channel_ids.is_empty() {
if let Some(tv_guide) = fp.epg {
debug!("found epg information for {}", &target.name);
}
let channel_ids: HashSet<_> = new_playlist.iter().flat_map(|g| &g.channels)
.filter_map(|c| c.header.borrow().epg_channel_id.clone()).collect();
if !channel_ids.is_empty() {
if let Some(epg) = guide.filter(&channel_ids) {
if let Some(epg) = tv_guide.filter(&epg_channel_ids) {
new_epg.push(epg);
}
} else if log_enabled!(Level::Debug) {
debug!("channel ids are empty");
}
} else if log_enabled!(Level::Debug) {
debug!("channel ids are empty");
}
});
src/processing/xtream_parser.rs
+5 -4
View File
@@ -50,7 +50,6 @@ pub(crate) fn parse_xtream_series_info(info: &Value, group_title: &str, input: &
audio_track: default_as_empty_rc_str(),
time_shift: default_as_empty_rc_str(),
rec: default_as_empty_rc_str(),
// source is meant to hold the original provider data
source: default_as_empty_rc_str(),
url: if episode.direct_source.is_empty() {
let ext = episode.container_extension.to_owned();
@@ -64,6 +63,7 @@ pub(crate) fn parse_xtream_series_info(info: &Value, group_title: &str, input: &
xtream_cluster: XtreamCluster::Series,
additional_properties: episode.get_additional_properties(&series_info),
series_fetched: false,
category_id: 0,
})
}).collect();
if result.is_empty() { Ok(None) } else { Ok(Some(result)) }
@@ -80,6 +80,7 @@ pub(crate) fn parse_xtream(input: &ConfigInput,
streams: &Value) -> Result<Option<Vec<PlaylistGroup>>, M3uFilterError> {
match map_to_xtream_category(category) {
Ok(mut categories) => {
let input_id = input.id.to_string();
let url = input.url.as_str();
let username = input.username.as_ref().map_or("", |v| v);
let password = input.password.as_ref().map_or("", |v| v);
@@ -108,8 +109,7 @@ pub(crate) fn parse_xtream(input: &ConfigInput,
audio_track: default_as_empty_rc_str(),
time_shift: default_as_empty_rc_str(),
rec: default_as_empty_rc_str(),
// source is meant to hold the original provider data
source: default_as_empty_rc_str(),
source: Rc::new(input_id.to_owned()),
url: if stream.direct_source.is_empty() {
let stream_base_url = match xtream_cluster {
XtreamCluster::Live => format!("{}/live/{}/{}/{}.ts", url, username, password, &stream.get_stream_id()),
@@ -134,6 +134,7 @@ pub(crate) fn parse_xtream(input: &ConfigInput,
xtream_cluster: xtream_cluster.clone(),
additional_properties: stream.get_additional_properties(),
series_fetched: false,
category_id: 0,
}),
};
grp.add(item);
@@ -146,7 +147,7 @@ pub(crate) fn parse_xtream(input: &ConfigInput,
id: cat.category_id.parse::<u32>().unwrap_or(0),
xtream_cluster: xtream_cluster.clone(),
title: Rc::clone(&cat.category_name),
channels: cat.channels.clone()
channels: cat.channels.clone(),
}
}).collect()))
}