Refactor plugin system and update dependencies

Refactored the plugin system, including changes to plugin loading and Twitch plugin implementation. Updated dependencies in Cargo.lock to their latest versions, improving compatibility and security. Adjusted downloader and download logic to align with the new plugin structure.

crates/biliup-cli/src/server/common/download.rs

@@ -7,6 +7,7 @@ use crate::server::core::monitor::RoomsHandle;
 use crate::server::core::plugin::{DownloadPlugin, StreamStatus};
 use crate::server::errors::{AppError, AppResult};
 use crate::server::infrastructure::context::{Context, Worker, WorkerStatus};
+use crate::server::infrastructure::models::hook_step::{process, process_video};
 use async_channel::{Receiver, Sender};
 use error_stack::{Report, ResultExt, bail};
 use std::fs;
@@ -445,8 +446,14 @@ impl DActor {
                     .toggle(worker.clone(), WorkerStatus::Working(task.clone()))
                     .await;
 
+                process(&[], &ctx.worker.get_streamer().preprocessor).await;
+
+                let option = &ctx.worker.get_streamer().downloaded_processor;
+
                 let result = task.execute(ctx, processor, &rooms_handle).await;
 
+                process(&[], option).await;
+
                 info!(
                     "Download workflow completed {} => {:?}",
                     worker.live_streamer.url,

crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs

@@ -294,7 +294,6 @@ impl FfmpegDownloader {
         }
         let status = spawn_log(child, &self.process_handle).await?;
 
-
         if let Some(file_path) = prev_file_path {
             // 重命名文件
             let no_ext = file_path.with_extension("");
@@ -357,9 +356,10 @@ impl Downloader for FfmpegDownloader {
     // }
 }
 
-
-async fn spawn_log(mut child: tokio::process::Child, process_handle: &RwLock<Option<tokio::process::Child>>) -> AppResult<ExitStatus> {
-
+async fn spawn_log(
+    mut child: tokio::process::Child,
+    process_handle: &RwLock<Option<tokio::process::Child>>,
+) -> AppResult<ExitStatus> {
     let stderr = child.stderr.take().ok_or(AppError::Custom(
         "failed to capture stderr pipe".to_string(),
     ))?;
@@ -378,7 +378,6 @@ async fn spawn_log(mut child: tokio::process::Child, process_handle: &RwLock<Opt
         }
     });
 
-
     // 确保读任务结束（忽略它们的返回错误以避免因提前关闭管道导致的 join 错）
     let _ = stderr_task.await;
 

crates/biliup-cli/src/server/core/plugin.rs

@@ -1,5 +1,5 @@
-pub mod yy;
 mod twitch;
+pub mod yy;
 
 use crate::server::common::construct_headers;
 use crate::server::common::util::{Recorder, parse_time};

crates/biliup-cli/src/server/core/plugin/twitch.rs

@@ -1,20 +1,19 @@
+use crate::server::errors::{AppError, AppResult};
+use error_stack::{IntoReport, ResultExt, bail, report};
 use regex::Regex;
 use reqwest::Client;
 use serde::{Deserialize, Serialize};
-use serde_json::{json, Value};
+use serde_json::{Value, json};
 use std::sync::Arc;
-use error_stack::{bail, report, IntoReport, ResultExt};
 use tokio::net::TcpListener;
 use tokio::process::{Child, Command};
 use tokio::sync::RwLock;
-use tokio::time::{sleep, Duration};
+use tokio::time::{Duration, sleep};
 use tracing::{error, warn};
-use crate::server::errors::{AppError, AppResult};
 
 // 常量定义
 const CLIENT_ID: &str = "kimne78kx3ncx6brgo4mv6wki5h1ko";
 
-
 // GraphQL 响应结构
 #[derive(Debug, Deserialize)]
 struct GqlResponse {
@@ -80,12 +79,14 @@ impl Twitch {
             raw_stream_url: None,
             proc: None,
             danmaku: None,
-            re: Regex::new(r"https?://(?:(?:www|go|m)\.)?twitch\.tv/(?P<id>[0-9_a-zA-Z]+)").unwrap(),
+            re: Regex::new(r"https?://(?:(?:www|go|m)\.)?twitch\.tv/(?P<id>[0-9_a-zA-Z]+)")
+                .unwrap(),
         }
     }
 
     pub async fn acheck_stream(&mut self, is_check: bool) -> AppResult<bool> {
-        let channel_name = self.re
+        let channel_name = self
+            .re
             .captures(&self.url)
             .and_then(|caps| caps.name("id"))
             .map(|m| m.as_str().to_lowercase())
@@ -121,7 +122,8 @@ impl Twitch {
 
         let gql: GqlResponse = TwitchUtils::post_gql(ops).await?;
 
-        let user = gql.data
+        let user = gql
+            .data
             .and_then(|d| d.user)
             .ok_or_else(|| AppError::Custom("获取错误".to_string()))?;
 
@@ -139,14 +141,22 @@ impl Twitch {
 
         if self.downloader == "streamlink" || self.downloader == "ffmpeg" {
             // 获取可用端口
-            let listener = TcpListener::bind("127.0.0.1:0").await.change_context(AppError::Unknown)?;
-            let port = listener.local_addr().change_context(AppError::Unknown)?.port().to_string();
+            let listener = TcpListener::bind("127.0.0.1:0")
+                .await
+                .change_context(AppError::Unknown)?;
+            let port = listener
+                .local_addr()
+                .change_context(AppError::Unknown)?
+                .port()
+                .to_string();
             drop(listener);
 
             let mut args = vec![
                 "--player-external-http",
-                "--player-external-http-port", &port,
-                "--player-external-http-interface", "localhost",
+                "--player-external-http-port",
+                &port,
+                "--player-external-http-interface",
+                "localhost",
             ];
 
             let mut extra_args = Vec::new();
@@ -156,7 +166,10 @@ impl Twitch {
             }
 
             if let Some(auth_token) = TwitchUtils::get_auth_token() {
-                extra_args.push(format!("--twitch-api-header=Authorization=OAuth {}", auth_token));
+                extra_args.push(format!(
+                    "--twitch-api-header=Authorization=OAuth {}",
+                    auth_token
+                ));
             }
 
             let port_str = port.to_string();
@@ -166,9 +179,7 @@ impl Twitch {
                 cmd.arg(arg);
             }
 
-            cmd.args(&args)
-                .arg(&self.url)
-                .arg("best");
+            cmd.args(&args).arg(&self.url).arg("best");
 
             let mut child = cmd.spawn().change_context(AppError::Unknown)?;
 
@@ -176,7 +187,11 @@ impl Twitch {
 
             // 等待进程启动
             for _ in 0..5 {
-                if child.try_wait().change_context(AppError::Unknown)?.is_some() {
+                if child
+                    .try_wait()
+                    .change_context(AppError::Unknown)?
+                    .is_some()
+                {
                     return Ok(false);
                 }
                 sleep(Duration::from_secs(1)).await;
@@ -185,7 +200,8 @@ impl Twitch {
             self.proc = Some(child);
             Ok(true)
         } else {
-            let token = stream.playback_access_token
+            let token = stream
+                .playback_access_token
                 .ok_or_else(|| AppError::Custom("No playback access token".to_string()))?;
 
             let query_params = [
@@ -199,7 +215,8 @@ impl Twitch {
                 ("token", token.value),
             ];
 
-            let query_string = serde_urlencoded::to_string(&query_params).change_context(AppError::Unknown)?;
+            let query_string =
+                serde_urlencoded::to_string(&query_params).change_context(AppError::Unknown)?;
             self.raw_stream_url = Some(format!(
                 "https://usher.ttvnw.net/api/channel/hls/{}.m3u8?{}",
                 channel_name, query_string
@@ -213,7 +230,8 @@ impl Twitch {
         let mut ops = Vec::new();
 
         for url in &check_urls {
-            let channel_name = self.re
+            let channel_name = self
+                .re
                 .captures(url)
                 .and_then(|caps| caps.name("id"))
                 .map(|m| m.as_str().to_lowercase())
@@ -263,7 +281,7 @@ impl Twitch {
     pub async fn close(&mut self) -> AppResult<()> {
         if let Some(mut proc) = self.proc.take() {
             match tokio::time::timeout(Duration::from_secs(5), proc.kill()).await {
-                Ok(Ok(_)) => {},
+                Ok(Ok(_)) => {}
                 Ok(Err(e)) => error!("terminate {} failed: {}", self.fname, e),
                 Err(_) => {
                     warn!("Timeout expired, force killing process");
@@ -298,13 +316,26 @@ impl TwitchUtils {
         // 最多重试一次（在 token 失效的情况下）
         for retry in 0..2 {
             let mut headers = reqwest::header::HeaderMap::new();
-            headers.insert("Content-Type", "text/plain;charset=UTF-8".parse().change_context(AppError::Unknown)?);
-            headers.insert("Client-ID", CLIENT_ID.parse().change_context(AppError::Unknown)?);
+            headers.insert(
+                "Content-Type",
+                "text/plain;charset=UTF-8"
+                    .parse()
+                    .change_context(AppError::Unknown)?,
+            );
+            headers.insert(
+                "Client-ID",
+                CLIENT_ID.parse().change_context(AppError::Unknown)?,
+            );
 
             // 第二次重试时不使用 auth_token（因为已经失效）
             if retry == 0 {
                 if let Some(auth_token) = Self::get_auth_token() {
-                    headers.insert("Authorization", format!("OAuth {}", auth_token).parse().change_context(AppError::Unknown)?);
+                    headers.insert(
+                        "Authorization",
+                        format!("OAuth {}", auth_token)
+                            .parse()
+                            .change_context(AppError::Unknown)?,
+                    );
                 }
             }
 
@@ -315,9 +346,11 @@ impl TwitchUtils {
                 .json(&ops)
                 .timeout(Duration::from_secs(15))
                 .send()
-                .await.change_context(AppError::Unknown)?;
+                .await
+                .change_context(AppError::Unknown)?;
 
-            resp.error_for_status_ref().change_context(AppError::Unknown)?;
+            resp.error_for_status_ref()
+                .change_context(AppError::Unknown)?;
             let gql: GqlResponse = resp.json().await.change_context(AppError::Unknown)?;
 
             if gql.error.as_deref() == Some("Unauthorized") {
@@ -346,11 +379,24 @@ impl TwitchUtils {
 
     async fn post_gql_chunk(ops: &[Value]) -> AppResult<Vec<GqlResponse>> {
         let mut headers = reqwest::header::HeaderMap::new();
-        headers.insert("Content-Type", "text/plain;charset=UTF-8".parse().change_context(AppError::Unknown)?);
-        headers.insert("Client-ID", CLIENT_ID.parse().change_context(AppError::Unknown)?);
+        headers.insert(
+            "Content-Type",
+            "text/plain;charset=UTF-8"
+                .parse()
+                .change_context(AppError::Unknown)?,
+        );
+        headers.insert(
+            "Client-ID",
+            CLIENT_ID.parse().change_context(AppError::Unknown)?,
+        );
 
         if let Some(auth_token) = Self::get_auth_token() {
-            headers.insert("Authorization", format!("OAuth {}", auth_token).parse().change_context(AppError::Unknown)?);
+            headers.insert(
+                "Authorization",
+                format!("OAuth {}", auth_token)
+                    .parse()
+                    .change_context(AppError::Unknown)?,
+            );
         }
 
         let client = Client::new();
@@ -360,9 +406,11 @@ impl TwitchUtils {
             .json(&ops)
             .timeout(Duration::from_secs(15))
             .send()
-            .await.change_context(AppError::Unknown)?;
+            .await
+            .change_context(AppError::Unknown)?;
 
-        resp.error_for_status_ref().change_context(AppError::Unknown)?;
+        resp.error_for_status_ref()
+            .change_context(AppError::Unknown)?;
         Ok(resp.json().await.change_context(AppError::Unknown)?)
     }
 }
@@ -377,4 +425,4 @@ impl DanmakuClient {
     fn new(url: String, filename: String) -> Self {
         Self { url, filename }
     }
-}
+}

crates/biliup-cli/src/server/infrastructure/models/hook_step.rs

@@ -6,7 +6,7 @@ use std::path::Path;
 use tokio::fs;
 use tokio::io::AsyncWriteExt;
 use tokio::process::Command;
-use tracing::info;
+use tracing::{error, info};
 
 /// 钩子步骤枚举：支持多种操作格式
 /// 既支持 key-value 形式（如 {run: "..."}），也支持纯字符串（如 "rm"）
@@ -51,14 +51,61 @@ impl HookStep {
         Ok(())
     }
 
+    /// 执行钩子步骤操作
+    ///
+    /// # 参数
+    /// * `video_paths` - 视频文件路径列表
+    ///
+    /// # 返回
+    /// 执行成功返回Ok(())，失败返回错误信息
+    pub async fn execute_with(&self, src: &[u8]) -> AppResult<()> {
+        match self {
+            HookStep::Run { run } => {
+                // 执行自定义命令
+                // 解析命令和参数
+                let parts: Vec<&str> = run.split_whitespace().collect();
+                if parts.is_empty() {
+                    bail!(AppError::Custom("Empty command".into()));
+                }
+
+                // 启动子进程，配置标准输入管道
+                let mut process = Command::new(parts[0])
+                    .args(&parts[1..])
+                    .stdin(std::process::Stdio::piped())
+                    .spawn()
+                    .change_context(AppError::Unknown)?;
+
+                // 将自定义输入写入标准输入
+                if let Some(mut stdin) = process.stdin.take() {
+                    stdin
+                        .write_all(src)
+                        .await
+                        .change_context(AppError::Unknown)?;
+                }
+
+                // 等待进程完成并检查退出状态
+                let status = process.wait().await.change_context(AppError::Unknown)?;
+                if !status.success() {
+                    bail!(AppError::Custom(format!(
+                        "Command failed with status: {}",
+                        status
+                    )));
+                }
+            }
+            cmd => {
+                // 未知命令，返回错误
+                bail!(AppError::Custom(format!("不支持的命令: {:?}", cmd)));
+            }
+        }
+        Ok(())
+    }
+
     /// 执行自定义命令，将视频路径作为标准输入传入
     ///
     /// # 参数
     /// * `cmd` - 要执行的命令字符串
     /// * `video_paths` - 视频文件路径列表
     async fn execute_command(&self, cmd: &str, video_paths: &[&Path]) -> AppResult<()> {
-        println!("Executing: {}", cmd);
-
         // 解析命令和参数
         let parts: Vec<&str> = cmd.split_whitespace().collect();
         if parts.is_empty() {
@@ -78,7 +125,7 @@ impl HookStep {
                 .iter()
                 .map(|p| p.to_string_lossy())
                 .reduce(|acc, e| Cow::from(acc.to_string() + "\n" + &*e))
-                .expect("At least one video path should exist");
+                .ok_or(AppError::Unknown)?;
             stdin
                 .write_all(paths_str.as_bytes())
                 .await
@@ -215,3 +262,16 @@ pub async fn process_video(video_path: &[&Path], processors: &[HookStep]) -> App
     info!("Video processing completed");
     Ok(())
 }
+
+pub async fn process(input: &[u8], processors: &Option<Vec<HookStep>>) {
+    if let Some(hooks) = processors {
+        // 依次执行每个处理器步骤
+        for processor in hooks {
+            info!(processor=?processor, "Starting processing...");
+            if let Err(e) = processor.execute_with(input).await {
+                error!(error=?e, "自定义处理执行出错");
+            }
+            info!(processor=?processor, "processing completed");
+        }
+    }
+}