feat: 支持配置通知器，在视频源处理或整个下载任务出现错误时会触发消息通知 (#526)

crates/bili_sync/src/bilibili/client.rs

@@ -134,4 +134,8 @@ impl BiliClient {
     pub async fn wbi_img(&self, credential: &Credential) -> Result<WbiImg> {
         credential.wbi_img(&self.client).await
     }
+
+    pub fn inner_client(&self) -> &reqwest::Client {
+        &self.client.0
+    }
 }

crates/bili_sync/src/config/current.rs

@@ -11,6 +11,7 @@ use crate::config::default::{default_auth_token, default_bind_address, default_t
 use crate::config::item::{
     ConcurrentLimit, NFOTimeType, SkipOption, default_collection_path, default_favorite_path, default_submission_path,
 };
+use crate::notifier::Notifier;
 use crate::utils::model::{load_db_config, save_db_config};
 
 pub static CONFIG_DIR: LazyLock<PathBuf> =
@@ -27,6 +28,8 @@ pub struct Config {
     pub skip_option: SkipOption,
     pub video_name: String,
     pub page_name: String,
+    #[serde(default)]
+    pub notifiers: Option<Vec<Notifier>>,
     #[serde(default = "default_favorite_path")]
     pub favorite_default_path: String,
     #[serde(default = "default_collection_path")]
@@ -98,6 +101,7 @@ impl Default for Config {
             skip_option: SkipOption::default(),
             video_name: "{{title}}".to_owned(),
             page_name: "{{bvid}}".to_owned(),
+            notifiers: None,
             favorite_default_path: default_favorite_path(),
             collection_default_path: default_collection_path(),
             submission_default_path: default_submission_path(),

crates/bili_sync/src/config/versioned_config.rs

@@ -8,7 +8,7 @@ use tokio::sync::{OnceCell, watch};
 use crate::bilibili::Credential;
 use crate::config::{CONFIG_DIR, Config};
 
-pub static VERSIONED_CONFIG: OnceCell<VersionedConfig> = OnceCell::const_new();
+static VERSIONED_CONFIG: OnceCell<VersionedConfig> = OnceCell::const_new();
 
 pub struct VersionedConfig {
     inner: ArcSwap<Config>,

crates/bili_sync/src/main.rs

@@ -8,6 +8,7 @@ mod config;
 mod database;
 mod downloader;
 mod error;
+mod notifier;
 mod task;
 mod utils;
 mod workflow;

crates/bili_sync/src/notifier/mod.rs

@@ -0,0 +1,46 @@
+use anyhow::Result;
+use futures::future;
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Clone, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase", tag = "type")]
+pub enum Notifier {
+    Telegram { bot_token: String, chat_id: String },
+    Webhook { url: String },
+}
+
+#[derive(Serialize)]
+struct WebhookPayload<'a> {
+    text: &'a str,
+}
+
+pub trait NotifierAllExt {
+    async fn notify_all(&self, client: &reqwest::Client, message: &str) -> Result<()>;
+}
+
+impl NotifierAllExt for Option<Vec<Notifier>> {
+    async fn notify_all(&self, client: &reqwest::Client, message: &str) -> Result<()> {
+        let Some(notifiers) = self else {
+            return Ok(());
+        };
+        future::join_all(notifiers.iter().map(|notifier| notifier.notify(client, message))).await;
+        Ok(())
+    }
+}
+
+impl Notifier {
+    pub async fn notify(&self, client: &reqwest::Client, message: &str) -> Result<()> {
+        match self {
+            Notifier::Telegram { bot_token, chat_id } => {
+                let url = format!("https://api.telegram.org/bot{}/sendMessage", bot_token);
+                let params = [("chat_id", chat_id.as_str()), ("text", message)];
+                client.post(&url).form(&params).send().await?;
+            }
+            Notifier::Webhook { url } => {
+                let payload = WebhookPayload { text: message };
+                client.post(url).json(&payload).send().await?;
+            }
+        }
+        Ok(())
+    }
+}

crates/bili_sync/src/task/video_downloader.rs

@@ -8,6 +8,7 @@ use tokio::time;
 use crate::adapter::VideoSource;
 use crate::bilibili::{self, BiliClient};
 use crate::config::{Config, TEMPLATE, VersionedConfig};
+use crate::notifier::NotifierAllExt;
 use crate::utils::model::get_enabled_video_sources;
 use crate::utils::task_notifier::TASK_STATUS_NOTIFIER;
 use crate::workflow::process_video_source;
@@ -20,7 +21,12 @@ pub async fn video_downloader(connection: DatabaseConnection, bili_client: Arc<B
         let mut config = VersionedConfig::get().snapshot();
         info!("开始执行本轮视频下载任务..");
         if let Err(e) = download_all_video_sources(&connection, &bili_client, &mut config, &mut anchor).await {
-            error!("本轮视频下载任务执行遇到错误：{:#}，跳过本轮执行", e);
+            let error_msg = format!("本轮视频下载任务执行遇到错误：{:#}", e);
+            error!("{error_msg}");
+            let _ = config
+                .notifiers
+                .notify_all(bili_client.inner_client(), &error_msg)
+                .await;
         } else {
             info!("本轮视频下载任务执行完毕");
         }
@@ -67,7 +73,12 @@ async fn download_all_video_sources(
     for video_source in video_sources {
         let display_name = video_source.display_name();
         if let Err(e) = process_video_source(video_source, &bili_client, connection, &template, config).await {
-            error!("处理 {} 时遇到错误：{:#}，跳过该视频源", display_name, e);
+            let error_msg = format!("处理 {} 时遇到错误：{:#}，跳过该视频源", display_name, e);
+            error!("{error_msg}");
+            let _ = config
+                .notifiers
+                .notify_all(bili_client.inner_client(), &error_msg)
+                .await;
         }
     }
     Ok(())