fix(douyu): ffmpeg / mesio / streamlink / 边录边传也用上网宿 expire=0，HEAD 探测 + 403 退回原直链 (#1719)

* fix(douyu): ffmpeg / mesio / streamlink / 边录边传也用追加 expire=0 的网宿直链，拉流前探测 403 退回原直链

#1713 只让 stream-gears 用上追加 expire=0 的直链（它在首连里自带 403 兜底），
其它下载器仍拿原直链，录斗鱼网宿房间每 300 s 断一次。

改为在下载公共层（DownloadTask::download）交出直链前，用同一组请求头对追加过
expire=0 的网宿直链探一次：403 改用原直链，其余结果保留追加后的直链。只等响应头、
不读 body，不是追加直链时不发请求。stream-gears 仍走自己的 connect() 兜底。

* fix(douyu): expire=0 探测改用 HEAD，只有 2xx 才保留追加后的直链

实测网宿 token 一经 GET 成功即作废：GET 探测拿到 200 后，下载器用同一直链再连
8 次里 6 次只收到一个 GOP（0.2–2.3 MB）就被断开；HEAD 同样校验签名（改写 expire
即 403），之后 GET 4/4 正常持续。探测失败或返回非 2xx 时保守退回原直链。

* fix(streamlink): 直链型 runtime 每次拉流改用本次解析出的直链

streamlink runtime 在录制开始时固定了第一次解析的直链，重试时 check_stream 刷新的
DownloadConfig::url 用不上：斗鱼网宿 token 连过一次就作废，断开后每次重连只拿到一个
GOP，1 秒一个文件；也绕过了 expire=0 探测。runtime 的 url 就是插件解析出的流直链时
（无 runtime_options），改用 DownloadConfig::url。

* fix(douyu): 追加 expire=0 的直链探测通过却零字节失败时，本次录制余下的拉流改用原直链

HEAD 探测与下载器实际的 GET 若被网宿判得不一样（HEAD 200、GET 403），下载器会
反复失败直到重试用尽。DownloadTask 记一个粘性标记：非 stream-gears 下载器用追加
直链拉流、一个字节没写出且没有正常结束，之后本任务不再追加、也不再探测。

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<broadcast::Sender<DanmakuEvent>>,
+    /// 追加 `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<DownloadStatus>) -> bool {
     )
 }
 
+/// 用追加 `expire=0` 的网宿直链拉流（探测已通过）却没写出任何字节就失败：
+/// 说明网宿对下载器的请求与对 HEAD 探测的判断不一致，不能再信探测结果。
+fn ws_expire_override_failed(
+    stream_url: &str,
+    result: &AppResult<DownloadStatus>,
+    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<DownloadStatus> = 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(

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;

crates/biliup-cli/src/server/core/downloader/streamlink.rs

@@ -116,6 +116,9 @@ pub struct StreamlinkDownloader {
     url: String,
     headers: HashMap<String, String>,
     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<String, String>) -> 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() {

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, String>,
+) -> 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<String>,
+    }
+
+    struct Cdn {
+        base: String,
+        hits: Arc<Mutex<Vec<Hit>>>,
+    }
+
+    /// `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<String, String> {
+        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());
+    }
+}

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))
 }