From 267bda7360c7dfe9bbd4b5e7a7d341de0985abc9 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Tue, 23 Sep 2025 05:14:28 +0800 Subject: [PATCH] Add and improve Rust doc comments across server modules Added and expanded Rust documentation comments (///) for structs, enums, traits, functions, and methods throughout the core and infrastructure modules. These comments clarify the purpose, parameters, and return values of types and functions, improving code readability and maintainability for developers. No functional or logic changes were made. --- .../src/server/core/download_manager.rs | 86 ++++++++++++- .../biliup-cli/src/server/core/downloader.rs | 44 +++++-- .../core/downloader/ffmpeg_downloader.rs | 18 +++ .../server/core/downloader/stream_gears.rs | 46 +++++-- crates/biliup-cli/src/server/core/monitor.rs | 114 ++++++++++++++---- crates/biliup-cli/src/server/core/plugin.rs | 34 +++++- .../server/infrastructure/connection_pool.rs | 11 +- .../src/server/infrastructure/context.rs | 18 +++ .../src/server/infrastructure/models.rs | 83 +++++++++++-- .../server/infrastructure/models/hook_step.rs | 31 ++++- .../src/server/infrastructure/repositories.rs | 40 ++++++ .../server/infrastructure/service_register.rs | 42 ++++++- .../src/server/infrastructure/users.rs | 68 ++++++++--- 13 files changed, 550 insertions(+), 85 deletions(-) diff --git a/crates/biliup-cli/src/server/core/download_manager.rs b/crates/biliup-cli/src/server/core/download_manager.rs index 42ed920..e5551fd 100644 --- a/crates/biliup-cli/src/server/core/download_manager.rs +++ b/crates/biliup-cli/src/server/core/download_manager.rs @@ -22,13 +22,23 @@ use std::time::Instant; use tokio::task::JoinHandle; use tracing::{error, info, warn}; +/// 下载管理器 +/// 负责管理特定平台的下载任务，包括监控器和插件 pub struct DownloadManager { + /// 监控器实例（可选，使用Mutex保护） pub monitor: Mutex>>, + /// 下载插件 plugin: Arc, + /// Actor处理器 actor_handle: Arc, } impl DownloadManager { + /// 创建新的下载管理器实例 + /// + /// # 参数 + /// * `plugin` - 下载插件实现 + /// * `actor_handle` - Actor处理器 pub fn new( plugin: impl DownloadPlugin + Send + Sync + 'static, actor_handle: Arc, @@ -40,8 +50,11 @@ impl DownloadManager { } } + /// 确保监控器存在，如果不存在则创建新的 + /// + /// # 返回 + /// 返回监控器的Arc引用 pub fn ensure_monitor(&self) -> Arc { - // Monitor::new(self.plugin.name(), url) self.monitor .lock() .unwrap() @@ -51,6 +64,13 @@ impl DownloadManager { .clone() } + /// 检查URL是否匹配此下载管理器的插件 + /// + /// # 参数 + /// * `url` - 要检查的URL + /// + /// # 返回 + /// 如果URL匹配返回true，否则返回false pub fn matches(&self, url: &str) -> bool { self.plugin.matches(url) } @@ -62,16 +82,22 @@ 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 { @@ -80,49 +106,61 @@ impl DActor { } } + /// 处理下载消息 + /// + /// # 参数 + /// * `msg` - 要处理的下载消息 async fn handle_message(&mut self, msg: DownloaderMessage) -> AppResult<()> { match msg { DownloaderMessage::Start(plugin, stream_info, ctx, rooms_handle) => { + // 获取配置和主播信息 let config = ctx.worker.get_config(); let streamer = ctx.worker.get_streamer(); - // info!(stream_info=?stream_info, "Create downloader"); + // 确定文件格式后缀 let suffix = streamer .format .unwrap_or_else(|| stream_info.suffix.clone()); - // println!("suffix: {suffix}"); + // 创建录制器 let recorder = Recorder::new( streamer.filename_prefix, &streamer.remark, &stream_info.title, &suffix, ); + // 创建下载器实例 let downloader = plugin .create_downloader(&stream_info, config, recorder) .await; + // 更新工作器状态为工作中 ctx.worker .change_status(Stage::Download, WorkerStatus::Working); + // 初始化弹幕客户端（如果存在） let mut danmaku_client = None; if let Some(danmaku) = ctx.extension.get::>() { - danmaku + let _ = danmaku .download(Box::new(|_| {})) .await .inspect_err(|e| error!(e=?e)); danmaku_client = Some(danmaku.clone()) } + // 创建分段事件处理通道 let (tx, rx) = bounded(16); let hook = { let room = ctx.worker.clone(); let sender = self.sender.clone(); let danmaku_client = danmaku_client.clone(); move |event: SegmentEvent| { + // 如果有弹幕客户端，触发滚动保存 if let Some(danmaku) = &danmaku_client { - danmaku + let _ = danmaku .rolling(&event.file_path.display().to_string()) .inspect_err(|e| error!(e)); } + // 处理分段事件 if event.segment_index == 0 { + // 第一个分段，发送到上传器 match sender.force_send(UploaderMessage::SegmentEvent( event, rx.clone(), @@ -135,6 +173,7 @@ impl DActor { Ok(None) => {} }; } else { + // 后续分段，发送到内部通道 match tx.clone().force_send(event) { Ok(Some(ret)) => { warn!(SegmentEvent = ?ret, "replace an existing message in the channel"); @@ -145,6 +184,7 @@ impl DActor { } } }; + // 开始下载 match downloader.download(Box::new(hook)).await { Ok(status) => { println!("Download completed with status: {:?}", status); @@ -153,32 +193,44 @@ impl DActor { error!("download error: {:?}", err); } }; + // 停止弹幕客户端 if let Some(danmaku) = &danmaku_client { - danmaku.stop().await.inspect_err(|e| error!(e)); + let _ = danmaku.stop().await.inspect_err(|e| error!(e)); } + // 更新工作器状态为空闲 ctx.worker .change_status(Stage::Download, WorkerStatus::Idle); + // 切换房间状态 rooms_handle.toggle(ctx.worker).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(event, rx, worker) => { @@ -349,26 +401,42 @@ impl UActor { } } +/// Actor处理器 +/// 管理下载和上传Actor的生命周期 pub struct ActorHandle { + /// 下载信号量数量 download_semaphore: u32, + /// 上传信号量数量 update_semaphore: u32, + /// 上传消息发送器 pub up_sender: Sender, + /// 下载消息发送器 pub down_sender: Sender, + /// 下载Actor任务句柄列表 d_kills: Vec>, + /// 上传Actor任务句柄列表 u_kills: Vec>, } impl ActorHandle { + /// 创建新的Actor处理器实例 + /// + /// # 参数 + /// * `download_semaphore` - 下载Actor数量 + /// * `update_semaphore` - 上传Actor数量 pub fn new(download_semaphore: u32, update_semaphore: u32) -> Self { + // 创建消息通道 let (up_tx, up_rx) = bounded(16); let (down_tx, down_rx) = bounded(1); let mut d_kills = Vec::new(); let mut u_kills = Vec::new(); + // 创建下载Actor for _ in 0..download_semaphore { let mut d_actor = DActor::new(down_rx.clone(), up_tx.clone()); let d_kill = tokio::spawn(async move { d_actor.run().await }); d_kills.push(d_kill) } + // 创建上传Actor for _ in 0..update_semaphore { let mut u_actor = UActor::new(up_rx.clone()); let u_kill = tokio::spawn(async move { u_actor.run().await }); @@ -386,12 +454,18 @@ impl ActorHandle { } } +/// 上传消息枚举 +/// 定义上传Actor可以处理的消息类型 #[derive(Debug)] pub enum UploaderMessage { + /// 分段事件消息，包含事件、接收器和工作器 SegmentEvent(SegmentEvent, Receiver, Arc), } +/// 下载消息枚举 +/// 定义下载Actor可以处理的消息类型 pub enum DownloaderMessage { + /// 开始下载消息，包含插件、流信息、上下文和房间句柄 Start( Arc, StreamInfo, diff --git a/crates/biliup-cli/src/server/core/downloader.rs b/crates/biliup-cli/src/server/core/downloader.rs index 36506d0..6335b48 100644 --- a/crates/biliup-cli/src/server/core/downloader.rs +++ b/crates/biliup-cli/src/server/core/downloader.rs @@ -1,4 +1,6 @@ +/// FFmpeg下载器实现 pub mod ffmpeg_downloader; +/// Stream-gears下载器实现 pub mod stream_gears; use crate::server::common::util::Recorder; @@ -9,6 +11,7 @@ use std::collections::HashMap; use std::path::PathBuf; /// 下载器配置 +/// 包含下载过程中需要的各种参数和设置 #[derive(Debug, Clone, Deserialize, Serialize)] pub struct DownloadConfig { /// 分段时长 (格式: "HH:MM:SS") @@ -20,41 +23,50 @@ pub struct DownloadConfig { /// HTTP请求头 pub headers: HashMap, - /// 输出文件名前缀 + /// 录制器实例 pub recorder: Recorder, - // 输出文件名前缀 - // pub filename_prefix: String, + /// 输出目录路径 pub output_dir: PathBuf, } impl DownloadConfig { + /// 生成输出文件名 + /// + /// # 返回 + /// 返回完整的输出文件路径 fn generate_output_filename(&self) -> PathBuf { self.output_dir.join(self.recorder.generate_path()) } } +/// 下载器类型枚举 +/// 定义支持的各种下载器类型 #[derive(Debug, Clone, Deserialize, Serialize)] #[serde(rename_all = "kebab-case")] pub enum DownloaderType { + /// Ytarchive下载器 Ytarchive, + /// 同步下载器 #[serde(rename = "sync-downloader")] SyncDownloader, /// 使用stream-gears #[serde(rename = "stream-gears")] StreamGears, + /// FFmpeg下载器 Ffmpeg, /// FFmpeg外部分段 FfmpegExternal, /// FFmpeg内部分段 FfmpegInternal, - /// Streamlink + /// Streamlink下载器 Streamlink, - /// yt-dlp + /// yt-dlp下载器 YtDlp, } /// 分段事件 +/// 当下载器完成一个分段时触发的事件 #[derive(Debug, Clone)] pub struct SegmentEvent { /// 分段文件路径 @@ -68,6 +80,7 @@ pub struct SegmentEvent { } /// 下载状态 +/// 表示下载器当前的状态 #[derive(Debug, Clone, PartialEq)] pub enum DownloadStatus { /// 正在下载 @@ -81,27 +94,40 @@ pub enum DownloadStatus { } /// 下载器基础trait +/// 定义所有下载器必须实现的基本接口 #[async_trait] pub trait Downloader: Send + Sync { /// 开始下载 + /// + /// # 参数 + /// * `callback` - 分段完成时的回调函数 + /// + /// # 返回 + /// 返回下载状态 async fn download( &self, - // sender: Sender, - // worker: Arc, callback: Box, ) -> AppResult; /// 停止下载 async fn stop(&self) -> Result<(), Box>; - /// 获取当前下载状态 - /// async fn get_status(&self) -> DownloadStatus; + /// 滚动保存（用于弹幕等） + /// + /// # 参数 + /// * `file_name` - 文件名 fn rolling(&self, file_name: &str) -> Result<(), Box> { Ok(()) } } /// 解析时长字符串 "HH:MM:SS" 为秒数 +/// +/// # 参数 +/// * `duration` - 时长字符串，格式为"HH:MM:SS" +/// +/// # 返回 +/// 返回总秒数 fn parse_duration(duration: &str) -> u64 { let parts: Vec<&str> = duration.split(':').collect(); if parts.len() == 3 { 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 7d5bd3f..d134ffe 100644 --- a/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs +++ b/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs @@ -14,10 +14,15 @@ use tokio::sync::RwLock; use tracing::info; /// FFmpeg下载器实现 +/// 使用FFmpeg进行直播流下载，支持内部和外部分段 pub struct FfmpegDownloader { + /// 下载配置 config: DownloadConfig, + /// 流URL url: String, + /// 下载状态 status: Arc>, + /// 进程句柄 process_handle: Arc>>, /// 额外的FFmpeg参数 @@ -28,6 +33,13 @@ pub struct FfmpegDownloader { } impl FfmpegDownloader { + /// 创建新的FFmpeg下载器实例 + /// + /// # 参数 + /// * `url` - 流URL + /// * `config` - 下载配置 + /// * `extra_args` - 额外的FFmpeg参数 + /// * `downloader_type` - 下载器类型 pub fn new( url: &str, config: DownloadConfig, @@ -45,6 +57,7 @@ impl FfmpegDownloader { } /// 构建内部分段模式的FFmpeg命令参数 + /// 使用FFmpeg的segment muxer进行自动分段 fn build_ffmpeg_args_internal_segment(&self) -> Vec { let mut args = Vec::new(); @@ -82,6 +95,7 @@ impl FfmpegDownloader { } /// 构建外部分段模式的FFmpeg命令参数 + /// 通过外部控制进行分段，每次录制固定时长或大小 fn build_ffmpeg_args_external_segment(&self) -> Vec { let mut args = Vec::new(); @@ -109,6 +123,7 @@ impl FfmpegDownloader { } /// 添加通用的输入参数 + /// 包括覆盖文件、HTTP头、超时设置等 fn append_common_input_args(&self, args: &mut Vec) { args.push("-y".to_string()); // 覆盖已存在文件 @@ -141,6 +156,7 @@ impl FfmpegDownloader { } /// 添加通用的输出参数 + /// 包括编码设置、格式特定参数等 fn append_common_output_args(&self, args: &mut Vec, format: &str) { // -c copy: 直接复制编码，不重新编码 // 减少CPU使用，保持原始质量 @@ -176,6 +192,7 @@ impl FfmpegDownloader { } /// 执行外部分段下载 + /// 每次录制一个完整的分段文件 async fn download_external( &self, callback: Box, @@ -252,6 +269,7 @@ impl FfmpegDownloader { } /// 执行内部分段下载 + /// 使用FFmpeg的segment muxer自动分段 async fn download_internal( &self, callback: Box, diff --git a/crates/biliup-cli/src/server/core/downloader/stream_gears.rs b/crates/biliup-cli/src/server/core/downloader/stream_gears.rs index 76a81b2..83220a8 100644 --- a/crates/biliup-cli/src/server/core/downloader/stream_gears.rs +++ b/crates/biliup-cli/src/server/core/downloader/stream_gears.rs @@ -14,17 +14,32 @@ use std::sync::{Arc, RwLock}; use std::time::SystemTime; use tracing::{debug, error, info}; -// 具体下载器实现 +/// Stream-gears下载器实现 +/// 使用stream-gears库进行直播流下载 pub struct StreamGears { + /// 流URL url: String, + /// HTTP请求头 header_map: HeaderMap, + /// 文件名 file_name: String, + /// 分段配置 segment: Segmentable, + /// 代理设置（可选） proxy: Option, + /// 下载状态 status: Arc>, } impl StreamGears { + /// 创建新的Stream-gears下载器实例 + /// + /// # 参数 + /// * `url` - 流URL + /// * `header_map` - HTTP请求头 + /// * `file_name` - 输出文件名 + /// * `segment` - 分段配置 + /// * `proxy` - 代理设置（可选） pub fn new( url: &str, header_map: HeaderMap, @@ -45,33 +60,36 @@ impl StreamGears { #[async_trait] impl Downloader for StreamGears { + /// 开始下载流 + /// + /// # 参数 + /// * `callback` - 分段完成时的回调函数 async fn download( &self, callback: Box, ) -> AppResult { let url = self.url.clone(); let file_name = self.file_name.clone(); - let headers_in = self.header_map.clone(); + let _headers_in = self.header_map.clone(); let proxy = self.proxy.clone(); - let status = Arc::clone(&self.status); - // response.read_exact(buf)?; - // let out = File::create(format!("{}.flv", file_name)).expect("Unable to create file."); - // let mut writer = BufWriter::new(out); - // let mut buf = [0u8; 8 * 1024]; - // response.copy_to(&mut writer)?; - // io::copy(&mut resp, &mut out).expect("Unable to copy the content."); + let _status = Arc::clone(&self.status); + + // 创建HTTP客户端 let client = StatelessClient::new(self.header_map.clone(), proxy.as_deref()); + // 获取可重试的响应 let response = client .retryable(&url) .await .change_context(AppError::Unknown)?; + // 创建连接 let mut connection = Connection::new(response); - // let buf = &mut [0u8; 9]; + // 读取帧头 let bytes = connection .read_frame(9) .await .change_context(AppError::Unknown)?; let mut i = 0; + // 创建分段回调钩子 let hook = move |s: &str| { i += 1; let event = SegmentEvent { @@ -83,10 +101,12 @@ impl Downloader for StreamGears { callback(event); }; + // 解析流头部，判断流类型 match header(&bytes) { Ok((_i, header)) => { debug!("header: {header:#?}"); info!("Downloading {}...", url); + // FLV流下载 let file = LifecycleFile::with_hook(&file_name, "flv", hook); httpflv::download(connection, file, self.segment.clone()).await; } @@ -95,6 +115,7 @@ impl Downloader for StreamGears { } Err(e) => { error!("{e}"); + // HLS流下载 let file = LifecycleFile::with_hook(&file_name, "ts", hook); hls::download(&url, &client, file, self.segment.clone()) .await @@ -104,9 +125,10 @@ impl Downloader for StreamGears { Ok(DownloadStatus::StreamEnded) } + /// 停止下载 async fn stop(&self) -> Result<(), Box> { - // 仅发出取消信号并更新状态。 - // 如上所述，如果底层下载函数不支持取消，这里不能真正中断正在进行的下载。 + // 仅发出取消信号并更新状态 + // 如果底层下载函数不支持取消，这里不能真正中断正在进行的下载 todo!() } } diff --git a/crates/biliup-cli/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs index b42173d..95f80e5 100644 --- a/crates/biliup-cli/src/server/core/monitor.rs +++ b/crates/biliup-cli/src/server/core/monitor.rs @@ -8,6 +8,13 @@ use tokio::sync::oneshot; use tokio::task::JoinHandle; use tracing::info; +/// 启动客户端监控循环 +/// +/// # 参数 +/// * `rooms_handle` - 房间处理器 +/// * `plugin` - 下载插件 +/// * `actor_handle` - Actor处理器 +/// * `interval` - 监控间隔（秒） async fn start_client( rooms_handle: Arc, plugin: Arc, @@ -17,15 +24,19 @@ async fn start_client( let platform_name = &rooms_handle.name; info!("start -> [{platform_name}]"); loop { + // 获取下一个要检查的房间 if let Some(room) = rooms_handle.next().await { let ulr = room.get_streamer().url; interval = room.get_config().event_loop_interval; info!("[{platform_name}] room: {ulr}"); let mut ctx = Context::new(room.clone()); + // 检查直播状态 match plugin.check_status(&mut ctx).await.unwrap() { StreamStatus::Live { stream_info } => { info!("room: {ulr} is live -> {:?}", stream_info); + // 更新状态为等待中 room.change_status(Stage::Download, WorkerStatus::Pending); + // 发送下载开始消息 if actor_handle .down_sender .send(DownloaderMessage::Start( @@ -37,6 +48,7 @@ async fn start_client( .await .is_ok() { + // 切换房间状态 rooms_handle.toggle(room).await; } } @@ -44,22 +56,34 @@ async fn start_client( StreamStatus::Unknown => {} }; } + // 等待下一次检查 tokio::time::sleep(Duration::from_secs(interval)).await; } } +/// 监控器 +/// 负责监控直播间状态并管理下载任务 #[derive(Debug)] pub struct Monitor { + /// 房间处理器 pub rooms_handle: Arc, + /// 监控任务句柄 kill: JoinHandle<()>, } impl Monitor { + /// 创建新的监控器实例 + /// + /// # 参数 + /// * `plugin` - 下载插件 + /// * `actor_handle` - Actor处理器 pub fn new( plugin: Arc, actor_handle: Arc, ) -> Self { + // 创建房间处理器 let handle = Arc::new(RoomsHandle::new(plugin.name())); + // 启动监控任务 let join_handle = tokio::spawn({ let handle = Arc::clone(&handle); async move { @@ -74,24 +98,37 @@ impl Monitor { } impl Drop for Monitor { + /// 监控器销毁时的清理逻辑 fn drop(&mut self) { + // 终止监控任务 self.kill.abort(); self.rooms_handle.kill.abort(); info!("Monitor [{}] killed", self.rooms_handle.name) } } +/// 房间处理器 +/// 管理多个直播间的状态和操作 #[derive(Debug)] pub struct RoomsHandle { + /// 平台名称 name: String, + /// 消息发送器 sender: Sender, + /// Actor任务句柄 kill: JoinHandle<()>, } impl RoomsHandle { + /// 创建新的房间处理器实例 + /// + /// # 参数 + /// * `name` - 平台名称 pub fn new(name: &str) -> Self { + // 创建消息通道 let (sender, receiver) = bounded(1); let mut actor = RoomsActor::new(receiver); + // 启动Actor任务 let kill = tokio::spawn(async move { actor.run().await }); Self { @@ -101,11 +138,22 @@ impl RoomsHandle { } } + /// 添加工作器到房间列表 + /// + /// # 参数 + /// * `worker` - 要添加的工作器 pub async fn add(&self, worker: Arc) { let msg = ActorMessage::Add(worker); let _ = self.sender.send(msg).await; } + /// 删除指定ID的工作器 + /// + /// # 参数 + /// * `id` - 要删除的工作器ID + /// + /// # 返回 + /// 返回剩余工作器数量 pub async fn del(&self, id: i64) -> usize { let (send, recv) = oneshot::channel(); let msg = ActorMessage::Del { @@ -113,54 +161,71 @@ impl RoomsHandle { id, }; - // Ignore send errors. If this send fails, so does the - // recv.await below. There's no reason to check the - // failure twice. + // 忽略发送错误。如果发送失败，下面的recv.await也会失败 + // 没有必要检查两次失败 let _ = self.sender.send(msg).await; recv.await.expect("Actor task has been killed") } + /// 获取下一个要处理的工作器 + /// + /// # 返回 + /// 返回下一个工作器，如果没有则返回None async fn next(&self) -> Option> { let (send, recv) = oneshot::channel(); let msg = ActorMessage::NextRoom { respond_to: send }; - // Ignore send errors. If this send fails, so does the - // recv.await below. There's no reason to check the - // failure twice. + // 忽略发送错误。如果发送失败，下面的recv.await也会失败 + // 没有必要检查两次失败 let _ = self.sender.send(msg).await; recv.await.expect("Actor task has been killed") } + /// 切换工作器状态（在活跃和等待列表之间） + /// + /// # 参数 + /// * `worker` - 要切换的工作器 pub async fn toggle(&self, worker: Arc) { let msg = ActorMessage::Toggle(worker.clone()); - // Ignore send errors. If this send fails, so does the - // recv.await below. There's no reason to check the - // failure twice. + // 忽略发送错误 let _ = self.sender.send(msg).await; } } +/// Actor消息枚举 +/// 定义RoomsActor可以处理的消息类型 enum ActorMessage { + /// 获取下一个房间 NextRoom { respond_to: oneshot::Sender>>, }, + /// 添加工作器 Add(Arc), + /// 删除工作器 Del { respond_to: oneshot::Sender, id: i64, }, + /// 切换工作器状态 Toggle(Arc), } +/// 房间Actor +/// 管理房间列表的内部Actor struct RoomsActor { + /// 消息接收器 receiver: Receiver, + /// 当前索引 index: usize, + /// 活跃房间列表 rooms: Vec>, + /// 等待房间列表 waiting: Vec>, } impl RoomsActor { + /// 创建新的房间Actor实例 fn new(receiver: Receiver) -> Self { Self { receiver, @@ -170,19 +235,19 @@ impl RoomsActor { } } + /// 运行Actor主循环 async fn run(&mut self) { while let Ok(msg) = self.receiver.recv().await { self.handle_message(msg); } } + /// 处理接收到的消息 fn handle_message(&mut self, msg: ActorMessage) { match msg { ActorMessage::NextRoom { respond_to } => { - // The `let _ =` ignores any errors when sending. - // - // This can happen if the `select!` macro is used - // to cancel waiting for the response. + // `let _ =` 忽略发送时的任何错误 + // 如果使用`select!`宏取消等待响应，可能会发生这种情况 let _ = respond_to.send(self.next()); } ActorMessage::Add(worker) => { @@ -190,10 +255,8 @@ impl RoomsActor { info!("Added room [{:?}]", self.rooms); } ActorMessage::Del { respond_to, id } => { - // The `let _ =` ignores any errors when sending. - // - // This can happen if the `select!` macro is used - // to cancel waiting for the response. + // `let _ =` 忽略发送时的任何错误 + // 如果使用`select!`宏取消等待响应，可能会发生这种情况 let _ = respond_to.send(self.del(id)); } ActorMessage::Toggle(worker) => { @@ -202,26 +265,30 @@ impl RoomsActor { } } + /// 获取下一个工作器（循环遍历） fn next(&mut self) -> Option> { - // 如果内部 Vec 是空的，迭代结束（虽然是循环迭代器，但空集合无法产生任何值） + // 如果内部Vec是空的，迭代结束（虽然是循环迭代器，但空集合无法产生任何值） if self.rooms.is_empty() { return None; } // 获取当前位置元素的克隆 - // 使用 .get() 并 .cloned() 是安全的做法 + // 使用.get()并.cloned()是安全的做法 let item = self.rooms[self.index].clone(); - // 更新 index 以便下一次调用，使用取模运算实现循环 + // 更新index以便下一次调用，使用取模运算实现循环 self.index = (self.index + 1) % self.rooms.len(); Some(item) } + /// 删除指定ID的工作器 fn del(&mut self, id: i64) -> usize { + // 从活跃房间列表中删除 if let Some(i) = self.rooms.iter().position(|x| x.live_streamer.id == id) { - self.rooms.remove(i); // 保序，但 O(n) + self.rooms.remove(i); // 保序，但O(n) } else if let Some(i) = self.waiting.iter().position(|x| x.live_streamer.id == id) { + // 从等待房间列表中删除 self.waiting.swap_remove(i); }; info!("Removed room [{:?}] {}", self.rooms, self.rooms.len()); @@ -229,12 +296,15 @@ impl RoomsActor { self.rooms.len() + self.waiting.len() } + /// 切换工作器状态，保持顺序 fn toggle_keep_order(&mut self, worker: &Arc) -> bool { + // 从活跃列表移动到等待列表 if let Some(i) = self.rooms.iter().position(|x| x == worker) { - let val = self.rooms.remove(i); // 保序，但 O(n) + 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 diff --git a/crates/biliup-cli/src/server/core/plugin.rs b/crates/biliup-cli/src/server/core/plugin.rs index 55232a9..809188c 100644 --- a/crates/biliup-cli/src/server/core/plugin.rs +++ b/crates/biliup-cli/src/server/core/plugin.rs @@ -10,42 +10,68 @@ use std::collections::HashMap; use std::path::Path; use time::OffsetDateTime; -// Stream information structures +/// 流信息结构 +/// 包含直播流的详细信息 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct StreamInfo { + /// 主播名称 pub name: String, + /// 直播间URL pub url: String, + /// 原始流URL pub raw_stream_url: String, + /// 直播标题 pub title: String, + /// 直播开始时间 #[serde(with = "time::serde::rfc3339")] - pub date: OffsetDateTime, // 保存 Python 的时间对象（如 time.struct_time） - // pub end_time: PyObject, // 同上 + pub date: OffsetDateTime, + /// 直播封面路径（可选） pub live_cover_path: Option, + /// 平台名称 pub platform: String, + /// 流请求头 pub stream_headers: HashMap, + /// 文件后缀 pub suffix: String, } +/// 流状态枚举 +/// 表示直播流的当前状态 #[derive(Debug, Clone, Serialize, Deserialize)] pub enum StreamStatus { + /// 正在直播，包含流信息 Live { stream_info: StreamInfo }, + /// 离线状态 Offline, + /// 未知状态 Unknown, } +/// 下载基础trait +/// 定义下载器的基本功能接口 #[async_trait] pub trait DownloadBase: Send + Sync { + /// 检查流是否可用 async fn check_stream(&self) -> Result>; + /// 获取流信息 async fn get_stream_info(&self) -> Result>; + /// 下载到指定路径 async fn download(&self, output_path: impl AsRef) -> Result<(), Report>; + /// 判断是否应该录制 fn should_record(&self, room_title: &str) -> bool; + /// 获取平台名称 fn get_platform_name(&self) -> &'static str; } +/// 下载插件trait +/// 定义下载插件必须实现的接口 #[async_trait] pub trait DownloadPlugin { + /// 检查URL是否匹配此插件 fn matches(&self, url: &str) -> bool; + /// 检查流状态 async fn check_status(&self, ctx: &mut Context) -> Result>; + /// 创建下载器实例 async fn create_downloader( &self, stream_info: &StreamInfo, @@ -53,7 +79,9 @@ pub trait DownloadPlugin { recorder: Recorder, ) -> Box; + /// 初始化弹幕客户端（可选） fn danmaku_init(&self) -> Option>; + /// 获取插件名称 fn name(&self) -> &str; } diff --git a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs index 2ed93fc..6f7fea9 100644 --- a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs +++ b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs @@ -9,10 +9,17 @@ use tracing::info; pub type ConnectionPool = Pool; /// 连接管理器 +/// 负责管理SQLite数据库连接池的创建和配置 pub struct ConnectionManager; impl ConnectionManager { /// 创建新的数据库连接池 + /// + /// # 参数 + /// * `path` - 数据库文件路径 + /// + /// # 返回 + /// 返回配置好的SQLite连接池 pub async fn new_pool(path: &str) -> AppResult { // 创建所有父级目录（如果不存在） if let Some(parent) = Path::new(path).parent() { @@ -30,7 +37,7 @@ impl ConnectionManager { .open(path) .change_context(AppError::Unknown)?; - // 创建连接池 + // 创建连接池，最大连接数设为2 let pool = SqlitePoolOptions::new() .max_connections(2) .connect(&db_url) @@ -39,7 +46,7 @@ impl ConnectionManager { "error while initializing the database connection pool".to_string(), ))?; - // 运行数据库迁移 + // 运行数据库迁移，确保数据库结构是最新的 info!("migrations enabled, running..."); sqlx::migrate!() .run(&pool) diff --git a/crates/biliup-cli/src/server/infrastructure/context.rs b/crates/biliup-cli/src/server/infrastructure/context.rs index 484741c..575db49 100644 --- a/crates/biliup-cli/src/server/infrastructure/context.rs +++ b/crates/biliup-cli/src/server/infrastructure/context.rs @@ -22,6 +22,9 @@ pub struct Context { impl Context { /// 创建新的上下文实例 + /// + /// # 参数 + /// * `worker` - 工作器实例的Arc引用 pub fn new(worker: Arc) -> Self { Self { worker, @@ -49,6 +52,12 @@ pub struct Worker { impl Worker { /// 创建新的工作器实例 + /// + /// # 参数 + /// * `live_streamer` - 直播主播信息 + /// * `upload_streamer` - 上传配置（可选） + /// * `config` - 全局配置的Arc引用 + /// * `client` - HTTP客户端 pub fn new( live_streamer: LiveStreamer, upload_streamer: Option, @@ -66,21 +75,28 @@ impl Worker { } /// 获取主播信息 + /// 返回当前工作器关联的直播主播信息 pub fn get_streamer(&self) -> LiveStreamer { self.live_streamer.clone() } /// 获取上传配置 + /// 返回当前工作器的上传配置（如果存在） pub fn get_upload_config(&self) -> Option { self.upload_streamer.clone() } /// 获取全局配置 + /// 返回当前的全局配置副本 pub fn get_config(&self) -> Config { self.config.read().unwrap().clone() } /// 更改工作器状态 + /// + /// # 参数 + /// * `stage` - 工作阶段（下载或上传） + /// * `status` - 新的工作状态 pub fn change_status(&self, stage: Stage, status: WorkerStatus) { match stage { Stage::Download => { @@ -94,12 +110,14 @@ impl Worker { } impl Drop for Worker { + /// 工作器销毁时的清理逻辑 fn drop(&mut self) { info!("Dropping worker {}", self.live_streamer.id); } } impl PartialEq for Worker { + /// 比较两个工作器是否相等（基于主播ID） fn eq(&self, other: &Self) -> bool { self.live_streamer.id == other.live_streamer.id } diff --git a/crates/biliup-cli/src/server/infrastructure/models.rs b/crates/biliup-cli/src/server/infrastructure/models.rs index aed5374..b2dbc8c 100644 --- a/crates/biliup-cli/src/server/infrastructure/models.rs +++ b/crates/biliup-cli/src/server/infrastructure/models.rs @@ -1,3 +1,4 @@ +/// 钩子步骤模块 pub mod hook_step; use chrono::NaiveDateTime; @@ -6,91 +7,151 @@ use ormlite::{Insert, Model}; use serde::{Deserialize, Serialize}; use serde_json::Value; +/// 上传配置模型 +/// 存储视频上传到B站时的各种配置参数 #[derive(Model, Debug, Clone, Serialize, Deserialize)] #[ormlite(table = "uploadstreamers")] pub struct UploadStreamer { + /// 主键ID pub id: i64, + /// 模板名称 pub template_name: String, + /// 视频标题 pub title: Option, + /// 分区ID pub tid: Option, + /// 版权类型（1-自制，2-转载） pub copyright: Option, + /// 转载来源 pub copyright_source: Option, + /// 封面路径 pub cover_path: Option, + /// 视频简介 pub description: Option, + /// 动态内容 pub dynamic: Option, + /// 定时发布时间 pub dtime: Option, + /// 杜比音效 pub dolby: Option, + /// Hi-Res音质 pub hires: Option, + /// 充电专属 pub charging_pay: Option, + /// 禁止转载 pub no_reprint: Option, + /// 上传者 pub uploader: Option, + /// 用户Cookie pub user_cookie: Option, + /// 标签列表（JSON格式） #[ormlite(json)] pub tags: Vec, // not null + /// 制作人员信息 pub credits: Option, + /// 开启精选评论 pub up_selection_reply: Option, + /// 关闭评论 pub up_close_reply: Option, + /// 关闭弹幕 pub up_close_danmu: Option, + /// 额外字段 pub extra_fields: Option, + /// 仅自己可见 pub is_only_self: Option, } +/// 主播信息模型 +/// 存储主播的基本信息和直播状态 #[derive(Model, Debug, Clone, Serialize, Deserialize)] #[ormlite(table = "streamerinfo")] pub struct StreamerInfo { + /// 主键ID pub id: i64, + /// 主播名称 pub name: String, + /// 直播间URL pub url: String, + /// 直播标题 pub title: String, + /// 直播时间 pub date: NaiveDateTime, + /// 直播封面路径 pub live_cover_path: String, } +/// 直播主播模型 +/// 存储直播主播的配置信息和录制参数 #[derive(Model, Debug, Clone, Serialize, Deserialize)] #[ormlite(table = "livestreamers")] pub struct LiveStreamer { + /// 主键ID #[ormlite(primary_key)] pub id: i64, + /// 直播间URL pub url: String, + /// 备注名称 pub remark: String, + /// 文件名前缀 pub filename_prefix: Option, + /// 录制时间范围 pub time_range: Option, - pub upload_streamers_id: Option, // FK，可空 + /// 关联的上传配置ID（外键，可空） + pub upload_streamers_id: Option, + /// 录制格式 pub format: Option, + /// 覆盖配置（"override"为关键字，字段名避让） #[ormlite(column = "override")] - pub override_: Option, // "override" 为关键字，字段名避让 + pub override_: Option, - // #[ormlite(json)] 数据库空与json空有区别所以这里不能用 - // pub preprocessor: Option>, - // 只能使用 sqlx::types::Json + /// 预处理器列表（JSON格式） + /// 注意：数据库空与json空有区别，所以这里不能用#[ormlite(json)] + /// 只能使用 sqlx::types::Json #[ormlite(json)] pub preprocessor: Option>, + /// 分段处理器列表（JSON格式） #[ormlite(json)] pub segment_processor: Option>, + /// 下载完成处理器列表（JSON格式） #[ormlite(json)] pub downloaded_processor: Option>, + /// 后处理器列表（JSON格式） #[ormlite(json)] pub postprocessor: Option>, + /// 可选参数 pub opt_args: Option, + /// 排除关键词 pub excluded_keywords: Option, } +/// 文件列表模型 +/// 存储录制文件的信息 #[derive(Model, Debug, Clone, Serialize, Deserialize)] #[ormlite(table = "filelist")] pub struct FileItem { + /// 主键ID pub id: i64, + /// 文件路径 pub file: String, - pub streamer_info_id: i64, // FK not null + /// 关联的主播信息ID（外键，非空） + pub streamer_info_id: i64, } +/// 配置模型 +/// 存储应用程序的配置信息 #[derive(Model, Debug, Clone, Serialize, Deserialize)] #[ormlite(table = "configuration")] pub struct Configuration { + /// 主键ID pub id: i64, + /// 配置键 pub key: String, - pub value: String, // TEXT + /// 配置值（TEXT类型） + pub value: String, } +/// 插入直播主播的数据结构 +/// 用于创建新的直播主播记录 #[derive(Insert, Debug, Serialize, Deserialize)] #[ormlite(returns = "LiveStreamer")] pub struct InsertLiveStreamer { @@ -118,6 +179,8 @@ pub struct InsertLiveStreamer { pub excluded_keywords: Option, } +/// 插入上传配置的数据结构 +/// 用于创建新的上传配置记录 #[derive(Model, Insert, Debug, Clone, Serialize, Deserialize)] #[ormlite(table = "uploadstreamers", returns = "UploadStreamer")] pub struct InsertUploadStreamer { @@ -147,9 +210,13 @@ pub struct InsertUploadStreamer { pub is_only_self: Option, } +/// 插入配置的数据结构 +/// 用于创建新的配置记录 #[derive(Insert, Debug, Clone, Serialize, Deserialize)] #[ormlite(returns = "Configuration")] pub struct InsertConfiguration { + /// 配置键 pub key: String, - pub value: String, // TEXT + /// 配置值（TEXT类型） + pub value: String, } diff --git a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs index 9bda71a..840f5f2 100644 --- a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs +++ b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs @@ -21,6 +21,12 @@ pub enum HookStep { impl HookStep { /// 执行钩子步骤操作 + /// + /// # 参数 + /// * `video_paths` - 视频文件路径列表 + /// + /// # 返回 + /// 执行成功返回Ok(())，失败返回错误信息 pub async fn execute(&self, video_paths: &[&Path]) -> Result<(), Box> { match self { HookStep::Run { run } => { @@ -36,6 +42,7 @@ impl HookStep { self.remove_file(video_paths).await?; } HookStep::Remove(cmd) => { + // 未知命令，返回错误 return Err(format!("Unknown command: {}", cmd).into()); } } @@ -43,6 +50,10 @@ impl HookStep { } /// 执行自定义命令，将视频路径作为标准输入传入 + /// + /// # 参数 + /// * `cmd` - 要执行的命令字符串 + /// * `video_paths` - 视频文件路径列表 async fn execute_command( &self, cmd: &str, @@ -56,7 +67,7 @@ impl HookStep { return Err("Empty command".into()); } - // 启动子进程 + // 启动子进程，配置标准输入管道 let mut process = Command::new(parts[0]) .args(&parts[1..]) .stdin(std::process::Stdio::piped()) @@ -72,7 +83,7 @@ impl HookStep { stdin.write_all(paths_str.as_bytes()).await?; } - // 等待进程完成 + // 等待进程完成并检查退出状态 let status = process.wait().await?; if !status.success() { return Err(format!("Command failed with status: {}", status).into()); @@ -82,6 +93,10 @@ impl HookStep { } /// 移动文件到指定目录 + /// + /// # 参数 + /// * `video_paths` - 视频文件路径列表 + /// * `target_dir` - 目标目录路径 async fn move_file( &self, video_paths: &[&Path], @@ -94,7 +109,7 @@ impl HookStep { fs::create_dir_all(target_path).await?; } - // 移动每个视频文件 + // 移动每个视频文件到目标目录 for video_path in video_paths { let file_name = video_path.file_name().ok_or("Invalid file name")?; let destination = target_path.join(file_name); @@ -110,7 +125,11 @@ impl HookStep { } /// 删除指定的视频文件 + /// + /// # 参数 + /// * `video_paths` - 要删除的视频文件路径列表 async fn remove_file(&self, video_paths: &[&Path]) -> Result<(), Box> { + // 逐个删除视频文件 for video_path in video_paths { println!("Removing: {}", video_path.display()); fs::remove_file(video_path).await?; @@ -120,6 +139,12 @@ impl HookStep { } /// 处理所有后处理器步骤 +/// 处理视频文件的主函数 +/// 按顺序执行所有处理器步骤 +/// +/// # 参数 +/// * `video_path` - 视频文件路径列表 +/// * `processors` - 处理器步骤列表 pub async fn process_video( video_path: &[&Path], processors: &[HookStep], diff --git a/crates/biliup-cli/src/server/infrastructure/repositories.rs b/crates/biliup-cli/src/server/infrastructure/repositories.rs index a361841..e40b5dc 100644 --- a/crates/biliup-cli/src/server/infrastructure/repositories.rs +++ b/crates/biliup-cli/src/server/infrastructure/repositories.rs @@ -7,6 +7,11 @@ use crate::server::infrastructure::models::{ use error_stack::ResultExt; use ormlite::{Insert, Model}; +/// 根据ID获取直播主播信息 +/// +/// # 参数 +/// * `pool` - 数据库连接池 +/// * `id` - 主播ID pub async fn get_streamer(pool: &ConnectionPool, id: i64) -> AppResult { LiveStreamer::select() .where_("id = ?") @@ -16,6 +21,11 @@ pub async fn get_streamer(pool: &ConnectionPool, id: i64) -> AppResult AppResult { let streamer = get_streamer(pool, id).await?; streamer @@ -41,6 +59,10 @@ pub async fn del_streamer(pool: &ConnectionPool, id: i64) -> AppResult AppResult> { LiveStreamer::select() .fetch_all(pool) @@ -48,6 +70,13 @@ pub async fn get_all_streamer(pool: &ConnectionPool) -> AppResult AppResult { let configuration = Configuration::select() .where_("key = 'config'") @@ -55,15 +84,22 @@ pub async fn get_config(pool: &ConnectionPool) -> AppResult { .await .change_context(AppError::Unknown)?; if let Some(configuration) = configuration { + // 从数据库中解析配置JSON let json: Config = serde_json::from_str(&configuration.value).change_context(AppError::Unknown)?; Ok(json) } else { + // 如果数据库中没有配置，返回默认配置 let config = Config::builder().streamers(Default::default()).build(); Ok(config) } } +/// 插入或更新全局配置到数据库 +/// +/// # 参数 +/// * `pool` - 数据库连接池 +/// * `config` - 要保存的配置 pub async fn insert_config(pool: &ConnectionPool, config: &Config) -> AppResult { let configuration = InsertConfiguration { key: "config".to_string(), @@ -75,6 +111,10 @@ pub async fn insert_config(pool: &ConnectionPool, config: &Config) -> AppResult< Ok(configuration) } +/// 获取所有上传配置 +/// +/// # 参数 +/// * `pool` - 数据库连接池 pub async fn get_all_uploader(pool: &ConnectionPool) -> AppResult> { UploadStreamer::select() .fetch_all(pool) diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs index 80dac67..9a0ba7f 100644 --- a/crates/biliup-cli/src/server/infrastructure/service_register.rs +++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs @@ -11,18 +11,33 @@ use error_stack::bail; use std::sync::{Arc, RwLock}; use tracing::info; +/// 服务注册器 +/// 负责管理应用程序中的各种服务实例，包括数据库连接池、工作器、下载管理器等 #[derive(FromRef, Clone)] pub struct ServiceRegister { + /// 数据库连接池 pub pool: ConnectionPool, + /// 工作器列表 pub workers: Arc>>>, + /// 下载管理器列表 pub managers: Arc>, + /// 全局配置 pub config: Arc>, + /// Actor处理器 pub actor_handle: Arc, + /// HTTP客户端 pub client: StatelessClient, } -/// A simple service container responsible for managing the various services our API endpoints will pull from through axum extensions. +/// 简单的服务容器，负责管理API端点通过axum扩展获取的各种服务 impl ServiceRegister { + /// 创建新的服务注册器实例 + /// + /// # 参数 + /// * `pool` - 数据库连接池 + /// * `config` - 全局配置 + /// * `actor_handle` - Actor处理器 + /// * `download_manager` - 下载管理器列表 pub fn new( pool: ConnectionPool, config: Arc>, @@ -30,6 +45,7 @@ impl ServiceRegister { download_manager: Vec, ) -> Self { info!("initializing utility services..."); + // 创建默认的HTTP客户端 let client = StatelessClient::default(); info!(config=?config); @@ -48,6 +64,13 @@ impl ServiceRegister { } } + /// 根据URL获取匹配的下载管理器 + /// + /// # 参数 + /// * `url` - 直播流URL + /// + /// # 返回 + /// 返回匹配的下载管理器引用，如果没有匹配的则返回None pub fn get_manager(&self, url: &str) -> Option<&DownloadManager> { self.managers .iter() @@ -55,25 +78,38 @@ impl ServiceRegister { .map(|v| v as _) } + /// 添加新的直播间到监控列表 + /// + /// # 参数 + /// * `monitor` - 监控器实例 + /// * `live_streamer` - 直播主播信息 + /// * `upload_streamer` - 上传配置（可选） pub async fn add_room( &self, monitor: Arc, live_streamer: LiveStreamer, upload_streamer: Option, ) -> AppResult> { + // 创建新的工作器实例 let worker = Arc::new(Worker::new( live_streamer, upload_streamer, self.config.clone(), self.client.clone(), )); + // 将工作器添加到监控器和工作器列表中 monitor.rooms_handle.add(worker.clone()).await; self.workers.write().unwrap().push(worker.clone()); info!("add {worker:?} success"); Ok(Some(())) } + /// 删除指定ID的直播间 + /// + /// # 参数 + /// * `id` - 要删除的直播间ID pub async fn del_room(&self, id: i64) -> AppResult<()> { + // 在工作器列表中查找要删除的工作器 let Some(i) = self .workers .read() @@ -84,15 +120,19 @@ impl ServiceRegister { return Err(error_stack::Report::new(AppError::Unknown)); }; + // 从工作器列表中移除 let removed = self.workers.write().unwrap().swap_remove(i); let url = &removed.live_streamer.url; + // 获取对应的下载管理器 let Some(manager) = self.get_manager(url) else { info!("not found url: {url}"); bail!(AppError::Unknown) }; + // 从监控器中删除房间 let monitor = manager.ensure_monitor(); let len = monitor.rooms_handle.del(id).await; info!("id: {id} removed, remained len {len}"); + // 如果没有剩余房间，清理监控器 if len == 0 { *manager.monitor.lock().unwrap() = None; } diff --git a/crates/biliup-cli/src/server/infrastructure/users.rs b/crates/biliup-cli/src/server/infrastructure/users.rs index 7e740a6..9619803 100644 --- a/crates/biliup-cli/src/server/infrastructure/users.rs +++ b/crates/biliup-cli/src/server/infrastructure/users.rs @@ -7,15 +7,19 @@ use serde::{Deserialize, Serialize}; use sqlx::FromRow; use tokio::task; +/// 用户数据结构 +/// 存储用户的基本信息，包括ID、用户名和密码哈希 #[derive(Clone, Serialize, Deserialize, FromRow)] pub struct User { + /// 用户ID id: i64, + /// 用户名 pub key: String, + /// 密码哈希值 value: String, } -// Here we've implemented `Debug` manually to avoid accidentally logging the -// password hash. +// 手动实现Debug trait以避免意外记录密码哈希 impl std::fmt::Debug for User { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { f.debug_struct("User") @@ -34,36 +38,45 @@ impl AuthUser for User { } fn session_auth_hash(&self) -> &[u8] { - self.value.as_bytes() // We use the password hash as the auth - // hash--what this means - // is when the user changes their password the - // auth session becomes invalid. + // 使用密码哈希作为认证哈希 + // 这意味着当用户更改密码时，认证会话将失效 + self.value.as_bytes() } } -// This allows us to extract the authentication fields from forms. We use this -// to authenticate requests with the backend. +// 认证凭据结构，用于从表单中提取认证字段 +// 用于与后端进行请求认证 #[derive(Debug, Clone, Deserialize)] pub struct Credentials { + /// 用户名 pub username: String, + /// 密码 pub password: String, + /// 登录后跳转的URL（可选） pub next: Option, } +/// 认证后端 +/// 负责处理用户认证相关的数据库操作 #[derive(Debug, Clone)] pub struct Backend { + /// 数据库连接池 db: ConnectionPool, } impl Backend { + /// 创建新的认证后端实例 pub fn new(db: ConnectionPool) -> Self { Self { db } } + /// 检查是否存在用户 + /// + /// # 返回 + /// 如果存在用户返回true，否则返回false pub async fn exists(&self) -> AppResult { - // check if a user corresponding to the given creds already exists... + // 检查是否已存在对应的用户 let user: Option = sqlx::query_as("select * from configuration where key = ? ") - // .bind(creds.username) .bind("biliup") .fetch_optional(&self.db) .await @@ -71,8 +84,15 @@ impl Backend { Ok(user.is_some()) } + /// 创建新用户 + /// + /// # 参数 + /// * `creds` - 用户凭据 + /// + /// # 返回 + /// 返回创建的用户信息 pub async fn create_user(&mut self, creds: Credentials) -> AppResult { - // create the new user account... + // 创建新用户账户 // 验证输入 if creds.username.is_empty() || creds.password.is_empty() { bail!(AppError::Custom("用户名和密码不能为空".into())) @@ -103,11 +123,14 @@ impl Backend { } } +/// 认证相关的错误类型 #[derive(Debug, thiserror::Error)] pub enum Error { + /// 数据库错误 #[error(transparent)] Sqlx(#[from] sqlx::Error), + /// 任务连接错误 #[error(transparent)] TaskJoin(#[from] task::JoinError), } @@ -117,26 +140,34 @@ impl AuthnBackend for Backend { type Credentials = Credentials; type Error = Error; + /// 认证用户 + /// + /// # 参数 + /// * `creds` - 用户凭据 + /// + /// # 返回 + /// 如果认证成功返回用户信息，否则返回None async fn authenticate( &self, creds: Self::Credentials, ) -> Result, Self::Error> { let user: Option = sqlx::query_as("select * from configuration where key = ? ") - // .bind(creds.username) .bind("biliup") .fetch_optional(&self.db) .await?; - // Verifying the password is blocking and potentially slow, so we'll do so via - // `spawn_blocking`. + // 密码验证是阻塞且可能较慢的操作，所以通过spawn_blocking执行 task::spawn_blocking(|| { - // We're using password-based authentication--this works by comparing our form - // input with an argon2 password hash. + // 使用基于密码的认证 - 通过比较表单输入与argon2密码哈希来工作 Ok(user.filter(|user| verify_password(creds.password, &user.value).is_ok())) }) .await? } + /// 根据用户ID获取用户信息 + /// + /// # 参数 + /// * `user_id` - 用户ID async fn get_user(&self, user_id: &UserId) -> Result, Self::Error> { let user = sqlx::query_as("select * from configuration where id = ?") .bind(user_id) @@ -147,7 +178,6 @@ impl AuthnBackend for Backend { } } -// We use a type alias for convenience. -// -// Note that we've supplied our concrete backend here. +// 为了方便使用的类型别名 +// 注意这里我们提供了具体的后端实现 pub type AuthSession = axum_login::AuthSession;