diff --git a/crates/biliup-cli/src/server/common/download.rs b/crates/biliup-cli/src/server/common/download.rs index 2bed38a..6159f3d 100644 --- a/crates/biliup-cli/src/server/common/download.rs +++ b/crates/biliup-cli/src/server/common/download.rs @@ -5,6 +5,7 @@ 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; +use crate::server::core::downloader::ws_expire::resolve_ws_expire_override; use crate::server::core::downloader::{ DanmakuClient, DownloadStatus, DownloaderRuntime, SegmentEvent, SegmentInfo, }; @@ -19,6 +20,7 @@ use biliup::downloader::preview::PreviewHub; use danmaku_client::DanmakuEvent; use error_stack::ResultExt; use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tokio::sync::{Mutex, Notify, broadcast}; use tokio_util::sync::CancellationToken; @@ -175,6 +177,9 @@ pub struct DownloadTask { /// 实时弹幕广播：弹幕客户端每解出一条就 `send` 一份给预览播放器； /// 平台没有弹幕实现（`stream.danmaku` 为 None）时为 `None`。 danmaku_tx: Option>, + /// 追加 `expire=0` 的网宿直链探测通过、下载器却一个字节没拿到就失败过： + /// 本任务余下的拉流不再追加，直接用原直链（stream-gears 有自己的首连兜底，不受影响）。 + ws_expire_rejected: AtomicBool, } /// 每路实时弹幕广播的槽位数；掉队的订阅者跳过丢掉的那几条继续收，不断开。 @@ -199,6 +204,7 @@ impl DownloadTask { source: std::sync::RwLock::new(LiveSource::from_stream(stream)), preview, danmaku_tx, + ws_expire_rejected: AtomicBool::new(false), } } @@ -303,9 +309,23 @@ impl DownloadTask { // 创建守卫确保清理 // 创建事件处理器 // 执行下载 + let bytes_before = self.meter.counter().total(); let components = self .download(&mut processor, ctx.clone(), danmaku_client.clone(), &stream) .await; + if !matches!(self.downloader, DownloaderRuntime::StreamGears(_)) + && ws_expire_override_failed( + &stream.raw_stream_url, + &components, + self.meter.counter().total() > bytes_before, + ) + && !self.ws_expire_rejected.swap(true, Ordering::Relaxed) + { + warn!( + url = url, + "追加 expire=0 的直链没拉到数据，本次录制余下的拉流改用原直链" + ); + } // 失败原因只藏在结束时的 Debug 输出里会让用户以为下载器“什么都没做” // （典型：边录边传缺少上传模板、cookie 失效）。 @@ -422,11 +442,16 @@ 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(); + // stream-gears 在自己的首连里处理网宿 403，其它下载器拉流前先探一次 + if !matches!(self.downloader, DownloaderRuntime::StreamGears(_)) { + download_config.url = match strip_ws_expire_override(&download_config.url) { + Some(original) if self.ws_expire_rejected.load(Ordering::Relaxed) => { + original.to_string() + } + _ => { + resolve_ws_expire_override(download_config.url, &download_config.headers).await + } + }; } download_config.bytes_written = self.meter.counter(); download_config.preview = self.preview.clone(); @@ -545,6 +570,18 @@ fn download_attempt_progressed(result: &AppResult) -> bool { ) } +/// 用追加 `expire=0` 的网宿直链拉流（探测已通过）却没写出任何字节就失败： +/// 说明网宿对下载器的请求与对 HEAD 探测的判断不一致，不能再信探测结果。 +fn ws_expire_override_failed( + stream_url: &str, + result: &AppResult, + wrote_bytes: bool, +) -> bool { + strip_ws_expire_override(stream_url).is_some() + && !wrote_bytes + && !download_attempt_progressed(result) +} + /// Exponential backoff delay for the current `retry_count`. /// /// `retry_count == 0` means "immediate next segment" (Duration::ZERO). Non-zero @@ -621,6 +658,24 @@ mod tests { use crate::server::errors::AppError; use error_stack::Report; + #[test] + fn ws_expire_override_counts_as_failed_only_without_data() { + let overridden = "https://ws1a.douyucdn.cn/live/a.flv?wsAuth=a&expire=300&fcdn=ws&expire=0"; + let original = "https://ws1a.douyucdn.cn/live/a.flv?wsAuth=a&expire=300&fcdn=ws"; + let failed = Ok(DownloadStatus::Error("FFmpeg error: Some(8)".into())); + let err: AppResult = Err(Report::new(AppError::Unknown)); + + assert!(ws_expire_override_failed(overridden, &failed, false)); + assert!(ws_expire_override_failed(overridden, &err, false)); + assert!(!ws_expire_override_failed(overridden, &failed, true)); + assert!(!ws_expire_override_failed( + overridden, + &Ok(DownloadStatus::StreamEnded), + false + )); + assert!(!ws_expire_override_failed(original, &failed, false)); + } + #[test] fn segment_completed_and_stream_ended_count_as_progress() { assert!(download_attempt_progressed(&Ok( diff --git a/crates/biliup-cli/src/server/core/downloader.rs b/crates/biliup-cli/src/server/core/downloader.rs index d9a0c31..92b092f 100644 --- a/crates/biliup-cli/src/server/core/downloader.rs +++ b/crates/biliup-cli/src/server/core/downloader.rs @@ -8,6 +8,7 @@ pub mod stream_gears; pub mod streamlink; /// 边录边传（零落盘流式上传） pub mod sync_downloader; +pub mod ws_expire; pub mod ytdlp; use crate::server::common::timerange; diff --git a/crates/biliup-cli/src/server/core/downloader/streamlink.rs b/crates/biliup-cli/src/server/core/downloader/streamlink.rs index 02ea60c..4bd3a6d 100644 --- a/crates/biliup-cli/src/server/core/downloader/streamlink.rs +++ b/crates/biliup-cli/src/server/core/downloader/streamlink.rs @@ -116,6 +116,9 @@ pub struct StreamlinkDownloader { url: String, headers: HashMap, output_mode: OutputMode, + /// `url` 就是插件解析出的流直链时，每次拉流改用 `DownloadConfig::url`： + /// 重试会重新解析出新直链（斗鱼网宿的 token 连过一次就作废），且经过了 expire=0 探测 + follow_stream_url: bool, } impl StreamlinkDownloader { @@ -125,9 +128,15 @@ impl StreamlinkDownloader { url, headers: HashMap::new(), output_mode: OutputMode::Pipe, // 默认管道模式 + follow_stream_url: false, } } + pub fn following_stream_url(mut self) -> Self { + self.follow_stream_url = true; + self + } + pub fn with_headers(mut self, headers: HashMap) -> Self { self.headers = headers; self @@ -176,7 +185,12 @@ impl StreamlinkDownloader { args.push("force".to_string()); args.push("--output".to_string()); args.push(output_file.to_string()); - args.push(streamlink_cli_url(&self.url)); + let url = if self.follow_stream_url { + &download_config.url + } else { + &self.url + }; + args.push(streamlink_cli_url(url)); args.push("best".to_string()); Ok(args) } @@ -524,6 +538,49 @@ async fn spawn_log( #[cfg(test)] mod tests { use super::*; + use crate::server::common::util::Recorder; + use crate::server::infrastructure::models::StreamerInfo; + + fn download_config(url: &str) -> DownloadConfig { + DownloadConfig { + url: url.to_string(), + segment_time: None, + time_range: None, + file_size: None, + headers: HashMap::new(), + recorder: Recorder::new( + None, + StreamerInfo::new("t", "u", "title", chrono::Utc::now(), ""), + ), + output_dir: ".".into(), + suffix: "flv".to_string(), + bytes_written: ByteCounter::new(), + preview: Default::default(), + } + } + + #[test] + fn stream_url_runtime_follows_each_attempts_url() { + let first = "https://ws1a.douyucdn.cn/live/a.flv?token=first&expire=300&fcdn=ws&expire=0"; + let fresh = "https://ws1a.douyucdn.cn/live/a.flv?token=fresh&expire=300&fcdn=ws"; + let args = |downloader: StreamlinkDownloader| { + downloader + .build_file_args(&download_config(fresh), "out.flv.part") + .unwrap() + }; + + let following = args( + StreamlinkDownloader::new(first.to_string(), Platform::Generic).following_stream_url(), + ); + assert!(following.contains(&streamlink_cli_url(fresh))); + assert!(!following.iter().any(|arg| arg.contains("token=first"))); + + let pinned = args(StreamlinkDownloader::new( + first.to_string(), + Platform::Generic, + )); + assert!(pinned.contains(&streamlink_cli_url(first))); + } #[test] fn progressive_flv_gets_httpstream_prefix() { diff --git a/crates/biliup-cli/src/server/core/downloader/ws_expire.rs b/crates/biliup-cli/src/server/core/downloader/ws_expire.rs new file mode 100644 index 0000000..60012f6 --- /dev/null +++ b/crates/biliup-cli/src/server/core/downloader/ws_expire.rs @@ -0,0 +1,178 @@ +//! 斗鱼网宿直链追加 `expire=0` 的拉流前探测。 +//! +//! stream-gears 在自己的首连里处理 403（见 `stream_gears::connect`）；ffmpeg / mesio / +//! streamlink / 边录边传把直链交给外部进程或库，拿不到首连的状态码，所以在交出去之前 +//! 先用同一组请求头探一次，确认网宿接受追加后的直链才用它，否则退回原直链。 + +use crate::server::common::construct_headers; +use biliup::downloader::live::strip_ws_expire_override; +use std::collections::HashMap; +use std::time::Duration; +use tracing::{info, warn}; + +const PROBE_TIMEOUT: Duration = Duration::from_secs(10); + +/// 返回交给下载器的直链：不是追加过 `expire=0` 的网宿直链时原样返回、不发请求。 +/// +/// 探测必须用 HEAD：网宿的 token 一经 GET 成功就作废，同一直链再 GET 只回一个 GOP +/// 就断开；HEAD 照样校验签名（不通过即 403），却不消耗 token。只有 HEAD 返回 2xx +/// 才保留追加后的直链，403、其它状态码与网络错误一律退回原直链。 +pub(crate) async fn resolve_ws_expire_override( + url: String, + headers: &HashMap, +) -> String { + let Some(original) = strip_ws_expire_override(&url) else { + return url; + }; + let original = original.to_string(); + let client = match construct_headers(headers).map(|headers| { + reqwest::Client::builder() + .default_headers(headers) + .connect_timeout(PROBE_TIMEOUT) + .timeout(PROBE_TIMEOUT) + .build() + }) { + Ok(Ok(client)) => client, + _ => return original, + }; + match client.head(&url).send().await { + Ok(response) if response.status().is_success() => { + info!(status = %response.status(), "网宿接受追加 expire=0 的直链"); + url + } + Ok(response) => { + warn!( + status = %response.status(), + "网宿拒绝了追加 expire=0 的直链，本次改用原直链，连接仍会按 expire 定时断开" + ); + original + } + Err(e) => { + warn!(error = %e, "探测追加 expire=0 的直链失败，本次改用原直链"); + original + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::Router; + use axum::extract::RawQuery; + use axum::http::{HeaderMap, Method, StatusCode}; + use axum::routing::any; + use std::sync::{Arc, Mutex}; + + #[derive(Debug, PartialEq)] + struct Hit { + method: Method, + user_agent: Option, + } + + struct Cdn { + base: String, + hits: Arc>>, + } + + /// `reject_duplicate`：模拟网宿修掉重复参数的解析差异，带重复 expire 的一律 403 + async fn wangsu(reject_duplicate: bool, status_otherwise: StatusCode) -> Cdn { + let hits = Arc::new(Mutex::new(Vec::new())); + let seen = hits.clone(); + let app = Router::new().route( + "/live/a.flv", + any( + move |method: Method, RawQuery(query): RawQuery, headers: HeaderMap| { + seen.lock().unwrap().push(Hit { + method, + user_agent: headers + .get("user-agent") + .map(|ua| ua.to_str().unwrap().to_string()), + }); + async move { + let expires = query + .unwrap_or_default() + .split('&') + .filter(|pair| pair.starts_with("expire=")) + .count(); + if reject_duplicate && expires > 1 { + StatusCode::FORBIDDEN + } else { + status_otherwise + } + } + }, + ), + ); + 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() }); + Cdn { + base: format!("http://{addr}/live/a.flv"), + hits, + } + } + + fn headers() -> HashMap { + HashMap::from([("User-Agent".to_string(), "biliup-test".to_string())]) + } + + #[tokio::test] + async fn keeps_override_when_wangsu_accepts_it() { + let cdn = wangsu(false, StatusCode::OK).await; + let overridden = format!("{}?wsAuth=a&token=t&expire=300&fcdn=ws&expire=0", cdn.base); + + let url = resolve_ws_expire_override(overridden.clone(), &headers()).await; + + assert_eq!(url, overridden); + assert_eq!( + *cdn.hits.lock().unwrap(), + [Hit { + method: Method::HEAD, + user_agent: Some("biliup-test".to_string()), + }] + ); + } + + #[tokio::test] + async fn falls_back_to_original_url_on_403() { + let cdn = wangsu(true, StatusCode::OK).await; + let original = format!("{}?wsAuth=a&token=t&expire=300&fcdn=ws", cdn.base); + + let url = resolve_ws_expire_override(format!("{original}&expire=0"), &headers()).await; + + assert_eq!(url, original); + assert_eq!(cdn.hits.lock().unwrap().len(), 1); + } + + #[tokio::test] + async fn falls_back_to_original_url_unless_probe_succeeds() { + let cdn = wangsu(false, StatusCode::METHOD_NOT_ALLOWED).await; + let original = format!("{}?wsAuth=a&expire=300&fcdn=ws", cdn.base); + assert_eq!( + resolve_ws_expire_override(format!("{original}&expire=0"), &headers()).await, + original + ); + + let unreachable = "http://127.0.0.1:1/live/a.flv?wsAuth=a&expire=300&fcdn=ws"; + assert_eq!( + resolve_ws_expire_override(format!("{unreachable}&expire=0"), &headers()).await, + unreachable + ); + } + + #[tokio::test] + async fn sends_nothing_for_urls_without_override() { + let cdn = wangsu(true, StatusCode::OK).await; + for url in [ + format!("{}?wsAuth=a&expire=300&fcdn=ws", cdn.base), + format!("{}?wsAuth=a&expire=300&fcdn=hw&expire=0", cdn.base), + "https://example.com/live.m3u8".to_string(), + ] { + assert_eq!( + resolve_ws_expire_override(url.clone(), &headers()).await, + url + ); + } + assert!(cdn.hits.lock().unwrap().is_empty()); + } +} diff --git a/crates/biliup-cli/src/server/core/live.rs b/crates/biliup-cli/src/server/core/live.rs index b3b3bb0..1ef3dd6 100644 --- a/crates/biliup-cli/src/server/core/live.rs +++ b/crates/biliup-cli/src/server/core/live.rs @@ -201,20 +201,24 @@ pub fn downloader_runtime( } fn streamlink_runtime(stream: &LiveStream) -> DownloaderRuntime { - let (url, platform) = match stream.runtime_options.as_ref() { + let (url, platform, from_stream_url) = match stream.runtime_options.as_ref() { Some(RuntimeOptions::Streamlink(StreamlinkOptions { url, platform })) => ( url.clone().unwrap_or_else(|| stream.raw_stream_url.clone()), streamlink_platform(platform), + false, ), // yt-dlp 型来源（如 YouTube）交给 streamlink 时，传入网页地址让其自行提取， // 而非已解析的 manifest 直链（对齐 youtube.py:96-101） Some(RuntimeOptions::YtDlp(options)) if !options.webpage_url.is_empty() => { - (options.webpage_url.clone(), Platform::Generic) + (options.webpage_url.clone(), Platform::Generic, false) } - _ => (stream.raw_stream_url.clone(), Platform::Generic), + _ => (stream.raw_stream_url.clone(), Platform::Generic, true), }; - let downloader = + let mut downloader = StreamlinkDownloader::new(url, platform).with_headers(stream.stream_headers.clone()); + if from_stream_url { + downloader = downloader.following_stream_url(); + } DownloaderRuntime::StreamLink(Streamlink::new(downloader)) }