diff --git a/Cargo.lock b/Cargo.lock index 178cfbd..6b7bd0d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -251,6 +251,7 @@ checksum = "021e862c184ae977658b36c4500f7feac3221ca5da43e3f25bd04ab6c79a29b5" dependencies = [ "axum-core", "axum-macros", + "base64", "bytes", "form_urlencoded", "futures-util", @@ -270,8 +271,10 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", + "sha1", "sync_wrapper", "tokio", + "tokio-tungstenite", "tower", "tower-layer", "tower-service", @@ -392,7 +395,7 @@ dependencies = [ [[package]] name = "biliup-cli" -version = "0.2.4" +version = "1.1.7" dependencies = [ "async-channel", "async-trait", @@ -904,6 +907,12 @@ dependencies = [ "syn", ] +[[package]] +name = "data-encoding" +version = "2.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a2330da5de22e8a3cb63252ce2abb30116bf5265e89c0e01bc17015ce30a476" + [[package]] name = "der" version = "0.7.10" @@ -3626,7 +3635,7 @@ checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" [[package]] name = "stream-gears" -version = "1.1.6" +version = "1.1.7" dependencies = [ "async-channel", "async-trait", @@ -3969,6 +3978,18 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio-tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7a9daff607c6d2bf6c16fd681ccb7eecc83e4e2cdc1ca067ffaadfca5de7f084" +dependencies = [ + "futures-util", + "log", + "tokio", + "tungstenite", +] + [[package]] name = "tokio-util" version = "0.7.16" @@ -4223,6 +4244,23 @@ version = "0.2.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" +[[package]] +name = "tungstenite" +version = "0.26.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4793cb5e56680ecbb1d843515b23b6de9a75eb04b66643e256a396d43be33c13" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand 0.9.1", + "sha1", + "thiserror 2.0.12", + "utf-8", +] + [[package]] name = "typenum" version = "1.18.0" @@ -4315,6 +4353,12 @@ version = "2.1.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "daf8dba3b7eb870caf1ddeed7bc9d2a049f3cfdfae7cb521b087cc33ae4c49da" +[[package]] +name = "utf-8" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09cc8ee72d2a9becf2f2febe0205bbed8fc6615b7cb429ad062dc7b7ddd036a9" + [[package]] name = "utf8_iter" version = "1.0.4" diff --git a/Cargo.toml b/Cargo.toml index 0820085..46dfb10 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,7 +7,7 @@ members = [ resolver = "3" [workspace.package] -version = "1.1.6" +version = "1.1.7" [workspace.dependencies] serde = { version = "1.0", features = ["derive"] } diff --git a/crates/biliup-cli/Cargo.toml b/crates/biliup-cli/Cargo.toml index 2629fb2..10fe301 100644 --- a/crates/biliup-cli/Cargo.toml +++ b/crates/biliup-cli/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "biliup-cli" -version = "0.2.4" +version.workspace = true edition = "2024" description = "Upload video to bilibili." license = "MIT OR Apache-2.0" @@ -33,7 +33,7 @@ chrono = { workspace = true, features = ["serde"] } indexmap = { version = "2.8.0" } async-trait = { workspace = true } humantime = "2.2.0" -axum = { version = "0.8", features = ["macros"] } +axum = { version = "0.8", features = ["macros", "ws"] } tower-http = { version = "0.6.2", features = ["cors"] } reqwest = { workspace = true } sqlx = { version = "0.8", features = [ "runtime-tokio-rustls", "sqlite" ] } diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs index 43fc132..8291e4c 100644 --- a/crates/biliup-cli/src/server/api/endpoints.rs +++ b/crates/biliup-cli/src/server/api/endpoints.rs @@ -1,4 +1,5 @@ use crate::server::config::Config; +use crate::server::core::download_manager::ActorHandle; use crate::server::errors::{AppError, report_to_response}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::context::Worker; @@ -11,19 +12,26 @@ use crate::server::infrastructure::repositories::{ del_streamer, get_all_streamer, get_upload_config, }; use crate::server::infrastructure::service_register::ServiceRegister; -use axum::extract::{Path, State}; +use axum::extract::ws::{Message, Utf8Bytes, WebSocket}; +use axum::extract::{Path, Query, State, WebSocketUpgrade}; use axum::http::StatusCode; use axum::response::{IntoResponse, Response}; use axum::{Json, debug_handler}; use biliup::credential::Credential; use error_stack::{Report, ResultExt}; use ormlite::{Insert, Model}; +use serde::Deserialize; use serde_json::json; +use std::collections::VecDeque; +use std::io; +use std::io::ErrorKind; +use std::path::PathBuf; use std::sync::{Arc, RwLock}; use std::time::{Duration, UNIX_EPOCH}; use tokio::fs; -use tokio::io::AsyncWriteExt; -use tracing::info; +use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader}; +use tokio::time::{MissedTickBehavior, interval}; +use tracing::{debug, error, info}; pub async fn get_streamers_endpoint( State(pool): State, @@ -411,7 +419,7 @@ pub async fn get_videos() -> Result>, Response> { let mut index = 1; // **use tokio::fs::read_dir** - if let Ok(mut entries) = fs::read_dir("../../../../..").await { + if let Ok(mut entries) = fs::read_dir(".").await { while let Ok(Some(entry)) = entries.next_entry().await { let path = entry.path(); let file_name = entry.file_name().to_string_lossy().into_owned(); @@ -445,3 +453,270 @@ pub async fn get_videos() -> Result>, Response> { } Ok(Json(file_list)) } + +pub async fn get_status( + State(service_register): State, + State(workers): State>>>>, + State(config): State>>, + State(actor_handle): State>, +) -> Result, Response> { + let mut sw = Vec::new(); + for worker in workers.read().unwrap().iter() { + sw.push(serde_json::json!({ + "downloader_status": format!("{:?}", worker.downloader_status.read().unwrap()), + "uploader_status": format!("{:?}", worker.uploader_status.read().unwrap()), + "live_streamer": worker.live_streamer, + "upload_streamer": worker.upload_streamer, + })) + } + + Ok(Json(serde_json::json!({ + "version": env!("CARGO_PKG_VERSION"), + "rooms": sw, + "download_semaphore": actor_handle.d_kills.len(), + "update_semaphore": actor_handle.u_kills.len(), + "config": config, + }))) +} + +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 dd3e58e..3c97d24 100644 --- a/crates/biliup-cli/src/server/app.rs +++ b/crates/biliup-cli/src/server/app.rs @@ -2,11 +2,13 @@ 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::errors::{AppError, AppResult}; use crate::server::infrastructure::service_register::ServiceRegister; use crate::server::infrastructure::users::Backend; use axum::http::HeaderValue; +use axum::routing::get; use axum_login::{AuthManagerLayerBuilder, login_required}; use error_stack::ResultExt; use std::net::SocketAddr; @@ -75,6 +77,7 @@ impl ApplicationController { .allow_origin("http://localhost:3000".parse::().unwrap()) .allow_methods(AllowMethods::any()), ) + .route("/v1/ws/logs", get(ws_logs)) // 获取视频列表 .fallback(static_handler); // 静态文件处理回退 // 启动HTTP服务器 diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index c0d5147..8fa85db 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -84,6 +84,15 @@ pub struct FileValidator { check_format: bool, } +impl FileValidator { + pub fn new(min_size: u64, check_format: bool) -> Self { + Self { + min_size, + check_format, + } + } +} + impl Default for FileValidator { fn default() -> Self { Self { @@ -148,8 +157,11 @@ impl SegmentEventProcessor { tx, rx, uploader, + file_validator: FileValidator::new( + worker.clone().config.read().unwrap().filtering_threshold * 1000 * 1000, + true, + ), worker, - file_validator: FileValidator::default(), } } diff --git a/crates/biliup-cli/src/server/core/download_manager.rs b/crates/biliup-cli/src/server/core/download_manager.rs index e90c240..0a87388 100644 --- a/crates/biliup-cli/src/server/core/download_manager.rs +++ b/crates/biliup-cli/src/server/core/download_manager.rs @@ -188,9 +188,9 @@ pub struct ActorHandle { /// 下载消息发送器 pub down_sender: Sender, /// 下载Actor任务句柄列表 - d_kills: Vec>, + pub(crate) d_kills: Vec>, /// 上传Actor任务句柄列表 - u_kills: Vec>, + pub(crate) u_kills: Vec>, } impl ActorHandle { diff --git a/crates/biliup-cli/src/server/router.rs b/crates/biliup-cli/src/server/router.rs index 2e550b6..d895319 100644 --- a/crates/biliup-cli/src/server/router.rs +++ b/crates/biliup-cli/src/server/router.rs @@ -3,10 +3,10 @@ use crate::server::api::bilibili_endpoints::{ }; use crate::server::api::endpoints::{ add_upload_streamer_endpoint, add_user_endpoint, delete_streamers_endpoint, - delete_template_endpoint, delete_user_endpoint, get_configuration, get_qrcode, + delete_template_endpoint, delete_user_endpoint, get_configuration, get_qrcode, get_status, get_streamer_info, get_streamers_endpoint, get_upload_streamer_endpoint, get_upload_streamers_endpoint, get_users_endpoint, get_videos, login_by_qrcode, - post_streamers_endpoint, put_configuration, put_streamers_endpoint, + post_streamers_endpoint, put_configuration, put_streamers_endpoint, ws_logs, }; use crate::server::infrastructure::service_register::ServiceRegister; use axum::Router; @@ -50,5 +50,6 @@ pub fn router(service_register: ServiceRegister) -> Router<()> { .route("/v1/login_by_qrcode", post(login_by_qrcode)) // 二维码登录 // 视频文件管理路由 .route("/v1/videos", get(get_videos)) // 获取视频列表 + .route("/v1/status", get(get_status)) .with_state(service_register) // 注入服务注册器状态 } diff --git a/crates/biliup/src/uploader/bilibili.rs b/crates/biliup/src/uploader/bilibili.rs index 6e70025..4580ecb 100644 --- a/crates/biliup/src/uploader/bilibili.rs +++ b/crates/biliup/src/uploader/bilibili.rs @@ -6,16 +6,14 @@ use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use std::collections::HashMap; +use bon::Builder; use std::fmt::{Display, Formatter}; use std::num::ParseIntError; use std::str::FromStr; use std::time::Duration; -use bon::Builder; use tracing::{info, warn}; - -#[derive(Serialize, Deserialize, Debug)] -#[derive(Builder)] +#[derive(Serialize, Deserialize, Debug, Builder)] #[cfg_attr(feature = "cli", derive(clap::Args))] pub struct Studio { /// 是否转载, 1-自制 2-转载 diff --git a/crates/stream-gears/src/lib.rs b/crates/stream-gears/src/lib.rs index 0f278bd..9427933 100644 --- a/crates/stream-gears/src/lib.rs +++ b/crates/stream-gears/src/lib.rs @@ -45,10 +45,13 @@ fn init_tracing_with_rotation() { // 按日期滚动，每天创建新文件 let file_appender = tracing_appender::rolling::RollingFileAppender::builder() .rotation(Rotation::DAILY) // rotate log files once every hour + .rotation(Rotation::NEVER) // rotate log files once every hour .filename_prefix("biliup") // log file names will be prefixed with `myapp.` + .filename_prefix("download") // log file names will be prefixed with `myapp.` .filename_suffix("log") // log file names will be suffixed with `.log` - .max_log_files(3) - .build("logs") // try to build an appender that stores log files in `/var/log` + // .max_log_files(3) + // .build("logs") // try to build an appender that stores log files in `/var/log` + .build("") // try to build an appender that stores log files in `/var/log` .expect("initializing rolling file appender failed"); // 或者按小时滚动 // let file_appender = tracing_appender::rolling::hourly("logs", "upload.log"); diff --git a/crates/stream-gears/src/uploader.rs b/crates/stream-gears/src/uploader.rs index 7b5d1e2..08ffbfc 100644 --- a/crates/stream-gears/src/uploader.rs +++ b/crates/stream-gears/src/uploader.rs @@ -9,12 +9,12 @@ use futures::StreamExt; use pyo3::prelude::*; use pyo3::pyclass; +use bon::Builder; use error_stack::ResultExt; use std::collections::HashMap; use std::path::PathBuf; use std::str::FromStr; use std::time::Instant; -use bon::Builder; use tracing::info; use biliup_cli::server::errors::{AppError, AppResult};