tuliprox/src/playlist_processor.rs

extern crate unidecode;

use std::cell::RefCell;
use std::sync::{Arc};
use std::thread;
use config::ConfigTarget;
use unidecode::unidecode;
use crate::{config, Config, valid_property};
use crate::config::{ConfigInput, InputAffix, InputType, ProcessTargets};
use crate::model_config::SortOrder::{Asc, Desc};
use crate::filter::{get_field_value, MockValueProcessor, set_field_value, ValueProvider};
use crate::m3u_repository::write_playlist;
use crate::model_m3u::{FieldAccessor, PlaylistGroup, PlaylistItem, PlaylistItemHeader};
use crate::mapping::{Mapping, MappingValueProcessor};
use crate::model_config::{ItemField, AFFIX_FIELDS, ProcessingOrder};
use crate::download::{get_m3u_playlist, get_xtream_playlist};
use crate::repository::save_playlist;

fn filter_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget, verbose: bool) -> Option<Vec<PlaylistGroup>> {
    if verbose { println!("Filtering {} groups", playlist.len()) }
    let mut new_playlist = Vec::new();
    playlist.iter_mut().for_each(|pg| {
        if verbose { println!("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, verbose) {
                channels.push(pli.clone());
            }
        });
        if verbose { println!("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(playlist: &mut [PlaylistGroup], input: &ConfigInput, verbose: bool) {
    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);
                            if verbose { println!("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);
                            if verbose { println!("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, verbose: bool) -> bool {
    let provider = ValueProvider { pli: RefCell::new(pli) };
    target.filter(&provider, verbose)
}

fn exec_rename(pli: &mut PlaylistItem, rename: &Option<Vec<config::ConfigRename>>, verbose: bool) {
    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);
                if verbose { println!("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, verbose: bool) -> 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);
                            if verbose { println!("Renamed group {} to {}", &grp.title, cap); }
                            grp.title = cap.into_owned();
                        }
                    }

                    grp.channels.iter_mut().for_each(|pli| exec_rename(pli, &target.rename, verbose));
                    new_playlist.push(grp);
                }
                return Some(new_playlist);
            }
            None
        }
        _ => None
    }
}

macro_rules! apply_pattern {
    ($pattern:expr, $provider:expr, $processor:expr, $verbose:expr) => {{
            match $pattern {
                Some(ptrn) => {
                    ptrn.filter($provider, $processor, $verbose);
                },
                _ => {}
            };
    }};
}

fn map_channel(channel: &PlaylistItem, mapping: &Mapping, verbose: bool) -> 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 verbose && mapping.match_as_ascii { println!("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, verbose) {
                        apply_pattern!(&m._pattern, &provider, &mut processor, verbose);
                    }
                }
                _ => {
                    apply_pattern!(&m._pattern, &provider, &mut processor, verbose);
                }
            };
        }
    }
    channel.clone()
}

fn map_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget, verbose: bool) -> Option<Vec<PlaylistGroup>> {
    if verbose { println!("Mapping") }
    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, verbose)).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
    }
}

fn process_source(cfg: Arc<Config>, source_idx: usize, user_targets: Arc<ProcessTargets>, verbose: bool) {
    let source = cfg.sources.get(source_idx).unwrap();
    let input = &source.input;
    if input.enabled || (user_targets.enabled && user_targets.has_input(input.id)) {
        let mut result = match input.input_type {
            InputType::M3u => get_m3u_playlist(&cfg, input, &cfg.working_dir, verbose),
            InputType::Xtream => get_xtream_playlist(input, &cfg.working_dir, verbose),
        };
        if let Some(playlist) = result.as_mut() {
            if playlist.is_empty() {
                if verbose { println!("Input file is empty") }
            } else {
                if verbose { println!("Input file has {} groups", 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(playlist, input, target, &cfg, verbose) {
                            Ok(_) => (),
                            Err(e) => println!("Failed to write file: {}", e)
                        }
                    }
                });
            }
        }
    }
}

pub(crate) fn process_sources(cfg: Config, user_targets: &ProcessTargets, verbose: bool) {
    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 verbose && process_parallel { println!("Using {} threads", thread_num) }

    for (index, _) in config.sources.iter().enumerate() {
        let cfg_clone = config.clone();
        let usr_targets = Arc::new(user_targets.clone());
        let process = move || process_source(cfg_clone, index, usr_targets, verbose);
        if process_parallel {
            let handles = &mut handle_list;
            handles.push(thread::spawn(process));
            if handles.len() as u8 >= thread_num {
                while let Some(handle) = handles.pop() {
                    let _ = handle.join();
                }
            }
        } else {
            process();
        }
    }
    for handle in handle_list {
        let _ = handle.join();
    }
}


type ProcessingPipe = Vec<fn(playlist: &mut [PlaylistGroup], target: &ConfigTarget, verbose: bool) -> Option<Vec<PlaylistGroup>>>;

pub(crate) fn process_playlist(playlist: &mut [PlaylistGroup],
                               input: &ConfigInput,
                               target: &ConfigTarget, cfg: &Config,
                               verbose: bool) -> Result<(), std::io::Error> {
    let pipe: 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]
        };

    if verbose { println!("Processing order is {}", &target.processing_order) }

    let mut new_playlist = playlist.to_owned();
    for f in pipe {
        let r = f(&mut new_playlist, target, verbose);
        if let Some(v) = r {
            new_playlist = v;
        }
    }

    apply_affixes(&mut new_playlist, input, verbose);

    sort_playlist(target, &mut new_playlist);
    let publish = target.publish;
    if target.filename.is_some() {
        let result = write_playlist(target, cfg, &mut new_playlist);
        match &result {
            Ok(..) => {},
            Err(e) => {
                println!("failed to write {:?}", e);
            }
        }
        if !publish {
            return result;
        }
    }

    if target.publish {
        return save_playlist(target, cfg, &mut new_playlist);
    }
    Err(std::io::Error::new(std::io::ErrorKind::Other, format!("Persisting playlist failed: {}", &target.name)))
}