From 08c8370c49c088f5bbe26131dfdb1663611d5712 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Fri, 5 Dec 2025 09:09:31 +0800 Subject: [PATCH] Refactor context and downloader logic for modularity This commit refactors the context handling in the server, introducing PluginContext and improving separation between plugin and runtime contexts. It updates downloader and uploader logic to use the new context structure, streamlines file suffix handling, and removes redundant fields and code. The changes enhance code maintainability, modularity, and clarity, especially around download/upload workflows and plugin integration. --- Cargo.lock | 4 +- Cargo.toml | 2 +- crates/biliup-cli/src/server/api/auth.rs | 2 +- crates/biliup-cli/src/server/api/endpoints.rs | 6 +- crates/biliup-cli/src/server/app.rs | 4 +- .../biliup-cli/src/server/common/download.rs | 68 +++----- crates/biliup-cli/src/server/common/upload.rs | 71 ++++---- crates/biliup-cli/src/server/common/util.rs | 17 +- .../biliup-cli/src/server/core/downloader.rs | 6 +- .../core/downloader/ffmpeg_downloader.rs | 11 +- .../src/server/core/downloader/streamlink.rs | 12 +- crates/biliup-cli/src/server/core/monitor.rs | 43 ++--- crates/biliup-cli/src/server/core/plugin.rs | 4 +- .../src/server/core/plugin/twitch.rs | 8 +- .../biliup-cli/src/server/core/plugin/yy.rs | 14 +- .../src/server/infrastructure/context.rs | 162 ++++++++++++++++-- .../server/infrastructure/models/hook_step.rs | 2 +- .../infrastructure/models/live_streamer.rs | 2 +- crates/stream-gears/src/server.rs | 14 +- 19 files changed, 269 insertions(+), 183 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c85adf9..babf9a3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -362,7 +362,7 @@ dependencies = [ [[package]] name = "biliup-cli" -version = "1.1.27" +version = "1.1.28" dependencies = [ "async-channel", "async-trait", @@ -3693,7 +3693,7 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" [[package]] name = "stream-gears" -version = "1.1.27" +version = "1.1.28" dependencies = [ "async-channel", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index c7e4b1f..a125b61 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ members = [ resolver = "3" [workspace.package] -version = "1.1.27" +version = "1.1.28" [workspace.dependencies] serde = { version = "1.0", features = ["derive"] } diff --git a/crates/biliup-cli/src/server/api/auth.rs b/crates/biliup-cli/src/server/api/auth.rs index 219929d..edcbc6d 100644 --- a/crates/biliup-cli/src/server/api/auth.rs +++ b/crates/biliup-cli/src/server/api/auth.rs @@ -12,7 +12,7 @@ use crate::server::infrastructure::users::{AuthSession, Credentials}; // to redirect after log in. #[derive(Debug, Deserialize)] pub struct NextUrl { - next: Option, + _next: Option, } pub fn router() -> Router<()> { diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index 43fb392..819e7ec 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -124,7 +124,6 @@ pub async fn put_streamers_endpoint( } pub async fn delete_streamers_endpoint( - State(service_register): State, State(managers): State>, State(pool): State, Path(id): Path, @@ -138,9 +137,7 @@ pub async fn delete_streamers_endpoint( // #[axum::debug_handler(state = ServiceRegister)] pub async fn pause_streamers_endpoint( - State(service_register): State, State(managers): State>, - State(pool): State, Path(id): Path, ) -> Result, Response> { let worker = managers.get_room_by_id(id).await; @@ -580,9 +577,8 @@ pub async fn post_uploads( Utc::now(), "", ), - "", ); - let studio = build_studio(&upload_config, &bilibili, videos, recorder).await?; + let studio = build_studio(&upload_config, &bilibili, videos, &recorder).await?; let response_data = submit_to_bilibili(&bilibili, &studio, submit_api.as_deref()).await?; info!("通过页面上传成功 {:?}", response_data); diff --git a/crates/biliup-cli/src/server/app.rs b/crates/biliup-cli/src/server/app.rs index ff59780..74d5c7d 100644 --- a/crates/biliup-cli/src/server/app.rs +++ b/crates/biliup-cli/src/server/app.rs @@ -47,7 +47,7 @@ impl ApplicationController { ); // 生成用于签名会话cookie的加密密钥 - let key = Key::generate(); + // let key = Key::generate(); // 配置会话管理层 let session_layer = SessionManagerLayer::new(session_store) @@ -102,7 +102,7 @@ impl ApplicationController { // 等待会话清理任务完成 match deletion_task.await { - Ok(Ok(val)) => { /* 正常完成 */ } + Ok(Ok(())) => { /* 正常完成 */ } Ok(Err(e)) => { // 真正业务错误 return Err(e).change_context(AppError::Unknown); diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index 0ab760b..5081f1f 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -5,7 +5,7 @@ use crate::server::core::downloader::{ SegmentInfo, }; use crate::server::core::monitor::Monitor; -use crate::server::core::plugin::{DownloadBase, StreamStatus}; +use crate::server::core::plugin::{DownloadBase, StreamInfoExt, StreamStatus}; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::context::{Context, Stage, WorkerStatus}; use crate::server::infrastructure::models::hook_step::process; @@ -101,7 +101,7 @@ impl FileValidator { /// 分段事件处理器 pub struct SegmentEventProcessor { - channel: Option<(Sender, Receiver)>, + channel: Option>, uploader: Sender, ctx: Context, file_validator: FileValidator, @@ -114,7 +114,7 @@ impl SegmentEventProcessor { channel: None, uploader, file_validator: FileValidator::new( - ctx.worker.clone().get_config().filtering_threshold * 1000 * 1000, + ctx.config().filtering_threshold * 1000 * 1000, true, ), ctx, @@ -145,9 +145,9 @@ impl SegmentEventProcessor { if let Some(prev) = res { warn!(SegmentEvent = ?prev, "replace an existing message in the channel"); } - self.channel = Some((tx, rx)); + self.channel = Some(tx); } - Some((tx, rx)) => { + Some(tx) => { // 发送到缓冲区 let res = tx .force_send(event) @@ -180,18 +180,18 @@ impl DownloadTask { pub(self) async fn execute( &self, - mut ctx: Context, + ctx: &Context, mut processor: SegmentEventProcessor, mut plugin: Box, rooms_handle: Arc, ) -> AppResult<()> { - let worker = ctx.worker.clone(); // 重试配置 let mut retry_count = 0; let max_retries = 3; // 最大重试次数 let base_delay = Duration::from_secs(2); // 基础延迟时间（2秒） - let max_delay = Duration::from_secs(worker.get_config().delay); // 最大延迟时间（60秒） - let url = worker.live_streamer.url.clone(); + let max_delay = Duration::from_secs(ctx.config().delay); // 最大延迟时间（60秒） + let url = ctx.live_streamer().url.clone(); + let mut stream_info_ext = ctx.stream_info_ext().clone(); // 可选的弹幕客户端 // 初始化弹幕客户端（如果存在） let danmaku_client = plugin.danmaku_init(); @@ -206,7 +206,12 @@ impl DownloadTask { // 创建事件处理器 // 执行下载 let components = self - .initialize_components(&mut processor, ctx.clone(), danmaku_client.clone()) + .initialize_components( + &mut processor, + ctx.clone(), + danmaku_client.clone(), + &stream_info_ext, + ) .await; info!("initialize_components completed: {url}"); @@ -215,11 +220,10 @@ impl DownloadTask { info!(url = url, "task is cancelled"); break components; } - ctx = ctx.clone(); // 检查流状态 match plugin.check_stream().await { Ok(StreamStatus::Live { stream_info }) => { - ctx.stream_info.raw_stream_url = stream_info.raw_stream_url; + stream_info_ext = *stream_info; info!( url = url, "Stream is still live, preparing to retry. attempt: {}", retry_count @@ -276,7 +280,7 @@ impl DownloadTask { } // 清理资源 // 确保状态更新和资源清理 - rooms_handle.wake_waker(worker.id()).await; + rooms_handle.wake_waker(ctx.worker_id()).await; info!("Download task completed: {:?}", result); self.done_notify.notify_one(); Ok(()) @@ -287,12 +291,10 @@ impl DownloadTask { processor: &mut SegmentEventProcessor, ctx: Context, danmaku_client: Option>, + ext: &StreamInfoExt, ) -> AppResult { // 获取配置和主播信息 - let config = ctx.worker.get_config(); - let streamer = ctx.worker.get_streamer(); - let stream_info = &ctx.stream_info; - let raw_stream_url = &stream_info.raw_stream_url; + let streamer = ctx.live_streamer(); // 执行下载 // let hook = processor.create_hook(danmaku_client.clone()); @@ -332,20 +334,9 @@ impl DownloadTask { } }; - let download_config = DownloadConfig { - /// 流URL - url: raw_stream_url.to_string(), - segment_time: config.segment_time.or_else(default_segment_time), - file_size: Some(config.file_size), // 2GB - headers: stream_info.stream_headers.clone(), - recorder: ctx.recorder.clone(), - // output_dir: PathBuf::from("./downloads") - output_dir: PathBuf::from("."), - }; let result = self .downloader - // .downloader(ctx.worker.get_config().downloader)? - .download(Box::new(hook), download_config) + .download(Box::new(hook), ctx.download_config(ext)) .await .change_context(AppError::Custom("Failed to download segment".into()))?; @@ -403,12 +394,9 @@ impl DActor { async fn handle_message(&mut self, msg: DownloaderMessage) { match msg { DownloaderMessage::Start(downloader, ctx) => { - let worker = ctx.worker.clone(); - // 创建下载任务 let runtime = downloader.downloader( - ctx.worker - .get_config() + ctx.config() .downloader .unwrap_or(DownloaderType::StreamGears), ); @@ -416,23 +404,21 @@ impl DActor { // 初始化组件 let processor = SegmentEventProcessor::new(self.sender.clone(), ctx.clone()); // 更新工作器状态为工作中 - worker - .change_status(Stage::Download, WorkerStatus::Working(task.clone())) + ctx.change_status(Stage::Download, WorkerStatus::Working(task.clone())) .await; - process(&[], &ctx.worker.get_streamer().preprocessor).await; + process(&[], &ctx.live_streamer().preprocessor).await; - let option = &ctx.worker.get_streamer().downloaded_processor; let _ = task - .execute(ctx, processor, downloader, self.rooms_handle.clone()) + .execute(&ctx, processor, downloader, self.rooms_handle.clone()) .await; - process(&[], option).await; + process(&[], &ctx.live_streamer().downloaded_processor).await; info!( "Download workflow completed {} => {:?}", - worker.live_streamer.url, - worker.downloader_status.read() + ctx.live_streamer().url, + ctx.status(Stage::Download) ); } } diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs index df31d17..c431c4b 100644 --- a/crates/biliup-cli/src/server/common/upload.rs +++ b/crates/biliup-cli/src/server/common/upload.rs @@ -1,5 +1,6 @@ use crate::UploadLine; use crate::server::common::util::Recorder; +use crate::server::config::Config; use crate::server::core::downloader::SegmentInfo; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus}; @@ -27,10 +28,9 @@ use tracing::{error, info}; // 辅助结构体 struct UploadContext { bilibili: BiliBili, - client: StatelessClient, line: Line, threads: usize, - upload_config: UploadStreamer, + client: StatelessClient, } #[derive(Default)] @@ -42,31 +42,34 @@ struct UploadedVideos { pub async fn process_with_upload( rx: Inspect, F>, ctx: &Context, - upload_config: UploadStreamer, + upload_config: &UploadStreamer, ) -> AppResult<()> where F: FnMut(&SegmentInfo), { info!(upload_config=?upload_config, "Starting process with upload"); // 1. 初始化上传环境 - let upload_context = initialize_upload_context(&ctx.worker, upload_config).await?; + let upload_context = + initialize_upload_context(&ctx.config(), &ctx.stateless_client(), upload_config).await?; // 2. 流水线处理视频上传 let uploaded_videos = pipeline_upload_videos(rx, &upload_context).await?; // 3. 提交到B站 if !uploaded_videos.videos.is_empty() { - let mut recorder = ctx.recorder.clone(); - recorder.filename_prefix = upload_context.upload_config.title.clone(); + let mut recorder = ctx + .recorder(ctx.stream_info_ext().streamer_info.clone()) + .clone(); + recorder.filename_prefix = upload_config.title.clone(); let studio = build_studio( - &upload_context.upload_config, + &upload_config, &upload_context.bilibili, uploaded_videos.videos, - recorder, + &recorder, ) .await?; - let submit_api = ctx.worker.get_config().submit_api.clone(); + let submit_api = ctx.config().submit_api.clone(); submit_to_bilibili(&upload_context.bilibili, &studio, submit_api.as_deref()).await?; } @@ -79,8 +82,9 @@ where } async fn initialize_upload_context( - worker: &Worker, - upload_config: UploadStreamer, + config: &Config, + client: &StatelessClient, + upload_config: &UploadStreamer, ) -> AppResult { // 登录处理 let cookie_file = upload_config @@ -90,21 +94,19 @@ async fn initialize_upload_context( let bilibili = login_by_cookies(&cookie_file, None) .await .change_context(AppError::Unknown)?; - let config = worker.get_config(); // 获取上传线路 - let line = get_upload_line(&worker.client, &config.lines).await?; + let line = get_upload_line(&client.client, &config.lines).await?; Ok(UploadContext { bilibili, - client: worker.client.clone(), line, - threads: worker.get_config().threads as usize, - upload_config, + threads: config.threads as usize, + client: client.clone(), }) } -async fn get_upload_line(client: &StatelessClient, line: &str) -> AppResult { +async fn get_upload_line(client: &reqwest::Client, line: &str) -> AppResult { let line = match line { "bda2" => line::bda2(), "bda" => line::bda(), @@ -112,7 +114,7 @@ async fn get_upload_line(client: &StatelessClient, line: &str) -> AppResult line::txa(), "bldsa" => line::bldsa(), "alia" => line::alia(), - _ => Probe::probe(&client.client).await.unwrap_or_default(), + _ => Probe::probe(client).await.unwrap_or_default(), }; Ok(line) } @@ -147,11 +149,13 @@ where } async fn upload_single_file(file_path: &Path, context: &UploadContext) -> AppResult