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;