diff --git a/crates/biliup-cli/src/server/api/bilibili_endpoints.rs b/crates/biliup-cli/src/server/api/bilibili_endpoints.rs index db5bbfe..e93d08d 100644 --- a/crates/biliup-cli/src/server/api/bilibili_endpoints.rs +++ b/crates/biliup-cli/src/server/api/bilibili_endpoints.rs @@ -12,16 +12,20 @@ use ormlite::Model; use std::collections::HashMap; use tracing::info; +/// B站投稿预处理端点 pub async fn archive_pre_endpoint( - Query(params): Query>, + Query(_params): Query>, State(pool): State, ) -> Result, Response> { + // 获取所有B站Cookie配置 let configurations = Configuration::select() .where_("key = 'bilibili-cookies'") .fetch_all(&pool) .await .change_context(AppError::Unknown) .map_err(report_to_response)?; + + // 尝试使用每个Cookie进行登录 for cookies in configurations { if let Ok(bili) = login_by_cookies(cookies.value, None).await { return Ok(Json( @@ -32,18 +36,24 @@ pub async fn archive_pre_endpoint( )); } } + + // 没有可用的Cookie Err(report_to_response(Report::from(AppError::Custom( "无可用 cookie 文件".to_string(), )))) } +/// 获取B站用户信息端点 pub async fn get_myinfo_endpoint( Query(params): Query>, ) -> Result, Response> { + // 使用指定用户的Cookie登录 let bili = login_by_cookies(¶ms["user"], None) .await .change_context(AppError::Custom(params["user"].to_string())) .map_err(report_to_response)?; + + // 获取用户信息 Ok(Json( bili.my_info() .await @@ -52,12 +62,14 @@ pub async fn get_myinfo_endpoint( )) } +/// 代理请求端点 pub async fn get_proxy_endpoint( State(client): State, Query(params): Query>, ) -> Result { info!(params = ¶ms["url"]); - // let bili = login_by_cookies(¶ms["user"]).await?; + + // 代理HTTP请求 client .client .get(¶ms["url"]) diff --git a/crates/biliup-cli/src/server/api/mod.rs b/crates/biliup-cli/src/server/api/mod.rs index 15df7f3..218dea0 100644 --- a/crates/biliup-cli/src/server/api/mod.rs +++ b/crates/biliup-cli/src/server/api/mod.rs @@ -1,4 +1,8 @@ +/// 认证相关API pub mod auth; +/// B站API端点 pub mod bilibili_endpoints; +/// 通用API端点 pub mod endpoints; +/// 单页应用静态文件处理 pub mod spa; diff --git a/crates/biliup-cli/src/server/api/spa.rs b/crates/biliup-cli/src/server/api/spa.rs index 16f1e39..1d0e3b5 100644 --- a/crates/biliup-cli/src/server/api/spa.rs +++ b/crates/biliup-cli/src/server/api/spa.rs @@ -2,41 +2,44 @@ use axum::http::{StatusCode, Uri, header}; use axum::response::{Html, IntoResponse, Response}; use rust_embed::Embed; +/// 默认首页文件名 static INDEX_HTML: &str = "index.html"; +/// 嵌入的静态资源 #[derive(Embed)] #[folder = "../../out/"] struct Assets; +/// 静态文件处理器，用于服务单页应用 pub async fn static_handler(uri: Uri) -> impl IntoResponse { let path = uri.path().trim_start_matches('/'); + // 根路径或直接访问index.html时返回首页 if path.is_empty() || path == INDEX_HTML { return index_html().await; } + // 尝试查找对应的HTML文件 let guess_html = path.to_owned() + ".html"; if let Some(html) = Assets::get(&guess_html) { return Html(html.data).into_response(); } + // 查找静态资源文件 match Assets::get(path) { Some(content) => { + // 根据文件扩展名推断MIME类型 let mime = mime_guess::from_path(path).first_or_octet_stream(); - ([(header::CONTENT_TYPE, mime.as_ref())], content.data).into_response() } None => { - // if path.contains('.') { - // return not_found().await; - // } - // - // index_html().await + // 文件不存在，返回404 not_found().await } } } +/// 返回首页HTML内容 async fn index_html() -> Response { match Assets::get(INDEX_HTML) { Some(content) => Html(content.data).into_response(), @@ -44,6 +47,7 @@ async fn index_html() -> Response { } } +/// 返回404错误响应 async fn not_found() -> Response { (StatusCode::NOT_FOUND, "404").into_response() } diff --git a/crates/biliup-cli/src/server/app.rs b/crates/biliup-cli/src/server/app.rs index 13cf0e5..08e4e46 100644 --- a/crates/biliup-cli/src/server/app.rs +++ b/crates/biliup-cli/src/server/app.rs @@ -19,70 +19,65 @@ use tower_sessions::{ExpiredDeletion, Expiry, SessionManagerLayer}; use tower_sessions_sqlx_store::SqliteStore; use tracing::info; +/// 应用程序控制器，负责启动和管理Web服务器 pub struct ApplicationController; impl ApplicationController { + /// 启动Web服务器 pub async fn serve(addr: &SocketAddr, service_register: ServiceRegister) -> AppResult<()> { - // Session layer. - // - // This uses `tower-sessions` to establish a layer that will provide the session - // as a request extension. + // 会话层配置 + // 使用 tower-sessions 建立会话层，将会话作为请求扩展提供 let session_store = SqliteStore::new(service_register.pool.clone()); session_store .migrate() .await .change_context(AppError::Unknown)?; + // 启动定期清理过期会话的任务 let deletion_task = tokio::task::spawn( session_store .clone() .continuously_delete_expired(tokio::time::Duration::from_secs(60)), ); - // Generate a cryptographic key to sign the session cookie. + // 生成用于签名会话cookie的加密密钥 let key = Key::generate(); + // 配置会话管理层 let session_layer = SessionManagerLayer::new(session_store) .with_secure(false) .with_expiry(Expiry::OnInactivity(Duration::days(1))) .with_signed(key); - // Auth service. - // - // This combines the session layer with our backend to establish the auth - // service which will provide the auth session as a request extension. + // 认证服务配置 + // 将会话层与后端结合，建立认证服务，将认证会话作为请求扩展提供 let backend = Backend::new(service_register.pool.clone()); let auth_layer = AuthManagerLayerBuilder::new(backend, session_layer).build(); - // let app = protected::router() - - // let vec = service_register.streamers_service.get_streamers().await?; - // build our application with a route - let enable_login_guard = true; + // 构建应用程序路由 + let enable_login_guard = true; // 是否启用登录保护 let mut app = server::router::router(service_register); if enable_login_guard { app = app - .route_layer(login_required!(Backend)) - .merge(auth::router()); + .route_layer(login_required!(Backend)) // 添加登录验证中间件 + .merge(auth::router()); // 合并认证路由 } app = app - .layer(auth_layer) + .layer(auth_layer) // 添加认证层 .layer( - // see https://docs.rs/tower-http/latest/tower_http/cors/index.html - // for more details - // - // pay attention that for some request types like posting content-type: application/json - // it is required to add ".allow_headers([http::header::CONTENT_TYPE])" - // or see this issue https://github.com/tokio-rs/axum/issues/849 + // CORS配置 - 跨域资源共享 + // 详见 https://docs.rs/tower-http/latest/tower_http/cors/index.html + // 注意：对于某些请求类型（如POST application/json）， + // 需要添加 ".allow_headers([http::header::CONTENT_TYPE])" + // 参考：https://github.com/tokio-rs/axum/issues/849 CorsLayer::new() .allow_headers([http::header::CONTENT_TYPE]) .allow_origin("http://localhost:3000".parse::().unwrap()) .allow_methods(AllowMethods::any()), ) - // .layer(Extension(client.clone())) - .fallback(static_handler); - // run our app with hyper - // `axum::Server` is a re-export of `hyper::Server` + .fallback(static_handler); // 静态文件处理回退 + + // 启动HTTP服务器 info!("routes initialized, listening on {}", addr); let listener = tokio::net::TcpListener::bind(addr) .await @@ -94,6 +89,7 @@ impl ApplicationController { .change_context(AppError::Unknown) .attach("error while starting API server")?; + // 等待会话清理任务完成 deletion_task .await .change_context(AppError::Unknown)? @@ -103,13 +99,16 @@ impl ApplicationController { } } +/// 优雅关闭信号处理 async fn shutdown_signal(deletion_task_abort_handle: AbortHandle) { + // 监听Ctrl+C信号 let ctrl_c = async { signal::ctrl_c() .await .expect("failed to install Ctrl+C handler"); }; + // Unix系统下监听SIGTERM信号 #[cfg(unix)] let terminate = async { signal::unix::signal(signal::unix::SignalKind::terminate()) @@ -118,9 +117,11 @@ async fn shutdown_signal(deletion_task_abort_handle: AbortHandle) { .await; }; + // 非Unix系统下使用pending future #[cfg(not(unix))] let terminate = std::future::pending::<()>(); + // 等待任一信号触发，然后中止清理任务 tokio::select! { _ = ctrl_c => { deletion_task_abort_handle.abort() }, _ = terminate => { deletion_task_abort_handle.abort() }, diff --git a/crates/biliup-cli/src/server/common/mod.rs b/crates/biliup-cli/src/server/common/mod.rs index 812d1ed..a946577 100644 --- a/crates/biliup-cli/src/server/common/mod.rs +++ b/crates/biliup-cli/src/server/common/mod.rs @@ -1 +1,2 @@ +/// 通用工具函数 pub mod util; diff --git a/crates/biliup-cli/src/server/common/util.rs b/crates/biliup-cli/src/server/common/util.rs index f5c4573..3ecfbae 100644 --- a/crates/biliup-cli/src/server/common/util.rs +++ b/crates/biliup-cli/src/server/common/util.rs @@ -1,16 +1,19 @@ -// main.rs use chrono::{Duration, Local}; use serde::{Deserialize, Serialize}; use std::path::{Path, PathBuf}; use url::Url; +/// 录制器配置结构体 #[derive(Debug, Clone, Deserialize, Serialize)] pub struct Recorder { + /// 文件名前缀模板 pub filename_prefix: Option, + /// 主播名称 pub fname: String, + /// 直播间标题 pub room_title: String, - /// 录制后保存文件格式 (mp4, ts, mkv, flv) - pub suffix: String, // 不含点，如 "mp4"、"flv" + /// 录制后保存文件格式 (mp4, ts, mkv, flv)，不含点 + pub suffix: String, } impl Recorder { diff --git a/crates/biliup-cli/src/server/config.rs b/crates/biliup-cli/src/server/config.rs index 57ad36a..860f767 100644 --- a/crates/biliup-cli/src/server/config.rs +++ b/crates/biliup-cli/src/server/config.rs @@ -5,172 +5,226 @@ use error_stack::bail; use serde::{Deserialize, Serialize}; use std::{collections::HashMap, path::Path, path::PathBuf}; +/// 全局配置结构体 #[derive(bon::Builder, Debug, Clone, Serialize, Deserialize)] pub struct Config { // ===== 全局录播与上传设置 ===== + /// 下载器类型：streamlink | ffmpeg | stream-gears | 自定义 #[serde(default)] - pub downloader: Option, // streamlink | ffmpeg | stream-gears | 自定义 + pub downloader: Option, + /// 文件大小限制（字节） #[builder(default = default_file_size())] #[serde(default = "default_file_size")] - pub file_size: u64, // Byte + pub file_size: u64, - // 形如 "00:00:00"；保留为字符串以保持直观（你也可以替换为Duration并写自定义反序列化） + /// 分段时间，格式如 "00:00:00"，保留为字符串以保持直观 #[serde(default = "default_segment_time")] pub segment_time: Option, + /// 过滤阈值（MB） #[builder(default = default_filtering_threshold())] #[serde(default = "default_filtering_threshold")] - pub filtering_threshold: u64, // MB + pub filtering_threshold: u64, + /// 文件名前缀 #[serde(default)] pub filename_prefix: Option, + /// 分段处理器是否并行执行 #[serde(default)] pub segment_processor_parallel: Option, + /// 上传器类型：Noop | bili_web | biliup-rs | 其他 #[serde(default)] - pub uploader: Option, // Noop | bili_web | biliup-rs | 其他 + pub uploader: Option, + /// 提交API类型：web | client #[serde(default)] - pub submit_api: Option, // web | client + pub submit_api: Option, + /// 上传线路：AUTO | alia | bda2 | bldsa | qn | tx | txa #[builder(default = default_lines())] #[serde(default = "default_lines")] - pub lines: String, // AUTO | alia | bda2 | bldsa | qn | tx | txa + pub lines: String, + /// 上传线程数 #[builder(default = default_threads())] #[serde(default = "default_threads")] pub threads: u32, + /// 延迟时间（秒） #[builder(default = default_delay())] #[serde(default = "default_delay")] pub delay: u64, + /// 事件循环间隔（秒） #[builder(default = default_event_loop_interval())] #[serde(default = "default_event_loop_interval")] pub event_loop_interval: u64, + /// 检查器休眠时间（秒） #[builder(default = default_checker_sleep())] #[serde(default = "default_checker_sleep")] pub checker_sleep: u64, + /// 连接池1大小 #[builder(default = default_pool1_size())] #[serde(default = "default_pool1_size")] pub pool1_size: u32, + /// 连接池2大小 #[builder(default = default_pool2_size())] #[serde(default = "default_pool2_size")] pub pool2_size: u32, - // ===== 各平台录播设置（顶层分散字段） ===== + // ===== 各平台录播设置 ===== + /// 是否使用直播封面 #[serde(default)] pub use_live_cover: Option, - // 斗鱼 + // 斗鱼平台设置 + /// 斗鱼CDN节点 #[serde(default)] pub douyu_cdn: Option, + /// 斗鱼弹幕录制 #[serde(default)] pub douyu_danmaku: Option, + /// 斗鱼码率 #[serde(default)] pub douyu_rate: Option, - // 虎牙 + // 虎牙平台设置 + /// 虎牙CDN节点 #[serde(default)] pub huya_cdn: Option, + /// 虎牙CDN回退 #[serde(default)] pub huya_cdn_fallback: Option, + /// 虎牙弹幕录制 #[serde(default)] pub huya_danmaku: Option, + /// 虎牙最大比率 #[serde(default)] pub huya_max_ratio: Option, - // 抖音 + // 抖音平台设置 + /// 抖音弹幕录制 #[serde(default)] pub douyin_danmaku: Option, + /// 抖音画质 #[serde(default)] pub douyin_quality: Option, - // 哔哩哔哩 + // 哔哩哔哩平台设置 + /// B站弹幕录制 #[serde(default)] pub bilibili_danmaku: Option, + /// B站弹幕详细信息 #[serde(default)] pub bilibili_danmaku_detail: Option, + /// B站弹幕原始数据 #[serde(default)] pub bilibili_danmaku_raw: Option, + /// B站协议类型：stream | hls_ts | hls_fmp4 #[serde(default)] - pub bili_protocol: Option, // stream | hls_ts | hls_fmp4 + pub bili_protocol: Option, + /// B站CDN节点列表 #[serde(default)] pub bili_cdn: Option>, + /// B站强制原画 #[serde(default)] pub bili_force_source: Option, + /// B站直播API #[serde(default)] pub bili_liveapi: Option, + /// B站回退API #[serde(default)] pub bili_fallback_api: Option, + /// B站CDN回退 #[serde(default)] pub bili_cdn_fallback: Option, + /// B站cn01节点替换 #[serde(default)] pub bili_replace_cn01: Option>, + /// B站画质编号 #[serde(default)] pub bili_qn: Option, - // YouTube + // YouTube平台设置 + /// YouTube首选视频编码 #[serde(default)] pub youtube_prefer_vcodec: Option, + /// YouTube首选音频编码 #[serde(default)] pub youtube_prefer_acodec: Option, + /// YouTube最大分辨率 #[serde(default)] pub youtube_max_resolution: Option, + /// YouTube最大视频大小 #[serde(default)] pub youtube_max_videosize: Option, + /// YouTube开始日期 #[serde(default)] pub youtube_after_date: Option, + /// YouTube结束日期 #[serde(default)] pub youtube_before_date: Option, + /// YouTube启用直播下载 #[serde(default)] pub youtube_enable_download_live: Option, + /// YouTube启用回放下载 #[serde(default)] pub youtube_enable_download_playback: Option, - // Twitch + // Twitch平台设置 + /// Twitch弹幕录制 #[serde(default)] pub twitch_danmaku: Option, + /// Twitch禁用广告 #[serde(default)] pub twitch_disable_ads: Option, - // TwitCasting + // TwitCasting平台设置 + /// TwitCasting弹幕录制 #[serde(default)] pub twitcasting_danmaku: Option, + /// TwitCasting密码 #[serde(default)] pub twitcasting_password: Option, - // 录制主播设置 + /// 录制主播配置映射 #[serde(default)] pub streamers: HashMap, - // 用户 cookie + /// 用户Cookie配置 #[serde(default)] pub user: Option, } +/// 主播配置结构体 #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct StreamerConfig { + /// 直播间URL列表 pub url: Vec, + /// 视频标题 #[serde(default)] pub title: Option, + /// 分区ID #[serde(default)] pub tid: Option, + /// 版权类型 #[serde(default)] pub copyright: Option, + /// 封面路径 #[serde(default)] pub cover_path: Option, - // 用字符串保留缩进和多行 + /// 视频描述（保留缩进和多行格式） #[serde(default)] pub description: Option, @@ -242,83 +296,116 @@ pub struct StreamerConfig { pub override_cfg: Option>, } +/// 制作人员信息 #[derive(Debug, Clone, Serialize, Deserialize)] pub struct Credit { + /// 用户名 pub username: String, + /// 用户ID pub uid: String, } +/// 用户配置结构体 #[derive(bon::Builder, Debug, Clone, Serialize, Deserialize, Default)] pub struct UserConfig { - // B 站 + // B站配置 + /// B站Cookie字符串 #[serde(default)] pub bili_cookie: Option, + /// B站Cookie文件路径 #[serde(default)] pub bili_cookie_file: Option, - // 抖音 + // 抖音配置 + /// 抖音Cookie #[serde(default)] pub douyin_cookie: Option, - // Twitch + // Twitch配置 + /// Twitch Cookie #[serde(default)] pub twitch_cookie: Option, - // YouTube + // YouTube配置 + /// YouTube Cookie文件路径 #[serde(default)] pub youtube_cookie: Option, - // NICO（使用了包含破折号的 key，使用 rename 保持一致） + // Niconico配置（使用rename保持与配置文件一致） + /// Niconico邮箱 #[serde(rename = "niconico-email", default)] pub niconico_email: Option, + /// Niconico密码 #[serde(rename = "niconico-password", default)] pub niconico_password: Option, + /// Niconico用户会话 #[serde(rename = "niconico-user-session", default)] pub niconico_user_session: Option, + /// Niconico清除凭据 #[serde(rename = "niconico-purge-credentials", default)] pub niconico_purge_credentials: Option, - // AfreecaTV + // AfreecaTV配置 + /// AfreecaTV用户名 #[serde(default)] pub afreecatv_username: Option, + /// AfreecaTV密码 #[serde(default)] pub afreecatv_password: Option, } +/// 默认文件大小：2.5GB fn default_file_size() -> u64 { 2_621_440_000 } +/// 默认分段时间：2小时 pub fn default_segment_time() -> Option { Some("02:00:00".to_string()) } +/// 默认过滤阈值：20MB fn default_filtering_threshold() -> u64 { 20 } + +/// 默认上传线路：自动选择 fn default_lines() -> String { "AUTO".to_string() } + +/// 默认线程数：3 fn default_threads() -> u32 { 3 } + +/// 默认延迟：300秒 fn default_delay() -> u64 { 300 } + +/// 默认事件循环间隔：30秒 fn default_event_loop_interval() -> u64 { 30 } + +/// 默认检查器休眠时间：10秒 fn default_checker_sleep() -> u64 { 10 } + +/// 默认连接池1大小：5 fn default_pool1_size() -> u32 { 5 } + +/// 默认连接池2大小：3 fn default_pool2_size() -> u32 { 3 } impl Config { + /// 从指定路径加载配置文件，如果不存在则创建默认配置 pub fn load_or_create>(path: P) -> AppResult { bail!(AppError::Custom(format!( "load_or_create: {:?}", diff --git a/crates/biliup-cli/src/server/core/mod.rs b/crates/biliup-cli/src/server/core/mod.rs index ac71670..0ff2280 100644 --- a/crates/biliup-cli/src/server/core/mod.rs +++ b/crates/biliup-cli/src/server/core/mod.rs @@ -1,4 +1,8 @@ +/// 下载管理器 pub mod download_manager; +/// 下载器接口和实现 pub mod downloader; +/// 监控器 pub mod monitor; +/// 插件系统 pub mod plugin; diff --git a/crates/biliup-cli/src/server/errors.rs b/crates/biliup-cli/src/server/errors.rs index 3849e61..fff2ca7 100644 --- a/crates/biliup-cli/src/server/errors.rs +++ b/crates/biliup-cli/src/server/errors.rs @@ -7,15 +7,19 @@ use std::collections::HashMap; use thiserror::Error; use tracing::log::error; +/// 应用程序错误类型枚举 #[derive(Error, Debug)] pub enum AppError { + /// 未知错误 #[error("Unknown Error")] Unknown, + /// 自定义错误消息 #[error("{0}")] Custom(String), } +/// 将错误报告转换为HTTP响应 pub fn report_to_response(report: impl Into>) -> Response { let report = report.into(); let (status, error_message) = match report.downcast_ref::() { @@ -24,33 +28,35 @@ pub fn report_to_response(report: impl Into>) -> Response { _ => (StatusCode::INTERNAL_SERVER_ERROR, report.to_string()), }; tracing::error!(error = ?report); - // I'm not a fan of the error specification, so for the sake of consistency, - // serialize singular errors as a map of vectors similar to the 422 validation responses + // 为了保持一致性，将单个错误序列化为类似422验证响应的映射向量格式 let body = Json(ApiError::new(error_message)); (status, body).into_response() } +/// API错误响应结构体 #[derive(Debug, Deserialize, Serialize)] pub struct ApiError { + /// 错误详情列表 pub errors: Vec>, + /// 错误消息 pub message: String, } impl ApiError { + /// 创建新的API错误 pub fn new(message: String) -> Self { let errors: Vec> = Vec::new(); + // 可以根据需要添加具体的错误字段信息 // errors.push(HashMap::from([ // (String::from("resource"), "Issue".to_string()), // (String::from("field"), "title".to_string()), // (String::from("code"), "missing_field".to_string()), - // (String::from("code"), "unprocessable".to_string()), - // (String::from("code"), "already_exists".to_string()), - // (String::from("code"), "invalid".to_string()), - // (String::from("code"), "missing".to_string()), + // ... // ])); Self { errors, message } } } +/// 应用程序结果类型别名 pub type AppResult = Result>; diff --git a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs index 3f393a1..2ed93fc 100644 --- a/crates/biliup-cli/src/server/infrastructure/connection_pool.rs +++ b/crates/biliup-cli/src/server/infrastructure/connection_pool.rs @@ -5,11 +5,14 @@ use sqlx::{Pool, Sqlite}; use std::path::Path; use tracing::info; +/// SQLite连接池类型别名 pub type ConnectionPool = Pool; +/// 连接管理器 pub struct ConnectionManager; impl ConnectionManager { + /// 创建新的数据库连接池 pub async fn new_pool(path: &str) -> AppResult { // 创建所有父级目录（如果不存在） if let Some(parent) = Path::new(path).parent() { @@ -17,13 +20,17 @@ impl ConnectionManager { .change_context(AppError::Unknown) .attach_with(|| path.to_string())?; // 创建 data/ 目录 } + let db_url = format!("sqlite://{path}"); - // Start by making a database connection. + + // 创建数据库文件（如果不存在） std::fs::OpenOptions::new() .write(true) .create(true) .open(path) .change_context(AppError::Unknown)?; + + // 创建连接池 let pool = SqlitePoolOptions::new() .max_connections(2) .connect(&db_url) @@ -32,7 +39,7 @@ impl ConnectionManager { "error while initializing the database connection pool".to_string(), ))?; - // Query builder syntax closely follows SQL syntax, translated into chained function calls. + // 运行数据库迁移 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 eb3f515..484741c 100644 --- a/crates/biliup-cli/src/server/infrastructure/context.rs +++ b/crates/biliup-cli/src/server/infrastructure/context.rs @@ -11,36 +11,44 @@ use serde::{Deserialize, Serialize}; use std::sync::{Arc, RwLock}; use tracing::info; +/// 应用程序上下文，包含工作器和扩展信息 #[derive(Debug, Clone)] pub struct Context { + /// 工作器实例 pub worker: Arc, + /// 扩展数据容器 pub extension: Extensions, } impl Context { + /// 创建新的上下文实例 pub fn new(worker: Arc) -> Self { Self { worker, extension: Default::default(), } } - - // pub async fn get_config(&self) -> AppResult { - // get_config(&self.pool).await - // } } +/// 工作器结构体，管理单个主播的录制和上传任务 #[derive(Debug)] pub struct Worker { + /// 下载器状态 pub downloader_status: RwLock, + /// 上传器状态 pub uploader_status: RwLock, + /// 直播主播信息 pub live_streamer: LiveStreamer, + /// 上传配置（可选） pub upload_streamer: Option, + /// 全局配置 pub config: Arc>, + /// HTTP客户端 pub client: StatelessClient, } impl Worker { + /// 创建新的工作器实例 pub fn new( live_streamer: LiveStreamer, upload_streamer: Option, @@ -57,29 +65,22 @@ impl Worker { } } + /// 获取主播信息 pub fn get_streamer(&self) -> LiveStreamer { - // get_streamer(&self.context.pool, self.id).await self.live_streamer.clone() } + /// 获取上传配置 pub fn get_upload_config(&self) -> Option { - // let Some(id) = self.get_streamer().await?.upload_streamers_id else { - // return Ok(None); - // }; - // - // UploadStreamer::select() - // .where_("id = ?") - // .bind(id) - // .fetch_optional(&self.context.pool) - // .await - // .change_context(AppError::Unknown) self.upload_streamer.clone() } + /// 获取全局配置 pub fn get_config(&self) -> Config { self.config.read().unwrap().clone() } + /// 更改工作器状态 pub fn change_status(&self, stage: Stage, status: WorkerStatus) { match stage { Stage::Download => { @@ -106,18 +107,23 @@ impl PartialEq for Worker { impl Eq for Worker {} +/// 工作阶段枚举 #[derive(Debug)] pub enum Stage { + /// 下载阶段 Download, + /// 上传阶段 Upload, } +/// 工作器状态枚举 #[derive(Clone, Copy, Debug, Serialize, Deserialize, Default, PartialEq)] pub enum WorkerStatus { - /// Stream is online. + /// 正在工作 Working, - /// The status of the stream could not be determined. + /// 等待中 Pending, + /// 空闲状态（默认） #[default] Idle, } diff --git a/crates/biliup-cli/src/server/infrastructure/dto.rs b/crates/biliup-cli/src/server/infrastructure/dto.rs index 198b82c..1ca8de0 100644 --- a/crates/biliup-cli/src/server/infrastructure/dto.rs +++ b/crates/biliup-cli/src/server/infrastructure/dto.rs @@ -2,10 +2,14 @@ use crate::server::infrastructure::context::WorkerStatus; use crate::server::infrastructure::models::LiveStreamer; use serde::{Deserialize, Serialize}; +/// 直播主播响应数据传输对象 +/// 包含主播信息和当前工作状态 #[derive(Serialize, Deserialize)] pub struct LiveStreamerResponse { + /// 主播基本信息（展开到顶层） #[serde(flatten)] pub inner: LiveStreamer, + /// 当前工作状态 pub status: WorkerStatus, } diff --git a/crates/biliup-cli/src/server/infrastructure/mod.rs b/crates/biliup-cli/src/server/infrastructure/mod.rs index 4eed25d..597364c 100644 --- a/crates/biliup-cli/src/server/infrastructure/mod.rs +++ b/crates/biliup-cli/src/server/infrastructure/mod.rs @@ -1,7 +1,14 @@ +/// 数据库连接池管理 pub mod connection_pool; +/// 应用程序上下文和工作器 pub mod context; +/// 数据传输对象 pub mod dto; +/// 数据模型定义 pub mod models; +/// 数据仓库层 pub mod repositories; +/// 服务注册器 pub mod service_register; +/// 用户认证相关 pub mod users; 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 1ae3358..9bda71a 100644 --- a/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs +++ b/crates/biliup-cli/src/server/infrastructure/models/hook_step.rs @@ -6,29 +6,33 @@ use tokio::io::AsyncWriteExt; use tokio::process::Command; use tracing::info; -// 钩子步骤：既支持 key-value 形式（如 {run: "..."}），也支持纯字符串（如 "rm"） +/// 钩子步骤枚举：支持多种操作格式 +/// 既支持 key-value 形式（如 {run: "..."}），也支持纯字符串（如 "rm"） #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(untagged)] pub enum HookStep { - // 处理 {run = "..."} 格式 + /// 执行命令格式：{run: "command"} Run { run: String }, - // 处理 {mv = "..."} 格式 + /// 移动文件格式：{mv: "target_dir"} Move { mv: String }, - // 处理 "rm" 字符串格式 + /// 删除文件格式："rm" Remove(String), } impl HookStep { - /// 执行后处理操作 + /// 执行钩子步骤操作 pub async fn execute(&self, video_paths: &[&Path]) -> Result<(), Box> { match self { HookStep::Run { run } => { + // 执行自定义命令 self.execute_command(run, video_paths).await?; } HookStep::Move { mv } => { + // 移动文件到指定目录 self.move_file(video_paths, mv).await?; } HookStep::Remove(cmd) if cmd == "rm" => { + // 删除文件 self.remove_file(video_paths).await?; } HookStep::Remove(cmd) => { @@ -38,7 +42,7 @@ impl HookStep { Ok(()) } - /// 执行命令，将视频路径作为标准输入传入 + /// 执行自定义命令，将视频路径作为标准输入传入 async fn execute_command( &self, cmd: &str, @@ -46,11 +50,13 @@ impl HookStep { ) -> Result<(), Box> { println!("Executing: {}", cmd); + // 解析命令和参数 let parts: Vec<&str> = cmd.split_whitespace().collect(); if parts.is_empty() { return Err("Empty command".into()); } + // 启动子进程 let mut process = Command::new(parts[0]) .args(&parts[1..]) .stdin(std::process::Stdio::piped()) @@ -58,11 +64,15 @@ impl HookStep { // 将视频文件路径写入标准输入 if let Some(mut stdin) = process.stdin.take() { - stdin - .write_all(video_paths.iter().map(|p| p.to_string_lossy()).reduce(|acc, e| Cow::from(acc.to_string() + "\n" + &*e)).expect("REASON").as_bytes()) - .await?; + let paths_str = video_paths + .iter() + .map(|p| p.to_string_lossy()) + .reduce(|acc, e| Cow::from(acc.to_string() + "\n" + &*e)) + .expect("At least one video path should exist"); + 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()); @@ -83,9 +93,10 @@ impl HookStep { if !target_path.exists() { 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); println!( @@ -98,7 +109,7 @@ impl HookStep { Ok(()) } - /// 删除文件 + /// 删除指定的视频文件 async fn remove_file(&self, video_paths: &[&Path]) -> Result<(), Box> { for video_path in video_paths { println!("Removing: {}", video_path.display()); @@ -108,14 +119,18 @@ impl HookStep { } } -/// 处理所有后处理器 +/// 处理所有后处理器步骤 pub async fn process_video( video_path: &[&Path], processors: &[HookStep], ) -> Result<(), Box> { - info!("Starting processing..."); + info!("Starting video processing..."); + + // 依次执行每个处理器步骤 for processor in processors { processor.execute(video_path).await?; } + + info!("Video processing completed"); Ok(()) } diff --git a/crates/biliup-cli/src/server/router.rs b/crates/biliup-cli/src/server/router.rs index 4c05a9c..8bb8631 100644 --- a/crates/biliup-cli/src/server/router.rs +++ b/crates/biliup-cli/src/server/router.rs @@ -12,46 +12,51 @@ use crate::server::infrastructure::service_register::ServiceRegister; use axum::Router; use axum::routing::{delete, get, post}; +/// 创建应用程序路由 pub fn router(service_register: ServiceRegister) -> Router<()> { Router::new() - // `GET /` goes to `root` + // 主播管理相关路由 .route( "/v1/streamers", - get(get_streamers_endpoint) - .post(post_streamers_endpoint) - .put(put_streamers_endpoint), + get(get_streamers_endpoint) // 获取主播列表 + .post(post_streamers_endpoint) // 添加主播 + .put(put_streamers_endpoint), // 更新主播 ) - .route("/v1/streamers/{id}", delete(delete_streamers_endpoint)) + .route("/v1/streamers/{id}", delete(delete_streamers_endpoint)) // 删除主播 + + // 配置管理路由 .route( "/v1/configuration", - get(get_configuration).put(put_configuration), + get(get_configuration).put(put_configuration), // 获取/更新配置 ) - .route("/v1/streamer-info", get(get_streamer_info)) - .route("/v1/upload/streamers", get(get_upload_streamers_endpoint)) + + // 主播信息路由 + .route("/v1/streamer-info", get(get_streamer_info)) // 获取主播信息 + + // 上传模板管理路由 + .route("/v1/upload/streamers", get(get_upload_streamers_endpoint)) // 获取上传模板列表 .route( "/v1/upload/streamers/{id}", - delete(delete_template_endpoint). - // .put(update_template_endpoint) - get(get_upload_streamer_endpoint), + delete(delete_template_endpoint) // 删除上传模板 + .get(get_upload_streamer_endpoint), // 获取单个上传模板 ) - // .route("/v1/streamers/{id}", get(get_streamer_endpoint)) - // .route("/v1/streamers/{id}", delete(delete_streamer_endpoint)) - // .route("/v1/streamers/{id}", put(update_streamer_endpoint)) - // .route("/v1/streamers", post(add_streamer_endpoint)) - // .route( - // "/v1/upload/streamers/:id", - // , - // ) - // .route("/v1/upload/streamers/:id", ) - .route("/v1/upload/streamers", post(add_upload_streamer_endpoint)) - .route("/v1/users", get(get_users_endpoint).post(add_user_endpoint)) - .route("/v1/users/{id}", delete(delete_user_endpoint)) - .route("/bili/archive/pre", get(archive_pre_endpoint)) - .route("/bili/space/myinfo", get(get_myinfo_endpoint)) - .route("/bili/proxy", get(get_proxy_endpoint)) - .route("/v1/get_qrcode", get(get_qrcode)) - .route("/v1/login_by_qrcode", post(login_by_qrcode)) - .route("/v1/videos", get(get_videos)) - // .layer(Extension(client.clone())) - .with_state(service_register) + .route("/v1/upload/streamers", post(add_upload_streamer_endpoint)) // 添加上传模板 + + // 用户管理路由 + .route("/v1/users", get(get_users_endpoint).post(add_user_endpoint)) // 获取用户列表/添加用户 + .route("/v1/users/{id}", delete(delete_user_endpoint)) // 删除用户 + + // B站API代理路由 + .route("/bili/archive/pre", get(archive_pre_endpoint)) // 投稿预处理 + .route("/bili/space/myinfo", get(get_myinfo_endpoint)) // 获取用户信息 + .route("/bili/proxy", get(get_proxy_endpoint)) // 代理请求 + + // 认证相关路由 + .route("/v1/get_qrcode", get(get_qrcode)) // 获取二维码 + .route("/v1/login_by_qrcode", post(login_by_qrcode)) // 二维码登录 + + // 视频文件管理路由 + .route("/v1/videos", get(get_videos)) // 获取视频列表 + + .with_state(service_register) // 注入服务注册器状态 }