Refactor upload pipeline and improve code consistency

Refactored upload pipeline to use stream inspection for file item insertion, improving modularity and resource handling. Cleaned up imports, improved formatting, and removed redundant logging and code. Minor fixes in module ordering and code style for better maintainability.

Cargo.lock

@@ -395,7 +395,7 @@ dependencies = [
 
 [[package]]
 name = "biliup-cli"
-version = "1.1.10"
+version = "1.1.11"
 dependencies = [
  "async-channel",
  "async-trait",
@@ -3641,7 +3641,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
 
 [[package]]
 name = "stream-gears"
-version = "1.1.10"
+version = "1.1.11"
 dependencies = [
  "async-channel",
  "async-trait",

Cargo.toml

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

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

@@ -67,8 +67,6 @@ pub async fn get_proxy_endpoint(
     State(client): State<StatelessClient>,
     Query(params): Query<HashMap<String, String>>,
 ) -> Result<Bytes, Response> {
-    info!(params = &params["url"]);
-
     // 代理HTTP请求
     client
         .client

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

@@ -3,13 +3,17 @@ use crate::server::common::upload::{build_studio, submit_to_bilibili, upload};
 use crate::server::common::util::Recorder;
 use crate::server::config::Config;
 use crate::server::core::download_manager::ActorHandle;
-use crate::server::errors::{report_to_response, AppError};
+use crate::server::core::downloader::Downloader;
+use crate::server::errors::{AppError, report_to_response};
 use crate::server::infrastructure::connection_pool::ConnectionPool;
-use crate::server::infrastructure::context::{Worker, WorkerStatus};
+use crate::server::infrastructure::context::{Stage, Worker, WorkerStatus};
 use crate::server::infrastructure::dto::LiveStreamerResponse;
+use crate::server::infrastructure::models::live_streamer::{InsertLiveStreamer, LiveStreamer};
+use crate::server::infrastructure::models::upload_streamer::{
+    InsertUploadStreamer, UploadStreamer,
+};
 use crate::server::infrastructure::models::{
-    Configuration, FileItem, InsertConfiguration,
-    StreamerInfo,
+    Configuration, FileItem, InsertConfiguration, StreamerInfo,
 };
 use crate::server::infrastructure::repositories::{
     del_streamer, get_all_streamer, get_upload_config,
@@ -19,10 +23,10 @@ use axum::extract::ws::{Message, Utf8Bytes, WebSocket};
 use axum::extract::{Path, Query, State, WebSocketUpgrade};
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
-use axum::{debug_handler, Json};
+use axum::{Json, debug_handler};
 use biliup::credential::Credential;
 use clap::ValueEnum;
-use error_stack::{bail, Report, ResultExt};
+use error_stack::{Report, ResultExt, bail};
 use ormlite::{Insert, Model};
 use serde::Deserialize;
 use serde_json::json;
@@ -34,21 +38,14 @@ use std::sync::{Arc, RwLock};
 use std::time::{Duration, UNIX_EPOCH};
 use tokio::fs;
 use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader};
-use tokio::time::{interval, MissedTickBehavior};
+use tokio::time::{MissedTickBehavior, interval};
 use tracing::{debug, error, info};
-use crate::server::core::downloader::Downloader;
-use crate::server::infrastructure::models::live_streamer::{InsertLiveStreamer, LiveStreamer};
-use crate::server::infrastructure::models::upload_streamer::{InsertUploadStreamer, UploadStreamer};
 
 pub async fn get_streamers_endpoint(
     State(pool): State<ConnectionPool>,
     State(workers): State<Arc<RwLock<Vec<Arc<Worker>>>>>,
 ) -> Result<Json<Vec<LiveStreamerResponse>>, Response> {
     let live_streamers = get_all_streamer(&pool).await.map_err(report_to_response)?;
-    info!(
-        "get_streamers_endpoint found {} live streamers",
-        live_streamers.len()
-    );
     Ok(Json(
         live_streamers
             .into_iter()
@@ -158,31 +155,32 @@ pub async fn pause_streamers_endpoint(
         .unwrap()
         .clone()
         .into_iter()
-        .find(|worker| {
-            worker.live_streamer.id == id
-        });
+        .find(|worker| worker.live_streamer.id == id);
     if let Some(w) = option {
-        let manager = service_register.get_manager(&w.live_streamer.url).ok_or(AppError::Unknown).map_err(report_to_response)?;
+        let manager = service_register
+            .get_manager(&w.live_streamer.url)
+            .ok_or(AppError::Unknown)
+            .map_err(report_to_response)?;
         let monitor = manager.ensure_monitor(pool.clone());
         let worker_status = w.downloader_status.read().unwrap().clone();
         let d = match worker_status {
             WorkerStatus::Working(d) => {
-                *w.downloader_status.write().unwrap() = WorkerStatus::Pause;
+                w.change_status(Stage::Download, WorkerStatus::Pause);
                 Some(d.clone())
-            },
+            }
             WorkerStatus::Pending => {
                 monitor.rooms_handle.toggle(w.clone()).await;
-                *w.downloader_status.write().unwrap() = WorkerStatus::Pause;
+                w.change_status(Stage::Download, WorkerStatus::Pause);
                 None
             }
             WorkerStatus::Idle => {
                 monitor.rooms_handle.toggle(w.clone()).await;
-                *w.downloader_status.write().unwrap() = WorkerStatus::Pause;
+                w.change_status(Stage::Download, WorkerStatus::Pause);
                 None
             }
             WorkerStatus::Pause => {
                 monitor.rooms_handle.toggle(w.clone()).await;
-                *w.downloader_status.write().unwrap() = WorkerStatus::Idle;
+                w.change_status(Stage::Download, WorkerStatus::Idle);
                 None
             }
         };
@@ -190,7 +188,7 @@ pub async fn pause_streamers_endpoint(
         if let Some(d) = d {
             d.stop().await.map_err(report_to_response)?;
         }
-        info!(workers=?service_register.workers, "successfully pause live streamers");
+        info!(workers=?&w.live_streamer.url, "successfully pause live streamers");
     }
 
     Ok(Json(()))

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

@@ -71,8 +71,7 @@ impl Drop for DownloadGuard {
             if let WorkerStatus::Pause = *worker.downloader_status.read().unwrap() {
             } else {
                 // 确保状态更新和资源清理
-                worker
-                    .change_status(Stage::Download, WorkerStatus::Idle);
+                worker.change_status(Stage::Download, WorkerStatus::Idle);
                 rooms_handle.toggle(worker).await;
             };
         });

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

@@ -4,8 +4,9 @@ use crate::server::core::download_manager::UActor;
 use crate::server::core::downloader::{SegmentEvent, SegmentInfo};
 use crate::server::errors::{AppError, AppResult};
 use crate::server::infrastructure::context::{Context, Worker};
-use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
+use crate::server::infrastructure::models::InsertFileItem;
 use crate::server::infrastructure::models::hook_step::process_video;
+use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
 use async_channel::Receiver;
 use biliup::bilibili::{BiliBili, Credit, ResponseData, Studio, Video};
 use biliup::client::StatelessClient;
@@ -17,12 +18,13 @@ use biliup::uploader::{VideoFile, line};
 use chrono::Local;
 use error_stack::ResultExt;
 use futures::StreamExt;
+use futures::stream::Inspect;
+use ormlite::Insert;
 use std::path::{Path, PathBuf};
 use std::str::FromStr;
 use std::time::Instant;
-use ormlite::Insert;
+use tokio::pin;
 use tracing::{error, info};
-use crate::server::infrastructure::models::InsertFileItem;
 
 // 辅助结构体
 struct UploadContext {
@@ -39,11 +41,14 @@ struct UploadedVideos {
     paths: Vec<PathBuf>,
 }
 
-pub async fn process_with_upload(
-    rx: Receiver<SegmentInfo>,
+pub async fn process_with_upload<F>(
+    rx: Inspect<Receiver<SegmentInfo>, F>,
     ctx: &Context,
     upload_config: UploadStreamer,
-) -> AppResult<()> {
+) -> 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?;
@@ -117,10 +122,13 @@ async fn get_upload_line(client: &StatelessClient, line: &str) -> AppResult<Line
     Ok(line)
 }
 
-async fn pipeline_upload_videos(
-    rx: Receiver<SegmentInfo>,
+async fn pipeline_upload_videos<F>(
+    mut rx: Inspect<Receiver<SegmentInfo>, F>,
     context: &UploadContext,
-) -> AppResult<UploadedVideos> {
+) -> AppResult<UploadedVideos>
+where
+    F: FnMut(&SegmentInfo),
+{
     // let mut desc_v2 = Vec::new();
     // for credit in context.upload_config.desc_v2_credit {
     //     desc_v2.push(Credit {
@@ -131,9 +139,9 @@ async fn pipeline_upload_videos(
     // }
 
     let mut uploaded = UploadedVideos::default();
-
+    pin!(rx);
     // 流式处理后续事件
-    while let Ok(event) = rx.recv().await {
+    while let Some(event) = rx.next().await {
         let video = upload_single_file(&event.prev_file_path, context).await?;
         uploaded.videos.push(video);
         uploaded.paths.push(event.prev_file_path);
@@ -267,19 +275,7 @@ 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 {
-        let paths: Vec<&Path> = video_paths.iter().map(|p| {
-            let pool = ctx.pool.clone();
-            let file = p.display().to_string();
-            tokio::spawn(async move {
-                let result = InsertFileItem {
-                    file,
-                    streamer_info_id: 0,
-                }.insert(&pool).await;
-                info!(result=?result);
-            });
-
-            p.as_path()
-        }).collect();
+        let paths: Vec<&Path> = video_paths.iter().map(|p| p.as_path()).collect();
         process_video(&paths, &processor).await?;
     }
     Ok(())

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

@@ -5,7 +5,9 @@ use crate::server::core::downloader::{Downloader, SegmentEvent, SegmentInfo};
 use crate::server::core::monitor::{Monitor, RoomsHandle};
 use crate::server::core::plugin::{DownloadPlugin, StreamInfoExt};
 use crate::server::errors::{AppError, AppResult};
+use crate::server::infrastructure::connection_pool::ConnectionPool;
 use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus};
+use crate::server::infrastructure::models::InsertFileItem;
 use crate::server::infrastructure::models::hook_step::process_video;
 use async_channel::{Receiver, Sender, bounded};
 use biliup::bilibili::{BiliBili, Studio, Video};
@@ -17,13 +19,14 @@ use biliup::uploader::{VideoFile, line};
 use core::fmt;
 use error_stack::ResultExt;
 use futures::StreamExt;
+use ormlite::Insert;
 use std::path::Path;
 use std::str::FromStr;
 use std::sync::{Arc, Mutex};
 use std::time::Instant;
+use tokio::pin;
 use tokio::task::JoinHandle;
 use tracing::{error, info, warn};
-use crate::server::infrastructure::connection_pool::ConnectionPool;
 
 /// 下载管理器
 /// 负责管理特定平台的下载任务，包括监控器和插件
@@ -62,7 +65,11 @@ impl DownloadManager {
             .lock()
             .unwrap()
             .get_or_insert_with(|| {
-                Arc::new(Monitor::new(self.plugin.clone(), self.actor_handle.clone(), pool))
+                Arc::new(Monitor::new(
+                    self.plugin.clone(),
+                    self.actor_handle.clone(),
+                    pool,
+                ))
             })
             .clone()
     }
@@ -156,11 +163,26 @@ impl UActor {
             UploaderMessage::SegmentEvent(rx, ctx) => {
                 ctx.worker
                     .change_status(Stage::Upload, WorkerStatus::Pending);
+                let inspect = rx.inspect(|f| {
+                    let pool = ctx.pool.clone();
+                    let streamer_info_id = ctx.stream_info.streamer_info.id;
+                    let file = f.prev_file_path.display().to_string();
+                    tokio::spawn(async move {
+                        let result = InsertFileItem {
+                            file,
+                            streamer_info_id,
+                        }
+                        .insert(&pool)
+                        .await;
+                        info!(result=?result, "Insert file");
+                    });
+                });
                 let result = match ctx.worker.get_upload_config() {
-                    Some(config) => process_with_upload(rx, &ctx, config).await,
+                    Some(config) => process_with_upload(inspect, &ctx, config).await,
                     None => {
                         let mut paths = Vec::new();
-                        while let Ok(event) = rx.recv().await {
+                        pin!(inspect);
+                        while let Some(event) = inspect.next().await {
                             paths.push(event.prev_file_path);
                         }
                         // 无上传配置时，直接执行后处理

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

@@ -304,7 +304,7 @@ impl FfmpegDownloader {
         // 将 stderr 打印到当前进程的 stderr
         let stderr_task = tokio::spawn(async move {
             while let Ok(Some(line)) = stderr_lines.next_line().await {
-                eprintln!("[ffmpeg] {line}");
+                info!("[ffmpeg] {line}");
             }
         });
 
@@ -314,7 +314,6 @@ impl FfmpegDownloader {
         let mut prev_file_path: Option<PathBuf> = None;
 
         while let Some(line) = reader.next_line().await.change_context(AppError::Unknown)? {
-            info!("Received line: {line}");
             // 解析文件名
             let file_path = PathBuf::from(line.trim());
 

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

@@ -1,16 +1,16 @@
 use crate::server::core::download_manager::{ActorHandle, DownloaderMessage};
 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::models::StreamerInfo;
 use async_channel::{Receiver, Sender, bounded};
-use std::sync::Arc;
-use std::time::Duration;
 use ormlite::Model;
 use ormlite::model::ModelBuilder;
+use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::oneshot;
 use tokio::task::JoinHandle;
 use tracing::{error, info};
-use crate::server::infrastructure::connection_pool::ConnectionPool;
-use crate::server::infrastructure::models::StreamerInfo;
 
 /// 启动客户端监控循环
 ///
@@ -44,14 +44,16 @@ async fn start_client(
                         .title(sql_no_id.title.clone())
                         .date(sql_no_id.date.clone())
                         .live_cover_path(sql_no_id.live_cover_path.clone())
-                        .insert(&ctx.pool).await {
-                        Ok(insert) => {insert}
+                        .insert(&ctx.pool)
+                        .await
+                    {
+                        Ok(insert) => insert,
                         Err(e) => {
                             error!(e=?e, "插入数据库失败");
-                            continue
+                            continue;
                         }
                     };
-                    info!(url=url, "room: is live -> 开播了");
+                    info!(url = url, "room: is live -> 开播了");
                     // 更新状态为等待中
                     room.change_status(Stage::Download, WorkerStatus::Pending);
                     stream_info.streamer_info = insert;
@@ -98,7 +100,7 @@ impl Monitor {
     pub fn new(
         plugin: Arc<dyn DownloadPlugin + Send + Sync>,
         actor_handle: Arc<ActorHandle>,
-        pool: ConnectionPool
+        pool: ConnectionPool,
     ) -> Self {
         // 创建房间处理器
         let handle = Arc::new(RoomsHandle::new(plugin.name()));
@@ -106,7 +108,7 @@ impl Monitor {
         let join_handle = tokio::spawn({
             let handle = Arc::clone(&handle);
             async move {
-                start_client(handle, plugin, actor_handle, pool,10).await;
+                start_client(handle, plugin, actor_handle, pool, 10).await;
             }
         });
         Self {
@@ -305,13 +307,14 @@ impl RoomsActor {
     fn del(&mut self, id: i64) -> usize {
         // 从活跃房间列表中删除
         if let Some(i) = self.rooms.iter().position(|x| x.live_streamer.id == id) {
+            info!("Removed room [{:?}] {}", self.rooms.len(), i);
             self.rooms.remove(i); // 保序，但O(n)
         } else if let Some(i) = self.waiting.iter().position(|x| x.live_streamer.id == id) {
+            info!("Deleting room [{:?}] {}", self.waiting.len(), i);
             // 从等待房间列表中删除
             self.waiting.swap_remove(i);
         };
-        info!("Removed room [{:?}] {}", self.rooms, self.rooms.len());
-        info!("Deleting room [{:?}] {}", self.waiting, self.waiting.len());
+
         self.rooms.len() + self.waiting.len()
     }
 

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

@@ -3,6 +3,7 @@ use crate::server::config::Config;
 use crate::server::core::downloader::Downloader;
 use crate::server::errors::{AppError, AppResult};
 use crate::server::infrastructure::context::{Context, Worker};
+use crate::server::infrastructure::models::StreamerInfo;
 use async_trait::async_trait;
 use chrono::{DateTime, Local};
 use error_stack::Report;
@@ -11,7 +12,6 @@ use std::collections::HashMap;
 use std::path::Path;
 use std::sync::Arc;
 use time::OffsetDateTime;
-use crate::server::infrastructure::models::StreamerInfo;
 
 /// 流信息结构
 /// 包含直播流的详细信息

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

@@ -4,6 +4,7 @@ use crate::server::core::plugin::StreamInfoExt;
 use crate::server::errors::{AppError, AppResult};
 use crate::server::infrastructure::connection_pool::ConnectionPool;
 use crate::server::infrastructure::models::live_streamer::LiveStreamer;
+use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
 use crate::server::infrastructure::repositories::{get_config, get_streamer};
 use axum::http::Extensions;
 use biliup::client::StatelessClient;
@@ -13,7 +14,6 @@ use ormlite::Model;
 use serde::{Deserialize, Serialize, Serializer};
 use std::sync::{Arc, RwLock};
 use tracing::info;
-use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
 
 /// 应用程序上下文，包含工作器和扩展信息
 #[derive(Debug, Clone)]
@@ -157,7 +157,7 @@ pub enum WorkerStatus {
     #[default]
     Idle,
     /// 下载暂停中
-    Pause
+    Pause,
 }
 
 // 简单 Debug：打印状态名，忽略内部 downloader
@@ -167,7 +167,7 @@ impl fmt::Debug for WorkerStatus {
             WorkerStatus::Working(_) => "Working",
             WorkerStatus::Pending => "Pending",
             WorkerStatus::Idle => "Idle",
-            WorkerStatus::Pause => "Pause"
+            WorkerStatus::Pause => "Pause",
         };
         f.write_str(name)
     }

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

@@ -1,7 +1,7 @@
 /// 钩子步骤模块
 pub mod hook_step;
-pub mod upload_streamer;
 pub mod live_streamer;
+pub mod upload_streamer;
 
 use chrono::{DateTime, Utc};
 use ormlite::{Insert, Model};

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

@@ -1,7 +1,7 @@
+use crate::server::infrastructure::models::hook_step::HookStep;
 use ormlite::{Insert, Model};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
-use crate::server::infrastructure::models::hook_step::HookStep;
 
 /// 直播主播模型
 /// 存储直播主播的配置信息和录制参数
@@ -77,4 +77,4 @@ pub struct InsertLiveStreamer {
     pub postprocessor: Option<Vec<HookStep>>,
     pub opt_args: Option<Value>,
     pub excluded_keywords: Option<Value>,
-}
+}

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

@@ -85,4 +85,4 @@ pub struct InsertUploadStreamer {
     pub up_close_danmu: Option<u8>,
     pub extra_fields: Option<String>,
     pub is_only_self: Option<i64>,
-}
+}

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

@@ -1,13 +1,11 @@
 use crate::server::config::Config;
 use crate::server::errors::{AppError, AppResult};
 use crate::server::infrastructure::connection_pool::ConnectionPool;
-use crate::server::infrastructure::models::{
-    Configuration, InsertConfiguration,
-};
-use error_stack::ResultExt;
-use ormlite::{Insert, Model};
 use crate::server::infrastructure::models::live_streamer::LiveStreamer;
 use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
+use crate::server::infrastructure::models::{Configuration, InsertConfiguration};
+use error_stack::ResultExt;
+use ormlite::{Insert, Model};
 
 /// 根据ID获取直播主播信息
 ///

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

@@ -5,12 +5,12 @@ use crate::server::errors::{AppError, AppResult};
 use crate::server::infrastructure::connection_pool::ConnectionPool;
 use crate::server::infrastructure::context::{Context, Worker, WorkerStatus};
 use crate::server::infrastructure::models::live_streamer::LiveStreamer;
+use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
 use axum::extract::FromRef;
 use biliup::client::StatelessClient;
 use error_stack::bail;
 use std::sync::{Arc, RwLock};
 use tracing::info;
-use crate::server::infrastructure::models::upload_streamer::UploadStreamer;
 
 /// 服务注册器
 /// 负责管理应用程序中的各种服务实例，包括数据库连接池、工作器、下载管理器等

crates/biliup-cli/src/server/router.rs

@@ -1,7 +1,14 @@
 use crate::server::api::bilibili_endpoints::{
     archive_pre_endpoint, get_myinfo_endpoint, get_proxy_endpoint,
 };
-use crate::server::api::endpoints::{add_upload_streamer_endpoint, add_user_endpoint, delete_streamers_endpoint, delete_template_endpoint, delete_user_endpoint, get_configuration, get_qrcode, get_status, get_streamer_info, get_streamers_endpoint, get_upload_streamer_endpoint, get_upload_streamers_endpoint, get_users_endpoint, get_videos, login_by_qrcode, pause_streamers_endpoint, post_streamers_endpoint, post_uploads, put_configuration, put_streamers_endpoint};
+use crate::server::api::endpoints::{
+    add_upload_streamer_endpoint, add_user_endpoint, delete_streamers_endpoint,
+    delete_template_endpoint, delete_user_endpoint, get_configuration, get_qrcode, get_status,
+    get_streamer_info, get_streamers_endpoint, get_upload_streamer_endpoint,
+    get_upload_streamers_endpoint, get_users_endpoint, get_videos, login_by_qrcode,
+    pause_streamers_endpoint, post_streamers_endpoint, post_uploads, put_configuration,
+    put_streamers_endpoint,
+};
 use crate::server::infrastructure::service_register::ServiceRegister;
 use axum::Router;
 use axum::body::Body;

crates/stream-gears/src/danmaku.rs

@@ -59,7 +59,9 @@ impl Downloader for DanmakuClient {
                 Ok::<_, PyErr>(())
             })
         })
-        .await.change_context(AppError::Unknown)?.change_context(AppError::Unknown)?;
+        .await
+        .change_context(AppError::Unknown)?
+        .change_context(AppError::Unknown)?;
         Ok(())
     }
 

crates/stream-gears/src/server.rs

@@ -12,6 +12,7 @@ use biliup_cli::server::core::plugin::{DownloadPlugin, StreamInfoExt, StreamStat
 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::models::StreamerInfo;
 use biliup_cli::server::infrastructure::repositories;
 use biliup_cli::server::infrastructure::repositories::get_upload_config;
 use biliup_cli::server::infrastructure::service_register::ServiceRegister;
@@ -33,7 +34,6 @@ use std::path::PathBuf;
 use std::sync::{Arc, LazyLock, RwLock};
 use time::OffsetDateTime;
 use tracing::{debug, info};
-use biliup_cli::server::infrastructure::models::StreamerInfo;
 
 #[derive(Debug)]
 pub struct PyPlugin {
@@ -145,37 +145,39 @@ async fn call_via_threads(
 ) -> PyResult<Option<(StreamInfoExt, Option<Py<PyAny>>)>> {
     let url = url.to_string();
     tokio::task::spawn_blocking(move || {
-        Python::attach(|py| -> PyResult<Option<(StreamInfoExt, Option<Py<PyAny>>)>> {
-            // 从 biliup.util 获取 loop（按你项目里真实的名字来取）
-            let util = PyModule::import(py, "biliup.common.util")?;
-            // 下面两行二选一（取决于 biliup.util 的 API）：
-            // let loop_obj: Py<PyAny> = util.getattr("loop")?.into_py(py);
-            // 或：
-            // let loop_obj: Py<PyAny> = util.call_method0("get_loop")?.into_py(py);
+        Python::attach(
+            |py| -> PyResult<Option<(StreamInfoExt, Option<Py<PyAny>>)>> {
+                // 从 biliup.util 获取 loop（按你项目里真实的名字来取）
+                let util = PyModule::import(py, "biliup.common.util")?;
+                // 下面两行二选一（取决于 biliup.util 的 API）：
+                // let loop_obj: Py<PyAny> = util.getattr("loop")?.into_py(py);
+                // 或：
+                // let loop_obj: Py<PyAny> = util.call_method0("get_loop")?.into_py(py);
 
-            // 这里假设是直接暴露了 util.loop
-            let loop_obj = util.getattr("loop")?;
+                // 这里假设是直接暴露了 util.loop
+                let loop_obj = util.getattr("loop")?;
 
-            let asyncio = PyModule::import(py, "asyncio")?;
+                let asyncio = PyModule::import(py, "asyncio")?;
 
-            // 生成协程 self.acheck_stream()
-            let instance = obj.bind(py).call1(("fname", url))?;
-            let coro = instance.call_method0("acheck_stream")?;
+                // 生成协程 self.acheck_stream()
+                let instance = obj.bind(py).call1(("fname", url))?;
+                let coro = instance.call_method0("acheck_stream")?;
 
-            // 调度到指定 loop
-            let fut = asyncio
-                .getattr("run_coroutine_threadsafe")?
-                .call1((coro, loop_obj))?;
+                // 调度到指定 loop
+                let fut = asyncio
+                    .getattr("run_coroutine_threadsafe")?
+                    .call1((coro, loop_obj))?;
 
-            let res = fut.call_method0("result")?;
-            let is_live = res.unbind().extract(py)?;
-            let info = if is_live {
-                Some(stream_info_from_py(&instance)?)
-            } else {
-                None
-            };
-            Ok(info)
-        })
+                let res = fut.call_method0("result")?;
+                let is_live = res.unbind().extract(py)?;
+                let info = if is_live {
+                    Some(stream_info_from_py(&instance)?)
+                } else {
+                    None
+                };
+                Ok(info)
+            },
+        )
     })
     .await
     .expect("spawn_blocking panicked")