Add WebSocket log streaming and status endpoint

Introduces a WebSocket endpoint for real-time log streaming and a new /v1/status API for server status. Refactors log file handling, updates dependencies (adds ws feature to axum), and improves file validation logic. Also updates workspace and crate versions to 1.1.7, and makes minor improvements to log rotation and struct derivations.

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"

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"] }

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" ] }

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<ConnectionPool>,
@@ -411,7 +419,7 @@ pub async fn get_videos() -> Result<Json<Vec<serde_json::Value>>, 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<Json<Vec<serde_json::Value>>, Response> {
     }
     Ok(Json(file_list))
 }
+
+pub async fn get_status(
+    State(service_register): State<ServiceRegister>,
+    State(workers): State<Arc<RwLock<Vec<Arc<Worker>>>>>,
+    State(config): State<Arc<RwLock<Config>>>,
+    State(actor_handle): State<Arc<ActorHandle>>,
+) -> Result<Json<serde_json::Value>, 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<String>,
+}
+pub async fn ws_logs(
+    ws: WebSocketUpgrade,
+    Query(query): Query<LogsQuery>,
+) -> 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<u64> {
+    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<String> = 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<PathBuf> {
+    // 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())
+}

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::<HeaderValue>().unwrap())
                     .allow_methods(AllowMethods::any()),
             )
+            .route("/v1/ws/logs", get(ws_logs)) // 获取视频列表
             .fallback(static_handler); // 静态文件处理回退
 
         // 启动HTTP服务器

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(),
         }
     }
 

crates/biliup-cli/src/server/core/download_manager.rs

@@ -188,9 +188,9 @@ pub struct ActorHandle {
     /// 下载消息发送器
     pub down_sender: Sender<DownloaderMessage>,
     /// 下载Actor任务句柄列表
-    d_kills: Vec<JoinHandle<()>>,
+    pub(crate) d_kills: Vec<JoinHandle<()>>,
     /// 上传Actor任务句柄列表
-    u_kills: Vec<JoinHandle<()>>,
+    pub(crate) u_kills: Vec<JoinHandle<()>>,
 }
 
 impl ActorHandle {

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) // 注入服务注册器状态
 }

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-转载

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");

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};