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