okooooooo/biliup
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor actor system and add upload status tracking

Browse Source 
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.
This commit is contained in:
ForgQi committed 2025-10-01 00:26:07 +08:00
1 parent e2da70b9ea
commit ea17305b0a
14 files changed
+188 -182
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+2 -2
View File
@@ -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",
Cargo.toml
+1 -1
View File
@@ -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"] }
app/(app)/streamers/page.tsx
+11 -3
View File
@@ -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() {
}
>
<div style={{ position: 'absolute', right: 20, top: 9 }}>{item.statusTag}</div>
<div style={{ display: 'flex', justifyContent: 'space-between' }}>
{item.upload_status === "Pending" ? <Badge count={<IconUpload />}> </Badge> : null}
<h3
style={{
color: 'var(--semi-color-text-0)',
fontWeight: 500,
maxWidth: '60%',
maxWidth: '80%',
overflow: 'hidden',
textOverflow: 'ellipsis',
whiteSpace: 'nowrap',
@@ -234,10 +238,13 @@ export default function Home() {
>
{item.remark}
</h3>
</div>
<Text style={{ width: '101%' }} ellipsis={{ showTooltip: true }} type="tertiary">
{item.url}
</Text>
<div
style={{
margin: '0',
@@ -279,6 +286,7 @@ export default function Home() {
</ButtonGroup>
</div>
</Card>
</List.Item>
)}
/>
app/lib/api-streamer.ts
+1
View File
@@ -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[];
crates/biliup-cli/src/server/api/endpoints.rs
+15 -12
View File
@@ -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(),
))
crates/biliup-cli/src/server/common/download.rs
+57 -2
View File
@@ -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<DownloaderMessage>,
/// 上传消息发送器
sender: Sender<UploaderMessage>,
}
impl DActor {
/// 创建新的下载Actor实例
pub fn new(receiver: Receiver<DownloaderMessage>, sender: Sender<UploaderMessage>) -> 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<dyn DownloadPlugin + Send + Sync>,
Context,
Arc<RoomsHandle>,
),
}
crates/biliup-cli/src/server/common/upload.rs
+76 -2
View File
@@ -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<UploaderMessage>,
}
impl UActor {
/// 创建新的上传Actor实例
pub fn new(receiver: Receiver<UploaderMessage>) -> 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<SegmentInfo>, Context),
}
crates/biliup-cli/src/server/config.rs
+3
View File
@@ -108,6 +108,9 @@ pub struct Config {
/// 虎牙最大比率
#[serde(default)]
pub huya_max_ratio: Option<u32>,
/// 虎牙 Flv or Hls
#[serde(default)]
pub huya_protocol: Option<String>,
// 抖音平台设置
/// 抖音弹幕录制
crates/biliup-cli/src/server/core/download_manager.rs
+6 -153
View File
@@ -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<DownloaderMessage>,
/// 上传消息发送器
sender: Sender<UploaderMessage>,
}
impl DActor {
/// 创建新的下载Actor实例
pub fn new(receiver: Receiver<DownloaderMessage>, sender: Sender<UploaderMessage>) -> 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<UploaderMessage>,
}
impl UActor {
/// 创建新的上传Actor实例
pub fn new(receiver: Receiver<UploaderMessage>) -> 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<SegmentInfo>, Context),
}
/// 下载消息枚举
/// 定义下载Actor可以处理的消息类型
pub enum DownloaderMessage {
/// 开始下载消息，包含插件、流信息、上下文和房间句柄
Start(
Arc<dyn DownloadPlugin + Send + Sync>,
Context,
Arc<RoomsHandle>,
),
}
impl Drop for ActorHandle {
fn drop(&mut self) {
// 发送端随 ActorHandle 一起被 drop，会关闭通道（如果没有其他 sender 克隆）。
crates/biliup-cli/src/server/core/monitor.rs
+4 -3
View File
@@ -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,"检查直播间出错"),
};
}
// 等待下一次检查
crates/biliup-cli/src/server/infrastructure/dto.rs
+2
View File
@@ -12,4 +12,6 @@ pub struct LiveStreamerResponse {
/// 当前工作状态
pub status: String,
/// 上传状态
pub upload_status: String,
}
crates/biliup-cli/src/server/infrastructure/models/live_streamer.rs
+1 -1
View File
@@ -32,7 +32,7 @@ pub struct LiveStreamer {
/// 注意：数据库空与json空有区别，所以这里不能用#[ormlite(json)]
/// 只能使用 sqlx::types::Json
#[ormlite(json)]
pub preprocessor: Option<Vec<String>>,
pub preprocessor: Option<Vec<HookStep>>,
/// 分段处理器列表（JSON格式）
#[ormlite(json)]
pub segment_processor: Option<Vec<HookStep>>,
crates/biliup-cli/src/server/infrastructure/service_register.rs
-1
View File
@@ -49,7 +49,6 @@ impl ServiceRegister {
// 创建默认的HTTP客户端
let client = StatelessClient::default();
info!("utility services initialized, building feature services...");
info!("feature services successfully initialized!");
crates/stream-gears/src/server.rs
+9 -2
View File
@@ -228,8 +228,15 @@ pub fn stream_info_from_py(
let live_cover_path: Option<String> = 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<String, String> = self_obj.getattr("stream_headers")?.extract()?;
self_obj.call_method1("update_headers", (&stream_headers,))?;
let stream_headers: HashMap<String, String> = 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<PyAny> = self_obj.getattr("danmaku")?.extract()?;
// danmaku 可能在条件下没有设置（比如 bilibili_danmaku 为 False）