diff --git a/Cargo.lock b/Cargo.lock index 6b7bd0d..49921fa 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -395,7 +395,7 @@ dependencies = [ [[package]] name = "biliup-cli" -version = "1.1.7" +version = "1.1.8" dependencies = [ "async-channel", "async-trait", @@ -1464,6 +1464,12 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "http-range-header" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9171a2ea8a68358193d15dd5d70c1c10a2afc3e7e4c5bc92bc9f025cebd7359c" + [[package]] name = "httparse" version = "1.10.1" @@ -3635,7 +3641,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" [[package]] name = "stream-gears" -version = "1.1.7" +version = "1.1.8" dependencies = [ "async-channel", "async-trait", @@ -4090,14 +4096,24 @@ checksum = "adc82fd73de2a9722ac5da747f12383d2bfdb93591ee6c58486e0097890f05f2" dependencies = [ "bitflags 2.9.1", "bytes", + "futures-core", "futures-util", "http", "http-body", + "http-body-util", + "http-range-header", + "httpdate", "iri-string", + "mime", + "mime_guess", + "percent-encoding", "pin-project-lite", + "tokio", + "tokio-util", "tower", "tower-layer", "tower-service", + "tracing", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 46dfb10..57fa007 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ members = [ resolver = "3" [workspace.package] -version = "1.1.7" +version = "1.1.8" [workspace.dependencies] serde = { version = "1.0", features = ["derive"] } diff --git a/crates/biliup-cli/Cargo.toml b/crates/biliup-cli/Cargo.toml index 10fe301..141db5a 100644 --- a/crates/biliup-cli/Cargo.toml +++ b/crates/biliup-cli/Cargo.toml @@ -34,7 +34,7 @@ indexmap = { version = "2.8.0" } async-trait = { workspace = true } humantime = "2.2.0" axum = { version = "0.8", features = ["macros", "ws"] } -tower-http = { version = "0.6.2", features = ["cors"] } +tower-http = { version = "0.6.2", features = ["fs", "cors"] } reqwest = { workspace = true } sqlx = { version = "0.8", features = [ "runtime-tokio-rustls", "sqlite" ] } regex = { workspace = true } diff --git a/crates/biliup-cli/migrations/2_fix_preprocessor_null.sql b/crates/biliup-cli/migrations/2_fix_preprocessor_null.sql index ecafba1..756de88 100644 --- a/crates/biliup-cli/migrations/2_fix_preprocessor_null.sql +++ b/crates/biliup-cli/migrations/2_fix_preprocessor_null.sql @@ -3,34 +3,34 @@ -- 修复 livestreamers 表的 JSON 字段 UPDATE livestreamers -SET preprocessor = '[]' -WHERE preprocessor IS NULL OR preprocessor = '' OR preprocessor = 'null'; +SET preprocessor = null +WHERE preprocessor = '' OR preprocessor = 'null'; UPDATE livestreamers -SET segment_processor = '[]' -WHERE segment_processor IS NULL OR segment_processor = '' OR segment_processor = 'null'; +SET segment_processor = null +WHERE segment_processor = '' OR segment_processor = 'null'; UPDATE livestreamers -SET downloaded_processor = '[]' -WHERE downloaded_processor IS NULL OR downloaded_processor = '' OR downloaded_processor = 'null'; +SET downloaded_processor = null +WHERE downloaded_processor = '' OR downloaded_processor = 'null'; UPDATE livestreamers -SET postprocessor = '[]' -WHERE postprocessor IS NULL OR postprocessor = '' OR postprocessor = 'null'; +SET postprocessor = null +WHERE postprocessor = '' OR postprocessor = 'null'; UPDATE livestreamers -SET override = '' -WHERE override IS NULL OR override = 'null'; +SET override = null +WHERE override = '' OR override = 'null'; UPDATE livestreamers -SET excluded_keywords = '' -WHERE excluded_keywords IS NULL OR excluded_keywords = 'null'; +SET excluded_keywords = null +WHERE excluded_keywords = '' OR excluded_keywords = 'null'; UPDATE livestreamers -SET opt_args = '' -WHERE opt_args IS NULL OR opt_args = 'null'; +SET opt_args = null +WHERE opt_args = '' OR opt_args = 'null'; -- 修复 uploadstreamers 表的 JSON 字段 UPDATE uploadstreamers -SET tags = '[]' -WHERE tags IS NULL OR tags = '' OR tags = 'null'; \ No newline at end of file +SET tags = null +WHERE tags = '' OR tags = 'null'; \ No newline at end of file diff --git a/crates/biliup-cli/src/cli.rs b/crates/biliup-cli/src/cli.rs index e765609..ab6582d 100644 --- a/crates/biliup-cli/src/cli.rs +++ b/crates/biliup-cli/src/cli.rs @@ -1,7 +1,8 @@ use biliup::uploader::bilibili::{Studio, Vid}; use biliup::uploader::util::SubmitOption; -use clap::{Parser, Subcommand, ValueEnum}; +use clap::{Parser, Subcommand}; +use biliup_cli::UploadLine; use std::path::PathBuf; #[derive(Parser)] @@ -148,25 +149,6 @@ pub enum Commands { }, } -#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)] -pub enum UploadLine { - Bldsa, - Cnbldsa, - Andsa, - Atdsa, - Bda2, - Cnbd, - Anbd, - Atbd, - Tx, - Cntx, - Antx, - Attx, - Bda, - Txa, - Alia, -} - fn human_size(s: &str) -> Result { let ret = match s.as_bytes() { [init @ .., b'K'] => parse_u8(init)? * 1000.0, diff --git a/crates/biliup-cli/src/lib.rs b/crates/biliup-cli/src/lib.rs index ab04306..f4e7c42 100644 --- a/crates/biliup-cli/src/lib.rs +++ b/crates/biliup-cli/src/lib.rs @@ -7,6 +7,7 @@ use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::connection_pool::ConnectionManager; use crate::server::infrastructure::repositories; use crate::server::infrastructure::service_register::ServiceRegister; +use clap::ValueEnum; use error_stack::ResultExt; use std::net::ToSocketAddrs; use std::sync::{Arc, RwLock}; @@ -41,3 +42,22 @@ pub async fn run(addr: (&str, u16)) -> AppResult<()> { .attach("could not initialize application routes")?; Ok(()) } + +#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, ValueEnum)] +pub enum UploadLine { + Bldsa, + Cnbldsa, + Andsa, + Atdsa, + Bda2, + Cnbd, + Anbd, + Atbd, + Tx, + Cntx, + Antx, + Attx, + Bda, + Txa, + Alia, +} diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index 8291e4c..52fd022 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -1,3 +1,6 @@ +use crate::UploadLine; +use crate::server::common::upload::{build_studio, submit_to_bilibili, upload}; +use crate::server::common::util::Recorder; use crate::server::config::Config; use crate::server::core::download_manager::ActorHandle; use crate::server::errors::{AppError, report_to_response}; @@ -18,6 +21,7 @@ use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::{Json, debug_handler}; use biliup::credential::Credential; +use clap::ValueEnum; use error_stack::{Report, ResultExt}; use ormlite::{Insert, Model}; use serde::Deserialize; @@ -91,7 +95,7 @@ pub async fn post_streamers_endpoint( info!(workers=?service_register.workers, "successfully inserted new live streamers"); Ok(Json(live_streamers)) } -#[debug_handler(state = ServiceRegister)] + pub async fn put_streamers_endpoint( State(service_register): State, State(pool): State, @@ -479,244 +483,52 @@ pub async fn get_status( }))) } -static ALLOWED_FILES: &[&str] = &["ds_update.log", "download.log", "upload.log"]; - -#[derive(Debug, Deserialize, Clone)] -pub struct LogsQuery { - file: Option, -} -pub async fn ws_logs( - ws: WebSocketUpgrade, - Query(query): Query, -) -> axum::response::Response { - ws.on_upgrade(move |socket| websocket_logs(socket, query)) +#[derive(Deserialize)] +pub struct PostUploads { + files: Vec, + params: UploadStreamer, } -async fn websocket_logs(mut ws: WebSocket, query: LogsQuery) { - // 参数获取与校验 - let file_param = query.file.unwrap_or_else(|| "ds_update.log".to_string()); - if !ALLOWED_FILES.contains(&file_param.as_str()) { - let _ = ws - .send(Message::Text( - format!("不允许访问请求的文件: {}", file_param).into(), - )) - .await; - let _ = ws.send(Message::Close(None)).await; - return; - } - - let log_file = PathBuf::from(&file_param); - - // 发送初始内容（最后50行）并获取当前大小 - let mut file_size = match send_last_lines(&mut ws, &log_file, 50).await { - Ok(size) => size, - Err(e) => { - match e.kind() { - ErrorKind::NotFound => { - let _ = ws - .send(Message::Text( - format!("日志文件 {} 不存在", log_file.display()).into(), - )) - .await; - } - _ => { - let _ = ws - .send(Message::Text(format!("读取日志文件错误: {}", e).into())) - .await; - error!("读取日志文件错误: {}", e); - } - } - let _ = ws.send(Message::Close(None)).await; - return; - } +#[debug_handler] +pub async fn post_uploads( + State(config): State>>, + Json(json_data): Json, +) -> Result, Response> { + let upload_config = json_data.params; + let (line, limit, submit_api) = { + let config = config.read().unwrap(); + let line = UploadLine::from_str(&config.lines, true).ok(); + let limit = config.threads; + let submit_api = config.submit_api.clone(); + (line, limit, submit_api) }; - - // 心跳/轮询间隔 - let mut tick = interval(Duration::from_millis(500)); - tick.set_missed_tick_behavior(MissedTickBehavior::Skip); - - // 主循环：同时处理客户端消息和文件更新 - loop { - tokio::select! { - maybe_msg = ws.recv() => { - match maybe_msg { - Some(Ok(Message::Close(_))) => { - let _ = ws.send(Message::Close(None)).await; - break; - } - Some(Ok(Message::Ping(payload))) => { - // 回应 PONG - let _ = ws.send(Message::Pong(payload)).await; - } - Some(Ok(_)) => { - // 其他消息不处理（Text/Binary等） - } - Some(Err(e)) => { - error!("WebSocket连接错误: {}", e); - break; - } - None => { - info!("WebSocket连接已关闭"); - break; - } - } - } - - _ = tick.tick() => { - // 文件是否存在 - let meta = match fs::metadata(&log_file).await { - Ok(m) => m, - Err(e) if e.kind() == ErrorKind::NotFound => { - let _ = ws.send(Message::Text(format!( - "日志文件 {} 不再存在", - log_file.display() - ).into())).await; - break; - } - Err(e) => { - let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await; - error!("websocket_logs错误: {}", e); - break; - } - }; - - let current_size = meta.len(); - - // 文件被截断 - if current_size < file_size { - let _ = ws.send(Message::Text(Utf8Bytes::from("日志文件被截断，重新加载...".to_string()))).await; - match send_last_lines(&mut ws, &log_file, 50).await { - Ok(size) => file_size = size, - Err(e) => { - let _ = ws.send(Message::Text(format!("读取日志文件错误: {}", e).into())).await; - error!("读取日志文件错误: {}", e); - break; - } - } - continue; - } - - // 文件新增内容 - if current_size > file_size { - if let Err(e) = send_new_lines_from_offset(&mut ws, &log_file, file_size).await { - let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await; - error!("websocket_logs错误: {}", e); - break; - } - file_size = current_size; - } - } - } - } - - let _ = ws.send(Message::Close(None)).await; - debug!("WebSocket日志会话结束: {}", file_param); -} - -// 发送最后 n 行，并返回当前文件大小 -async fn send_last_lines( - ws: &mut WebSocket, - path: &std::path::Path, - n: usize, -) -> std::io::Result { - let meta = fs::metadata(path).await?; - let file_size = meta.len(); - - let file = fs::File::open(path).await?; - let mut reader = BufReader::new(file); - let mut lines = reader.lines(); - - let mut buf: VecDeque = VecDeque::with_capacity(n); - while let Some(line) = lines.next_line().await? { - if buf.len() == n { - buf.pop_front(); - } - buf.push_back(line); - } - for line in buf { - ws.send(Message::Text(Utf8Bytes::from(line))) + info!("通过页面开始上传"); + let (bilibili, videos) = upload( + upload_config + .user_cookie + .as_deref() + .unwrap_or("cookies.json"), + None, + line, + &json_data.files, + limit as usize, + ) + .await + .map_err(report_to_response)?; + if !videos.is_empty() { + let recorder = Recorder::new( + upload_config.title.clone(), + &upload_config.template_name, + "stream_title", + "", + ); + let studio = build_studio(&upload_config, &bilibili, videos, recorder) .await - .map_err(|e| { - std::io::Error::new( - ErrorKind::ConnectionAborted, - format!("发送WebSocket消息失败: {}", e), - ) - })?; - } - Ok(file_size) -} - -// 从偏移量开始读取新增内容，并逐行发送 -async fn send_new_lines_from_offset( - ws: &mut WebSocket, - path: &std::path::Path, - offset: u64, -) -> std::io::Result<()> { - let mut file = fs::File::open(path).await?; - file.seek(std::io::SeekFrom::Start(offset)).await?; - - // 直接读到字符串（UTF-8），若遇到非UTF-8可换成读bytes+lossy - let mut s = String::new(); - if let Err(e) = file.read_to_string(&mut s).await { - // 如果遇到非UTF-8数据，降级为 lossy - let mut bytes = Vec::new(); - file.seek(std::io::SeekFrom::Start(offset)).await?; - file.read_to_end(&mut bytes).await?; - s = String::from_utf8_lossy(&bytes).into_owned(); - if e.kind() != ErrorKind::InvalidData { - // 非编码错误也要汇报 - error!("读取日志文件新内容失败: {}", e); - } - } - - for line in s.lines() { - ws.send(Message::Text(Utf8Bytes::from(line.to_string()))) + .map_err(report_to_response)?; + let response_data = submit_to_bilibili(&bilibili, &studio, submit_api.as_deref()) .await - .map_err(|e| { - std::io::Error::new( - ErrorKind::ConnectionAborted, - format!("发送WebSocket消息失败: {}", e), - ) - })?; + .map_err(report_to_response)?; + info!("通过页面上传成功 {:?}", response_data); } - Ok(()) -} - -async fn resolve_latest_log_path( - dir: &std::path::Path, - prefix: &str, - suffix: &str, -) -> io::Result { - // 1) 活跃文件 prefix.log - let active = dir.join(format!("{}.{}", prefix, suffix)); - if fs::metadata(&active).await.is_ok() { - return Ok(active); - } - - // 2) 回退到归档文件 prefix.*.log 中最新的一个 - let mut rd = fs::read_dir(dir).await?; - let pre = format!("{}.", prefix); - let suf = format!(".{}", suffix); - - let mut candidates: Vec<(String, PathBuf)> = Vec::new(); - while let Some(ent) = rd.next_entry().await? { - let path = ent.path(); - if !path.is_file() { - continue; - } - let Some(name) = path.file_name().and_then(|s| s.to_str()) else { - continue; - }; - if name.starts_with(&pre) && name.ends_with(&suf) { - candidates.push((name.to_string(), path)); - } - } - - if candidates.is_empty() { - return Err(io::Error::new(ErrorKind::NotFound, "no log file found")); - } - - // tracing-appender 的归档名使用 YYYY-MM-DD，中间按字符串排序即时间顺序 - candidates.sort_by(|a, b| a.0.cmp(&b.0)); - Ok(candidates.last().unwrap().1.clone()) + Ok(Json(serde_json::json!({}))) } diff --git a/crates/biliup-cli/src/server/api/mod.rs b/crates/biliup-cli/src/server/api/mod.rs index 218dea0..51d4c0e 100644 --- a/crates/biliup-cli/src/server/api/mod.rs +++ b/crates/biliup-cli/src/server/api/mod.rs @@ -6,3 +6,4 @@ pub mod bilibili_endpoints; pub mod endpoints; /// 单页应用静态文件处理 pub mod spa; +pub mod ws; diff --git a/crates/biliup-cli/src/server/api/ws.rs b/crates/biliup-cli/src/server/api/ws.rs new file mode 100644 index 0000000..0b837d5 --- /dev/null +++ b/crates/biliup-cli/src/server/api/ws.rs @@ -0,0 +1,255 @@ +use axum::extract::ws::{Message, Utf8Bytes, WebSocket}; +use axum::extract::{Query, WebSocketUpgrade}; +use serde::Deserialize; +use std::collections::VecDeque; +use std::io; +use std::io::ErrorKind; +use std::path::PathBuf; +use std::time::Duration; +use tokio::fs; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, BufReader}; +use tokio::time::{MissedTickBehavior, interval}; +use tracing::{debug, error, info}; + +static ALLOWED_FILES: &[&str] = &["ds_update.log", "download.log", "upload.log"]; + +#[derive(Debug, Deserialize, Clone)] +pub struct LogsQuery { + file: Option, +} + +pub async fn ws_logs( + ws: WebSocketUpgrade, + Query(query): Query, +) -> axum::response::Response { + ws.on_upgrade(move |socket| websocket_logs(socket, query)) +} + +async fn websocket_logs(mut ws: WebSocket, query: LogsQuery) { + // 参数获取与校验 + let file_param = query.file.unwrap_or_else(|| "ds_update.log".to_string()); + if !ALLOWED_FILES.contains(&file_param.as_str()) { + let _ = ws + .send(Message::Text( + format!("不允许访问请求的文件: {}", file_param).into(), + )) + .await; + let _ = ws.send(Message::Close(None)).await; + return; + } + + let log_file = PathBuf::from(&file_param); + + // 发送初始内容（最后50行）并获取当前大小 + let mut file_size = match send_last_lines(&mut ws, &log_file, 50).await { + Ok(size) => size, + Err(e) => { + match e.kind() { + ErrorKind::NotFound => { + let _ = ws + .send(Message::Text( + format!("日志文件 {} 不存在", log_file.display()).into(), + )) + .await; + } + _ => { + let _ = ws + .send(Message::Text(format!("读取日志文件错误: {}", e).into())) + .await; + error!("读取日志文件错误: {}", e); + } + } + let _ = ws.send(Message::Close(None)).await; + return; + } + }; + + // 心跳/轮询间隔 + let mut tick = interval(Duration::from_millis(500)); + tick.set_missed_tick_behavior(MissedTickBehavior::Skip); + + // 主循环：同时处理客户端消息和文件更新 + loop { + tokio::select! { + maybe_msg = ws.recv() => { + match maybe_msg { + Some(Ok(Message::Close(_))) => { + let _ = ws.send(Message::Close(None)).await; + break; + } + Some(Ok(Message::Ping(payload))) => { + // 回应 PONG + let _ = ws.send(Message::Pong(payload)).await; + } + Some(Ok(_)) => { + // 其他消息不处理（Text/Binary等） + } + Some(Err(e)) => { + error!("WebSocket连接错误: {}", e); + break; + } + None => { + info!("WebSocket连接已关闭"); + break; + } + } + } + + _ = tick.tick() => { + // 文件是否存在 + let meta = match fs::metadata(&log_file).await { + Ok(m) => m, + Err(e) if e.kind() == ErrorKind::NotFound => { + let _ = ws.send(Message::Text(format!( + "日志文件 {} 不再存在", + log_file.display() + ).into())).await; + break; + } + Err(e) => { + let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await; + error!("websocket_logs错误: {}", e); + break; + } + }; + + let current_size = meta.len(); + + // 文件被截断 + if current_size < file_size { + let _ = ws.send(Message::Text(Utf8Bytes::from("日志文件被截断，重新加载...".to_string()))).await; + match send_last_lines(&mut ws, &log_file, 50).await { + Ok(size) => file_size = size, + Err(e) => { + let _ = ws.send(Message::Text(format!("读取日志文件错误: {}", e).into())).await; + error!("读取日志文件错误: {}", e); + break; + } + } + continue; + } + + // 文件新增内容 + if current_size > file_size { + if let Err(e) = send_new_lines_from_offset(&mut ws, &log_file, file_size).await { + let _ = ws.send(Message::Text(format!("监控日志文件错误: {}", e).into())).await; + error!("websocket_logs错误: {}", e); + break; + } + file_size = current_size; + } + } + } + } + + let _ = ws.send(Message::Close(None)).await; + debug!("WebSocket日志会话结束: {}", file_param); +} + +// 发送最后 n 行，并返回当前文件大小 +async fn send_last_lines( + ws: &mut WebSocket, + path: &std::path::Path, + n: usize, +) -> std::io::Result { + let meta = fs::metadata(path).await?; + let file_size = meta.len(); + + let file = fs::File::open(path).await?; + let mut reader = BufReader::new(file); + let mut lines = reader.lines(); + + let mut buf: VecDeque = VecDeque::with_capacity(n); + while let Some(line) = lines.next_line().await? { + if buf.len() == n { + buf.pop_front(); + } + buf.push_back(line); + } + for line in buf { + ws.send(Message::Text(Utf8Bytes::from(line))) + .await + .map_err(|e| { + std::io::Error::new( + ErrorKind::ConnectionAborted, + format!("发送WebSocket消息失败: {}", e), + ) + })?; + } + Ok(file_size) +} + +// 从偏移量开始读取新增内容，并逐行发送 +async fn send_new_lines_from_offset( + ws: &mut WebSocket, + path: &std::path::Path, + offset: u64, +) -> std::io::Result<()> { + let mut file = fs::File::open(path).await?; + file.seek(std::io::SeekFrom::Start(offset)).await?; + + // 直接读到字符串（UTF-8），若遇到非UTF-8可换成读bytes+lossy + let mut s = String::new(); + if let Err(e) = file.read_to_string(&mut s).await { + // 如果遇到非UTF-8数据，降级为 lossy + let mut bytes = Vec::new(); + file.seek(std::io::SeekFrom::Start(offset)).await?; + file.read_to_end(&mut bytes).await?; + s = String::from_utf8_lossy(&bytes).into_owned(); + if e.kind() != ErrorKind::InvalidData { + // 非编码错误也要汇报 + error!("读取日志文件新内容失败: {}", e); + } + } + + for line in s.lines() { + ws.send(Message::Text(Utf8Bytes::from(line.to_string()))) + .await + .map_err(|e| { + std::io::Error::new( + ErrorKind::ConnectionAborted, + format!("发送WebSocket消息失败: {}", e), + ) + })?; + } + Ok(()) +} + +async fn resolve_latest_log_path( + dir: &std::path::Path, + prefix: &str, + suffix: &str, +) -> io::Result { + // 1) 活跃文件 prefix.log + let active = dir.join(format!("{}.{}", prefix, suffix)); + if fs::metadata(&active).await.is_ok() { + return Ok(active); + } + + // 2) 回退到归档文件 prefix.*.log 中最新的一个 + let mut rd = fs::read_dir(dir).await?; + let pre = format!("{}.", prefix); + let suf = format!(".{}", suffix); + + let mut candidates: Vec<(String, PathBuf)> = Vec::new(); + while let Some(ent) = rd.next_entry().await? { + let path = ent.path(); + if !path.is_file() { + continue; + } + let Some(name) = path.file_name().and_then(|s| s.to_str()) else { + continue; + }; + if name.starts_with(&pre) && name.ends_with(&suf) { + candidates.push((name.to_string(), path)); + } + } + + if candidates.is_empty() { + return Err(io::Error::new(ErrorKind::NotFound, "no log file found")); + } + + // tracing-appender 的归档名使用 YYYY-MM-DD，中间按字符串排序即时间顺序 + candidates.sort_by(|a, b| a.0.cmp(&b.0)); + Ok(candidates.last().unwrap().1.clone()) +} diff --git a/crates/biliup-cli/src/server/app.rs b/crates/biliup-cli/src/server/app.rs index 3c97d24..7af0328 100644 --- a/crates/biliup-cli/src/server/app.rs +++ b/crates/biliup-cli/src/server/app.rs @@ -2,8 +2,8 @@ use axum::http; use crate::server; use crate::server::api::auth; -use crate::server::api::endpoints::ws_logs; use crate::server::api::spa::static_handler; +use crate::server::api::ws::ws_logs; use crate::server::errors::{AppError, AppResult}; use crate::server::infrastructure::service_register::ServiceRegister; use crate::server::infrastructure::users::Backend; @@ -48,8 +48,8 @@ impl ApplicationController { // 配置会话管理层 let session_layer = SessionManagerLayer::new(session_store) .with_secure(false) - .with_expiry(Expiry::OnInactivity(Duration::days(1))) - .with_signed(key); + .with_expiry(Expiry::OnInactivity(Duration::days(7))); + // .with_signed(key); // 认证服务配置 // 将会话层与后端结合，建立认证服务，将认证会话作为请求扩展提供 diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index 8fa85db..e8008f1 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -144,13 +144,13 @@ pub struct SegmentEventProcessor { tx: Sender, rx: Receiver, uploader: Sender, - worker: Arc, + ctx: Context, file_validator: FileValidator, } impl SegmentEventProcessor { /// 创建处理器 - pub fn new(uploader: Sender, worker: Arc) -> Self { + pub fn new(uploader: Sender, ctx: Context) -> Self { let (tx, rx) = async_channel::bounded(32); // Use tokio channel for async Self { @@ -158,10 +158,17 @@ impl SegmentEventProcessor { rx, uploader, file_validator: FileValidator::new( - worker.clone().config.read().unwrap().filtering_threshold * 1000 * 1000, + ctx.worker + .clone() + .config + .read() + .unwrap() + .filtering_threshold + * 1000 + * 1000, true, ), - worker, + ctx, } } @@ -175,7 +182,7 @@ impl SegmentEventProcessor { .uploader .force_send(UploaderMessage::SegmentEvent( self.rx.clone(), - self.worker.clone(), + self.ctx.clone(), )) .change_context(AppError::Custom("Failed to send to uploader".to_string()))?; if let Some(prev) = res { @@ -239,7 +246,6 @@ impl SegmentEventProcessor { /// 下载任务 pub struct DownloadTask { plugin: Arc, - stream_info: StreamInfo, ctx: Context, rooms_handle: Arc, } @@ -252,13 +258,11 @@ struct DownloadComponents { impl DownloadTask { pub fn new( plugin: Arc, - stream_info: StreamInfo, ctx: Context, rooms_handle: Arc, ) -> Self { Self { plugin, - stream_info, ctx, rooms_handle, } @@ -277,7 +281,7 @@ impl DownloadTask { ); // 创建事件处理器 - let processor = SegmentEventProcessor::new(uploader.clone(), self.ctx.worker.clone()); + let processor = SegmentEventProcessor::new(uploader.clone(), self.ctx.clone()); // 启动弹幕客户端 if let Some(ref client) = components.danmaku_client { @@ -307,15 +311,16 @@ impl DownloadTask { // 获取配置和主播信息 let config = self.ctx.worker.get_config(); let streamer = self.ctx.worker.get_streamer(); + let stream_info = &self.ctx.stream_info; // 确定文件格式后缀 let suffix = streamer .format - .unwrap_or_else(|| self.stream_info.suffix.clone()); + .unwrap_or_else(|| stream_info.suffix.clone()); // 创建录制器 let recorder = Recorder::new( streamer.filename_prefix, &streamer.remark, - &self.stream_info.title, + &stream_info.title, &suffix, ); // 可选的弹幕客户端 @@ -329,7 +334,7 @@ impl DownloadTask { // 创建下载器实例 let downloader = self .plugin - .create_downloader(&self.stream_info, config, recorder) + .create_downloader(stream_info, config, recorder) .await; Ok(DownloadComponents { @@ -342,7 +347,7 @@ impl DownloadTask { // 启动弹幕下载逻辑 println!( "Starting danmaku client for stream: {}", - self.stream_info.url + self.ctx.stream_info.url ); client.download(Box::new(|_| {})).await?; Ok(()) diff --git a/crates/biliup-cli/src/server/common/upload.rs b/crates/biliup-cli/src/server/common/upload.rs index b216561..fe8cdf4 100644 --- a/crates/biliup-cli/src/server/common/upload.rs +++ b/crates/biliup-cli/src/server/common/upload.rs @@ -1,16 +1,20 @@ +use crate::UploadLine; +use crate::server::common::util::Recorder; use crate::server::core::download_manager::UActor; use crate::server::core::downloader::{SegmentEvent, SegmentInfo}; use crate::server::errors::{AppError, AppResult}; -use crate::server::infrastructure::context::Worker; +use crate::server::infrastructure::context::{Context, Worker}; use crate::server::infrastructure::models::UploadStreamer; use crate::server::infrastructure::models::hook_step::process_video; use async_channel::Receiver; -use biliup::bilibili::{BiliBili, ResponseData, Studio, Video}; +use biliup::bilibili::{BiliBili, Credit, ResponseData, Studio, Video}; use biliup::client::StatelessClient; use biliup::credential::login_by_cookies; +use biliup::error::Kind; use biliup::uploader::line::{Line, Probe}; use biliup::uploader::util::SubmitOption; use biliup::uploader::{VideoFile, line}; +use chrono::Local; use error_stack::ResultExt; use futures::StreamExt; use std::path::{Path, PathBuf}; @@ -35,24 +39,38 @@ struct UploadedVideos { pub async fn process_with_upload( rx: Receiver, - worker: &Worker, + ctx: &Context, upload_config: UploadStreamer, ) -> AppResult<()> { info!(upload_config=?upload_config, "Starting process with upload"); // 1. 初始化上传环境 - let upload_context = initialize_upload_context(&worker, upload_config).await?; + let upload_context = initialize_upload_context(&ctx.worker, upload_config).await?; // 2. 流水线处理视频上传 let uploaded_videos = pipeline_upload_videos(rx, &upload_context).await?; // 3. 提交到B站 if !uploaded_videos.videos.is_empty() { - submit_to_bilibili(&upload_context, uploaded_videos.videos, &worker).await?; + let recorder = Recorder::new( + upload_context.upload_config.title.clone(), + &ctx.worker.live_streamer.remark, + &ctx.stream_info.title, + "", + ); + let studio = build_studio( + &upload_context.upload_config, + &upload_context.bilibili, + uploaded_videos.videos, + recorder, + ) + .await?; + let submit_api = ctx.worker.config.read().unwrap().submit_api.clone(); + submit_to_bilibili(&upload_context.bilibili, &studio, submit_api.as_deref()).await?; } // 4. 执行后处理 if !uploaded_videos.paths.is_empty() { - execute_postprocessor(uploaded_videos.paths, &worker).await?; + execute_postprocessor(uploaded_videos.paths, &ctx.worker).await?; } Ok(()) @@ -102,7 +120,7 @@ async fn pipeline_upload_videos( context: &UploadContext, ) -> AppResult { // let mut desc_v2 = Vec::new(); - // for credit in desc_v2_credit { + // for credit in context.upload_config.desc_v2_credit { // desc_v2.push(Credit { // type_id: credit.type_id, // raw_text: credit.raw_text, @@ -167,12 +185,11 @@ async fn upload_single_file(file_path: &Path, context: &UploadContext) -> AppRes Ok(video) } -async fn submit_to_bilibili( - context: &UploadContext, - videos: Vec