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

Refactor async worker status and improve file move logic

Browse Source 
Refactored worker status management to use async RwLock and updated all related methods to be async. Improved file move logic in hook steps to support cross-filesystem moves by falling back to copy+delete when necessary. Updated API endpoints, download/upload logic, and monitor state transitions to be fully async and more robust. Also added a Debug implementation for Credentials to avoid logging sensitive data, and made minor UI tweaks to disable certain developer options.
This commit is contained in:
ForgQi committed 2025-10-22 02:13:37 +08:00
1 parent 2fcdb0fe7b
commit a0b6800c0a
17 files changed
+347 -190
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

Cargo.lock
Generated
+3 -2
View File
@@ -395,7 +395,7 @@ dependencies = [
[[package]]
name = "biliup-cli"
version = "1.1.16"
version = "1.1.17"
dependencies = [
"async-channel",
"async-trait",
@@ -414,6 +414,7 @@ dependencies = [
"image",
"indexmap",
"indicatif",
"libc",
"mime_guess",
"nom 8.0.0",
"ormlite",
@@ -3641,7 +3642,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3"
[[package]]
name = "stream-gears"
version = "1.1.16"
version = "1.1.17"
dependencies = [
"async-channel",
"async-trait",
Cargo.toml
+1 -1
View File
@@ -7,7 +7,7 @@ members = [
resolver = "3"
[workspace.package]
version = "1.1.16"
version = "1.1.17"
[workspace.dependencies]
serde = { version = "1.0", features = ["derive"] }
app/ui/StreamerActions/PauseButton.tsx
+1 -1
View File
@@ -45,7 +45,7 @@ export const PauseButton: React.FC<PauseButtonProps> = ({
onError?.(error as Error);
}
};
console.log("????????????", streamer.status)
return (
<Button onClick={handlePause} icon={streamer.status === 'Pause'? <IconPlay />: <IconPause />} theme="borderless" aria-label="暂停" />
);
app/ui/plugins/developer.tsx
+2
View File
@@ -52,6 +52,7 @@ const Developer: React.FC = () => {
formApi.setValue('LOGGING.root.handlers', ['console'])
}}
showClear={true}
disabled
>
<Select.Option value="DEBUG">DEBUG</Select.Option>
<Select.Option value="INFO">INFO</Select.Option>
@@ -72,6 +73,7 @@ const Developer: React.FC = () => {
formApi.setValue('LOGGING.loggers.biliup.handlers', ['file'])
}}
showClear={true}
disabled
>
<Select.Option value="DEBUG">DEBUG</Select.Option>
<Select.Option value="INFO">INFO</Select.Option>
crates/biliup-cli/Cargo.toml
+1
View File
@@ -57,6 +57,7 @@ tower-sessions-sqlx-store = { version = "0.15.0", features = ["sqlite"] }
async-channel = { version = "2.5.0" }
nom = "8"
tokio-util = "0.7.16"
libc = "0.2.174"
[features]
default = []
python-bridge = []
crates/biliup-cli/src/server/api/endpoints.rs
+61 -55
View File
@@ -25,7 +25,7 @@ use axum::http::StatusCode;
use axum::response::{IntoResponse, Response};
use axum::{Json, debug_handler};
use biliup::credential::Credential;
use chrono::Local;
use chrono::{Local, Utc};
use clap::ValueEnum;
use error_stack::{Report, ResultExt, bail};
use ormlite::{Insert, Model};
@@ -47,29 +47,29 @@ pub async fn get_streamers_endpoint(
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)?;
Ok(Json(
live_streamers
let mut results = Vec::new();
for x in live_streamers {
let option = workers
.read()
.unwrap()
.clone()
.into_iter()
.map(|x| {
let option = workers
.read()
.unwrap()
.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(),
))
.find(|worker| worker.live_streamer.id == x.id);
let status = match option.as_ref() {
Some(t) => format!("{:?}", *t.downloader_status.read().await),
None => String::new(),
};
results.push(LiveStreamerResponse {
status,
inner: x,
upload_status: option
.map(|t| format!("{:?}", *t.uploader_status.read().unwrap()))
.unwrap_or_default(),
});
}
Ok(Json(results))
}
pub async fn post_streamers_endpoint(
@@ -147,7 +147,7 @@ pub async fn delete_streamers_endpoint(
Ok(Json(live_streamers))
}
#[axum::debug_handler(state = ServiceRegister)]
// #[axum::debug_handler(state = ServiceRegister)]
pub async fn pause_streamers_endpoint(
State(service_register): State<ServiceRegister>,
State(pool): State<ConnectionPool>,
@@ -166,33 +166,32 @@ pub async fn pause_streamers_endpoint(
.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 {
let worker_status = w.downloader_status.read().await.clone();
match worker_status {
WorkerStatus::Working(d) => {
w.change_status(Stage::Download, WorkerStatus::Pause);
Some(d.clone())
}
WorkerStatus::Pending => {
monitor.rooms_handle.toggle(w.clone()).await;
w.change_status(Stage::Download, WorkerStatus::Pause);
None
}
WorkerStatus::Idle => {
monitor.rooms_handle.toggle(w.clone()).await;
w.change_status(Stage::Download, WorkerStatus::Pause);
None
d.stop().await.map_err(report_to_response)?;
tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
monitor
.rooms_handle
.toggle(w.clone(), WorkerStatus::Pause)
.await;
info!(workers=?&w.live_streamer.url, "successfully pause live streamers");
}
WorkerStatus::Pause => {
monitor.rooms_handle.toggle(w.clone()).await;
w.change_status(Stage::Download, WorkerStatus::Idle);
None
monitor
.rooms_handle
.toggle(w.clone(), WorkerStatus::Idle)
.await;
info!(workers=?&w.live_streamer.url, "successfully start live streamers");
}
_ => {
monitor
.rooms_handle
.toggle(w.clone(), WorkerStatus::Pause)
.await;
info!("非预期状态")
}
};
if let Some(d) = d {
d.stop().await.map_err(report_to_response)?;
}
info!(workers=?&w.live_streamer.url, "successfully pause live streamers");
}
Ok(Json(()))
@@ -513,20 +512,23 @@ pub async fn get_videos() -> Result<Json<Vec<serde_json::Value>>, Response> {
Ok(Json(file_list))
}
// #[axum::debug_handler(state = ServiceRegister)]
pub async fn get_status(
State(service_register): State<ServiceRegister>,
State(workers): State<Arc<RwLock<Vec<Arc<Worker>>>>>,
State(config): State<Arc<RwLock<Config>>>,
State(actor_handle): State<Arc<ActorHandle>>,
) -> Result<Json<serde_json::Value>, Response> {
let workers_clone = workers.read().unwrap().clone(); // 如果 Vec 本身可以克隆
let mut sw = Vec::new();
for worker in workers.read().unwrap().iter() {
for worker in &workers_clone {
sw.push(serde_json::json!({
"downloader_status": format!("{:?}", worker.downloader_status.read().unwrap()),
"uploader_status": format!("{:?}", worker.uploader_status.read().unwrap()),
"live_streamer": worker.live_streamer,
"upload_streamer": worker.upload_streamer,
}))
"downloader_status": format!("{:?}", worker.downloader_status.read().await),
"uploader_status": format!("{:?}", worker.uploader_status.read().unwrap()),
"live_streamer": worker.live_streamer,
"upload_streamer": worker.upload_streamer,
}));
}
Ok(Json(serde_json::json!({
@@ -573,10 +575,14 @@ pub async fn post_uploads(
if !videos.is_empty() {
let recorder = Recorder::new(
upload_config.title.clone(),
&upload_config.template_name,
"stream_title",
StreamerInfo::new(
&upload_config.template_name,
"stream_title",
"",
Utc::now(),
"",
),
"",
Local::now(),
);
let studio = build_studio(&upload_config, &bilibili, videos, recorder)
.await
crates/biliup-cli/src/server/app.rs
+20 -5
View File
@@ -19,7 +19,7 @@ use tower_http::cors::{AllowMethods, CorsLayer};
use tower_sessions::cookie::Key;
use tower_sessions::{ExpiredDeletion, Expiry, SessionManagerLayer};
use tower_sessions_sqlx_store::SqliteStore;
use tracing::info;
use tracing::{error, info};
/// 应用程序控制器，负责启动和管理Web服务器
pub struct ApplicationController;
@@ -48,6 +48,7 @@ impl ApplicationController {
// 配置会话管理层
let session_layer = SessionManagerLayer::new(session_store)
.with_secure(false)
.with_name("biliup.sid")
.with_expiry(Expiry::OnInactivity(Duration::days(7)));
// .with_signed(key);
@@ -93,10 +94,24 @@ impl ApplicationController {
.attach("error while starting API server")?;
// 等待会话清理任务完成
deletion_task
.await
.change_context(AppError::Unknown)?
.change_context(AppError::Unknown)?;
match deletion_task.await {
Ok(Ok(val)) => { /* 正常完成 */ }
Ok(Err(e)) => {
// 真正业务错误
return Err(e).change_context(AppError::Unknown);
}
Err(join_err) if join_err.is_cancelled() => {
info!("Deletion task cancelled on shutdown");
}
Err(join_err) if join_err.is_panic() => {
error!("Deletion task panicked: {join_err}");
return Err(AppError::Unknown.into());
}
Err(join_err) => {
error!("Join error: {join_err}");
return Err(AppError::Unknown.into());
}
}
Ok(())
}
crates/biliup-cli/src/server/common/download.rs
+51 -40
View File
@@ -46,36 +46,41 @@ impl DownloadGuard {
rooms_handle: Arc<RoomsHandle>,
downloader: Arc<dyn Downloader + Send + Sync>,
) -> Self {
// 更新工作器状态为工作中
worker.change_status(Stage::Download, WorkerStatus::Working(downloader.clone()));
Self {
worker,
danmaku_client,
rooms_handle,
}
}
}
impl Drop for DownloadGuard {
fn drop(&mut self) {
async fn cleanup(&mut self) {
// 异步清理任务
let danmaku = self.danmaku_client.clone();
let rooms_handle = self.rooms_handle.clone();
let worker = self.worker.clone();
tokio::spawn(async move {
if let Some(client) = danmaku {
if let Err(e) = client.stop().await {
error!("Error stopping danmaku client: {}", e);
}
if let Some(client) = danmaku {
if let Err(e) = client.stop().await {
error!("Error stopping danmaku client: {}", e);
}
if let WorkerStatus::Pause = *worker.downloader_status.read().unwrap() {
} else {
// 确保状态更新和资源清理
worker.change_status(Stage::Download, WorkerStatus::Idle);
rooms_handle.toggle(worker).await;
};
});
}
// 确保状态更新和资源清理
rooms_handle
.toggle(worker.clone(), WorkerStatus::Idle)
.await;
info!(
"{} => {:?}",
worker.live_streamer.url,
worker.downloader_status.read().await
);
}
}
impl Drop for DownloadGuard {
fn drop(&mut self) {
let worker = self.worker.clone();
info!("{}完成资源清理", worker.live_streamer.url);
}
}
@@ -256,6 +261,7 @@ pub struct DownloadTask {
struct DownloadComponents {
downloader: Arc<dyn Downloader + Send + Sync>,
danmaku_client: Option<Arc<dyn Downloader>>,
uploader: Sender<UploaderMessage>,
}
impl DownloadTask {
@@ -271,20 +277,9 @@ impl DownloadTask {
}
}
pub async fn execute(self, uploader: &Sender<UploaderMessage>) -> AppResult<()> {
// 初始化组件
let components = self.initialize_components().await?;
// 创建守卫确保清理
let _guard = DownloadGuard::new(
self.ctx.worker.clone(),
components.danmaku_client.clone(),
self.rooms_handle.clone(),
components.downloader.clone(),
);
pub async fn execute(self, components: DownloadComponents) -> AppResult<()> {
// 创建事件处理器
let processor = SegmentEventProcessor::new(uploader.clone(), self.ctx.clone());
let processor = SegmentEventProcessor::new(components.uploader.clone(), self.ctx.clone());
// 启动弹幕客户端
if let Some(ref client) = components.danmaku_client {
@@ -310,7 +305,7 @@ impl DownloadTask {
Ok(())
}
async fn initialize_components(&self) -> AppResult<DownloadComponents> {
async fn initialize_components(&self, uploader: Sender<UploaderMessage>) -> DownloadComponents {
// 获取配置和主播信息
let config = self.ctx.worker.get_config();
let streamer = self.ctx.worker.get_streamer();
@@ -330,10 +325,11 @@ impl DownloadTask {
.create_downloader(stream_info, config, self.ctx.recorder.clone())
.await;
Ok(DownloadComponents {
DownloadComponents {
downloader,
danmaku_client,
})
uploader,
}
}
async fn start_danmaku(&self, client: &Arc<dyn Downloader>) -> AppResult<()> {
@@ -365,9 +361,7 @@ impl DActor {
/// 运行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);
}
self.handle_message(msg).await
}
}
@@ -375,16 +369,33 @@ impl DActor {
///
/// # 参数
/// * `msg` - 要处理的下载消息
async fn handle_message(&mut self, msg: DownloaderMessage) -> AppResult<()> {
async fn handle_message(&mut self, msg: DownloaderMessage) {
match msg {
DownloaderMessage::Start(plugin, ctx, rooms_handle) => {
let worker = ctx.worker.clone();
// 创建下载任务
let task = DownloadTask::new(plugin, ctx, rooms_handle);
let task = DownloadTask::new(plugin, ctx, rooms_handle.clone());
// 初始化组件
let components = task.initialize_components(self.sender.clone()).await;
// 创建守卫确保清理
let mut guard = DownloadGuard::new(
worker.clone(),
components.danmaku_client.clone(),
rooms_handle.clone(),
components.downloader.clone(),
);
// 更新工作器状态为工作中
rooms_handle
.toggle(worker, WorkerStatus::Working(components.downloader.clone()))
.await;
// 执行下载（使用 Result 链式处理）
task.execute(&self.sender).await?;
let result = task.execute(components).await;
Ok(())
guard.cleanup().await;
info!("Handling message: {:?} done", result);
}
}
}
crates/biliup-cli/src/server/common/upload.rs
+3 -2
View File
@@ -384,7 +384,8 @@ impl UActor {
match msg {
UploaderMessage::SegmentEvent(rx, ctx) => {
ctx.worker
.change_status(Stage::Upload, WorkerStatus::Pending);
.change_status(Stage::Upload, WorkerStatus::Pending)
.await;
let inspect = rx.inspect(|f| {
let pool = ctx.pool.clone();
let streamer_info_id = ctx.stream_info.streamer_info.id;
@@ -417,7 +418,7 @@ impl UActor {
// 可以添加错误通知机制
}
info!(url=ctx.stream_info.streamer_info.url, result=?result, "后处理执行完毕：Finished processing segment event");
ctx.worker.change_status(Stage::Upload, WorkerStatus::Idle);
ctx.worker.change_status(Stage::Upload, WorkerStatus::Idle).await;
}
}
}
crates/biliup-cli/src/server/common/util.rs
+17 -21
View File
@@ -1,3 +1,4 @@
use crate::server::infrastructure::models::StreamerInfo;
use chrono::{DateTime, Duration, Local};
use serde::{Deserialize, Serialize};
use std::path::{Path, PathBuf};
@@ -8,30 +9,18 @@ use url::Url;
pub struct Recorder {
/// filename_prefix 文件名前缀模板
pub filename_prefix: Option<String>,
/// 主播名称
pub fname: String,
/// 直播间标题
pub room_title: String,
/// 直播间信息
pub streamer_info: StreamerInfo,
/// 录制后保存文件格式 (mp4, ts, mkv, flv)，不含点
pub suffix: String,
/// 开播时间
date_time: DateTime<Local>,
}
impl Recorder {
pub fn new(
filename_prefix: Option<String>,
fname: &str,
room_title: &str,
suffix: &str,
date_time: DateTime<Local>,
) -> Self {
pub fn new(filename_prefix: Option<String>, streamer_info: StreamerInfo, suffix: &str) -> Self {
Self {
filename_prefix,
fname: fname.to_string(),
room_title: room_title.to_string(),
streamer_info,
suffix: suffix.to_string(),
date_time,
}
}
@@ -40,15 +29,16 @@ impl Recorder {
let raw = if let Some(prefix) = &self.filename_prefix {
self.template_with(prefix)
} else {
format!("{}%Y-%m-%dT%H_%M_%S", self.fname)
format!("{}%Y-%m-%dT%H_%M_%S", self.streamer_info.name)
};
sanitize_filename(&raw)
}
fn template_with(&self, template: &str) -> String {
template
.replace("{streamer}", &self.fname)
.replace("{title}", &self.room_title)
.replace("{streamer}", &self.streamer_info.name)
.replace("{title}", &self.streamer_info.title)
.replace("{url}", &self.streamer_info.url)
}
/// 生成“基名”（不带扩展名），时间冲突时按秒+1继续尝试，直到唯一
@@ -68,11 +58,17 @@ impl Recorder {
/// 生成“基名”（不带扩展名）
pub fn format_filename(&self) -> String {
let template = self.filename_template();
self.date_time.format(&template).to_string()
self.streamer_info
.date
.with_timezone(&Local)
.format(&template)
.to_string()
}
pub fn format(&self, template: &str) -> String {
self.date_time
self.streamer_info
.date
.with_timezone(&Local)
.format(&self.template_with(template))
.to_string()
}
crates/biliup-cli/src/server/core/monitor.rs
+94 -27
View File
@@ -13,7 +13,7 @@ use std::sync::Arc;
use std::time::Duration;
use tokio::sync::oneshot;
use tokio::task::JoinHandle;
use tracing::{error, info};
use tracing::{debug, error, info};
/// 启动客户端监控循环
///
@@ -58,7 +58,8 @@ async fn start_client(
};
info!(url = url, "room: is live -> 开播了");
// 更新状态为等待中
room.change_status(Stage::Download, WorkerStatus::Pending);
room.change_status(Stage::Download, WorkerStatus::Pending)
.await;
stream_info.streamer_info = insert;
let streamer = room.get_streamer();
@@ -71,10 +72,8 @@ async fn start_client(
streamer
.filename_prefix
.or(room.get_config().filename_prefix.clone()),
&streamer.remark,
&stream_info.streamer_info.title,
stream_info.streamer_info.clone(),
&suffix,
stream_info.streamer_info.date.with_timezone(&Local),
);
// 修改 ctx
ctx.stream_info = stream_info;
@@ -90,8 +89,7 @@ async fn start_client(
.await
.is_ok()
{
// 切换房间状态
rooms_handle.toggle(room).await;
info!("成功开始录制 {}", url)
}
}
Ok(StreamStatus::Offline) => {}
@@ -229,11 +227,14 @@ impl RoomsHandle {
///
/// # 参数
/// * `worker` - 要切换的工作器
pub async fn toggle(&self, worker: Arc<Worker>) {
let msg = ActorMessage::Toggle(worker.clone());
pub async fn toggle(&self, worker: Arc<Worker>, status: WorkerStatus) {
let (send, recv) = oneshot::channel();
let msg = ActorMessage::Toggle(send, worker.clone(), status);
// 忽略发送错误
let _ = self.sender.send(msg).await;
recv.await.expect("Actor task has been killed")
}
}
@@ -252,7 +253,7 @@ enum ActorMessage {
id: i64,
},
/// 切换工作器状态
Toggle(Arc<Worker>),
Toggle(oneshot::Sender<()>, Arc<Worker>, WorkerStatus),
}
/// 房间Actor
@@ -282,12 +283,12 @@ impl RoomsActor {
/// 运行Actor主循环
async fn run(&mut self) {
while let Ok(msg) = self.receiver.recv().await {
self.handle_message(msg);
self.handle_message(msg).await;
}
}
/// 处理接收到的消息
fn handle_message(&mut self, msg: ActorMessage) {
async fn handle_message(&mut self, msg: ActorMessage) {
match msg {
ActorMessage::NextRoom { respond_to } => {
// `let _ =` 忽略发送时的任何错误
@@ -303,8 +304,10 @@ impl RoomsActor {
// 如果使用`select!`宏取消等待响应，可能会发生这种情况
let _ = respond_to.send(self.del(id));
}
ActorMessage::Toggle(worker) => {
self.toggle_keep_order(&worker);
ActorMessage::Toggle(sender, worker, status) => {
self.toggle_keep_order(&worker, status).await;
// `let _ =` 忽略发送时的任何错误
let _ = sender.send(());
}
}
}
@@ -342,19 +345,83 @@ impl RoomsActor {
}
/// 切换工作器状态，保持顺序
fn toggle_keep_order(&mut self, worker: &Arc<Worker>) -> bool {
// 从活跃列表移动到等待列表
if let Some(i) = self.rooms.iter().position(|x| x == worker) {
let val = self.rooms.remove(i); // 保序，但O(n)
self.waiting.push(val);
true
} else if let Some(i) = self.waiting.iter().position(|x| x == worker) {
// 从等待列表移动到活跃列表
let val = self.waiting.swap_remove(i);
self.rooms.push(val);
true
} else {
false
async fn toggle_keep_order(&mut self, worker: &Arc<Worker>, status: WorkerStatus) {
let mut write_guard = worker.downloader_status.write().await;
match (&*write_guard, &status) {
(WorkerStatus::Working(_), WorkerStatus::Idle) => {
if let Some(i) = self.waiting.iter().position(|x| x == worker) {
// 从等待列表移动到活跃列表
let val = self.waiting.swap_remove(i);
self.rooms.push(val);
*write_guard = status;
} else {
error!(
url = worker.live_streamer.url,
"working 状态应该在等待队列中"
)
}
}
(WorkerStatus::Idle | WorkerStatus::Pending, WorkerStatus::Working(_)) => {
// 从活跃列表移动到等待列表
if let Some(i) = self.rooms.iter().position(|x| x == worker) {
let val = self.rooms.remove(i); // 保序，但O(n)
self.waiting.push(val);
*write_guard = status;
} else {
error!(url = worker.live_streamer.url, "idle状态应该在活跃列表中");
}
}
(WorkerStatus::Pause, WorkerStatus::Idle) => {
if let Some(i) = self.waiting.iter().position(|x| x == worker) {
// 从等待列表移动到活跃列表
let val = self.waiting.swap_remove(i);
self.rooms.push(val);
*write_guard = status;
} else {
error!(
url = worker.live_streamer.url,
"working 状态应该在等待队列中"
)
}
}
(WorkerStatus::Idle, WorkerStatus::Pause) => {
// 从活跃列表移动到等待列表
if let Some(i) = self.rooms.iter().position(|x| x == worker) {
let val = self.rooms.remove(i); // 保序，但O(n)
self.waiting.push(val);
*write_guard = status;
} else {
error!(url = worker.live_streamer.url, "idle状态应该在活跃列表中");
}
}
(WorkerStatus::Working(_), WorkerStatus::Pause) => {
// 从活跃列表移动到等待列表
if let Some(i) = self.rooms.iter().position(|x| x == worker) {
let val = self.rooms.remove(i); // 保序，但O(n)
self.waiting.push(val);
*write_guard = status;
} else {
error!(url = worker.live_streamer.url, "idle状态应该在活跃列表中");
}
}
remaining => {
error!("非法的状态转移: {:?}", remaining);
}
}
// worker.change_status(Stage::Download, status).await;
// // 从活跃列表移动到等待列表
// if let Some(i) = self.rooms.iter().position(|x| x == worker) {
// let val = self.rooms.remove(i); // 保序，但O(n)
// self.waiting.push(val);
// true
// } else if let Some(i) = self.waiting.iter().position(|x| x == worker) {
// // 从等待列表移动到活跃列表
// let val = self.waiting.swap_remove(i);
// self.rooms.push(val);
// true
// } else {
// false
// }
}
}
crates/biliup-cli/src/server/infrastructure/context.rs
+4 -4
View File
@@ -48,7 +48,7 @@ impl Context {
#[derive(Debug)]
pub struct Worker {
/// 下载器状态
pub downloader_status: RwLock<WorkerStatus>,
pub downloader_status: tokio::sync::RwLock<WorkerStatus>,
/// 上传器状态
pub uploader_status: RwLock<WorkerStatus>,
/// 直播主播信息
@@ -76,7 +76,7 @@ impl Worker {
client: StatelessClient,
) -> Self {
Self {
downloader_status: RwLock::new(Default::default()),
downloader_status: tokio::sync::RwLock::new(Default::default()),
uploader_status: Default::default(),
live_streamer,
upload_streamer,
@@ -108,10 +108,10 @@ impl Worker {
/// # 参数
/// * `stage` - 工作阶段（下载或上传）
/// * `status` - 新的工作状态
pub fn change_status(&self, stage: Stage, status: WorkerStatus) {
pub async fn change_status(&self, stage: Stage, status: WorkerStatus) {
match stage {
Stage::Download => {
*self.downloader_status.write().unwrap() = status;
*self.downloader_status.write().await = status;
}
Stage::Upload => {
*self.uploader_status.write().unwrap() = status;
crates/biliup-cli/src/server/infrastructure/models.rs
+19
View File
@@ -25,6 +25,25 @@ pub struct StreamerInfo {
pub live_cover_path: String,
}
impl StreamerInfo {
pub fn new(
name: &str,
url: &str,
title: &str,
date: DateTime<Utc>,
live_cover_path: &str,
) -> Self {
Self {
id: -1,
name: name.to_string(),
url: url.to_string(),
title: title.to_string(),
date,
live_cover_path: live_cover_path.to_string(),
}
}
}
/// 文件列表模型
/// 存储录制文件的信息
#[derive(Model, Debug, Clone, Serialize, Deserialize)]
crates/biliup-cli/src/server/infrastructure/models/hook_step.rs
+56 -29
View File
@@ -104,48 +104,75 @@ impl HookStep {
async fn move_file(&self, video_paths: &[&Path], target_dir: &str) -> AppResult<()> {
let target_path = Path::new(target_dir);
// 创建目标目录（如果不存在）
if !target_path.exists() {
fs::create_dir_all(target_path)
.await
.change_context(AppError::Unknown)?;
}
// 移动每个视频文件到目标目录
for video_path in video_paths {
let file_name = video_path
.file_name()
.ok_or(AppError::Custom("Invalid file name".to_string()))?;
let destination = target_path.join(file_name);
self.move_single_file(video_path, target_path).await?;
info!(
"Moving {} to {}",
video_path.display(),
destination.display()
);
fs::rename(video_path, destination)
.await
.change_context(AppError::Unknown)?;
let buf = video_path.with_extension("xml");
if buf.exists() {
let file_name = buf
.file_name()
.ok_or(AppError::Custom("Invalid file name".to_string()))?;
let destination = target_path.join(file_name);
fs::rename(&buf, &destination)
.await
.change_context(AppError::Unknown)?;
info!(
"移动弹幕文件: Moving {} to {}",
buf.display(),
destination.display()
);
// 移动对应的 XML 文件
let xml_path = video_path.with_extension("xml");
if xml_path.exists() {
info!("移动弹幕文件: {}", xml_path.display());
self.move_single_file(&xml_path, target_path).await?;
}
}
Ok(())
}
/// 移动单个文件，支持跨文件系统
async fn move_single_file(&self, source: &Path, target_dir: &Path) -> AppResult<()> {
let file_name = source
.file_name()
.ok_or(AppError::Custom("Invalid file name".to_string()))?;
let destination = target_dir.join(file_name);
info!("Moving {} to {}", source.display(), destination.display());
// 先尝试 rename（快速，仅同文件系统）
match fs::rename(source, &destination).await {
Ok(_) => Ok(()),
Err(e) => {
// 检查是否是跨文件系统错误
if HookStep::is_cross_device_error(&e) {
info!("检测到跨文件系统操作，使用复制+删除模式");
// 复制文件
fs::copy(source, &destination)
.await
.change_context(AppError::Unknown)?;
// 删除原文件
fs::remove_file(source)
.await
.change_context(AppError::Unknown)?;
Ok(())
} else {
Err(e).change_context(AppError::Unknown)?
}
}
}
}
/// 判断是否是跨设备/文件系统错误
fn is_cross_device_error(error: &std::io::Error) -> bool {
#[cfg(unix)]
{
error.raw_os_error() == Some(libc::EXDEV)
}
#[cfg(windows)]
{
error.raw_os_error() == Some(17) // ERROR_NOT_SAME_DEVICE
}
#[cfg(not(any(unix, windows)))]
{
error.kind() == std::io::ErrorKind::Other
}
}
/// 删除指定的视频文件
///
/// # 参数
crates/biliup-cli/src/server/infrastructure/service_register.rs
+1 -1
View File
@@ -136,7 +136,7 @@ impl ServiceRegister {
if len == 0 {
*manager.monitor.lock().unwrap() = None;
}
let downloader = match &*removed.downloader_status.read().unwrap() {
let downloader = match &*removed.downloader_status.read().await {
WorkerStatus::Working(downloader) => Some(downloader.clone()),
_ => None,
};
crates/biliup-cli/src/server/infrastructure/users.rs
+12 -1
View File
@@ -46,7 +46,7 @@ impl AuthUser for User {
// 认证凭据结构，用于从表单中提取认证字段
// 用于与后端进行请求认证
#[derive(Debug, Clone, Deserialize)]
#[derive(Clone, Deserialize)]
pub struct Credentials {
/// 用户名
pub username: String,
@@ -56,6 +56,17 @@ pub struct Credentials {
pub next: Option<String>,
}
// 手动实现Debug trait以避免意外记录密码哈希
impl std::fmt::Debug for Credentials {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("Credentials")
.field("next", &self.next)
.field("username", &self.username)
.field("password", &"[redacted]")
.finish()
}
}
/// 认证后端
/// 负责处理用户认证相关的数据库操作
#[derive(Debug, Clone)]
crates/biliup-cli/src/server/router.rs
+1 -1
View File
@@ -28,7 +28,7 @@ pub fn router(service_register: ServiceRegister) -> Router<()> {
.put(put_streamers_endpoint), // 更新主播
)
.route("/v1/streamers/{id}", delete(delete_streamers_endpoint)) // 删除主播
.route("/v1/streamers/{id}/pause", put(pause_streamers_endpoint)) // 删除主播
.route("/v1/streamers/{id}/pause", put(pause_streamers_endpoint))
// 配置管理路由
.route(
"/v1/configuration",