From bd1ccdd0ff99b6801583b5cf23353b0c7108e423 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Mon, 29 Sep 2025 04:19:22 +0800 Subject: [PATCH] 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 | 4 +- Cargo.toml | 2 +- .../src/server/api/bilibili_endpoints.rs | 2 - crates/biliup-cli/src/server/api/endpoints.rs | 46 ++++++++------- .../biliup-cli/src/server/common/download.rs | 3 +- crates/biliup-cli/src/server/common/upload.rs | 44 +++++++-------- .../src/server/core/download_manager.rs | 30 ++++++++-- .../core/downloader/ffmpeg_downloader.rs | 3 +- crates/biliup-cli/src/server/core/monitor.rs | 27 +++++---- crates/biliup-cli/src/server/core/plugin.rs | 2 +- .../src/server/infrastructure/context.rs | 6 +- .../src/server/infrastructure/models.rs | 2 +- .../infrastructure/models/live_streamer.rs | 4 +- .../infrastructure/models/upload_streamer.rs | 2 +- .../src/server/infrastructure/repositories.rs | 8 +-- .../server/infrastructure/service_register.rs | 2 +- crates/biliup-cli/src/server/router.rs | 9 ++- crates/stream-gears/src/danmaku.rs | 4 +- crates/stream-gears/src/server.rs | 56 ++++++++++--------- 19 files changed, 140 insertions(+), 116 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index a44839e..d17ac20 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index 77e7727..2aaaf7c 100644 --- a/Cargo.toml +++ b/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"] } diff --git a/crates/biliup-cli/src/server/api/bilibili_endpoints.rs b/crates/biliup-cli/src/server/api/bilibili_endpoints.rs index e3dbcaf..954cd5d 100644 --- a/crates/biliup-cli/src/server/api/bilibili_endpoints.rs +++ b/crates/biliup-cli/src/server/api/bilibili_endpoints.rs @@ -67,8 +67,6 @@ pub async fn get_proxy_endpoint( State(client): State, Query(params): Query>, ) -> Result { - info!(params = ¶ms["url"]); - // 代理HTTP请求 client .client diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index ce883ea..c0e49eb 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/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, State(workers): State>>>>, ) -> Result>, 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(())) diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index 5341579..a607696 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/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; }; }); diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs index e2f6520..13a0241 100644 --- a/crates/biliup-cli/src/server/common/upload.rs +++ b/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, } -pub async fn process_with_upload( - rx: Receiver, +pub async fn process_with_upload( + rx: Inspect, 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, +async fn pipeline_upload_videos( + mut rx: Inspect, F>, context: &UploadContext, -) -> AppResult { +) -> AppResult +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, 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(()) diff --git a/crates/biliup-cli/src/server/core/download_manager.rs b/crates/biliup-cli/src/server/core/download_manager.rs index 2815df2..687e260 100644 --- a/crates/biliup-cli/src/server/core/download_manager.rs +++ b/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); } // 无上传配置时，直接执行后处理 diff --git a/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs b/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs index 8c7f2b4..c10f008 100644 --- a/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs +++ b/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 = None; while let Some(line) = reader.next_line().await.change_context(AppError::Unknown)? { - info!("Received line: {line}"); // 解析文件名 let file_path = PathBuf::from(line.trim()); diff --git a/crates/biliup-cli/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs index 171c197..623ad96 100644 --- a/crates/biliup-cli/src/server/core/monitor.rs +++ b/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, actor_handle: Arc, - 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() } diff --git a/crates/biliup-cli/src/server/core/plugin.rs b/crates/biliup-cli/src/server/core/plugin.rs index 098e762..a926731 100644 --- a/crates/biliup-cli/src/server/core/plugin.rs +++ b/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; /// 流信息结构 /// 包含直播流的详细信息 diff --git a/crates/biliup-cli/src/server/infrastructure/context.rs b/crates/biliup-cli/src/server/infrastructure/context.rs index 6ae1d0a..ca98548 100644 --- a/crates/biliup-cli/src/server/infrastructure/context.rs +++ b/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) } diff --git a/crates/biliup-cli/src/server/infrastructure/models.rs b/crates/biliup-cli/src/server/infrastructure/models.rs index a0cc628..3ba438d 100644 --- a/crates/biliup-cli/src/server/infrastructure/models.rs +++ b/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}; diff --git a/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs b/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs index 81ab8c5..cb43be9 100644 --- a/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs +++ b/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>, pub opt_args: Option, pub excluded_keywords: Option, -} \ No newline at end of file +} diff --git a/crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs b/crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs index 0b8d1df..32f5024 100644 --- a/crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs +++ b/crates/biliup-cli/src/server/infrastructure/models/upload_streamer.rs @@ -85,4 +85,4 @@ pub struct InsertUploadStreamer { pub up_close_danmu: Option, pub extra_fields: Option, pub is_only_self: Option, -} \ No newline at end of file +} diff --git a/crates/biliup-cli/src/server/infrastructure/repositories.rs b/crates/biliup-cli/src/server/infrastructure/repositories.rs index 92b6d8a..ac77fe1 100644 --- a/crates/biliup-cli/src/server/infrastructure/repositories.rs +++ b/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获取直播主播信息 /// diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs index 745c6f7..ff1c7b3 100644 --- a/crates/biliup-cli/src/server/infrastructure/service_register.rs +++ b/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; /// 服务注册器 /// 负责管理应用程序中的各种服务实例，包括数据库连接池、工作器、下载管理器等 diff --git a/crates/biliup-cli/src/server/router.rs b/crates/biliup-cli/src/server/router.rs index 37956d2..d37bc39 100644 --- a/crates/biliup-cli/src/server/router.rs +++ b/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; diff --git a/crates/stream-gears/src/danmaku.rs b/crates/stream-gears/src/danmaku.rs index f1f38d9..9280b92 100644 --- a/crates/stream-gears/src/danmaku.rs +++ b/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(()) } diff --git a/crates/stream-gears/src/server.rs b/crates/stream-gears/src/server.rs index 3d938b0..5605c4c 100644 --- a/crates/stream-gears/src/server.rs +++ b/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>)>> { let url = url.to_string(); tokio::task::spawn_blocking(move || { - Python::attach(|py| -> PyResult>)>> { - // 从 biliup.util 获取 loop（按你项目里真实的名字来取） - let util = PyModule::import(py, "biliup.common.util")?; - // 下面两行二选一（取决于 biliup.util 的 API）： - // let loop_obj: Py = util.getattr("loop")?.into_py(py); - // 或： - // let loop_obj: Py = util.call_method0("get_loop")?.into_py(py); + Python::attach( + |py| -> PyResult>)>> { + // 从 biliup.util 获取 loop（按你项目里真实的名字来取） + let util = PyModule::import(py, "biliup.common.util")?; + // 下面两行二选一（取决于 biliup.util 的 API）： + // let loop_obj: Py = util.getattr("loop")?.into_py(py); + // 或： + // let loop_obj: Py = 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")