mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-30 10:42:29 +01:00
Added target watch for group content changes
This commit is contained in:
euzu
@@ -0,0 +1,173 @@
|
||||
use std::cell::RefCell;
|
||||
use crate::model::config::Config;
|
||||
use crate::model::model_m3u::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, XtreamCluster};
|
||||
|
||||
fn token_value(it: &mut std::str::Chars) -> String {
|
||||
if let Some(oc) = it.next() {
|
||||
if oc == '"' {
|
||||
return get_value(it);
|
||||
}
|
||||
}
|
||||
String::from("")
|
||||
}
|
||||
|
||||
fn get_value(it: &mut std::str::Chars) -> String {
|
||||
let mut result: Vec<char> = vec![];
|
||||
for oc in it.by_ref() {
|
||||
if oc == '"' {
|
||||
break;
|
||||
}
|
||||
result.push(oc);
|
||||
}
|
||||
result.iter().collect::<String>()
|
||||
}
|
||||
|
||||
fn token_till(it: &mut std::str::Chars, stop_char: char) -> Option<String> {
|
||||
let mut result: Vec<char> = vec![];
|
||||
for ch in it.by_ref() {
|
||||
if ch == stop_char {
|
||||
break;
|
||||
} else if ch.is_whitespace() && result.is_empty() {
|
||||
continue;
|
||||
} else {
|
||||
result.push(ch);
|
||||
}
|
||||
}
|
||||
if !result.is_empty() { Some(result.iter().collect::<String>()) } else { None }
|
||||
}
|
||||
|
||||
fn skip_digit(it: &mut std::str::Chars) -> Option<char> {
|
||||
loop {
|
||||
match it.next() {
|
||||
Some(c) => {
|
||||
if !(c == '-' || c == '+' || c.is_ascii_digit()) {
|
||||
return Some(c);
|
||||
}
|
||||
}
|
||||
None => return None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn create_empty_playlistitem_header(content: &String) -> PlaylistItemHeader {
|
||||
PlaylistItemHeader {
|
||||
id: String::from(""),
|
||||
name: String::from(""),
|
||||
logo: String::from(""),
|
||||
logo_small: String::from(""),
|
||||
group: String::from("Unknown"),
|
||||
title: String::from(""),
|
||||
parent_code: String::from(""),
|
||||
audio_track: String::from(""),
|
||||
time_shift: String::from(""),
|
||||
rec: String::from(""),
|
||||
source: String::from(content),
|
||||
xtream_cluster: XtreamCluster::Live,
|
||||
additional_properties: None,
|
||||
}
|
||||
}
|
||||
|
||||
fn process_header(video_suffixes: &Vec<&str>,content: &String, url: String) -> PlaylistItemHeader {
|
||||
let mut plih = create_empty_playlistitem_header(content);
|
||||
let mut it = content.chars();
|
||||
let line_token = token_till(&mut it, ':');
|
||||
if line_token == Some(String::from("#EXTINF")) {
|
||||
let mut c = skip_digit(&mut it);
|
||||
loop {
|
||||
if c.is_none() {
|
||||
break;
|
||||
}
|
||||
match c.unwrap() {
|
||||
',' => plih.title = get_value(&mut it),
|
||||
_ => {
|
||||
let token = token_till(&mut it, '=');
|
||||
if let Some(t) = token {
|
||||
let value = token_value(&mut it);
|
||||
match t.as_str() {
|
||||
"tvg-id" => plih.id = value,
|
||||
"tvg-name" => plih.name = value,
|
||||
"group-title" => if !value.is_empty() { plih.group = value },
|
||||
"parent-code" => plih.parent_code = value,
|
||||
"audio-track" => plih.audio_track = value,
|
||||
"tvg-logo" => plih.logo = value,
|
||||
"tvg-logo-small" => plih.logo_small = value,
|
||||
"timeshift" => plih.time_shift = value,
|
||||
"tvg-rec" => plih.rec = value,
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
c = it.next();
|
||||
}
|
||||
}
|
||||
|
||||
for suffix in video_suffixes {
|
||||
if url.ends_with(suffix) {
|
||||
// TODO find Series based on group or configured names
|
||||
plih.xtream_cluster = XtreamCluster::Video;
|
||||
break;
|
||||
}
|
||||
}
|
||||
plih
|
||||
}
|
||||
|
||||
|
||||
|
||||
pub(crate) fn parse_m3u(cfg: &Config, lines: &Vec<String>) -> Vec<PlaylistGroup> {
|
||||
let mut groups: std::collections::HashMap<String, Vec<PlaylistItem>> = std::collections::HashMap::new();
|
||||
let mut sort_order: Vec<String> = vec![];
|
||||
let mut header: Option<String> = None;
|
||||
let mut group: Option<String> = None;
|
||||
|
||||
let video_suffixes = if let Some(suffixes) = &cfg.video_suffix {
|
||||
suffixes.iter().map(|s| s.as_str()).collect()
|
||||
} else {
|
||||
vec![".mp4", ".mkv", ".avi"]
|
||||
};
|
||||
|
||||
for line in lines {
|
||||
if line.starts_with("#EXTINF") {
|
||||
header = Some(String::from(line));
|
||||
continue;
|
||||
}
|
||||
if line.starts_with("#EXTGRP") {
|
||||
group = Some(String::from(&line[8..]));
|
||||
continue;
|
||||
}
|
||||
if line.starts_with('#') {
|
||||
continue;
|
||||
}
|
||||
if let Some(header_value) = header {
|
||||
let item = PlaylistItem { header: RefCell::new(process_header(&video_suffixes, &header_value, String::from(line))), url: String::from(line) };
|
||||
if let Some(group_value) = group {
|
||||
if item.header.borrow().group.is_empty() {
|
||||
item.header.borrow_mut().group = group_value;
|
||||
}
|
||||
}
|
||||
let key = String::from(&item.header.borrow().group);
|
||||
// let key2 = String::from(&item.header.group);
|
||||
match groups.entry(key.clone()) {
|
||||
std::collections::hash_map::Entry::Vacant(e) => {
|
||||
e.insert(vec![item]);
|
||||
sort_order.push(key);
|
||||
}
|
||||
std::collections::hash_map::Entry::Occupied(mut e) => { e.get_mut().push(item); }
|
||||
}
|
||||
}
|
||||
header = None;
|
||||
group = None;
|
||||
}
|
||||
|
||||
let mut result: Vec<PlaylistGroup> = vec![];
|
||||
for (grp_id, (key, channels)) in (1_i32..).zip(groups.into_iter()) {
|
||||
let cluster = channels.first().map(|pli| pli.header.borrow().xtream_cluster.clone());
|
||||
result.push(PlaylistGroup { id: grp_id, xtream_cluster: cluster.unwrap(), title: key, channels });
|
||||
}
|
||||
result.sort_by(|f, s| {
|
||||
let i1 = sort_order.iter().position(|r| r == &f.title).unwrap();
|
||||
let i2 = sort_order.iter().position(|r| r == &s.title).unwrap();
|
||||
i1.cmp(&i2)
|
||||
});
|
||||
result
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
pub(crate) mod m3u_parser;
|
||||
pub(crate) mod xtream_parser;
|
||||
pub(crate) mod playlist_processor;
|
||||
pub(crate) mod playlist_watch;
|
||||
@@ -0,0 +1,425 @@
|
||||
extern crate unidecode;
|
||||
|
||||
use std::cell::RefCell;
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::thread;
|
||||
use log::{debug, error, info};
|
||||
use unidecode::unidecode;
|
||||
use crate::{model::config, Config, valid_property, create_m3u_filter_error_result};
|
||||
use crate::model::config::{ConfigTarget, default_as_default, InputAffix, InputType, ProcessTargets};
|
||||
use crate::model::model_config::{SortOrder::{Asc, Desc}, ItemField, AFFIX_FIELDS, ProcessingOrder};
|
||||
use crate::filter::{get_field_value, MockValueProcessor, set_field_value, ValueProvider};
|
||||
use crate::repository::m3u_repository::write_playlist;
|
||||
use crate::model::model_m3u::{FetchedPlaylist, FieldAccessor, PlaylistGroup, PlaylistItem, PlaylistItemHeader};
|
||||
use crate::model::mapping::{Mapping, MappingValueProcessor};
|
||||
use crate::download::{get_m3u_playlist, get_xtream_playlist};
|
||||
use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind};
|
||||
use crate::processing::playlist_watch::process_group_watch;
|
||||
use crate::repository::xtream_repository::xtream_save_playlist;
|
||||
|
||||
fn filter_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
|
||||
debug!("Filtering {} groups", playlist.len());
|
||||
let mut new_playlist = Vec::new();
|
||||
playlist.iter_mut().for_each(|pg| {
|
||||
debug!("Filtering group {} with {} items", pg.title, pg.channels.len());
|
||||
let mut channels = Vec::new();
|
||||
pg.channels.iter_mut().for_each(|pli| {
|
||||
if is_valid(pli, target) {
|
||||
channels.push(pli.clone());
|
||||
}
|
||||
});
|
||||
debug!("Filtered group {} has now {} items", pg.title, channels.len());
|
||||
if !channels.is_empty() {
|
||||
new_playlist.push(PlaylistGroup {
|
||||
id: pg.id,
|
||||
title: pg.title.clone(),
|
||||
channels,
|
||||
xtream_cluster: pg.xtream_cluster.clone(),
|
||||
});
|
||||
}
|
||||
});
|
||||
Some(new_playlist)
|
||||
}
|
||||
|
||||
fn apply_affixes(fetched_playlists: &mut [FetchedPlaylist]) {
|
||||
fetched_playlists.iter_mut().for_each(|fetched_playlist| {
|
||||
let FetchedPlaylist { input, playlist } = fetched_playlist;
|
||||
if input.suffix.is_some() || input.prefix.is_some() {
|
||||
let validate_affix = |a: &Option<InputAffix>| match a {
|
||||
Some(affix) => {
|
||||
valid_property!(&affix.field.as_str(), AFFIX_FIELDS) && !affix.value.is_empty()
|
||||
}
|
||||
_ => false
|
||||
};
|
||||
|
||||
let apply_prefix = validate_affix(&input.prefix);
|
||||
let apply_suffix = validate_affix(&input.suffix);
|
||||
|
||||
if apply_prefix || apply_suffix {
|
||||
let get_affix_applied_value = |header: &mut PlaylistItemHeader, affix: &InputAffix, prefix: bool| {
|
||||
if let Some(field_value) = header.get_field(affix.field.as_str()) {
|
||||
return if prefix {
|
||||
format!("{}{}", &affix.value, field_value.as_str())
|
||||
} else {
|
||||
format!("{}{}", field_value.as_str(), &affix.value)
|
||||
};
|
||||
}
|
||||
String::from(&affix.value)
|
||||
};
|
||||
|
||||
playlist.iter_mut().for_each(|group| {
|
||||
group.channels.iter_mut().for_each(|channel| {
|
||||
if apply_suffix {
|
||||
if let Some(suffix) = &input.suffix {
|
||||
let value = get_affix_applied_value(&mut channel.header.borrow_mut(), suffix, false);
|
||||
debug!("Applying input suffix: {}={}", &suffix.field, &value);
|
||||
channel.header.borrow_mut().set_field(&suffix.field, value.as_str());
|
||||
}
|
||||
}
|
||||
if apply_prefix {
|
||||
if let Some(prefix) = &input.prefix {
|
||||
let value = get_affix_applied_value(&mut channel.header.borrow_mut(), prefix, true);
|
||||
debug!("Applying input prefix: {}={}", &prefix.field, &value);
|
||||
channel.header.borrow_mut().set_field(&prefix.field, value.as_str());
|
||||
}
|
||||
}
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
fn sort_playlist(target: &ConfigTarget, new_playlist: &mut [PlaylistGroup]) {
|
||||
if let Some(sort) = &target.sort {
|
||||
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 { unidecode(&a.title) } else { String::from(&a.title) };
|
||||
let value_b = if *match_as_ascii { unidecode(&b.title) } else { String::from(&b.title) };
|
||||
let ordering = value_a.partial_cmp(&value_b).unwrap();
|
||||
match group_sort.order {
|
||||
Asc => ordering,
|
||||
Desc => ordering.reverse()
|
||||
}
|
||||
});
|
||||
}
|
||||
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 { unidecode(&group.title) } else { String::from(&group.title) };
|
||||
let is_match = regexp.is_match(group_title.as_str());
|
||||
if is_match {
|
||||
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 { unidecode(&raw_value_a) } else { raw_value_a };
|
||||
let value_b = if *match_as_ascii { 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 is_valid(pli: &mut PlaylistItem, target: &ConfigTarget) -> bool {
|
||||
let provider = ValueProvider { pli: RefCell::new(pli) };
|
||||
target.filter(&provider)
|
||||
}
|
||||
|
||||
fn exec_rename(pli: &mut PlaylistItem, rename: &Option<Vec<config::ConfigRename>>) {
|
||||
if let Some(renames) = rename {
|
||||
if !renames.is_empty() {
|
||||
let result = pli;
|
||||
for r in renames {
|
||||
let value = get_field_value(result, &r.field);
|
||||
let cap = r.re.as_ref().unwrap().replace_all(value.as_str(), &r.new_name);
|
||||
debug!("Renamed {}={} to {}", &r.field, value, cap);
|
||||
let value = cap.into_owned();
|
||||
set_field_value(result, &r.field, value);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn rename_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
|
||||
match &target.rename {
|
||||
Some(renames) => {
|
||||
if !renames.is_empty() {
|
||||
let mut new_playlist: Vec<PlaylistGroup> = Vec::new();
|
||||
for g in playlist {
|
||||
let mut grp = g.clone();
|
||||
for r in renames {
|
||||
if let ItemField::Group = r.field {
|
||||
let cap = r.re.as_ref().unwrap().replace_all(&grp.title, &r.new_name);
|
||||
debug!("Renamed group {} to {} for {}", &grp.title, cap, target.name);
|
||||
grp.title = cap.into_owned();
|
||||
}
|
||||
}
|
||||
|
||||
grp.channels.iter_mut().for_each(|pli| exec_rename(pli, &target.rename));
|
||||
new_playlist.push(grp);
|
||||
}
|
||||
return Some(new_playlist);
|
||||
}
|
||||
None
|
||||
}
|
||||
_ => None
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! apply_pattern {
|
||||
($pattern:expr, $provider:expr, $processor:expr) => {{
|
||||
match $pattern {
|
||||
Some(ptrn) => {
|
||||
ptrn.filter($provider, $processor);
|
||||
},
|
||||
_ => {}
|
||||
};
|
||||
}};
|
||||
}
|
||||
|
||||
fn map_channel(channel: &PlaylistItem, mapping: &Mapping) -> PlaylistItem {
|
||||
if !mapping.mapper.is_empty() {
|
||||
let header = channel.header.borrow();
|
||||
let channel_name = if mapping.match_as_ascii { unidecode(&header.name) } else { String::from(&header.name) };
|
||||
if mapping.match_as_ascii { debug!("Decoded {} for matching to {}", &header.name, &channel_name); };
|
||||
drop(header);
|
||||
let ref_chan = RefCell::new(channel);
|
||||
let provider = ValueProvider { pli: ref_chan.clone() };
|
||||
let mut mock_processor = MockValueProcessor {};
|
||||
|
||||
for m in &mapping.mapper {
|
||||
let mut processor = MappingValueProcessor { pli: ref_chan.clone(), mapper: RefCell::new(m) };
|
||||
match &m._filter {
|
||||
Some(filter) => {
|
||||
if filter.filter(&provider, &mut mock_processor) {
|
||||
apply_pattern!(&m._pattern, &provider, &mut processor);
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
apply_pattern!(&m._pattern, &provider, &mut processor);
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
channel.clone()
|
||||
}
|
||||
|
||||
fn map_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Option<Vec<PlaylistGroup>> {
|
||||
if target._mapping.is_some() {
|
||||
let new_playlist: Vec<PlaylistGroup> = playlist.iter().map(|playlist_group| {
|
||||
let mut grp = playlist_group.clone();
|
||||
let mappings = target._mapping.as_ref().unwrap();
|
||||
mappings.iter().filter(|mapping| !mapping.mapper.is_empty()).for_each(|mapping|
|
||||
grp.channels = grp.channels.iter_mut().map(|chan| map_channel(chan, mapping)).collect());
|
||||
grp
|
||||
}).collect();
|
||||
|
||||
// if the group names are changed, restructure channels to the right groups
|
||||
// we use
|
||||
let mut new_groups: Vec<PlaylistGroup> = Vec::new();
|
||||
let mut grp_id: i32 = 0;
|
||||
for playlist_group in new_playlist {
|
||||
for channel in &playlist_group.channels {
|
||||
let cluster = &channel.header.borrow().xtream_cluster;
|
||||
let title = &channel.header.borrow().group;
|
||||
match new_groups.iter_mut().find(|x| &*x.title == title) {
|
||||
Some(grp) => grp.channels.push(channel.clone()),
|
||||
_ => {
|
||||
grp_id += 1;
|
||||
new_groups.push(PlaylistGroup {
|
||||
id: grp_id,
|
||||
title: String::from(title),
|
||||
channels: vec![channel.clone()],
|
||||
xtream_cluster: cluster.clone(),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(new_groups)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
}
|
||||
|
||||
// If no input is enabled but the user set the target as command line argument,
|
||||
// we force the input to be enabled.
|
||||
// If there are enabled input, then only these are used.
|
||||
fn is_input_enabled(enabled_inputs: usize, input_enabled: bool, input_id: u16, user_targets: &Arc<ProcessTargets>) -> bool {
|
||||
if enabled_inputs == 0 {
|
||||
return user_targets.enabled && user_targets.has_input(input_id);
|
||||
}
|
||||
input_enabled
|
||||
}
|
||||
|
||||
fn process_source(cfg: Arc<Config>, source_idx: usize, user_targets: Arc<ProcessTargets>) -> Vec<M3uFilterError> {
|
||||
let source = cfg.sources.get(source_idx).unwrap();
|
||||
let mut all_playlist = Vec::new();
|
||||
let enabled_inputs = source.inputs.iter().filter(|item| item.enabled).count();
|
||||
let mut errors = vec![];
|
||||
for input in &source.inputs {
|
||||
//if input.enabled || (user_targets.enabled && user_targets.has_input(input.id)) {
|
||||
if is_input_enabled(enabled_inputs, input.enabled, input.id, &user_targets) {
|
||||
let (playlist, mut error_list) = match input.input_type {
|
||||
InputType::M3u => get_m3u_playlist(&cfg, input, &cfg.working_dir),
|
||||
InputType::Xtream => get_xtream_playlist(input, &cfg.working_dir),
|
||||
};
|
||||
error_list.drain(..).for_each(|err| errors.push(err));
|
||||
if !playlist.is_empty() {
|
||||
all_playlist.push(
|
||||
FetchedPlaylist {
|
||||
input,
|
||||
playlist,
|
||||
}
|
||||
);
|
||||
} else {
|
||||
info!("source is empty {}", input.url);
|
||||
let input_name = match &input.name {
|
||||
None => input.url.as_str(),
|
||||
Some(name_val) => name_val.as_str()
|
||||
};
|
||||
errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("source is empty {}", input_name)));
|
||||
}
|
||||
}
|
||||
}
|
||||
if all_playlist.is_empty() {
|
||||
debug!("Source at {} input is empty", source_idx);
|
||||
errors.push(M3uFilterError::new(M3uFilterErrorKind::Notify, format!("Source at {} input is empty", source_idx)));
|
||||
} else {
|
||||
debug!("Input has {} groups", all_playlist.len());
|
||||
source.targets.iter().for_each(|target| {
|
||||
let should_process = (!user_targets.enabled && target.enabled)
|
||||
|| (user_targets.enabled && user_targets.has_target(target.id));
|
||||
if should_process {
|
||||
match process_playlist(&mut all_playlist, target, &cfg) {
|
||||
Ok(_) => {}
|
||||
Err(err) => errors.push(err)
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
errors
|
||||
}
|
||||
|
||||
pub(crate) fn process_sources(cfg: Config, user_targets: &ProcessTargets) -> Vec<M3uFilterError> {
|
||||
let config = Arc::new(cfg);
|
||||
let mut handle_list = vec![];
|
||||
let thread_num = config.threads;
|
||||
let process_parallel = thread_num > 1 && config.sources.len() > 1;
|
||||
if process_parallel { debug!("Using {} threads", thread_num); }
|
||||
|
||||
let errors = Arc::new(Mutex::<Vec<M3uFilterError>>::new(vec![]));
|
||||
for (index, _) in config.sources.iter().enumerate() {
|
||||
let cfg_clone = config.clone();
|
||||
let usr_targets = Arc::new(user_targets.clone());
|
||||
let shared_errors = errors.clone();
|
||||
let process = move || {
|
||||
process_source(cfg_clone, index, usr_targets).drain(..).for_each(|err| shared_errors.lock().unwrap().push(err));
|
||||
};
|
||||
if process_parallel {
|
||||
let handles = &mut handle_list;
|
||||
handles.push(thread::spawn(process));
|
||||
if handles.len() as u8 >= thread_num {
|
||||
handles.drain(..).for_each(|handle| { let _ = handle.join(); });
|
||||
}
|
||||
} else {
|
||||
process();
|
||||
}
|
||||
}
|
||||
for handle in handle_list {
|
||||
let _ = handle.join();
|
||||
}
|
||||
Arc::try_unwrap(errors).unwrap().into_inner().unwrap()
|
||||
}
|
||||
|
||||
|
||||
type ProcessingPipe = Vec<fn(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Option<Vec<PlaylistGroup>>>;
|
||||
|
||||
fn get_processing_pipe(target: &ConfigTarget) -> ProcessingPipe {
|
||||
match &target.processing_order {
|
||||
ProcessingOrder::Frm => vec![filter_playlist, rename_playlist, map_playlist],
|
||||
ProcessingOrder::Fmr => vec![filter_playlist, map_playlist, rename_playlist],
|
||||
ProcessingOrder::Rfm => vec![rename_playlist, filter_playlist, map_playlist],
|
||||
ProcessingOrder::Rmf => vec![rename_playlist, map_playlist, filter_playlist],
|
||||
ProcessingOrder::Mfr => vec![map_playlist, filter_playlist, rename_playlist],
|
||||
ProcessingOrder::Mrf => vec![map_playlist, rename_playlist, filter_playlist]
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn process_playlist(playlists: &mut [FetchedPlaylist],
|
||||
target: &ConfigTarget, cfg: &Config) -> Result<(), M3uFilterError> {
|
||||
let pipe = get_processing_pipe(target);
|
||||
|
||||
debug!("Processing order is {}", &target.processing_order);
|
||||
|
||||
let mut new_fetched_playlist : Vec<FetchedPlaylist> = vec![];
|
||||
playlists.iter_mut().for_each(|fpl| {
|
||||
let mut new_fpl = FetchedPlaylist {
|
||||
input: fpl.input,
|
||||
playlist: fpl.playlist.clone(),
|
||||
};
|
||||
for f in &pipe {
|
||||
let playlist = &mut new_fpl.playlist;
|
||||
let r = f(playlist, target);
|
||||
if let Some(v) = r {
|
||||
new_fpl.playlist = v;
|
||||
}
|
||||
}
|
||||
new_fetched_playlist.push(new_fpl);
|
||||
});
|
||||
|
||||
apply_affixes(&mut new_fetched_playlist);
|
||||
let mut new_playlist = vec![];
|
||||
new_fetched_playlist.iter_mut().for_each(|fp| {
|
||||
fp.playlist.drain(..).for_each(|group| new_playlist.push(group));
|
||||
});
|
||||
|
||||
if !new_playlist.is_empty() {
|
||||
sort_playlist(target, &mut new_playlist);
|
||||
|
||||
if target.watch.is_some() {
|
||||
if default_as_default().eq_ignore_ascii_case(&target.name) {
|
||||
error!("cant watch a target with no unique name");
|
||||
} else {
|
||||
let titles = target.watch.as_ref().unwrap();
|
||||
new_playlist.iter().for_each(|pl| {
|
||||
if titles.contains(&pl.title) {
|
||||
process_group_watch(cfg, &target.name, pl)
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
let publish = target.publish;
|
||||
if target.filename.is_some() {
|
||||
let result = write_playlist(target, cfg, &mut new_playlist);
|
||||
match &result {
|
||||
Ok(..) => {}
|
||||
Err(e) => {
|
||||
error!("failed to write playlist {} {}", target.name, e);
|
||||
}
|
||||
}
|
||||
if !publish {
|
||||
return result;
|
||||
}
|
||||
}
|
||||
|
||||
if target.publish {
|
||||
return xtream_save_playlist(target, cfg, &mut new_playlist);
|
||||
}
|
||||
create_m3u_filter_error_result!(M3uFilterErrorKind::Notify, "Persisting playlist failed: {}", &target.name)
|
||||
} else {
|
||||
info!("Playlist is empty: {}", &target.name);
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
use std::collections::BTreeSet;
|
||||
use std::path::PathBuf;
|
||||
use log::{error, info};
|
||||
use regex::Regex;
|
||||
use crate::messaging::send_message;
|
||||
use crate::model::config::Config;
|
||||
use crate::model::model_m3u::PlaylistGroup;
|
||||
use crate::utils;
|
||||
|
||||
pub(crate) fn process_group_watch(cfg: &Config, target_name: &str, pl: &PlaylistGroup) {
|
||||
let mut new_tree = BTreeSet::new();
|
||||
pl.channels.iter().for_each(|chan| {
|
||||
let header = chan.header.borrow();
|
||||
let title = if header.title.is_empty() { header.title.to_string() } else { header.name.to_string() };
|
||||
new_tree.insert(title);
|
||||
});
|
||||
|
||||
let filename_re = Regex::new(r"[^A-Za-z0-9_.-\\s]").unwrap();
|
||||
let watch_filename = format!("watch_{}.bin", filename_re.replace_all(target_name, "_"));
|
||||
match utils::get_file_path(&cfg.working_dir, Some(std::path::PathBuf::from(&watch_filename))) {
|
||||
Some(path) => {
|
||||
let save_path = path.clone();
|
||||
if path.exists() {
|
||||
match load_tree(&path) {
|
||||
Some(loaded_tree) => {
|
||||
// Find elements in set2 but not in set1
|
||||
let added_difference: BTreeSet<String> = new_tree.difference(&loaded_tree).cloned().collect();
|
||||
let removed_difference: BTreeSet<String> = loaded_tree.difference(&new_tree).cloned().collect();
|
||||
if !added_difference.is_empty() || !removed_difference.is_empty() {
|
||||
handle_watch_notification(cfg, added_difference, removed_difference);
|
||||
}
|
||||
}
|
||||
None => {
|
||||
error!("failed to load watch_file {}", &path.into_os_string().into_string().unwrap());
|
||||
}
|
||||
}
|
||||
}
|
||||
match save_tree(&save_path, new_tree) {
|
||||
Ok(_) => {}
|
||||
Err(err) => {
|
||||
error!("failed to write watch_file {}: {}", &save_path.into_os_string().into_string().unwrap(), err)
|
||||
}
|
||||
}
|
||||
}
|
||||
None => {
|
||||
error!("failed to write watch_file {}", &watch_filename);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn handle_watch_notification(cfg: &Config, added: BTreeSet<String>, removed: BTreeSet<String>) {
|
||||
let added_entries = added.iter().map(|name| name.to_string()).collect::<Vec<String>>().join("\n\t");
|
||||
let removed_entries = removed.iter().map(|name| name.to_string()).collect::<Vec<String>>().join("\n\t");
|
||||
|
||||
let mut message = vec![];
|
||||
if !added_entries.is_empty() {
|
||||
message.push("added: [\n\t".to_string());
|
||||
message.push(added_entries);
|
||||
message.push("\n]\n".to_string());
|
||||
}
|
||||
if !removed_entries.is_empty() {
|
||||
message.push("removed: [\n\t".to_string());
|
||||
message.push(removed_entries);
|
||||
message.push("\n]\n".to_string());
|
||||
}
|
||||
|
||||
if !message.is_empty() {
|
||||
info!("{}", message.join("").as_str());
|
||||
send_message(&cfg.messaging, message.join("").as_str())
|
||||
}
|
||||
}
|
||||
|
||||
fn load_tree(path: &PathBuf) -> Option<BTreeSet<String>> {
|
||||
match std::fs::read(path) {
|
||||
Ok(encoded) => {
|
||||
let decoded: BTreeSet<String> = bincode::deserialize(&encoded[..]).unwrap();
|
||||
Some(decoded)
|
||||
}
|
||||
Err(_) => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn save_tree(path: &PathBuf, tree: BTreeSet<String>) -> std::io::Result<()> {
|
||||
let encoded: Vec<u8> = bincode::serialize(&tree).unwrap();
|
||||
std::fs::write(path, encoded)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,298 @@
|
||||
use std::cell::RefCell;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::atomic::{AtomicI32, Ordering};
|
||||
|
||||
use serde::{Deserialize, Deserializer, Serialize};
|
||||
use serde::de::DeserializeOwned;
|
||||
use serde_json::Value;
|
||||
use crate::create_m3u_filter_error_result;
|
||||
use crate::m3u_filter_error::{M3uFilterError, M3uFilterErrorKind};
|
||||
|
||||
use crate::model::model_config::default_as_empty_str;
|
||||
use crate::model::model_m3u::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, XtreamCluster};
|
||||
|
||||
fn default_as_empty_list() -> Vec<PlaylistItem> { vec![] }
|
||||
|
||||
fn deserialize_number_from_string<'de, D, T: DeserializeOwned>(
|
||||
deserializer: D,
|
||||
) -> Result<Option<T>, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
// we define a local enum type inside of the function
|
||||
// because it is untagged, serde will deserialize as the first variant
|
||||
// that it can
|
||||
#[derive(Deserialize)]
|
||||
#[serde(untagged)]
|
||||
enum MaybeNumber<U> {
|
||||
// if it can be parsed as Option<T>, it will be
|
||||
Value(Option<U>),
|
||||
// otherwise try parsing as a string
|
||||
NumberString(String),
|
||||
}
|
||||
|
||||
// deserialize into local enum
|
||||
let value: MaybeNumber<T> = Deserialize::deserialize(deserializer)?;
|
||||
match value {
|
||||
// if parsed as T or None, return that
|
||||
MaybeNumber::Value(value) => Ok(value),
|
||||
|
||||
// (if it is any other string)
|
||||
MaybeNumber::NumberString(string) => {
|
||||
match serde_json::from_str::<T>(string.as_str()) {
|
||||
Ok(val) => Ok(Some(val)),
|
||||
Err(_) => Ok(None)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn value_to_string_array(value: &[Value]) -> Option<Vec<String>> {
|
||||
Some(value.iter().filter_map(value_to_string).collect())
|
||||
}
|
||||
|
||||
fn value_to_string(v: &Value) -> Option<String> {
|
||||
match v {
|
||||
Value::Bool(value) => Some(value.to_string()),
|
||||
Value::Number(value) => Some(value.to_string()),
|
||||
Value::String(value) => Some(value.to_string()),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn deserialize_as_option_string<'de, D>(deserializer: D) -> Result<Option<String>, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let value: Value = Deserialize::deserialize(deserializer)?;
|
||||
|
||||
match &value {
|
||||
Value::String(s) => Ok(Some(s.clone())),
|
||||
Value::Number(s) => Ok(Some(s.to_string())),
|
||||
_ => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
fn deserialize_as_string<'de, D>(deserializer: D) -> Result<String, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
let value: Value = Deserialize::deserialize(deserializer)?;
|
||||
|
||||
match &value {
|
||||
Value::String(s) => Ok(s.clone()),
|
||||
_ => Ok(value.to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn deserialize_as_string_array<'de, D>(deserializer: D) -> Result<Option<Vec<String>>, D::Error>
|
||||
where
|
||||
D: Deserializer<'de>,
|
||||
{
|
||||
Value::deserialize(deserializer).map(|v| match v {
|
||||
Value::Array(value) => value_to_string_array(&value),
|
||||
_ => None,
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct XtreamCategory {
|
||||
#[serde(deserialize_with = "deserialize_as_string")]
|
||||
pub category_id: String,
|
||||
#[serde(deserialize_with = "deserialize_as_string")]
|
||||
pub category_name: String,
|
||||
//pub parent_id: i32,
|
||||
#[serde(default = "default_as_empty_list")]
|
||||
pub channels: Vec<PlaylistItem>,
|
||||
}
|
||||
|
||||
impl XtreamCategory {
|
||||
fn add(&mut self, item: PlaylistItem) {
|
||||
self.channels.push(item);
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize)]
|
||||
struct XtreamStream {
|
||||
#[serde(default, deserialize_with = "deserialize_as_string")]
|
||||
pub name: String,
|
||||
#[serde(default, deserialize_with = "deserialize_as_string")]
|
||||
pub category_id: String,
|
||||
#[serde(default, deserialize_with = "deserialize_number_from_string")]
|
||||
pub stream_id: Option<i32>,
|
||||
#[serde(default, deserialize_with = "deserialize_number_from_string")]
|
||||
pub series_id: Option<i32>,
|
||||
#[serde(default = "default_as_empty_str", deserialize_with = "deserialize_as_string")]
|
||||
pub stream_icon: String,
|
||||
#[serde(default = "default_as_empty_str", deserialize_with = "deserialize_as_string")]
|
||||
pub direct_source: String,
|
||||
|
||||
// optional attributes
|
||||
#[serde(default, deserialize_with = "deserialize_as_string_array")]
|
||||
backdrop_path: Option<Vec<String>>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
added: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
cast: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
container_extension: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
cover: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
director: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
episode_run_time: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
genre: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
last_modified: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
plot: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_number_from_string")]
|
||||
rating: Option<f64>,
|
||||
#[serde(default, deserialize_with = "deserialize_number_from_string")]
|
||||
rating_5based: Option<f64>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
release_date: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
stream_type: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
title: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
year: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
youtube_trailer: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_as_option_string")]
|
||||
epg_channel_id: Option<String>,
|
||||
#[serde(default, deserialize_with = "deserialize_number_from_string")]
|
||||
tv_archive: Option<i32>,
|
||||
#[serde(default, deserialize_with = "deserialize_number_from_string")]
|
||||
tv_archive_duration: Option<i32>,
|
||||
}
|
||||
|
||||
macro_rules! add_str_property_if_exists {
|
||||
($vec:expr, $prop:expr, $prop_name:expr) => {
|
||||
$prop.as_ref().map(|v| $vec.push((String::from($prop_name), Value::String(v.to_string()))));
|
||||
}
|
||||
}
|
||||
macro_rules! add_i64_property_if_exists {
|
||||
($vec:expr, $prop:expr, $prop_name:expr) => {
|
||||
$prop.as_ref().map(|v| $vec.push((String::from($prop_name), Value::Number(serde_json::value::Number::from(i64::from(*v))))));
|
||||
}
|
||||
}
|
||||
|
||||
macro_rules! add_f64_property_if_exists {
|
||||
($vec:expr, $prop:expr, $prop_name:expr) => {
|
||||
$prop.as_ref().map(|v| $vec.push((String::from($prop_name), Value::Number(serde_json::value::Number::from_f64(f64::from(*v)).unwrap()))));
|
||||
}
|
||||
}
|
||||
|
||||
impl XtreamStream {
|
||||
pub(crate) fn get_stream_id(&self) -> String {
|
||||
self.stream_id.map_or_else(|| self.series_id.map_or_else(|| String::from(""), |seid| format!("{}", seid)), |sid| format!("{}", sid))
|
||||
}
|
||||
|
||||
pub(crate) fn get_additional_properties(&self) -> Option<Vec<(String, Value)>> {
|
||||
let mut result = vec![];
|
||||
if let Some(bdpath) = self.backdrop_path.as_ref() {
|
||||
if !bdpath.is_empty() {
|
||||
result.push((String::from("backdrop_path"), Value::Array(Vec::from([Value::String(String::from(bdpath.get(0).unwrap()))]))));
|
||||
}
|
||||
}
|
||||
add_str_property_if_exists!(result, self.added, "added");
|
||||
add_str_property_if_exists!(result, self.cast, "cast");
|
||||
add_str_property_if_exists!(result, self.container_extension, "container_extension");
|
||||
add_str_property_if_exists!(result, self.cover, "cover");
|
||||
add_str_property_if_exists!(result, self.director, "director");
|
||||
add_str_property_if_exists!(result, self.episode_run_time, "episode_run_time");
|
||||
add_str_property_if_exists!(result, self.genre, "genre");
|
||||
add_str_property_if_exists!(result, self.last_modified, "last_modified");
|
||||
add_str_property_if_exists!(result, self.plot, "plot");
|
||||
add_f64_property_if_exists!(result, self.rating, "rating");
|
||||
add_f64_property_if_exists!(result, self.rating_5based, "rating_5based");
|
||||
add_str_property_if_exists!(result, self.release_date, "release_date");
|
||||
add_str_property_if_exists!(result, self.stream_type, "stream_type");
|
||||
add_str_property_if_exists!(result, self.title, "title");
|
||||
add_str_property_if_exists!(result, self.year, "year");
|
||||
add_str_property_if_exists!(result, self.youtube_trailer, "youtube_trailer");
|
||||
add_str_property_if_exists!(result, self.epg_channel_id, "epg_channel_id");
|
||||
add_i64_property_if_exists!(result, self.tv_archive, "tv_archive");
|
||||
add_i64_property_if_exists!(result, self.tv_archive_duration, "tv_archive_duration");
|
||||
if result.is_empty() { None } else { Some(result) }
|
||||
}
|
||||
}
|
||||
|
||||
fn process_category(category: &serde_json::Value) -> Result<Vec<XtreamCategory>, M3uFilterError> {
|
||||
match serde_json::from_value::<Vec<XtreamCategory>>(category.to_owned()) {
|
||||
Ok(category_list) => Ok(category_list),
|
||||
Err(err) => {
|
||||
create_m3u_filter_error_result!(M3uFilterErrorKind::Notify, "Failed to process categories {}", &err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
fn process_streams(xtream_cluster: &XtreamCluster, streams: &serde_json::Value) -> Result<Vec<XtreamStream>, M3uFilterError> {
|
||||
match serde_json::from_value::<Vec<XtreamStream>>(streams.to_owned()) {
|
||||
Ok(stream_list) => Ok(stream_list),
|
||||
Err(err) => {
|
||||
create_m3u_filter_error_result!(M3uFilterErrorKind::Notify, "Failed to process streams {:?}: {}", xtream_cluster, &err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn parse_xtream(cat_id_cnt: &AtomicI32, xtream_cluster: &XtreamCluster,
|
||||
category: &serde_json::Value, streams: &serde_json::Value,
|
||||
stream_base_url: &String) -> Result<Option<Vec<PlaylistGroup>>, M3uFilterError> {
|
||||
match process_category(category) {
|
||||
Ok(mut categories) => {
|
||||
return match process_streams(xtream_cluster, streams) {
|
||||
Ok(streams) => {
|
||||
let group_map: HashMap::<String, RefCell<XtreamCategory>> =
|
||||
categories.drain(..).map(|category|
|
||||
(String::from(&category.category_id), RefCell::new(category))
|
||||
).collect();
|
||||
|
||||
for stream in streams {
|
||||
if let Some(group) = group_map.get(stream.category_id.as_str()) {
|
||||
let mut grp = group.borrow_mut();
|
||||
let title = String::from(&grp.category_name);
|
||||
let item = PlaylistItem {
|
||||
header: RefCell::new(PlaylistItemHeader {
|
||||
id: stream.get_stream_id(),
|
||||
name: String::from(&stream.name),
|
||||
logo: String::from(&stream.stream_icon),
|
||||
logo_small: "".to_string(),
|
||||
group: title,
|
||||
title: String::from(&stream.name),
|
||||
parent_code: "".to_string(),
|
||||
audio_track: "".to_string(),
|
||||
time_shift: "".to_string(),
|
||||
rec: "".to_string(),
|
||||
source: String::from(&stream.direct_source),
|
||||
xtream_cluster: xtream_cluster.clone(),
|
||||
additional_properties: stream.get_additional_properties(),
|
||||
}),
|
||||
url: format!("{}/{}", stream_base_url, stream.get_stream_id()),
|
||||
};
|
||||
grp.add(item);
|
||||
}
|
||||
}
|
||||
|
||||
Ok(Some(group_map.values().map(|category| {
|
||||
let cat = category.borrow();
|
||||
cat_id_cnt.fetch_add(1, Ordering::Relaxed);
|
||||
PlaylistGroup {
|
||||
id: cat_id_cnt.load(Ordering::Relaxed),
|
||||
xtream_cluster: xtream_cluster.clone(),
|
||||
title: String::from(&cat.category_name),
|
||||
channels: cat.channels.clone(),
|
||||
}
|
||||
}).collect()))
|
||||
}
|
||||
Err(err) => Err(err)
|
||||
};
|
||||
}
|
||||
Err(err) => Err(err)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user