From ea17305b0a0e184531b29cee80ced70b771effec Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Wed, 1 Oct 2025 00:26:07 +0800 Subject: [PATCH] Refactor actor system and add upload status tracking Refactored download and upload actor implementations into separate modules for better maintainability. Added upload_status field to LiveStreamerEntity and API responses, and updated the frontend to display upload status. Also added Huya protocol config option and improved Huya stream header handling in stream-gears. Minor bug fixes and code cleanups included. --- Cargo.lock | 4 +- Cargo.toml | 2 +- app/(app)/streamers/page.tsx | 14 +- app/lib/api-streamer.ts | 1 + crates/biliup-cli/src/server/api/endpoints.rs | 27 +-- .../biliup-cli/src/server/common/download.rs | 59 ++++++- crates/biliup-cli/src/server/common/upload.rs | 78 ++++++++- crates/biliup-cli/src/server/config.rs | 3 + .../src/server/core/download_manager.rs | 159 +----------------- crates/biliup-cli/src/server/core/monitor.rs | 7 +- .../src/server/infrastructure/dto.rs | 2 + .../infrastructure/models/live_streamer.rs | 2 +- .../server/infrastructure/service_register.rs | 1 - crates/stream-gears/src/server.rs | 11 +- 14 files changed, 188 insertions(+), 182 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index acc4c83..c7e4895 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -395,7 +395,7 @@ dependencies = [ [[package]] name = "biliup-cli" -version = "1.1.13" +version = "1.1.14" dependencies = [ "async-channel", "async-trait", @@ -3641,7 +3641,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" [[package]] name = "stream-gears" -version = "1.1.13" +version = "1.1.14" dependencies = [ "async-channel", "async-trait", diff --git a/Cargo.toml b/Cargo.toml index 0fc36b9..0d6f288 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ members = [ resolver = "3" [workspace.package] -version = "1.1.13" +version = "1.1.14" [workspace.dependencies] serde = { version = "1.0", features = ["derive"] } diff --git a/app/(app)/streamers/page.tsx b/app/(app)/streamers/page.tsx index 54bda31..eacca58 100644 --- a/app/(app)/streamers/page.tsx +++ b/app/(app)/streamers/page.tsx @@ -7,7 +7,7 @@ import { Typography, Popconfirm, Notification, - Card, Dropdown, + Card, Dropdown, Badge, } from '@douyinfe/semi-ui' import { IconHelpCircle, @@ -15,7 +15,7 @@ import { IconVideoListStroked, IconEdit2Stroked, IconDeleteStroked, - IconWrench, IconTreeTriangleDown, IconPause, IconPlay, + IconWrench, IconTreeTriangleDown, IconPause, IconPlay, IconLock, IconUpload, } from '@douyinfe/semi-icons' import { List, ButtonGroup } from '@douyinfe/semi-ui' import React, { useState } from 'react' @@ -101,6 +101,7 @@ export default function Home() { console.log(values); delete values.status delete values.statusTag + delete values.upload_status if (values?.postprocessor) { values.postprocessor = values.postprocessor.map( ({ cmd, value }: { cmd: string; value: string }) => (cmd === 'rm' ? 'rm' : { [cmd]: value }) @@ -221,12 +222,15 @@ export default function Home() { } >
{item.statusTag}
+
+ {item.upload_status === "Pending" ? }> : null} +

{item.remark}

+
+ {item.url} +
+ )} /> diff --git a/app/lib/api-streamer.ts b/app/lib/api-streamer.ts index 7961926..b3880f8 100644 --- a/app/lib/api-streamer.ts +++ b/app/lib/api-streamer.ts @@ -101,6 +101,7 @@ export interface LiveStreamerEntity { split_size?: number; upload_id?: number; status?: string; + upload_status?: string; statusTag?: React.ReactNode; format?: string; time_range?: string | Date[]; diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index c0e49eb..426448d 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -49,20 +49,23 @@ pub async fn get_streamers_endpoint( Ok(Json( live_streamers .into_iter() - .map(|x| LiveStreamerResponse { - status: workers + .map(|x| { + let option = workers .read() .unwrap() - .iter() - .find_map(|worker| { - if worker.live_streamer.id == x.id { - Some(format!("{:?}", *worker.downloader_status.read().unwrap())) - } else { - None - } - }) - .unwrap_or_default(), - inner: x, + .clone() + .into_iter() + .find(|worker| worker.live_streamer.id == x.id); + LiveStreamerResponse { + status: option + .as_ref() + .map(|t| format!("{:?}", *t.downloader_status.read().unwrap())) + .unwrap_or_default(), + inner: x, + upload_status: option + .map(|t| format!("{:?}", *t.uploader_status.read().unwrap())) + .unwrap_or_default(), + } }) .collect(), )) diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index 15cb25c..0c49072 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -1,5 +1,5 @@ +use crate::server::common::upload::UploaderMessage; use crate::server::common::util::Recorder; -use crate::server::core::download_manager::{DownloaderMessage, UploaderMessage}; use crate::server::core::downloader::{DownloadStatus, Downloader, SegmentEvent, SegmentInfo}; use crate::server::core::monitor::RoomsHandle; use crate::server::core::plugin::{DownloadPlugin, StreamInfoExt}; @@ -112,7 +112,8 @@ impl FileValidator { if size < self.min_size { bail!(AppError::Custom(format!( - "File too small: {size} bytes, minimum: {} bytes", + "File {} too small: {size} bytes, minimum: {} bytes", + path.display(), self.min_size ))); } @@ -354,3 +355,57 @@ impl DownloadTask { Ok(()) } } + +/// 下载Actor +/// 负责处理下载相关的消息和任务 +pub struct DActor { + /// 下载消息接收器 + receiver: Receiver, + /// 上传消息发送器 + sender: Sender, +} + +impl DActor { + /// 创建新的下载Actor实例 + pub fn new(receiver: Receiver, sender: Sender) -> Self { + Self { receiver, sender } + } + + /// 运行Actor主循环，处理接收到的消息 + pub(crate) async fn run(&mut self) { + while let Ok(msg) = self.receiver.recv().await { + if let Err(e) = self.handle_message(msg).await { + error!("Error handling message: {}", e); + } + } + } + + /// 处理下载消息 + /// + /// # 参数 + /// * `msg` - 要处理的下载消息 + async fn handle_message(&mut self, msg: DownloaderMessage) -> AppResult<()> { + match msg { + DownloaderMessage::Start(plugin, ctx, rooms_handle) => { + // 创建下载任务 + let task = DownloadTask::new(plugin, ctx, rooms_handle); + + // 执行下载（使用 Result 链式处理） + task.execute(&self.sender).await?; + + Ok(()) + } + } + } +} + +/// 下载消息枚举 +/// 定义下载Actor可以处理的消息类型 +pub enum DownloaderMessage { + /// 开始下载消息，包含插件、流信息、上下文和房间句柄 + Start( + Arc, + Context, + Arc, + ), +} diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs index 13a0241..c81b856 100644 --- a/crates/biliup-cli/src/server/common/upload.rs +++ b/crates/biliup-cli/src/server/common/upload.rs @@ -1,9 +1,8 @@ use crate::UploadLine; use crate::server::common::util::Recorder; -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::context::{Context, Stage, Worker, WorkerStatus}; use crate::server::infrastructure::models::InsertFileItem; use crate::server::infrastructure::models::hook_step::process_video; use crate::server::infrastructure::models::upload_streamer::UploadStreamer; @@ -359,3 +358,78 @@ pub async fn upload( Ok((bilibili, videos)) } + +/// 上传Actor +/// 负责处理上传相关的消息和任务 +pub struct UActor { + /// 上传消息接收器 + receiver: Receiver, +} + +impl UActor { + /// 创建新的上传Actor实例 + pub fn new(receiver: Receiver) -> Self { + Self { receiver } + } + + /// 运行Actor主循环，处理接收到的消息 + pub(crate) async fn run(&mut self) { + while let Ok(msg) = self.receiver.recv().await { + self.handle_message(msg).await; + } + } + + /// 处理上传消息 + /// + /// # 参数 + /// * `msg` - 要处理的上传消息 + async fn handle_message(&mut self, msg: UploaderMessage) { + match msg { + 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(inspect, &ctx, config).await, + None => { + let mut paths = Vec::new(); + pin!(inspect); + while let Some(event) = inspect.next().await { + paths.push(event.prev_file_path); + } + // 无上传配置时，直接执行后处理 + execute_postprocessor(paths, &ctx).await + } + }; + + if let Err(e) = &result { + 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); + } + } + } +} + +/// 上传消息枚举 +/// 定义上传Actor可以处理的消息类型 +#[derive(Debug)] +pub enum UploaderMessage { + /// 分段事件消息，包含事件、接收器和工作器 + SegmentEvent(Receiver, Context), +} diff --git a/crates/biliup-cli/src/server/config.rs b/crates/biliup-cli/src/server/config.rs index 760af3d..0d8e40c 100644 --- a/crates/biliup-cli/src/server/config.rs +++ b/crates/biliup-cli/src/server/config.rs @@ -108,6 +108,9 @@ pub struct Config { /// 虎牙最大比率 #[serde(default)] pub huya_max_ratio: Option, + /// 虎牙 Flv or Hls + #[serde(default)] + pub huya_protocol: Option, // 抖音平台设置 /// 抖音弹幕录制 diff --git a/crates/biliup-cli/src/server/core/download_manager.rs b/crates/biliup-cli/src/server/core/download_manager.rs index 6bc969f..f6780f9 100644 --- a/crates/biliup-cli/src/server/core/download_manager.rs +++ b/crates/biliup-cli/src/server/core/download_manager.rs @@ -1,33 +1,14 @@ -use crate::server::common::download::DownloadTask; -use crate::server::common::upload::{execute_postprocessor, process_with_upload}; -use crate::server::common::util::Recorder; -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::common::download::{DActor, DownloaderMessage}; +use crate::server::common::upload::{UActor, UploaderMessage}; +use crate::server::core::downloader::Downloader; +use crate::server::core::monitor::Monitor; +use crate::server::core::plugin::DownloadPlugin; 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}; -use biliup::client::StatelessClient; -use biliup::credential::login_by_cookies; -use biliup::uploader::line::{Line, Probe}; -use biliup::uploader::util::SubmitOption; -use biliup::uploader::{VideoFile, line}; +use async_channel::{Sender, bounded}; 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}; - /// 下载管理器 /// 负责管理特定平台的下载任务，包括监控器和插件 pub struct DownloadManager { @@ -92,115 +73,6 @@ impl fmt::Debug for DownloadManager { } } -/// 下载Actor -/// 负责处理下载相关的消息和任务 -pub struct DActor { - /// 下载消息接收器 - receiver: Receiver, - /// 上传消息发送器 - sender: Sender, -} - -impl DActor { - /// 创建新的下载Actor实例 - pub fn new(receiver: Receiver, sender: Sender) -> Self { - Self { receiver, sender } - } - - /// 运行Actor主循环，处理接收到的消息 - async fn run(&mut self) { - while let Ok(msg) = self.receiver.recv().await { - if let Err(e) = self.handle_message(msg).await { - error!("Error handling message: {}", e); - } - } - } - - /// 处理下载消息 - /// - /// # 参数 - /// * `msg` - 要处理的下载消息 - async fn handle_message(&mut self, msg: DownloaderMessage) -> AppResult<()> { - match msg { - DownloaderMessage::Start(plugin, ctx, rooms_handle) => { - // 创建下载任务 - let task = DownloadTask::new(plugin, ctx, rooms_handle); - - // 执行下载（使用 Result 链式处理） - task.execute(&self.sender).await?; - - Ok(()) - } - } - } -} -/// 上传Actor -/// 负责处理上传相关的消息和任务 -pub struct UActor { - /// 上传消息接收器 - receiver: Receiver, -} - -impl UActor { - /// 创建新的上传Actor实例 - pub fn new(receiver: Receiver) -> Self { - Self { receiver } - } - - /// 运行Actor主循环，处理接收到的消息 - async fn run(&mut self) { - while let Ok(msg) = self.receiver.recv().await { - self.handle_message(msg).await; - } - } - - /// 处理上传消息 - /// - /// # 参数 - /// * `msg` - 要处理的上传消息 - async fn handle_message(&mut self, msg: UploaderMessage) { - match msg { - 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(inspect, &ctx, config).await, - None => { - let mut paths = Vec::new(); - pin!(inspect); - while let Some(event) = inspect.next().await { - paths.push(event.prev_file_path); - } - // 无上传配置时，直接执行后处理 - execute_postprocessor(paths, &ctx).await - } - }; - - if let Err(e) = &result { - 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); - } - } - } -} - /// Actor处理器 /// 管理下载和上传Actor的生命周期 pub struct ActorHandle { @@ -254,25 +126,6 @@ impl ActorHandle { } } -/// 上传消息枚举 -/// 定义上传Actor可以处理的消息类型 -#[derive(Debug)] -pub enum UploaderMessage { - /// 分段事件消息，包含事件、接收器和工作器 - SegmentEvent(Receiver, Context), -} - -/// 下载消息枚举 -/// 定义下载Actor可以处理的消息类型 -pub enum DownloaderMessage { - /// 开始下载消息，包含插件、流信息、上下文和房间句柄 - Start( - Arc, - Context, - Arc, - ), -} - impl Drop for ActorHandle { fn drop(&mut self) { // 发送端随 ActorHandle 一起被 drop，会关闭通道（如果没有其他 sender 克隆）。 diff --git a/crates/biliup-cli/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs index cbf0465..91361f1 100644 --- a/crates/biliup-cli/src/server/core/monitor.rs +++ b/crates/biliup-cli/src/server/core/monitor.rs @@ -1,4 +1,5 @@ -use crate::server::core::download_manager::{ActorHandle, DownloaderMessage}; +use crate::server::common::download::DownloaderMessage; +use crate::server::core::download_manager::ActorHandle; use crate::server::core::plugin::{DownloadPlugin, StreamStatus}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus}; @@ -74,8 +75,8 @@ async fn start_client( } } Ok(StreamStatus::Offline) => {} - Ok(StreamStatus::Unknown) => {}, - Err(e) => error!(e=?e, ctx=ctx.worker.live_streamer.url,"检查直播间出错") + Ok(StreamStatus::Unknown) => {} + Err(e) => error!(e=?e, ctx=ctx.worker.live_streamer.url,"检查直播间出错"), }; } // 等待下一次检查 diff --git a/crates/biliup-cli/src/server/infrastructure/dto.rs b/crates/biliup-cli/src/server/infrastructure/dto.rs index 79aae6e..d88b1fc 100644 --- a/crates/biliup-cli/src/server/infrastructure/dto.rs +++ b/crates/biliup-cli/src/server/infrastructure/dto.rs @@ -12,4 +12,6 @@ pub struct LiveStreamerResponse { /// 当前工作状态 pub status: String, + /// 上传状态 + pub upload_status: String, } 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 cb43be9..824d712 100644 --- a/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs +++ b/crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs @@ -32,7 +32,7 @@ pub struct LiveStreamer { /// 注意：数据库空与json空有区别，所以这里不能用#[ormlite(json)] /// 只能使用 sqlx::types::Json #[ormlite(json)] - pub preprocessor: Option>, + pub preprocessor: Option>, /// 分段处理器列表（JSON格式） #[ormlite(json)] pub segment_processor: Option>, diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs index bce7e4a..150c185 100644 --- a/crates/biliup-cli/src/server/infrastructure/service_register.rs +++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs @@ -49,7 +49,6 @@ impl ServiceRegister { // 创建默认的HTTP客户端 let client = StatelessClient::default(); - info!("utility services initialized, building feature services..."); info!("feature services successfully initialized!"); diff --git a/crates/stream-gears/src/server.rs b/crates/stream-gears/src/server.rs index c686dd2..35c3290 100644 --- a/crates/stream-gears/src/server.rs +++ b/crates/stream-gears/src/server.rs @@ -228,8 +228,15 @@ pub fn stream_info_from_py( let live_cover_path: Option = self_obj.getattr("live_cover_path")?.extract()?; let _is_download: bool = self_obj.getattr("is_download")?.extract()?; let platform: String = self_obj.getattr("platform")?.extract()?; - let stream_headers: HashMap = self_obj.getattr("stream_headers")?.extract()?; - self_obj.call_method1("update_headers", (&stream_headers,))?; + + let stream_headers: HashMap = if platform == "Huya" { + let stream_headers = self_obj.getattr("stream_headers")?; + self_obj.call_method1("update_headers", (&stream_headers,))?; + stream_headers.extract()? + } else { + self_obj.getattr("stream_headers")?.extract()? + }; + let danmaku_init = self_obj.call_method0("danmaku_init")?; // let platform: Option = self_obj.getattr("danmaku")?.extract()?; // danmaku 可能在条件下没有设置（比如 bilibili_danmaku 为 False）