Add Streamlink downloader and refactor Twitch plugin

Introduces a new Streamlink downloader implementation supporting multiple platforms and output modes. Refactors the Twitch plugin to use async traits and separates TwitchDownloader logic. Cleans up endpoint logging and updates workspace version to 1.1.22.

Cargo.lock

@@ -362,7 +362,7 @@ dependencies = [
 
 [[package]]
 name = "biliup-cli"
-version = "1.1.21"
+version = "1.1.22"
 dependencies = [
  "async-channel",
  "async-trait",
@@ -3643,7 +3643,7 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
 
 [[package]]
 name = "stream-gears"
-version = "1.1.21"
+version = "1.1.22"
 dependencies = [
  "async-channel",
  "async-trait",

Cargo.toml

@@ -7,7 +7,7 @@ members = [
 resolver = "3"
 
 [workspace.package]
-version = "1.1.21"
+version = "1.1.22"
 
 [workspace.dependencies]
 serde = { version = "1.0", features = ["derive"] }

crates/biliup-cli/src/server/api/endpoints.rs

@@ -164,7 +164,6 @@ pub async fn pause_streamers_endpoint(
         match worker_status {
             WorkerStatus::Working(d) => {
                 d.stop().await.map_err(report_to_response)?;
-
                 monitor
                     .rooms_handle
                     .toggle(w.clone(), WorkerStatus::Pause)
@@ -183,7 +182,6 @@ pub async fn pause_streamers_endpoint(
                     .rooms_handle
                     .toggle(w.clone(), WorkerStatus::Pause)
                     .await;
-                info!("非预期状态")
             }
         };
     }

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

@@ -2,6 +2,7 @@
 pub mod ffmpeg_downloader;
 /// Stream-gears下载器实现
 pub mod stream_gears;
+mod streamlink;
 
 use crate::server::common::util::Recorder;
 use crate::server::errors::AppResult;

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

@@ -0,0 +1,218 @@
+use error_stack::{IntoReport, Result, ResultExt};
+use std::collections::HashMap;
+use std::process::{Child, Command, Stdio};
+use tokio::time::{sleep, Duration};
+use url::Url;
+use crate::server::errors::{AppError, AppResult};
+
+#[derive(Debug, Clone)]
+pub enum Platform {
+    Bilibili,
+    Twitch { disable_ads: bool },
+    Generic,
+}
+
+#[derive(Debug, Clone)]
+pub enum OutputMode {
+    /// 管道模式：streamlink输出到stdout，由父进程读取
+    Pipe,
+    /// HTTP服务器模式：streamlink启动本地HTTP服务器
+    HttpServer { port: u16 },
+}
+
+pub struct StreamlinkDownloader {
+    platform: Platform,
+    url: String,
+    headers: HashMap<String, String>,
+    output_mode: OutputMode,
+    proc: Option<Child>,
+}
+
+impl StreamlinkDownloader {
+    pub fn new(url: String, platform: Platform) -> Self {
+        Self {
+            platform,
+            url,
+            headers: HashMap::new(),
+            output_mode: OutputMode::Pipe, // 默认管道模式
+            proc: None,
+        }
+    }
+
+    pub fn with_headers(mut self, headers: HashMap<String, String>) -> Self {
+        self.headers = headers;
+        self
+    }
+
+    pub fn with_output_mode(mut self, mode: OutputMode) -> Self {
+        self.output_mode = mode;
+        self
+    }
+
+    /// 启动streamlink进程
+    pub async fn start(&mut self) -> AppResult<StreamOutput> {
+        let mut cmd = Command::new("streamlink");
+
+        // 基础参数
+        cmd.args([
+            "--stream-segment-threads", "3",
+            "--hls-playlist-reload-attempts", "1",
+        ]);
+
+        // 添加HTTP headers
+        for (key, value) in &self.headers {
+            cmd.args(["--http-header", &format!("{}={}", key, value)]);
+        }
+
+        // 平台特定处理
+        let platform_args = self.build_platform_args().await?;
+        for arg in platform_args {
+            cmd.arg(arg);
+        }
+
+        // 配置输出模式
+        let output = match &self.output_mode {
+            OutputMode::Pipe => {
+                cmd.args([&self.url, "best", "-O"]);
+                cmd.stdout(Stdio::piped());
+
+                let child = cmd.spawn().change_context(AppError::Unknown)?;
+                StreamOutput::Pipe(child)
+            }
+            OutputMode::HttpServer { port } => {
+                cmd.args([
+                    "--player-external-http",
+                    "--player-external-http-port", &port.to_string(),
+                    "--player-external-http-interface", "localhost",
+                    &self.url,
+                    "best",
+                ]);
+
+                let mut child = cmd.spawn().change_context(AppError::Unknown)?;
+
+                // 等待HTTP服务器启动
+                self.wait_for_startup(&mut child).await?;
+
+                StreamOutput::Http {
+                    url: format!("http://localhost:{}", port),
+                    process: child,
+                }
+            }
+        };
+
+        Ok(output)
+    }
+
+    /// 构建平台特定参数
+    async fn build_platform_args(&self) -> AppResult<Vec<String>> {
+        let mut args = Vec::new();
+
+        match &self.platform {
+            Platform::Bilibili => {
+                // Bilibili需要保留特定URL参数，否则segment请求会404
+                args.extend(self.parse_bilibili_params()?);
+            }
+            Platform::Twitch { disable_ads } => {
+                if *disable_ads {
+                    args.push("--twitch-disable-ads".to_string());
+                }
+
+                // 添加认证token（如果存在）
+                if let Some(token) = Self::get_twitch_auth_token() {
+                    args.push(format!("--twitch-api-header=Authorization=OAuth {}", token));
+                }
+            }
+            Platform::Generic => {}
+        }
+
+        Ok(args)
+    }
+
+    /// 解析Bilibili URL参数（白名单过滤）
+    fn parse_bilibili_params(&self) -> AppResult<Vec<String>> {
+        let mut params = Vec::new();
+
+        let url = Url::parse(&self.url).change_context(AppError::Unknown)?;
+        let query = url.query().unwrap_or("");
+
+        // 白名单参数
+        let mut whitelist = vec![
+            "uparams", "upsig", "sigparams", "sign",
+            "flvsk", "sk", "mid", "site"
+        ];
+
+        // 动态扩展白名单
+        let query_pairs: HashMap<_, _> = url.query_pairs().collect();
+
+        if let Some(sigparams) = query_pairs.get("sigparams") {
+            whitelist.extend(sigparams.split(',').map(|s| s.trim()));
+        }
+        if let Some(uparams) = query_pairs.get("uparams") {
+            whitelist.extend(uparams.split(',').map(|s| s.trim()));
+        }
+
+        // 过滤参数
+        for (key, value) in url.query_pairs() {
+            if whitelist.contains(&key.as_ref()) {
+                params.push("--http-query-param".to_string());
+                params.push(format!("{}={}", key, value));
+            }
+        }
+
+        Ok(params)
+    }
+
+    /// 等待HTTP服务器启动
+    async fn wait_for_startup(&self, child: &mut Child) -> AppResult<()> {
+        for _ in 0..5 {
+            if child.try_wait().change_context(AppError::Unknown)?.is_some() {
+                return Err(AppError::Unknown.into_report());
+            }
+            sleep(Duration::from_secs(1)).await;
+        }
+        Ok(())
+    }
+
+    fn get_twitch_auth_token() -> Option<String> {
+        // 从配置文件或环境变量读取
+        std::env::var("TWITCH_AUTH_TOKEN").ok()
+    }
+}
+
+/// Streamlink输出类型
+pub enum StreamOutput {
+    /// 管道输出（直接读取stdout）
+    Pipe(Child),
+    /// HTTP服务器输出
+    Http { url: String, process: Child },
+}
+
+impl StreamOutput {
+    /// 获取可读的输入源（用于FFmpeg等）
+    pub fn get_input_uri(&self) -> String {
+        match self {
+            StreamOutput::Pipe(_) => "pipe:0".to_string(),
+            StreamOutput::Http { url, .. } => url.clone(),
+        }
+    }
+
+    /// 获取stdout（仅管道模式）
+    pub fn take_stdout(&mut self) -> Option<std::process::ChildStdout> {
+        match self {
+            StreamOutput::Pipe(child) => child.stdout.take(),
+            StreamOutput::Http { .. } => None,
+        }
+    }
+}
+
+impl Drop for StreamOutput {
+    fn drop(&mut self) {
+        let child = match self {
+            StreamOutput::Pipe(ref mut c) => c,
+            StreamOutput::Http { ref mut process, .. } => process,
+        };
+
+        let _ = child.kill(); // 强制终止
+        let _ = child.wait(); // 回收资源
+    }
+}

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

@@ -1,19 +1,54 @@
 use crate::server::errors::{AppError, AppResult};
-use error_stack::{IntoReport, ResultExt, bail, report};
+use error_stack::{IntoReport, ResultExt, bail, report, Report};
 use regex::Regex;
 use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 use std::sync::Arc;
+use async_trait::async_trait;
 use tokio::net::TcpListener;
 use tokio::process::{Child, Command};
 use tokio::sync::RwLock;
 use tokio::time::{Duration, sleep};
 use tracing::{error, warn};
+use crate::server::core::downloader::Downloader;
+use crate::server::core::plugin::{DownloadPlugin, StreamStatus};
+use crate::server::infrastructure::context::Context;
 
 // 常量定义
 const CLIENT_ID: &str = "kimne78kx3ncx6brgo4mv6wki5h1ko";
 
+pub struct Twitch {
+    re: Regex,
+}
+
+impl Twitch {
+    fn new() -> Twitch {
+        Twitch {
+            re: Regex::new(r"https?://(?:(?:www|go|m)\.)?twitch\.tv/(?P<id>[0-9_a-zA-Z]+)")
+            .unwrap(),
+        }
+    }
+}
+
+#[async_trait]
+impl DownloadPlugin for Twitch {
+    fn matches(&self, url: &str) -> bool {
+        self.re.is_match(url)
+    }
+
+    async fn check_status(&self, ctx: &mut Context) -> Result<StreamStatus, Report<AppError>> {
+        todo!()
+    }
+
+    fn danmaku_init(&self) -> Option<Box<dyn Downloader>> {
+        None
+    }
+
+    fn name(&self) -> &str {
+        "Twitch"
+    }
+}
 // GraphQL 响应结构
 #[derive(Debug, Deserialize)]
 struct GqlResponse {
@@ -50,8 +85,7 @@ struct PlaybackAccessToken {
 }
 
 // Twitch 下载器
-pub struct Twitch {
-    re: Regex,
+pub struct TwitchDownloader {
     fname: String,
     url: String,
     suffix: String,
@@ -65,7 +99,7 @@ pub struct Twitch {
     danmaku: Option<DanmakuClient>,
 }
 
-impl Twitch {
+impl TwitchDownloader {
     pub fn new(fname: String, url: String, suffix: Option<String>) -> Self {
         Self {
             fname,
@@ -79,8 +113,6 @@ 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(),
         }
     }