Optimizations

* New Features

Added series information display with episodes, cast, and metadata in playlist explorer
Introduced multi-cluster playlist endpoints (live, VOD, series) for flexible content retrieval
Added series info lookup and filtering capabilities
Implemented sidebar navigation for source editor with dedicated pages (Main, Options, Staged, Advanced, Alias)
Added filter state controls (All, Selected, Deselected) for playlist category management
Enabled Trakt configuration toggle to control integration
Performance Improvements

Optimized B+Tree indexing for faster queries
Reduced startup latency with disk-based operations and optional memory caching
Improved string memory handling through efficient reference-counted strings
UI Enhancements

Enhanced playlist explorer with series details and episode grouping by season
Added new UI icons (Options, Advanced, Staging, Alias, Trakt)
Improved form layouts with toolbar controls
Enhanced radio button group styling for better visual hierarchy

* New Features

Added series information display with episodes, cast, and metadata in playlist explorer
Introduced multi-cluster playlist endpoints (live, VOD, series) for flexible content retrieval
Added series info lookup and filtering capabilities
Implemented sidebar navigation for source editor with dedicated pages (Main, Options, Staged, Advanced, Alias)
Added filter state controls (All, Selected, Deselected) for playlist category management
Enabled Trakt configuration toggle to control integration
Performance Improvements

Optimized B+Tree indexing for faster queries
Reduced startup latency with disk-based operations and optional memory caching
Improved string memory handling through efficient reference-counted strings
UI Enhancements

Enhanced playlist explorer with series details and episode grouping by season
Added new UI icons (Options, Advanced, Staging, Alias, Trakt)
Improved form layouts with toolbar controls
Enhanced radio button group styling for better visual hierarchy

* New Features

Added series information display with episodes, cast, and metadata in playlist explorer
Introduced multi-cluster playlist endpoints (live, VOD, series) for flexible content retrieval
Added series info lookup and filtering capabilities
Implemented sidebar navigation for source editor with dedicated pages (Main, Options, Staged, Advanced, Alias)
Added filter state controls (All, Selected, Deselected) for playlist category management
Enabled Trakt configuration toggle to control integration
Performance Improvements

Optimized B+Tree indexing for faster queries
Reduced startup latency with disk-based operations and optional memory caching
Improved string memory handling through efficient reference-counted strings
UI Enhancements

Enhanced playlist explorer with series details and episode grouping by season
Added new UI icons (Options, Advanced, Staging, Alias, Trakt)
Improved form layouts with toolbar controls
Enhanced radio button group styling for better visual hierarchy

* New Features

Added series information display with episodes, cast, and metadata in playlist explorer
Introduced multi-cluster playlist endpoints (live, VOD, series) for flexible content retrieval
Added series info lookup and filtering capabilities
Implemented sidebar navigation for source editor with dedicated pages (Main, Options, Staged, Advanced, Alias)
Added filter state controls (All, Selected, Deselected) for playlist category management
Enabled Trakt configuration toggle to control integration
Performance Improvements

Optimized B+Tree indexing for faster queries
Reduced startup latency with disk-based operations and optional memory caching
Improved string memory handling through efficient reference-counted strings
UI Enhancements

