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