euzu/tuliprox
1
0
Fork 0
mirror of https://github.com/euzu/tuliprox.git synced 2026-10-28 09:42:26 +01:00
Code Issues Packages Projects Releases 5 Wiki Activity

Messaging templates (#534)

Browse Source 
Messaging templates
This commit is contained in:
euzu
2026-01-25 12:41:18 +01:00
committed by GitHub
parent 4c27f13d47
commit a407f209a8
33 changed files with 1074 additions and 228 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/src/processing/playlist_watch.rs
+8 -17
View File
@@ -1,14 +1,14 @@
use std::collections::BTreeSet;
use std::sync::Arc;
use std::path::{Path};
use log::{error, info};
use shared::model::{MsgKind, PlaylistGroup};
use log::{error};
use shared::model::{PlaylistGroup};
use crate::messaging::{send_message};
use crate::model::Config;
use crate::model::{AppConfig, MessageContent, WatchChanges};
use crate::utils;
use crate::utils::{binary_deserialize, binary_serialize, file_exists_async};
pub async fn process_group_watch(client: &reqwest::Client, cfg: &Config, target_name: &str, pl: &PlaylistGroup) {
pub async fn process_group_watch(app_config: &Arc<AppConfig>, client: &reqwest::Client, target_name: &str, pl: &PlaylistGroup) {
let mut new_tree = BTreeSet::new();
pl.channels.iter().for_each(|chan| {
let header = &chan.header;
@@ -17,6 +17,7 @@ pub async fn process_group_watch(client: &reqwest::Client, cfg: &Config, target_
});
let watch_filename = format!("{}/{}.bin", utils::sanitize_filename(target_name), utils::sanitize_filename(&pl.title));
let cfg = app_config.config.load();
match utils::get_file_path(&cfg.working_dir, Some(std::path::PathBuf::from(&watch_filename))) {
Some(path) => {
let save_path = path.as_path();
@@ -28,7 +29,7 @@ pub async fn process_group_watch(client: &reqwest::Client, cfg: &Config, target_
let removed_difference: BTreeSet<Arc<str>> = loaded_tree.difference(&new_tree).cloned().collect();
if !added_difference.is_empty() || !removed_difference.is_empty() {
changed = true;
handle_watch_notification(client, cfg, &added_difference, &removed_difference, target_name, &pl.title).await;
handle_watch_notification(app_config, client, &added_difference, &removed_difference, target_name, &pl.title).await;
}
} else {
error!("failed to load watch_file {}", &path.to_str().unwrap_or_default());
@@ -52,15 +53,7 @@ pub async fn process_group_watch(client: &reqwest::Client, cfg: &Config, target_
}
}
#[derive(Debug, serde::Serialize)]
struct WatchChanges {
pub target: String,
pub group: String,
pub added: Vec<String>,
pub removed: Vec<String>,
}
async fn handle_watch_notification(client: &reqwest::Client, cfg: &Config, added: &BTreeSet<Arc<str>>, removed: &BTreeSet<Arc<str>>, target_name: &str, group_name: &str) {
async fn handle_watch_notification(app_config: &Arc<AppConfig>, client: &reqwest::Client, added: &BTreeSet<Arc<str>>, removed: &BTreeSet<Arc<str>>, target_name: &str, group_name: &str) {
let added = added.iter().map(std::string::ToString::to_string).collect::<Vec<String>>();
let removed = removed.iter().map(std::string::ToString::to_string).collect::<Vec<String>>();
if !added.is_empty() || !removed.is_empty() {
@@ -72,9 +65,7 @@ async fn handle_watch_notification(client: &reqwest::Client, cfg: &Config, added
removed
};
let msg = serde_json::to_string_pretty(&changes).unwrap_or_else(|_| "Error: Failed to serialize watch changes".to_string());
info!("{}", &msg);
send_message(client, MsgKind::Watch, cfg.messaging.as_ref(), &msg).await;
send_message(app_config, client, MessageContent::Watch(changes)).await;
}
}
backend/src/processing/processor/playlist.rs
+13 -26
View File
@@ -1,4 +1,4 @@
use crate::model::{AppConfig, Config, ConfigFavourites, ConfigInput, ConfigRename, ReverseProxyDisabledHeaderConfig, TVGuide};
use crate::model::{AppConfig, ConfigFavourites, ConfigInput, ConfigRename, ReverseProxyDisabledHeaderConfig, TVGuide};
use crate::utils::m3u;
use crate::utils::xtream;
use crate::utils::{epg, StepMeasureCallback};
@@ -9,7 +9,8 @@ use tokio::sync::{Mutex, OwnedRwLockWriteGuard, RwLock};
use tokio::task::JoinSet;
use crate::api::model::{EventManager, EventMessage, PlaylistStorageState, UpdateGuard};
use crate::messaging::send_message_json;
use crate::messaging::{send_message};
use crate::model::messaging::{MessageContent};
use crate::model::Epg;
@@ -37,7 +38,7 @@ use log::{debug, error, info, log_enabled, warn, Level};
use shared::error::{get_errors_notify_message, notify_err, TuliproxError};
use shared::foundation::filter::{get_field_value, set_field_value, Filter, ValueAccessor, ValueProvider};
use shared::model::xtream_const::XTREAM_CLUSTER;
use shared::model::{CounterModifier, FieldGetAccessor, FieldSetAccessor, InputType, ItemField, MsgKind,
use shared::model::{CounterModifier, FieldGetAccessor, FieldSetAccessor, InputType, ItemField,
PlaylistGroup, PlaylistItem, PlaylistItemType, PlaylistUpdateState,
ProcessingOrder, XtreamCluster};
use shared::utils::{create_alias_uuid, default_as_default, interner_gc, Internable};
@@ -762,8 +763,7 @@ async fn process_playlist_for_target(ctx: &PlaylistProcessingContext,
map_playlist_counter(target, &mut flat_new_playlist);
step.tick("assigning channel counter");
let config = ctx.config.config.load();
if process_watch(&config, &ctx.client, target, &flat_new_playlist).await {
if process_watch(&ctx.config, &ctx.client, target, &flat_new_playlist).await {
step.tick("group watches");
}
let result = persist_playlist(&ctx.config, &mut flat_new_playlist, flatten_tvguide(&new_epg).as_ref(), target, ctx.playlist_state.as_ref()).await;
@@ -842,7 +842,7 @@ async fn process_epg(processed_fetched_playlists: &mut Vec<FetchedPlaylist<'_>>)
(new_epg, new_playlist)
}
async fn process_watch(cfg: &Config, client: &reqwest::Client, target: &ConfigTarget, new_playlist: &[PlaylistGroup]) -> bool {
async fn process_watch(app_config: &Arc<AppConfig>, client: &reqwest::Client, target: &ConfigTarget, new_playlist: &[PlaylistGroup]) -> bool {
if let Some(watches) = &target.watch {
if default_as_default().eq_ignore_ascii_case(&target.name) {
error!("can't watch a target with no unique name");
@@ -853,7 +853,7 @@ async fn process_watch(cfg: &Config, client: &reqwest::Client, target: &ConfigTa
new_playlist
.iter()
.filter(|pl| watches.iter().any(|r| r.is_match(&pl.title)))
.map(|pl| process_group_watch(client, cfg, &target.name, pl))
.map(|pl| process_group_watch(app_config, client, &target.name, pl))
).for_each_concurrent(16, |f| f).await;
true
@@ -898,25 +898,14 @@ pub async fn exec_processing(client: &reqwest::Client, app_config: Arc<AppConfig
for err in &errors {
error!("{}", err.message);
}
let config = app_config.config.load();
let messaging = config.messaging.as_ref();
if !stats.is_empty() {
match serde_json::to_value(&stats) {
Ok(val) => {
match serde_json::to_string(&serde_json::Value::Object(
serde_json::map::Map::from_iter([("stats".to_string(), val)]))) {
Ok(stats_msg) => {
// print stats
info!("{stats_msg}");
// send stats
send_message_json(client, MsgKind::Stats, messaging, stats_msg.as_str()).await;
}
Err(err) => error!("Failed to serialize playlist stats {err}"),
}
}
Err(err) => error!("Failed to serialize playlist stats {err}")
// print stats
if let Ok(stats_msg) = serde_json::to_string(&stats) {
info!("stats: {stats_msg}");
}
// send stats
send_message(&app_config, client, MessageContent::event_stats(stats)).await;
}
// send errors
@@ -924,9 +913,7 @@ pub async fn exec_processing(client: &reqwest::Client, app_config: Arc<AppConfig
if let Some(events) = &event_manager {
events.send_event(EventMessage::PlaylistUpdate(PlaylistUpdateState::Failure));
}
if let Ok(error_msg) = serde_json::to_string(&serde_json::Value::Object(serde_json::map::Map::from_iter([("errors".to_string(), serde_json::Value::String(message))]))) {
send_message_json(client, MsgKind::Error, messaging, error_msg.as_str()).await;
}
send_message(&app_config, client, MessageContent::event_error(message)).await;
} else if let Some(events) = &event_manager {
events.send_event(EventMessage::PlaylistUpdate(PlaylistUpdateState::Success));
}