diff --git a/app/lib/api-streamer.ts b/app/lib/api-streamer.ts index 7a7d7c7..f23ad4d 100644 --- a/app/lib/api-streamer.ts +++ b/app/lib/api-streamer.ts @@ -124,6 +124,8 @@ export interface LiveUrlInfo { /** 过期时间估计（Unix 秒），直链里没有可识别的过期参数时为 null */ expires_at: number | null; direct: DirectCapability; + /** 是后端 5 s 去抖窗口内复用的上一次结果，不是新取的 */ + cached: boolean; } /** 直播预览的取流方式（全局配置 preview_transport），空值视同 relay */ diff --git a/app/lib/use-dashboard.ts b/app/lib/use-dashboard.ts index 54b1856..ab1b808 100644 --- a/app/lib/use-dashboard.ts +++ b/app/lib/use-dashboard.ts @@ -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 { - return fetcher(`/v1/streamers/${id}/live-url`, { cache: 'no-store' }) +/** + * 浏览器直连模式：让后端向平台新取一条 CDN 直链（不复用录制那条）。 + * 后端对同一房间 5 s 内的重复请求复用上一次结果；播放失败后的重取传 `fresh`，一定拿新 token。 + */ +export function fetchLiveUrl(id: number, opts: { fresh?: boolean } = {}): Promise { + const query = opts.fresh ? '?fresh=1' : '' + return fetcher(`/v1/streamers/${id}/live-url${query}`, { cache: 'no-store' }) } /** diff --git a/app/ui/LivePreview.tsx b/app/ui/LivePreview.tsx index 5eaa228..2fa313d 100644 --- a/app/ui/LivePreview.tsx +++ b/app/ui/LivePreview.tsx @@ -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({ diff --git a/crates/biliup-cli/src/server/api/live_preview.rs b/crates/biliup-cli/src/server/api/live_preview.rs index 0a246c6..babab4e 100644 --- a/crates/biliup-cli/src/server/api/live_preview.rs +++ b/crates/biliup-cli/src/server/api/live_preview.rs @@ -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 { /// `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, 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, +} + +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> { + static CACHE: OnceLock>> = + OnceLock::new(); + CACHE.get_or_init(|| std::sync::Mutex::new(HashMap::new())) +} + +/// 去抖窗口内有没有可复用的直链。 +pub fn recent_live_url(id: i64, now: Instant) -> Option { + 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>, Path(id): Path, + Query(query): Query, ) -> 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!( diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index 486f8fb..2bed38a 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -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 { diff --git a/crates/biliup-cli/src/server/core/downloader/stream_gears.rs b/crates/biliup-cli/src/server/core/downloader/stream_gears.rs index d700005..7e15176 100644 --- a/crates/biliup-cli/src/server/core/downloader/stream_gears.rs +++ b/crates/biliup-cli/src/server/core/downloader/stream_gears.rs @@ -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, download_config: DownloadConfig, ) -> AppResult { - 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) { + 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); + } +} diff --git a/crates/biliup/src/downloader/live/douyu.rs b/crates/biliup/src/downloader/live/douyu.rs index 3634fc9..ce02595 100644 --- a/crates/biliup/src/downloader/live/douyu.rs +++ b/crates/biliup/src/downloader/live/douyu.rs @@ -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); diff --git a/crates/biliup/src/downloader/live/mod.rs b/crates/biliup/src/downloader/live/mod.rs index ff838e6..31cc805 100644 --- a/crates/biliup/src/downloader/live/mod.rs +++ b/crates/biliup/src/downloader/live/mod.rs @@ -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;