feat: live-url 同一房间 5 秒去抖(失败重取带 fresh=1 绕过);斗鱼网宿直链追加 expire=0,不再每 5 分钟断流 (#1713)

* feat(preview): live-url 同一房间 5 秒去抖,播放失败后的重取带 fresh=1 绕过

每次打开预览都向平台 API 取一条新直链(带 cookie 时消耗登录态配额,B 站接口有频控)。同一房间
5 s 内的重复请求(StrictMode 双调用、连点、同一房间多个小窗)复用上一次结果并标 cached: true,
不再打平台;前端播放失败后的重取传 ?fresh=1 一定拿新 token(抖音某节点 404、直链到期都靠它)。
缓存插入时顺手清掉一分钟前的条目,表不会长。

Part of #1708

* feat(douyu): 网宿直链末尾追加重复的 expire=0,不再每 5 分钟断流切文件

斗鱼网宿 CDN 按直链里的 expire=300 每 300 s 关闭一次连接,stream-gears 录斗鱼因此
5 分钟一个文件。实测网宿的 wsAuth 只校验第一个 expire、连接寿命取最后一个:保留原值、
末尾再追加 &expire=0 能通过校验且不再断开;删掉或改写原值都会 403。

- douyu:只对网宿直链(host 为 ws*.douyucdn.cn 或 fcdn=ws)、且只有一个非 0 expire 时追加;
  火山(force_hs)、华为等其它线路与已是 expire=0 的直链不动。
- stream-gears:首连 403 且是追加过的直链时,本次改用原直链再连一次(同一 token 仍有效),
  之后照旧按 300 s 断开续录;每次拉流最多多一次请求,不进外层重试。
- 其它下载器(ffmpeg / mesio / streamlink / 边录边传)与预览直连没有这层兜底,照旧用原直链。
This commit is contained in:
Noah authored and GitHub committed 2026-09-23 11:06:06 +00:00
1 parent 48dc8fd65b
commit 3ab1a4630b
8 files changed
+320 -16

No files matched your search

+2
View File
@@ -124,6 +124,8 @@ export interface LiveUrlInfo {
/** 过期时间估计(Unix 秒),直链里没有可识别的过期参数时为 null */
expires_at: number | null;
direct: DirectCapability;
/** 是后端 5 s 去抖窗口内复用的上一次结果,不是新取的 */
cached: boolean;
}
/** 直播预览的取流方式(全局配置 preview_transport),空值视同 relay */
+7 -3
View File
@@ -121,9 +121,13 @@ export function livePreviewUrl(id: number, snapshotMs?: number): string {
return snapshotMs === undefined ? base : `${base}?snapshot_ms=${Math.max(0, Math.round(snapshotMs))}`
}
/** 浏览器直连模式:向后端要当前录制中那条流的 CDN 直链。 */
export function fetchLiveUrl(id: number): Promise<LiveUrlInfo> {
return fetcher(`/v1/streamers/${id}/live-url`, { cache: 'no-store' })
/**
* 浏览器直连模式:让后端向平台新取一条 CDN 直链(不复用录制那条)。
* 后端对同一房间 5 s 内的重复请求复用上一次结果;播放失败后的重取传 `fresh`,一定拿新 token。
*/
export function fetchLiveUrl(id: number, opts: { fresh?: boolean } = {}): Promise<LiveUrlInfo> {
const query = opts.fresh ? '?fresh=1' : ''
return fetcher(`/v1/streamers/${id}/live-url${query}`, { cache: 'no-store' })
}
/**
+1 -1
View File
@@ -219,7 +219,7 @@ export function LivePreviewPlayer({
}
setPhase('reconnecting')
setMessage('直连中断,重新获取直链…')
fetchLiveUrl(streamer.id).then(
fetchLiveUrl(streamer.id, { fresh: true }).then(
(info) => {
if (info.direct.capable && info.url) {
setOverride({
@@ -18,13 +18,13 @@ use axum::extract::{Path, Query, State};
use axum::http::{HeaderValue, StatusCode, header};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use biliup::downloader::live::LiveStatus;
use biliup::downloader::live::{LiveStatus, strip_ws_expire_override};
use biliup::downloader::preview::{PreviewFormat, PreviewHub, SubscribeError, Subscription};
use bytes::{Bytes, BytesMut};
use danmaku_client::DanmakuEvent;
use std::collections::VecDeque;
use std::collections::{HashMap, VecDeque};
use std::sync::{Arc, OnceLock};
use std::time::Duration;
use std::time::{Duration, Instant};
use tokio::sync::broadcast::error::RecvError;
use tokio::sync::{OwnedSemaphorePermit, Semaphore, broadcast};
use tracing::{debug, info, warn};
@@ -175,7 +175,7 @@ pub fn estimate_expiry(stream_url: &str) -> Option<i64> {
/// `GET /v1/streamers/{id}/live-url` 的响应:当前录制中那条流的 CDN 直链。
#[derive(Debug, Clone, serde::Serialize, PartialEq, Eq)]
pub struct LiveUrlResponse {
/// 直链,含 CDN 参数;与录制用的是同一条
/// 向平台新取的直链,含 CDN 参数;与录制那条不是同一个 token
pub url: String,
/// 浏览器该用哪个播放器:`flv`(mpegts.js)/ `hls`(hls.js,TS 或 fMP4 分片都行);后缀猜不出为 `null`
pub format: Option<&'static str>,
@@ -183,6 +183,52 @@ pub struct LiveUrlResponse {
/// 过期时间估计(Unix 秒);直链里没有可识别的过期参数时为 `null`
pub expires_at: Option<i64>,
pub direct: DirectCapability,
/// 这条是 [`LIVE_URL_DEBOUNCE`] 内复用的上一次结果(同一房间短时间内重复打开),不是新取的
pub cached: bool,
}
/// 同一房间两次 `live-url` 之间少于这个间隔就复用上一次的直链,不再打平台 API
/// (StrictMode 双调用、连点、同一房间几个小窗同时打开)。播放失败后的重取带 `?fresh=1` 绕过它。
pub const LIVE_URL_DEBOUNCE: Duration = Duration::from_secs(5);
#[derive(Debug, Default, serde::Deserialize)]
pub struct LiveUrlQuery {
/// 为 `1` / `true` 时忽略去抖缓存,一定向平台新取(前端播放失败后的重取用)
#[serde(default)]
pub fresh: Option<String>,
}
impl LiveUrlQuery {
fn wants_fresh(&self) -> bool {
matches!(self.fresh.as_deref(), Some("1") | Some("true"))
}
}
/// 每个房间最近一次成功取到的直链与时刻。条目只在 [`LIVE_URL_DEBOUNCE`] 内有意义,插入时顺手
/// 清掉一分钟前的,表不会长。
fn live_url_cache() -> &'static std::sync::Mutex<HashMap<i64, (Instant, LiveUrlResponse)>> {
static CACHE: OnceLock<std::sync::Mutex<HashMap<i64, (Instant, LiveUrlResponse)>>> =
OnceLock::new();
CACHE.get_or_init(|| std::sync::Mutex::new(HashMap::new()))
}
/// 去抖窗口内有没有可复用的直链。
pub fn recent_live_url(id: i64, now: Instant) -> Option<LiveUrlResponse> {
let cache = live_url_cache().lock().unwrap();
cache
.get(&id)
.filter(|(at, _)| now.duration_since(*at) < LIVE_URL_DEBOUNCE)
.map(|(_, response)| LiveUrlResponse {
cached: true,
..response.clone()
})
}
/// 记下这次取到的直链;顺手清掉过期很久的条目。
pub fn remember_live_url(id: i64, now: Instant, response: &LiveUrlResponse) {
let mut cache = live_url_cache().lock().unwrap();
cache.retain(|_, (at, _)| now.duration_since(*at) < Duration::from_secs(60));
cache.insert(id, (now, response.clone()));
}
/// `GET /v1/streamers/{id}/live-url`:浏览器直连模式用。
@@ -194,6 +240,7 @@ pub struct LiveUrlResponse {
pub async fn get_live_url(
State(managers): State<Arc<DownloadManager>>,
Path(id): Path<i64>,
Query(query): Query<LiveUrlQuery>,
) -> Response {
let Some(worker) = managers.get_room_by_id(id).await else {
return (StatusCode::NOT_FOUND, "直播间不存在").into_response();
@@ -211,9 +258,17 @@ pub async fn get_live_url(
format: direct_format(&source.url),
platform: source.platform,
url: String::new(),
cached: false,
};
return no_store(axum::Json(response).into_response());
}
let now = Instant::now();
if !query.wants_fresh()
&& let Some(recent) = recent_live_url(id, now)
{
debug!(id, "预览直链在去抖窗口内,复用上一次的");
return no_store(axum::Json(recent).into_response());
}
let room_url = worker.get_streamer().url.clone();
let Some(plugin) = managers.plugin_for(&room_url).await else {
return (StatusCode::SERVICE_UNAVAILABLE, "找不到该直播间的平台插件").into_response();
@@ -236,15 +291,20 @@ pub async fn get_live_url(
.into_response();
}
};
let url = fresh.raw_stream_url;
// 浏览器那边没有 403 兜底,直连照旧给原直链
let url = strip_ws_expire_override(&fresh.raw_stream_url)
.map(str::to_owned)
.unwrap_or(fresh.raw_stream_url);
let response = LiveUrlResponse {
direct: direct_capability(&fresh.platform),
expires_at: estimate_expiry(&url),
format: direct_format(&url),
platform: fresh.platform,
url,
cached: false,
};
info!(id, "已向平台取到预览直链");
remember_live_url(id, now, &response);
info!(id, fresh = query.wants_fresh(), "已向平台取到预览直链");
no_store(axum::Json(response).into_response())
}
@@ -806,6 +866,61 @@ mod tests {
assert_eq!(direct_format("https://x/y.mp4"), None);
}
/// 去抖:5 s 内同一房间复用上一次的直链(标 `cached`),不同房间互不影响,过了窗口就不复用。
#[test]
fn live_url_is_debounced_per_room() {
let sample = |url: &str| LiveUrlResponse {
url: url.into(),
format: Some("flv"),
platform: "douyu".into(),
expires_at: None,
direct: DirectCapability {
capable: true,
reason: None,
},
cached: false,
};
let t0 = Instant::now();
assert!(recent_live_url(9001, t0).is_none());
remember_live_url(9001, t0, &sample("https://cdn/a.flv?token=1"));
let hit = recent_live_url(9001, t0 + Duration::from_secs(4)).unwrap();
assert!(hit.cached);
assert_eq!(hit.url, "https://cdn/a.flv?token=1");
assert!(recent_live_url(9002, t0 + Duration::from_secs(1)).is_none());
assert!(recent_live_url(9001, t0 + LIVE_URL_DEBOUNCE).is_none());
// 新取的覆盖旧的
remember_live_url(
9001,
t0 + Duration::from_secs(10),
&sample("https://cdn/a.flv?token=2"),
);
assert_eq!(
recent_live_url(9001, t0 + Duration::from_secs(11))
.unwrap()
.url,
"https://cdn/a.flv?token=2"
);
assert!(
LiveUrlQuery {
fresh: Some("1".into())
}
.wants_fresh()
);
assert!(
LiveUrlQuery {
fresh: Some("true".into())
}
.wants_fresh()
);
assert!(
!LiveUrlQuery {
fresh: Some("0".into())
}
.wants_fresh()
);
assert!(!LiveUrlQuery::default().wants_fresh());
}
#[test]
fn expiry_is_read_from_known_query_parameters() {
assert_eq!(
@@ -14,7 +14,7 @@ use crate::server::errors::{AppError, AppResult};
use crate::server::infrastructure::context::{Context, Stage, WorkerStatus};
use crate::server::infrastructure::models::hook_step::process;
use async_channel::Sender;
use biliup::downloader::live::{LivePlugin, LiveStatus, LiveStream};
use biliup::downloader::live::{LivePlugin, LiveStatus, LiveStream, strip_ws_expire_override};
use biliup::downloader::preview::PreviewHub;
use danmaku_client::DanmakuEvent;
use error_stack::ResultExt;
@@ -422,6 +422,12 @@ impl DownloadTask {
// 获取配置和主播信息
let streamer = ctx.live_streamer();
let mut download_config = ctx.download_config(stream);
// 只有 stream-gears 会在网宿 403 时退回原直链,其它下载器仍用原直链
if !matches!(self.downloader, DownloaderRuntime::StreamGears(_))
&& let Some(original) = strip_ws_expire_override(&download_config.url)
{
download_config.url = original.to_string();
}
download_config.bytes_written = self.meter.counter();
download_config.preview = self.preview.clone();
if let crate::server::core::downloader::DownloaderRuntime::Sync(sync) = &self.downloader {
@@ -5,15 +5,17 @@ use crate::server::errors::{AppError, AppResult};
use biliup::client::StatelessClient;
use biliup::downloader::flv_parser::header;
use biliup::downloader::httpflv::Connection;
use biliup::downloader::live::strip_ws_expire_override;
use biliup::downloader::preview::PreviewFormat;
use biliup::downloader::util::{LifecycleFile, Segmentable};
use biliup::downloader::{hls, httpflv};
use error_stack::{ResultExt, bail};
use nom::Err;
use reqwest::{Response, StatusCode};
use std::path::PathBuf;
use std::sync::RwLock;
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info};
use tracing::{debug, error, info, warn};
/// Stream-gears下载器实现
/// 使用stream-gears库进行直播流下载
@@ -45,7 +47,6 @@ impl StreamGears {
mut callback: Box<dyn FnMut(SegmentEvent) + Send + Sync + 'a>,
download_config: DownloadConfig,
) -> AppResult<DownloadStatus> {
let url = download_config.url.clone();
let file_name = download_config.recorder.filename_template();
let headers_in = construct_headers(&download_config.headers).map_err(AppError::Custom)?;
let proxy = self.proxy.clone();
@@ -61,8 +62,7 @@ impl StreamGears {
// 创建HTTP客户端
let client = StatelessClient::new(headers_in, proxy.as_deref());
// 获取可重试的响应
let response = client
.retryable(&url)
let (url, response) = connect(&client, download_config.url.clone())
.await
.change_context(AppError::Unknown)?;
// 创建连接
@@ -121,6 +121,25 @@ impl StreamGears {
}
}
/// 首连;斗鱼网宿直链追加的 `expire=0` 被 403 时,本次改用原直链再连一次。
/// 返回实际连上的直链。
async fn connect(client: &StatelessClient, url: String) -> reqwest::Result<(String, Response)> {
match client.retryable(&url).await {
Err(e) if e.status() == Some(StatusCode::FORBIDDEN) => {
let Some(original) = strip_ws_expire_override(&url) else {
return Err(e);
};
warn!(
"网宿拒绝了追加 expire=0 的直链(403),本次改用原直链,连接仍会按 expire 定时断开"
);
let original = original.to_string();
let response = client.retryable(&original).await?;
Ok((original, response))
}
result => result.map(|response| (url, response)),
}
}
impl StreamGears {
/// 开始下载流
///
@@ -149,3 +168,72 @@ impl StreamGears {
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use axum::Router;
use axum::extract::RawQuery;
use axum::http::StatusCode as HttpStatus;
use axum::routing::get;
use reqwest::header::HeaderMap;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
/// 模拟网宿修掉重复参数的解析差异:带重复 expire 的一律 403,原直链 200
async fn wangsu_rejecting_duplicate_expire() -> (String, Arc<AtomicUsize>) {
let hits = Arc::new(AtomicUsize::new(0));
let counter = hits.clone();
let app = Router::new().route(
"/live/a.flv",
get(move |RawQuery(query): RawQuery| {
counter.fetch_add(1, Ordering::SeqCst);
async move {
let expires = query
.unwrap_or_default()
.split('&')
.filter(|pair| pair.starts_with("expire="))
.count();
if expires > 1 {
(HttpStatus::FORBIDDEN, "Invalid Request")
} else {
(HttpStatus::OK, "FLV")
}
}
}),
);
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}/live/a.flv"), hits)
}
#[tokio::test]
async fn connect_falls_back_to_original_url_once_on_403() {
let (base, hits) = wangsu_rejecting_duplicate_expire().await;
let original = format!("{base}?wsAuth=a&token=t&expire=300&fcdn=ws");
let client = StatelessClient::new(HeaderMap::new(), None);
let (url, response) = connect(&client, format!("{original}&expire=0"))
.await
.unwrap();
assert_eq!(url, original);
assert_eq!(response.status(), StatusCode::OK);
assert_eq!(hits.load(Ordering::SeqCst), 2);
}
#[tokio::test]
async fn connect_does_not_retry_403_on_other_urls() {
let (base, hits) = wangsu_rejecting_duplicate_expire().await;
let client = StatelessClient::new(HeaderMap::new(), None);
// 不是网宿直链:403 原样返回,不另发请求
let err = connect(&client, format!("{base}?expire=300&fcdn=hw&expire=0"))
.await
.unwrap_err();
assert_eq!(err.status(), Some(StatusCode::FORBIDDEN));
assert_eq!(hits.load(Ordering::SeqCst), 1);
}
}
@@ -26,6 +26,11 @@ const DOUYU_HS_CDN: &str = "hs-h5";
const DOUYU_P2P_DOMAIN_TCT: &str = "hdltctwk.douyucdn.cn";
const DOUYU_P2PSDK_APIS: [&str; 2] = ["https://sdkapiv4.douyucdn.cn", "https://sdkapi.douyucdn.cn"];
const DOUYU_USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36";
/// 网宿按直链里**最后一个** `expire` 限制单条连接寿命(`expire=300` 即 300 s 断一次),
/// `wsAuth` 却只校验**第一个**:保留原值、末尾再追加 `expire=0` 能过校验且不再按时断开,
/// 删掉或改写原值则 403。这是 CDN 未文档化的行为,被 403 时拉流端用
/// [`strip_ws_expire_override`] 退回原直链。
const WS_EXPIRE_OVERRIDE: &str = "&expire=0";
pub struct Douyu {
re: Regex,
@@ -140,6 +145,7 @@ impl<'a> DouyuLive<'a> {
let play_info = self.get_web_play_info(&room_id).await?;
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 raw_stream_url = with_ws_expire_override(raw_stream_url);
let avatar_url = room_info.avatar_url();
Ok(LiveStatus::Live {
@@ -690,6 +696,45 @@ fn md5_hex(input: String) -> String {
format!("{:x}", hasher.finalize())
}
/// 网宿直链:host 为 `ws*.douyucdn.cn`,或带 `fcdn=ws`。
/// 不看 `rtmp_cdn`:`douyu_force_hs` 构造火山直链后它仍是 `ws-h5`。
fn is_wangsu_stream(url: &Url) -> bool {
let ws_host = url.host_str().is_some_and(|host| {
host.starts_with("ws")
&& (host.ends_with(".douyucdn.cn") || host.ends_with(".douyucdn2.cn"))
});
ws_host
|| url
.query_pairs()
.any(|(key, value)| key == "fcdn" && value == "ws")
}
/// 网宿直链里只有一个 `expire` 且不为 0 时才需要追加。
fn ws_expire_needs_override(url: &str) -> bool {
let Ok(parsed) = Url::parse(url) else {
return false;
};
if !is_wangsu_stream(&parsed) {
return false;
}
let mut expires = parsed.query_pairs().filter(|(key, _)| key == "expire");
matches!((expires.next(), expires.next()), (Some((_, value)), None) if value != "0")
}
fn with_ws_expire_override(url: String) -> String {
if ws_expire_needs_override(&url) {
url + WS_EXPIRE_OVERRIDE
} else {
url
}
}
/// 若是追加过 `expire=0` 的网宿直链,返回追加前的原直链;否则 `None`。
pub fn strip_ws_expire_override(url: &str) -> Option<&str> {
url.strip_suffix(WS_EXPIRE_OVERRIDE)
.filter(|original| ws_expire_needs_override(original))
}
fn parse_stream_url(input: &str) -> LiveResult<(String, Vec<(String, String)>)> {
let parsed = Url::parse(input)
.map_err(|err| LiveError::custom(format!("解析斗鱼 huos 源链接失败: {err}")))?;
@@ -931,6 +976,50 @@ mod tests {
);
}
const WS_URL: &str = "https://ws1a.douyucdn.cn/live/24422abc_4000.flv?wsAuth=a1&token=t1&logo=0&expire=300&did=d&origin=dy&fcdn=ws&fo=0&mix=0&isp=";
#[test]
fn ws_expire_override_appends_after_original_expire() {
let overridden = with_ws_expire_override(WS_URL.to_string());
assert_eq!(overridden, format!("{WS_URL}&expire=0"));
assert_eq!(strip_ws_expire_override(&overridden), Some(WS_URL));
// 已追加过的不再追加
assert_eq!(with_ws_expire_override(overridden.clone()), overridden);
// 只靠 fcdn=ws 也能认出网宿
let by_fcdn = "https://cdn.example/live/a.flv?wsAuth=a&expire=300&fcdn=ws";
assert_eq!(
with_ws_expire_override(by_fcdn.to_string()),
format!("{by_fcdn}&expire=0")
);
}
#[test]
fn ws_expire_override_skips_expire_zero() {
let url = WS_URL.replace("expire=300", "expire=0");
assert_eq!(with_ws_expire_override(url.clone()), url);
assert_eq!(strip_ws_expire_override(&url), None);
}
#[test]
fn ws_expire_override_skips_non_wangsu() {
for url in [
"https://hw3.douyucdn2.cn/live/abc_4000.flv?wsAuth=a&token=t&expire=300&fcdn=hw",
"http://openflv-huos.douyucdn2.cn/live/abc_4000.xs?wsAuth=a&expire=300&fcdn=hs&txSecret=s&txTime=t&domain=hdltctwk.douyucdn.cn",
"https://tc-tct.douyucdn2.cn/dyliveflv1/abc_4000.flv?wsAuth=a&expire=300&fcdn=tct",
] {
assert_eq!(with_ws_expire_override(url.to_string()), url);
let appended = format!("{url}&expire=0");
assert_eq!(strip_ws_expire_override(&appended), None);
}
}
#[test]
fn ws_expire_override_skips_url_without_expire() {
let url = "https://ws1a.douyucdn.cn/live/abc_4000.flv?wsAuth=a&token=t&fcdn=ws";
assert_eq!(with_ws_expire_override(url.to_string()), url);
assert_eq!(strip_ws_expire_override(&format!("{url}&expire=0")), None);
}
#[tokio::test]
async fn maybe_build_huos_url_falls_back_when_build_fails() {
let encrypt_key_cache = Mutex::new(None);
+1 -1
View File
@@ -35,7 +35,7 @@ pub use bigo::Bigo;
pub use bilibili::Bilibili;
pub use cc::CC;
pub use douyin::Douyin;
pub use douyu::Douyu;
pub use douyu::{Douyu, strip_ws_expire_override};
pub use general::General;
pub use huya::Huya;
pub use inke::Inke;