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-22 17:33:18 +01:00
parent 646cc426ba
commit c7bc0beb96
25 changed files with 1844 additions and 1462 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/processor/library.rs
+161 -168
View File
@@ -1,9 +1,7 @@
use crate::library::{MediaMetadata, MetadataAsyncIter, MetadataCacheEntry};
use crate::model::{AppConfig, ConfigInput, XtreamSeriesInfo, XtreamSeriesInfoEpisode, XtreamSeriesInfoEpisodeInfo, XtreamSeriesInfoInfo};
use crate::model::{XtreamMovieData, XtreamMovieInfo, XtreamMovieInfoDetails};
use serde_json::value::RawValue;
use crate::model::{AppConfig, ConfigInput};
use shared::error::TuliproxError;
use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemHeader, PlaylistItemType, XtreamCluster};
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 std::path::Path;
use std::sync::Arc;
@@ -56,10 +54,7 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
match metadata {
MediaMetadata::Movie(_) => {
let additional_properties = metadata_cache_entry_to_xtream_movie_info(entry).and_then(|info| {
let raw = serde_json::to_string(&info).ok()?;
RawValue::from_string(raw).ok()
});
let additional_properties = metadata_cache_entry_to_xtream_movie_info(entry);
vec![PlaylistItem {
header: PlaylistItemHeader {
uuid: string_to_uuid_type(&entry.uuid),
@@ -77,8 +72,49 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
}
MediaMetadata::Series(_series) => {
let mut items = vec![];
if let Some(xtream_series_info) = metadata_cache_entry_to_xtream_series_info(entry) {
let additional_properties = serde_json::to_string(&xtream_series_info).ok().and_then(|r| RawValue::from_string(r).ok());
if let Some(additional_properties) = metadata_cache_entry_to_xtream_series_info(entry) {
let mut episodes = vec![];
if let StreamProperties::Series(series_properties) = &additional_properties {
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 container_extension = Path::new(&episode.direct_source)
.extension()
.and_then(|s| s.to_str())
.map(ToString::to_string).unwrap_or_default();
episodes.push(PlaylistItem {
header: PlaylistItemHeader {
id: episode.id.to_string(),
// 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),
name: episode.title.clone(),
group: group_name.to_string(),
title: episode.title.clone(),
url: episode.direct_source.clone(),
xtream_cluster: XtreamCluster::Series,
item_type: PlaylistItemType::LocalSeries,
category_id: 0,
input_name: input_name.to_string(),
additional_properties: Some(StreamProperties::Episode(EpisodeStreamProperties {
episode_id: episode.id,
episode: episode.episode_num,
season: episode.season,
added: Some(episode.added.clone()),
release_date: Some(episode.release_date.clone()),
tmdb: episode.tmdb,
movie_image: episode.movie_image.clone(),
container_extension,
audio: None,
video: None,
})),
..Default::default()
}
});
}
}
}
}
let series_info = PlaylistItem {
header: PlaylistItemHeader {
@@ -91,34 +127,13 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
xtream_cluster: XtreamCluster::Series,
item_type: PlaylistItemType::LocalSeriesInfo,
input_name: input_name.to_string(),
additional_properties,
additional_properties: Some(additional_properties),
..PlaylistItemHeader::default()
}
};
items.push(series_info);
if let Some(episodes) = xtream_series_info.episodes.as_ref() {
for episode in episodes {
items.push(PlaylistItem {
header: PlaylistItemHeader {
id: episode.id.to_string(),
// 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),
name: episode.title.clone(),
group: group_name.to_string(),
title: episode.title.clone(),
url: episode.direct_source.clone(),
xtream_cluster: XtreamCluster::Series,
item_type: PlaylistItemType::LocalSeries,
category_id: 0,
input_name: input_name.to_string(),
additional_properties: episode.get_additional_properties(&xtream_series_info),
..Default::default()
}
});
}
}
items.extend(episodes);
}
items
@@ -128,7 +143,7 @@ fn to_playlist_item(entry: &MetadataCacheEntry, input_name: &str, group_name: &s
pub fn metadata_cache_entry_to_xtream_movie_info(
entry: &MetadataCacheEntry,
) -> Option<XtreamMovieInfo> {
) -> Option<StreamProperties> {
let movie = match &entry.metadata {
MediaMetadata::Movie(m) => m,
MediaMetadata::Series(_) => return None,
@@ -137,96 +152,123 @@ pub fn metadata_cache_entry_to_xtream_movie_info(
let container_extension = Path::new(&entry.file_path)
.extension()
.and_then(|s| s.to_str())
.map(ToString::to_string);
.map(ToString::to_string).unwrap_or_default();
let actor_names = movie.actors.as_ref().map(|a| a.iter().map(|a| a.name.clone()).collect::<Vec<_>>().join(", "));
let info = XtreamMovieInfoDetails {
kinopoisk_url: movie.tmdb_id.map(|id| format!("https://www.themoviedb.org/movie/{id}")),
tmdb_id: movie.tmdb_id.map(|id| id.to_string()),
name: Some(movie.title.clone()),
o_name: movie.original_title.clone(),
cover_big: movie.poster.clone(),
movie_image: movie.poster.clone(),
releasedate: movie.year.map(|y| format!("{y}-01-01")),
episode_run_time: movie.runtime,
director: movie.directors.as_ref().map(|d| d.join(", ")),
youtube_trailer: movie.videos.as_ref().and_then(|v| v.iter().find(|video| video.site.eq_ignore_ascii_case("youtube")).map(|video| video.key.clone())),
actors: actor_names.clone(),
cast: actor_names,
genre: movie.genres.as_ref().map(|g| g.join(", ")),
description: movie.plot.clone(),
plot: movie.plot.clone(),
age: None,
mpaa_rating: movie.mpaa.clone(),
rating_count_kinopoisk: 0,
country: None,
backdrop_path: movie
.fanart
.as_ref()
.filter(|s| !s.is_empty())
.map(|f| vec![f.clone()])
.or_else(|| {
movie.poster
.as_ref()
.filter(|s| !s.is_empty())
.map(|p| vec![p.clone()])
})
.unwrap_or_default(),
duration_secs: movie.runtime.map(|r| (r * 60).to_string()),
duration: movie.runtime.map(|r| {
let h = r / 60;
let m = r % 60;
format!("{h:02}:{m:02}:00")
}),
video: Vec::new(),
audio: Vec::new(),
bitrate: 0,
rating: movie.rating.map(|r| format!("{r:.2}")),
runtime: movie.runtime.map(|r| (r * 60).to_string()),
status: Some("Released".to_string()),
};
let movie_data = XtreamMovieData {
stream_id: 0,
let properties = VideoStreamProperties {
name: movie.title.clone(),
added: Some(entry.file_modified.to_string()),
category_id: 0,
category_ids: vec![0],
container_extension,
custom_sid: None,
stream_id: 0,
stream_icon: String::new(),
direct_source: String::new(),
custom_sid: None,
added: entry.file_modified.to_string(),
container_extension,
rating: movie.rating,
rating_5based: None,
stream_type: "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,
details: Some(VideoStreamDetailProperties {
kinopoisk_url: movie.tmdb_id.map(|id| format!("https://www.themoviedb.org/movie/{id}")),
o_name: movie.original_title.clone(),
cover_big: movie.poster.clone(),
movie_image: movie.poster.clone(),
release_date: movie.year.map(|y| format!("{y}-01-01")),
episode_run_time: movie.runtime,
director: movie.directors.as_ref().map(|d| d.join(", ")),
youtube_trailer: movie.videos.as_ref().and_then(|v| v.iter().find(|video| video.site.eq_ignore_ascii_case("youtube")).map(|video| video.key.clone())),
actors: actor_names.clone(),
cast: actor_names,
genre: movie.genres.as_ref().map(|g| g.join(", ")),
description: movie.plot.clone(),
plot: movie.plot.clone(),
age: None,
mpaa_rating: movie.mpaa.clone(),
rating_count_kinopoisk: 0,
country: None,
backdrop_path: movie
.fanart
.as_ref()
.filter(|s| !s.is_empty())
.map(|f| vec![f.clone()])
.or_else(|| {
movie.poster
.as_ref()
.filter(|s| !s.is_empty())
.map(|p| vec![p.clone()])
}),
duration_secs: movie.runtime.map(|r| (r * 60).to_string()),
duration: movie.runtime.map(|r| {
let h = r / 60;
let m = r % 60;
format!("{h:02}:{m:02}:00")
}),
video: None,
audio: None,
bitrate: 0,
runtime: movie.runtime.map(|r| (r * 60).to_string()),
status: Some("Released".to_string()),
}),
};
Some(XtreamMovieInfo { info, movie_data })
Some(StreamProperties::Video(Box::new(properties)))
}
pub fn metadata_cache_entry_to_xtream_series_info(
entry: &MetadataCacheEntry,
) -> Option<XtreamSeriesInfo> {
) -> Option<StreamProperties> {
let series = match &entry.metadata {
MediaMetadata::Movie(_) => return None,
MediaMetadata::Series(m) => m,
};
let actor_names = series.actors.as_ref().map(|a| a.iter().map(|a| a.name.clone()).collect::<Vec<_>>().join(", ")).unwrap_or_default();
let release_date = series.year.map(|y| format!("{y}-01-01")).unwrap_or_default();
let release_date = series.year.map(|y| format!("{y}-01-01"));
let youtube_trailer = series.videos.as_ref().and_then(|v| v.iter().find(|video| video.site.eq_ignore_ascii_case("youtube")).map(|video| video.key.clone())).unwrap_or_default();
let info = XtreamSeriesInfoInfo {
let episodes = series.episodes.as_ref().map(|episodes| {
episodes.iter().filter(|episode| !episode.file_path.is_empty()).map(|episode| {
let container_extension = Path::new(&episode.file_path)
.extension()
.and_then(|s| s.to_str())
.map(ToString::to_string)
.unwrap_or_default();
let episode_release_date = episode.aired.as_ref().map(ToString::to_string).unwrap_or_default();
let tmdb_id = (episode.tmdb_id > 0).then_some(episode.tmdb_id);
SeriesStreamDetailEpisodeProperties {
id: tmdb_id.unwrap_or_default(),
episode_num: episode.episode,
season: episode.season,
title: episode.title.clone(),
container_extension,
custom_sid: String::new(),
added: episode.file_modified.to_string(),
direct_source: episode.file_path.clone(),
tmdb: tmdb_id,
release_date: episode_release_date.clone(),
plot: episode.plot.clone(),
duration_secs: episode.runtime.map_or(0, |r| r * 60),
duration: episode.runtime
.map(|r| format!("{:02}:{:02}:00", r / 60, r % 60))
.unwrap_or_default(),
movie_image: episode.thumb.clone().unwrap_or_default(),
audio: None,
video: None,
bitrate: 0,
rating: None,
}
}).collect::<Vec<_>>()
});
let properties = SeriesStreamProperties {
name: series.title.clone(),
cover: series.poster.clone().unwrap_or_default(),
plot: series.plot.clone().unwrap_or_default(),
cast: actor_names,
director: series.directors.as_ref().map(|d| d.join(", ")).unwrap_or_default(),
genre: series.genres.as_ref().map(|d| d.join(", ")).unwrap_or_default(),
release_date: release_date.clone(),
releaseDate: release_date.clone(),
releasedate: release_date.clone(),
last_modified: series.last_updated.to_string(),
rating: series.rating.unwrap_or(0f64),
rating_5based: 0.0,
category_id: 0,
series_id: 0,
backdrop_path: series
.fanart
.as_ref()
@@ -237,73 +279,24 @@ pub fn metadata_cache_entry_to_xtream_series_info(
.filter(|s| !s.is_empty())
.map(|p| vec![p.clone()])
}),
trailer: youtube_trailer.clone(),
cast: actor_names,
cover: series.poster.clone().unwrap_or_default(),
director: series.directors.as_ref().map(|d| d.join(", ")).unwrap_or_default(),
episode_run_time: None,
genre: series.genres.as_ref().map(|d| d.join(", ")),
last_modified: Some(series.last_updated.to_string()),
plot: series.plot.clone(),
rating: series.rating.unwrap_or(0f64),
rating_5based: 0.0,
release_date,
youtube_trailer,
episode_run_time: String::from("0"),
category_id: 0,
tmdb_id: series.tmdb_id,
tmdb: series.tmdb_id,
year: series.year,
details: Some(SeriesStreamDetailProperties {
year: series.year,
// seasons are not delivered through xtream get_series_info.
episodes,
}),
};
// seasons are not delivered through xtream get_series_info.
let seasons = Some(vec![]); /*series.seasons.as_ref()
.map(|seasons| seasons.iter().map(|season| XtreamSeriesInfoSeason {
air_date: season.air_date.clone().unwrap_or_default(),
episode_count: season.episode_count,
id: season.id,
name: season.name.clone(),
overview: season.overview.clone().unwrap_or_default(),
season_number: season.season_number,
vote_average: season.vote_average,
cover: season.poster_path.as_ref().cloned().unwrap_or_default(),
cover_big: season.poster_path.as_ref().cloned().unwrap_or_default(),
}).collect());
*/
let episodes = series.episodes.as_ref().map(|episodes| {
episodes.iter().filter(|episode| !episode.file_path.is_empty()).map(|episode| {
let container_extension = Path::new(&episode.file_path)
.extension()
.and_then(|s| s.to_str())
.map(ToString::to_string)
.unwrap_or_default();
let episode_release_date = episode.aired.as_ref().map(ToString::to_string).unwrap_or_default();
let tmdb_id = (episode.tmdb_id > 0).then_some(episode.tmdb_id);
XtreamSeriesInfoEpisode {
id: episode.id,
episode_num: episode.episode,
season: episode.season,
title: episode.title.clone(),
container_extension,
info: Some(XtreamSeriesInfoEpisodeInfo {
id: tmdb_id,
tmdb_id,
tmdb: tmdb_id,
season: episode.season,
release_date: episode_release_date.clone(),
releaseDate: episode_release_date.clone(),
releasedate: episode_release_date,
plot: episode.plot.clone().unwrap_or_default(),
duration_secs: episode.runtime.map_or(0, |r| r * 60),
duration: episode.runtime
.map(|r| format!("{:02}:{:02}:00", r / 60, r % 60))
.unwrap_or_default(),
movie_image: episode.thumb.clone().unwrap_or_default(),
video: None,
audio: None,
bitrate: 0,
rating: episode.rating.unwrap_or(0.0),
}),
custom_sid: String::new(),
added: episode.file_modified.to_string(),
direct_source: episode.file_path.clone(),
}
}).collect::<Vec<_>>()
});
Some(XtreamSeriesInfo { seasons, info: Some(info), episodes })
Some(StreamProperties::Series(Box::new(properties)))
}
backend/src/processing/processor/playlist.rs
+6 -6
View File
@@ -19,6 +19,7 @@ use crate::model::{InputStats, PlaylistStats, SourceStats, TargetStats};
use crate::processing::parser::xmltv::flatten_tvguide;
use crate::processing::playlist_watch::process_group_watch;
use crate::processing::processor::epg::process_playlist_epg;
use crate::processing::processor::library;
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;
@@ -34,7 +35,6 @@ use shared::foundation::filter::{get_field_value, set_field_value, Filter, Value
use shared::model::{CounterModifier, FieldGetAccessor, FieldSetAccessor, InputType, ItemField, MsgKind, PlaylistEntry, PlaylistGroup, PlaylistItem, PlaylistUpdateState, ProcessingOrder, UUIDType, XtreamCluster};
use shared::utils::{default_as_default, hash_bytes};
use std::time::Instant;
use crate::processing::processor::library;
fn is_valid(pli: &PlaylistItem, filter: &Filter) -> bool {
let provider = ValueProvider { pli };
@@ -111,17 +111,17 @@ fn filter_playlist(playlist: &mut [PlaylistGroup], target: &ConfigTarget) -> Opt
fn assign_channel_no_playlist(new_playlist: &mut [PlaylistGroup]) {
let assigned_chnos: HashSet<u32> = new_playlist.iter().flat_map(|g| &g.channels)
.filter(|c| !c.header.chno.is_empty())
.map(|c| c.header.chno.as_str())
.flat_map(str::parse::<u32>).collect();
.filter(|c| !c.header.chno == 0)
.map(|c| c.header.chno)
.collect();
let mut chno = 1;
for group in new_playlist {
for chan in &mut group.channels {
if chan.header.chno.is_empty() {
if chan.header.chno == 0 {
while assigned_chnos.contains(&chno) {
chno += 1;
}
chan.header.chno = chno.to_string();
chan.header.chno = chno;
chno += 1;
}
}
backend/src/processing/processor/trakt.rs
+9 -21
View File
@@ -1,13 +1,12 @@
use std::borrow::Cow;
use crate::model::{ConfigTarget, TraktListItem, TraktMatchItem};
use shared::model::{FieldGetAccessor, FieldSetAccessor, PlaylistGroup, PlaylistItem, TraktContentType, XtreamCluster};
use crate::model::{TraktConfig, TraktListConfig, TraktMatchResult};
use shared::error::TuliproxError;
use crate::utils::{TraktClient, extract_year_from_title, normalize_title_for_matching};
use shared::utils::{get_u32_from_serde_value};
use shared::utils::{CONSTANTS};
use crate::utils::{extract_year_from_title, normalize_title_for_matching, TraktClient};
use crate::utils::{trace_if_enabled, with};
use log::{debug, info, trace, warn};
use shared::error::TuliproxError;
use shared::model::{FieldGetAccessor, FieldSetAccessor, PlaylistGroup, PlaylistItem, TraktContentType, XtreamCluster};
use shared::utils::CONSTANTS;
use std::borrow::Cow;
use strsim::normalized_levenshtein;
fn extract_quality(value: &str) -> Option<&str> {
@@ -39,22 +38,11 @@ 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(additional_props) = &item.header.additional_properties {
let props_str = additional_props.get();
if let Ok(props) = serde_json::from_str::<serde_json::Map<String, serde_json::Value>>(props_str) {
return props
.get("tmdb_id")
.and_then(get_u32_from_serde_value)
.filter(|&id| id != 0)
.or_else(|| {
props.get("info")
.and_then(|info| info.get("tmdb_id"))
.and_then(get_u32_from_serde_value)
.filter(|&id| id != 0)
});
}
if let Some(props) = &item.header.additional_properties {
props.get_tmdb_id()
} else {
None
}
None
}
fn calculate_year_bonus(playlist_year: Option<u32>, trakt_year: Option<u32>) -> f64 {
backend/src/processing/processor/xtream.rs
+2 -16
View File
@@ -3,7 +3,6 @@ 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::model::normalize_release_date;
use crate::repository::storage::get_input_storage_path;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
@@ -13,7 +12,6 @@ use crate::repository::storage_const;
use crate::repository::xtream_repository::xtream_get_record_file_path;
use crate::utils;
use crate::utils::xtream;
use serde_json::{from_str, to_string, Value};
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> {
let mut result = None;
@@ -34,18 +32,6 @@ pub(in crate::processing) async fn playlist_resolve_download_playlist_item(clien
result
}
pub(in crate::processing) fn normalize_json_content(content: String) -> String {
match from_str::<Value>(&content) {
Ok(mut json_value) => {
if let Some(info) = json_value.get_mut("info").and_then(Value::as_object_mut) {
normalize_release_date(info);
}
to_string(&json_value).unwrap_or(content)
},
Err(_) => content,
}
}
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) => {
@@ -77,9 +63,9 @@ pub(in crate::processing) async fn create_resolve_info_wal_files(cfg: &Config, i
}
}
pub(in crate::processing) fn should_update_info(pli: &mut PlaylistItem, processed_provider_ids: &HashMap<u32, u64>, field: &str) -> (bool, u32, u64) {
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.get_additional_property_as_u64(field);
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()
backend/src/processing/processor/xtream_series.rs
+49 -56
View File
@@ -1,9 +1,7 @@
use crate::model::{AppConfig, ConfigTarget};
use crate::model::FetchedPlaylist;
use crate::model::{XtreamSeriesEpisode, XtreamSeriesInfoEpisode};
use crate::processing::parser::xtream::parse_xtream_series_info;
use crate::processing::processor::playlist::ProcessingPipe;
use crate::processing::processor::xtream::normalize_json_content;
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;
@@ -16,8 +14,8 @@ use crate::utils::{bincode_serialize, IO_BUFFER_SIZE};
use log::{error, info, log_enabled, warn, Level};
use shared::error::{info_err, notify_err};
use shared::error::TuliproxError;
use shared::model::InputType;
use shared::model::{PlaylistGroup, PlaylistItem, PlaylistItemType, XtreamCluster};
use shared::model::{InputType, SeriesStreamProperties, StreamProperties, XtreamSeriesInfo};
use shared::model::{PlaylistGroup, PlaylistItemType, XtreamCluster};
use std::collections::{HashMap, HashSet};
use std::time::Instant;
use tokio::io::AsyncWriteExt;
@@ -31,26 +29,23 @@ async fn read_processed_series_info_ids(cfg: &AppConfig, errors: &mut Vec<Tulipr
async fn write_series_episode_record_to_wal_file(
writer: &mut tokio::io::BufWriter<&mut tokio::fs::File>,
provider_id: u32,
episode: &XtreamSeriesInfoEpisode,
additional_properties: Option<&StreamProperties>,
) -> std::io::Result<usize> {
let series_episode = XtreamSeriesEpisode::from(episode);
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);
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");
}
error!("Cant write to WAL file, content length exceeds u32");
}
Ok(0)
}
fn should_update_series_info(pli: &mut PlaylistItem, processed_provider_ids: &HashMap<u32, u64>) -> (bool, u32, u64) {
should_update_info(pli, processed_provider_ids, crate::model::XC_TAG_SERIES_INFO_LAST_MODIFIED)
}
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
@@ -81,33 +76,36 @@ async fn playlist_resolve_series_info(cfg: &AppConfig, client: &reqwest::Client,
// LocalSeriesInfo entries are not resolved!
for plg in &mut fpl.playlistgroups {
if plg.xtream_cluster != XtreamCluster::Series {
continue;
}
for pli in &mut plg.channels {
if pli.header.item_type != PlaylistItemType::SeriesInfo {
continue;
}
let (should_update, provider_id, ts) = should_update_series_info(pli, &processed_info_ids);
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 let Some(content) = playlist_resolve_download_playlist_item(client, pli, fpl.input, errors, resolve_delay, XtreamCluster::Series).await {
let normalized_content = normalize_json_content(content);
let normalized_str = normalized_content.as_str();
handle_error_and_return!(write_series_info_to_wal_file(provider_id, ts, normalized_str, &mut content_writer, &mut record_writer).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 += normalized_str.len();
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}")));
// 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.
// This makes the data available for subsequent processing steps like STRM export.
pli.header.additional_properties = Some(StreamProperties::Series(Box::new(series_stream_props)));
}
}
}
@@ -145,6 +143,7 @@ async fn playlist_resolve_series_info(cfg: &AppConfig, client: &reqwest::Client,
// - 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<'_>,
@@ -192,36 +191,30 @@ async fn process_series_info(
warn!("Series info content is empty, skipping series with provider id: {provider_id}");
continue;
}
match serde_json::from_str::<serde_json::Value>(&content) {
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() })
};
match parse_xtream_series_info(&series_content, &group, &series_name, input) {
Ok(Some(mut series)) => {
for (episode, 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, episode).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}")));
}
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}"))) }
}
Err(err) => { errors.push(info_err!(format!("Failed to write to series episode wal file: {err}"))) }
}
group_series.extend(series.into_iter().map(|(_, pli)| pli));
}
Ok(None) => {}
Err(err) => {
errors.push(err);
}
group_series.extend(series.into_iter());
}
}
Err(err) => errors.push(info_err!(format!("Failed to parse JSON: {err}"))),
backend/src/processing/processor/xtream_vod.rs
+26 -70
View File
@@ -1,19 +1,15 @@
use crate::model::{AppConfig, ConfigTarget};
use crate::model::FetchedPlaylist;
use crate::processing::processor::xtream::normalize_json_content;
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 serde_json::value::RawValue;
use serde_json::{Map, Value};
use shared::error::notify_err;
use shared::error::TuliproxError;
use shared::model::{InputType, PlaylistEntry};
use shared::model::{PlaylistItem, PlaylistItemType, XtreamCluster};
use shared::utils::{get_string_from_serde_value, get_u32_from_serde_value, get_u64_from_serde_value};
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;
@@ -24,47 +20,6 @@ async fn read_processed_vod_info_ids(cfg: &AppConfig, errors: &mut Vec<TuliproxE
read_processed_info_ids(cfg, errors, fpl, PlaylistItemType::Video, |record: &InputVodInfoRecord| record.ts).await
}
fn extract_info_record_from_vod_info(content: &str) -> Option<(u32, InputVodInfoRecord)> {
let doc = serde_json::from_str::<Map<String, Value>>(content).ok()?;
let movie_data = doc.get(crate::model::XC_TAG_VOD_INFO_MOVIE_DATA)?.as_object()?;
let provider_id = get_u32_from_serde_value(
movie_data.get(crate::model::XC_TAG_VOD_INFO_STREAM_ID)?,
)?;
let added = movie_data
.get(crate::model::XC_TAG_VOD_INFO_ADDED)
.and_then(get_u64_from_serde_value)
.unwrap_or(0);
let info_section = doc.get(crate::model::XC_TAG_VOD_INFO_INFO)?.as_object()?;
let tmdb_id = info_section
.get(crate::model::XC_TAG_VOD_INFO_TMDB_ID)
.and_then(get_u32_from_serde_value)
.filter(|&id| id != 0)
.or_else(|| {
info_section
.get(crate::model::XC_TAG_VOD_INFO_TMDB)
.and_then(get_u32_from_serde_value)
})
.unwrap_or(0);
let release_date = info_section
.get(crate::model::XC_TAG_VOD_INFO_RELEASEDATE)
.and_then(get_string_from_serde_value);
Some((provider_id, InputVodInfoRecord {
tmdb_id,
ts: added,
release_date,
}))
}
fn should_update_vod_info(pli: &mut PlaylistItem, processed_provider_ids: &HashMap<u32, u64>) -> (bool, u32, u64) {
should_update_info(pli, processed_provider_ids, crate::model::XC_TAG_VOD_INFO_ADDED)
}
pub async fn playlist_resolve_vod(app_config: &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; }
@@ -90,7 +45,7 @@ pub async fn playlist_resolve_vod(app_config: &AppConfig, client: &reqwest::Clie
// 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).count();
info!("Found {vod_info_count} vod info to resolve");
let mut last_log_time = Instant::now();
@@ -100,25 +55,32 @@ pub async fn playlist_resolve_vod(app_config: &AppConfig, client: &reqwest::Clie
for plg in &mut fpl.playlistgroups {
for pli in &mut plg.channels {
// LocalVideo files are not resolved
if pli.header.xtream_cluster != XtreamCluster::Video || pli.header.item_type != PlaylistItemType::Video {
if pli.header.item_type != PlaylistItemType::Video {
continue;
}
let (should_update, provider_id, _ts) = should_update_vod_info(pli, &processed_info_ids);
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 let Some(content) = if pli.get_item_type().is_local() {
pli.header.additional_properties.as_ref().map(ToString::to_string)
} else {
playlist_resolve_download_playlist_item(client, pli, fpl.input, errors, resolve_delay, XtreamCluster::Video).await
} {
let normalized_content = normalize_json_content(content);
let normalized_str: &str = &normalized_content;
if let Some((provider_id, info_record)) = extract_info_record_from_vod_info(normalized_str) {
let ts = info_record.ts;
handle_error_and_return!(write_vod_info_to_wal_file(provider_id, normalized_str, &info_record, &mut content_writer, &mut record_writer).await,
if pli.get_item_type().is_local() {
continue;
}
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, ts);
processed_info_ids.insert(provider_id, added);
content_updated = true;
write_counter += normalized_str.len();
write_counter += content.len();
// periodic flush to bound BufWriter memory
if write_counter >= IO_BUFFER_SIZE {
write_counter = 0;
@@ -132,13 +94,7 @@ pub async fn playlist_resolve_vod(app_config: &AppConfig, client: &reqwest::Clie
// Update in-memory playlist items with the newly fetched vod info.
// This makes the data available for subsequent processing steps like STRM export.
if let Ok(value) = serde_json::from_str::<Value>(normalized_str) {
if let Some(info_content) = value.get("info") {
if let Ok(raw_info) = serde_json::to_string(info_content) {
pli.header.additional_properties = serde_json::from_str::<Box<RawValue>>(&raw_info).ok();
}
}
}
pli.header.additional_properties = Some(StreamProperties::Video(Box::new(video_stream_props)));
}
}
}