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

@@ -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",

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"] }

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<String>,
+    _next: Option<String>,
 }
 
 pub fn router() -> Router<()> {

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<ServiceRegister>,
     State(managers): State<Arc<DownloadManager>>,
     State(pool): State<ConnectionPool>,
     Path(id): Path<i64>,
@@ -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<ServiceRegister>,
     State(managers): State<Arc<DownloadManager>>,
-    State(pool): State<ConnectionPool>,
     Path(id): Path<i64>,
 ) -> Result<Json<()>, 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);

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);

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<SegmentInfo>, Receiver<SegmentInfo>)>,
+    channel: Option<Sender<SegmentInfo>>,
     uploader: Sender<UploaderMessage>,
     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<dyn DownloadBase>,
         rooms_handle: Arc<Monitor>,
     ) -> 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<Arc<dyn DanmakuClient + Send + Sync>>,
+        ext: &StreamInfoExt,
     ) -> AppResult<DownloadStatus> {
         // 获取配置和主播信息
-        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)
                 );
             }
         }

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<F>(
     rx: Inspect<Receiver<SegmentInfo>, 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<UploadContext> {
     // 登录处理
     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<Line> {
+async fn get_upload_line(client: &reqwest::Client, line: &str) -> AppResult<Line> {
     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" => 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<Video> {
-    let bilibili = &context.bilibili;
-    let limit = context.threads;
-    let client = &context.client;
-    let line = &context.line;
     let video_path = file_path;
+    let UploadContext {
+        bilibili,
+        line,
+        threads: limit,
+        client,
+    } = context;
 
     info!(
         "开始上传文件：{:?}",
@@ -172,7 +176,7 @@ async fn upload_single_file(file_path: &Path, context: &UploadContext) -> AppRes
     let instant = Instant::now();
 
     let video = uploader
-        .upload(client.clone(), limit, |vs| {
+        .upload(client.clone(), *limit, |vs| {
             vs.map(|vs| {
                 let chunk = vs?;
                 let len = chunk.len();
@@ -230,7 +234,7 @@ pub(crate) async fn build_studio(
     upload_config: &UploadStreamer,
     bilibili: &BiliBili,
     videos: Vec<Video>,
-    recorder: Recorder,
+    recorder: &Recorder,
 ) -> AppResult<Studio> {
     // 使用 Builder 模式简化构建
     let mut studio: Studio = Studio::builder()
@@ -275,9 +279,9 @@ pub(crate) async fn build_studio(
 }
 
 pub async fn execute_postprocessor(video_paths: Vec<PathBuf>, ctx: &Context) -> AppResult<()> {
-    if let Some(processor) = ctx.worker.get_streamer().postprocessor {
+    if let Some(processor) = &ctx.live_streamer().postprocessor {
         let paths: Vec<&Path> = video_paths.iter().map(|p| p.as_path()).collect();
-        process_video(&paths, &processor).await?;
+        process_video(&paths, processor).await?;
     }
     Ok(())
 }
@@ -388,12 +392,11 @@ impl UActor {
     async fn handle_message(&mut self, msg: UploaderMessage) {
         match msg {
             UploaderMessage::SegmentEvent(rx, ctx) => {
-                ctx.worker
-                    .change_status(Stage::Upload, WorkerStatus::Pending)
+                ctx.change_status(Stage::Upload, WorkerStatus::Pending)
                     .await;
                 let inspect = rx.inspect(|f| {
-                    let pool = ctx.pool.clone();
-                    let streamer_info_id = ctx.stream_info.streamer_info.id;
+                    let pool = ctx.pool().clone();
+                    let streamer_info_id = ctx.id();
                     let file = f.prev_file_path.display().to_string();
                     tokio::spawn(async move {
                         let result = InsertFileItem {
@@ -405,7 +408,7 @@ impl UActor {
                         info!(result=?result, "Insert file");
                     });
                 });
-                let result = match ctx.worker.get_upload_config() {
+                let result = match ctx.upload_config() {
                     Some(config) => process_with_upload(inspect, &ctx, config).await,
                     None => {
                         let mut paths = Vec::new();
@@ -422,10 +425,8 @@ impl UActor {
                     error!("Process segment event failed: {}", e);
                     // 可以添加错误通知机制
                 }
-                info!(url=ctx.stream_info.streamer_info.url, result=?result, "后处理执行完毕：Finished processing segment event");
-                ctx.worker
-                    .change_status(Stage::Upload, WorkerStatus::Idle)
-                    .await;
+                info!(url=ctx.live_streamer().url, result=?result, "后处理执行完毕：Finished processing segment event");
+                ctx.change_status(Stage::Upload, WorkerStatus::Idle).await;
             }
         }
     }

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

@@ -11,16 +11,13 @@ pub struct Recorder {
     pub filename_prefix: Option<String>,
     /// 直播间信息
     pub streamer_info: StreamerInfo,
-    /// 录制后保存文件格式 (mp4, ts, mkv, flv)，不含点
-    pub suffix: String,
 }
 
 impl Recorder {
-    pub fn new(filename_prefix: Option<String>, streamer_info: StreamerInfo, suffix: &str) -> Self {
+    pub fn new(filename_prefix: Option<String>, streamer_info: StreamerInfo) -> Self {
         Self {
             filename_prefix,
             streamer_info,
-            suffix: suffix.to_string(),
         }
     }
 
@@ -42,13 +39,13 @@ impl Recorder {
     }
 
     /// 生成“基名”（不带扩展名），时间冲突时按秒+1继续尝试，直到唯一
-    pub fn generate_filename(&self) -> String {
+    pub fn generate_filename(&self, suffix: &str) -> String {
         let template = self.filename_template();
         let mut t = Local::now();
 
         loop {
             let base = t.format(&template).to_string();
-            if !self.exists_with_suffix(&base) {
+            if !self.exists_with_suffix(&base, suffix) {
                 return base;
             }
             t += Duration::seconds(1);
@@ -74,12 +71,12 @@ impl Recorder {
     }
 
     /// 直接生成带扩展名的完整路径（当前目录下）
-    pub fn generate_path(&self) -> PathBuf {
-        PathBuf::from(self.generate_filename()).with_extension(&self.suffix)
+    pub fn generate_path(&self, suffix: &str) -> PathBuf {
+        PathBuf::from(self.generate_filename(suffix)).with_extension(suffix)
     }
 
-    fn exists_with_suffix(&self, base: &str) -> bool {
-        Path::new(base).with_extension(&self.suffix).exists()
+    fn exists_with_suffix(&self, base: &str, suffix: &str) -> bool {
+        Path::new(base).with_extension(&suffix).exists()
     }
 }
 

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

@@ -35,6 +35,8 @@ pub struct DownloadConfig {
 
     /// 输出目录路径
     pub output_dir: PathBuf,
+
+    pub suffix: String,
 }
 
 impl DownloadConfig {
@@ -42,8 +44,8 @@ impl DownloadConfig {
     ///
     /// # 返回
     /// 返回完整的输出文件路径
-    fn generate_output_filename(&self) -> PathBuf {
-        self.output_dir.join(self.recorder.generate_path())
+    fn generate_output_filename(&self, suffix: &str) -> PathBuf {
+        self.output_dir.join(self.recorder.generate_path(suffix))
     }
 }
 

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

@@ -15,8 +15,6 @@ use tracing::info;
 /// FFmpeg下载器实现
 /// 使用FFmpeg进行直播流下载，支持内部和外部分段
 pub struct FfmpegDownloader {
-    /// 下载状态
-    status: Arc<RwLock<DownloadStatus>>,
     /// 进程句柄
     process_handle: Arc<RwLock<Option<tokio::process::Child>>>,
 
@@ -37,7 +35,6 @@ impl FfmpegDownloader {
     /// * `downloader_type` - 下载器类型
     pub fn new(extra_args: Vec<String>, downloader_type: DownloaderType) -> Self {
         Self {
-            status: Arc::new(RwLock::new(DownloadStatus::Downloading)),
             process_handle: Arc::new(RwLock::new(None)),
             extra_args,
             downloader_type,
@@ -60,7 +57,7 @@ impl FfmpegDownloader {
         args.extend(["-f".to_string(), "segment".to_string()]);
         args.extend([
             "-segment_format".to_string(),
-            download_config.recorder.suffix.to_string(),
+            download_config.suffix.to_string(),
         ]);
         // -segment_list pipe:1: 将分段文件名输出到stdout
         // 这样我们可以实时获取新生成的分段文件
@@ -113,7 +110,7 @@ impl FfmpegDownloader {
         }
 
         // 添加通用输出参数
-        self.append_common_output_args(&mut args, &download_config.recorder.suffix);
+        self.append_common_output_args(&mut args, &download_config.suffix);
 
         args
     }
@@ -194,7 +191,7 @@ impl FfmpegDownloader {
         download_config: DownloadConfig,
     ) -> AppResult<DownloadStatus> {
         let args = self.build_ffmpeg_args_external_segment(&download_config);
-        let output_file = download_config.generate_output_filename();
+        let output_file = download_config.generate_output_filename(&download_config.suffix);
 
         let mut cmd = Command::new("ffmpeg");
         cmd.args(&args)
@@ -243,7 +240,7 @@ impl FfmpegDownloader {
             .arg(format!(
                 "{}.{}.part",
                 download_config.recorder.filename_template(),
-                download_config.recorder.suffix
+                download_config.suffix
             ))
             .stdin(Stdio::null())
             .stdout(Stdio::piped())

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

@@ -242,21 +242,15 @@ impl StreamOutput {
         }
     }
 
-    pub fn stop(&mut self) {
+    pub async fn stop(&mut self) {
         info!("准备停止stream terminated");
         let child = match self {
             StreamOutput::Pipe(c) => c,
             StreamOutput::Http { process, .. } => process,
         };
 
-        let _ = child.kill(); // 强制终止
-        let _ = child.wait(); // 回收资源
+        let _ = child.kill().await; // 强制终止
+        let _ = child.wait().await; // 回收资源
         info!("成功stream terminated");
     }
 }
-
-impl Drop for StreamOutput {
-    fn drop(&mut self) {
-        self.stop()
-    }
-}

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

@@ -2,10 +2,9 @@ use crate::server::common::download::DownloaderMessage;
 use crate::server::common::util::Recorder;
 use crate::server::core::plugin::{DownloadPlugin, StreamStatus};
 use crate::server::infrastructure::connection_pool::ConnectionPool;
-use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus};
+use crate::server::infrastructure::context::{Context, PluginContext, Stage, Worker, WorkerStatus};
 use crate::server::infrastructure::models::StreamerInfo;
 use async_channel::Sender;
-use futures::SinkExt;
 use ormlite::Model;
 use ormlite::model::ModelBuilder;
 use std::collections::hash_map::Entry;
@@ -23,7 +22,6 @@ pub struct Monitor {
     /// 消息发送器
     sender: tokio::sync::mpsc::Sender<ActorMessage>,
     /// Actor任务句柄
-    kill: JoinHandle<()>,
     pool: ConnectionPool,
     /// 下载消息发送器
     down_sender: Sender<DownloaderMessage>,
@@ -55,11 +53,10 @@ impl Monitor {
         let (sender, receiver) = tokio::sync::mpsc::channel(1);
         let mut actor = RoomsActor::new(receiver);
         // 启动Actor任务
-        let kill = tokio::spawn(async move { actor.run().await });
+        let _kill = tokio::spawn(async move { actor.run().await });
 
         Self {
             sender,
-            kill,
             pool,
             down_sender,
             monitors: Default::default(),
@@ -84,9 +81,9 @@ impl Monitor {
             // 更新状态为等待中
             room.change_status(Stage::Download, WorkerStatus::Pending)
                 .await;
-            let url = room.get_streamer().url;
+            let url = room.get_streamer().url.clone();
             let interval = room.get_config().event_loop_interval;
-            let mut ctx = Context::new(room.clone(), self.pool.clone());
+            let mut ctx = PluginContext::new(room.clone(), self.pool.clone());
             // 检查直播状态
             let mut downloader = plugin.create_downloader(&mut ctx);
             match downloader.check_stream().await {
@@ -98,7 +95,7 @@ impl Monitor {
                         .title(sql_no_id.title.clone())
                         .date(sql_no_id.date)
                         .live_cover_path(sql_no_id.live_cover_path.clone())
-                        .insert(&ctx.pool)
+                        .insert(ctx.pool())
                         .await
                     {
                         Ok(insert) => insert,
@@ -109,28 +106,16 @@ impl Monitor {
                     };
                     info!(url = url, "room: is live -> 开播了");
 
-                    stream_info.streamer_info = insert;
-
-                    let streamer = room.get_streamer();
-                    // 确定文件格式后缀
-                    let suffix = streamer
-                        .format
-                        .unwrap_or_else(|| stream_info.suffix.clone());
-                    // 创建录制器
-                    let recorder = Recorder::new(
-                        streamer
-                            .filename_prefix
-                            .or(room.get_config().filename_prefix.clone()),
-                        stream_info.streamer_info.clone(),
-                        &suffix,
-                    );
                     // 修改 ctx
-                    ctx.stream_info = *stream_info;
-                    ctx.recorder = recorder;
+                    // stream_info.streamer_info = insert;
+                    let context = ctx.to_context(insert.id, *stream_info);
+                    // context
+                    // *ctx.mut_stream_info_ext() = *stream_info;
+
                     // 发送下载开始消息
                     if self
                         .down_sender
-                        .send(DownloaderMessage::Start(downloader, ctx))
+                        .send(DownloaderMessage::Start(downloader, context))
                         .await
                         .is_ok()
                     {
@@ -139,11 +124,11 @@ impl Monitor {
                 }
                 Ok(StreamStatus::Offline) => {
                     self.wake_waker(room.id()).await;
-                    debug!(url = ctx.worker.live_streamer.url, "未开播")
+                    debug!(url = ctx.live_streamer().url, "未开播")
                 }
                 Err(e) => {
                     self.wake_waker(room.id()).await;
-                    error!(e=?e, ctx=ctx.worker.live_streamer.url,"检查直播间出错")
+                    error!(e=?e, ctx=ctx.live_streamer().url,"检查直播间出错")
                 }
             };
             // 等待下一次检查
@@ -407,9 +392,9 @@ impl RoomsActor {
     }
 
     /// 运行Actor主循环
+    /// 处理接收到的消息
     async fn run(&mut self) {
         while let Some(msg) = self.receiver.recv().await {
-            /// 处理接收到的消息
             match msg {
                 ActorMessage::NextRoom {
                     respond_to,

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

@@ -3,7 +3,7 @@ pub mod yy;
 
 use crate::server::core::downloader::{DanmakuClient, DownloaderRuntime, DownloaderType};
 use crate::server::errors::AppError;
-use crate::server::infrastructure::context::Context;
+use crate::server::infrastructure::context::{Context, PluginContext};
 use crate::server::infrastructure::models::StreamerInfo;
 use async_trait::async_trait;
 use error_stack::Report;
@@ -62,7 +62,7 @@ pub trait DownloadPlugin {
     fn matches(&self, url: &str) -> bool;
 
     /// 创建下载器实例
-    fn create_downloader(&self, ctx: &mut Context) -> Box<dyn DownloadBase>;
+    fn create_downloader(&self, ctx: &mut PluginContext) -> Box<dyn DownloadBase>;
 
     /// 获取插件名称
     fn name(&self) -> &str;

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

@@ -5,7 +5,7 @@ use crate::server::core::downloader::streamlink::{Platform, Streamlink, Streamli
 use crate::server::core::downloader::{DanmakuClient, DownloaderRuntime, DownloaderType};
 use crate::server::core::plugin::{DownloadBase, DownloadPlugin, StreamInfoExt, StreamStatus};
 use crate::server::errors::{AppError, AppResult};
-use crate::server::infrastructure::context::Context;
+use crate::server::infrastructure::context::{Context, PluginContext};
 use crate::server::infrastructure::models::StreamerInfo;
 use async_trait::async_trait;
 use chrono::Utc;
@@ -68,10 +68,10 @@ impl DownloadPlugin for Twitch {
         self.re.is_match(url)
     }
 
-    fn create_downloader(&self, ctx: &mut Context) -> Box<dyn DownloadBase> {
+    fn create_downloader(&self, ctx: &mut PluginContext) -> Box<dyn DownloadBase> {
         Box::new(TwitchDownloader::new(
-            &ctx.worker.live_streamer.remark,
-            ctx.worker.live_streamer.url.clone(),
+            &ctx.live_streamer().remark,
+            ctx.live_streamer().url.clone(),
             self.re.clone(),
         ))
     }

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

@@ -1,7 +1,7 @@
 use crate::server::common::util::media_ext_from_url;
 use crate::server::core::plugin::{DownloadBase, DownloadPlugin, StreamInfoExt, StreamStatus};
 use crate::server::errors::AppError;
-use crate::server::infrastructure::context::Context;
+use crate::server::infrastructure::context::{Context, PluginContext};
 use crate::server::infrastructure::models::StreamerInfo;
 use async_trait::async_trait;
 use chrono::Utc;
@@ -177,15 +177,11 @@ impl DownloadPlugin for YY {
         self.re.is_match(url)
     }
 
-    fn create_downloader(&self, ctx: &mut Context) -> Box<dyn DownloadBase> {
-        let url = ctx.worker.live_streamer.url.to_string();
-        let name = ctx.worker.live_streamer.remark.to_string();
+    fn create_downloader(&self, ctx: &mut PluginContext) -> Box<dyn DownloadBase> {
+        let url = ctx.live_streamer().url.to_string();
+        let name = ctx.live_streamer().remark.to_string();
 
-        Box::new(YYDownloader::new(
-            ctx.worker.client.client.clone(),
-            &url,
-            &name,
-        ))
+        Box::new(YYDownloader::new(ctx.client(), &url, &name))
     }
 
     fn name(&self) -> &str {

crates/biliup-cli/src/server/infrastructure/context.rs

@@ -1,14 +1,16 @@
 use crate::server::common::download::DownloadTask;
 use crate::server::common::util::Recorder;
-use crate::server::config::{Config, ConfigPatch};
+use crate::server::config::{Config, default_segment_time};
+use crate::server::core::downloader::DownloadConfig;
 use crate::server::core::plugin::StreamInfoExt;
 use crate::server::infrastructure::connection_pool::ConnectionPool;
+use crate::server::infrastructure::models::StreamerInfo;
 use crate::server::infrastructure::models::live_streamer::LiveStreamer;
 use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
 use axum::http::Extensions;
 use biliup::client::StatelessClient;
 use core::fmt;
-use ormlite::Model;
+use std::path::PathBuf;
 use std::sync::{Arc, RwLock};
 use struct_patch::Patch;
 use tracing::{error, info};
@@ -16,13 +18,13 @@ use tracing::{error, info};
 /// 应用程序上下文，包含工作器和扩展信息
 #[derive(Debug, Clone)]
 pub struct Context {
+    id: i64,
     /// 工作器实例
-    pub worker: Arc<Worker>,
-    pub stream_info: StreamInfoExt,
+    worker: Arc<Worker>,
+    stream_info: StreamInfoExt,
     /// 扩展数据容器
-    pub extension: Extensions,
-    pub pool: ConnectionPool,
-    pub recorder: Recorder,
+    extension: Extensions,
+    pool: ConnectionPool,
 }
 
 impl Context {
@@ -30,13 +32,94 @@ impl Context {
     ///
     /// # 参数
     /// * `worker` - 工作器实例的Arc引用
-    pub fn new(worker: Arc<Worker>, pool: ConnectionPool) -> Self {
+    pub fn new(
+        id: i64,
+        worker: Arc<Worker>,
+        pool: ConnectionPool,
+        stream_info: StreamInfoExt,
+        extension: Extensions,
+    ) -> Self {
         Self {
+            id,
             worker,
-            stream_info: Default::default(),
-            extension: Default::default(),
+            stream_info,
+            extension,
             pool,
-            recorder: Default::default(),
+        }
+    }
+
+    pub fn worker_id(&self) -> i64 {
+        self.worker.id()
+    }
+
+    pub fn id(&self) -> i64 {
+        self.id
+    }
+
+    pub fn live_streamer(&self) -> &LiveStreamer {
+        &self.worker.get_streamer()
+    }
+
+    pub fn stateless_client(&self) -> &StatelessClient {
+        &self.worker.client
+    }
+
+    pub fn config(&self) -> Config {
+        self.worker.get_config()
+    }
+
+    pub fn pool(&self) -> &ConnectionPool {
+        &self.pool
+    }
+
+    pub async fn change_status(&self, stage: Stage, status: WorkerStatus) {
+        self.worker.change_status(stage, status).await;
+    }
+
+    pub fn status(&self, stage: Stage) -> WorkerStatus {
+        match stage {
+            Stage::Download => self.worker.downloader_status.read().unwrap().clone(),
+            Stage::Upload => self.worker.uploader_status.read().unwrap().clone(),
+        }
+    }
+
+    pub fn upload_config(&self) -> &Option<UploadStreamer> {
+        self.worker.get_upload_config()
+    }
+
+    pub fn recorder(&self, streamer_info: StreamerInfo) -> Recorder {
+        // 创建录制器
+        Recorder::new(
+            self.live_streamer()
+                .filename_prefix
+                .clone()
+                .or(self.config().filename_prefix.clone()),
+            streamer_info,
+        )
+    }
+
+    pub fn stream_info_ext(&self) -> &StreamInfoExt {
+        &self.stream_info
+    }
+
+    pub fn download_config(&self, ext: &StreamInfoExt) -> DownloadConfig {
+        let config = self.config();
+        // 确定文件格式后缀
+        let suffix = self
+            .live_streamer()
+            .format
+            .clone()
+            .unwrap_or_else(|| ext.suffix.to_string());
+        DownloadConfig {
+            // 流URL
+            url: ext.raw_stream_url.to_string(),
+            segment_time: config.segment_time.or_else(default_segment_time),
+            file_size: Some(config.file_size), // 2GB
+            headers: ext.stream_headers.clone(),
+            recorder: self.recorder(ext.streamer_info.clone()),
+            // output_dir: PathBuf::from("./downloads")
+            output_dir: PathBuf::from("."),
+            suffix,
         }
     }
 }
@@ -93,14 +176,14 @@ impl Worker {
 
     /// 获取主播信息
     /// 返回当前工作器关联的直播主播信息
-    pub fn get_streamer(&self) -> LiveStreamer {
-        self.live_streamer.clone()
+    pub fn get_streamer(&self) -> &LiveStreamer {
+        &self.live_streamer
     }
 
     /// 获取上传配置
     /// 返回当前工作器的上传配置（如果存在）
-    pub fn get_upload_config(&self) -> Option<UploadStreamer> {
-        self.upload_streamer.clone()
+    pub fn get_upload_config(&self) -> &Option<UploadStreamer> {
+        &self.upload_streamer
     }
 
     /// 获取覆写配置
@@ -201,3 +284,52 @@ impl fmt::Debug for WorkerStatus {
         f.write_str(name)
     }
 }
+
+/// 应用程序上下文，包含工作器和扩展信息
+#[derive(Debug, Clone)]
+pub struct PluginContext {
+    /// 工作器实例
+    worker: Arc<Worker>,
+    pool: ConnectionPool,
+    extension: Extensions,
+}
+
+impl PluginContext {
+    pub fn new(worker: Arc<Worker>, pool: ConnectionPool) -> Self {
+        Self {
+            worker,
+            pool,
+            extension: Default::default(),
+        }
+    }
+
+    pub fn to_context(&self, id: i64, stream_info: StreamInfoExt) -> Context {
+        Context::new(
+            id,
+            self.worker.clone(),
+            self.pool.clone(),
+            stream_info,
+            self.extension.clone(),
+        )
+    }
+
+    pub fn config(&self) -> Config {
+        self.worker.get_config()
+    }
+
+    pub fn live_streamer(&self) -> &LiveStreamer {
+        &self.worker.get_streamer()
+    }
+
+    pub fn upload_config(&self) -> &Option<UploadStreamer> {
+        self.worker.get_upload_config()
+    }
+
+    pub fn pool(&self) -> &ConnectionPool {
+        &self.pool
+    }
+
+    pub fn client(&self) -> reqwest::Client {
+        self.worker.client.client.clone()
+    }
+}

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

@@ -4,7 +4,7 @@ use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
 use std::path::Path;
 use tokio::fs;
-use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
+use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 use tokio::process::Command;
 use tracing::{error, info};
 

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

@@ -1,4 +1,4 @@
-use crate::server::config::{Config, ConfigPatch};
+use crate::server::config::ConfigPatch;
 use crate::server::infrastructure::models::hook_step::HookStep;
 use ormlite::{Insert, Model};
 use serde::{Deserialize, Serialize};

crates/stream-gears/src/server.rs

@@ -11,7 +11,7 @@ use biliup_cli::server::core::downloader::DanmakuClient;
 use biliup_cli::server::core::plugin::{DownloadBase, DownloadPlugin, StreamInfoExt, StreamStatus};
 use biliup_cli::server::errors::{AppError, AppResult};
 use biliup_cli::server::infrastructure::connection_pool::ConnectionManager;
-use biliup_cli::server::infrastructure::context::{Context, Worker};
+use biliup_cli::server::infrastructure::context::{Context, PluginContext, Worker};
 use biliup_cli::server::infrastructure::models::StreamerInfo;
 use biliup_cli::server::infrastructure::repositories;
 use biliup_cli::server::infrastructure::repositories::get_upload_config;
@@ -19,7 +19,7 @@ use biliup_cli::server::infrastructure::service_register::ServiceRegister;
 use biliup_cli::uploader::{append, list, login, renew, show, upload_by_command, upload_by_config};
 use chrono::Utc;
 use clap::Parser;
-use error_stack::{FutureExt, Report, ResultExt};
+use error_stack::{Report, ResultExt};
 use fancy_regex::Regex;
 use pyo3::prelude::PyDictMethods;
 use pyo3::prelude::{PyAnyMethods, PyListMethods, PyModule};
@@ -135,7 +135,7 @@ impl PyDownloader {
                                 self_obj.getattr("stream_headers")?.extract()?
                             };
 
-                            let danmaku_init = self_obj.call_method0("danmaku_init")?;
+                            let _danmaku_init = self_obj.call_method0("danmaku_init")?;
                             // let platform: Option<PyAny> = self_obj.getattr("danmaku")?.extract()?;
                             // danmaku 可能在条件下没有设置（比如 bilibili_danmaku 为 False）
                             let self_danmaku = self_obj.getattr("danmaku")?;
@@ -197,14 +197,14 @@ impl DownloadPlugin for PyPlugin {
         false
     }
 
-    fn create_downloader(&self, ctx: &mut Context) -> Box<dyn DownloadBase> {
-        let url = ctx.worker.live_streamer.url.to_string();
-        let remark = ctx.worker.live_streamer.remark.to_string();
+    fn create_downloader(&self, ctx: &mut PluginContext) -> Box<dyn DownloadBase> {
+        let url = ctx.live_streamer().url.to_string();
+        let remark = ctx.live_streamer().remark.to_string();
         Box::new(PyDownloader::new(
             self.plugin.clone(),
             url,
             remark,
-            ctx.worker.get_config(),
+            ctx.config(),
         ))
     }