{item.remark || item.url}
+
+ {avatarSrc ?
{info?.title ? (
{item.remark || item.url}
+ {info.title}
) : null}
@@ -419,6 +434,22 @@ export default function StreamersPage() {
{platformName(item.url)}
-
- {name}
+ >,
+}
+
+impl Default for ImageProxy {
+ fn default() -> Self {
+ Self::new()
+ }
+}
+
+impl ImageProxy {
+ pub fn new() -> Self {
+ Self {
+ client: image_client(),
+ cache: Mutex::new(HashMap::new()),
+ }
+ }
+
+ fn cached(&self, url: &str, now: Instant) -> Option {
+ let cache = self.cache.lock().unwrap();
+ cache
+ .get(url)
+ .filter(|image| now.duration_since(image.fetched_at) < CACHE_TTL)
+ .cloned()
+ }
+
+ fn store(&self, url: String, image: CachedImage, now: Instant) {
+ let mut cache = self.cache.lock().unwrap();
+ cache.retain(|_, image| now.duration_since(image.fetched_at) < CACHE_TTL);
+ if cache.len() >= MAX_CACHE_ENTRIES
+ && let Some(oldest) = cache
+ .iter()
+ .min_by_key(|(_, image)| image.fetched_at)
+ .map(|(url, _)| url.clone())
+ {
+ cache.remove(&oldest);
+ }
+ cache.insert(url, image);
+ }
+
+ /// 取图：命中缓存直接返回，否则带 Referer 去上游拉一次并缓存。
+ async fn fetch(
+ &self,
+ url: &str,
+ referer: &str,
+ kind: LiveImage,
+ ) -> Result {
+ let url = upstream_url(url, kind);
+ let now = Instant::now();
+ if let Some(image) = self.cached(&url, now) {
+ return Ok(image);
+ }
+ let FetchedImage { content_type, body } = fetch_image(&self.client, &url, referer).await?;
+ let image = CachedImage {
+ fetched_at: Instant::now(),
+ content_type,
+ body,
+ };
+ self.store(url, image.clone(), image.fetched_at);
+ Ok(image)
+ }
+}
+
+/// `GET /v1/streamers/{id}/cover`：正在录制的直播间封面，服务端转发并短暂缓存。
+pub async fn get_live_cover(
+ State(managers): State>,
+ State(proxy): State>,
+ Path(id): Path,
+) -> Response {
+ let worker = managers.get_room_by_id(id).await;
+ let (url, referer) = match resolve_target(worker.as_deref(), LiveImage::Cover) {
+ Ok(target) => target,
+ Err(rejection) => return rejection.into_response(),
+ };
+ match proxy.fetch(&url, &referer, LiveImage::Cover).await {
+ Ok(image) => image_response(&image.content_type, image.body, "private, max-age=300"),
+ Err(reason) => {
+ warn!(url, referer, reason, "拉取直播间封面失败");
+ (StatusCode::BAD_GATEWAY, format!("拉取封面失败: {reason}")).into_response()
+ }
+ }
+}
+
+/// `GET /v1/streamers/{id}/avatar`：主播头像，读 `data/avatar/` 里的本地文件。
+///
+/// 正在录制且本地还没有这个地址的头像时，现场下载一次落盘；下载失败退回旧文件。
+/// 不在录制时只读本地，没有就 404。
+pub async fn get_live_avatar(
+ State(managers): State>,
+ Path(id): Path,
+) -> Response {
+ let Some(worker) = managers.get_room_by_id(id).await else {
+ return (StatusCode::NOT_FOUND, "直播间不存在").into_response();
+ };
+ serve_avatar(&worker, avatar_store()).await
+}
+
+async fn serve_avatar(worker: &Worker, store: &AvatarStore) -> Response {
+ let id = worker.live_streamer.id;
+ let stored = match resolve_target(Some(worker), LiveImage::Avatar) {
+ Ok((url, referer)) => match store.ensure(id, &url, &referer).await {
+ Ok(stored) => Some(stored),
+ Err(reason) => {
+ warn!(id, url, reason, "下载主播头像失败，尝试用本地旧文件");
+ store.load(id).await
+ }
+ },
+ Err(_) => store.load(id).await,
+ };
+ let Some(stored) = stored else {
+ return (StatusCode::NOT_FOUND, "该直播间没有可用的头像").into_response();
+ };
+ match tokio::fs::read(store.image_path(&stored)).await {
+ Ok(bytes) => image_response(
+ &stored.content_type,
+ Bytes::from(bytes),
+ "private, max-age=3600",
+ ),
+ Err(e) => {
+ warn!(id, error = %e, "读取本地头像失败");
+ (StatusCode::INTERNAL_SERVER_ERROR, "读取本地头像失败").into_response()
+ }
+ }
+}
+
+/// 从正在录制的 worker 里取出图片地址，以及请求它时要带的 Referer（直播间页面地址）。
+fn resolve_target(
+ worker: Option<&Worker>,
+ kind: LiveImage,
+) -> Result<(String, String), (StatusCode, String)> {
+ let Some(worker) = worker else {
+ return Err((StatusCode::NOT_FOUND, "直播间不存在".to_string()));
+ };
+ let media = match &*worker.downloader_status.read().unwrap() {
+ WorkerStatus::Working(task) => task.live_media(),
+ _ => return Err((StatusCode::NOT_FOUND, "直播间未在录制".to_string())),
+ };
+ let url = match kind {
+ LiveImage::Cover => media.cover_url,
+ LiveImage::Avatar => media.avatar_url,
+ };
+ let Some(url) = url.filter(|url| url.starts_with("http://") || url.starts_with("https://"))
+ else {
+ return Err((
+ StatusCode::NOT_FOUND,
+ format!("该直播间没有可用的{}", kind.label()),
+ ));
+ };
+ Ok((url, worker.live_streamer.url.clone()))
+}
+
+fn image_response(content_type: &str, body: Bytes, cache_control: &'static str) -> Response {
+ let mut response = body.into_response();
+ let headers = response.headers_mut();
+ if let Ok(value) = HeaderValue::from_str(content_type) {
+ headers.insert(header::CONTENT_TYPE, value);
+ }
+ headers.insert(
+ header::CACHE_CONTROL,
+ HeaderValue::from_static(cache_control),
+ );
+ response
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use crate::server::common::download::DownloadTask;
+ use crate::server::config::Config;
+ use crate::server::core::downloader::{DownloaderRuntime, DownloaderType};
+ use crate::server::infrastructure::context::Stage;
+ use crate::server::infrastructure::models::live_streamer::LiveStreamer;
+ use axum::body::to_bytes;
+ use axum::routing::get;
+ use axum::{Router, http::HeaderMap};
+ use biliup::downloader::live::{DownloaderHint, LiveStream};
+ use std::sync::RwLock;
+
+ fn streamer(url: &str) -> LiveStreamer {
+ LiveStreamer {
+ id: 7,
+ url: url.to_string(),
+ remark: "x".to_string(),
+ filename_prefix: None,
+ time_range: None,
+ upload_streamers_id: None,
+ format: None,
+ override_cfg: None,
+ preprocessor: None,
+ segment_processor: None,
+ downloaded_processor: None,
+ postprocessor: None,
+ opt_args: None,
+ excluded_keywords: None,
+ }
+ }
+
+ fn live_stream(cover: &str, avatar: Option<&str>) -> LiveStream {
+ LiveStream {
+ name: "n".into(),
+ url: "https://live.bilibili.com/1".into(),
+ title: "t".into(),
+ date: chrono::Utc::now(),
+ live_cover_url: cover.into(),
+ avatar_url: avatar.map(str::to_string),
+ raw_stream_url: "https://cdn.example/live.flv".into(),
+ platform: "bilibili".into(),
+ stream_headers: HashMap::new(),
+ suffix: "flv".into(),
+ danmaku: None,
+ downloader_hint: DownloaderHint::StreamGears,
+ runtime_options: None,
+ }
+ }
+
+ fn idle_worker() -> Worker {
+ Worker::new(
+ streamer("https://live.bilibili.com/1"),
+ None,
+ Arc::new(RwLock::new(Config::default())),
+ Default::default(),
+ )
+ }
+
+ async fn working_worker(cover: &str, avatar: Option<&str>) -> Worker {
+ let worker = idle_worker();
+ let stream = live_stream(cover, avatar);
+ let task = DownloadTask::new(
+ DownloaderRuntime::from_type(DownloaderType::StreamGears),
+ &stream,
+ );
+ worker
+ .change_status(Stage::Download, WorkerStatus::Working(Arc::new(task)))
+ .await;
+ worker
+ }
+
+ #[tokio::test]
+ async fn only_a_recording_room_with_a_url_resolves_to_a_target() {
+ assert_eq!(
+ resolve_target(None, LiveImage::Cover)
+ .err()
+ .map(|(status, _)| status),
+ Some(StatusCode::NOT_FOUND)
+ );
+ assert_eq!(
+ resolve_target(Some(&idle_worker()), LiveImage::Cover)
+ .err()
+ .map(|(status, _)| status),
+ Some(StatusCode::NOT_FOUND)
+ );
+
+ let working = working_worker("https://i0.hdslb.com/cover.jpg", None).await;
+ assert_eq!(
+ resolve_target(Some(&working), LiveImage::Cover).unwrap(),
+ (
+ "https://i0.hdslb.com/cover.jpg".to_string(),
+ "https://live.bilibili.com/1".to_string()
+ )
+ );
+ // 没有头像 -> 404，而不是去请求一个空地址
+ assert_eq!(
+ resolve_target(Some(&working), LiveImage::Avatar)
+ .err()
+ .map(|(status, _)| status),
+ Some(StatusCode::NOT_FOUND)
+ );
+ // 非 http(s) 地址不代理
+ let odd = working_worker("data:image/png;base64,AAAA", None).await;
+ assert_eq!(
+ resolve_target(Some(&odd), LiveImage::Cover)
+ .err()
+ .map(|(status, _)| status),
+ Some(StatusCode::NOT_FOUND)
+ );
+ }
+
+ #[tokio::test]
+ async fn cover_proxy_fetches_with_the_room_page_as_referer_and_caches_by_url() {
+ let seen: Arc>>> = Arc::default();
+ let app = Router::new().route(
+ "/cover.jpg",
+ get({
+ let seen = seen.clone();
+ move |headers: HeaderMap| {
+ let seen = seen.clone();
+ async move {
+ seen.lock().unwrap().push(
+ headers
+ .get(header::REFERER)
+ .and_then(|v| v.to_str().ok())
+ .map(str::to_string),
+ );
+ // 上游没标 content-type，靠文件头识别
+ Bytes::from_static(&[0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10])
+ }
+ }
+ }),
+ );
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+
+ let proxy = ImageProxy::new();
+ let url = format!("http://{addr}/cover.jpg");
+ let referer = "https://live.bilibili.com/1";
+ let first = proxy.fetch(&url, referer, LiveImage::Cover).await.unwrap();
+ assert_eq!(first.content_type, "image/jpeg");
+ assert_eq!(first.body.len(), 6);
+ let second = proxy.fetch(&url, referer, LiveImage::Cover).await.unwrap();
+ assert_eq!(second.body, first.body);
+ assert_eq!(
+ *seen.lock().unwrap(),
+ vec![Some(referer.to_string())],
+ "第二次应命中缓存，上游只被请求一次，且带直播间页面作 Referer"
+ );
+
+ let response = image_response(
+ &first.content_type,
+ first.body.clone(),
+ "private, max-age=300",
+ );
+ assert_eq!(response.status(), StatusCode::OK);
+ assert_eq!(
+ response.headers().get(header::CONTENT_TYPE).unwrap(),
+ "image/jpeg"
+ );
+ assert_eq!(
+ response.headers().get(header::CACHE_CONTROL).unwrap(),
+ "private, max-age=300"
+ );
+ assert_eq!(
+ to_bytes(response.into_body(), usize::MAX).await.unwrap(),
+ first.body
+ );
+ }
+
+ #[tokio::test]
+ async fn non_image_upstream_responses_are_not_forwarded() {
+ let app = Router::new()
+ .route("/html", get(|| async { "403" }))
+ .route(
+ "/missing",
+ get(|| async { (StatusCode::FORBIDDEN, "denied") }),
+ );
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+
+ let proxy = ImageProxy::new();
+ let err = proxy
+ .fetch(
+ &format!("http://{addr}/html"),
+ "https://live.bilibili.com/1",
+ LiveImage::Cover,
+ )
+ .await
+ .unwrap_err();
+ assert!(err.contains("不是图片"), "{err}");
+ let err = proxy
+ .fetch(
+ &format!("http://{addr}/missing"),
+ "https://live.bilibili.com/1",
+ LiveImage::Cover,
+ )
+ .await
+ .unwrap_err();
+ assert!(err.contains("403"), "{err}");
+ }
+
+ #[test]
+ fn cache_expires_and_evicts_the_oldest_entry_when_full() {
+ let proxy = ImageProxy::new();
+ let t0 = Instant::now();
+ let image = |at: Instant| CachedImage {
+ fetched_at: at,
+ content_type: "image/png".into(),
+ body: Bytes::from_static(b"x"),
+ };
+ proxy.store("a".into(), image(t0), t0);
+ assert!(proxy.cached("a", t0 + Duration::from_secs(10)).is_some());
+ assert!(proxy.cached("a", t0 + CACHE_TTL).is_none(), "到期后失效");
+
+ for i in 0..MAX_CACHE_ENTRIES {
+ let at = t0 + Duration::from_secs(i as u64);
+ proxy.store(format!("u{i}"), image(at), at);
+ }
+ let now = t0 + Duration::from_secs(MAX_CACHE_ENTRIES as u64);
+ proxy.store("newest".into(), image(now), now);
+ let cache = proxy.cache.lock().unwrap();
+ assert!(cache.len() <= MAX_CACHE_ENTRIES);
+ assert!(!cache.contains_key("u0"), "最旧的条目被淘汰");
+ assert!(cache.contains_key("newest"));
+ }
+
+ /// 头像接口：录制中首次请求会现场下载落盘；之后（包括不在录制时）直接读本地文件
+ #[tokio::test]
+ async fn avatar_endpoint_downloads_once_then_serves_the_local_file() {
+ let hits = Arc::new(Mutex::new(0usize));
+ let app = Router::new().route(
+ "/face.png",
+ get({
+ let hits = hits.clone();
+ move || {
+ let hits = hits.clone();
+ async move {
+ *hits.lock().unwrap() += 1;
+ (
+ [("content-type", "image/png")],
+ Bytes::from_static(&[
+ 0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A, 1,
+ ]),
+ )
+ }
+ }
+ }),
+ );
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+
+ let dir = tempfile::tempdir().unwrap();
+ let store = AvatarStore::new(dir.path());
+ let url = format!("http://{addr}/face.png");
+ let working = working_worker("", Some(&url)).await;
+
+ let response = serve_avatar(&working, &store).await;
+ assert_eq!(response.status(), StatusCode::OK);
+ assert_eq!(
+ response.headers().get(header::CONTENT_TYPE).unwrap(),
+ "image/png"
+ );
+ assert_eq!(
+ to_bytes(response.into_body(), usize::MAX)
+ .await
+ .unwrap()
+ .len(),
+ 9
+ );
+ assert_eq!(*hits.lock().unwrap(), 1);
+
+ // 第二次：命中本地文件，上游不再被请求；即使房间已不在录制也能读到
+ let again = serve_avatar(&working, &store).await;
+ let offline = serve_avatar(&idle_worker(), &store).await;
+ assert_eq!(again.status(), StatusCode::OK);
+ assert_eq!(
+ offline.status(),
+ StatusCode::OK,
+ "本地已有文件时不在录制也能读到"
+ );
+ assert_eq!(*hits.lock().unwrap(), 1, "第二次不该再请求上游");
+ assert!(dir.path().join("7.png").is_file());
+ assert!(dir.path().join("7.json").is_file());
+ }
+}
diff --git a/crates/biliup-cli/src/server/api/mod.rs b/crates/biliup-cli/src/server/api/mod.rs
index 51d4c0e..43b4eee 100644
--- a/crates/biliup-cli/src/server/api/mod.rs
+++ b/crates/biliup-cli/src/server/api/mod.rs
@@ -4,6 +4,8 @@ pub mod auth;
pub mod bilibili_endpoints;
/// 通用API端点
pub mod endpoints;
+/// 录制中直播间的封面 / 头像图片代理
+pub mod live_media;
/// 单页应用静态文件处理
pub mod spa;
pub mod ws;
diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs
index 335bfcb..94bf626 100644
--- a/crates/biliup-cli/src/server/common/download.rs
+++ b/crates/biliup-cli/src/server/common/download.rs
@@ -1,5 +1,7 @@
+use crate::server::common::live_image::spawn_avatar_download;
use crate::server::common::recording_policy;
use crate::server::common::sync::SyncSession;
+use crate::server::common::throughput::{RateMeter, Sampler};
use crate::server::common::upload::UploaderMessage;
use crate::server::common::util::FileValidator;
use crate::server::core::downloader::cover_downloader;
@@ -114,16 +116,38 @@ impl SegmentEventProcessor {
}
}
+/// 正在录制的直播间的封面与头像地址，供界面透出。
+///
+/// 只放在内存里，随下载任务消亡；每次 `check_stream` 拿到新的流信息就刷新一遍。
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct LiveMedia {
+ pub cover_url: Option,
+ pub avatar_url: Option,
+}
+
+impl LiveMedia {
+ pub fn from_stream(stream: &LiveStream) -> Self {
+ let non_empty = |s: &str| (!s.trim().is_empty()).then(|| s.to_string());
+ Self {
+ cover_url: non_empty(&stream.live_cover_url),
+ avatar_url: stream.avatar_url.as_deref().and_then(non_empty),
+ }
+ }
+}
+
/// 下载任务
pub struct DownloadTask {
token: CancellationToken,
done_notify: Notify,
downloader: DownloaderRuntime,
sync_session: Option>>,
+ /// 写盘速率表；各下载器拿它的计数器句柄累加，采样任务随 `execute` 启停。
+ meter: Arc,
+ media: std::sync::RwLock,
}
impl DownloadTask {
- pub fn new(downloader: DownloaderRuntime) -> Self {
+ pub fn new(downloader: DownloaderRuntime, stream: &LiveStream) -> Self {
let sync_session = matches!(&downloader, DownloaderRuntime::Sync(_))
.then(|| Arc::new(Mutex::new(SyncSession::default())));
Self {
@@ -131,9 +155,39 @@ impl DownloadTask {
done_notify: Notify::new(),
downloader,
sync_session,
+ meter: Arc::new(RateMeter::new()),
+ media: std::sync::RwLock::new(LiveMedia::from_stream(stream)),
}
}
+ /// 最近一个滑动窗口内的写盘速率（字节/秒）。
+ ///
+ /// 边录边传与 yt-dlp 的字节不经过计数器，报告 `None` 而不是一个恒为 0 的假数。
+ pub fn bytes_per_sec(&self) -> Option {
+ match &self.downloader {
+ DownloaderRuntime::Sync(_) | DownloaderRuntime::YtDlp(_) => None,
+ _ => self.meter.bytes_per_sec(),
+ }
+ }
+
+ /// 当前直播间的封面与头像地址。
+ pub fn live_media(&self) -> LiveMedia {
+ self.media.read().unwrap().clone()
+ }
+
+ fn refresh_media(&self, ctx: &Context, stream: &LiveStream) {
+ let media = LiveMedia::from_stream(stream);
+ let changed = self.media.read().unwrap().avatar_url != media.avatar_url;
+ if changed {
+ spawn_avatar_download(
+ ctx.live_streamer().id,
+ media.avatar_url.clone(),
+ ctx.live_streamer().url.clone(),
+ );
+ }
+ *self.media.write().unwrap() = media;
+ }
+
pub(self) async fn execute(
&self,
ctx: &Context,
@@ -141,6 +195,8 @@ impl DownloadTask {
plugin: Arc,
rooms_handle: Arc,
) -> AppResult<()> {
+ // 速率采样任务与本次录制同寿命，句柄 drop 时随之停止
+ let _sampler = Sampler::spawn(self.meter.clone());
// 重试配置
let mut retry_count = 0;
let max_retries = 3; // 最大重试次数
@@ -198,6 +254,7 @@ impl DownloadTask {
stream: next_stream,
}) => {
stream = *next_stream;
+ self.refresh_media(ctx, &stream);
info!(
url = url,
"Stream is still live, preparing to retry. attempt: {}", retry_count
@@ -294,7 +351,8 @@ impl DownloadTask {
) -> AppResult {
// 获取配置和主播信息
let streamer = ctx.live_streamer();
- let download_config = ctx.download_config(stream);
+ let mut download_config = ctx.download_config(stream);
+ download_config.bytes_written = self.meter.counter();
if let crate::server::core::downloader::DownloaderRuntime::Sync(sync) = &self.downloader {
info!(
page_url = streamer.url,
@@ -413,13 +471,20 @@ pub async fn start_download_workflow(
sender: Sender,
rooms_handle: Arc,
) {
- let task = Arc::new(DownloadTask::new(downloader_runtime(
- ctx.config().downloader,
+ let task = Arc::new(DownloadTask::new(
+ downloader_runtime(ctx.config().downloader, ctx.live_stream()),
ctx.live_stream(),
- )));
+ ));
ctx.change_status(Stage::Download, WorkerStatus::Working(task.clone()))
.await;
+ // 主播头像几乎不变：开播时下载一次存到 data/avatar/，地址没变就不再下载
+ spawn_avatar_download(
+ ctx.live_streamer().id,
+ task.live_media().avatar_url,
+ ctx.live_streamer().url.clone(),
+ );
+
tokio::spawn({
let streamer_info = ctx.streamer_info();
let live_cover_url = streamer_info.live_cover_path.clone();
diff --git a/crates/biliup-cli/src/server/common/live_image.rs b/crates/biliup-cli/src/server/common/live_image.rs
new file mode 100644
index 0000000..8acfb51
--- /dev/null
+++ b/crates/biliup-cli/src/server/common/live_image.rs
@@ -0,0 +1,478 @@
+//! 直播间封面 / 主播头像的抓取与落盘。
+//!
+//! 页面直接 `
` 会带上 biliup 自己的 Referer，
+//! B 站等图片 CDN 对站外 Referer 返回 403，所以都由服务端以直播间页面地址作 Referer 去取。
+//! 封面每场直播都可能换，只在内存里短暂缓存（见 `api::live_media`）；
+//! 头像几乎不变，开播时下载一次存到 `data/avatar/`，地址变了才重新下载。
+
+use axum::http::{HeaderValue, header};
+use bytes::Bytes;
+use reqwest::Client;
+use serde::{Deserialize, Serialize};
+use std::path::{Path, PathBuf};
+use std::sync::LazyLock;
+use std::time::Duration;
+use tokio::sync::Mutex;
+use tracing::{info, warn};
+
+/// 单张图片上限；封面 / 头像通常几十到几百 KB，超出的不缓存、不落盘、不转发
+pub const MAX_IMAGE_BYTES: usize = 4 * 1024 * 1024;
+const FETCH_TIMEOUT: Duration = Duration::from_secs(15);
+const USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0 Safari/537.36";
+/// 头像落盘目录，与 `use_live_cover` 的 `data/cover` 并列
+const AVATAR_DIR: &str = "data/avatar";
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LiveImage {
+ Cover,
+ Avatar,
+}
+
+impl LiveImage {
+ pub fn label(self) -> &'static str {
+ match self {
+ LiveImage::Cover => "封面",
+ LiveImage::Avatar => "头像",
+ }
+ }
+
+ /// B 站图片 CDN 的缩放后缀（`@宽w_高h_1c` = 等比裁切填满）。
+ /// 卡片上的封面按 16:9 缩略图显示，头像只有 22px，用不到原图。
+ fn bfs_size_suffix(self) -> &'static str {
+ match self {
+ LiveImage::Cover => "@640w_360h_1c.jpg",
+ LiveImage::Avatar => "@128w_128h_1c.jpg",
+ }
+ }
+}
+
+/// 向上游请求时实际使用的地址。
+///
+/// B 站的 `*.hdslb.com` 原图可能有上千万像素、几 MB（实测 11520×8640、4 MB），
+/// 直接转发给页面既慢又占缓存，借 CDN 的缩放后缀取一张缩略图；已经带后缀的地址不再处理。
+/// 其它平台原样使用。
+pub fn upstream_url(url: &str, kind: LiveImage) -> String {
+ let is_bfs = url::Url::parse(url)
+ .ok()
+ .and_then(|parsed| parsed.host_str().map(|host| host.ends_with(".hdslb.com")))
+ .unwrap_or(false);
+ if is_bfs && !url.contains('@') && !url.contains('?') {
+ format!("{url}{}", kind.bfs_size_suffix())
+ } else {
+ url.to_string()
+ }
+}
+
+/// 图片抓取用的 HTTP 客户端：浏览器 UA、15 秒总超时
+pub fn image_client() -> Client {
+ Client::builder()
+ .user_agent(USER_AGENT)
+ .timeout(FETCH_TIMEOUT)
+ .build()
+ .expect("reqwest client for live images")
+}
+
+#[derive(Debug, Clone)]
+pub struct FetchedImage {
+ pub content_type: String,
+ pub body: Bytes,
+}
+
+/// 带 Referer 拉一张图。非 2xx、超过大小上限、上游不是图片都算失败。
+pub async fn fetch_image(
+ client: &Client,
+ url: &str,
+ referer: &str,
+) -> Result {
+ let mut request = client.get(url);
+ if let Ok(referer) = HeaderValue::from_str(referer) {
+ request = request.header(header::REFERER, referer);
+ }
+ let response = request.send().await.map_err(|e| e.to_string())?;
+ let status = response.status();
+ if !status.is_success() {
+ return Err(format!("上游返回 {status}"));
+ }
+ if response
+ .content_length()
+ .is_some_and(|len| len > MAX_IMAGE_BYTES as u64)
+ {
+ return Err(format!("图片超过 {MAX_IMAGE_BYTES} 字节"));
+ }
+ let upstream_type = response
+ .headers()
+ .get(header::CONTENT_TYPE)
+ .and_then(|value| value.to_str().ok())
+ .map(|value| value.to_string());
+ let body = response.bytes().await.map_err(|e| e.to_string())?;
+ if body.len() > MAX_IMAGE_BYTES {
+ return Err(format!("图片超过 {MAX_IMAGE_BYTES} 字节"));
+ }
+ let content_type = image_content_type(upstream_type.as_deref(), &body)
+ .ok_or_else(|| "上游返回的不是图片".to_string())?;
+ Ok(FetchedImage { content_type, body })
+}
+
+/// 上游标了 `image/*` 就用上游的；否则按文件头识别常见格式，都认不出就拒绝。
+pub fn image_content_type(upstream: Option<&str>, body: &[u8]) -> Option {
+ if let Some(upstream) = upstream {
+ let essence = upstream.split(';').next().unwrap_or_default().trim();
+ if essence.starts_with("image/") {
+ return Some(essence.to_string());
+ }
+ }
+ sniff_image_type(body).map(str::to_string)
+}
+
+fn sniff_image_type(body: &[u8]) -> Option<&'static str> {
+ if body.starts_with(&[0xFF, 0xD8, 0xFF]) {
+ Some("image/jpeg")
+ } else if body.starts_with(&[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A]) {
+ Some("image/png")
+ } else if body.starts_with(b"GIF87a") || body.starts_with(b"GIF89a") {
+ Some("image/gif")
+ } else if body.len() >= 12 && &body[0..4] == b"RIFF" && &body[8..12] == b"WEBP" {
+ Some("image/webp")
+ } else if body.len() >= 12 && &body[4..8] == b"ftyp" && &body[8..12] == b"avif" {
+ Some("image/avif")
+ } else {
+ None
+ }
+}
+
+/// 文件扩展名按 Content-Type 定，认不出的存成 `.img`（响应时仍按记录的 Content-Type 返回）
+fn extension_for(content_type: &str) -> &'static str {
+ match content_type {
+ "image/jpeg" => "jpg",
+ "image/png" => "png",
+ "image/gif" => "gif",
+ "image/webp" => "webp",
+ "image/avif" => "avif",
+ _ => "img",
+ }
+}
+
+/// 已落盘的头像：图片文件旁有一份同名 `.json` 记录来源地址与类型，
+/// 服务重启后据此判断要不要重新下载。
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct StoredAvatar {
+ /// 平台给的原始地址（不含 B 站缩放后缀），变了才重新下载
+ pub url: String,
+ pub content_type: String,
+ /// 图片文件名（相对头像目录）
+ pub file: String,
+}
+
+/// 主播头像的本地存储：`/<主播 id>.` + `/<主播 id>.json`。
+pub struct AvatarStore {
+ dir: PathBuf,
+ client: Client,
+ /// 同一时刻只让一个下载在写盘，避免开播时的预下载和页面首次请求撞车
+ write_lock: Mutex<()>,
+}
+
+/// 进程内共用的头像存储，录制流程与 Web 接口都用它。
+pub fn avatar_store() -> &'static AvatarStore {
+ static STORE: LazyLock = LazyLock::new(|| AvatarStore::new(AVATAR_DIR));
+ &STORE
+}
+
+impl AvatarStore {
+ pub fn new(dir: impl Into) -> Self {
+ Self {
+ dir: dir.into(),
+ client: image_client(),
+ write_lock: Mutex::new(()),
+ }
+ }
+
+ fn meta_path(&self, id: i64) -> PathBuf {
+ self.dir.join(format!("{id}.json"))
+ }
+
+ /// 已落盘且文件确实存在的头像。
+ pub async fn load(&self, id: i64) -> Option {
+ let text = tokio::fs::read_to_string(self.meta_path(id)).await.ok()?;
+ let stored: StoredAvatar = serde_json::from_str(&text).ok()?;
+ tokio::fs::metadata(self.dir.join(&stored.file))
+ .await
+ .ok()
+ .filter(|meta| meta.is_file())?;
+ Some(stored)
+ }
+
+ pub fn image_path(&self, stored: &StoredAvatar) -> PathBuf {
+ self.dir.join(&stored.file)
+ }
+
+ /// 确保本地有 `url` 对应的头像：已存且地址相同就直接用；否则带 Referer 下载后写盘。
+ /// 下载失败时返回错误，磁盘上的旧文件原样保留。
+ pub async fn ensure(&self, id: i64, url: &str, referer: &str) -> Result {
+ if !(url.starts_with("http://") || url.starts_with("https://")) {
+ return Err("头像地址不是 http(s)".to_string());
+ }
+ if let Some(stored) = self.load(id).await
+ && stored.url == url
+ {
+ return Ok(stored);
+ }
+ let _guard = self.write_lock.lock().await;
+ // 拿到锁后再看一次，前一个持锁者可能刚刚下完同一张
+ if let Some(stored) = self.load(id).await
+ && stored.url == url
+ {
+ return Ok(stored);
+ }
+ let image =
+ fetch_image(&self.client, &upstream_url(url, LiveImage::Avatar), referer).await?;
+ let stored = StoredAvatar {
+ url: url.to_string(),
+ content_type: image.content_type.clone(),
+ file: format!("{id}.{}", extension_for(&image.content_type)),
+ };
+ let previous = self.load(id).await;
+ self.write(id, &stored, &image.body)
+ .await
+ .map_err(|e| format!("写入头像失败: {e}"))?;
+ if let Some(previous) = previous
+ && previous.file != stored.file
+ {
+ let _ = tokio::fs::remove_file(self.dir.join(previous.file)).await;
+ }
+ info!(id, url, file = %self.image_path(&stored).display(), "主播头像已保存到本地");
+ Ok(stored)
+ }
+
+ /// 先写临时文件再改名，避免页面读到半张图或半份 json
+ async fn write(&self, id: i64, stored: &StoredAvatar, body: &[u8]) -> std::io::Result<()> {
+ tokio::fs::create_dir_all(&self.dir).await?;
+ write_atomically(&self.image_path(stored), body).await?;
+ write_atomically(
+ &self.meta_path(id),
+ serde_json::to_string_pretty(stored)
+ .map_err(std::io::Error::other)?
+ .as_bytes(),
+ )
+ .await
+ }
+}
+
+async fn write_atomically(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
+ let tmp = path.with_extension(format!(
+ "{}.tmp",
+ path.extension()
+ .and_then(|e| e.to_str())
+ .unwrap_or_default()
+ ));
+ tokio::fs::write(&tmp, bytes).await?;
+ tokio::fs::rename(&tmp, path).await
+}
+
+/// 开播 / 换流时在后台把头像存到本地。失败只记日志，不影响录制；页面请求时还会再试一次。
+pub fn spawn_avatar_download(id: i64, url: Option, referer: String) {
+ let Some(url) = url else { return };
+ tokio::spawn(async move {
+ if let Err(reason) = avatar_store().ensure(id, &url, &referer).await {
+ warn!(id, url, reason, "下载主播头像失败");
+ }
+ });
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use axum::routing::get;
+ use axum::{Router, http::HeaderMap};
+ use std::sync::{Arc, Mutex as StdMutex};
+
+ #[test]
+ fn bilibili_bfs_urls_get_a_thumbnail_suffix_and_others_pass_through() {
+ assert_eq!(
+ upstream_url(
+ "https://i0.hdslb.com/bfs/live/user_cover/abc.jpg",
+ LiveImage::Cover
+ ),
+ "https://i0.hdslb.com/bfs/live/user_cover/abc.jpg@640w_360h_1c.jpg"
+ );
+ assert_eq!(
+ upstream_url("https://i1.hdslb.com/bfs/face/def.jpg", LiveImage::Avatar),
+ "https://i1.hdslb.com/bfs/face/def.jpg@128w_128h_1c.jpg"
+ );
+ // 已带缩放参数的不重复追加
+ assert_eq!(
+ upstream_url(
+ "https://i0.hdslb.com/bfs/live/abc.jpg@100w_100h.jpg",
+ LiveImage::Cover
+ ),
+ "https://i0.hdslb.com/bfs/live/abc.jpg@100w_100h.jpg"
+ );
+ // 其它平台（含签名 query 的抖音、虎牙截图）原样使用
+ for url in [
+ "https://p3-webcast.douyinpic.com/img/x~tplv-obj.image",
+ "https://tx-live-cover.msstatic.com/huyalive/a/20260922.jpg?sign=abc",
+ "https://rpic.douyucdn.cn/asrpic/260922/1_src.avif/dy4",
+ ] {
+ assert_eq!(upstream_url(url, LiveImage::Cover), url);
+ }
+ }
+
+ #[test]
+ fn content_type_prefers_upstream_image_type_then_sniffs() {
+ assert_eq!(
+ image_content_type(Some("image/webp; charset=binary"), b"junk").as_deref(),
+ Some("image/webp")
+ );
+ assert_eq!(
+ image_content_type(
+ Some("application/octet-stream"),
+ &[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A]
+ )
+ .as_deref(),
+ Some("image/png")
+ );
+ let avif = [
+ 0u8, 0, 0, 0x1c, b'f', b't', b'y', b'p', b'a', b'v', b'i', b'f',
+ ];
+ assert_eq!(
+ image_content_type(None, &avif).as_deref(),
+ Some("image/avif")
+ );
+ assert_eq!(image_content_type(Some("text/html"), b""), None);
+ }
+
+ /// (请求路径, Referer) 记录
+ type Seen = Arc)>>>;
+
+ /// 一个记下每次请求 Referer 的本地图片服务；`/broken` 返回 403
+ async fn image_server() -> (String, Seen) {
+ let seen: Seen = Arc::default();
+ let record = {
+ let seen = seen.clone();
+ move |path: &str, headers: &HeaderMap| {
+ seen.lock().unwrap().push((
+ path.to_string(),
+ headers
+ .get(header::REFERER)
+ .and_then(|v| v.to_str().ok())
+ .map(str::to_string),
+ ));
+ }
+ };
+ let jpeg = Bytes::from_static(&[0xFF, 0xD8, 0xFF, 0xE0, 0x00, 0x10, 0x4A, 0x46]);
+ let png = Bytes::from_static(&[0x89, b'P', b'N', b'G', 0x0D, 0x0A, 0x1A, 0x0A, 0, 0]);
+ let app = Router::new()
+ .route(
+ "/a.jpg",
+ get({
+ let record = record.clone();
+ let jpeg = jpeg.clone();
+ move |headers: HeaderMap| {
+ let (record, jpeg) = (record.clone(), jpeg.clone());
+ async move {
+ record("/a.jpg", &headers);
+ jpeg
+ }
+ }
+ }),
+ )
+ .route(
+ "/b.png",
+ get({
+ let record = record.clone();
+ move |headers: HeaderMap| {
+ let (record, png) = (record.clone(), png.clone());
+ async move {
+ record("/b.png", &headers);
+ ([("content-type", "image/png")], png)
+ }
+ }
+ }),
+ )
+ .route(
+ "/broken",
+ get(move |headers: HeaderMap| {
+ let record = record.clone();
+ async move {
+ record("/broken", &headers);
+ (axum::http::StatusCode::FORBIDDEN, "denied")
+ }
+ }),
+ );
+ let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+ let addr = listener.local_addr().unwrap();
+ tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
+ (format!("http://{addr}"), seen)
+ }
+
+ #[tokio::test]
+ async fn avatar_is_downloaded_once_with_referer_and_survives_a_restart() {
+ let (base, seen) = image_server().await;
+ let dir = tempfile::tempdir().unwrap();
+ let store = AvatarStore::new(dir.path());
+ let referer = "https://live.bilibili.com/1";
+
+ let stored = store
+ .ensure(7, &format!("{base}/a.jpg"), referer)
+ .await
+ .unwrap();
+ assert_eq!(stored.file, "7.jpg");
+ assert_eq!(stored.content_type, "image/jpeg");
+ assert!(dir.path().join("7.jpg").is_file());
+ assert!(dir.path().join("7.json").is_file());
+ assert_eq!(
+ std::fs::read(dir.path().join("7.jpg")).unwrap()[..3],
+ [0xFF, 0xD8, 0xFF]
+ );
+
+ // 同一地址再来一次：不请求上游
+ let again = store
+ .ensure(7, &format!("{base}/a.jpg"), referer)
+ .await
+ .unwrap();
+ assert_eq!(again, stored);
+ assert_eq!(
+ *seen.lock().unwrap(),
+ vec![("/a.jpg".to_string(), Some(referer.to_string()))]
+ );
+
+ // 「重启」：新的 store 实例读同一目录
+ let reopened = AvatarStore::new(dir.path());
+ assert_eq!(reopened.load(7).await, Some(stored.clone()));
+ assert_eq!(reopened.load(8).await, None);
+
+ // 地址变了才重新下载；类型变了旧文件被清掉
+ let switched = reopened
+ .ensure(7, &format!("{base}/b.png"), referer)
+ .await
+ .unwrap();
+ assert_eq!(switched.file, "7.png");
+ assert!(dir.path().join("7.png").is_file());
+ assert!(!dir.path().join("7.jpg").exists(), "旧扩展名的文件应被删除");
+ assert_eq!(seen.lock().unwrap().len(), 2);
+ }
+
+ #[tokio::test]
+ async fn a_failed_download_keeps_the_previous_avatar() {
+ let (base, _seen) = image_server().await;
+ let dir = tempfile::tempdir().unwrap();
+ let store = AvatarStore::new(dir.path());
+ let ok = store
+ .ensure(3, &format!("{base}/a.jpg"), "https://live.bilibili.com/1")
+ .await
+ .unwrap();
+
+ let err = store
+ .ensure(3, &format!("{base}/broken"), "https://live.bilibili.com/1")
+ .await
+ .unwrap_err();
+ assert!(err.contains("403"), "{err}");
+ assert_eq!(store.load(3).await, Some(ok), "失败时保留旧头像");
+
+ assert!(
+ store
+ .ensure(3, "data:image/png;base64,AAAA", "x")
+ .await
+ .is_err()
+ );
+ }
+}
diff --git a/crates/biliup-cli/src/server/common/mod.rs b/crates/biliup-cli/src/server/common/mod.rs
index c15546c..4d3e799 100644
--- a/crates/biliup-cli/src/server/common/mod.rs
+++ b/crates/biliup-cli/src/server/common/mod.rs
@@ -3,10 +3,14 @@ use std::collections::HashMap;
use std::str::FromStr;
pub mod download;
+/// 直播间封面 / 主播头像的抓取与落盘
+pub mod live_image;
/// 录制准入策略
pub mod recording_policy;
/// 边录边传（ffmpeg stdout → 流式投稿）
pub mod sync;
+/// 录制写盘速率采样
+pub mod throughput;
/// 录制时间范围判定
pub mod timerange;
pub mod upload;
diff --git a/crates/biliup-cli/src/server/common/throughput.rs b/crates/biliup-cli/src/server/common/throughput.rs
new file mode 100644
index 0000000..747ef51
--- /dev/null
+++ b/crates/biliup-cli/src/server/common/throughput.rs
@@ -0,0 +1,349 @@
+//! 录制写盘速率：进程内写盘的下载器在写出点往 [`ByteCounter`] 里累加，子进程落盘的
+//! 下载器（ffmpeg / streamlink）由 [`SubprocessProgress`] 解析其 stderr 进度行累加；
+//! [`RateMeter`] 按固定间隔采样，用滑动窗口内首尾两次采样的差值算出 bytes/s，
+//! 供 `/v1/streamers` 透出。
+//!
+//! 采样与计算都在录制热路径之外进行；写盘路径上只有一次原子加法。
+
+use biliup::downloader::util::ByteCounter;
+use std::collections::VecDeque;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+use tokio::task::JoinHandle;
+
+/// 滑动窗口长度。httpflv 只在关键帧才把缓存的 tag 批量落盘，一个 GOP 可达数秒，
+/// 窗口必须明显长于 GOP，否则速率会在 0 与峰值之间抖动。
+pub const WINDOW: Duration = Duration::from_secs(10);
+/// 采样间隔
+pub const SAMPLE_INTERVAL: Duration = Duration::from_secs(1);
+/// 最新一次采样超过这个时长没有更新（采样任务已停）时不再报告速率
+const STALE_AFTER: Duration = Duration::from_secs(3);
+
+/// 单路录制的写盘速率表。
+#[derive(Debug, Default)]
+pub struct RateMeter {
+ counter: ByteCounter,
+ samples: Mutex>,
+}
+
+impl RateMeter {
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ /// 交给下载器累加的计数器句柄。
+ pub fn counter(&self) -> ByteCounter {
+ self.counter.clone()
+ }
+
+ /// 已写盘总字节数。
+ pub fn total(&self) -> u64 {
+ self.counter.total()
+ }
+
+ /// 记录一次采样并丢掉窗口之外的旧样本。
+ pub fn sample(&self) {
+ self.sample_at(Instant::now());
+ }
+
+ fn sample_at(&self, now: Instant) {
+ let total = self.counter.total();
+ let mut samples = self.samples.lock().unwrap();
+ samples.push_back((now, total));
+ while let Some(&(t, _)) = samples.front() {
+ if now.duration_since(t) > WINDOW && samples.len() > 1 {
+ samples.pop_front();
+ } else {
+ break;
+ }
+ }
+ }
+
+ /// 窗口内的平均写盘速率（字节/秒）。
+ ///
+ /// 采样不足两次、或采样已停止更新时返回 `None`，界面据此显示「—」而不是一个过期数字。
+ pub fn bytes_per_sec(&self) -> Option {
+ self.bytes_per_sec_at(Instant::now())
+ }
+
+ fn bytes_per_sec_at(&self, now: Instant) -> Option {
+ let samples = self.samples.lock().unwrap();
+ let &(newest_at, newest) = samples.back()?;
+ if now.duration_since(newest_at) > STALE_AFTER {
+ return None;
+ }
+ let &(oldest_at, oldest) = samples.front()?;
+ let elapsed = newest_at.duration_since(oldest_at);
+ if elapsed < SAMPLE_INTERVAL {
+ return None;
+ }
+ let rate = newest.saturating_sub(oldest) as f64 / elapsed.as_secs_f64();
+ Some(rate.round() as u64)
+ }
+}
+
+/// 后台按 [`SAMPLE_INTERVAL`] 采样，直到句柄被 drop（随之 abort）。
+pub struct Sampler(JoinHandle<()>);
+
+impl Sampler {
+ pub fn spawn(meter: Arc) -> Self {
+ Self(tokio::spawn(async move {
+ let mut ticker = tokio::time::interval(SAMPLE_INTERVAL);
+ ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
+ loop {
+ ticker.tick().await;
+ meter.sample();
+ }
+ }))
+ }
+}
+
+impl Drop for Sampler {
+ fn drop(&mut self) {
+ self.0.abort();
+ }
+}
+
+/// 子进程落盘的下载器（ffmpeg / streamlink）不经过本进程写盘，改为解析它们打到 stderr 的
+/// 进度行：两者都会周期性报告「本次进程累计写出多少字节」，这里把增量累加到计数器。
+/// 不读盘、不新开任务，就在既有的 stderr 读取协程里顺带解析；解析不出来只是没有速率，
+/// 不影响录制。
+#[derive(Debug, Default)]
+pub struct SubprocessProgress {
+ /// 当前子进程最近一次报告的累计字节数
+ last_total: u64,
+}
+
+/// ffmpeg `-progress` 输出的字段名（`stream_N_M_q` 另行按前后缀匹配）
+const FFMPEG_PROGRESS_KEYS: &[&str] = &[
+ "frame",
+ "fps",
+ "bitrate",
+ "total_size",
+ "out_time_us",
+ "out_time_ms",
+ "out_time",
+ "dup_frames",
+ "drop_frames",
+ "speed",
+ "progress",
+];
+
+impl SubprocessProgress {
+ /// 记录一次累计值：只加增量；回退（换了输出文件）时只重置基准，不倒扣。
+ pub fn advance_to(&mut self, total: u64, counter: &ByteCounter) {
+ if total > self.last_total {
+ counter.add(total - self.last_total);
+ }
+ self.last_total = total;
+ }
+
+ /// ffmpeg `-progress pipe:2` 的 `key=value` 行，其中 `total_size=N` 是累计写出字节。
+ /// 返回 true 表示这一行是进度行（调用方不必再当普通日志打印）。
+ pub fn observe_ffmpeg(&mut self, line: &str, counter: &ByteCounter) -> bool {
+ let Some((key, value)) = line.split_once('=') else {
+ return false;
+ };
+ let key = key.trim();
+ let is_progress_key = FFMPEG_PROGRESS_KEYS.contains(&key)
+ || (key.starts_with("stream_") && key.ends_with("_q"));
+ if !is_progress_key {
+ return false;
+ }
+ if key == "total_size"
+ && let Ok(total) = value.trim().parse::()
+ {
+ self.advance_to(total, counter);
+ }
+ true
+ }
+
+ /// streamlink `--progress=force` 的 `[download] Written 12.34 MiB to (5s @ 2.50 MiB/s)`。
+ /// 返回 true 表示这一行是进度行。
+ pub fn observe_streamlink(&mut self, line: &str, counter: &ByteCounter) -> bool {
+ let Some(total) = streamlink_written_bytes(line) else {
+ return false;
+ };
+ self.advance_to(total, counter);
+ true
+ }
+}
+
+/// 从 streamlink 进度行里取出累计写出字节数（`Written 12.34 MiB`，单位为 bytes / KiB / MiB / GiB / TiB）
+pub fn streamlink_written_bytes(line: &str) -> Option {
+ let rest = line.split_once("[download]")?.1.trim_start();
+ let rest = rest.strip_prefix("Written")?.trim_start();
+ let mut parts = rest.splitn(3, ' ');
+ let amount: f64 = parts.next()?.parse().ok()?;
+ let unit = parts.next()?;
+ let scale: f64 = match unit {
+ "bytes" | "B" => 1.0,
+ "KiB" | "KB" => 1024.0,
+ "MiB" | "MB" => 1024.0 * 1024.0,
+ "GiB" | "GB" => 1024.0 * 1024.0 * 1024.0,
+ "TiB" | "TB" => 1024.0 * 1024.0 * 1024.0 * 1024.0,
+ _ => return None,
+ };
+ Some((amount * scale).round() as u64)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn needs_two_samples_at_least_one_interval_apart() {
+ let meter = RateMeter::new();
+ let t0 = Instant::now();
+ assert_eq!(meter.bytes_per_sec_at(t0), None);
+
+ meter.counter().add(1000);
+ meter.sample_at(t0);
+ assert_eq!(meter.bytes_per_sec_at(t0), None, "只有一个样本算不出速率");
+
+ meter.counter().add(2000);
+ meter.sample_at(t0 + Duration::from_secs(2));
+ assert_eq!(
+ meter.bytes_per_sec_at(t0 + Duration::from_secs(2)),
+ Some(1000)
+ );
+ }
+
+ #[test]
+ fn rate_is_averaged_over_the_window_and_old_samples_fall_off() {
+ let meter = RateMeter::new();
+ let t0 = Instant::now();
+ // 每秒写 100 字节，持续 30 秒
+ for i in 0..=30u64 {
+ meter.sample_at(t0 + Duration::from_secs(i));
+ meter.counter().add(100);
+ }
+ let now = t0 + Duration::from_secs(30);
+ assert_eq!(meter.bytes_per_sec_at(now), Some(100));
+ assert!(
+ meter.samples.lock().unwrap().len() <= (WINDOW.as_secs() as usize) + 1,
+ "窗口之外的样本应被丢掉"
+ );
+
+ // 停写超过一个窗口后窗口内没有增量，速率归零而不是沿用旧值
+ for i in 31..=41u64 {
+ meter.sample_at(t0 + Duration::from_secs(i));
+ }
+ assert_eq!(
+ meter.bytes_per_sec_at(t0 + Duration::from_secs(41)),
+ Some(0)
+ );
+ }
+
+ #[test]
+ fn a_stale_meter_reports_nothing() {
+ let meter = RateMeter::new();
+ let t0 = Instant::now();
+ meter.sample_at(t0);
+ meter.counter().add(500);
+ meter.sample_at(t0 + Duration::from_secs(1));
+ assert_eq!(
+ meter.bytes_per_sec_at(t0 + Duration::from_secs(2)),
+ Some(500)
+ );
+ assert_eq!(
+ meter.bytes_per_sec_at(t0 + Duration::from_secs(10)),
+ None,
+ "采样任务停了之后不能继续报旧速率"
+ );
+ }
+
+ #[test]
+ fn ffmpeg_progress_lines_feed_total_size_and_are_recognised() {
+ let counter = ByteCounter::new();
+ let mut progress = SubprocessProgress::default();
+ // 实测 ffmpeg 6.1 `-progress pipe:2` 的一个进度块
+ let block = [
+ "frame=0",
+ "fps=0.00",
+ "stream_0_0_q=0.0",
+ "bitrate=N/A",
+ "total_size=N/A",
+ "out_time_us=N/A",
+ "out_time_ms=N/A",
+ "out_time=N/A",
+ "dup_frames=0",
+ "drop_frames=0",
+ "speed=N/A",
+ "progress=continue",
+ ];
+ for line in block {
+ assert!(
+ progress.observe_ffmpeg(line, &counter),
+ "{line} 应被识别为进度行"
+ );
+ }
+ assert_eq!(counter.total(), 0, "N/A 不计");
+
+ assert!(progress.observe_ffmpeg("total_size=63269", &counter));
+ assert!(progress.observe_ffmpeg("total_size=100000", &counter));
+ assert_eq!(counter.total(), 100_000);
+ // 新进程从 0 重新计：不倒扣，只重置基准
+ assert!(progress.observe_ffmpeg("total_size=5000", &counter));
+ assert_eq!(counter.total(), 100_000);
+ assert!(progress.observe_ffmpeg("total_size=7000", &counter));
+ assert_eq!(counter.total(), 102_000);
+
+ // 普通日志行不是进度行
+ for line in [
+ "ffmpeg version 6.1.1 Copyright (c) 2000-2023",
+ "[flv @ 0x55] Failed to update header with correct duration.",
+ "Input #0, flv, from 'https://x/y.flv?a=b':",
+ "",
+ ] {
+ assert!(!progress.observe_ffmpeg(line, &counter), "{line:?}");
+ }
+ }
+
+ #[test]
+ fn streamlink_progress_lines_are_parsed_in_binary_units() {
+ assert_eq!(
+ streamlink_written_bytes("[download] Written 61.78 KiB to /tmp/x.flv.part (0s)"),
+ Some(63_263)
+ );
+ assert_eq!(
+ streamlink_written_bytes(
+ "[download] Written 488.00 KiB to /tmp/x.flv.part (4s @ 243.92 KiB/s)"
+ ),
+ Some(499_712)
+ );
+ assert_eq!(
+ streamlink_written_bytes("[download] Written 1.50 MiB to a b c.ts (9s @ 1.2 MiB/s)"),
+ Some(1_572_864)
+ );
+ assert_eq!(
+ streamlink_written_bytes("[download] Written 2.00 GiB to x (1h)"),
+ Some(2 * 1024 * 1024 * 1024)
+ );
+ assert_eq!(
+ streamlink_written_bytes("[download] Written 512 bytes to x (0s)"),
+ Some(512)
+ );
+ for line in [
+ "[cli][info] Found matching plugin http for URL httpstream://...",
+ "[cli][info] Writing output to",
+ "[download] Written weird MiB to x",
+ "",
+ ] {
+ assert_eq!(streamlink_written_bytes(line), None, "{line:?}");
+ }
+
+ let counter = ByteCounter::new();
+ let mut progress = SubprocessProgress::default();
+ assert!(progress.observe_streamlink(
+ "[download] Written 240.00 KiB to /tmp/x.flv.part (2s)",
+ &counter
+ ));
+ assert!(progress.observe_streamlink(
+ "[download] Written 488.00 KiB to /tmp/x.flv.part (4s @ 243.92 KiB/s)",
+ &counter
+ ));
+ assert_eq!(counter.total(), 499_712);
+ assert!(!progress.observe_streamlink("[cli][info] Stream ended", &counter));
+ }
+}
diff --git a/crates/biliup-cli/src/server/core/downloader.rs b/crates/biliup-cli/src/server/core/downloader.rs
index 406cf91..cbf98be 100644
--- a/crates/biliup-cli/src/server/core/downloader.rs
+++ b/crates/biliup-cli/src/server/core/downloader.rs
@@ -20,6 +20,7 @@ use crate::server::core::downloader::sync_downloader::SyncDownloader;
use crate::server::core::downloader::ytdlp::YouTubeDownloader;
use crate::server::errors::{AppError, AppResult};
use async_trait::async_trait;
+use biliup::downloader::util::ByteCounter;
use danmaku_client::{DanmakuRecorder, RecorderConfig, RecorderHandle};
use error_stack::Report;
use serde::{Deserialize, Serialize};
@@ -53,6 +54,11 @@ pub struct DownloadConfig {
pub output_dir: PathBuf,
pub suffix: String,
+
+ /// 本次录制任务的写盘字节累计，由各下载器在写出点累加，供界面显示实时速率。
+ /// 只是一个原子计数器的句柄：不参与序列化，也不影响任何下载控制流。
+ #[serde(skip)]
+ pub bytes_written: ByteCounter,
}
impl DownloadConfig {
diff --git a/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs b/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs
index 297d53c..bc451cc 100644
--- a/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs
+++ b/crates/biliup-cli/src/server/core/downloader/ffmpeg_downloader.rs
@@ -1,9 +1,11 @@
+use crate::server::common::throughput::SubprocessProgress;
+use crate::server::common::util::redact_process_debug;
use crate::server::core::downloader;
use crate::server::core::downloader::{
DownloadConfig, DownloadStatus, DownloaderType, SegmentEvent, SegmentInfo,
};
use crate::server::errors::{AppError, AppResult};
-use crate::server::common::util::redact_process_debug;
+use biliup::downloader::util::ByteCounter;
use error_stack::{ResultExt, bail};
use std::path::PathBuf;
use std::process::{ExitStatus, Stdio};
@@ -11,7 +13,7 @@ use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::Command;
use tokio::sync::RwLock;
-use tracing::info;
+use tracing::{debug, info};
/// FFmpeg下载器实现
/// 使用FFmpeg进行直播流下载，支持内部和外部分段
@@ -127,6 +129,12 @@ impl FfmpegDownloader {
fn append_common_input_args(&self, args: &mut Vec, download_config: &DownloadConfig) {
args.push("-y".to_string()); // 覆盖已存在文件
+ // -progress pipe:2: 把 key=value 形式的进度（含累计写出字节 total_size）打到 stderr，
+ // 由 spawn_log 解析出写盘速率；不受 -loglevel 影响。
+ // -nostats: 关掉同样写 stderr、以 \r 刷新的单行统计，避免与进度行混在一起
+ args.extend(["-progress".to_string(), "pipe:2".to_string()]);
+ args.push("-nostats".to_string());
+
// HTTP headers
// -headers: 设置HTTP请求头，格式为"Key: Value\r\n"
// 用于传递User-Agent、Cookie等信息
@@ -200,9 +208,10 @@ impl FfmpegDownloader {
let args = self.build_ffmpeg_args_external_segment(&download_config);
let output_file = download_config.generate_output_filename(&download_config.suffix);
+ let part_file = format!("{}.part", output_file.display());
let mut cmd = Command::new("ffmpeg");
cmd.args(&args)
- .arg(format!("{}.part", output_file.display()))
+ .arg(&part_file)
.stdin(Stdio::null())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
@@ -210,9 +219,13 @@ impl FfmpegDownloader {
let child = cmd.spawn().change_context(AppError::Unknown)?;
- let status = spawn_log(child, &self.process_handle).await?;
+ let status = spawn_log(
+ child,
+ &self.process_handle,
+ download_config.bytes_written.clone(),
+ )
+ .await?;
// 退出时，重命名文件
- let part_file = format!("{}.part", output_file.display());
tokio::fs::rename(&part_file, &output_file)
.await
.change_context(AppError::Custom(String::from("退出时，重命名文件")))?;
@@ -297,7 +310,12 @@ impl FfmpegDownloader {
segment_index += 1;
prev_file_path = Some(file_path);
}
- let status = spawn_log(child, &self.process_handle).await?;
+ let status = spawn_log(
+ child,
+ &self.process_handle,
+ download_config.bytes_written.clone(),
+ )
+ .await?;
if let Some(file_path) = prev_file_path {
// 重命名文件
@@ -361,9 +379,12 @@ impl FfmpegDownloader {
// }
}
+/// 等待 ffmpeg 结束，期间把 stderr 转成日志；`-progress` 的进度行不打日志，
+/// 只把 `total_size` 的增量累加到 `bytes_written`（写盘速率的来源）。
async fn spawn_log(
mut child: tokio::process::Child,
process_handle: &RwLock>,
+ bytes_written: ByteCounter,
) -> AppResult {
let stderr = child.stderr.take().ok_or(AppError::Custom(
"failed to capture stderr pipe".to_string(),
@@ -376,9 +397,14 @@ async fn spawn_log(
}
let mut stderr_lines = BufReader::new(stderr).lines();
- // 将 stderr 打印到当前进程的 stderr
+ // 将 stderr 打印到当前进程的 stderr；进度行只解析不打印
let stderr_task = tokio::spawn(async move {
+ let mut progress = SubprocessProgress::default();
while let Ok(Some(line)) = stderr_lines.next_line().await {
+ if progress.observe_ffmpeg(&line, &bytes_written) {
+ debug!("[ffmpeg] {line}");
+ continue;
+ }
info!("[ffmpeg] {line}");
}
});
@@ -444,6 +470,21 @@ mod tests {
FfmpegDownloader::new(Vec::new(), DownloaderType::FfmpegInternal)
}
+ #[test]
+ fn both_modes_ask_ffmpeg_for_machine_readable_progress_on_stderr() {
+ for args in [
+ external().build_ffmpeg_args_external_segment(&config(None, None)),
+ internal().build_ffmpeg_args_internal_segment(&config(None, None)),
+ ] {
+ assert_eq!(value_of(&args, "-progress"), Some("pipe:2".to_string()));
+ assert!(args.contains(&"-nostats".to_string()));
+ // 全局选项必须在 -i 之前
+ let progress_at = args.iter().position(|a| a == "-progress").unwrap();
+ let input_at = args.iter().position(|a| a == "-i").unwrap();
+ assert!(progress_at < input_at);
+ }
+ }
+
#[test]
fn without_a_time_range_the_segment_time_reaches_ffmpeg_unchanged() {
let args = external().build_ffmpeg_args_external_segment(&config(Some("01:00:00"), None));
diff --git a/crates/biliup-cli/src/server/core/downloader/mesio.rs b/crates/biliup-cli/src/server/core/downloader/mesio.rs
index 5f70d6c..5c15432 100644
--- a/crates/biliup-cli/src/server/core/downloader/mesio.rs
+++ b/crates/biliup-cli/src/server/core/downloader/mesio.rs
@@ -11,6 +11,7 @@ use crate::server::core::downloader::{
self, DownloadConfig, DownloadStatus, SegmentEvent, SegmentInfo,
};
use crate::server::errors::{AppError, AppResult};
+use biliup::downloader::util::ByteCounter;
use error_stack::Report;
use flv_fix::{
ContinuityMode, FlvPipeline, FlvPipelineConfig, FlvWriter, FlvWriterConfig, ScriptFillerConfig,
@@ -154,6 +155,7 @@ impl Mesio {
seg_rx,
token.clone(),
callback.as_mut(),
+ (download_config.bytes_written.clone(), |item| item.size()),
)
.await;
handle.cancel();
@@ -206,6 +208,7 @@ impl Mesio {
seg_rx,
token.clone(),
callback.as_mut(),
+ (download_config.bytes_written.clone(), |item| item.size()),
)
.await;
handle.cancel();
@@ -335,6 +338,10 @@ fn segment_complete_hook(
///
/// 结束条件：拉流结束或被取消后 `input_tx` 关闭 → 管线排空 → writer 关闭最后一个
/// 文件并退出 → writer 被 drop 使 `seg_rx` 断开，回调排空后收尾。
+///
+/// `byte_meter`：写盘速率计数器与「一个管线条目占多少字节」的取值函数。
+/// 修复管线的 writer 来自外部 crate，拿不到逐条写出量，因此在条目进入管线前计数；
+/// 修复只增删极少量 tag，与实际落盘量相差可忽略。
#[allow(clippy::too_many_arguments)]
async fn run_pipeline<'a, P, W>(
common: &PipelineConfig,
@@ -345,6 +352,7 @@ async fn run_pipeline<'a, P, W>(
mut seg_rx: UnboundedReceiver,
token: CancellationToken,
callback: &mut (dyn FnMut(SegmentEvent) + Send + Sync + 'a),
+ byte_meter: (ByteCounter, fn(&P::Item) -> usize),
) -> Result
where
P: PipelineProvider,
@@ -361,8 +369,12 @@ where
let writer_task = tokio::task::spawn_blocking(move || writer.run(output_rx));
let forward = tokio::spawn(async move {
+ let (bytes_written, item_size) = byte_meter;
let mut items = items;
while let Some(item) = items.next().await {
+ if let Ok(item) = &item {
+ bytes_written.add(item_size(item) as u64);
+ }
if input_tx.send(item).await.is_err() {
debug!("mesio 管线已关闭，停止转发");
break;
@@ -490,6 +502,7 @@ mod tests {
payload.extend_from_slice(&data);
payload.extend_from_slice(&((11 + data.len()) as u32).to_be_bytes());
}
+ let payload_len = payload.len();
let body = bytes::Bytes::from(payload);
let app = Router::new().route(
@@ -517,7 +530,9 @@ mod tests {
recorder,
output_dir: dir.path().to_path_buf(),
suffix: "flv".to_string(),
+ bytes_written: ByteCounter::new(),
};
+ let bytes_written = config.bytes_written.clone();
let seen: Arc>> = Arc::default();
let sink = seen.clone();
@@ -534,6 +549,11 @@ mod tests {
.expect("download");
assert_eq!(status, DownloadStatus::StreamEnded);
+ assert_eq!(
+ bytes_written.total(),
+ payload_len as u64,
+ "进入管线的字节应等于源 FLV 的长度（文件头 + 全部 tag）"
+ );
let seen = seen.lock().unwrap();
assert!(
seen.len() >= 2,
diff --git a/crates/biliup-cli/src/server/core/downloader/stream_gears.rs b/crates/biliup-cli/src/server/core/downloader/stream_gears.rs
index a912bb7..db90fd8 100644
--- a/crates/biliup-cli/src/server/core/downloader/stream_gears.rs
+++ b/crates/biliup-cli/src/server/core/downloader/stream_gears.rs
@@ -96,7 +96,8 @@ impl StreamGears {
debug!("header: {header:#?}");
info!("Downloading {}...", url);
// FLV流下载
- let file = LifecycleFile::with_hook(&file_name, "flv", hook);
+ let file = LifecycleFile::with_hook(&file_name, "flv", hook)
+ .with_counter(download_config.bytes_written.clone());
httpflv::download(connection, file, segment.clone()).await;
}
Err(Err::Incomplete(needed)) => {
@@ -105,7 +106,8 @@ impl StreamGears {
Err(e) => {
error!("{e}");
// HLS流下载
- let file = LifecycleFile::with_hook(&file_name, "ts", hook);
+ let file = LifecycleFile::with_hook(&file_name, "ts", hook)
+ .with_counter(download_config.bytes_written.clone());
hls::download(&url, &client, file, segment.clone())
.await
.change_context(AppError::Unknown)?;
diff --git a/crates/biliup-cli/src/server/core/downloader/streamlink.rs b/crates/biliup-cli/src/server/core/downloader/streamlink.rs
index 10d17e9..02ea60c 100644
--- a/crates/biliup-cli/src/server/core/downloader/streamlink.rs
+++ b/crates/biliup-cli/src/server/core/downloader/streamlink.rs
@@ -1,6 +1,8 @@
-use crate::server::core::downloader::{DownloadConfig, DownloadStatus, SegmentEvent, SegmentInfo};
+use crate::server::common::throughput::SubprocessProgress;
use crate::server::common::util::redact_process_debug;
+use crate::server::core::downloader::{DownloadConfig, DownloadStatus, SegmentEvent, SegmentInfo};
use crate::server::errors::{AppError, AppResult};
+use biliup::downloader::util::ByteCounter;
use error_stack::ResultExt;
use std::collections::HashMap;
use std::process::{ExitStatus, Stdio};
@@ -9,7 +11,7 @@ use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::process::{Child, ChildStdout, Command};
use tokio::sync::RwLock;
use tokio::time::Duration;
-use tracing::info;
+use tracing::{debug, info};
use url::Url;
#[derive(Debug, Clone)]
@@ -70,7 +72,12 @@ impl Streamlink {
info!(cmd = %redact_process_debug(&cmd), "Starting streamlink download");
let child = cmd.spawn().change_context(AppError::Unknown)?;
- let status = spawn_log(child, &self.process_handle).await?;
+ let status = spawn_log(
+ child,
+ &self.process_handle,
+ download_config.bytes_written.clone(),
+ )
+ .await?;
if tokio::fs::try_exists(&part_file)
.await
@@ -164,6 +171,9 @@ impl StreamlinkDownloader {
args.push(segment_time);
}
args.push("--force".to_string());
+ // stderr 不是终端时也按周期打「[download] Written …」进度行，spawn_log 据此算写盘速率
+ args.push("--progress".to_string());
+ args.push("force".to_string());
args.push("--output".to_string());
args.push(output_file.to_string());
args.push(streamlink_cli_url(&self.url));
@@ -438,15 +448,36 @@ fn is_progressive_http_ext(ext: &str) -> bool {
)
}
+/// 等待 streamlink 结束，期间把 stdout / stderr 转成日志；`--progress=force` 的进度行
+/// 只解析不打印，把累计写出字节的增量累加到 `bytes_written`（写盘速率的来源）。
+/// `--output` 到文件时 streamlink 把控制台输出（含进度）打在 stdout，两条流都解析。
async fn spawn_log(
mut child: Child,
process_handle: &RwLock>,
+ bytes_written: ByteCounter,
) -> AppResult {
+ let progress = Arc::new(std::sync::Mutex::new(SubprocessProgress::default()));
+ let log_or_track = {
+ let progress = progress.clone();
+ move |line: String| {
+ if progress
+ .lock()
+ .unwrap()
+ .observe_streamlink(&line, &bytes_written)
+ {
+ debug!("[streamlink] {line}");
+ } else {
+ info!("[streamlink] {line}");
+ }
+ }
+ };
+
let mut stderr_task = child.stderr.take().map(|stderr| {
let mut stderr_lines = BufReader::new(stderr).lines();
+ let log_or_track = log_or_track.clone();
tokio::spawn(async move {
while let Ok(Some(line)) = stderr_lines.next_line().await {
- info!("[streamlink] {line}");
+ log_or_track(line);
}
})
});
@@ -455,7 +486,7 @@ async fn spawn_log(
let mut stdout_lines = BufReader::new(stdout).lines();
tokio::spawn(async move {
while let Ok(Some(line)) = stdout_lines.next_line().await {
- info!("[streamlink] {line}");
+ log_or_track(line);
}
})
});
@@ -567,5 +598,11 @@ mod tests {
"expected httpstream URL in args: {args:?}"
);
assert!(args.iter().any(|a| a == "best"));
+ // 写盘速率靠解析进度行，stderr 不是终端时也要让 streamlink 打出来
+ assert!(
+ args.windows(2)
+ .any(|w| w[0] == "--progress" && w[1] == "force"),
+ "expected --progress force in args: {args:?}"
+ );
}
}
diff --git a/crates/biliup-cli/src/server/infrastructure/context.rs b/crates/biliup-cli/src/server/infrastructure/context.rs
index d0d7bc7..f95f3d5 100644
--- a/crates/biliup-cli/src/server/infrastructure/context.rs
+++ b/crates/biliup-cli/src/server/infrastructure/context.rs
@@ -129,6 +129,7 @@ impl Context {
// output_dir: PathBuf::from("./downloads")
output_dir: PathBuf::from("."),
suffix,
+ bytes_written: Default::default(),
}
}
}
diff --git a/crates/biliup-cli/src/server/infrastructure/dto.rs b/crates/biliup-cli/src/server/infrastructure/dto.rs
index 04d4042..a004d69 100644
--- a/crates/biliup-cli/src/server/infrastructure/dto.rs
+++ b/crates/biliup-cli/src/server/infrastructure/dto.rs
@@ -13,4 +13,12 @@ pub struct LiveStreamerResponse {
pub status: String,
/// 上传状态
pub upload_status: String,
+
+ /// 正在录制时最近一个滑动窗口内的写盘速率（字节/秒）。
+ /// 未录制、尚无采样、或下载器不经过本进程写盘（边录边传、yt-dlp）时为 `null`。
+ pub live_bytes_per_sec: Option,
+ /// 正在录制的直播间封面地址；图片本身请走 `GET /v1/streamers/{id}/cover` 代理。
+ pub live_cover_url: Option,
+ /// 正在录制的主播头像地址；图片本身请走 `GET /v1/streamers/{id}/avatar` 代理。
+ pub live_avatar_url: Option,
}
diff --git a/crates/biliup-cli/src/server/infrastructure/service_register.rs b/crates/biliup-cli/src/server/infrastructure/service_register.rs
index b636ff1..81ef40e 100644
--- a/crates/biliup-cli/src/server/infrastructure/service_register.rs
+++ b/crates/biliup-cli/src/server/infrastructure/service_register.rs
@@ -1,4 +1,5 @@
use crate::LogHandle;
+use crate::server::api::live_media::ImageProxy;
use crate::server::config::Config;
use crate::server::core::download_manager::DownloadManager;
use crate::server::infrastructure::connection_pool::ConnectionPool;
@@ -27,6 +28,9 @@ pub struct ServiceRegister {
pub client: StatelessClient,
pub log_handle: LogHandle,
+
+ /// 录制中直播间封面 / 头像的图片代理（抓取客户端 + 内存缓存）
+ pub image_proxy: Arc,
}
/// 简单的服务容器，负责管理API端点通过axum扩展获取的各种服务
@@ -63,6 +67,7 @@ impl ServiceRegister {
config: config.clone(),
client,
log_handle,
+ image_proxy: Arc::new(ImageProxy::new()),
}
}
diff --git a/crates/biliup-cli/src/server/router.rs b/crates/biliup-cli/src/server/router.rs
index ce13285..72e7a0e 100644
--- a/crates/biliup-cli/src/server/router.rs
+++ b/crates/biliup-cli/src/server/router.rs
@@ -9,6 +9,7 @@ use crate::server::api::endpoints::{
login_by_qrcode, pause_streamers_endpoint, post_streamers_endpoint, post_uploads,
put_configuration, put_streamers_endpoint,
};
+use crate::server::api::live_media::{get_live_avatar, get_live_cover};
use crate::server::infrastructure::service_register::ServiceRegister;
use axum::Router;
use axum::body::Body;
@@ -33,6 +34,9 @@ pub fn router(service_register: ServiceRegister) -> Router<()> {
)
.route("/v1/streamers/{id}", delete(delete_streamers_endpoint)) // 删除主播
.route("/v1/streamers/{id}/pause", put(pause_streamers_endpoint))
+ // 正在录制的直播间封面 / 主播头像（服务端带 Referer 转发，避开图片 CDN 防盗链）
+ .route("/v1/streamers/{id}/cover", get(get_live_cover))
+ .route("/v1/streamers/{id}/avatar", get(get_live_avatar))
// 配置管理路由
.route(
"/v1/configuration",
diff --git a/crates/biliup/src/downloader/flv_writer.rs b/crates/biliup/src/downloader/flv_writer.rs
index 042630f..0870f1a 100644
--- a/crates/biliup/src/downloader/flv_writer.rs
+++ b/crates/biliup/src/downloader/flv_writer.rs
@@ -70,6 +70,9 @@ impl<'a> FlvFile<'a> {
self.buf_writer.write_all(body)?;
// write 允许部分写入，短写会静默丢字节并破坏 FLV 结构，必须用 write_all
self.buf_writer.write_all(previous_tag_size)?;
+ self.file
+ .bytes_written
+ .add((11 + body.len() + previous_tag_size.len()) as u64);
Ok(previous_tag_size.len())
}
diff --git a/crates/biliup/src/downloader/hls.rs b/crates/biliup/src/downloader/hls.rs
index a879f2b..279f65d 100644
--- a/crates/biliup/src/downloader/hls.rs
+++ b/crates/biliup/src/downloader/hls.rs
@@ -1,5 +1,5 @@
use crate::downloader::error::{Error, Result};
-use crate::downloader::util::{LifecycleFile, Segmentable};
+use crate::downloader::util::{ByteCounter, LifecycleFile, Segmentable};
use m3u8_rs::{MediaPlaylist, Playlist};
use std::fs::File;
@@ -100,6 +100,7 @@ pub async fn download(
media_url.join(&segment.uri)?,
client,
&mut ts_file.buf_writer,
+ &ts_file.file.bytes_written,
)
.await?;
splitting.increase_size(length);
@@ -134,13 +135,19 @@ pub async fn download(
Ok(())
}
-async fn download_to_file(url: Url, client: &StatelessClient, out: &mut impl Write) -> Result {
+async fn download_to_file(
+ url: Url,
+ client: &StatelessClient,
+ out: &mut impl Write,
+ bytes_written: &ByteCounter,
+) -> Result {
debug!("url: {url}");
let mut response = client.retryable(url.as_str()).await?;
let mut length: u64 = 0;
while let Some(chunk) = response.chunk().await? {
length += chunk.len() as u64;
out.write_all(&chunk)?;
+ bytes_written.add(chunk.len() as u64);
}
// let mut out = File::options()
// .append(true)
diff --git a/crates/biliup/src/downloader/httpflv.rs b/crates/biliup/src/downloader/httpflv.rs
index 6dfbd44..82d637e 100644
--- a/crates/biliup/src/downloader/httpflv.rs
+++ b/crates/biliup/src/downloader/httpflv.rs
@@ -331,15 +331,9 @@ mod tests {
Ok(())
}
- /// 回归测试：纯视频流（没有任何音频标签）在首次分段时不应 panic。
- ///
- /// 该流只包含一个 onMetaData 脚本标签和一个 H264 序列头关键帧，`aac_sequence_header`
- /// 全程为 `None`。修复前，分段重建逻辑会对 `aac_sequence_header` 执行
- /// `expect("aac_sequence_header does not exist")` 而 panic，导致纯视频直播录制中断。
- #[tokio::test]
- async fn pure_video_stream_segments_without_panic() -> Result<(), Box> {
- use crate::downloader::util::{LifecycleFile, Segmentable};
-
+ /// 一段最小的纯视频 FLV 流体（不含 9 字节文件头）：
+ /// onMetaData 脚本标签 + 一个 H264 序列头关键帧，没有任何音频标签。
+ fn pure_video_flv_body() -> Vec {
let mut data: Vec = Vec::new();
// parse_flv 起始会先读取 4 字节（上一个 tag 的大小），这里给占位。
data.extend_from_slice(&[0, 0, 0, 0]);
@@ -366,8 +360,21 @@ mod tests {
]);
data.extend_from_slice(&video_body);
data.extend_from_slice(&[0, 0, 0, 0]); // previous_tag_size
+ data
+ }
- let http_resp = http::Response::builder().status(200).body(data)?;
+ /// 回归测试：纯视频流（没有任何音频标签）在首次分段时不应 panic。
+ ///
+ /// 该流只包含一个 onMetaData 脚本标签和一个 H264 序列头关键帧，`aac_sequence_header`
+ /// 全程为 `None`。修复前，分段重建逻辑会对 `aac_sequence_header` 执行
+ /// `expect("aac_sequence_header does not exist")` 而 panic，导致纯视频直播录制中断。
+ #[tokio::test]
+ async fn pure_video_stream_segments_without_panic() -> Result<(), Box> {
+ use crate::downloader::util::{LifecycleFile, Segmentable};
+
+ let http_resp = http::Response::builder()
+ .status(200)
+ .body(pure_video_flv_body())?;
let resp = reqwest::Response::from(http_resp);
let connection = super::Connection::new(resp);
@@ -382,4 +389,29 @@ mod tests {
super::parse_flv(connection, file, segment).await?;
Ok(())
}
+
+ /// 写盘字节计数：只统计真正经 `write_tag` 落盘的 tag（11 字节头 + 数据 + 4 字节 previous tag size）。
+ ///
+ /// 这段流里只有 onMetaData 脚本标签会在遇到关键帧时被写出（14 字节数据 → 29 字节），
+ /// 关键帧本身留在缓存里等待下一个关键帧，流结束时并未落盘，因此不计入。
+ #[tokio::test]
+ async fn written_bytes_are_counted_on_the_shared_counter()
+ -> Result<(), Box> {
+ use crate::downloader::util::{ByteCounter, LifecycleFile, Segmentable};
+
+ let http_resp = http::Response::builder()
+ .status(200)
+ .body(pure_video_flv_body())?;
+ let connection = super::Connection::new(reqwest::Response::from(http_resp));
+
+ let dir = tempfile::tempdir()?;
+ let file_stem = dir.path().join("counted");
+ let counter = ByteCounter::new();
+ let file =
+ LifecycleFile::new(file_stem.to_str().unwrap(), "flv").with_counter(counter.clone());
+
+ super::parse_flv(connection, file, Segmentable::new(None, None)).await?;
+ assert_eq!(counter.total(), 11 + 14 + 4);
+ Ok(())
+ }
}
diff --git a/crates/biliup/src/downloader/live/acfun.rs b/crates/biliup/src/downloader/live/acfun.rs
index 39e8585..bcf8b0b 100644
--- a/crates/biliup/src/downloader/live/acfun.rs
+++ b/crates/biliup/src/downloader/live/acfun.rs
@@ -79,6 +79,7 @@ impl AcfunLive {
title: data.caption,
date: Utc::now(),
live_cover_url: String::new(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "flv".to_string()),
raw_stream_url,
platform: "acfun".to_string(),
diff --git a/crates/biliup/src/downloader/live/afreecatv.rs b/crates/biliup/src/downloader/live/afreecatv.rs
index 6f8897d..56b052c 100644
--- a/crates/biliup/src/downloader/live/afreecatv.rs
+++ b/crates/biliup/src/downloader/live/afreecatv.rs
@@ -125,6 +125,7 @@ impl AfreecaTVLive {
title: channel.title,
date: Utc::now(),
live_cover_url: String::new(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "m3u8".to_string()),
raw_stream_url,
platform: "afreecatv".to_string(),
diff --git a/crates/biliup/src/downloader/live/bigo.rs b/crates/biliup/src/downloader/live/bigo.rs
index d46b223..5d79302 100644
--- a/crates/biliup/src/downloader/live/bigo.rs
+++ b/crates/biliup/src/downloader/live/bigo.rs
@@ -91,6 +91,7 @@ impl BigoLive {
title: data.room_topic.unwrap_or(room_id),
date: Utc::now(),
live_cover_url: String::new(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "m3u8".to_string()),
raw_stream_url,
platform: "bigo".to_string(),
diff --git a/crates/biliup/src/downloader/live/bilibili.rs b/crates/biliup/src/downloader/live/bilibili.rs
index 7a77e16..120384e 100644
--- a/crates/biliup/src/downloader/live/bilibili.rs
+++ b/crates/biliup/src/downloader/live/bilibili.rs
@@ -142,6 +142,7 @@ impl BilibiliLive {
title: profile.title,
date: Utc::now(),
live_cover_url: profile.cover,
+ avatar_url: (!profile.avatar.is_empty()).then_some(profile.avatar),
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| {
if raw_stream_url.contains(".m3u8") {
"m3u8".to_string()
@@ -252,53 +253,7 @@ impl BilibiliLive {
.await
.map_err(|err| LiveError::custom(format!("解析 B 站直播间信息失败: {err}")))?;
- if room_info.get("code").and_then(|code| code.as_i64()) != Some(0) {
- return Err(LiveError::custom(format!(
- "获取 B 站直播间信息错误: {}",
- room_info
- .get("message")
- .or_else(|| room_info.get("msg"))
- .and_then(|msg| msg.as_str())
- .unwrap_or_default()
- )));
- }
-
- let room = room_info
- .get("data")
- .and_then(|data| data.get("room_info"))
- .ok_or_else(|| LiveError::custom("B 站直播间信息为空"))?;
- if room.get("live_status").and_then(|status| status.as_i64()) != Some(1) {
- return Ok(None);
- }
-
- Ok(Some(BiliRoomProfile {
- room_id: room
- .get("room_id")
- .and_then(|room_id| room_id.as_u64())
- .ok_or_else(|| LiveError::custom("B 站真实房间号为空"))?,
- uid: room
- .get("uid")
- .and_then(|uid| uid.as_u64())
- .unwrap_or_default(),
- live_start_time: room
- .get("live_start_time")
- .and_then(|time| time.as_u64())
- .unwrap_or_default(),
- special_type: room
- .get("special_type")
- .and_then(|special_type| special_type.as_u64())
- .unwrap_or_default(),
- title: room
- .get("title")
- .and_then(|title| title.as_str())
- .unwrap_or_default()
- .to_string(),
- cover: room
- .get("cover")
- .and_then(|cover| cover.as_str())
- .unwrap_or_default()
- .to_string(),
- }))
+ parse_room_profile(&room_info)
}
async fn get_stream_candidates(
@@ -654,6 +609,64 @@ struct BiliRoomProfile {
special_type: u64,
title: String,
cover: String,
+ /// 主播头像（`anchor_info.base_info.face`），与房间信息来自同一次响应
+ avatar: String,
+}
+
+/// 解析 `getInfoByRoom` 的响应；未开播返回 `Ok(None)`。
+fn parse_room_profile(room_info: &Value) -> LiveResult> {
+ if room_info.get("code").and_then(|code| code.as_i64()) != Some(0) {
+ return Err(LiveError::custom(format!(
+ "获取 B 站直播间信息错误: {}",
+ room_info
+ .get("message")
+ .or_else(|| room_info.get("msg"))
+ .and_then(|msg| msg.as_str())
+ .unwrap_or_default()
+ )));
+ }
+
+ let room = room_info
+ .get("data")
+ .and_then(|data| data.get("room_info"))
+ .ok_or_else(|| LiveError::custom("B 站直播间信息为空"))?;
+ if room.get("live_status").and_then(|status| status.as_i64()) != Some(1) {
+ return Ok(None);
+ }
+
+ Ok(Some(BiliRoomProfile {
+ room_id: room
+ .get("room_id")
+ .and_then(|room_id| room_id.as_u64())
+ .ok_or_else(|| LiveError::custom("B 站真实房间号为空"))?,
+ uid: room
+ .get("uid")
+ .and_then(|uid| uid.as_u64())
+ .unwrap_or_default(),
+ live_start_time: room
+ .get("live_start_time")
+ .and_then(|time| time.as_u64())
+ .unwrap_or_default(),
+ special_type: room
+ .get("special_type")
+ .and_then(|special_type| special_type.as_u64())
+ .unwrap_or_default(),
+ title: room
+ .get("title")
+ .and_then(|title| title.as_str())
+ .unwrap_or_default()
+ .to_string(),
+ cover: room
+ .get("cover")
+ .and_then(|cover| cover.as_str())
+ .unwrap_or_default()
+ .to_string(),
+ avatar: room_info
+ .pointer("/data/anchor_info/base_info/face")
+ .and_then(Value::as_str)
+ .unwrap_or_default()
+ .to_string(),
+ }))
}
struct BiliStreamCandidate {
@@ -853,4 +866,38 @@ mod tests {
assert_eq!(candidates.len(), 1);
assert!(candidates[0].url.starts_with("https://a.example.com/"));
}
+
+ /// 头像与房间信息来自同一次 getInfoByRoom 响应，字段缺失时留空而不是报错
+ #[test]
+ fn room_profile_takes_cover_and_anchor_face_from_one_response() {
+ let response = serde_json::json!({
+ "code": 0,
+ "data": {
+ "room_info": {
+ "room_id": 1, "uid": 2, "live_status": 1, "live_start_time": 3,
+ "special_type": 0, "title": "t",
+ "cover": "https://i0.hdslb.com/bfs/live/cover.jpg"
+ },
+ "anchor_info": {
+ "base_info": { "face": "https://i1.hdslb.com/bfs/face/face.jpg" }
+ }
+ }
+ });
+ let profile = parse_room_profile(&response).unwrap().unwrap();
+ assert_eq!(profile.cover, "https://i0.hdslb.com/bfs/live/cover.jpg");
+ assert_eq!(profile.avatar, "https://i1.hdslb.com/bfs/face/face.jpg");
+
+ let without_anchor = serde_json::json!({
+ "code": 0,
+ "data": { "room_info": { "room_id": 1, "live_status": 1, "title": "t" } }
+ });
+ let profile = parse_room_profile(&without_anchor).unwrap().unwrap();
+ assert_eq!(profile.avatar, "");
+
+ let offline = serde_json::json!({
+ "code": 0,
+ "data": { "room_info": { "room_id": 1, "live_status": 0 } }
+ });
+ assert!(parse_room_profile(&offline).unwrap().is_none());
+ }
}
diff --git a/crates/biliup/src/downloader/live/cc.rs b/crates/biliup/src/downloader/live/cc.rs
index 61cedd1..9132d61 100644
--- a/crates/biliup/src/downloader/live/cc.rs
+++ b/crates/biliup/src/downloader/live/cc.rs
@@ -87,6 +87,7 @@ impl CCLive {
title,
date: Utc::now(),
live_cover_url: String::new(),
+ avatar_url: None,
raw_stream_url,
platform: "cc".to_string(),
stream_headers: HashMap::new(),
diff --git a/crates/biliup/src/downloader/live/douyin.rs b/crates/biliup/src/downloader/live/douyin.rs
index c0a7261..66773e9 100644
--- a/crates/biliup/src/downloader/live/douyin.rs
+++ b/crates/biliup/src/downloader/live/douyin.rs
@@ -110,13 +110,9 @@ impl<'a> DouyinLive<'a> {
.and_then(Value::as_str)
.unwrap_or_default()
.to_string();
- let cover = room_info
- .pointer("/cover/url_list")
- .and_then(Value::as_array)
- .and_then(|url_list| url_list.first())
- .and_then(Value::as_str)
- .unwrap_or_default()
- .to_string();
+ let cover = first_url(room_info.pointer("/cover/url_list")).unwrap_or_default();
+ // web 与 H5 两个接口的房间对象里都带 owner.avatar_thumb
+ let avatar_url = first_url(room_info.pointer("/owner/avatar_thumb/url_list"));
Ok(LiveStatus::Live {
stream: Box::new(LiveStream {
@@ -125,6 +121,7 @@ impl<'a> DouyinLive<'a> {
title,
date: Utc::now(),
live_cover_url: cover,
+ avatar_url,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "flv".to_string()),
raw_stream_url,
platform: "douyin".to_string(),
@@ -478,6 +475,16 @@ impl<'a> DouyinLive<'a> {
}
}
+/// 抖音图片字段形如 `{"url_list": ["https://...", ...]}`，取第一个非空地址
+fn first_url(url_list: Option<&Value>) -> Option {
+ url_list?
+ .as_array()?
+ .iter()
+ .filter_map(Value::as_str)
+ .find(|url| !url.is_empty())
+ .map(ToString::to_string)
+}
+
fn common_params() -> Vec<(String, String)> {
vec![
("app_name".to_string(), "douyin_web".to_string()),
@@ -994,6 +1001,27 @@ mod tests {
);
}
+ #[test]
+ fn first_url_picks_the_first_non_empty_entry_and_tolerates_missing_fields() {
+ let room = serde_json::json!({
+ "cover": { "url_list": ["", "https://p3.douyinpic.com/cover.jpg"] },
+ "owner": { "avatar_thumb": { "url_list": ["https://p3.douyinpic.com/face.jpeg"] } }
+ });
+ assert_eq!(
+ first_url(room.pointer("/cover/url_list")).as_deref(),
+ Some("https://p3.douyinpic.com/cover.jpg")
+ );
+ assert_eq!(
+ first_url(room.pointer("/owner/avatar_thumb/url_list")).as_deref(),
+ Some("https://p3.douyinpic.com/face.jpeg")
+ );
+ assert_eq!(
+ first_url(room.pointer("/owner/avatar_large/url_list")),
+ None
+ );
+ assert_eq!(first_url(Some(&serde_json::json!([]))), None);
+ }
+
#[tokio::test]
#[ignore = "需要外网访问抖音直播间"]
async fn douyin_live_room_resolves_and_streams_flv() {
diff --git a/crates/biliup/src/downloader/live/douyu.rs b/crates/biliup/src/downloader/live/douyu.rs
index a7e4bdf..3634fc9 100644
--- a/crates/biliup/src/downloader/live/douyu.rs
+++ b/crates/biliup/src/downloader/live/douyu.rs
@@ -141,13 +141,15 @@ impl<'a> DouyuLive<'a> {
let raw_stream_url = format!("{}/{}", play_info.rtmp_url, play_info.rtmp_live);
let raw_stream_url = self.maybe_build_huos_url(raw_stream_url).await;
+ let avatar_url = room_info.avatar_url();
Ok(LiveStatus::Live {
stream: Box::new(LiveStream {
name: self.name.clone(),
url: self.url.clone(),
title: room_info.room_name,
date: Utc::now(),
- live_cover_url: String::new(),
+ live_cover_url: room_info.room_pic.unwrap_or_default(),
+ avatar_url,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "flv".to_string()),
raw_stream_url,
platform: "douyu".to_string(),
@@ -759,6 +761,33 @@ struct RoomInfo {
show_status: i64,
#[serde(rename = "videoLoop")]
video_loop: i64,
+ /// 直播间封面（betard 里的 `room_pic`，通常是 avif），缺失时为空
+ #[serde(default)]
+ room_pic: Option,
+ /// 主播头像直链
+ #[serde(default)]
+ owner_avatar: Option,
+ /// 头像的多尺寸版本，形如 `{"big": ..., "middle": ..., "small": ...}`；
+ /// 个别房间只给这一份，作为 `owner_avatar` 缺失时的后备
+ #[serde(default)]
+ avatar: Option,
+}
+
+impl RoomInfo {
+ fn avatar_url(&self) -> Option {
+ let non_empty = |url: &str| (!url.is_empty()).then(|| url.to_string());
+ self.owner_avatar
+ .as_deref()
+ .and_then(non_empty)
+ .or_else(|| {
+ let avatar = self.avatar.as_ref()?;
+ ["middle", "big", "small"]
+ .iter()
+ .filter_map(|size| avatar.get(size).and_then(Value::as_str))
+ .find_map(non_empty)
+ .or_else(|| avatar.as_str().and_then(non_empty))
+ })
+ }
}
#[derive(Deserialize)]
@@ -1095,4 +1124,43 @@ mod tests {
assert_eq!(cached.server_aligned_now(local_now + 10), server_now + 10);
assert!(cached.is_valid(cached.server_aligned_now(local_now)));
}
+
+ /// betard 的 room 对象里 `room_pic` 是封面、`owner_avatar` / `avatar.{big,middle,small}` 是头像；
+ /// 老样本没有这些键时仍要能反序列化
+ #[test]
+ fn betard_room_exposes_cover_and_avatar_with_fallbacks() {
+ let full: BetardResponse = serde_json::from_str(
+ r#"{"room":{"room_name":"n","show_status":1,"videoLoop":0,
+ "room_pic":"https://rpic.douyucdn.cn/asrpic/260922/1_src.avif/dy4",
+ "owner_avatar":"https://apic.douyucdn.cn/face_big.jpg",
+ "avatar":{"big":"https://apic.douyucdn.cn/face_big.jpg","middle":"https://apic.douyucdn.cn/face_middle.jpg","small":""}}}"#,
+ )
+ .unwrap();
+ let room = full.room.unwrap();
+ assert_eq!(
+ room.room_pic.as_deref(),
+ Some("https://rpic.douyucdn.cn/asrpic/260922/1_src.avif/dy4")
+ );
+ assert_eq!(
+ room.avatar_url().as_deref(),
+ Some("https://apic.douyucdn.cn/face_big.jpg")
+ );
+
+ let only_sizes: BetardResponse = serde_json::from_str(
+ r#"{"room":{"room_name":"n","show_status":1,"videoLoop":0,
+ "avatar":{"big":"","middle":"https://apic.douyucdn.cn/face_middle.jpg"}}}"#,
+ )
+ .unwrap();
+ assert_eq!(
+ only_sizes.room.unwrap().avatar_url().as_deref(),
+ Some("https://apic.douyucdn.cn/face_middle.jpg")
+ );
+
+ let legacy: BetardResponse =
+ serde_json::from_str(r#"{"room":{"room_name":"n","show_status":1,"videoLoop":0}}"#)
+ .unwrap();
+ let room = legacy.room.unwrap();
+ assert_eq!(room.room_pic, None);
+ assert_eq!(room.avatar_url(), None);
+ }
}
diff --git a/crates/biliup/src/downloader/live/general.rs b/crates/biliup/src/downloader/live/general.rs
index 2ba0758..34a454e 100644
--- a/crates/biliup/src/downloader/live/general.rs
+++ b/crates/biliup/src/downloader/live/general.rs
@@ -97,6 +97,7 @@ impl GeneralLive {
title: selection.title.clone(),
date: Utc::now(),
live_cover_url: selection.thumbnail.clone(),
+ avatar_url: None,
raw_stream_url: selection.raw_stream_url.clone(),
platform: "general".to_string(),
stream_headers: HashMap::new(),
diff --git a/crates/biliup/src/downloader/live/huya.rs b/crates/biliup/src/downloader/live/huya.rs
index 7d875bd..e43618f 100644
--- a/crates/biliup/src/downloader/live/huya.rs
+++ b/crates/biliup/src/downloader/live/huya.rs
@@ -152,6 +152,7 @@ impl<'a> HuyaLive<'a> {
title: profile.title,
date: Utc::now(),
live_cover_url: profile.cover,
+ avatar_url: (!profile.avatar.is_empty()).then_some(profile.avatar),
suffix: media_ext_from_url(&raw_stream_url)
.unwrap_or_else(|| self.huya_protocol.extension().to_string()),
raw_stream_url,
@@ -280,6 +281,7 @@ impl<'a> HuyaLive<'a> {
.and_then(Value::as_str)
.unwrap_or_default()
.replace("http://", "https://"),
+ avatar: huya_image_url(live_data.get("avatar180")),
max_bitrate: live_data
.get("bitRate")
.and_then(Value::as_u64)
@@ -355,6 +357,7 @@ impl<'a> HuyaLive<'a> {
.and_then(|cover| cover.as_str())
.unwrap_or_default()
.replace("http://", "https://"),
+ avatar: huya_image_url(live_info.get("avatar180")),
max_bitrate: live_info
.get("bitRate")
.and_then(|bitrate| bitrate.as_u64())
@@ -561,11 +564,21 @@ impl<'a> HuyaLive<'a> {
struct HuyaRoomProfile {
title: String,
cover: String,
+ /// 主播头像（`avatar180`），网页 `gameLiveInfo` 与小程序 `liveData` 里都有
+ avatar: String,
max_bitrate: u32,
bitrate_info: Vec,
stream_info: Vec,
}
+/// 虎牙图片地址常带 `http://`，与 `screenshot` 一样统一成 https；缺失时为空串
+fn huya_image_url(value: Option<&Value>) -> String {
+ value
+ .and_then(Value::as_str)
+ .unwrap_or_default()
+ .replace("http://", "https://")
+}
+
enum HuyaProtocol {
Flv,
Hls,
@@ -889,4 +902,16 @@ mod tests {
assert_eq!(value["text"], "}; { ]");
assert_eq!(value["items"][0]["value"], 1);
}
+
+ #[test]
+ fn huya_image_url_upgrades_to_https_and_tolerates_missing_field() {
+ let info = serde_json::json!({
+ "avatar180": "http://huyaimg.msstatic.com/avatar/1099/f6/face_180_135.jpg?1"
+ });
+ assert_eq!(
+ huya_image_url(info.get("avatar180")),
+ "https://huyaimg.msstatic.com/avatar/1099/f6/face_180_135.jpg?1"
+ );
+ assert_eq!(huya_image_url(info.get("screenshot")), "");
+ }
}
diff --git a/crates/biliup/src/downloader/live/inke.rs b/crates/biliup/src/downloader/live/inke.rs
index 6f66381..cb56e59 100644
--- a/crates/biliup/src/downloader/live/inke.rs
+++ b/crates/biliup/src/downloader/live/inke.rs
@@ -90,6 +90,7 @@ impl InkeLive {
title: data.title.unwrap_or(uid),
date: Utc::now(),
live_cover_url: data.image.unwrap_or_default(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "flv".to_string()),
raw_stream_url,
platform: "inke".to_string(),
diff --git a/crates/biliup/src/downloader/live/kilakila.rs b/crates/biliup/src/downloader/live/kilakila.rs
index a729aa1..87aeab4 100644
--- a/crates/biliup/src/downloader/live/kilakila.rs
+++ b/crates/biliup/src/downloader/live/kilakila.rs
@@ -96,6 +96,7 @@ impl KilakilaLive {
title: response.b.title.unwrap_or(room_id),
date: Utc::now(),
live_cover_url: response.b.cover_url.unwrap_or_default(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| {
if self.protocol == "flv" {
"flv".to_string()
diff --git a/crates/biliup/src/downloader/live/kuaishou.rs b/crates/biliup/src/downloader/live/kuaishou.rs
index fa4a861..7271f7e 100644
--- a/crates/biliup/src/downloader/live/kuaishou.rs
+++ b/crates/biliup/src/downloader/live/kuaishou.rs
@@ -92,6 +92,7 @@ impl KuaishouLive {
title,
date: Utc::now(),
live_cover_url: live_stream.cover_url.unwrap_or_default(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "flv".to_string()),
raw_stream_url,
platform: "kuaishou".to_string(),
diff --git a/crates/biliup/src/downloader/live/missevan.rs b/crates/biliup/src/downloader/live/missevan.rs
index 35b26c0..4b0881d 100644
--- a/crates/biliup/src/downloader/live/missevan.rs
+++ b/crates/biliup/src/downloader/live/missevan.rs
@@ -92,6 +92,7 @@ impl MissevanLive {
title,
date: Utc::now(),
live_cover_url: cover,
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "flv".to_string()),
raw_stream_url,
platform: "missevan".to_string(),
diff --git a/crates/biliup/src/downloader/live/mod.rs b/crates/biliup/src/downloader/live/mod.rs
index 64495fb..ff838e6 100644
--- a/crates/biliup/src/downloader/live/mod.rs
+++ b/crates/biliup/src/downloader/live/mod.rs
@@ -336,6 +336,9 @@ pub struct LiveStream {
pub title: String,
pub date: DateTime,
pub live_cover_url: String,
+ /// 主播头像地址；平台响应里没有或尚未解析的平台为 `None`。
+ #[serde(default)]
+ pub avatar_url: Option,
pub raw_stream_url: String,
pub platform: String,
pub stream_headers: HashMap,
diff --git a/crates/biliup/src/downloader/live/niconico.rs b/crates/biliup/src/downloader/live/niconico.rs
index 0df7ae0..08bb3ab 100644
--- a/crates/biliup/src/downloader/live/niconico.rs
+++ b/crates/biliup/src/downloader/live/niconico.rs
@@ -79,6 +79,7 @@ impl NiconicoLive {
title,
date: Utc::now(),
live_cover_url: String::new(),
+ avatar_url: None,
raw_stream_url: self.url.clone(),
platform: "niconico".to_string(),
stream_headers: HashMap::new(),
diff --git a/crates/biliup/src/downloader/live/picarto.rs b/crates/biliup/src/downloader/live/picarto.rs
index cd79119..74a3370 100644
--- a/crates/biliup/src/downloader/live/picarto.rs
+++ b/crates/biliup/src/downloader/live/picarto.rs
@@ -152,6 +152,7 @@ impl PicartoLive {
title,
date: Utc::now(),
live_cover_url: stream.thumbnail_image.unwrap_or_default(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "m3u8".to_string()),
raw_stream_url,
platform: "picarto".to_string(),
diff --git a/crates/biliup/src/downloader/live/ttinglive.rs b/crates/biliup/src/downloader/live/ttinglive.rs
index 5630311..58d0567 100644
--- a/crates/biliup/src/downloader/live/ttinglive.rs
+++ b/crates/biliup/src/downloader/live/ttinglive.rs
@@ -80,6 +80,7 @@ impl TTingLiveLive {
title: info.title.unwrap_or(room_id),
date: Utc::now(),
live_cover_url: info.thumb_url.unwrap_or_default(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "m3u8".to_string()),
raw_stream_url,
platform: "ttinglive".to_string(),
diff --git a/crates/biliup/src/downloader/live/twitcasting.rs b/crates/biliup/src/downloader/live/twitcasting.rs
index a9a9d8f..c7627da 100644
--- a/crates/biliup/src/downloader/live/twitcasting.rs
+++ b/crates/biliup/src/downloader/live/twitcasting.rs
@@ -94,6 +94,7 @@ impl TwitcastingLive {
title: room_page.title,
date: Utc::now(),
live_cover_url: String::new(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "m3u8".to_string()),
raw_stream_url,
platform: "twitcasting".to_string(),
diff --git a/crates/biliup/src/downloader/live/twitch.rs b/crates/biliup/src/downloader/live/twitch.rs
index 8b60b4f..3ca7fbf 100644
--- a/crates/biliup/src/downloader/live/twitch.rs
+++ b/crates/biliup/src/downloader/live/twitch.rs
@@ -272,6 +272,7 @@ impl TwitchLive {
title: stream.title.unwrap_or_default(),
date: Utc::now(),
live_cover_url: stream.preview_image_url.unwrap_or_default(),
+ avatar_url: None,
suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "m3u8".to_string()),
raw_stream_url,
platform: "twitch".to_string(),
@@ -386,6 +387,7 @@ impl TwitchVideosLive {
title: selection.title,
date: Utc::now(),
live_cover_url: selection.thumbnail.clone(),
+ avatar_url: None,
suffix: "mp4".to_string(),
raw_stream_url,
platform: "twitch".to_string(),
diff --git a/crates/biliup/src/downloader/live/youtube.rs b/crates/biliup/src/downloader/live/youtube.rs
index 961e3f5..c90c2e1 100644
--- a/crates/biliup/src/downloader/live/youtube.rs
+++ b/crates/biliup/src/downloader/live/youtube.rs
@@ -138,6 +138,7 @@ impl YoutubeLive {
title: selection.title.clone(),
date: Utc::now(),
live_cover_url: selection.thumbnail.clone(),
+ avatar_url: None,
raw_stream_url,
platform: "youtube".to_string(),
stream_headers: HashMap::new(),
diff --git a/crates/biliup/src/downloader/live/yy.rs b/crates/biliup/src/downloader/live/yy.rs
index 3828740..c40862d 100644
--- a/crates/biliup/src/downloader/live/yy.rs
+++ b/crates/biliup/src/downloader/live/yy.rs
@@ -149,6 +149,7 @@ impl YYLive {
title: String::new(),
date: Utc::now(),
live_cover_url: String::new(),
+ avatar_url: None,
raw_stream_url: raw_stream_url.clone(),
platform: "yy".to_string(),
stream_headers: HashMap::new(),
diff --git a/crates/biliup/src/downloader/util.rs b/crates/biliup/src/downloader/util.rs
index 3d57b1b..ef8c250 100644
--- a/crates/biliup/src/downloader/util.rs
+++ b/crates/biliup/src/downloader/util.rs
@@ -1,12 +1,36 @@
use chrono::{DateTime, Local};
use std::fs;
use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
use std::time::Duration;
use tracing::{error, info};
pub type CallbackFn<'a> = Box;
+/// 已写盘字节的原子累计，供录制线程之外（如 Web 接口）读取实时速率。
+///
+/// 写盘路径上只做一次 `fetch_add`：不加锁、不 await、不会失败，
+/// 因此不改变录制的任何控制流。`Clone` 得到的是同一计数器的另一个句柄。
+#[derive(Debug, Clone, Default)]
+pub struct ByteCounter(Arc);
+
+impl ByteCounter {
+ pub fn new() -> Self {
+ Self::default()
+ }
+
+ #[inline]
+ pub fn add(&self, bytes: u64) {
+ self.0.fetch_add(bytes, Ordering::Relaxed);
+ }
+
+ pub fn total(&self) -> u64 {
+ self.0.load(Ordering::Relaxed)
+ }
+}
+
#[derive(Debug)]
pub enum Segment {
Time(Duration, Duration),
@@ -232,6 +256,8 @@ pub struct LifecycleFile<'a> {
pub path: PathBuf,
pub hook: CallbackFn<'a>,
pub extension: &'static str,
+ /// 写入这一系列分段文件的字节累计（跨分段，不随 `create_new` 归零）。
+ pub bytes_written: ByteCounter,
}
impl<'a> LifecycleFile<'a> {
@@ -249,9 +275,16 @@ impl<'a> LifecycleFile<'a> {
path: Default::default(),
hook: Box::new(hook),
extension,
+ bytes_written: ByteCounter::new(),
}
}
+ /// 让写盘字节累计到调用方持有的计数器上（例如按录制任务汇总速率）。
+ pub fn with_counter(mut self, counter: ByteCounter) -> Self {
+ self.bytes_written = counter;
+ self
+ }
+
pub fn create(&mut self) -> Result<&Path, std::io::Error> {
// 构建最终文件名
self.file_name = format!(
@@ -310,6 +343,20 @@ mod tests {
Ok(())
}
+ #[test]
+ fn byte_counter_clones_share_one_total() {
+ let counter = ByteCounter::new();
+ let handle = counter.clone();
+ counter.add(10);
+ handle.add(5);
+ assert_eq!(counter.total(), 15);
+ assert_eq!(handle.total(), 15);
+
+ let file = LifecycleFile::new("x", "flv").with_counter(counter.clone());
+ file.bytes_written.add(1);
+ assert_eq!(counter.total(), 16);
+ }
+
#[test]
fn test_segmentation_logic() -> Result<(), Box> {
// 测试时间分割
+ {avatarSrc ?
- {duration ? (
- {duration}
+ {live ? (
+
+ {duration ? {duration} : null}
+
+ {rate ?? '—'}
+
+
) : lastRec ? (
{lastRec}
) : null}
diff --git a/app/ui/streamer-card.module.scss b/app/ui/streamer-card.module.scss
index 9382054..cdb1351 100644
--- a/app/ui/streamer-card.module.scss
+++ b/app/ui/streamer-card.module.scss
@@ -129,15 +129,50 @@
flex-shrink: 0;
}
+/* ===== 录制中:直播间封面(16:9) ===== */
+.cover {
+ position: relative;
+ margin: 0 0 10px;
+ aspect-ratio: 16 / 9;
+ border-radius: 8px;
+ overflow: hidden;
+ background: var(--semi-color-fill-1);
+}
+.cover img {
+ object-fit: cover;
+ opacity: 0;
+ transition: opacity 0.25s ease;
+}
+.cover[data-state='ok'] img {
+ opacity: 1;
+}
+
/* ===== 名称 + 时长 ===== */
.nameRow {
display: flex;
- align-items: baseline;
+ align-items: center;
justify-content: space-between;
gap: 10px;
margin-bottom: 3px;
}
+.nameWrap {
+ display: flex;
+ align-items: center;
+ gap: 7px;
+ min-width: 0;
+}
+
+/* 录制中:主播头像小圆图 */
+.avatar {
+ width: 22px;
+ height: 22px;
+ border-radius: 50%;
+ object-fit: cover;
+ flex-shrink: 0;
+ background: var(--semi-color-fill-1);
+}
+
.name {
font-size: 15px;
font-weight: 650;
@@ -148,6 +183,14 @@
min-width: 0;
}
+/* 直播中:右侧时长 + 写盘速率 */
+.liveStats {
+ display: inline-flex;
+ align-items: baseline;
+ gap: 8px;
+ flex-shrink: 0;
+}
+
/* 直播中:实时时长(绿色) */
.liveInfo {
font-size: 12.5px;
@@ -158,6 +201,21 @@
font-variant-numeric: tabular-nums;
}
+/* 直播中:写盘速率(最近 10 秒平均);无采样时为「—」 */
+.rate {
+ font-size: 12px;
+ font-weight: 600;
+ color: var(--semi-color-text-2);
+ white-space: nowrap;
+ flex-shrink: 0;
+ font-variant-numeric: tabular-nums;
+ padding: 1px 7px;
+ border-radius: 999px;
+ background: var(--semi-color-fill-0);
+ min-width: 2.6em;
+ text-align: center;
+}
+
/* 未开播:最近录制时间 */
.recInfo {
font-size: 12px;
diff --git a/crates/biliup-cli/src/server/api/endpoints.rs b/crates/biliup-cli/src/server/api/endpoints.rs
index e1d3424..1404d70 100644
--- a/crates/biliup-cli/src/server/api/endpoints.rs
+++ b/crates/biliup-cli/src/server/api/endpoints.rs
@@ -63,9 +63,16 @@ pub async fn get_streamers_endpoint(
.into_iter()
.find(|worker| worker.live_streamer.id == x.id);
- let status = match option.as_ref() {
- Some(t) => format!("{:?}", *t.downloader_status.read().unwrap()),
- None => String::new(),
+ let (status, live) = match option.as_ref() {
+ Some(t) => {
+ let downloader_status = t.downloader_status.read().unwrap();
+ let live = match &*downloader_status {
+ WorkerStatus::Working(task) => Some((task.bytes_per_sec(), task.live_media())),
+ _ => None,
+ };
+ (format!("{:?}", *downloader_status), live)
+ }
+ None => (String::new(), None),
};
// 被录制策略挡下时报告具体原因，否则「不在时间范围内」和「单纯没开播」
// 在界面上都是「空闲」，用户无从判断配置有没有生效。
@@ -75,6 +82,8 @@ pub async fn get_streamers_endpoint(
Some(reason) if status != "Working" => reason.to_string(),
_ => status,
};
+ let (live_bytes_per_sec, live_media) = live.unzip();
+ let live_media = live_media.unwrap_or_default();
results.push(LiveStreamerResponse {
status,
@@ -82,6 +91,9 @@ pub async fn get_streamers_endpoint(
upload_status: option
.map(|t| format!("{:?}", *t.uploader_status.read().unwrap()))
.unwrap_or_default(),
+ live_bytes_per_sec: live_bytes_per_sec.flatten(),
+ live_cover_url: live_media.cover_url,
+ live_avatar_url: live_media.avatar_url,
});
}
Ok(Json(results))
diff --git a/crates/biliup-cli/src/server/api/live_media.rs b/crates/biliup-cli/src/server/api/live_media.rs
new file mode 100644
index 0000000..facd062
--- /dev/null
+++ b/crates/biliup-cli/src/server/api/live_media.rs
@@ -0,0 +1,494 @@
+//! 正在录制的直播间封面 / 主播头像接口。
+//!
+//! - 封面每场直播都可能换：服务端以直播间页面地址作 Referer 去取（页面直连图片 CDN 会被
+//! B 站等按站外 Referer 拒绝），按图片 URL 做一份短暂的内存缓存后转发。
+//! - 头像几乎不变：开播时下载一次存到 `data/avatar/`，这里直接读本地文件；
+//! 本地还没有（或地址变了）时现场下载一次再返回。
+//!
+//! 只按主播 id 取 worker 里已经拿到的地址，不接受任意 URL，避免变成开放代理。
+//! 路由注册在 `router()` 里，`--auth` 时与 `/v1/streamers` 同一道登录校验。
+
+use crate::server::common::live_image::{
+ AvatarStore, FetchedImage, LiveImage, avatar_store, fetch_image, image_client, upstream_url,
+};
+use crate::server::core::download_manager::DownloadManager;
+use crate::server::infrastructure::context::{Worker, WorkerStatus};
+use axum::extract::{Path, State};
+use axum::http::{HeaderValue, StatusCode, header};
+use axum::response::{IntoResponse, Response};
+use bytes::Bytes;
+use reqwest::Client;
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+use tracing::warn;
+
+/// 封面在内存里的缓存时间
+const CACHE_TTL: Duration = Duration::from_secs(300);
+/// 缓存条目上限（按图片 URL）。默认 pool1_size = 5，远用不到这个数
+const MAX_CACHE_ENTRIES: usize = 64;
+
+#[derive(Debug, Clone)]
+struct CachedImage {
+ fetched_at: Instant,
+ content_type: String,
+ body: Bytes,
+}
+
+/// 封面的抓取客户端与内存缓存，进程内唯一，放在 `ServiceRegister` 里。
+pub struct ImageProxy {
+ client: Client,
+ cache: Mutex
+ {name}
+