From 58e7d3c80fbd5777957fc1e666d8e0ea9b923e74 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Wed, 5 Nov 2025 01:44:31 +0800 Subject: [PATCH] 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 | 4 +- Cargo.toml | 2 +- crates/biliup-cli/src/server/api/endpoints.rs | 2 - .../biliup-cli/src/server/core/downloader.rs | 1 + .../src/server/core/downloader/streamlink.rs | 218 ++++++++++++++++++ .../src/server/core/plugin/twitch.rs | 44 +++- 6 files changed, 260 insertions(+), 11 deletions(-) create mode 100644 crates/biliup-cli/src/server/core/downloader/streamlink.rs diff --git a/Cargo.lock b/Cargo.lock index 8973b8b..51bcae8 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index e1ce7b9..0aef312 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index 6da82e0..3d4ddab 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/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!("非预期状态") } }; } diff --git a/crates/biliup-cli/src/server/core/downloader.rs b/crates/biliup-cli/src/server/core/downloader.rs index 8b7f24f..453acd3 100644 --- a/crates/biliup-cli/src/server/core/downloader.rs +++ b/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; diff --git a/crates/biliup-cli/src/server/core/downloader/streamlink.rs b/crates/biliup-cli/src/server/core/downloader/streamlink.rs new file mode 100644 index 0000000..2a05f93 --- /dev/null +++ b/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, + output_mode: OutputMode, + proc: Option, +} + +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) -> 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 { + 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> { + 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> { + 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 { + // 从配置文件或环境变量读取 + 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 { + 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(); // 回收资源 + } +} \ No newline at end of file diff --git a/crates/biliup-cli/src/server/core/plugin/twitch.rs b/crates/biliup-cli/src/server/core/plugin/twitch.rs index 86b39cb..fbf9d15 100644 --- a/crates/biliup-cli/src/server/core/plugin/twitch.rs +++ b/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[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> { + todo!() + } + + fn danmaku_init(&self) -> Option> { + 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, } -impl Twitch { +impl TwitchDownloader { pub fn new(fname: String, url: String, suffix: Option) -> 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[0-9_a-zA-Z]+)") - .unwrap(), } }