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

stream properties refactored

Browse Source
This commit is contained in:
euzu
2025-12-26 14:09:44 +01:00
parent 76bde0fcc4
commit d5207e886f
102 changed files with 7817 additions and 2546 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/library.rs
+12 -8
View File
@@ -1,12 +1,12 @@
use crate::library::{MediaMetadata, MetadataAsyncIter, MetadataCacheEntry};
use crate::model::{AppConfig, ConfigInput};
use shared::error::TuliproxError;
use shared::model::{EpisodeStreamProperties, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, SeriesStreamProperties, StreamProperties, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster};
use shared::utils::{generate_playlist_uuid, string_to_uuid_type};
use shared::model::{EpisodeStreamProperties, PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, SeriesStreamDetailEpisodeProperties, SeriesStreamDetailProperties, SeriesStreamProperties, StreamProperties, UUIDType, VideoStreamDetailProperties, VideoStreamProperties, XtreamCluster};
use shared::utils::{generate_playlist_uuid};
use std::path::Path;
use std::sync::Arc;
pub async fn get_library_playlist(_client: &reqwest::Client, app_config: &Arc<AppConfig>, input: &Arc<ConfigInput>) -> (Vec<PlaylistGroup>, Vec<TuliproxError>) {
pub async fn download_library_playlist(_client: &reqwest::Client, app_config: &Arc<AppConfig>, input: &Arc<ConfigInput>) -> (Vec<PlaylistGroup>, Vec<TuliproxError>) {
let config = &*app_config.config.load();
let Some(library_config) = config.library.as_ref() else { return (vec![], vec![]) };
if !library_config.enabled { return (vec![], vec![]); }
@@ -57,10 +57,11 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
let additional_properties = metadata_cache_entry_to_xtream_movie_info(entry);
vec![PlaylistItem {
header: PlaylistItemHeader {
uuid: string_to_uuid_type(&entry.uuid),
uuid: UUIDType::from_valid_uuid(&entry.uuid),
name: metadata.title().to_string(),
group: group_name.to_string(),
title: metadata.title().to_string(),
logo: metadata.poster().map_or_else(String::new, ToString::to_string),
url: format!("file://{}", entry.file_path),
xtream_cluster: XtreamCluster::Video,
additional_properties,
@@ -78,6 +79,7 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
if let Some(details_props) = series_properties.details.as_ref() {
if let Some(prop_episodes) = details_props.episodes.as_ref() {
for episode in prop_episodes {
let logo = if episode.movie_image.is_empty() { metadata.poster().map_or_else(String::new, ToString::to_string) } else { episode.movie_image.clone() };
let container_extension = Path::new(&episode.direct_source)
.extension()
.and_then(|s| s.to_str())
@@ -88,6 +90,7 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
// we use parent_code for local series to find the parent series info and straighten the virtual_ids
parent_code: entry.uuid.clone(),
uuid: generate_playlist_uuid(input_name, &episode.id.to_string(), PlaylistItemType::LocalSeries, &episode.direct_source),
logo: logo.clone(),
name: episode.title.clone(),
group: group_name.to_string(),
title: episode.title.clone(),
@@ -103,7 +106,7 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
added: Some(episode.added.clone()),
release_date: Some(episode.release_date.clone()),
tmdb: episode.tmdb,
movie_image: episode.movie_image.clone(),
movie_image: logo,
container_extension,
audio: None,
video: None,
@@ -118,11 +121,12 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
let series_info = PlaylistItem {
header: PlaylistItemHeader {
uuid: string_to_uuid_type(&entry.uuid),
uuid: UUIDType::from_valid_uuid(&entry.uuid),
id: entry.uuid.clone(),
name: metadata.title().to_string(),
group: group_name.to_string(),
title: metadata.title().to_string(),
logo: metadata.poster().map_or_else(String::new, ToString::to_string),
url: format!("file://{}", entry.file_path),
xtream_cluster: XtreamCluster::Series,
item_type: PlaylistItemType::LocalSeriesInfo,
@@ -167,7 +171,7 @@ pub fn metadata_cache_entry_to_xtream_movie_info(
container_extension,
rating: movie.rating,
rating_5based: None,
stream_type: "movie".to_string(),
stream_type: Some("movie".to_string()),
trailer: movie.videos.as_ref().and_then(|v| v.iter().find(|video| video.site.eq_ignore_ascii_case("youtube")).map(|video| video.key.clone())),
tmdb: movie.tmdb_id,
is_adult: 0,
@@ -246,7 +250,7 @@ pub fn metadata_cache_entry_to_xtream_series_info(
season: episode.season,
title: episode.title.clone(),
container_extension,
custom_sid: String::new(),
custom_sid: None,
added: episode.file_modified.to_string(),
direct_source: episode.file_path.clone(),
tmdb: tmdb_id,
backend/src/processing/processor/mod.rs
-22
View File
@@ -8,28 +8,6 @@ mod sort;
pub mod trakt;
mod library;
#[macro_export]
macro_rules! handle_error {
($stmt:expr, $map_err:expr) => {
if let Err(err) = $stmt {
$map_err(err);
}
};
}
use handle_error;
#[macro_export]
macro_rules! handle_error_and_return {
($stmt:expr, $map_err:expr) => {
if let Err(err) = $stmt {
$map_err(err);
return Default::default();
}
};
}
use handle_error_and_return;
//
// fn get_resolve_<cluster>_options(target: &ConfigTarget, fpl: &FetchedPlaylist) -> (bool, u16)
//
backend/src/processing/processor/playlist.rs
+20 -10
View File
@@ -1,8 +1,7 @@
use crate::model::{AppConfig, ConfigFavourites, ConfigInput, ConfigRename};
use crate::model::{AppConfig, Config, ConfigFavourites, ConfigInput, ConfigRename};
use crate::utils::epg;
use crate::utils::m3u;
use crate::utils::xtream;
use crate::Config;
use std::collections::{HashMap, HashSet};
use std::path::PathBuf;
use std::sync::Arc;
@@ -24,7 +23,7 @@ use crate::processing::processor::sort::sort_playlist;
use crate::processing::processor::trakt::process_trakt_categories_for_target;
use crate::processing::processor::xtream_series::playlist_resolve_series;
use crate::processing::processor::xtream_vod::playlist_resolve_vod;
use crate::repository::playlist_repository::persist_playlist;
use crate::repository::playlist_repository::{persist_input_playlist, persist_playlist};
use crate::utils::StepMeasure;
use crate::utils::{debug_if_enabled, trace_if_enabled};
use deunicode::deunicode;
@@ -173,7 +172,7 @@ fn rename_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Opt
fn create_alias_uuid(base_uuid: &UUIDType, mapping_id: &str) -> UUIDType {
let mut data = Vec::with_capacity(base_uuid.len() + mapping_id.len());
data.extend_from_slice(base_uuid);
data.extend_from_slice(base_uuid.as_ref());
data.extend_from_slice(mapping_id.as_bytes());
hash_bytes(&data)
}
@@ -309,10 +308,10 @@ fn is_target_enabled(target: &ConfigTarget, user_targets: &ProcessTargets) -> bo
async fn playlist_download_from_input(client: &reqwest::Client, app_config: &Arc<AppConfig>, input: &Arc<ConfigInput>) -> (Vec<PlaylistGroup>, Vec<TuliproxError>) {
let config = &*app_config.config.load();
match input.input_type {
InputType::M3u => m3u::get_m3u_playlist(client, config, input).await,
InputType::Xtream => xtream::get_xtream_playlist(config, client, input).await,
InputType::M3u => m3u::download_m3u_playlist(client, config, input).await,
InputType::Xtream => xtream::download_xtream_playlist(config, client, input).await,
InputType::M3uBatch | InputType::XtreamBatch => (vec![], vec![]),
InputType::Library => library::get_library_playlist(client, app_config, input).await,
InputType::Library => library::download_library_playlist(client, app_config, input).await,
}
}
@@ -332,13 +331,25 @@ async fn process_source(client: &reqwest::Client, app_config: Arc<AppConfig>, so
if is_input_enabled(input, &user_targets) {
source_downloaded = true;
let start_time = Instant::now();
let (mut playlistgroups, mut error_list) = playlist_download_from_input(client, &app_config, input).await;
// Download playlist for input
let (playlistgroups, mut error_list) = {
let (downloaded_playlist, mut download_err) = playlist_download_from_input(client, &app_config, input).await;
let (playlist, error) = persist_input_playlist(&app_config, input, downloaded_playlist).await;
error!("Failed to persist input playlist {}", input.name);
if let Some(err) = error {
download_err.push(err);
}
(playlist, download_err)
};
// Download epg for input
let (tvguide, mut tvguide_errors) = if error_list.is_empty() {
let working_dir = &app_config.config.load().working_dir;
epg::get_xmltv(client, input, working_dir).await
} else {
(None, vec![])
};
errors.append(&mut error_list);
errors.append(&mut tvguide_errors);
let group_count = playlistgroups.len();
@@ -348,7 +359,6 @@ async fn process_source(client: &reqwest::Client, app_config: Arc<AppConfig>, so
info!("Source is empty {input_name}");
errors.push(notify_err!(format!("Source is empty {input_name}")));
} else {
playlistgroups.iter_mut().for_each(PlaylistGroup::on_load);
source_playlists.push(
FetchedPlaylist {
input,
@@ -533,7 +543,7 @@ fn flatten_groups(playlistgroups: Vec<PlaylistGroup>) -> Vec<PlaylistGroup> {
}
#[allow(clippy::too_many_arguments)]
async fn process_playlist_for_target(app_config: &AppConfig,
async fn process_playlist_for_target(app_config: &Arc<AppConfig>,
client: &reqwest::Client,
playlists: &mut [FetchedPlaylist<'_>],
target: &ConfigTarget,
backend/src/processing/processor/trakt.rs
+1 -10
View File
@@ -36,15 +36,6 @@ fn is_compatible_content_type(cluster: XtreamCluster, content_type: TraktContent
}
}
/// Extract TMDB ID from playlist item
fn extract_tmdb_id_from_playlist_item(item: &PlaylistItem) -> Option<u32> {
if let Some(props) = &item.header.additional_properties {
props.get_tmdb_id()
} else {
None
}
}
fn calculate_year_bonus(playlist_year: Option<u32>, trakt_year: Option<u32>) -> f64 {
if let (Some(p_year), Some(t_year)) = (playlist_year, trakt_year) {
if p_year == t_year {
@@ -218,7 +209,7 @@ fn match_trakt_items_with_playlist<'a>(
if is_compatible_content_type(channel.header.xtream_cluster, list_config.content_type) {
let normalized_title = normalize_title_for_matching(&channel.header.title);
let channel_year = extract_year_from_title(&channel.header.title);
let channel_tmdb_id = extract_tmdb_id_from_playlist_item(channel);
let channel_tmdb_id = channel.get_tmdb_id();
if let Some(matched) = find_best_match_for_item((channel, normalized_title, channel_year, channel_tmdb_id), &trakt_match_items, list_config) {
matches.push(matched);
}
backend/src/processing/processor/xtream.rs
+4 -84
View File
@@ -1,16 +1,7 @@
use shared::error::{info_err, notify_err};
use shared::error::{str_to_io_error, TuliproxError};
use crate::model::{AppConfig, Config, ConfigInput, InputSource};
use crate::model::{FetchedPlaylist};
use shared::model::{PlaylistEntry, PlaylistItem, PlaylistItemType, XtreamCluster};
use crate::repository::storage::get_input_storage_path;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::path::PathBuf;
use crate::repository::bplustree::BPlusTree;
use crate::repository::storage_const;
use crate::repository::xtream_repository::xtream_get_record_file_path;
use crate::utils;
use shared::error::{info_err};
use shared::error::{TuliproxError};
use crate::model::{ConfigInput, InputSource};
use shared::model::{PlaylistEntry, PlaylistItem, XtreamCluster};
use crate::utils::xtream;
pub(in crate::processing) async fn playlist_resolve_download_playlist_item(client: &reqwest::Client, pli: &PlaylistItem, input: &ConfigInput, errors: &mut Vec<TuliproxError>, resolve_delay: u16, cluster: XtreamCluster) -> Option<String> {
@@ -31,74 +22,3 @@ pub(in crate::processing) async fn playlist_resolve_download_playlist_item(clien
}
result
}
pub(in crate::processing) async fn create_resolve_episode_wal_files(cfg: &Config, input: &ConfigInput) -> Option<(tokio::fs::File, PathBuf)> {
match get_input_storage_path(&input.name, &cfg.working_dir) {
Ok(storage_path) => {
let info_path = storage_path.join(format!("{}.{}", crate::model::XC_FILE_SERIES_EPISODE_RECORD, storage_const::FILE_SUFFIX_WAL));
let info_file = utils::async_append_or_create_file(&info_path).await.ok()?;
Some((info_file, info_path))
}
Err(_) => None
}
}
pub(in crate::processing) async fn create_resolve_info_wal_files(cfg: &Config, input: &ConfigInput, cluster: XtreamCluster) -> Option<(tokio::fs::File, tokio::fs::File, PathBuf, PathBuf)> {
match get_input_storage_path(&input.name, &cfg.working_dir) {
Ok(storage_path) => {
if let Some(file_prefix) = match cluster {
XtreamCluster::Live => None,
XtreamCluster::Video => Some(crate::model::XC_FILE_VOD_INFO),
XtreamCluster::Series => Some(crate::model::XC_FILE_SERIES_INFO)
} {
let content_path = storage_path.join(format!("{file_prefix}_content.{}", storage_const::FILE_SUFFIX_WAL));
let info_path = storage_path.join(format!("{file_prefix}_record.{}", storage_const::FILE_SUFFIX_WAL));
let content_file = utils::async_append_or_create_file(&content_path).await.ok()?;
let info_file = utils::async_append_or_create_file(&info_path).await.ok()?;
return Some((content_file, info_file, content_path, info_path));
}
None
}
Err(_) => None
}
}
pub(in crate::processing) fn should_update_info(pli: &mut PlaylistItem, processed_provider_ids: &HashMap<u32, u64>) -> (bool, u32, u64) {
let Some(provider_id) = pli.header.get_provider_id() else { return (false, 0, 0) };
let last_modified = pli.header.additional_properties.as_ref().and_then(shared::model::StreamProperties::get_last_modified);
let old_timestamp = processed_provider_ids.get(&provider_id);
(old_timestamp.is_none()
|| last_modified.is_none()
|| *old_timestamp.unwrap() != last_modified.unwrap(), provider_id, last_modified.unwrap_or(0))
}
pub(in crate::processing) async fn read_processed_info_ids<V, F>(cfg: &AppConfig, errors: &mut Vec<TuliproxError>, fpl: &FetchedPlaylist<'_>,
item_type: PlaylistItemType, extract_ts: F) -> HashMap<u32, u64>
where
F: Fn(&V) -> u64,
V: Serialize + for<'de> Deserialize<'de> + Clone,
{
let mut processed_info_ids = HashMap::new();
let fpl_name = &fpl.input.name;
let file_path = match get_input_storage_path(fpl_name, &cfg.config.load().working_dir)
.map(|storage_path| xtream_get_record_file_path(&storage_path, item_type)).and_then(|opt| opt.ok_or_else(|| str_to_io_error("Not supported")))
{
Ok(file_path) => file_path,
Err(err) => {
errors.push(notify_err!(format!("Could not create storage path for input {fpl_name}: {err}")));
return processed_info_ids;
}
};
{
let file_lock = cfg.file_locks.read_lock(&file_path).await;
if let Ok(info_records) = BPlusTree::<u32, V>::load(&file_path) {
info_records.iter().for_each(|(provider_id, record)| {
processed_info_ids.insert(*provider_id, extract_ts(record));
});
}
drop(file_lock);
}
processed_info_ids
}
backend/src/processing/processor/xtream_series.rs
+139 -195
View File
@@ -2,105 +2,67 @@ use crate::model::{AppConfig, ConfigTarget};
use crate::model::FetchedPlaylist;
use crate::processing::parser::xtream::parse_xtream_series_info;
use crate::processing::processor::playlist::ProcessingPipe;
use crate::processing::processor::xtream::{create_resolve_episode_wal_files, create_resolve_info_wal_files, playlist_resolve_download_playlist_item, read_processed_info_ids, should_update_info};
use crate::processing::processor::{create_resolve_options_function_for_xtream_target, handle_error, handle_error_and_return};
use crate::repository::storage::get_input_storage_path;
use crate::repository::xtream_repository::{write_series_info_to_wal_file, xtream_get_info_file_paths,
xtream_update_input_info_file, xtream_update_input_series_episodes_record_from_wal_file,
xtream_update_input_series_record_from_wal_file};
use crate::repository::IndexedDocumentReader;
use crate::utils;
use crate::utils::{bincode_serialize, IO_BUFFER_SIZE};
use log::{error, info, log_enabled, warn, Level};
use shared::error::{info_err, notify_err};
use crate::processing::processor::xtream::{playlist_resolve_download_playlist_item};
use crate::processing::processor::{create_resolve_options_function_for_xtream_target};
use log::{error, info, log_enabled, Level};
use shared::error::TuliproxError;
use shared::model::{InputType, SeriesStreamProperties, StreamProperties, XtreamSeriesInfo};
use shared::model::{InputType, PlaylistEntry, SeriesStreamProperties, StreamProperties, XtreamSeriesInfo};
use shared::model::{PlaylistGroup, PlaylistItemType, XtreamCluster};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::Instant;
use tokio::io::AsyncWriteExt;
use crate::repository::storage::get_input_storage_path;
use crate::repository::xtream_repository::persists_input_series_info;
create_resolve_options_function_for_xtream_target!(series);
async fn read_processed_series_info_ids(cfg: &AppConfig, errors: &mut Vec<TuliproxError>, fpl: &FetchedPlaylist<'_>) -> HashMap<u32, u64> {
read_processed_info_ids(cfg, errors, fpl, PlaylistItemType::SeriesInfo, |ts: &u64| *ts).await
}
async fn playlist_resolve_series_info(app_config: &Arc<AppConfig>, client: &reqwest::Client, errors: &mut Vec<TuliproxError>,
fpl: &mut FetchedPlaylist<'_>, resolve_delay: u16) -> Vec<PlaylistGroup> {
async fn write_series_episode_record_to_wal_file(
writer: &mut tokio::io::BufWriter<&mut tokio::fs::File>,
provider_id: u32,
additional_properties: Option<&StreamProperties>,
) -> std::io::Result<usize> {
if let Some(StreamProperties::Episode(series_episode)) = additional_properties {
if let Ok(content_bytes) = bincode_serialize(&series_episode) {
writer.write_all(&provider_id.to_le_bytes()).await?;
let content_len = content_bytes.len();
if let Ok(len) = u32::try_from(content_len) {
writer.write_all(&len.to_le_bytes()).await?;
writer.write_all(&content_bytes).await?;
return Ok(content_len + 4usize);
}
error!("Cant write to WAL file, content length exceeds u32");
let input = fpl.input;
let working_dir = &app_config.config.load().working_dir;
let storage_path = match get_input_storage_path(&input.name, working_dir) {
Ok(storage_path) => storage_path,
Err(err) => {
error!("Can't resolve vod, input storage directory for input '{}' failed: {err}", input.name);
return vec![];
}
}
Ok(0)
}
async fn playlist_resolve_series_info(cfg: &AppConfig, client: &reqwest::Client, errors: &mut Vec<TuliproxError>,
fpl: &mut FetchedPlaylist<'_>, resolve_delay: u16) -> bool {
// TODO read existing WAL File and import it to avoid duplicate requests
let mut processed_info_ids: HashMap<u32, u64> = read_processed_series_info_ids(cfg, errors, fpl).await;
let mut fetched_in_run: HashSet<u32> = HashSet::new();
// we can't write to the indexed-document directly because of the write lock and time-consuming operation.
// All readers would be waiting for the lock and the app would be unresponsive.
// We collect the content into a wal file and write it once we collected everything.
let Some((mut wal_content_file, mut wal_record_file, wal_content_path, wal_record_path))
= create_resolve_info_wal_files(&cfg.config.load(), fpl.input, XtreamCluster::Series).await
else { return !processed_info_ids.is_empty(); };
let mut content_writer = utils::async_file_writer(&mut wal_content_file);
let mut record_writer = utils::async_file_writer(&mut wal_record_file);
let mut content_updated = false;
};
let series_info_count = fpl.playlistgroups.iter()
.filter(|&plg| plg.xtream_cluster == XtreamCluster::Series)
.flat_map(|plg| &plg.channels)
.filter(|&pli| pli.header.item_type == PlaylistItemType::SeriesInfo).count();
.filter(|&pli| pli.header.xtream_cluster == XtreamCluster::Series
&& pli.header.item_type == PlaylistItemType::SeriesInfo
&& !pli.has_details()).count();
info!("Found {series_info_count} series info to resolve");
let mut last_log_time = Instant::now();
let mut processed_series_info_count = 0;
let mut write_counter = 0usize;
// LocalSeriesInfo entries are not resolved!
let input = fpl.input;
let mut result: Vec<PlaylistGroup> = vec![];
for plg in &mut fpl.playlistgroups {
let mut group_series = vec![];
for pli in &mut plg.channels {
if pli.header.item_type != PlaylistItemType::SeriesInfo {
processed_series_info_count += 1;
if pli.header.xtream_cluster != XtreamCluster::Series
|| pli.header.item_type != PlaylistItemType::SeriesInfo
|| pli.has_details() {
continue;
}
let (should_update, provider_id, ts) = should_update_info(pli, &processed_info_ids);
if should_update && provider_id != 0 && fetched_in_run.insert(provider_id) {
let Some(provider_id) = pli.get_provider_id() else { continue; };
let (group, series_name) = {
let header = &pli.header;
(header.group.clone(), if header.name.is_empty() { header.title.clone() } else { header.name.clone() })
};
if provider_id != 0 {
if let Some(content) = playlist_resolve_download_playlist_item(client, pli, fpl.input, errors, resolve_delay, XtreamCluster::Series).await {
if let Ok(info) = serde_json::from_str::<XtreamSeriesInfo>(&content) {
let series_stream_props = SeriesStreamProperties::from_info(&info, pli);
handle_error_and_return!(write_series_info_to_wal_file(provider_id, ts, &serde_json::to_string(&series_stream_props).unwrap_or_else(|_|String::new()),
&mut content_writer, &mut record_writer).await,
|err| errors.push(notify_err!(format!("Failed to resolve series, could not write to wal file {err}"))));
processed_info_ids.insert(provider_id, ts);
content_updated = true;
write_counter += content.len();
// periodic flush to bound BufWriter memory
if write_counter >= IO_BUFFER_SIZE {
write_counter = 0;
if let Err(err) = content_writer.flush().await {
errors.push(notify_err!(format!("Failed periodic flush of wal content writer {err}")));
}
if let Err(err) = record_writer.flush().await {
errors.push(notify_err!(format!("Failed periodic flush of wal record writer {err}")));
}
// the input db needs to be updated
let _ = persists_input_series_info(app_config, &storage_path, pli.header.xtream_cluster, &input.name, provider_id, &series_stream_props).await;
// extract episodes from info
if let Some(episodes) = parse_xtream_series_info(&pli.get_uuid(), &series_stream_props, &group, &series_name, input) {
group_series.extend(episodes.into_iter());
}
// Update in-memory playlist items with the newly fetched vod info.
@@ -109,116 +71,11 @@ async fn playlist_resolve_series_info(cfg: &AppConfig, client: &reqwest::Client,
}
}
}
if log_enabled!(Level::Info) {
processed_series_info_count += 1;
if last_log_time.elapsed().as_secs() >= 30 {
if log_enabled!(Level::Info)
&& last_log_time.elapsed().as_secs() >= 30 {
info!("resolved {processed_series_info_count}/{series_info_count} series info");
last_log_time = Instant::now();
}
}
}
}
info!("resolved {processed_series_info_count}/{series_info_count} series info");
// content_wal contains the provider_id and series_info with episode listing
// record_wal contains provider_id and timestamp
if content_updated {
handle_error!(content_writer.flush().await,
|err| errors.push(notify_err!(format!("Failed to resolve series, could not write to wal file {err}"))));
handle_error!(record_writer.flush().await,
|err| errors.push(notify_err!(format!("Failed to resolve series tmdb, could not write to wal file {err}"))));
handle_error!(content_writer.get_ref().sync_all().await, |err| errors.push(notify_err!(format!("Failed to sync series info to wal file {err}"))));
handle_error!(record_writer.get_ref().sync_all().await, |err| errors.push(notify_err!(format!("Failed to sync series info record to wal file {err}"))));
drop(content_writer);
drop(record_writer);
drop(wal_content_file);
drop(wal_record_file);
handle_error!(xtream_update_input_info_file(cfg, fpl.input, &wal_content_path, XtreamCluster::Series).await,
|err| errors.push(err));
handle_error!(xtream_update_input_series_record_from_wal_file(cfg, fpl.input, &wal_record_path).await,
|err| errors.push(err));
}
// we updated now
// - series_info.db which contains the original series_info json
// - series_record.db which contains the series_info provider_id and timestamp
!processed_info_ids.is_empty()
}
async fn process_series_info(
app_config: &AppConfig,
fpl: &mut FetchedPlaylist<'_>,
errors: &mut Vec<TuliproxError>,
) -> Vec<PlaylistGroup> {
let mut result: Vec<PlaylistGroup> = vec![];
let input = fpl.input;
let config = app_config.config.load();
let Ok(Some((info_path, idx_path))) = get_input_storage_path(&input.name, &config.working_dir)
.map(|storage_path| xtream_get_info_file_paths(&storage_path, XtreamCluster::Series))
else {
errors.push(notify_err!("Failed to open input info file for series".to_string()));
return result;
};
let mut write_counter = 0usize;
let _file_lock = app_config.file_locks.read_lock(&info_path).await;
// Contains the Series Info with episode listing
let Ok(mut info_reader) = IndexedDocumentReader::<u32, String>::new(&info_path, &idx_path) else { return result; };
let Some((mut wal_file, wal_path)) = create_resolve_episode_wal_files(&config, input).await else {
errors.push(notify_err!("Could not create wal file for series episodes record".to_string()));
return result;
};
let mut wal_writer = utils::async_file_writer(&mut wal_file);
for plg in fpl
.playlistgroups
.iter_mut()
.filter(|plg| plg.xtream_cluster == XtreamCluster::Series)
{
let mut group_series = vec![];
// Resolve does not handle LocalSeriesInfo
for pli in plg
.channels
.iter_mut()
.filter(|pli| pli.header.item_type == PlaylistItemType::SeriesInfo)
{
let Some(provider_id) = pli.header.get_provider_id() else { continue; };
let Ok(content) = info_reader.get(&provider_id) else { continue; };
if content.is_empty() {
warn!("Series info content is empty, skipping series with provider id: {provider_id}");
continue;
}
match serde_json::from_str::<SeriesStreamProperties>(&content) {
Ok(series_content) => {
let (group, series_name) = {
let header = &pli.header;
(header.group.clone(), if header.name.is_empty() { header.title.clone() } else { header.name.clone() })
};
if let Some(mut series) = parse_xtream_series_info(&series_content, &group, &series_name, input) {
for pli_episode in &mut series {
let Some(provider_id) = &pli_episode.header.get_provider_id() else { continue; };
match write_series_episode_record_to_wal_file(&mut wal_writer, *provider_id, pli.header.additional_properties.as_ref()).await {
Ok(written_bytes) => {
write_counter += written_bytes;
// periodic flush to bound BufWriter memory
if write_counter >= IO_BUFFER_SIZE {
write_counter = 0;
if let Err(err) = wal_writer.flush().await {
errors.push(notify_err!(format!("Failed periodic flush of wal content writer {err}")));
}
}
}
Err(err) => { errors.push(info_err!(format!("Failed to write to series episode wal file: {err}"))) }
}
}
group_series.extend(series.into_iter());
}
}
Err(err) => errors.push(info_err!(format!("Failed to parse JSON: {err}"))),
}
}
if !group_series.is_empty() {
result.push(PlaylistGroup {
@@ -229,19 +86,107 @@ async fn process_series_info(
});
}
}
handle_error!(wal_writer.flush().await, |err| errors.push(notify_err!(format!("Failed to resolve series episodes, could not write to wal file {err}"))));
handle_error!(wal_writer.get_ref().sync_all().await, |err| errors.push(notify_err!(format!("Failed to sync series info to wal file {err}"))));
drop(wal_writer);
drop(wal_file);
handle_error!(xtream_update_input_series_episodes_record_from_wal_file(app_config, input, &wal_path).await,
|err| errors.push(err));
info!("resolved {processed_series_info_count}/{series_info_count} series info");
result
}
//
// async fn process_series_info(
// app_config: &AppConfig,
// fpl: &mut FetchedPlaylist<'_>,
// errors: &mut Vec<TuliproxError>,
// ) -> Vec<PlaylistGroup> {
// let mut result: Vec<PlaylistGroup> = vec![];
// let input = fpl.input;
// let config = app_config.config.load();
// let Ok(Some((info_path, idx_path))) = get_input_storage_path(&input.name, &config.working_dir)
// .map(|storage_path| xtream_get_info_file_paths(&storage_path, XtreamCluster::Series))
// else {
// errors.push(notify_err!("Failed to open input info file for series".to_string()));
// return result;
// };
//
// let mut write_counter = 0usize;
//
// let _file_lock = app_config.file_locks.read_lock(&info_path).await;
//
// // Contains the Series Info with episode listing
// let Ok(mut info_reader) = IndexedDocumentReader::<u32, String>::new(&info_path, &idx_path) else { return result; };
//
// let Some((mut wal_file, wal_path)) = create_resolve_episode_wal_files(&config, input).await else {
// errors.push(notify_err!("Could not create wal file for series episodes record".to_string()));
// return result;
// };
// let mut wal_writer = utils::async_file_writer(&mut wal_file);
//
// for plg in fpl
// .playlistgroups
// .iter_mut()
// .filter(|plg| plg.xtream_cluster == XtreamCluster::Series)
// {
// let mut group_series = vec![];
//
// // Resolve does not handle LocalSeriesInfo
// for pli in plg
// .channels
// .iter_mut()
// .filter(|pli| pli.header.item_type == PlaylistItemType::SeriesInfo)
// {
// let Some(provider_id) = pli.header.get_provider_id() else { continue; };
// let Ok(content) = info_reader.get(&provider_id) else { continue; };
// if content.is_empty() {
// warn!("Series info content is empty, skipping series with provider id: {provider_id}");
// continue;
// }
// match serde_json::from_str::<SeriesStreamProperties>(&content) {
// Ok(series_content) => {
// let (group, series_name) = {
// let header = &pli.header;
// (header.group.clone(), if header.name.is_empty() { header.title.clone() } else { header.name.clone() })
// };
// if let Some(mut series) = parse_xtream_series_info(&series_content, &group, &series_name, input) {
// for pli_episode in &mut series {
// let Some(provider_id) = &pli_episode.header.get_provider_id() else { continue; };
// match write_series_episode_record_to_wal_file(&mut wal_writer, *provider_id, pli.header.additional_properties.as_ref()).await {
// Ok(written_bytes) => {
// write_counter += written_bytes;
// // periodic flush to bound BufWriter memory
// if write_counter >= IO_BUFFER_SIZE {
// write_counter = 0;
// if let Err(err) = wal_writer.flush().await {
// errors.push(notify_err!(format!("Failed periodic flush of wal content writer {err}")));
// }
// }
// }
// Err(err) => { errors.push(info_err!(format!("Failed to write to series episode wal file: {err}"))) }
// }
// }
// group_series.extend(series.into_iter());
// }
// }
// Err(err) => errors.push(info_err!(format!("Failed to parse JSON: {err}"))),
// }
// }
// if !group_series.is_empty() {
// result.push(PlaylistGroup {
// id: plg.id,
// title: plg.title.clone(),
// channels: group_series,
// xtream_cluster: XtreamCluster::Series,
// });
// }
// }
//
// handle_error!(wal_writer.flush().await, |err| errors.push(notify_err!(format!("Failed to resolve series episodes, could not write to wal file {err}"))));
// handle_error!(wal_writer.get_ref().sync_all().await, |err| errors.push(notify_err!(format!("Failed to sync series info to wal file {err}"))));
//
// drop(wal_writer);
// drop(wal_file);
// handle_error!(xtream_update_input_series_episodes_record_from_wal_file(app_config, input, &wal_path).await,
// |err| errors.push(err));
// result
// }
pub async fn playlist_resolve_series(cfg: &AppConfig,
pub async fn playlist_resolve_series(cfg: &Arc<AppConfig>,
client: &reqwest::Client,
target: &ConfigTarget,
errors: &mut Vec<TuliproxError>,
@@ -252,8 +197,7 @@ pub async fn playlist_resolve_series(cfg: &AppConfig,
let (resolve_series, resolve_delay) = get_resolve_series_options(target, processed_fpl);
if !resolve_series { return; }
if !playlist_resolve_series_info(cfg, client, errors, processed_fpl, resolve_delay).await { return; }
let series_playlist = process_series_info(cfg, provider_fpl, errors).await;
let series_playlist = playlist_resolve_series_info(cfg, client, errors, processed_fpl, resolve_delay).await;
if series_playlist.is_empty() { return; }
// original content saved into original list
for plg in &series_playlist {
backend/src/processing/processor/xtream_vod.rs
+30 -93
View File
@@ -1,131 +1,68 @@
use std::sync::Arc;
use crate::model::FetchedPlaylist;
use crate::model::{AppConfig, ConfigTarget};
use crate::processing::processor::xtream::{create_resolve_info_wal_files, playlist_resolve_download_playlist_item, read_processed_info_ids, should_update_info};
use crate::processing::processor::{create_resolve_options_function_for_xtream_target, handle_error, handle_error_and_return};
use crate::repository::xtream_repository::{write_vod_info_to_wal_file, xtream_update_input_info_file, xtream_update_input_vod_record_from_wal_file, InputVodInfoRecord};
use crate::utils;
use crate::utils::IO_BUFFER_SIZE;
use log::{info, log_enabled, Level};
use shared::error::notify_err;
use crate::processing::processor::xtream::{playlist_resolve_download_playlist_item};
use crate::processing::processor::{create_resolve_options_function_for_xtream_target};
use crate::repository::xtream_repository::{persists_input_vod_info};
use log::{error, info, log_enabled, Level};
use shared::error::TuliproxError;
use shared::model::{InputType, PlaylistEntry, StreamProperties, VideoStreamProperties, XtreamVideoInfo};
use shared::model::{PlaylistItemType, XtreamCluster};
use std::collections::{HashMap, HashSet};
use std::time::Instant;
use tokio::io::AsyncWriteExt;
use crate::repository::storage::get_input_storage_path;
create_resolve_options_function_for_xtream_target!(vod);
async fn read_processed_vod_info_ids(cfg: &AppConfig, errors: &mut Vec<TuliproxError>, fpl: &FetchedPlaylist<'_>) -> HashMap<u32, u64> {
read_processed_info_ids(cfg, errors, fpl, PlaylistItemType::Video, |record: &InputVodInfoRecord| record.ts).await
}
pub async fn playlist_resolve_vod(app_config: &AppConfig, client: &reqwest::Client, target: &ConfigTarget, errors: &mut Vec<TuliproxError>, fpl: &mut FetchedPlaylist<'_>) {
pub async fn playlist_resolve_vod(app_config: &Arc<AppConfig>, client: &reqwest::Client, target: &ConfigTarget, errors: &mut Vec<TuliproxError>, fpl: &mut FetchedPlaylist<'_>) {
let (resolve_movies, resolve_delay) = get_resolve_vod_options(target, fpl);
if !resolve_movies { return; }
// TODO read existing WAL File and import it to avoid duplicate requests
// we cant write to the indexed-document directly because of the write lock and time-consuming operation.
// All readers would be waiting for the lock and the app would be unresponsive.
// We collect the content into a wal file and write it once we collected everything.
let config = app_config.config.load();
let Some((mut wal_content_file, mut wal_record_file, wal_content_path, wal_record_path))
= create_resolve_info_wal_files(&config, fpl.input, XtreamCluster::Video).await
else { return; };
let mut processed_info_ids: HashMap<u32, u64> = read_processed_vod_info_ids(app_config, errors, fpl).await;
let mut fetched_in_run: HashSet<u32> = HashSet::new();
let mut content_writer = utils::async_file_writer(&mut wal_content_file);
let mut record_writer = utils::async_file_writer(&mut wal_record_file);
let mut content_updated = false;
let input = fpl.input;
let working_dir = &app_config.config.load().working_dir;
let storage_path = match get_input_storage_path(&input.name, working_dir) {
Ok(storage_path) => storage_path,
Err(err) => {
error!("Can't resolve vod, input storage directory for input '{}' failed: {err}", input.name);
return;
}
};
// LocalVideo entries are not resolved!
// TODO merge both filters to one
let vod_info_count = fpl.playlistgroups.iter()
.flat_map(|plg| &plg.channels)
.filter(|&pli| pli.header.xtream_cluster == XtreamCluster::Video && pli.header.item_type == PlaylistItemType::Video).count();
.filter(|pli| pli.header.xtream_cluster == XtreamCluster::Video
&& pli.header.item_type == PlaylistItemType::Video
&& !pli.has_details()).count();
info!("Found {vod_info_count} vod info to resolve");
let mut last_log_time = Instant::now();
let mut processed_vod_info_count = 0;
let mut write_counter = 0usize;
for plg in &mut fpl.playlistgroups {
for pli in &mut plg.channels {
// LocalVideo files are not resolved
if pli.header.item_type != PlaylistItemType::Video {
processed_vod_info_count += 1;
if pli.header.xtream_cluster != XtreamCluster::Video
|| pli.header.item_type != PlaylistItemType::Video
|| pli.has_details() {
continue;
}
let (should_update, provider_id, _ts) = should_update_info(pli, &processed_info_ids);
if should_update && provider_id != 0 && fetched_in_run.insert(provider_id) {
if pli.get_item_type().is_local() {
continue;
}
let Some(provider_id) = pli.get_provider_id() else { continue; };
if provider_id != 0 {
if let Some(content) = playlist_resolve_download_playlist_item(client, pli, fpl.input, errors, resolve_delay, XtreamCluster::Video).await {
if let Ok(info) = serde_json::from_str::<XtreamVideoInfo>(&content) {
let provider_id = info.movie_data.stream_id;
let added = info.movie_data.added.parse::<u64>().unwrap_or(0);
let tmdb_id = info.info.tmdb_id.parse::<u32>().unwrap_or(0);
let info_record = InputVodInfoRecord {
tmdb_id,
ts: added,
release_date: info.info.releasedate.clone(),
};
let video_stream_props = VideoStreamProperties::from_info(&info, pli);
handle_error_and_return!(write_vod_info_to_wal_file(provider_id, &serde_json::to_string(&video_stream_props).unwrap_or_else(|_|String::new()),
&info_record, &mut content_writer, &mut record_writer).await,
|err| errors.push(notify_err!(format!("Failed to resolve vod, could not write to wal file {err}"))));
processed_info_ids.insert(provider_id, added);
content_updated = true;
write_counter += content.len();
// periodic flush to bound BufWriter memory
if write_counter >= IO_BUFFER_SIZE {
write_counter = 0;
if let Err(err) = content_writer.flush().await {
errors.push(notify_err!(format!("Failed periodic flush of wal content writer {err}")));
}
if let Err(err) = record_writer.flush().await {
errors.push(notify_err!(format!("Failed periodic flush of wal record writer {err}")));
}
}
// Update in-memory playlist items with the newly fetched vod info.
let _ = persists_input_vod_info(app_config, &storage_path, pli.header.xtream_cluster, &input.name, provider_id, &video_stream_props).await;
// This makes the data available for subsequent processing steps like STRM export.
pli.header.additional_properties = Some(StreamProperties::Video(Box::new(video_stream_props)));
}
}
}
if log_enabled!(Level::Info) {
processed_vod_info_count += 1;
if last_log_time.elapsed().as_secs() >= 30 {
info!("resolved {processed_vod_info_count}/{vod_info_count} vod info");
last_log_time = Instant::now();
}
if log_enabled!(Level::Info) && last_log_time.elapsed().as_secs() >= 30 {
info!("resolved {processed_vod_info_count}/{vod_info_count} vod info");
last_log_time = Instant::now();
}
}
}
info!("resolved {processed_vod_info_count}/{vod_info_count} vod info");
if content_updated {
// TODO better approach for transactional updates is multiplexed WAL file.
// final flush & sync with proper error handling
handle_error!(content_writer.flush().await,
|err| errors.push(notify_err!(format!("Failed to resolve vod, could not write to wal file {err}"))));
handle_error!(record_writer.flush().await,
|err| errors.push(notify_err!(format!("Failed to resolve vod tmdb, could not write to wal file {err}"))));
handle_error!(content_writer.get_ref().sync_all().await, |err| errors.push(notify_err!(format!("Failed to sync vod info to wal file {err}"))));
handle_error!(record_writer.get_ref().sync_all().await, |err| errors.push(notify_err!(format!("Failed to sync vod info record to wal file {err}"))));
// drop writers and files to release handles
drop(content_writer);
drop(record_writer);
drop(wal_content_file);
drop(wal_record_file);
handle_error!(xtream_update_input_info_file(app_config, fpl.input, &wal_content_path, XtreamCluster::Video).await,
|err| errors.push(err));
handle_error!(xtream_update_input_vod_record_from_wal_file(app_config, fpl.input, &wal_record_path).await,
|err| errors.push(err));
}
}