feat(ui): 录制中房间显示封面、头像与实时码率(直播预览阶段一) (#1709)

* feat(downloader): 录制写盘字节计数与实时速率,/v1/streamers 新增 live_bytes_per_sec

各下载器在写出点往 `ByteCounter`(Arc<AtomicU64>)累加,写盘路径上只多一次
`fetch_add`,不加锁、不 await、不改任何控制流:

- stream-gears/httpflv:`FlvFile::write_tag` 按 11 + 数据 + 4 字节计
- stream-gears/hls:`download_to_file` 每个 chunk 落盘后计
- mesio:条目进入修复管线前按 `FlvData::size()` / `HlsData::size()` 计
  (writer 来自外部 crate,拿不到逐条写出量;修复只增删极少量 tag)
- ffmpeg / streamlink:子进程自己落盘,改为每秒读一次 `.part` 长度做差分

`DownloadTask` 持有 `RateMeter`:1 秒采样、10 秒滑动窗口取首尾差值。窗口明显长于
GOP,避免 httpflv 按关键帧批量落盘造成的抖动;采样任务随 `execute` 启停。

`GET /v1/streamers` 每项新增 `live_bytes_per_sec`(字节/秒):未录制、采样不足、
或下载器不经本进程写盘(边录边传、yt-dlp)时为 null。

Part of #1708

* feat(live): 解析主播头像、补斗鱼封面,新增录制中封面/头像的同源代理端点

`LiveStream` 新增 `avatar_url: Option<String>`:
- B 站:`getInfoByRoom` 同一次响应里的 `anchor_info.base_info.face`
- 抖音:房间对象 `owner.avatar_thumb.url_list[0]`(web / H5 两个接口都有)
- 斗鱼:betard 的 `owner_avatar`,缺失时退回 `avatar.{middle,big,small}`;
  封面 `room_pic` 一并补上(此前写死空串)
- 虎牙:`gameLiveInfo` / `liveData` 的 `avatar180`,与 screenshot 一样升到 https
- 其余平台为 None,不改行为

`DownloadTask` 在内存里保存当前录制的封面与头像地址(每次 check_stream 刷新,
随任务消亡,不落库、不加迁移)。`GET /v1/streamers` 每项新增 `live_cover_url` /
`live_avatar_url`(未录制为 null)。

新增 `GET /v1/streamers/{id}/cover` 与 `/avatar`:页面直接引用图片 CDN 会带上 biliup
的 Referer 被 B 站等拒绝(403),改由服务端以直播间页面地址作 Referer 拉取后转发,
按图片 URL 做 5 分钟内存缓存(上限 64 条、单图 4MB)。只按主播 id 取 worker 里已有的
地址,不接受任意 URL;上游不是图片时不转发。路由注册在 `router()` 内,`--auth` 时
与其它 /v1 接口同一道登录校验。

Part of #1708

* feat(ui): 录制中卡片显示直播间封面、主播头像与实时写盘速率

`StreamerCard`(主页「进行中」与直播管理网格共用)在状态为录制中时:
- 顶部加 16:9 封面缩略图,懒加载,图片走 `/v1/streamers/{id}/cover` 同源代理;
  加载中显示占位底色,加载失败整块消失,卡片回到原有样式
- 名称前加 22px 头像圆图(`/v1/streamers/{id}/avatar`),失败则不占位
- 时长右侧加写盘速率,按 B/s、KB/s、MB/s 自动换单位;后端未采样(null)显示「—」

图片地址带原始 URL 的短哈希作查询参数,换一场直播封面变了就不会沿用旧缓存。
未开播 / 暂停的卡片不变。直播管理列表视图新增「码率」列,主播名前带头像。

不引入新依赖:头像与封面用 next/image(项目已配置 unoptimized),速率格式化与
代理地址拼接放在 `use-dashboard.ts`。

Part of #1708

* perf(live): B 站封面/头像经代理时改取 CDN 缩略图

实测 B 站 `user_cover` 原图可达 11520×8640、4 MB,卡片上只显示成 16:9 缩略图和 22px
头像,原样转发既慢又占缓存。对 `*.hdslb.com` 的地址借 CDN 的 `@宽w_高h_1c` 后缀取
640×360 / 128×128 的裁切图(同一张图从 4 MB 降到约 25 KB);已带后缀或带 query 的
地址、以及其它平台的地址原样使用。

Part of #1708

* refactor(downloader): ffmpeg / streamlink 写盘速率改为解析 stderr 进度行,不再读盘

review 意见:每秒 stat 一次 `.part` 长度会额外占磁盘 IO。改为让子进程自己报告:

- ffmpeg 加 `-progress pipe:2 -nostats`,把 key=value 进度块打到 stderr(不受
  `-loglevel quiet` 影响),`total_size=N` 就是本进程累计写出字节
- streamlink 加 `--progress force`,stderr 不是终端时也按周期打
  `[download] Written 12.34 MiB to <file> (5s @ 2.50 MiB/s)`

都在既有的 `spawn_log` stderr 读取协程里顺带解析(`SubprocessProgress`),进度行降为
debug 日志不再刷 info;累计值只加增量、回退时重置基准。解析不出来只是没有速率,
不影响录制。删掉 `FileSizeProbe`。

Part of #1708

* feat(live): 主播头像开播时下载一次存到 data/avatar/,接口直接读本地文件

review 意见:头像几乎不变,不必每次经代理拉取。

- 新增 `common/live_image.rs`:图片抓取(带 Referer、大小与类型校验、B 站缩略图后缀)
  从 `api/live_media.rs` 挪到这里共用;新增 `AvatarStore`,按主播 id 存
  `data/avatar/<id>.<ext>` + `<id>.json`(记录来源地址与 Content-Type),先写临时文件
  再改名,同一时刻只允许一个下载写盘
- 开播进入录制流程时后台下载一次;后续 check_stream 拿到的地址变了才重新下载;
  下载失败保留旧文件、只记日志
- `GET /v1/streamers/{id}/avatar` 改为读本地文件(`Cache-Control: max-age=3600`);
  正在录制而本地还没有时现场下载一次再返回,失败退回旧文件;不在录制时只读本地,
  没有就 404。服务重启后凭 `<id>.json` 直接复用,不重复下载
- 封面仍走内存缓存的代理(每场直播都会换,落盘意义不大)

Part of #1708

* fix(downloader): streamlink 进度行打在 stdout,stdout / stderr 都解析

实测 `--output` 到文件时 streamlink 把控制台输出(含 `[download] Written …`)
打在 stdout 而不是 stderr,只解析 stderr 会得到恒为 0 的速率。两条流共用一个
进度跟踪器,避免重复累加。

Part of #1708
This commit is contained in:
Noah authored and GitHub committed 2026-09-22 06:34:11 +00:00
1 parent 45ab54ba90
commit 5ce14df240
47 files changed
+2101 -102

No files matched your search

+10
View File
@@ -182,10 +182,20 @@
accent-color: var(--semi-color-primary);
cursor: pointer;
}
.cellNameRow {
display: flex;
align-items: center;
gap: 7px;
}
.cellName {
font-weight: 600;
color: var(--semi-color-text-0);
}
/* 列表视图的写盘速率:等宽数字,无采样时为「—」 */
.cellRate {
font-variant-numeric: tabular-nums;
white-space: nowrap;
}
.cellSub {
font-size: 12px;
color: var(--semi-color-text-2);
+34 -3
View File
@@ -33,8 +33,14 @@ import {
import { PauseButton } from '@/app/ui/StreamerActions/PauseButton'
import { platformName, streamerStatusTag } from '@/app/lib/status'
import { hookStepListToForm, formListToHookStep } from '@/app/lib/postprocessor'
import { timeAgo, STREAMERS_REFRESH_MS, SLOW_REFRESH_MS } from '@/app/lib/use-dashboard'
import StreamerCard from '@/app/ui/StreamerCard'
import {
timeAgo,
formatRate,
liveImageUrl,
STREAMERS_REFRESH_MS,
SLOW_REFRESH_MS,
} from '@/app/lib/use-dashboard'
import StreamerCard, { LiveAvatar } from '@/app/ui/StreamerCard'
import PageHeader from '../components/PageHeader'
import styles from './page.module.scss'
@@ -385,6 +391,7 @@ export default function StreamersPage() {
<th>状态</th>
<th>主播</th>
<th>平台</th>
<th>码率</th>
<th>最近录制</th>
<th style={{ textAlign: 'right' }}>操作</th>
</tr>
@@ -392,6 +399,11 @@ export default function StreamersPage() {
<tbody>
{filtered.map((item) => {
const info = infoByUrl.get(item.url)
const live = item.status === 'Working'
const avatarSrc = live
? liveImageUrl(item.id, 'avatar', item.live_avatar_url)
: null
const rate = live ? formatRate(item.live_bytes_per_sec) : null
return (
<tr
key={item.id}
@@ -409,7 +421,10 @@ export default function StreamersPage() {
</td>
<td>{streamerStatusTag(item.status)}</td>
<td>
<div className={styles.cellName}>{item.remark || item.url}</div>
<div className={styles.cellNameRow}>
{avatarSrc ? <LiveAvatar key={avatarSrc} src={avatarSrc} /> : null}
<div className={styles.cellName}>{item.remark || item.url}</div>
</div>
{info?.title ? (
<div className={styles.cellSub}>{info.title}</div>
) : null}
@@ -419,6 +434,22 @@ export default function StreamersPage() {
{platformName(item.url)}
</Text>
</td>
<td>
<Text
type={rate ? 'secondary' : 'tertiary'}
size="small"
className={styles.cellRate}
title={
live
? rate
? '写盘速率(最近 10 秒平均)'
: '写盘速率:尚无采样'
: undefined
}
>
{live ? (rate ?? '—') : '—'}
</Text>
</td>
<td>
<Text type="tertiary" size="small">
{info?.date ? timeAgo(info.date) : '—'}
+6
View File
@@ -105,6 +105,12 @@ export interface LiveStreamerEntity {
upload_streamers_id?: number | null;
status?: string;
upload_status?: string;
/** 录制中最近一个滑动窗口的写盘速率(字节/秒);未录制 / 尚无采样 / 下载器不支持时为 null */
live_bytes_per_sec?: number | null;
/** 录制中的直播间封面地址;图片请走 /v1/streamers/{id}/cover 代理 */
live_cover_url?: string | null;
/** 录制中的主播头像地址;图片请走 /v1/streamers/{id}/avatar 代理 */
live_avatar_url?: string | null;
statusTag?: React.ReactNode;
format?: string;
time_range?: string | Date[];
+28 -1
View File
@@ -1,6 +1,6 @@
'use client'
import useSWR from 'swr'
import { fetcher, LiveStreamerEntity, StreamerInfo, FileList } from './api-streamer'
import { API_BASE, fetcher, LiveStreamerEntity, StreamerInfo, FileList } from './api-streamer'
import { platformName } from './status'
/**
@@ -52,6 +52,33 @@ export function formatDuration(sec: number): string {
return `${Math.floor(sec / 3600)}h ${Math.floor((sec % 3600) / 60)}min`
}
/**
* 写盘速率(字节/秒) → "1.2 MB/s" / "680 KB/s"。
* 后端没有采样(null / undefined)时返回 null,由调用方显示「—」。
*/
export function formatRate(bytesPerSec: number | null | undefined): string | null {
if (bytesPerSec === null || bytesPerSec === undefined || !Number.isFinite(bytesPerSec)) return null
const b = Math.max(0, bytesPerSec)
if (b >= 1 << 20) return `${(b / (1 << 20)).toFixed(b >= 10 << 20 ? 0 : 1)} MB/s`
if (b >= 1 << 10) return `${(b / (1 << 10)).toFixed(0)} KB/s`
return `${Math.round(b)} B/s`
}
/**
* 录制中直播间的封面 / 头像同源代理地址。
* 带上原始地址的短哈希作查询参数:封面换了(新一场直播)浏览器就不会沿用旧缓存。
*/
export function liveImageUrl(
id: number,
kind: 'cover' | 'avatar',
sourceUrl: string | null | undefined
): string | null {
if (!sourceUrl) return null
let h = 5381
for (let i = 0; i < sourceUrl.length; i++) h = ((h << 5) + h + sourceUrl.charCodeAt(i)) | 0
return `${API_BASE}/v1/streamers/${id}/${kind}?v=${(h >>> 0).toString(36)}`
}
const DAY = 86400
const EVENT_WINDOW = DAY // 事件流只展示最近 24h
+70 -8
View File
@@ -1,8 +1,9 @@
'use client'
import React from 'react'
import React, { useState } from 'react'
import Image from 'next/image'
import { LiveStreamerEntity, StreamerInfo } from '@/app/lib/api-streamer'
import { streamerStatusMeta, uploadStatusTag, platformName } from '@/app/lib/status'
import { timeAgo, formatDuration } from '@/app/lib/use-dashboard'
import { timeAgo, formatDuration, formatRate, liveImageUrl } from '@/app/lib/use-dashboard'
import styles from './streamer-card.module.scss'
export interface StreamerCardProps {
@@ -35,9 +36,50 @@ function platTint(color: string): string {
return color + '1a'
}
/**
* 录制中的直播间封面缩略图(16:9,懒加载)。
* 加载失败就整块消失,卡片回到没有封面时的样式;加载中显示占位底色。
* 调用方用 key={src} 挂载,封面地址变化时状态自然重置。
*/
export function LiveCover({ src, alt }: { src: string; alt: string }) {
const [state, setState] = useState<'loading' | 'ok' | 'error'>('loading')
if (state === 'error') return null
return (
<figure className={styles.cover} data-state={state}>
<Image
src={src}
alt={alt}
fill
sizes="(max-width: 600px) 100vw, 400px"
loading="lazy"
onLoad={() => setState('ok')}
onError={() => setState('error')}
/>
</figure>
)
}
/** 录制中的主播头像小圆图;加载失败则不占位。 */
export function LiveAvatar({ src }: { src: string }) {
const [failed, setFailed] = useState(false)
if (failed) return null
return (
<Image
className={styles.avatar}
src={src}
alt=""
aria-hidden="true"
width={22}
height={22}
loading="lazy"
onError={() => setFailed(true)}
/>
)
}
/**
* 全局统一的直播间卡片:
* 状态徽章 + 平台色 chip + 名称 + 实时时长(直播中)/ 最近录制(未开播)
* 状态徽章 + 平台色 chip + [录制中:封面 / 头像 / 写盘速率] + 名称 + 实时时长(直播中)/ 最近录制(未开播)
* + 标题 + URL + 上传状态。主页与直播管理页共用。
*/
export default function StreamerCard({
@@ -60,6 +102,11 @@ export default function StreamerCard({
}
const lastRec = info?.date && !live ? timeAgo(info.date) : null
// 录制中才有封面 / 头像 / 速率;图片走同源代理,避开图片 CDN 的 Referer 校验
const coverSrc = live ? liveImageUrl(streamer.id, 'cover', streamer.live_cover_url) : null
const avatarSrc = live ? liveImageUrl(streamer.id, 'avatar', streamer.live_avatar_url) : null
const rate = live ? formatRate(streamer.live_bytes_per_sec) : null
const plat = platformName(streamer.url)
const platColor = PLAT_COLORS[plat] ?? DEFAULT_PLAT_COLOR
@@ -100,13 +147,28 @@ export default function StreamerCard({
</span>
</div>
{/* 名称 + 实时时长 / 最近录制 */}
{/* 录制中:直播间封面 */}
{coverSrc ? <LiveCover key={coverSrc} src={coverSrc} alt={`${name} 的直播间封面`} /> : null}
{/* 名称(录制中带头像)+ 实时时长与写盘速率 / 最近录制 */}
<div className={styles.nameRow}>
<div className={styles.name} title={name}>
{name}
<div className={styles.nameWrap}>
{avatarSrc ? <LiveAvatar key={avatarSrc} src={avatarSrc} /> : null}
<div className={styles.name} title={name}>
{name}
</div>
</div>
{duration ? (
<span className={styles.liveInfo}>{duration}</span>
{live ? (
<span className={styles.liveStats}>
{duration ? <span className={styles.liveInfo}>{duration}</span> : null}
<span
className={styles.rate}
title={rate ? '写盘速率(最近 10 秒平均)' : '写盘速率:尚无采样'}
aria-label={rate ? `写盘速率 ${rate}` : '写盘速率尚无采样'}
>
{rate ?? '—'}
</span>
</span>
) : lastRec ? (
<span className={styles.recInfo}>{lastRec}</span>
) : null}
+59 -1
View File
@@ -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;
+15 -3
View File
@@ -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))
@@ -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<HashMap<String, CachedImage>>,
}
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<CachedImage> {
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<CachedImage, String> {
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<Arc<DownloadManager>>,
State(proxy): State<Arc<ImageProxy>>,
Path(id): Path<i64>,
) -> 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<Arc<DownloadManager>>,
Path(id): Path<i64>,
) -> 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<Mutex<Vec<Option<String>>>> = 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 { "<html>403</html>" }))
.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());
}
}
+2
View File
@@ -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;
@@ -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<String>,
pub avatar_url: Option<String>,
}
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<Arc<Mutex<SyncSession>>>,
/// 写盘速率表;各下载器拿它的计数器句柄累加,采样任务随 `execute` 启停。
meter: Arc<RateMeter>,
media: std::sync::RwLock<LiveMedia>,
}
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<u64> {
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<dyn LivePlugin + Send + Sync>,
rooms_handle: Arc<Monitor>,
) -> 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<DownloadStatus> {
// 获取配置和主播信息
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<UploaderMessage>,
rooms_handle: Arc<Monitor>,
) {
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();
@@ -0,0 +1,478 @@
//! 直播间封面 / 主播头像的抓取与落盘。
//!
//! 页面直接 `<img src="https://i0.hdslb.com/...">` 会带上 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<FetchedImage, String> {
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<String> {
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,
}
/// 主播头像的本地存储:`<dir>/<主播 id>.<ext>` + `<dir>/<主播 id>.json`。
pub struct AvatarStore {
dir: PathBuf,
client: Client,
/// 同一时刻只让一个下载在写盘,避免开播时的预下载和页面首次请求撞车
write_lock: Mutex<()>,
}
/// 进程内共用的头像存储,录制流程与 Web 接口都用它。
pub fn avatar_store() -> &'static AvatarStore {
static STORE: LazyLock<AvatarStore> = LazyLock::new(|| AvatarStore::new(AVATAR_DIR));
&STORE
}
impl AvatarStore {
pub fn new(dir: impl Into<PathBuf>) -> 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<StoredAvatar> {
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<StoredAvatar, String> {
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<String>, 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"<html>"), None);
}
/// (请求路径, Referer) 记录
type Seen = Arc<StdMutex<Vec<(String, Option<String>)>>>;
/// 一个记下每次请求 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()
);
}
}
@@ -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;
@@ -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<VecDeque<(Instant, u64)>>,
}
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<u64> {
self.bytes_per_sec_at(Instant::now())
}
fn bytes_per_sec_at(&self, now: Instant) -> Option<u64> {
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<RateMeter>) -> 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::<u64>()
{
self.advance_to(total, counter);
}
true
}
/// streamlink `--progress=force` 的 `[download] Written 12.34 MiB to <file> (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<u64> {
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));
}
}
@@ -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 {
@@ -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<String>, 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<Option<tokio::process::Child>>,
bytes_written: ByteCounter,
) -> AppResult<ExitStatus> {
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));
@@ -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<SegmentClosed>,
token: CancellationToken,
callback: &mut (dyn FnMut(SegmentEvent) + Send + Sync + 'a),
byte_meter: (ByteCounter, fn(&P::Item) -> usize),
) -> Result<WriterStats, String>
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<Mutex<Vec<SegmentInfo>>> = 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,
@@ -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)?;
@@ -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<Option<Child>>,
bytes_written: ByteCounter,
) -> AppResult<ExitStatus> {
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:?}"
);
}
}
@@ -129,6 +129,7 @@ impl Context {
// output_dir: PathBuf::from("./downloads")
output_dir: PathBuf::from("."),
suffix,
bytes_written: Default::default(),
}
}
}
@@ -13,4 +13,12 @@ pub struct LiveStreamerResponse {
pub status: String,
/// 上传状态
pub upload_status: String,
/// 正在录制时最近一个滑动窗口内的写盘速率(字节/秒)。
/// 未录制、尚无采样、或下载器不经过本进程写盘(边录边传、yt-dlp)时为 `null`。
pub live_bytes_per_sec: Option<u64>,
/// 正在录制的直播间封面地址;图片本身请走 `GET /v1/streamers/{id}/cover` 代理。
pub live_cover_url: Option<String>,
/// 正在录制的主播头像地址;图片本身请走 `GET /v1/streamers/{id}/avatar` 代理。
pub live_avatar_url: Option<String>,
}
@@ -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<ImageProxy>,
}
/// 简单的服务容器,负责管理API端点通过axum扩展获取的各种服务
@@ -63,6 +67,7 @@ impl ServiceRegister {
config: config.clone(),
client,
log_handle,
image_proxy: Arc::new(ImageProxy::new()),
}
}
+4
View File
@@ -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",
@@ -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())
}
+9 -2
View File
@@ -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<u64> {
async fn download_to_file(
url: Url,
client: &StatelessClient,
out: &mut impl Write,
bytes_written: &ByteCounter,
) -> Result<u64> {
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)
+42 -10
View File
@@ -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<dyn std::error::Error>> {
use crate::downloader::util::{LifecycleFile, Segmentable};
/// 一段最小的纯视频 FLV 流体(不含 9 字节文件头):
/// onMetaData 脚本标签 + 一个 H264 序列头关键帧,没有任何音频标签。
fn pure_video_flv_body() -> Vec<u8> {
let mut data: Vec<u8> = 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<dyn std::error::Error>> {
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<dyn std::error::Error>> {
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(())
}
}
@@ -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(),
@@ -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(),
@@ -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(),
+94 -47
View File
@@ -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<Option<BiliRoomProfile>> {
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());
}
}
+1
View File
@@ -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(),
+35 -7
View File
@@ -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<String> {
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() {
+69 -1
View File
@@ -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<String>,
/// 主播头像直链
#[serde(default)]
owner_avatar: Option<String>,
/// 头像的多尺寸版本,形如 `{"big": ..., "middle": ..., "small": ...}`;
/// 个别房间只给这一份,作为 `owner_avatar` 缺失时的后备
#[serde(default)]
avatar: Option<Value>,
}
impl RoomInfo {
fn avatar_url(&self) -> Option<String> {
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);
}
}
@@ -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(),
+25
View File
@@ -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<Value>,
stream_info: Vec<Value>,
}
/// 虎牙图片地址常带 `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")), "");
}
}
@@ -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(),
@@ -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()
@@ -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(),
@@ -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(),
+3
View File
@@ -336,6 +336,9 @@ pub struct LiveStream {
pub title: String,
pub date: DateTime<Utc>,
pub live_cover_url: String,
/// 主播头像地址;平台响应里没有或尚未解析的平台为 `None`。
#[serde(default)]
pub avatar_url: Option<String>,
pub raw_stream_url: String,
pub platform: String,
pub stream_headers: HashMap<String, String>,
@@ -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(),
@@ -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(),
@@ -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(),
@@ -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(),
@@ -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(),
@@ -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(),
+1
View File
@@ -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(),
+47
View File
@@ -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<dyn FnMut(&str) + Send + Sync + 'a>;
/// 已写盘字节的原子累计,供录制线程之外(如 Web 接口)读取实时速率。
///
/// 写盘路径上只做一次 `fetch_add`:不加锁、不 await、不会失败,
/// 因此不改变录制的任何控制流。`Clone` 得到的是同一计数器的另一个句柄。
#[derive(Debug, Clone, Default)]
pub struct ByteCounter(Arc<AtomicU64>);
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<dyn std::error::Error>> {
// 测试时间分割