From e41acef883868a55da7c1d7b2cc5bff4d59ea79d Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Sat, 27 Sep 2025 01:13:18 +0800 Subject: [PATCH] Refactor upload logic and add WebSocket log endpoint Refactored upload logic to decouple upload and submission, moving UploadLine and related logic to biliup-cli crate root. Added a new WebSocket log endpoint in a dedicated ws.rs module and routed it accordingly. Updated database migration to set certain fields to null instead of empty strings. Adjusted context and download/upload manager logic to pass StreamInfo and Context as needed. Added static file serving route and improved filename generation for uploads. Bumped version to 1.1.8 and enabled 'fs' feature for tower-http. --- Cargo.lock | 20 +- Cargo.toml | 2 +- crates/biliup-cli/Cargo.toml | 2 +- .../migrations/2_fix_preprocessor_null.sql | 32 +- crates/biliup-cli/src/cli.rs | 22 +- crates/biliup-cli/src/lib.rs | 20 ++ crates/biliup-cli/src/server/api/endpoints.rs | 282 +++--------------- crates/biliup-cli/src/server/api/mod.rs | 1 + crates/biliup-cli/src/server/api/ws.rs | 255 ++++++++++++++++ crates/biliup-cli/src/server/app.rs | 6 +- .../biliup-cli/src/server/common/download.rs | 31 +- crates/biliup-cli/src/server/common/upload.rs | 152 ++++++++-- crates/biliup-cli/src/server/common/util.rs | 9 +- crates/biliup-cli/src/server/config.rs | 10 +- .../src/server/core/download_manager.rs | 20 +- crates/biliup-cli/src/server/core/monitor.rs | 5 +- crates/biliup-cli/src/server/core/plugin.rs | 7 +- .../src/server/infrastructure/context.rs | 5 +- .../src/server/infrastructure/models.rs | 7 +- crates/biliup-cli/src/server/router.rs | 18 +- crates/biliup-cli/src/uploader.rs | 2 +- crates/biliup/src/uploader/bilibili.rs | 2 +- crates/stream-gears/src/server.rs | 4 +- crates/stream-gears/src/uploader.rs | 122 +++----- 24 files changed, 596 insertions(+), 440 deletions(-) create mode 100644 crates/biliup-cli/src/server/api/ws.rs 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