Enhanced playlist explorer with series details and episode grouping by season
Added new UI icons (Options, Advanced, Staging, Alias, Trakt)
Improved form layouts with toolbar controls
Enhanced radio button group styling for better visual hierarchy
This commit is contained in:
euzu
2026-01-20 10:09:58 +01:00
committed by GitHub
parent 39c4fca9ca
commit 16fcd7f882
139 changed files with 5094 additions and 2797 deletions
+115 -35
View File
@@ -1,12 +1,16 @@
use std::borrow::Cow;
use crate::model::ProxyUserCredentials;
use shared::utils::{CONSTANTS, HLS_PREFIX};
use crate::utils::{deobfuscate_text, obfuscate_text};
use shared::utils::{CONSTANTS, HLS_PREFIX};
use std::str;
use url::Url;
use shared::concat_string;
const TOKEN_SEPARATOR: char = '\x1F';
const TOKEN_SEPARATOR_STR: &str = "\x1F";
fn create_hls_session_token_and_url(secret: &[u8], session_token: &str, stream_url: &str) -> Option<String> {
if let Ok(cookie_value) = obfuscate_text(secret, &format!("{session_token}{TOKEN_SEPARATOR}{stream_url}")) {
if let Ok(cookie_value) = obfuscate_text(secret, &concat_string!(session_token, TOKEN_SEPARATOR_STR, stream_url)) {
return Some(cookie_value);
}
None
@@ -15,7 +19,7 @@ fn create_hls_session_token_and_url(secret: &[u8], session_token: &str, stream_u
pub fn get_hls_session_token_and_url_from_token(secret: &[u8], token: &str) -> Option<(Option<String>, String)> {
if let Ok(decrypted) = deobfuscate_text(secret, token) {
let parts: Vec<&str> = decrypted.split(TOKEN_SEPARATOR).collect();
if !parts.is_empty() && parts.len() == 2 {
if parts.len() == 2 {
let session_token: String = parts[0].to_string();
let stream_url: String = parts[1].to_string();
return Some((Some(session_token), stream_url));
@@ -24,7 +28,6 @@ pub fn get_hls_session_token_and_url_from_token(secret: &[u8], token: &str) -> O
None
}
pub struct RewriteHlsProps<'a> {
pub secret: &'a [u8; 16],
pub base_url: &'a str,
@@ -35,33 +38,43 @@ pub struct RewriteHlsProps<'a> {
pub user_token: Option<&'a str>,
}
fn rewrite_hls_url(input: &str, replacement: &str) -> String {
if replacement.starts_with('/') {
let parts = input.splitn(4, '/').collect::<Vec<&str>>();
if parts.len() < 4 {
return replacement.to_string(); // less than 3 Slashes → replace all
}
format!("{}/{}/{}{}", parts[0], parts[1], parts[2], replacement)
} else {
match input.rsplitn(2, '/').collect::<Vec<&str>>().as_slice() {
[_after, before] => format!("{before}/{replacement}"),
[_only] => replacement.to_string(),
_ => input.to_string(),
}
/// Rewrites an HLS URI relative to a base playlist URL.
/// Absolute URIs are returned unchanged.
pub fn rewrite_hls_url<'a>(base: &'a str, reference: &'a str) -> Cow<'a, str> {
// absolute URI → passthrough
if Url::parse(reference).is_ok() {
return Cow::Borrowed(reference);
}
let Ok(base_url) = Url::parse(base) else {
return Cow::Borrowed(reference);
};
base_url.join(reference).map_or_else(|_| Cow::Borrowed(reference), |u| Cow::Owned(u.to_string()))
}
fn rewrite_uri_attrib(line: &str, props: &RewriteHlsProps) -> String {
if let Some(caps) = CONSTANTS.re_hls_uri.captures(line) {
let uri = &caps[1];
let target_url = &rewrite_hls_url(&props.hls_url, uri);
if let Some(user_token) = &props.user_token {
if let Some(token) = create_hls_session_token_and_url(props.secret, user_token, target_url) {
return CONSTANTS.re_hls_uri.replace(line, format!(r#"URI="{token}""#)).to_string();
}
}
}
line.to_string()
fn rewrite_uri_attrib<'a>(line: &'a str, props: &RewriteHlsProps) -> Cow<'a, str> {
let Some(caps) = CONSTANTS.re_hls_uri.captures(line) else {
return Cow::Borrowed(line);
};
let uri = &caps[1];
let rewritten = rewrite_hls_url(&props.hls_url, uri);
let final_uri = if let Some(user_token) = &props.user_token {
create_hls_session_token_and_url(
props.secret,
user_token,
&rewritten,
).map(Cow::Owned).unwrap_or(rewritten)
} else {
rewritten
};
Cow::Owned(CONSTANTS
.re_hls_uri
.replace(line, format!(r#"URI="{final_uri}""#))
.to_string())
}
pub fn rewrite_hls(user: &ProxyUserCredentials, props: &RewriteHlsProps) -> String {
@@ -69,7 +82,6 @@ pub fn rewrite_hls(user: &ProxyUserCredentials, props: &RewriteHlsProps) -> Stri
let password = &user.password;
let mut result = Vec::new();
for line in props.content.lines() {
if line.trim().is_empty() {
continue;
}
@@ -77,16 +89,12 @@ pub fn rewrite_hls(user: &ProxyUserCredentials, props: &RewriteHlsProps) -> Stri
// skip comments
if line.starts_with('#') {
let rewritten = rewrite_uri_attrib(line, props);
result.push(rewritten);
result.push(rewritten.to_string());
continue;
}
// target url
let target_url = if line.starts_with("http") {
line.to_string()
} else {
rewrite_hls_url(&props.hls_url, line)
};
let target_url = rewrite_hls_url(&props.hls_url, line);
if let Some(user_token) = &props.user_token {
if let Some(token) = create_hls_session_token_and_url(props.secret, user_token, &target_url) {
let url = format!(
@@ -110,6 +118,7 @@ pub fn rewrite_hls(user: &ProxyUserCredentials, props: &RewriteHlsProps) -> Stri
mod test {
use rand::RngCore;
use shared::utils::u32_to_base64;
use crate::processing::parser::hls::{rewrite_hls_url};
#[test]
fn test_token_size() {
@@ -119,4 +128,75 @@ mod test {
}
}
#[test]
fn rewrite_http_relative_segment() {
let base = "http://example.com/hls/playlist.m3u8";
let uri = "seg001.ts";
let out = rewrite_hls_url(base, uri);
assert_eq!(out, "http://example.com/hls/seg001.ts");
}
#[test]
fn rewrite_http_root_relative_segment() {
let base = "http://example.com/hls/playlist.m3u8";
let uri = "/media/seg001.ts";
let out = rewrite_hls_url(base, uri);
assert_eq!(out, "http://example.com/media/seg001.ts");
}
#[test]
fn rewrite_http_parent_directory() {
let base = "http://example.com/hls/level1/playlist.m3u8";
let uri = "../seg001.ts";
let out = rewrite_hls_url(base, uri);
assert_eq!(out, "http://example.com/hls/seg001.ts");
}
#[test]
fn rewrite_https_absolute_passthrough() {
let base = "http://example.com/hls/playlist.m3u8";
let uri = "https://cdn.example.org/video/seg.ts";
let out = rewrite_hls_url(base, uri);
assert_eq!(out, uri);
}
#[test]
fn rewrite_file_relative_segment() {
let base = "file:///mnt/media/hls/playlist.m3u8";
let uri = "seg001.ts";
let out = rewrite_hls_url(base, uri);
assert_eq!(out, "file:///mnt/media/hls/seg001.ts");
}
#[test]
fn rewrite_file_parent_directory() {
let base = "file:///mnt/media/hls/level1/playlist.m3u8";
let uri = "../seg001.ts";
let out = rewrite_hls_url(base, uri);
assert_eq!(out, "file:///mnt/media/hls/seg001.ts");
}
#[test]
fn rewrite_file_absolute_passthrough() {
let base = "file:///mnt/media/hls/playlist.m3u8";
let uri = "file:///mnt/other/seg.ts";
let out = rewrite_hls_url(base, uri);
assert_eq!(out, uri);
}
#[test]
fn rewrite_hls_fragment() {
let base = "http://example.com/hls/playlist.m3u8";
let fragment = "seg.ts#t=10";
let out = rewrite_hls_url(base, fragment);
assert_eq!(out, "http://example.com/hls/seg.ts#t=10");
}
}
+52 -52
View File
@@ -1,25 +1,26 @@
use crate::model::{Config, ConfigInput};
use crate::utils::request::DynReader;
use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, XtreamCluster};
use shared::utils::{default_supported_video_extensions, extract_id_from_url, StringInterner};
use shared::utils::{default_supported_video_extensions, extract_id_from_url, Internable};
use std::borrow::BorrowMut;
use std::sync::Arc;
use tokio::io::AsyncBufReadExt;
use indexmap::IndexMap;
use shared::concat_string;
// other implementations like calculating text_distance on all titles took too much time
// we keep it now as simple as possible and less memory intensive.
fn get_title_group(text: &str) -> String {
fn get_title_group(text: &Arc<str>) -> Arc<str> {
let alphabetic_only: String = text.chars().map(|c| if c.is_alphanumeric() { c } else { ' ' }).collect();
let parts = alphabetic_only.split_whitespace();
let mut combination = String::new();
for p in parts {
combination = concat_string!(&combination, " " , p);
if combination.len() > 2 {
return combination;
return combination.intern();
}
}
text.to_string()
text.clone()
}
#[inline]
@@ -90,11 +91,11 @@ fn skip_digit(it: &mut std::str::Chars) -> Option<char> {
}
}
fn create_empty_playlistitem_header(interner: &mut StringInterner, input_name: &str, url: String) -> PlaylistItemHeader {
fn create_empty_playlistitem_header(input_name: &Arc<str>, url: String) -> PlaylistItemHeader {
PlaylistItemHeader {
url,
url: Arc::from(url),
category_id: 0,
input_name: interner.intern(input_name),
input_name: Arc::clone(input_name),
..Default::default()
}
}
@@ -110,7 +111,7 @@ macro_rules! process_header_fields {
};
}
fn process_header(interner: &mut StringInterner, input_name: &str, video_suffixes: &[String], content: &str, url: String) -> PlaylistItemHeader {
fn process_header(input_name: &Arc<str>, video_suffixes: &[String], content: &str, url: String) -> PlaylistItemHeader {
let url_id = extract_id_from_url(&url);
let url_types = if video_suffixes.iter().any(|suffix| url.ends_with(suffix)) {
// TODO find Series based on group or configured names
@@ -119,7 +120,7 @@ fn process_header(interner: &mut StringInterner, input_name: &str, video_suffixe
None
};
let mut plih = create_empty_playlistitem_header(interner, input_name, url);
let mut plih = create_empty_playlistitem_header(input_name, url);
let mut it = content.chars();
let mut stack = String::with_capacity(64);
let line_token = token_till(&mut stack, &mut it, ':', false);
@@ -133,7 +134,7 @@ fn process_header(interner: &mut StringInterner, input_name: &str, video_suffixe
if chr.is_whitespace() {
// skip
} else if chr == ',' {
plih.title = get_value(&mut stack, &mut it);
plih.title = get_value(&mut stack, &mut it).intern();
} else {
stack.push(chr);
let token = token_till(&mut stack, &mut it, '=', true);
@@ -146,19 +147,22 @@ fn process_header(interner: &mut StringInterner, input_name: &str, video_suffixe
}
} else if token == "tvg-chno" {
plih.chno = value.parse::<u32>().unwrap_or(0);
} else if token == "group-title" {
plih.group = value.intern();
} else if token == "tvg-id" {
plih.id = value.intern();
} else if token == "tvg-name" {
plih.name = value.intern();
} else if token == "tvg-logo" {
plih.logo = value.intern();
} else if token == "tvg-logo-small" {
plih.logo_small = value.intern();
} else {
process_header_fields!(plih, token.as_str(),
(id, "tvg-id"),
(name, "tvg-name"),
(parent_code, "parent-code"),
(audio_track, "audio-track"),
(logo, "tvg-logo"),
(logo_small, "tvg-logo-small"),
(time_shift, "timeshift"),
(rec, "tvg-rec"); value.clone());
if token == "group-title" {
plih.group = interner.intern(&value);
}
(rec, "tvg-rec"); value.intern());
}
}
}
@@ -170,14 +174,14 @@ fn process_header(interner: &mut StringInterner, input_name: &str, video_suffixe
if plih.id.is_empty() {
plih.epg_channel_id = None;
if let Some(pid) = provider_id {
plih.id = pid;
plih.id = pid.intern();
} else if let Some(chanid) = url_id {
plih.id = chanid;
plih.id = chanid.intern();
}
} else {
plih.epg_channel_id = Some(plih.id.clone());
plih.epg_channel_id = Some(Arc::clone(&plih.id));
if let Some(pid) = provider_id {
plih.id = pid;
plih.id = pid.intern();
}
}
}
@@ -204,8 +208,7 @@ fn process_header(interner: &mut StringInterner, input_name: &str, video_suffixe
pub async fn consume_m3u<F: FnMut(PlaylistItem)>(cfg: &Config, input: &ConfigInput, lines: DynReader, mut visit: F) {
let mut header: Option<String> = None;
let mut group: Option<String> = None;
let mut interner = StringInterner::new();
let input_name = input.name.as_str();
let input_name = &input.name;
let video_suffixes = match cfg.video.as_ref() {
Some(config) => {
@@ -228,19 +231,19 @@ pub async fn consume_m3u<F: FnMut(PlaylistItem)>(cfg: &Config, input: &ConfigInp
continue;
}
if let Some(header_value) = header {
let mut item = PlaylistItem { header: process_header(&mut interner, input_name, &video_suffixes, &header_value, line) };
let mut item = PlaylistItem { header: process_header(input_name, &video_suffixes, &header_value, line) };
let header = &mut item.header;
header.source_ordinal = ord_counter;
ord_counter += 1;
if header.group.is_empty() {
if let Some(group_value) = group {
header.group = interner.intern(&group_value);
} else {
let current_title = header.title.clone();
header.group = interner.intern(&get_title_group(current_title.as_str()));
if header.group.is_empty() {
if let Some(group_value) = group {
header.group = group_value.intern();
} else {
let group = get_title_group(&header.title);
header.group = group;
}
}
}
visit(item);
visit(item);
}
header = None;
group = None;
@@ -274,53 +277,50 @@ pub async fn parse_m3u(cfg: &Config, input: &ConfigInput, lines: DynReader) -> V
#[cfg(test)]
mod test {
use shared::utils::Internable;
use crate::processing::parser::m3u::process_header;
use shared::utils::StringInterner;
#[test]
fn test_process_header_1() {
let mut interner = StringInterner::new();
let input: &str = "hello";
let input = "hello".intern();
let video_suffixes = Vec::new();
let url = "http://hello.de/hello.ts";
let line = r#"#EXTINF:-1 channel-id="abc-seven" tvg-id="abc-seven" tvg-logo="https://abc.nz/.images/seven.png" tvg-chno="7" group-title="Sydney" , Seven"#;
let pli = process_header(&mut interner, input, &video_suffixes, line, url.to_string());
assert_eq!(pli.title, "Seven");
assert_eq!(pli.id, "abc-seven");
assert_eq!(pli.logo, "https://abc.nz/.images/seven.png");
let pli = process_header(&input, &video_suffixes, line, url.to_string());
assert_eq!(pli.title, "Seven".intern());
assert_eq!(pli.id, "abc-seven".intern());
assert_eq!(pli.logo, "https://abc.nz/.images/seven.png".intern());
assert_eq!(pli.chno, 7);
assert_eq!(&*pli.group, "Sydney");
}
#[test]
fn test_process_header_2() {
let mut interner = StringInterner::new();
let input: &str = "hello";
let input = "hello".intern();
let video_suffixes = Vec::new();
let url = "http://hello.de/hello.ts";
let line = r#"#EXTINF:-1 channel-id="abc-seven" tvg-id="abc-seven" tvg-logo="https://abc.nz/.images/seven.png" tvg-chno="7" group-title="Sydney", Seven"#;
let pli = process_header(&mut interner, input, &video_suffixes, line, url.to_string());
assert_eq!(pli.title, "Seven");
assert_eq!(pli.id, "abc-seven");
assert_eq!(pli.logo, "https://abc.nz/.images/seven.png");
let pli = process_header(&input, &video_suffixes, line, url.to_string());
assert_eq!(pli.title, "Seven".intern());
assert_eq!(pli.id, "abc-seven".intern());
assert_eq!(pli.logo, "https://abc.nz/.images/seven.png".intern());
assert_eq!(pli.chno, 7);
assert_eq!(&*pli.group, "Sydney");
}
#[test]
fn test_process_header_xui_id() {
let mut interner = StringInterner::new();
let input: &str = "hello";
let input = "hello".intern();
let video_suffixes = Vec::new();
let url = "http://hello.de/hello.ts";
let line = r#"#EXTINF:-1 tvg-id="abc-seven" xui-id="provider-123" group-title="Sydney", Seven"#;
let pli = process_header(&mut interner, input, &video_suffixes, line, url.to_string());
assert_eq!(pli.title, "Seven");
assert_eq!(pli.id, "provider-123"); // Should use xui-id
assert_eq!(pli.epg_channel_id, Some("abc-seven".to_string())); // Should preserve original tvg-id
let pli = process_header(&input, &video_suffixes, line, url.to_string());
assert_eq!(pli.title, "Seven".intern());
assert_eq!(pli.id, "provider-123".intern()); // Should use xui-id
assert_eq!(pli.epg_channel_id, Some("abc-seven".intern())); // Should preserve original tvg-id
assert_eq!(&*pli.group, "Sydney");
}
}
+8 -6
View File
@@ -7,7 +7,7 @@ use dashmap::DashMap;
use quick_xml::events::{BytesStart, BytesText, Event};
use rayon::iter::{IntoParallelRefIterator, ParallelIterator};
use shared::model::EpgNamePrefix;
use shared::utils::{deunicode_string, CONSTANTS};
use shared::utils::{deunicode_string, Internable, CONSTANTS};
use std::borrow::Cow;
use std::cell::RefCell;
use std::cmp::min;
@@ -488,7 +488,7 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option<Epg> {
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 count = tv_guides.iter().map(|tvg| tvg.children.len()).sum();
let channel_mapping: DashMap<String, i16> = DashMap::with_capacity(count);
let channel_mapping: DashMap<Arc<str>, i16> = DashMap::with_capacity(count);
let mut sorted_guides = tv_guides.to_vec();
// sort by priority
@@ -499,14 +499,15 @@ pub fn flatten_tvguide(tv_guides: &[Epg]) -> Option<Epg> {
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) {
let chan_id = chan_id.intern();
let should_add = {
// if not stored
!channel_mapping.contains_key(chan_id) ||
!channel_mapping.contains_key(&chan_id) ||
// or if priority is higher (less means higher priority)
channel_mapping.get(chan_id).as_deref().is_none_or(|&priority| guide.priority < priority)
channel_mapping.get(&chan_id).as_deref().is_none_or(|&priority| guide.priority < priority)
};
if should_add {
if let Some(mut existing) = channel_mapping.get_mut(chan_id) {
if let Some(mut existing) = channel_mapping.get_mut(&chan_id) {
if guide.priority < *existing {
*existing = guide.priority;
children.push(c.clone());
@@ -522,7 +523,8 @@ 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(stored_priority) = channel_mapping.get(chan_id) {
let chan_id = chan_id.intern();
if let Some(stored_priority) = channel_mapping.get(&chan_id) {
if *stored_priority == guide.priority {
children.push(c.clone());
}
+126 -125
View File
@@ -1,19 +1,18 @@
use std::borrow::Cow;
use crate::model::ConfigInput;
use crate::model::{XtreamCategory};
use crate::model::XtreamCategory;
use crate::utils::request::DynReader;
use crate::utils::xtream::get_xtream_stream_url_base;
use shared::error::{notify_err_res, notify_err, TuliproxError};
use indexmap::IndexMap;
use serde::Deserializer;
use shared::error::{notify_err, notify_err_res, TuliproxError};
use shared::model::UUIDType;
use shared::model::{EpisodeStreamProperties, LiveStreamProperties, PlaylistGroup, PlaylistItem,
PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties,
SeriesStreamProperties, StreamProperties, UUIDType, VideoStreamProperties,
SeriesStreamProperties, StreamProperties, VideoStreamProperties,
XtreamCluster, XtreamPlaylistItem};
use shared::utils::{generate_playlist_uuid, trim_last_slash, StringInterner};
use indexmap::IndexMap;
use shared::utils::{generate_playlist_uuid, trim_last_slash, Internable};
use std::sync::Arc;
use tokio::task::spawn_blocking;
use serde::Deserializer;
async fn map_to_xtream_category(categories: DynReader) -> Result<Vec<XtreamCategory>, TuliproxError> {
spawn_blocking(move || {
@@ -51,46 +50,46 @@ async fn map_to_xtream_streams(xtream_cluster: XtreamCluster, streams: DynReader
}).await.map_err(|e| notify_err!("Mapping xtream streams failed: {e}"))?
}
fn create_xtream_series_episode_url<'a>(url: &'a str, username: &'a str, password: &'a str, episode: &'a SeriesStreamDetailEpisodeProperties) -> Cow<'a, str> {
fn create_xtream_series_episode_url(url: &str, username: &str, password: &str, episode: &SeriesStreamDetailEpisodeProperties) -> Arc<str> {
if episode.direct_source.is_empty() {
let ext = episode.container_extension.clone();
let stream_base_url = format!("{url}/series/{username}/{password}/{}.{ext}", episode.id);
Cow::Owned(stream_base_url)
stream_base_url.into()
} else {
Cow::Borrowed(&episode.direct_source)
Arc::clone(&episode.direct_source)
}
}
pub fn parse_xtream_series_info(parent_uuid: &UUIDType, series_info: &SeriesStreamProperties,
group_title: &str, series_name: &str, input: &ConfigInput,
interner: &mut StringInterner) -> Option<Vec<PlaylistItem>> {
pub fn parse_xtream_series_info(parent_uuid: &UUIDType, series_info: &SeriesStreamProperties,
group_title: &str, series_name: &Arc<str>, input: &ConfigInput) -> Option<Vec<PlaylistItem>> {
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);
if let Some(episodes) = series_info.details.as_ref().and_then(|d| d.episodes.as_ref()) {
let result: Vec<PlaylistItem> = episodes.iter().map(|episode| {
let episode_id = episode.id.to_string();
let episode_url = create_xtream_series_episode_url(url, username, password, episode);
let episode_info = EpisodeStreamProperties::from_series(series_info, episode);
PlaylistItem {
header: PlaylistItemHeader {
id: episode.id.to_string(),
uuid: generate_playlist_uuid(&input.name, &episode.id.to_string(), PlaylistItemType::Series, &episode_url),
// we use parent_code to track the parent series
parent_code: parent_uuid.to_string(),
name: series_name.to_string(),
logo: episode.movie_image.clone(),
group: interner.intern(group_title),
title: episode.title.clone(),
url: episode_url.to_string(),
item_type: PlaylistItemType::Series,
xtream_cluster: XtreamCluster::Series,
additional_properties: Some(StreamProperties::Episode(episode_info)),
category_id: 0,
input_name: interner.intern(&input.name),
..Default::default()
}
}
PlaylistItem {
header: PlaylistItemHeader {
uuid: generate_playlist_uuid(&input.name, &episode_id, PlaylistItemType::Series, &episode_url),
id: episode_id.into(),
// we use parent_code to track the parent series
parent_code: parent_uuid.intern(),
name: Arc::clone(series_name),
logo: Arc::clone(&episode.movie_image),
group: group_title.intern(),
title: Arc::clone(&episode.title),
url: episode_url,
item_type: PlaylistItemType::Series,
xtream_cluster: XtreamCluster::Series,
additional_properties: Some(StreamProperties::Episode(episode_info)),
category_id: 0,
input_name: input.name.intern(),
..Default::default()
}
}
}).collect();
return if result.is_empty() { None } else { Some(result) };
}
@@ -122,12 +121,12 @@ pub fn get_xtream_url(xtream_cluster: XtreamCluster, url: &str,
stream_base_url
}
pub fn create_xtream_url<'a>(xtream_cluster: XtreamCluster, url: &'a str, username: &'a str, password: &'a str,
stream: &'a StreamProperties, live_stream_use_prefix: bool, live_stream_without_extension: bool) -> Cow<'a, str> {
pub fn create_xtream_url(xtream_cluster: XtreamCluster, url: &str, username: &str, password: &str,
stream: &StreamProperties, live_stream_use_prefix: bool, live_stream_without_extension: bool) -> Arc<str> {
stream.get_direct_source().unwrap_or_else(||
Cow::Owned(get_xtream_url(xtream_cluster, url, username, password, stream.get_stream_id(),
get_xtream_url(xtream_cluster, url, username, password, stream.get_stream_id(),
stream.get_container_extension().as_deref(),
live_stream_use_prefix, live_stream_without_extension))
live_stream_use_prefix, live_stream_without_extension).into()
)
}
@@ -137,7 +136,6 @@ pub async fn parse_xtream(input: &ConfigInput,
streams: DynReader) -> Result<Option<Vec<PlaylistGroup>>, TuliproxError> {
match map_to_xtream_category(categories).await {
Ok(xtream_categories) => {
let mut interner = StringInterner::new();
let input_name = input.name.clone();
let url = input.url.as_str();
let username = input.username.as_ref().map_or("", |v| v);
@@ -149,7 +147,7 @@ pub async fn parse_xtream(input: &ConfigInput,
xtream_categories.into_iter().map(|category| (category.category_id, category)).collect();
let mut unknown_grp = XtreamCategory {
category_id: 0u32,
category_name: "Unknown".to_string(),
category_name: "Unknown".intern(),
channels: vec![],
};
@@ -165,19 +163,19 @@ pub async fn parse_xtream(input: &ConfigInput,
let item_type = PlaylistItemType::from(xtream_cluster);
let mut item = PlaylistItem {
header: PlaylistItemHeader {
id: stream.get_stream_id().to_string(),
id: stream.get_stream_id().intern(),
uuid: generate_playlist_uuid(&input_name, &stream.get_stream_id().to_string(), item_type, &stream_url),
name: stream.get_name().to_string(),
logo: stream.get_stream_icon().to_string(),
group: interner.intern(category_name),
title: stream.get_name().to_string(),
url: stream_url.to_string(),
epg_channel_id: stream.get_epg_channel_id(),
name: Arc::clone(&stream.get_name()),
logo: Arc::clone(&stream.get_stream_icon()),
group: Arc::clone(category_name),
title: Arc::clone(&stream.get_name()),
url: stream_url.clone(),
epg_channel_id: stream.get_epg_channel_id().clone(),
item_type,
xtream_cluster,
category_id: stream.get_category_id(),
additional_properties: Some(stream),
category_id: 0,
input_name: interner.intern(&input_name),
input_name: input_name.clone(),
..Default::default()
},
};
@@ -197,7 +195,7 @@ pub async fn parse_xtream(input: &ConfigInput,
PlaylistGroup {
id: category.category_id,
xtream_cluster,
title: interner.intern(&category.category_name),
title: Arc::clone(&category.category_name),
channels: category.channels.clone(),
}
}).collect()))
@@ -215,112 +213,115 @@ pub async fn parse_xtream_streaming<F>(
categories: DynReader,
streams: DynReader,
mut on_item: F,
) -> Result<Vec<XtreamCategory>, TuliproxError>
where
F: FnMut(XtreamPlaylistItem) -> Result<(), TuliproxError> + Send + 'static
) -> Result<Vec<XtreamCategory>, TuliproxError>
where
F: FnMut(XtreamPlaylistItem) -> Result<(), TuliproxError> + Send + 'static,
{
// 1. Parse Categories
let xtream_categories = map_to_xtream_category(categories).await?;
// 2. Prepare for Stream Parsing
let mut interner = StringInterner::new();
let input_name = input.name.clone();
let url = input.url.as_str().to_string();
let url = input.url.as_str().to_string();
let username = input.username.as_ref().map_or("", |v| v).to_string();
let password = input.password.as_ref().map_or("", |v| v).to_string();
let options = input.options.clone();
// Map categories for lookup
let group_map: IndexMap<u32, String> = xtream_categories.iter().map(|c| (c.category_id, c.category_name.clone())).collect();
let unknown_group_name = "Unknown".to_string();
let group_map: IndexMap<u32, Arc<str>> = xtream_categories.iter().map(|c| (c.category_id, c.category_name.clone())).collect();
let unknown_group_name = "Unknown".intern();
spawn_blocking(move || {
let reader = tokio_util::io::SyncIoBridge::new(streams);
let mut deserializer = serde_json::Deserializer::from_reader(reader);
let (live_stream_use_prefix, live_stream_without_extension) = options.as_ref()
.map_or((true, false), |o| (o.xtream_live_stream_use_prefix, o.xtream_live_stream_without_extension));
.map_or((true, false), |o| (o.xtream_live_stream_use_prefix, o.xtream_live_stream_without_extension));
let mut source_ordinal = 0u32;
match xtream_cluster {
XtreamCluster::Live => {
let mut on_stream = |stream: LiveStreamProperties| {
source_ordinal += 1;
let stream_prop = StreamProperties::Live(stream);
process_stream_item(&input_name, &url, &username, &password,
xtream_cluster, &mut interner, &group_map, &unknown_group_name,
stream_prop, &mut on_item, live_stream_use_prefix, live_stream_without_extension, source_ordinal)
};
let visitor = XtreamItemVisitor { on_item: &mut on_stream, _marker: std::marker::PhantomData };
deserializer.deserialize_any(visitor).map_err(|e| notify_err!("JSON parse error: {e}"))?;
}
XtreamCluster::Video => {
let mut on_stream = |stream: VideoStreamProperties| {
source_ordinal += 1;
let stream_prop = StreamProperties::Video(Box::new(stream));
process_stream_item(&input_name, &url, &username, &password,
xtream_cluster, &mut interner, &group_map, &unknown_group_name,
stream_prop, &mut on_item, live_stream_use_prefix, live_stream_without_extension, source_ordinal)
};
let visitor = XtreamItemVisitor { on_item: &mut on_stream, _marker: std::marker::PhantomData };
deserializer.deserialize_any(visitor).map_err(|e| notify_err!("JSON parse error: {e}"))?;
}
XtreamCluster::Series => {
let mut on_stream = |stream: SeriesStreamProperties| {
source_ordinal += 1;
let stream_prop = StreamProperties::Series(Box::new(stream));
process_stream_item(&input_name, &url, &username, &password,
xtream_cluster, &mut interner, &group_map, &unknown_group_name,
stream_prop, &mut on_item, live_stream_use_prefix, live_stream_without_extension, source_ordinal)
};
let visitor = XtreamItemVisitor { on_item: &mut on_stream, _marker: std::marker::PhantomData };
deserializer.deserialize_any(visitor).map_err(|e| notify_err!("JSON parse error: {e}"))?;
}
XtreamCluster::Live => {
let mut on_stream = |stream: LiveStreamProperties| {
source_ordinal += 1;
let stream_prop = StreamProperties::Live(stream);
process_stream_item(&input_name, &url, &username, &password,
xtream_cluster, &group_map, &unknown_group_name,
stream_prop, &mut on_item, live_stream_use_prefix, live_stream_without_extension, source_ordinal)
};
let visitor = XtreamItemVisitor { on_item: &mut on_stream, _marker: std::marker::PhantomData };
deserializer.deserialize_any(visitor).map_err(|e| notify_err!("JSON parse error: {e}"))?;
}
XtreamCluster::Video => {
let mut on_stream = |stream: VideoStreamProperties| {
source_ordinal += 1;
let stream_prop = StreamProperties::Video(Box::new(stream));
process_stream_item(&input_name, &url, &username, &password,
xtream_cluster, &group_map, &unknown_group_name,
stream_prop, &mut on_item, live_stream_use_prefix, live_stream_without_extension, source_ordinal)
};
let visitor = XtreamItemVisitor { on_item: &mut on_stream, _marker: std::marker::PhantomData };
deserializer.deserialize_any(visitor).map_err(|e| notify_err!("JSON parse error: {e}"))?;
}
XtreamCluster::Series => {
let mut on_stream = |stream: SeriesStreamProperties| {
source_ordinal += 1;
let stream_prop = StreamProperties::Series(Box::new(stream));
process_stream_item(&input_name, &url, &username, &password,
xtream_cluster, &group_map, &unknown_group_name,
stream_prop, &mut on_item, live_stream_use_prefix, live_stream_without_extension, source_ordinal)
};
let visitor = XtreamItemVisitor { on_item: &mut on_stream, _marker: std::marker::PhantomData };
deserializer.deserialize_any(visitor).map_err(|e| notify_err!("JSON parse error: {e}"))?;
}
}
Ok(())
}).await.map_err(|e| notify_err!("Streaming parse failed: {e}"))??;
Ok(xtream_categories)
Ok(xtream_categories)
}
#[allow(clippy::too_many_arguments)]
fn process_stream_item<F>(
input_name: &str, url: &str, username: &str, password: &str,
cluster: XtreamCluster, interner: &mut StringInterner,
group_map: &IndexMap<u32, String>, unknown_group_name: &str,
input_name: &Arc<str>,
url: &str, username: &str, password: &str,
cluster: XtreamCluster,
group_map: &IndexMap<u32, Arc<str>>,
unknown_group_name: &Arc<str>,
mut stream: StreamProperties,
callback: &mut F,
live_stream_use_prefix: bool, live_stream_without_extension: bool,
live_stream_use_prefix: bool,
live_stream_without_extension: bool,
source_ordinal: u32,
) -> Result<(), TuliproxError>
where F: FnMut(XtreamPlaylistItem) -> Result<(), TuliproxError>
where
F: FnMut(XtreamPlaylistItem) -> Result<(), TuliproxError>,
{
stream.prepare();
let category_id = stream.get_category_id();
let category_name = group_map.get(&category_id).map_or(unknown_group_name, String::as_str);
let stream_url = create_xtream_url(cluster, url, username, password, &stream, live_stream_use_prefix, live_stream_without_extension);
let item_type = PlaylistItemType::from(cluster);
let item = PlaylistItem {
header: PlaylistItemHeader {
id: stream.get_stream_id().to_string(),
uuid: generate_playlist_uuid(input_name, &stream.get_stream_id().to_string(), item_type, &stream_url),
name: stream.get_name().to_string(),
logo: stream.get_stream_icon().to_string(),
group: interner.intern(category_name),
title: stream.get_name().to_string(),
url: stream_url.to_string(),
epg_channel_id: stream.get_epg_channel_id(),
item_type,
xtream_cluster: cluster,
additional_properties: Some(stream),
category_id,
source_ordinal,
input_name: interner.intern(input_name),
..Default::default()
},
};
stream.prepare();
let category_id = stream.get_category_id();
let category_name = group_map.get(&category_id).unwrap_or(unknown_group_name);
let stream_url = create_xtream_url(cluster, url, username, password, &stream, live_stream_use_prefix, live_stream_without_extension);
let item_type = PlaylistItemType::from(cluster);
let item = PlaylistItem {
header: PlaylistItemHeader {
id: stream.get_stream_id().intern(),
uuid: generate_playlist_uuid(input_name, &stream.get_stream_id().to_string(), item_type, &stream_url),
name: stream.get_name(),
logo: stream.get_stream_icon(),
group: category_name.clone(),
title: stream.get_name(),
url: stream_url,
epg_channel_id: stream.get_epg_channel_id(),
item_type,
xtream_cluster: cluster,
additional_properties: Some(stream),
category_id,
source_ordinal,
input_name: Arc::clone(input_name),
..Default::default()
},
};
// if let Some(StreamProperties::Series(props)) = item.header.additional_properties.as_mut() {
// // We need to set category_id for Series properties just like parse_xtream might expect or use?
@@ -329,7 +330,7 @@ where F: FnMut(XtreamPlaylistItem) -> Result<(), TuliproxError>
// let _ = props;
// }
callback(XtreamPlaylistItem::from(&item))
callback(XtreamPlaylistItem::from(&item))
}
struct XtreamItemVisitor<'a, T, F> {