From 083313257fbff8639a3f09c9bd67ef6659678746 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Tue, 14 Jul 2026 00:32:29 +0800 Subject: [PATCH] Improve downloader reliability and add caching MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add process-level caching for expensive API calls: login tokens (Afreecatv, Douyin), encryption keys (Douyu), room IDs (Douyu, Huya) - Implement Huya mini-app API option (huya_mobile_api) with WUP protocol support for getCdnTokenInfoEx - Add cdn_fallback for Huya with health checks and protocol negotiation - Improve Bilibili stream selection with protocol fallback (hls_fmp4→stream) and qn-based grouping - Fix Twitch auth token invalidation handling and improve cookie file management - Refactor YouTube entry selection with proper playlist recursion and date filtering - Add retry logic for network failures (Douyu betard) and improve error recovery - Share WBI signer state across Bilibili check_stream calls to respect 2-hour key cache interval - Always pass --download-archive for YouTube (including live) to prevent post-stream duplicates - Add field aliases (inke: live_name, kilakila: backPic) and fix Niconico suffix (flv→ts) --- crates/biliup-cli/src/server/config.rs | 3 + .../src/server/core/downloader/ytdlp.rs | 6 +- crates/biliup-cli/src/server/core/live.rs | 2 + .../biliup/src/downloader/live/afreecatv.rs | 54 ++- crates/biliup/src/downloader/live/bigo.rs | 2 + crates/biliup/src/downloader/live/bilibili.rs | 283 ++++++++--- crates/biliup/src/downloader/live/douyin.rs | 82 +++- crates/biliup/src/downloader/live/douyu.rs | 315 ++++++++++-- crates/biliup/src/downloader/live/huya.rs | 420 ++++++++++++++-- crates/biliup/src/downloader/live/huya_wup.rs | 453 ++++++++++++++++++ crates/biliup/src/downloader/live/inke.rs | 1 + crates/biliup/src/downloader/live/kilakila.rs | 2 +- crates/biliup/src/downloader/live/mod.rs | 5 + crates/biliup/src/downloader/live/niconico.rs | 2 +- crates/biliup/src/downloader/live/twitch.rs | 296 ++++++++---- crates/biliup/src/downloader/live/wbi.rs | 60 ++- crates/biliup/src/downloader/live/youtube.rs | 297 ++++++++++-- 17 files changed, 1946 insertions(+), 337 deletions(-) create mode 100644 crates/biliup/src/downloader/live/huya_wup.rs diff --git a/crates/biliup-cli/src/server/config.rs b/crates/biliup-cli/src/server/config.rs index d22af99..8bce724 100644 --- a/crates/biliup-cli/src/server/config.rs +++ b/crates/biliup-cli/src/server/config.rs @@ -122,6 +122,9 @@ pub struct Config { /// 虎牙是否保留 imgplus 流名 #[serde(default)] pub huya_imgplus: Option, + /// 虎牙走小程序 API 获取房间信息 + #[serde(default)] + pub huya_mobile_api: Option, /// 虎牙编码参数 #[serde(default)] pub huya_codec: Option, diff --git a/crates/biliup-cli/src/server/core/downloader/ytdlp.rs b/crates/biliup-cli/src/server/core/downloader/ytdlp.rs index 71aa219..3017998 100644 --- a/crates/biliup-cli/src/server/core/downloader/ytdlp.rs +++ b/crates/biliup-cli/src/server/core/downloader/ytdlp.rs @@ -230,9 +230,9 @@ impl YouTubeDownloader { if let Some(proxy) = &self.cfg.proxy { cmd.arg("--proxy").arg(proxy); } - if !self.cfg.is_live - && let Some(archive) = &self.cfg.download_archive - { + if let Some(archive) = &self.cfg.download_archive { + // 直播也传 --download-archive：yt-dlp 在直播下载完成后同样会写入 archive， + // 避免直播结束后同一视频被当作回放重复下载（对齐 youtube.py:312, 331-334） cmd.arg("--download-archive").arg(archive); } diff --git a/crates/biliup-cli/src/server/core/live.rs b/crates/biliup-cli/src/server/core/live.rs index fbacb29..009ac3a 100644 --- a/crates/biliup-cli/src/server/core/live.rs +++ b/crates/biliup-cli/src/server/core/live.rs @@ -78,12 +78,14 @@ fn live_options(config: &Config) -> LiveOptions { }, huya: HuyaOptions { cdn: config.huya_cdn.clone().unwrap_or_default(), + cdn_fallback: config.huya_cdn_fallback.unwrap_or(false), max_ratio: config.huya_max_ratio.unwrap_or(0), protocol: config .huya_protocol .clone() .unwrap_or_else(|| "Flv".to_string()), imgplus: config.huya_imgplus.unwrap_or(true), + mobile_api: config.huya_mobile_api.unwrap_or(false), codec: config .huya_codec .clone() diff --git a/crates/biliup/src/downloader/live/afreecatv.rs b/crates/biliup/src/downloader/live/afreecatv.rs index c2552f4..6f8897d 100644 --- a/crates/biliup/src/downloader/live/afreecatv.rs +++ b/crates/biliup/src/downloader/live/afreecatv.rs @@ -8,15 +8,26 @@ use regex::Regex; use reqwest::header::{COOKIE, HeaderMap, HeaderValue, REFERER, USER_AGENT}; use serde::Deserialize; use std::collections::HashMap; +use std::time::{Duration, Instant}; +use tokio::sync::Mutex; const AFREECATV_CHANNEL_API_URL: &str = "https://live.afreecatv.com/afreeca/player_live_api.php"; const AFREECATV_LOGIN_URL: &str = "https://login.afreecatv.com/app/LoginAction.php"; const AFREECATV_REFERER: &str = "https://play.afreecatv.com/"; const AFREECATV_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"; const AFREECATV_QUALITY: &str = "original"; +const AFREECATV_LOGIN_TTL: Duration = Duration::from_secs(7 * 24 * 60 * 60); + +struct CachedLogin { + cookie: String, + fetched_at: Instant, + username: String, + password: String, +} pub struct AfreecaTV { re: Regex, + login_cache: Mutex>, } impl Default for AfreecaTV { @@ -29,8 +40,38 @@ impl AfreecaTV { pub fn new() -> Self { Self { re: Regex::new(r"https?://(.*?)\.afreecatv\.com/(\w+)(?:/\d+)?").unwrap(), + login_cache: Mutex::new(None), } } + + /// 复用 7 天内的登录 cookie，避免每次 check_stream 都请求 LoginAction.php。 + /// 凭据与缓存不一致时视为缓存失效；登录失败不缓存。 + async fn login_cookie(&self, live: &AfreecaTVLive) -> LiveResult> { + let (Some(username), Some(password)) = (&live.username, &live.password) else { + return Ok(None); + }; + if username.is_empty() || password.is_empty() { + return Ok(None); + } + + let mut cache = self.login_cache.lock().await; + if let Some(cached) = cache.as_ref() + && cached.fetched_at.elapsed() < AFREECATV_LOGIN_TTL + && cached.username == *username + && cached.password == *password + { + return Ok(Some(cached.cookie.clone())); + } + + let cookie = live.login().await?; + *cache = cookie.as_ref().map(|cookie| CachedLogin { + cookie: cookie.clone(), + fetched_at: Instant::now(), + username: username.clone(), + password: password.clone(), + }); + Ok(cookie) + } } #[async_trait] @@ -44,7 +85,9 @@ impl LivePlugin for AfreecaTV { } async fn check_stream(&self, request: LiveRequest) -> LiveResult { - AfreecaTVLive::new(request).check_stream().await + let live = AfreecaTVLive::new(request); + let cookie = self.login_cookie(&live).await?; + live.check_stream(cookie.as_deref()).await } } @@ -67,14 +110,13 @@ impl AfreecaTVLive { } } - async fn check_stream(&self) -> LiveResult { - let cookie = self.login().await?; + async fn check_stream(&self, cookie: Option<&str>) -> LiveResult { let username = self.room_username()?; - let Some(channel) = self.channel_info(&username, cookie.as_deref()).await? else { + let Some(channel) = self.channel_info(&username, cookie).await? else { return Ok(LiveStatus::Offline); }; - let aid = self.aid(&username, &channel.bno, cookie.as_deref()).await?; - let raw_stream_url = self.stream_url(&channel, &aid, cookie.as_deref()).await?; + let aid = self.aid(&username, &channel.bno, cookie).await?; + let raw_stream_url = self.stream_url(&channel, &aid, cookie).await?; Ok(LiveStatus::Live { stream: Box::new(LiveStream { diff --git a/crates/biliup/src/downloader/live/bigo.rs b/crates/biliup/src/downloader/live/bigo.rs index f11ee27..d46b223 100644 --- a/crates/biliup/src/downloader/live/bigo.rs +++ b/crates/biliup/src/downloader/live/bigo.rs @@ -107,6 +107,8 @@ impl BigoLive { .split('/') .filter(|part| !part.is_empty()) .next_back() + .and_then(|part| part.split(['?', '#']).next()) + .filter(|part| !part.is_empty()) .map(str::to_string) .ok_or_else(|| LiveError::custom("Bigo 直播间地址错误")) } diff --git a/crates/biliup/src/downloader/live/bilibili.rs b/crates/biliup/src/downloader/live/bilibili.rs index 55eeb9a..abd13e0 100644 --- a/crates/biliup/src/downloader/live/bilibili.rs +++ b/crates/biliup/src/downloader/live/bilibili.rs @@ -12,6 +12,7 @@ use serde_json::Value; use std::collections::{BTreeMap, HashMap}; use std::path::PathBuf; use std::time::{SystemTime, UNIX_EPOCH}; +use tracing::{error, warn}; use url::Url; const BILIBILI_API_BASE: &str = "https://api.live.bilibili.com"; @@ -21,6 +22,8 @@ const WBI_WEB_LOCATION: &str = "444.8"; pub struct Bilibili { re: Regex, + /// WBI key 状态跨 check_stream 共享，避免每轮重建导致 2 小时缓存失效 + wbi_signer: WbiSigner, } impl Default for Bilibili { @@ -32,8 +35,8 @@ impl Default for Bilibili { impl Bilibili { pub fn new() -> Self { Self { - re: Regex::new(r"(?:https?://)?(?:b23\.tv|(?:(?:www|m|live)\.)?bilibili\.com)") - .unwrap(), + re: Regex::new(r"(?:https?://)?(?:b23\.tv|live\.bilibili\.com)").unwrap(), + wbi_signer: WbiSigner::new(), } } } @@ -49,7 +52,9 @@ impl LivePlugin for Bilibili { } async fn check_stream(&self, request: LiveRequest) -> LiveResult { - BilibiliLive::new(request).check_stream().await + BilibiliLive::new(request, self.wbi_signer.clone()) + .check_stream() + .await } } @@ -74,11 +79,11 @@ struct BilibiliLive { } impl BilibiliLive { - fn new(request: LiveRequest) -> Self { + fn new(request: LiveRequest, wbi_signer: WbiSigner) -> Self { let options = request.options.bilibili; let credentials = request.credentials; Self { - wbi_signer: WbiSigner::new(request.client.clone()), + wbi_signer, client: request.client, url: request.url, name: request.name, @@ -111,7 +116,22 @@ impl BilibiliLive { return Ok(LiveStatus::Offline); }; self.room_id = Some(profile.room_id); - let candidates = self.get_stream_candidates(&profile, headers).await?; + let mut candidates = self + .get_stream_candidates(&profile, headers.clone(), &self.protocol) + .await?; + if candidates.is_empty() && self.protocol == "hls_fmp4" { + // fmp4 流可能尚未转码完成：等待窗口内按未开播处理，超时后回退 stream(flv) 协议 + if now_unix().saturating_sub(profile.live_start_time) <= self.hls_transcode_timeout { + warn!("{}: 暂未提供 hls_fmp4 流，等待下一次检测", self.name); + return Ok(LiveStatus::Offline); + } + candidates = self + .get_stream_candidates(&profile, headers, "stream") + .await?; + } + if candidates.is_empty() { + return Err(LiveError::custom(format!("获取 {} 流失败", self.protocol))); + } let raw_stream_url = self.select_stream_url(&candidates).await?; let danmaku = self.danmaku_source(); @@ -214,7 +234,9 @@ impl BilibiliLive { let mut params = BTreeMap::new(); params.insert("room_id".to_string(), room_id.to_string()); params.insert("web_location".to_string(), WBI_WEB_LOCATION.to_string()); - self.wbi_signer.sign(&mut params, &headers).await?; + self.wbi_signer + .sign(&self.client, &mut params, &headers) + .await?; let room_info: Value = self .client @@ -283,8 +305,8 @@ impl BilibiliLive { &self, profile: &BiliRoomProfile, headers: HeaderMap, + protocol: &str, ) -> LiveResult> { - let mut last_error = None; for api in &self.api_list { match self .request_play_info(api, profile.room_id, headers.clone()) @@ -292,17 +314,19 @@ impl BilibiliLive { { Ok(play_info) => { if let Some(candidates) = self - .parse_play_info(api, profile, &play_info, &headers) - .await? + .parse_play_info(api, profile, &play_info, &headers, protocol) + .await + && !candidates.is_empty() { return Ok(candidates); } } - Err(err) => last_error = Some(err), + Err(err) => error!("{}: {api} 获取 play_info 失败: {err}", self.name), } } - Err(last_error.unwrap_or_else(|| LiveError::custom("获取 B 站直播流失败"))) + // 与 Python 一致：空结果照常返回，等待/协议回退交给上层处理 + Ok(Vec::new()) } async fn request_play_info( @@ -320,7 +344,9 @@ impl BilibiliLive { params.insert("codec".to_string(), "0".to_string()); params.insert("dolby".to_string(), "5".to_string()); params.insert("web_location".to_string(), WBI_WEB_LOCATION.to_string()); - self.wbi_signer.sign(&mut params, &headers).await?; + self.wbi_signer + .sign(&self.client, &mut params, &headers) + .await?; let play_info: Value = self .client @@ -353,24 +379,23 @@ impl BilibiliLive { profile: &BiliRoomProfile, play_info: &Value, headers: &HeaderMap, - ) -> LiveResult>> { - let streams = match play_info + protocol: &str, + ) -> Option> { + let streams = play_info .get("data") .and_then(|data| data.get("playurl_info")) .and_then(|info| info.get("playurl")) .and_then(|playurl| playurl.get("stream")) - .and_then(Value::as_array) - { - Some(streams) => streams, - None => return Ok(None), - }; + .and_then(Value::as_array)?; - if self.protocol == "hls_fmp4" { - if let Some(candidates) = self + if protocol == "hls_fmp4" { + match self .master_m3u8_candidates(api, profile, play_info, headers) - .await? + .await { - return Ok(Some(candidates)); + Ok(Some(candidates)) => return Some(candidates), + Ok(None) => {} + Err(err) => error!("{}: {api} 获取 m3u8 失败: {err}", self.name), } let candidates = streams @@ -383,22 +408,17 @@ impl BilibiliLive { .and_then(|format| format.get("codec")) .and_then(Value::as_array) .and_then(|codecs| codecs.first()) - .and_then(parse_codec_urls); - if let Some(candidates) = &candidates - && self.qn >= 10000 + .and_then(parse_codec_urls)?; + // fmp4 可能没有原画 + if matches!(self.qn, 10000 | 25000) && !candidates.iter().any(|candidate| candidate.qn == self.qn) { - return Ok(Some(Vec::new())); - } - if candidates.is_some() { - return Ok(candidates); - } - if now_unix().saturating_sub(profile.live_start_time) <= self.hls_transcode_timeout { - return Ok(Some(Vec::new())); + return None; } + return Some(candidates); } - Ok(streams + streams .first() .and_then(|stream| stream.get("format")) .and_then(Value::as_array) @@ -406,7 +426,7 @@ impl BilibiliLive { .and_then(|format| format.get("codec")) .and_then(Value::as_array) .and_then(|codecs| codecs.first()) - .and_then(parse_codec_urls)) + .and_then(parse_codec_urls) } async fn master_m3u8_candidates( @@ -435,7 +455,7 @@ impl BilibiliLive { return Ok(None); } - let mid = self.login_mid(headers).await?.unwrap_or(profile.uid); + let mid = self.login_mid(headers).await.unwrap_or(profile.uid); let text = self .client .get(format!("{api}/xlive/play-gateway/master/url")) @@ -466,25 +486,41 @@ impl BilibiliLive { Ok(parse_master_m3u8(&text)) } - async fn login_mid(&self, headers: &HeaderMap) -> LiveResult> { - let value: Value = self + /// 获取登录用户 mid。与 Python 一致：nav 请求失败或未登录时返回 None， + /// 由调用方以主播 uid 兜底，不中断整次取流。 + async fn login_mid(&self, headers: &HeaderMap) -> Option { + let response = match self .client .get("https://api.bilibili.com/x/web-interface/nav") .headers(headers.clone()) .send() .await - .map_err(|err| LiveError::custom(format!("获取 B 站登录状态失败: {err}")))? - .json() - .await - .map_err(|err| LiveError::custom(format!("解析 B 站登录状态失败: {err}")))?; - Ok(value + { + Ok(response) => response, + Err(err) => { + error!("{}: 获取 B 站登录状态失败: {err}", self.name); + return None; + } + }; + let value: Value = match response.json().await { + Ok(value) => value, + Err(err) => { + error!("{}: 解析 B 站登录状态失败: {err}", self.name); + return None; + } + }; + let mid = value .get("data") .and_then(|data| data.get("isLogin")) .and_then(Value::as_bool) .filter(|is_login| *is_login) .and_then(|_| value.get("data")) .and_then(|data| data.get("mid")) - .and_then(Value::as_u64)) + .and_then(Value::as_u64); + if mid.is_none() { + warn!("{}: 未登录，或将只能录制到最低画质。", self.name); + } + mid } async fn check_url_healthy(&self, url: &str) -> Option { @@ -544,18 +580,24 @@ impl BilibiliLive { } async fn select_stream_url(&self, candidates: &[BiliStreamCandidate]) -> LiveResult { - let selected = if !self.cdn.is_empty() - && let Some(candidate) = self - .cdn - .iter() - .find_map(|cdn| candidates.iter().find(|candidate| &candidate.cdn == cdn)) - { - candidate + let first = candidates + .first() + .ok_or_else(|| LiveError::custom("B 站可用直播流为空"))?; + // 按 bili_qn 精确匹配画质，选不中取第一组（master m3u8 已按 qn 降序，即最高画质） + let target_qn = if candidates.iter().any(|candidate| candidate.qn == self.qn) { + self.qn } else { - candidates - .first() - .ok_or_else(|| LiveError::custom("B 站可用直播流为空"))? + first.qn }; + let group: Vec<&BiliStreamCandidate> = candidates + .iter() + .filter(|candidate| candidate.qn == target_qn) + .collect(); + let selected = self + .cdn + .iter() + .find_map(|cdn| group.iter().copied().find(|candidate| &candidate.cdn == cdn)) + .unwrap_or(group[0]); if !self.cdn_fallback { return Ok(selected.url.clone()); @@ -563,7 +605,7 @@ impl BilibiliLive { if let Some(url) = self.check_url_healthy(&selected.url).await { return Ok(url); } - for candidate in candidates { + for candidate in &group { if let Some(url) = self.check_url_healthy(&candidate.url).await { return Ok(url); } @@ -636,28 +678,52 @@ fn parse_codec_urls(codec: &Value) -> Option> { fn parse_master_m3u8(text: &str) -> Option> { let mut current_qn = None; - let mut candidates = Vec::new(); + let mut groups: BTreeMap> = BTreeMap::new(); for line in text.lines().map(str::trim) { if line.starts_with("#EXT-X-STREAM-INF:") { - current_qn = parse_stream_inf_qn(line); - } else if !line.is_empty() && !line.starts_with('#') { - candidates.push(BiliStreamCandidate { - qn: current_qn.unwrap_or_default(), - cdn: String::new(), - url: line.to_string(), + // 仅保留 CODECS 含 avc 的变体（跳过 hevc/av1） + current_qn = parse_stream_inf_qn(line).filter(|_| { + parse_stream_inf_codecs(line) + .is_some_and(|codecs| codecs.to_ascii_lowercase().contains("avc")) }); - current_qn = None; + } else if line.starts_with("http") + && let Some(qn) = current_qn + { + let cdn = extract_cdn(line); + if cdn.is_empty() { + continue; + } + let group = groups.entry(qn).or_default(); + if !group.iter().any(|candidate| candidate.cdn == cdn) { + group.push(BiliStreamCandidate { + qn, + cdn, + url: line.to_string(), + }); + } } } + // 按 qn 降序展开，供选流时缺省取最高画质 + let candidates: Vec<_> = groups + .into_iter() + .rev() + .flat_map(|(_, group)| group) + .collect(); (!candidates.is_empty()).then_some(candidates) } fn parse_stream_inf_qn(line: &str) -> Option { - line.split(',').find_map(|part| { - let (key, value) = part.split_once('=')?; - matches!(key.trim(), "QN" | "QUALITY" | "BILI-QN") - .then(|| value.trim().trim_matches('"').parse().ok())? - }) + Regex::new(r"BILI-QN=(\d+)") + .unwrap() + .captures(line) + .and_then(|captures| captures[1].parse().ok()) +} + +fn parse_stream_inf_codecs(line: &str) -> Option { + Regex::new(r#"CODECS="([^"]+)""#) + .unwrap() + .captures(line) + .map(|captures| captures[1].to_string()) } fn first_variant_uri(text: &str) -> Option { @@ -704,3 +770,82 @@ fn normalize_api(value: Option<&str>) -> String { }; value.trim_end_matches('/').to_string() } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn matches_only_live_and_short_urls() { + let plugin = Bilibili::new(); + assert!(plugin.matches("https://live.bilibili.com/12345")); + assert!(plugin.matches("http://live.bilibili.com/12345")); + assert!(plugin.matches("live.bilibili.com/12345")); + assert!(plugin.matches("https://b23.tv/abc123")); + assert!(plugin.matches("b23.tv/abc123")); + // 普通视频页/主站不应被认领 + assert!(!plugin.matches("https://www.bilibili.com/video/BV1xx411c7mD")); + assert!(!plugin.matches("https://m.bilibili.com/video/BV1xx411c7mD")); + assert!(!plugin.matches("https://bilibili.com/12345")); + assert!(!plugin.matches("https://space.bilibili.com/1")); + } + + const MASTER_M3U8: &str = concat!( + "#EXTM3U\n", + "#EXT-X-STREAM-INF:BANDWIDTH=5000000,CODECS=\"hev1.1.6.L120.90,mp4a.40.2\",RESOLUTION=1920x1080,BILI-QN=20000\n", + "https://cn-hevc.example.com/live-bvc/0/live_1_2/index.m3u8?cdn=cn-gotcha09&expires=1\n", + "#EXT-X-STREAM-INF:BANDWIDTH=2000000,CODECS=\"avc1.640028,mp4a.40.2\",RESOLUTION=1280x720,BILI-QN=400\n", + "https://cn-sd.example.com/live-bvc/0/live_1_2/index.m3u8?cdn=cn-gotcha02&expires=1\n", + "#EXT-X-STREAM-INF:BANDWIDTH=4000000,CODECS=\"avc1.640032,mp4a.40.2\",RESOLUTION=1920x1080,BILI-QN=10000\n", + "https://cn-hd.example.com/live-bvc/0/live_1_2/index.m3u8?cdn=cn-gotcha01&expires=1\n", + "#EXT-X-STREAM-INF:BANDWIDTH=4000000,CODECS=\"avc1.640032,mp4a.40.2\",RESOLUTION=1920x1080,BILI-QN=10000\n", + "https://ov-hd.example.com/live-bvc/0/live_1_2/index.m3u8?cdn=ov-gotcha05&expires=1\n", + ); + + #[test] + fn parse_master_m3u8_filters_non_avc_and_sorts_by_qn_desc() { + let candidates = parse_master_m3u8(MASTER_M3U8).unwrap(); + + // hevc 变体被过滤 + assert!(candidates.iter().all(|candidate| candidate.qn != 20000)); + // 按 qn 降序：两个 10000 在前，400 在最后 + assert_eq!( + candidates + .iter() + .map(|candidate| candidate.qn) + .collect::>(), + vec![10000, 10000, 400] + ); + // cdn 从 URL 的 cdn= 参数提取 + assert_eq!(candidates[0].cdn, "cn-gotcha01"); + assert_eq!(candidates[1].cdn, "ov-gotcha05"); + assert_eq!(candidates[2].cdn, "cn-gotcha02"); + assert!(candidates[0].url.starts_with("https://cn-hd.example.com/")); + } + + #[test] + fn parse_master_m3u8_skips_variants_without_qn_or_codecs() { + let text = concat!( + "#EXTM3U\n", + "#EXT-X-STREAM-INF:BANDWIDTH=4000000,CODECS=\"avc1.640032,mp4a.40.2\"\n", + "https://a.example.com/index.m3u8?cdn=cn-gotcha01\n", + "#EXT-X-STREAM-INF:BANDWIDTH=4000000,BILI-QN=10000\n", + "https://b.example.com/index.m3u8?cdn=cn-gotcha02\n", + ); + assert!(parse_master_m3u8(text).is_none()); + } + + #[test] + fn parse_master_m3u8_dedups_same_qn_and_cdn() { + let text = concat!( + "#EXTM3U\n", + "#EXT-X-STREAM-INF:CODECS=\"avc1.640032\",BILI-QN=10000\n", + "https://a.example.com/index.m3u8?cdn=cn-gotcha01\n", + "#EXT-X-STREAM-INF:CODECS=\"avc1.640032\",BILI-QN=10000\n", + "https://b.example.com/index.m3u8?cdn=cn-gotcha01\n", + ); + let candidates = parse_master_m3u8(text).unwrap(); + assert_eq!(candidates.len(), 1); + assert!(candidates[0].url.starts_with("https://a.example.com/")); + } +} diff --git a/crates/biliup/src/downloader/live/douyin.rs b/crates/biliup/src/downloader/live/douyin.rs index 39b9d81..530a1c9 100644 --- a/crates/biliup/src/downloader/live/douyin.rs +++ b/crates/biliup/src/downloader/live/douyin.rs @@ -13,6 +13,8 @@ use reqwest::header::{COOKIE, HeaderMap, HeaderValue, REFERER, USER_AGENT}; use serde_json::{Value, json}; use std::collections::HashMap; use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::sync::Mutex; +use tracing::{debug, warn}; const DOUYIN_LIVE_URL: &str = "https://live.douyin.com/"; const DOUYIN_WEBCAST_ENTER_URL: &str = "https://live.douyin.com/webcast/room/web/enter/"; @@ -24,6 +26,8 @@ const DEFAULT_TTWID: &str = "1%7Cu7ogdHsSmHtxbt4hjDCNvcLfVJz78CTM0TTWU8Hio8w%7C1 pub struct Douyin { re: Regex, + /// 进程级 ttwid 缓存，首次成功获取后复用（对应 Python DouyinUtils._douyin_ttwid 类级缓存） + ttwid: Mutex>, } impl Default for Douyin { @@ -36,6 +40,7 @@ impl Douyin { pub fn new() -> Self { Self { re: Regex::new(r"https?://(?:(?:www|m|live|v)\.)?douyin\.com").unwrap(), + ttwid: Mutex::new(None), } } } @@ -51,11 +56,11 @@ impl LivePlugin for Douyin { } async fn check_stream(&self, request: LiveRequest) -> LiveResult { - DouyinLive::new(request).check_stream().await + DouyinLive::new(request, &self.ttwid).check_stream().await } } -struct DouyinLive { +struct DouyinLive<'a> { client: Client, url: String, name: String, @@ -68,10 +73,11 @@ struct DouyinLive { web_rid: Option, room_id: Option, sec_uid: Option, + ttwid_cache: &'a Mutex>, } -impl DouyinLive { - fn new(request: LiveRequest) -> Self { +impl<'a> DouyinLive<'a> { + fn new(request: LiveRequest, ttwid_cache: &'a Mutex>) -> Self { let options = request.options.douyin; Self { client: request.client, @@ -86,12 +92,15 @@ impl DouyinLive { web_rid: None, room_id: None, sec_uid: None, + ttwid_cache, } } async fn check_stream(&mut self) -> LiveResult { self.ensure_cookie().await; - self.resolve_room_keys().await?; + if !self.resolve_room_keys().await? { + return Ok(LiveStatus::Offline); + } let Some(room_info) = self.get_room_info().await? else { return Ok(LiveStatus::Offline); }; @@ -132,7 +141,7 @@ impl DouyinLive { self.cookie.push(';'); } if !self.cookie.contains("ttwid") { - let ttwid = fetch_ttwid(&self.client).await; + let ttwid = self.cached_ttwid().await; self.cookie.push_str(&format!("ttwid={ttwid};")); } if !self.cookie.contains("odin_ttid=") { @@ -145,6 +154,21 @@ impl DouyinLive { } } + /// 进程内首次成功获取 ttwid 后复用；获取失败回退 DEFAULT_TTWID 但不缓存失败结果 + async fn cached_ttwid(&self) -> String { + let mut cached = self.ttwid_cache.lock().await; + if let Some(ttwid) = cached.as_deref() { + return ttwid.to_string(); + } + match fetch_ttwid(&self.client).await { + Some(ttwid) => { + *cached = Some(ttwid.clone()); + ttwid + } + None => DEFAULT_TTWID.to_string(), + } + } + fn headers(&self) -> HeaderMap { let mut headers = HeaderMap::new(); headers.insert(USER_AGENT, HeaderValue::from_static(DOUYIN_USER_AGENT)); @@ -157,7 +181,8 @@ impl DouyinLive { headers } - async fn resolve_room_keys(&mut self) -> LiveResult<()> { + /// 解析直播间标识；返回 false 表示已确定未开播（如用户主页无 web_rid） + async fn resolve_room_keys(&mut self) -> LiveResult { if self.url.contains("v.douyin") { let resp = self .client @@ -171,11 +196,11 @@ impl DouyinLive { self.sec_uid = capture(&url, r"[?&]sec_user_id=([^&]+)"); self.room_id = capture(&url, r"/reflow/(\d+)").or_else(|| capture(&url, r"room_id=(\d+)")); - return Ok(()); + return Ok(true); } if url.contains("isedouyin.com/share/user") { self.sec_uid = capture(&url, r"[?&]sec_uid=([^&]+)"); - return Ok(()); + return Ok(true); } self.url = url; } @@ -189,7 +214,7 @@ impl DouyinLive { .unwrap_or_default(); if matches!(sec_uid.len(), 55 | 76) { self.sec_uid = Some(sec_uid.to_string()); - return Ok(()); + return Ok(true); } let text = self @@ -210,7 +235,11 @@ impl DouyinLive { .map(|cow| cow.to_string()) .ok_or_else(|| LiveError::custom("抖音房间号获取失败"))?; self.web_rid = capture(&render_data, r#""web_rid":"([^"]+)""#); - return Ok(()); + if self.web_rid.is_none() { + debug!(name = %self.name, url = %self.url, "未开播"); + return Ok(false); + } + return Ok(true); } let web_rid = self @@ -223,7 +252,7 @@ impl DouyinLive { .filter(|part| !part.is_empty()) .ok_or_else(|| LiveError::custom("抖音直播间地址错误"))?; self.web_rid = Some(web_rid); - Ok(()) + Ok(true) } async fn get_web_room_info(&self, web_rid: &str) -> LiveResult { @@ -286,10 +315,22 @@ impl DouyinLive { { self.sec_uid = Some(sec_uid.to_string()); } - if response.pointer("/data/user").is_none() - && response.pointer("/data/prompts").and_then(Value::as_str) == Some("直播已结束") - { - return Ok(None); + if response.pointer("/data/user").is_none() { + let prompts = response + .pointer("/data/prompts") + .and_then(Value::as_str) + .unwrap_or_default(); + if prompts == "直播已结束" { + return Ok(None); + } + // 对应 Python 在此直接抛异常（可能是用户被封禁）；Rust 保留 H5 回退，仅提示 + warn!( + name = %self.name, + url = %self.url, + prompts, + response = %response, + "抖音 web 端接口未返回用户信息，可能被风控或封禁，尝试 H5 接口" + ); } } @@ -467,7 +508,7 @@ fn capture(input: &str, pattern: &str) -> Option { .map(|captures| captures[1].to_string()) } -async fn fetch_ttwid(client: &Client) -> String { +async fn fetch_ttwid(client: &Client) -> Option { let resp = client .post(DOUYIN_UNION_REGISTER_URL) .header(USER_AGENT, DOUYIN_USER_AGENT) @@ -480,10 +521,8 @@ async fn fetch_ttwid(client: &Client) -> String { "fid": "" })) .send() - .await; - let Ok(resp) = resp else { - return DEFAULT_TTWID.to_string(); - }; + .await + .ok()?; resp.headers() .get_all("set-cookie") .iter() @@ -495,7 +534,6 @@ async fn fetch_ttwid(client: &Client) -> String { .strip_prefix("ttwid=") .map(ToString::to_string) }) - .unwrap_or_else(|| DEFAULT_TTWID.to_string()) } fn generate_ms_token() -> String { diff --git a/crates/biliup/src/downloader/live/douyu.rs b/crates/biliup/src/downloader/live/douyu.rs index ebc60ae..39cc4d4 100644 --- a/crates/biliup/src/downloader/live/douyu.rs +++ b/crates/biliup/src/downloader/live/douyu.rs @@ -5,6 +5,7 @@ use super::{ use async_trait::async_trait; use chrono::Utc; use md5::{Digest, Md5}; +use rand::Rng; use rand::seq::SliceRandom; use regex::Regex; use reqwest::Client; @@ -12,7 +13,8 @@ use serde::Deserialize; use serde_json::Value; use std::collections::HashMap; use std::time::{SystemTime, UNIX_EPOCH}; -use tracing::warn; +use tokio::sync::{Mutex, RwLock}; +use tracing::{debug, warn}; use url::{Url, form_urlencoded}; const DOUYU_DEFAULT_DID: &str = "10000000000000000000000000001501"; @@ -26,6 +28,12 @@ const DOUYU_USER_AGENT: &str = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleW pub struct Douyu { re: Regex, + /// 进程级加密密钥 + 配套 UA 缓存（对应 Python DouyuUtils.WhiteEncryptKey / UserAgent 类级缓存）。 + /// Mutex 同时充当 single-flight 锁（对应 Python DouyuUtils._lock / _update_key_event）： + /// 并发刷新在此串行，后到者直接命中前者刷新的缓存。 + encrypt_key: Mutex>, + /// 进程级 url -> 真实房间号缓存（对应 Python get_real_rid 的 alru_cache） + real_room_id: RwLock>, } impl Default for Douyu { @@ -38,6 +46,8 @@ impl Douyu { pub fn new() -> Self { Self { re: Regex::new(r"https?://(?:(?:www|m)\.)?douyu\.com").unwrap(), + encrypt_key: Mutex::new(None), + real_room_id: RwLock::new(HashMap::new()), } } } @@ -53,11 +63,29 @@ impl LivePlugin for Douyu { } async fn check_stream(&self, request: LiveRequest) -> LiveResult { - DouyuLive::new(request).check_stream().await + DouyuLive::new(request, &self.encrypt_key, &self.real_room_id) + .check_stream() + .await } } -struct DouyuLive { +/// 缓存的加密密钥与刷新时使用的随机 UA。enc_data 会校验 UA， +/// 因此 getEncryption 与 getH5PlayV1 必须使用同一 UA，密钥与 UA 需成对缓存。 +#[derive(Clone)] +struct CachedEncryptKey { + key: WhiteEncryptKey, + user_agent: String, +} + +impl CachedEncryptKey { + /// 对应 Python DouyuUtils.is_key_valid：expire_at 取自 getEncryption 响应， + /// 缺失时默认 0（永远过期，每次都会重新请求）。 + fn is_valid(&self, now: u64) -> bool { + self.key.expire_at > now + } +} + +struct DouyuLive<'a> { client: Client, url: String, name: String, @@ -67,10 +95,16 @@ struct DouyuLive { douyu_disable_interactive_game: bool, douyu_danmaku: bool, room_id: Option, + encrypt_key_cache: &'a Mutex>, + real_room_id_cache: &'a RwLock>, } -impl DouyuLive { - fn new(request: LiveRequest) -> Self { +impl<'a> DouyuLive<'a> { + fn new( + request: LiveRequest, + encrypt_key_cache: &'a Mutex>, + real_room_id_cache: &'a RwLock>, + ) -> Self { let options = request.options.douyu; Self { client: request.client, @@ -82,6 +116,8 @@ impl DouyuLive { douyu_disable_interactive_game: options.disable_interactive_game, douyu_danmaku: options.danmaku, room_id: None, + encrypt_key_cache, + real_room_id_cache, } } @@ -153,6 +189,11 @@ impl DouyuLive { return Err(LiveError::custom("直播间地址错误")); }; + // 对应 Python get_real_rid 的 alru_cache：同一 url 只请求一次移动端页面 + if let Some(rid) = self.real_room_id_cache.read().await.get(&self.url) { + return Ok(rid.clone()); + } + let mobile_url = format!("https://{DOUYU_MOBILE_DOMAIN}/{short_id}"); let text = self .client @@ -169,7 +210,12 @@ impl DouyuLive { .unwrap() .captures(&text) { - return Ok(caps[1].to_string()); + let rid = caps[1].to_string(); + self.real_room_id_cache + .write() + .await + .insert(self.url.clone(), rid.clone()); + return Ok(rid); } if short_id.chars().all(|ch| ch.is_ascii_digit()) { @@ -180,20 +226,32 @@ impl DouyuLive { } async fn get_room_info(&self, room_id: &str) -> LiveResult> { - let resp: BetardResponse = self - .client - .get(format!("https://{DOUYU_WEB_DOMAIN}/betard/{room_id}")) - .header("referer", format!("https://{DOUYU_WEB_DOMAIN}")) - .send() - .await - .map_err(|err| { - LiveError::custom(format!("获取斗鱼直播间信息失败 room_id: {room_id}: {err}")) - })? - .json() - .await - .map_err(|err| { - LiveError::custom(format!("解析斗鱼直播间信息失败 room_id: {room_id}: {err}")) - })?; + // 对应 douyu.py：网络层错误重试 3 次，缓解 #1376 海外请求失败问题； + // 非网络错误（如 JSON 解析失败）不重试 + let mut body = None; + let mut last_error = None; + for _ in 0..3 { + match self.fetch_room_info_text(room_id).await { + Ok(text) => { + body = Some(text); + break; + } + Err(err) => { + debug!(error = ?err, room_id, "请求斗鱼直播间信息失败，重试"); + last_error = Some(err); + } + } + } + let Some(body) = body else { + let err = last_error.expect("betard retry loop runs at least once"); + return Err(LiveError::custom(format!( + "获取斗鱼直播间信息失败 room_id: {room_id}: {err}" + ))); + }; + + let resp: BetardResponse = serde_json::from_str(&body).map_err(|err| { + LiveError::custom(format!("解析斗鱼直播间信息失败 room_id: {room_id}: {err}")) + })?; let Some(room) = resp.room else { return Ok(None); @@ -207,6 +265,16 @@ impl DouyuLive { Ok(Some(room)) } + async fn fetch_room_info_text(&self, room_id: &str) -> Result { + self.client + .get(format!("https://{DOUYU_WEB_DOMAIN}/betard/{room_id}")) + .header("referer", format!("https://{DOUYU_WEB_DOMAIN}")) + .send() + .await? + .text() + .await + } + async fn has_interactive_game(&self, room_id: &str) -> LiveResult { let data: Value = self .client @@ -262,11 +330,8 @@ impl DouyuLive { } async fn request_web_play_info(&self, room_id: &str, cdn: &str) -> LiveResult { - let now = SystemTime::now() - .duration_since(UNIX_EPOCH) - .map_err(|err| LiveError::custom(format!("获取系统时间失败: {err}")))? - .as_secs(); - let encrypt_key = self.update_key().await?; + let now = unix_now()?; + let (encrypt_key, user_agent) = self.update_key().await?; let auth = sign_stream(&encrypt_key, room_id, now); let form = vec![ @@ -291,7 +356,7 @@ impl DouyuLive { "https://{DOUYU_WEB_DOMAIN}/lapi/live/getH5PlayV1/{room_id}" )) .header("referer", format!("https://{DOUYU_WEB_DOMAIN}")) - .header("user-agent", DOUYU_USER_AGENT) + .header("user-agent", &user_agent) .form(&form) .send() .await @@ -395,14 +460,28 @@ impl DouyuLive { Ok(tx_secret) } - async fn update_key(&self) -> LiveResult { + /// 获取加密密钥及其配套 UA。对应 Python DouyuUtils.sign 中 + /// `is_key_valid() or update_key()` 的缓存逻辑：密钥未过期直接复用， + /// 过期才重新请求 getEncryption。整个过程持有 Mutex，天然 single-flight。 + async fn update_key(&self) -> LiveResult<(WhiteEncryptKey, String)> { + let mut cache = self.encrypt_key_cache.lock().await; + + let now = unix_now()?; + if let Some(cached) = cache.as_ref() + && cached.is_valid(now) + { + return Ok((cached.key.clone(), cached.user_agent.clone())); + } + + // 防风控，每次刷新密钥随机 UA（对应 douyu.py DouyuUtils.update_key） + let user_agent = random_chrome_user_agent(); let rsp: EncryptionResponse = self .client .get(format!( "https://{DOUYU_WEB_DOMAIN}/wgapi/livenc/liveweb/websec/getEncryption" )) .query(&[("did", DOUYU_DEFAULT_DID)]) - .header("user-agent", DOUYU_USER_AGENT) + .header("user-agent", &user_agent) .send() .await .map_err(|err| LiveError::custom(format!("获取斗鱼加密密钥失败: {err}")))? @@ -418,8 +497,14 @@ impl DouyuLive { ))); } - rsp.data - .ok_or_else(|| LiveError::custom("斗鱼加密密钥为空")) + let key = rsp + .data + .ok_or_else(|| LiveError::custom("斗鱼加密密钥为空"))?; + *cache = Some(CachedEncryptKey { + key: key.clone(), + user_agent: user_agent.clone(), + }); + Ok((key, user_agent)) } fn danmaku_source(&self) -> Option { @@ -440,6 +525,22 @@ impl DouyuLive { } } +fn unix_now() -> LiveResult { + Ok(SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_err(|err| LiveError::custom(format!("获取系统时间失败: {err}")))? + .as_secs()) +} + +/// 生成随机 Chrome UA（对应 biliup/plugins/__init__.py 的 random_user_agent()， +/// 桌面端格式，Chrome 大版本随机 100~120） +fn random_chrome_user_agent() -> String { + let chrome_version: u32 = rand::thread_rng().gen_range(100..=120); + format!( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/{chrome_version}.0.0.0 Safari/537.36" + ) +} + fn sign_stream(encrypt_key: &WhiteEncryptKey, room_id: &str, ts: u64) -> String { let salt = if encrypt_key.is_special == 1 { String::new() @@ -547,6 +648,10 @@ struct WhiteEncryptKey { is_special: i64, key: String, enc_data: String, + /// getEncryption 响应携带的过期时间（10 位 Unix 时间戳）。 + /// 对应 Python `WhiteEncryptKey.get('expire_at', 0)`，缺失时默认 0 + #[serde(default)] + expire_at: u64, } #[derive(Deserialize)] @@ -583,6 +688,37 @@ struct XP2PTxSecret { mod tests { use super::*; + fn make_live<'a>( + url: &str, + encrypt_key_cache: &'a Mutex>, + real_room_id_cache: &'a RwLock>, + ) -> DouyuLive<'a> { + DouyuLive { + client: Client::new(), + url: url.to_string(), + name: "test".to_string(), + douyu_cdn: DOUYU_HS_CDN.to_string(), + douyu_force_hs: true, + douyu_rate: 0, + douyu_disable_interactive_game: false, + douyu_danmaku: false, + room_id: None, + encrypt_key_cache, + real_room_id_cache, + } + } + + fn make_key(expire_at: u64) -> WhiteEncryptKey { + WhiteEncryptKey { + rand_str: "rand".to_string(), + enc_time: 1, + is_special: 0, + key: "key".to_string(), + enc_data: "enc".to_string(), + expire_at, + } + } + #[test] fn build_huos_url_rewrites_fcdn_and_appends_secret() { let (stream_id, params) = parse_stream_url( @@ -640,17 +776,13 @@ mod tests { #[tokio::test] async fn maybe_build_huos_url_falls_back_when_build_fails() { - let live = DouyuLive { - client: Client::new(), - url: "https://www.douyu.com/10568722".to_string(), - name: "test".to_string(), - douyu_cdn: DOUYU_HS_CDN.to_string(), - douyu_force_hs: true, - douyu_rate: 0, - douyu_disable_interactive_game: false, - douyu_danmaku: false, - room_id: None, - }; + let encrypt_key_cache = Mutex::new(None); + let real_room_id_cache = RwLock::new(HashMap::new()); + let live = make_live( + "https://www.douyu.com/10568722", + &encrypt_key_cache, + &real_room_id_cache, + ); let raw_stream_url = "not a url".to_string(); @@ -659,4 +791,105 @@ mod tests { raw_stream_url ); } + + #[test] + fn random_chrome_user_agent_matches_python_format() { + // 对应 Python random_user_agent()：Chrome 大版本 100~120，桌面端固定格式 + let re = Regex::new( + r"^Mozilla/5\.0 \(Windows NT 10\.0; Win64; x64\) AppleWebKit/537\.36 \(KHTML, like Gecko\) Chrome/(\d+)\.0\.0\.0 Safari/537\.36$", + ) + .unwrap(); + + for _ in 0..50 { + let ua = random_chrome_user_agent(); + let caps = re.captures(&ua).unwrap_or_else(|| panic!("UA 格式错误: {ua}")); + let version: u32 = caps[1].parse().unwrap(); + assert!( + (100..=120).contains(&version), + "Chrome 版本超出范围: {version}" + ); + } + } + + #[test] + fn cached_encrypt_key_expiry_follows_expire_at() { + let cached = CachedEncryptKey { + key: make_key(1000), + user_agent: random_chrome_user_agent(), + }; + + assert!(cached.is_valid(999)); + // 对应 Python is_key_valid 的 `expire_at > int(time.time())`：等于当前时间视为过期 + assert!(!cached.is_valid(1000)); + assert!(!cached.is_valid(1001)); + + let no_expiry = CachedEncryptKey { + key: make_key(0), + user_agent: random_chrome_user_agent(), + }; + assert!(!no_expiry.is_valid(0)); + } + + #[test] + fn white_encrypt_key_missing_expire_at_defaults_to_zero() { + let key: WhiteEncryptKey = serde_json::from_str( + r#"{"rand_str":"r","enc_time":2,"is_special":0,"key":"k","enc_data":"e"}"#, + ) + .unwrap(); + assert_eq!(key.expire_at, 0); + + let key: WhiteEncryptKey = serde_json::from_str( + r#"{"rand_str":"r","enc_time":2,"is_special":0,"key":"k","enc_data":"e","expire_at":1234567890}"#, + ) + .unwrap(); + assert_eq!(key.expire_at, 1234567890); + } + + #[tokio::test] + async fn update_key_reuses_cached_key_and_user_agent_pair() { + let user_agent = random_chrome_user_agent(); + let encrypt_key_cache = Mutex::new(Some(CachedEncryptKey { + key: make_key(unix_now().unwrap() + 3600), + user_agent: user_agent.clone(), + })); + let real_room_id_cache = RwLock::new(HashMap::new()); + let live = make_live( + "https://www.douyu.com/10568722", + &encrypt_key_cache, + &real_room_id_cache, + ); + + // 缓存未过期时直接复用，不发起网络请求，且密钥与 UA 成对返回 + let (key, ua) = live.update_key().await.unwrap(); + assert_eq!(key.rand_str, "rand"); + assert_eq!(key.enc_data, "enc"); + assert_eq!(ua, user_agent); + } + + #[tokio::test] + async fn resolve_room_id_uses_cached_real_room_id() { + let url = "https://www.douyu.com/somename"; + let encrypt_key_cache = Mutex::new(None); + let real_room_id_cache = RwLock::new(HashMap::from([( + url.to_string(), + "10568722".to_string(), + )])); + let live = make_live(url, &encrypt_key_cache, &real_room_id_cache); + + // 命中缓存时不请求移动端页面 + assert_eq!(live.resolve_room_id().await.unwrap(), "10568722"); + } + + #[tokio::test] + async fn resolve_room_id_prefers_rid_query_param() { + let encrypt_key_cache = Mutex::new(None); + let real_room_id_cache = RwLock::new(HashMap::new()); + let live = make_live( + "https://www.douyu.com/topic/xyz?rid=123456", + &encrypt_key_cache, + &real_room_id_cache, + ); + + assert_eq!(live.resolve_room_id().await.unwrap(), "123456"); + } } diff --git a/crates/biliup/src/downloader/live/huya.rs b/crates/biliup/src/downloader/live/huya.rs index d046b18..8f28615 100644 --- a/crates/biliup/src/downloader/live/huya.rs +++ b/crates/biliup/src/downloader/live/huya.rs @@ -1,6 +1,6 @@ use super::{ DanmakuSource, DownloaderHint, LiveError, LivePlugin, LiveRequest, LiveResult, LiveStatus, - LiveStream, media_ext_from_url, + LiveStream, huya_wup, media_ext_from_url, }; use async_trait::async_trait; use base64::Engine; @@ -13,12 +13,17 @@ use reqwest::Client; use serde_json::Value; use std::collections::HashMap; use std::time::{SystemTime, UNIX_EPOCH}; +use tokio::sync::RwLock; +use tracing::{error, info, warn}; const HUYA_WEB_BASE_URL: &str = "https://www.huya.com"; +const HUYA_MP_BASE_URL: &str = "https://mp.huya.com"; const HUYA_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"; pub struct Huya { re: Regex, + /// 别名房间号 -> 真实数字房间号，进程内缓存避免每轮轮询重复解析 + real_rid_cache: RwLock>, } impl Default for Huya { @@ -31,6 +36,7 @@ impl Huya { pub fn new() -> Self { Self { re: Regex::new(r"https?://(?:(?:www|m)\.)?huya\.com").unwrap(), + real_rid_cache: RwLock::new(HashMap::new()), } } } @@ -46,11 +52,13 @@ impl LivePlugin for Huya { } async fn check_stream(&self, request: LiveRequest) -> LiveResult { - HuyaLive::new(request).check_stream().await + HuyaLive::new(request, &self.real_rid_cache) + .check_stream() + .await } } -struct HuyaLive { +struct HuyaLive<'a> { client: Client, url: String, name: String, @@ -58,12 +66,15 @@ struct HuyaLive { huya_max_ratio: u32, huya_protocol: HuyaProtocol, huya_imgplus: bool, + huya_cdn_fallback: bool, + huya_mobile_api: bool, huya_codec: String, huya_danmaku: bool, + real_rid_cache: &'a RwLock>, } -impl HuyaLive { - fn new(request: LiveRequest) -> Self { +impl<'a> HuyaLive<'a> { + fn new(request: LiveRequest, real_rid_cache: &'a RwLock>) -> Self { let options = request.options.huya; Self { client: request.client, @@ -73,27 +84,67 @@ impl HuyaLive { huya_max_ratio: options.max_ratio, huya_protocol: HuyaProtocol::from_config(&options.protocol), huya_imgplus: options.imgplus, + huya_cdn_fallback: options.cdn_fallback, + huya_mobile_api: options.mobile_api, huya_codec: options.codec, huya_danmaku: options.danmaku, + real_rid_cache, } } async fn check_stream(&self) -> LiveResult { - let page = self.get_room_page().await?; - let Some(profile) = self.extract_room_profile(&page)? else { + let room_id = self.resolve_room_id().await?; + let Some(mut profile) = self.get_room_profile(&room_id).await? else { return Ok(LiveStatus::Offline); }; - if profile.title.starts_with("回放") - || profile.title.starts_with("重播") - || profile.title.ends_with("回放") - || profile.title.ends_with("重播") - { + if is_replay_title(&profile.title) { return Ok(LiveStatus::Offline); } - let stream_urls = self.build_stream_urls(&profile.stream_info)?; - let raw_stream_url = self.select_stream_url(&stream_urls, &profile)?; + let stream_urls = self.build_stream_urls(&profile.stream_info).await?; + let (mut selected_cdn, mut raw_stream_url) = + self.select_stream_url(&stream_urls, &profile)?; + + // HTTPS 的直播流只允许连接一次：健康检查会消耗掉这次连接， + // 因此 cdn_fallback 无论是否回退，最后都要重新取一次流地址 + if self.huya_cdn_fallback { + if !self.check_url_healthy(&raw_stream_url).await { + let cdn_list: Vec<&str> = + stream_urls.iter().map(|(cdn, _)| cdn.as_str()).collect(); + info!(name = %self.name, "cdn_fallback 顺序尝试 {cdn_list:?}"); + let mut fallback = None; + for (cdn, url) in &stream_urls { + if cdn == &selected_cdn { + continue; + } + info!(name = %self.name, "cdn_fallback-{cdn}"); + if self.check_url_healthy(url).await { + fallback = Some(cdn.clone()); + break; + } + } + let Some(fallback) = fallback else { + error!(name = %self.name, "cdn_fallback 所有链接无法使用"); + return Ok(LiveStatus::Offline); + }; + info!(name = %self.name, "cdn_fallback 回退到 {fallback}"); + selected_cdn = fallback; + } + + let Some(next_profile) = self.get_room_profile(&room_id).await? else { + return Ok(LiveStatus::Offline); + }; + profile = next_profile; + let stream_urls = self.build_stream_urls(&profile.stream_info).await?; + let url = stream_urls + .iter() + .find(|(cdn, _)| cdn == &selected_cdn) + .or_else(|| stream_urls.first()) + .map(|(_, url)| url) + .ok_or_else(|| LiveError::custom("虎牙可用 CDN 为空"))?; + raw_stream_url = self.add_ratio(url, &profile.bitrate_info, profile.max_bitrate); + } Ok(LiveStatus::Live { stream: Box::new(LiveStream { @@ -114,15 +165,132 @@ impl HuyaLive { }) } - async fn get_room_page(&self) -> LiveResult { - let room_id = self + /// 解析房间号：URL 路径本身是数字则直接使用， + /// 否则请求房间页取 TT_ROOM_DATA.profileRoom（结果进程内缓存） + async fn resolve_room_id(&self) -> LiveResult { + let path = self .url .split("huya.com/") .nth(1) - .and_then(|part| part.split('?').next()) + .and_then(|part| part.split(['?', '#']).next()) .filter(|part| !part.is_empty()) .ok_or_else(|| LiveError::custom("虎牙直播间地址错误"))?; + if path.chars().all(|c| c.is_ascii_digit()) { + return Ok(path.to_string()); + } + + if let Some(rid) = self.real_rid_cache.read().await.get(path) { + return Ok(rid.clone()); + } + + let page = self.fetch_room_page(path).await?; + let room_data = extract_json_after(&page, r"var\s+TT_ROOM_DATA\s*=\s*", ';')?; + let rid = room_data + .get("profileRoom") + .and_then(|rid| { + rid.as_u64() + .map(|rid| rid.to_string()) + .or_else(|| rid.as_str().map(str::to_string)) + }) + .filter(|rid| !rid.is_empty() && rid != "0") + .ok_or_else(|| LiveError::custom("找不到这个主播"))?; + self.real_rid_cache + .write() + .await + .insert(path.to_string(), rid.clone()); + Ok(rid) + } + + async fn get_room_profile(&self, room_id: &str) -> LiveResult> { + if self.huya_mobile_api { + self.get_room_profile_mobile(room_id).await + } else { + let page = self.fetch_room_page(room_id).await?; + self.extract_room_profile(&page) + } + } + + async fn get_room_profile_mobile(&self, room_id: &str) -> LiveResult> { + let text = self + .client + .get(format!("{HUYA_MP_BASE_URL}/cache.php")) + .query(&[ + ("m", "Live"), + ("do", "profileRoom"), + ("roomid", room_id), + ("showSecret", "1"), + ]) + .header("referer", &self.url) + .header("user-agent", HUYA_USER_AGENT) + .send() + .await + .map_err(|err| LiveError::custom(format!("请求虎牙小程序接口失败: {err}")))? + .text() + .await + .map_err(|err| LiveError::custom(format!("读取虎牙小程序接口失败: {err}")))?; + + let value: Value = serde_json::from_str(&decode_html_entities(&text)) + .map_err(|err| LiveError::custom(format!("解析虎牙小程序接口失败: {err}")))?; + if value.get("status").and_then(Value::as_i64) != Some(200) { + let message = value + .get("message") + .and_then(Value::as_str) + .unwrap_or("未知错误"); + return Err(LiveError::custom(format!("虎牙小程序接口错误: {message}"))); + } + + let data = value + .get("data") + .ok_or_else(|| LiveError::custom("虎牙小程序接口数据为空"))?; + if data.get("liveStatus").and_then(Value::as_str) != Some("ON") { + return Ok(None); + } + let live_data = data + .get("liveData") + .ok_or_else(|| LiveError::custom("虎牙直播信息为空"))?; + // bitRateInfo 是内嵌 JSON 字符串；缺失视为未推流 + let Some(bitrate_json) = live_data + .get("bitRateInfo") + .and_then(Value::as_str) + .filter(|info| !info.is_empty()) + else { + return Ok(None); + }; + let bitrate_info: Vec = serde_json::from_str(bitrate_json) + .map_err(|err| LiveError::custom(format!("解析虎牙码率信息失败: {err}")))?; + + let stream_info = data + .get("stream") + .and_then(|stream| stream.get("baseSteamInfoList")) + .and_then(Value::as_array) + .cloned() + .unwrap_or_default(); + if stream_info.is_empty() { + return Ok(None); + } + + Ok(Some(HuyaRoomProfile { + title: live_data + .get("introduction") + .and_then(Value::as_str) + .unwrap_or_default() + .to_string(), + cover: live_data + .get("screenshot") + .and_then(Value::as_str) + .unwrap_or_default() + .replace("http://", "https://"), + max_bitrate: live_data + .get("bitRate") + .and_then(Value::as_u64) + .unwrap_or_default() as u32, + bitrate_info, + stream_info, + })) + } + + async fn fetch_room_page(&self, room_id: &str) -> LiveResult { let text = self .client .get(format!("{HUYA_WEB_BASE_URL}/{room_id}")) @@ -197,8 +365,10 @@ impl HuyaLive { })) } - fn build_stream_urls(&self, streams_info: &[Value]) -> LiveResult> { + async fn build_stream_urls(&self, streams_info: &[Value]) -> LiveResult> { let mut streams = Vec::new(); + // 同一房间所有 CDN 的 stream_name 相同，防盗链参数只需计算一次 + let mut cached_anticode: Option = None; for stream in streams_info { let priority = stream @@ -212,23 +382,114 @@ impl HuyaLive { let stream_name = self.get_stream_name(json_str(stream, "sStreamName")?); let cdn = json_str(stream, "sCdnType")?.to_string(); let suffix = json_str(stream, self.huya_protocol.suffix_key())?; - let anti_code = json_str(stream, self.huya_protocol.anticode_key())?; let base_url = json_str(stream, self.huya_protocol.url_key())?.replace("http://", "https://"); - let anti_code = build_anticode(&stream_name, anti_code)?; - let url = format!( - "{base_url}/{stream_name}.{suffix}?{anti_code}&codec={}", - self.huya_codec - ); + let presenter_uid = stream + .get("lPresenterUid") + .and_then(|uid| { + uid.as_u64() + .or_else(|| uid.as_str().and_then(|uid| uid.parse().ok())) + }) + .unwrap_or_default(); + + if cached_anticode.is_none() { + // 小程序 API 且保留 imgplus 时直接使用接口返回的防盗链参数； + // 其余情况用 WUP getCdnTokenInfoEx 取新 token 重建（imgplus=false 改写了 + // stream_name，页面自带的防盗链参数对改写后的流名无效） + let anticode = if self.huya_mobile_api && self.huya_imgplus { + json_str(stream, self.huya_protocol.anticode_key())?.to_string() + } else { + match self.get_cdn_token_info_ex(&stream_name).await { + Ok(token) => build_anticode(&stream_name, &token, presenter_uid)?, + Err(err) => { + warn!( + name = %self.name, + err = %err, + "虎牙 getCdnTokenInfoEx 失败，回退页面防盗链参数" + ); + build_anticode( + &stream_name, + json_str(stream, self.huya_protocol.anticode_key())?, + presenter_uid, + )? + } + } + }; + cached_anticode = Some(format!("{anticode}&codec={}", self.huya_codec)); + } + let anti_code = cached_anticode.as_deref().unwrap_or_default(); + let url = format!("{base_url}/{stream_name}.{suffix}?{anti_code}"); streams.push((cdn, priority, url)); } - streams.sort_by(|a, b| b.1.cmp(&a.1)); + streams.sort_by_key(|(_, priority, _)| std::cmp::Reverse(*priority)); Ok(streams .into_iter() .filter(|(cdn, _, _)| !matches!(cdn.as_str(), "HY" | "HUYA" | "HYZJ")) .map(|(cdn, _, url)| (cdn, url)) - .collect()) + .collect()) } + + async fn get_cdn_token_info_ex(&self, stream_name: &str) -> LiveResult { + let ua = huya_wup::random_hyapp_ua(); + let payload = huya_wup::encode_get_cdn_token_ex(stream_name, &ua); + let url = if rand::thread_rng().gen_bool(0.5) { + huya_wup::WUP_YST_URL + } else { + huya_wup::WUP_MAIN_URL + }; + let body = self + .client + .post(url) + .body(payload) + .send() + .await + .map_err(|err| LiveError::custom(format!("请求虎牙 getCdnTokenInfoEx 失败: {err}")))? + .bytes() + .await + .map_err(|err| LiveError::custom(format!("读取虎牙 getCdnTokenInfoEx 失败: {err}")))?; + huya_wup::decode_get_cdn_token_ex(&body) + .filter(|token| !token.is_empty()) + .ok_or_else(|| LiveError::custom("解析虎牙 getCdnTokenInfoEx 响应失败")) + } + + /// 流地址健康检查：能成功建立连接（m3u8 还要求首个变体列表可取）即视为可用。 + /// 注意这次连接会被服务端计数，调用方需要在检查后重新取流地址。 + async fn check_url_healthy(&self, url: &str) -> bool { + let response = match self + .client + .get(url) + .header("referer", HUYA_WEB_BASE_URL) + .header("user-agent", HUYA_USER_AGENT) + .send() + .await + { + Ok(response) if response.status().is_success() => response, + _ => return false, + }; + + if !url.contains(".m3u8") { + return true; + } + let Ok(playlist) = response.text().await else { + return false; + }; + let Some(variant) = first_m3u8_variant(&playlist) else { + // 没有变体说明本身就是媒体列表，能取到即健康 + return true; + }; + let variant_url = match resolve_relative_url(url, &variant) { + Some(resolved) => resolved, + None => return false, + }; + matches!( + self.client + .get(variant_url) + .header("referer", HUYA_WEB_BASE_URL) + .header("user-agent", HUYA_USER_AGENT) + .send() + .await, + Ok(response) if response.status().is_success() + ) } fn get_stream_name(&self, stream_name: &str) -> String { @@ -243,15 +504,17 @@ impl HuyaLive { &self, stream_urls: &[(String, String)], profile: &HuyaRoomProfile, - ) -> LiveResult { - let selected_url = stream_urls + ) -> LiveResult<(String, String)> { + let (cdn, url) = stream_urls .iter() .find(|(cdn, _)| !self.huya_cdn.is_empty() && cdn == &self.huya_cdn) .or_else(|| stream_urls.first()) - .map(|(_, url)| url) .ok_or_else(|| LiveError::custom("虎牙可用 CDN 为空"))?; - Ok(self.add_ratio(selected_url, &profile.bitrate_info, profile.max_bitrate)) + Ok(( + cdn.clone(), + self.add_ratio(url, &profile.bitrate_info, profile.max_bitrate), + )) } fn add_ratio(&self, url: &str, bitrate_info: &[Value], max_bitrate: u32) -> String { @@ -262,10 +525,11 @@ impl HuyaLive { let selected_ratio = bitrate_info .iter() .filter_map(|info| { - let bitrate = info - .get("iBitRate") - .and_then(|bitrate| bitrate.as_u64()) - .unwrap_or(max_bitrate as u64) as u32; + // 与 Python 语义一致：iBitRate 缺失或为 0 都视为原画码率 max_bitrate + let bitrate = match info.get("iBitRate").and_then(|bitrate| bitrate.as_u64()) { + Some(bitrate) if bitrate > 0 => bitrate as u32, + _ => max_bitrate, + }; (bitrate <= self.huya_max_ratio).then_some(bitrate) }) .max(); @@ -341,6 +605,17 @@ impl HuyaProtocol { } } +fn is_replay_title(title: &str) -> bool { + let head: String = title.chars().take(3).collect(); + let tail: String = title + .chars() + .skip(title.chars().count().saturating_sub(3)) + .collect(); + ["回放", "重播"] + .iter() + .any(|key| head.contains(key) || tail.contains(key)) +} + fn extract_json_after(page: &str, pattern: &str, end: char) -> LiveResult { let re = Regex::new(pattern).unwrap(); let Some(mat) = re.find(page) else { @@ -412,7 +687,11 @@ fn find_json_value_end(input: &str, start: usize) -> Option { None } -fn build_anticode(stream_name: &str, anti_code: &str) -> LiveResult { +fn build_anticode( + stream_name: &str, + anti_code: &str, + presenter_uid: u64, +) -> LiveResult { let query = serde_urlencoded::from_str::>(anti_code) .map_err(|err| LiveError::custom(format!("解析虎牙防盗链参数失败: {err}")))?; let Some(fm) = query.get("fm") else { @@ -424,7 +703,12 @@ fn build_anticode(stream_name: &str, anti_code: &str) -> LiveResult { .cloned() .unwrap_or_else(|| "huya_live".to_string()); let platform_id = query.get("t").cloned().unwrap_or_else(|| "100".to_string()); - let uid = generate_random_uid(); + let is_wap = matches!(platform_id.parse::(), Ok(103)); + let uid = if presenter_uid == 0 { + generate_random_uid() + } else { + presenter_uid + }; let now = SystemTime::now() .duration_since(UNIX_EPOCH) .map_err(|err| LiveError::custom(format!("获取系统时间失败: {err}")))?; @@ -454,7 +738,9 @@ fn build_anticode(stream_name: &str, anti_code: &str) -> LiveResult { ws_time = format!("{:x}", now_secs + 24 * 60 * 60); } - let secret_str = format!("{secret_prefix}_{convert_uid}_{stream_name}_{secret_hash}_{ws_time}"); + // wap 平台(t=103)用原始 uid 参与 wsSecret 计算，其余平台用 convert_uid + let calc_uid = if is_wap { uid } else { convert_uid }; + let secret_str = format!("{secret_prefix}_{calc_uid}_{stream_name}_{secret_hash}_{ws_time}"); let ws_secret = md5_hex(secret_str); let fs = query .get("fs") @@ -462,9 +748,19 @@ fn build_anticode(stream_name: &str, anti_code: &str) -> LiveResult { .unwrap_or_else(|| "bgct".to_string()); let fm = urlencoding::encode(query.get("fm").map(String::as_str).unwrap_or_default()); - Ok(format!( - "wsSecret={ws_secret}&wsTime={ws_time}&seqid={seq_id}&ctype={ctype}&ver=1&fs={fs}&fm={fm}&t={platform_id}&u={convert_uid}" - )) + let base = format!( + "wsSecret={ws_secret}&wsTime={ws_time}&seqid={seq_id}&ctype={ctype}&ver=1&fs={fs}&fm={fm}&t={platform_id}" + ); + Ok(if is_wap { + let mut rng = rand::thread_rng(); + let ws_time_secs = u64::from_str_radix(&ws_time, 16).unwrap_or_default(); + let ct = ((ws_time_secs as f64 + rng.gen_range(0.0..1.0)) * 1000.0) as u64; + let uuid = (((ct % 10_000_000_000) as f64 + rng.gen_range(0.0..1.0)) * 1e3 + % f64::from(u32::MAX)) as u64; + format!("{base}&uid={uid}&uuid={uuid}") + } else { + format!("{base}&u={convert_uid}") + }) } fn json_str<'a>(value: &'a Value, key: &str) -> LiveResult<&'a str> { @@ -508,10 +804,54 @@ fn decode_html_entities(input: &str) -> String { .replace(">", ">") } +/// 返回 master m3u8 中第一个 #EXT-X-STREAM-INF 变体的 URI；媒体列表返回 None +fn first_m3u8_variant(playlist: &str) -> Option { + let mut lines = playlist.lines().map(str::trim); + while let Some(line) = lines.next() { + if line.starts_with("#EXT-X-STREAM-INF") { + for candidate in lines.by_ref() { + if !candidate.is_empty() && !candidate.starts_with('#') { + return Some(candidate.to_string()); + } + } + } + } + None +} + +fn resolve_relative_url(base: &str, target: &str) -> Option { + if target.starts_with("http://") || target.starts_with("https://") { + return Some(target.to_string()); + } + url::Url::parse(base) + .ok()? + .join(target) + .ok() + .map(String::from) +} + #[cfg(test)] mod tests { use super::*; + #[test] + fn replay_title_matches_keyword_within_first_or_last_three_chars() { + // 关键词整体出现在前三个字符内 + assert!(is_replay_title("【回放】昨天的比赛")); + assert!(is_replay_title("回放：昨天的比赛")); + assert!(is_replay_title("重播比赛")); + // 关键词整体出现在后三个字符内 + assert!(is_replay_title("昨天的比赛回放")); + assert!(is_replay_title("昨天的比赛重播】")); + // 标题不足三个字符时取整个标题 + assert!(is_replay_title("回放")); + // 关键词跨越前三/后三边界或出现在中间时不过滤 + assert!(!is_replay_title("精彩回放合集视频节目")); + assert!(!is_replay_title("每日【回放】剪辑合集")); + assert!(!is_replay_title("正在直播")); + assert!(!is_replay_title("")); + } + #[test] fn extract_stream_json_stops_before_player_config_closing_brace() { let page = r#" diff --git a/crates/biliup/src/downloader/live/huya_wup.rs b/crates/biliup/src/downloader/live/huya_wup.rs new file mode 100644 index 0000000..22150f0 --- /dev/null +++ b/crates/biliup/src/downloader/live/huya_wup.rs @@ -0,0 +1,453 @@ +//! 虎牙 WUP(getCdnTokenInfoEx) 请求所需的最小 tars 编解码。 +//! +//! 只实现该接口用到的子集：TarsV3 uni-packet（`map` 载荷）、 +//! `HUYA.GetCdnTokenExReq`/`HUYA.UserId` 结构编码与 `HUYA.GetCdnTokenExRsp` 解码。 + +use rand::Rng; + +pub(crate) const WUP_MAIN_URL: &str = "https://wup.huya.com"; +pub(crate) const WUP_YST_URL: &str = "https://snmhuya.yst.aisee.tv/liveui/getCdnTokenInfoEx"; + +const TYPE_INT8: u8 = 0; +const TYPE_INT16: u8 = 1; +const TYPE_INT32: u8 = 2; +const TYPE_INT64: u8 = 3; +const TYPE_STRING1: u8 = 6; +const TYPE_STRING4: u8 = 7; +const TYPE_MAP: u8 = 8; +const TYPE_LIST: u8 = 9; +const TYPE_STRUCT_BEGIN: u8 = 10; +const TYPE_STRUCT_END: u8 = 11; +const TYPE_ZERO: u8 = 12; +const TYPE_BYTES: u8 = 13; + +struct TarsOutputStream { + buffer: Vec, +} + +impl TarsOutputStream { + fn new() -> Self { + Self { buffer: Vec::new() } + } + + fn write_head(&mut self, tag: u8, ty: u8) { + if tag < 15 { + self.buffer.push((tag << 4) | ty); + } else { + self.buffer.push(0xF0 | ty); + self.buffer.push(tag); + } + } + + fn write_int8(&mut self, tag: u8, value: i8) { + if value == 0 { + self.write_head(tag, TYPE_ZERO); + } else { + self.write_head(tag, TYPE_INT8); + self.buffer.push(value as u8); + } + } + + fn write_int16(&mut self, tag: u8, value: i16) { + if (-128..=127).contains(&value) { + self.write_int8(tag, value as i8); + } else { + self.write_head(tag, TYPE_INT16); + self.buffer.extend_from_slice(&value.to_be_bytes()); + } + } + + fn write_int32(&mut self, tag: u8, value: i32) { + if (-32768..=32767).contains(&value) { + self.write_int16(tag, value as i16); + } else { + self.write_head(tag, TYPE_INT32); + self.buffer.extend_from_slice(&value.to_be_bytes()); + } + } + + fn write_int64(&mut self, tag: u8, value: i64) { + if (i32::MIN as i64..=i32::MAX as i64).contains(&value) { + self.write_int32(tag, value as i32); + } else { + self.write_head(tag, TYPE_INT64); + self.buffer.extend_from_slice(&value.to_be_bytes()); + } + } + + fn write_string(&mut self, tag: u8, value: &str) { + let bytes = value.as_bytes(); + if bytes.len() <= 255 { + self.write_head(tag, TYPE_STRING1); + self.buffer.push(bytes.len() as u8); + } else { + self.write_head(tag, TYPE_STRING4); + self.buffer + .extend_from_slice(&(bytes.len() as u32).to_be_bytes()); + } + self.buffer.extend_from_slice(bytes); + } + + fn write_bytes(&mut self, tag: u8, value: &[u8]) { + self.write_head(tag, TYPE_BYTES); + self.write_head(0, TYPE_INT8); + self.write_int32(0, value.len() as i32); + self.buffer.extend_from_slice(value); + } + + fn write_map_str_bytes(&mut self, tag: u8, entries: &[(&str, &[u8])]) { + self.write_head(tag, TYPE_MAP); + self.write_int32(0, entries.len() as i32); + for (key, value) in entries { + self.write_string(0, key); + self.write_bytes(1, value); + } + } + + fn write_empty_str_map(&mut self, tag: u8) { + self.write_head(tag, TYPE_MAP); + self.write_int32(0, 0); + } + + fn write_struct_begin(&mut self, tag: u8) { + self.write_head(tag, TYPE_STRUCT_BEGIN); + } + + fn write_struct_end(&mut self) { + self.write_head(0, TYPE_STRUCT_END); + } +} + +struct TarsInputStream<'a> { + data: &'a [u8], + pos: usize, +} + +impl<'a> TarsInputStream<'a> { + fn new(data: &'a [u8]) -> Self { + Self { data, pos: 0 } + } + + fn peek_head(&self) -> Option<(u8, u8)> { + let byte = *self.data.get(self.pos)?; + let tag = (byte >> 4) & 0x0F; + let ty = byte & 0x0F; + if tag >= 15 { + Some((*self.data.get(self.pos + 1)?, ty)) + } else { + Some((tag, ty)) + } + } + + fn read_head(&mut self) -> Option<(u8, u8)> { + let byte = *self.data.get(self.pos)?; + let tag = (byte >> 4) & 0x0F; + let ty = byte & 0x0F; + self.pos += 1; + if tag >= 15 { + let tag = *self.data.get(self.pos)?; + self.pos += 1; + Some((tag, ty)) + } else { + Some((tag, ty)) + } + } + + fn skip_to_tag(&mut self, target: u8) -> bool { + while let Some((tag, ty)) = self.peek_head() { + if ty == TYPE_STRUCT_END || tag > target { + return false; + } + if tag == target { + return true; + } + self.read_head(); + self.skip_field(ty); + } + false + } + + fn skip_field(&mut self, ty: u8) { + match ty { + TYPE_INT8 => self.pos += 1, + TYPE_INT16 => self.pos += 2, + TYPE_INT32 => self.pos += 4, + TYPE_INT64 => self.pos += 8, + TYPE_STRING1 => { + if let Some(len) = self.data.get(self.pos).copied() { + self.pos += 1 + len as usize; + } + } + TYPE_STRING4 => { + if let Some(len) = self.read_u32_be() { + self.pos += len as usize; + } + } + TYPE_MAP => { + let size = self.read_int_value().unwrap_or(0); + for _ in 0..size.saturating_mul(2) { + if let Some((_, ty)) = self.read_head() { + self.skip_field(ty); + } + } + } + TYPE_LIST => { + let size = self.read_int_value().unwrap_or(0); + for _ in 0..size { + if let Some((_, ty)) = self.read_head() { + self.skip_field(ty); + } + } + } + TYPE_BYTES => { + self.read_head(); + let size = self.read_int_value().unwrap_or(0); + self.pos += size as usize; + } + TYPE_STRUCT_BEGIN => { + while let Some((_, ty)) = self.read_head() { + if ty == TYPE_STRUCT_END { + break; + } + self.skip_field(ty); + } + } + _ => {} + } + } + + fn read_u32_be(&mut self) -> Option { + let bytes = self.data.get(self.pos..self.pos + 4)?; + self.pos += 4; + Some(u32::from_be_bytes(bytes.try_into().ok()?)) + } + + /// 读取一个已消费掉 head 的整数值（带 head 的话先 read_head）。 + fn read_int_value(&mut self) -> Option { + let (_, ty) = self.read_head()?; + self.read_int_body(ty) + } + + fn read_int_body(&mut self, ty: u8) -> Option { + match ty { + TYPE_ZERO => Some(0), + TYPE_INT8 => { + let v = *self.data.get(self.pos)? as i8; + self.pos += 1; + Some(v as i64) + } + TYPE_INT16 => { + let bytes = self.data.get(self.pos..self.pos + 2)?; + self.pos += 2; + Some(i16::from_be_bytes(bytes.try_into().ok()?) as i64) + } + TYPE_INT32 => { + let bytes = self.data.get(self.pos..self.pos + 4)?; + self.pos += 4; + Some(i32::from_be_bytes(bytes.try_into().ok()?) as i64) + } + TYPE_INT64 => { + let bytes = self.data.get(self.pos..self.pos + 8)?; + self.pos += 8; + Some(i64::from_be_bytes(bytes.try_into().ok()?)) + } + _ => None, + } + } + + fn read_string_body(&mut self, ty: u8) -> Option { + let len = match ty { + TYPE_STRING1 => { + let len = *self.data.get(self.pos)? as usize; + self.pos += 1; + len + } + TYPE_STRING4 => self.read_u32_be()? as usize, + _ => return None, + }; + let bytes = self.data.get(self.pos..self.pos + len)?; + self.pos += len; + Some(String::from_utf8_lossy(bytes).to_string()) + } + + fn read_bytes_body(&mut self) -> Option> { + self.read_head()?; + let size = self.read_int_value()? as usize; + let bytes = self.data.get(self.pos..self.pos + size)?; + self.pos += size; + Some(bytes.to_vec()) + } + + fn read_string(&mut self, tag: u8) -> Option { + if !self.skip_to_tag(tag) { + return None; + } + let (_, ty) = self.read_head()?; + self.read_string_body(ty) + } + + fn read_bytes(&mut self, tag: u8) -> Option> { + if !self.skip_to_tag(tag) { + return None; + } + let (_, ty) = self.read_head()?; + if ty != TYPE_BYTES { + return None; + } + self.read_bytes_body() + } +} + +/// 生成随机的虎牙客户端 sHuYaUA（对应 Python UAGenerator.get_random_hyapp_ua）。 +pub(crate) fn random_hyapp_ua() -> String { + let mut rng = rand::thread_rng(); + // (平台短名, 版本, 是否安卓系——安卓系追加随机 build 号与 api level) + let configs = [ + ("adr", "13.1.0", true), + ("ios", "13.1.0", false), + ("huya_nftv", "2.6.10", true), + ("pc_exe", "7000000", false), + ]; + let (platform, version, android) = configs[rng.gen_range(0..configs.len())]; + if android { + format!( + "{platform}&{version}.{}&official&{}", + rng.gen_range(3000..=5000), + rng.gen_range(28..=36) + ) + } else { + format!("{platform}&{version}&official") + } +} + +/// 编码 getCdnTokenInfoEx 的 TarsV3 uni-packet 请求体。 +pub(crate) fn encode_get_cdn_token_ex(stream_name: &str, huya_ua: &str) -> Vec { + // HUYA.GetCdnTokenExReq，写在 tag 0 + let mut req = TarsOutputStream::new(); + req.write_struct_begin(0); + req.write_string(0, ""); // sFlvUrl + req.write_string(1, stream_name); // sStreamName + req.write_int32(2, 0); // iLoopTime + req.write_struct_begin(3); // tId: HUYA.UserId + req.write_int64(0, 0); // lUid + req.write_string(1, ""); // sGuid + req.write_string(2, ""); // sToken + req.write_string(3, huya_ua); // sHuYaUA + req.write_string(4, ""); // sCookie + req.write_int32(5, 0); // iTokenType + req.write_string(6, ""); // sDeviceId + req.write_string(7, ""); // sQIMEI + req.write_struct_end(); + req.write_int32(4, 66); // iAppId + req.write_struct_end(); + + // TarsV3 载荷：map { "tReq": } + let mut payload = TarsOutputStream::new(); + payload.write_map_str_bytes(0, &[("tReq", &req.buffer)]); + + // RequestPacket + let mut packet = TarsOutputStream::new(); + packet.write_int16(1, 3); // iVersion = 3 + packet.write_int8(2, 0); // cPacketType + packet.write_int32(3, 0); // iMessageType + packet.write_int32(4, 1); // iRequestId + packet.write_string(5, "liveui"); // sServantName + packet.write_string(6, "getCdnTokenInfoEx"); // sFuncName + packet.write_bytes(7, &payload.buffer); // sBuffer + packet.write_int32(8, 0); // iTimeout + packet.write_empty_str_map(9); // context + packet.write_empty_str_map(10); // status + + let mut out = Vec::with_capacity(4 + packet.buffer.len()); + out.extend_from_slice(&((4 + packet.buffer.len()) as u32).to_be_bytes()); + out.extend_from_slice(&packet.buffer); + out +} + +/// 从 getCdnTokenInfoEx 响应中解出 sFlvToken。 +pub(crate) fn decode_get_cdn_token_ex(body: &[u8]) -> Option { + let data = body.get(4..)?; + let mut packet = TarsInputStream::new(data); + let payload = packet.read_bytes(7)?; + + // map，找 "tRsp" + let mut map = TarsInputStream::new(&payload); + let (tag, ty) = map.read_head()?; + if tag != 0 || ty != TYPE_MAP { + return None; + } + let size = map.read_int_value()?; + for _ in 0..size { + let (_, key_ty) = map.read_head()?; + let key = map.read_string_body(key_ty)?; + let (_, value_ty) = map.read_head()?; + if value_ty != TYPE_BYTES { + return None; + } + let value = map.read_bytes_body()?; + if key == "tRsp" { + // HUYA.GetCdnTokenExRsp 结构体在 tag 0，sFlvToken 在结构体内 tag 0 + let mut rsp = TarsInputStream::new(&value); + let (_, ty) = rsp.read_head()?; + if ty != TYPE_STRUCT_BEGIN { + return None; + } + return rsp.read_string(0); + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + // 黄金数据由移植前的 Python tars 实现生成： + // stream_name = "1234567-1234567-5299160-2469038-10057-A-0-1-imgplus" + // sHuYaUA = "adr&13.1.0.4321&official&30" + const GOLDEN_REQ: &str = "000000a210032c3c400156066c6976657569661167657443646e546f6b656e496e666f45787d0000740800010604745265711d0000670a06001633313233343536372d313233343536372d353239393136302d323436393033382d31303035372d412d302d312d696d67706c75732c3a0c16002600361b6164722631332e312e302e34333231266f6666696369616c26333046005c660076000b40420b8c980ca80c"; + // 同一实现编码的响应包，sFlvToken 见下方断言，iExpireTime = 1699999999 + const GOLDEN_RSP: &str = "0000009210032c3c400156066c6976657569661167657443646e546f6b656e496e666f45787d0000640800010604745273701d0000570a064e77735365637265743d61626331323326777354696d653d36366626666d3d6447567a6446397a64484a6c5957302533442663747970653d687579615f6c69766526743d3130302666733d62676374126553f0ff0b8c980ca80c"; + + fn hex_decode(input: &str) -> Vec { + (0..input.len()) + .step_by(2) + .map(|idx| u8::from_str_radix(&input[idx..idx + 2], 16).unwrap()) + .collect() + } + + #[test] + fn encode_matches_python_reference_bytes() { + let encoded = encode_get_cdn_token_ex( + "1234567-1234567-5299160-2469038-10057-A-0-1-imgplus", + "adr&13.1.0.4321&official&30", + ); + assert_eq!(encoded, hex_decode(GOLDEN_REQ)); + } + + #[test] + fn decode_extracts_flv_token_from_python_reference_bytes() { + let token = decode_get_cdn_token_ex(&hex_decode(GOLDEN_RSP)).unwrap(); + assert_eq!( + token, + "wsSecret=abc123&wsTime=66f&fm=dGVzdF9zdHJlYW0%3D&ctype=huya_live&t=100&fs=bgct" + ); + } + + #[test] + fn decode_rejects_truncated_body() { + let mut body = hex_decode(GOLDEN_RSP); + body.truncate(20); + assert!(decode_get_cdn_token_ex(&body).is_none()); + } + + #[test] + fn random_hyapp_ua_has_expected_shape() { + for _ in 0..50 { + let ua = random_hyapp_ua(); + let parts: Vec<&str> = ua.split('&').collect(); + assert!(parts.len() == 3 || parts.len() == 4, "unexpected ua: {ua}"); + assert_eq!(parts[2], "official"); + } + } +} diff --git a/crates/biliup/src/downloader/live/inke.rs b/crates/biliup/src/downloader/live/inke.rs index 62e2948..82eac17 100644 --- a/crates/biliup/src/downloader/live/inke.rs +++ b/crates/biliup/src/downloader/live/inke.rs @@ -119,6 +119,7 @@ struct InkeData { status: bool, #[serde(default)] live_addr: Vec, + #[serde(alias = "live_name")] title: Option, image: Option, } diff --git a/crates/biliup/src/downloader/live/kilakila.rs b/crates/biliup/src/downloader/live/kilakila.rs index eda44be..a729aa1 100644 --- a/crates/biliup/src/downloader/live/kilakila.rs +++ b/crates/biliup/src/downloader/live/kilakila.rs @@ -153,6 +153,6 @@ struct KilakilaBody { #[serde(rename = "hlsPlayUrl")] hls_play_url: Option, title: Option, - #[serde(rename = "coverUrl")] + #[serde(rename = "coverUrl", alias = "backPic")] cover_url: Option, } diff --git a/crates/biliup/src/downloader/live/mod.rs b/crates/biliup/src/downloader/live/mod.rs index 080e9ea..4e854ec 100644 --- a/crates/biliup/src/downloader/live/mod.rs +++ b/crates/biliup/src/downloader/live/mod.rs @@ -15,6 +15,7 @@ mod douyin; mod douyu; mod general; mod huya; +mod huya_wup; mod inke; mod kilakila; mod kuaishou; @@ -190,9 +191,11 @@ impl Default for DouyuOptions { #[derive(Debug, Clone)] pub struct HuyaOptions { pub cdn: String, + pub cdn_fallback: bool, pub max_ratio: u32, pub protocol: String, pub imgplus: bool, + pub mobile_api: bool, pub codec: String, pub danmaku: bool, } @@ -201,9 +204,11 @@ impl Default for HuyaOptions { fn default() -> Self { Self { cdn: String::new(), + cdn_fallback: false, max_ratio: 0, protocol: "Flv".to_string(), imgplus: true, + mobile_api: false, codec: "264".to_string(), danmaku: false, } diff --git a/crates/biliup/src/downloader/live/niconico.rs b/crates/biliup/src/downloader/live/niconico.rs index b71de80..0df7ae0 100644 --- a/crates/biliup/src/downloader/live/niconico.rs +++ b/crates/biliup/src/downloader/live/niconico.rs @@ -82,7 +82,7 @@ impl NiconicoLive { raw_stream_url: self.url.clone(), platform: "niconico".to_string(), stream_headers: HashMap::new(), - suffix: "flv".to_string(), + suffix: "ts".to_string(), danmaku: None, downloader_hint: DownloaderHint::Streamlink, runtime_options: Some(RuntimeOptions::Streamlink(StreamlinkOptions { diff --git a/crates/biliup/src/downloader/live/twitch.rs b/crates/biliup/src/downloader/live/twitch.rs index c2d9531..5bcd7c0 100644 --- a/crates/biliup/src/downloader/live/twitch.rs +++ b/crates/biliup/src/downloader/live/twitch.rs @@ -10,16 +10,44 @@ use reqwest::Client; use reqwest::header::{HeaderMap, HeaderValue}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; +use std::collections::hash_map::DefaultHasher; use std::collections::{HashMap, HashSet}; +use std::hash::{Hash, Hasher}; use std::path::PathBuf; use std::process::Stdio; +use std::sync::RwLock; use tokio::fs; use tokio::process::Command; use tokio::time::Duration; +use tracing::warn; const CLIENT_ID: &str = "kimne78kx3ncx6brgo4mv6wki5h1ko"; const GQL_URL: &str = "https://gql.twitch.tv/gql"; +/// Twitch 已失效的 auth_token（Twitch 与 TwitchVideos 共享，进程内长期记忆）。 +/// 对应 Python 版 TwitchUtils._invalid_auth_token 类级属性。 +static INVALID_AUTH_TOKEN: RwLock> = RwLock::new(None); + +/// 返回可用的 auth_token：未配置、为空或已被记为失效时返回 None。 +fn effective_auth_token(auth_token: Option<&str>) -> Option { + let auth_token = auth_token.filter(|token| !token.is_empty())?; + let invalid = INVALID_AUTH_TOKEN + .read() + .unwrap_or_else(|err| err.into_inner()); + (invalid.as_deref() != Some(auth_token)).then(|| auth_token.to_string()) +} + +/// 将 auth_token 记为失效，后续请求跳过；仅首次记录时告警。 +fn mark_auth_token_invalid(auth_token: &str) { + let mut invalid = INVALID_AUTH_TOKEN + .write() + .unwrap_or_else(|err| err.into_inner()); + if invalid.as_deref() != Some(auth_token) { + *invalid = Some(auth_token.to_string()); + warn!("Twitch Cookie已失效请及时更换，后续操作将忽略Twitch Cookie"); + } +} + pub struct TwitchVideos { re: Regex, } @@ -143,9 +171,7 @@ impl TwitchLive { } "#, "variables": { "channel_name": channel_name } - }), - self.twitch_auth_token.as_deref(), - ) + })) .await?; let user = gql @@ -194,7 +220,7 @@ impl TwitchLive { url: Some(self.url.clone()), platform: StreamlinkPlatform::Twitch { disable_ads: self.twitch_disable_ads, - auth_token: self.twitch_auth_token.clone(), + auth_token: effective_auth_token(self.twitch_auth_token.as_deref()), }, })), }), @@ -209,16 +235,13 @@ impl TwitchLive { .ok_or_else(|| LiveError::custom("Twitch 直播间地址错误")) } - async fn post_gql( - &self, - ops: T, - auth_token: Option<&str>, - ) -> LiveResult { - for retry in 0..2 { + async fn post_gql(&self, ops: T) -> LiveResult { + let mut auth_token = effective_auth_token(self.twitch_auth_token.as_deref()); + loop { let resp = self .client .post(GQL_URL) - .headers(gql_headers((retry == 0).then_some(auth_token).flatten())?) + .headers(gql_headers(auth_token.as_deref())?) .json(&ops) .timeout(Duration::from_secs(15)) .send() @@ -235,11 +258,15 @@ impl TwitchLive { .await .map_err(|err| LiveError::custom(format!("解析 Twitch GQL 失败: {err}")))?; if gql.error.as_deref() == Some("Unauthorized") { - continue; + // 带 token 被拒时记为失效并改用匿名请求重试一次 + if let Some(token) = auth_token.take() { + mark_auth_token_invalid(&token); + continue; + } + return Err(LiveError::custom("请求 Twitch GQL 错误: Unauthorized")); } return Ok(gql); } - Err(LiveError::custom("Twitch Cookie 已失效")) } fn danmaku_source(&self) -> Option { @@ -310,90 +337,69 @@ impl TwitchVideosLive { is_live: false, use_live_cover: false, cover_url: (!selection.thumbnail.is_empty()).then_some(selection.thumbnail), - cookies_file: None, + cookies_file: self.auth_cookie_file().await, prefer_vcodec: None, prefer_acodec: None, max_filesize: None, max_height: None, download_archive: Some(PathBuf::from("archive.txt")), - extra_ytdlp_args: self.auth_cookie_args(), + extra_ytdlp_args: Vec::new(), })), }), }) } - async fn extract_info(&self, url: &str, process: bool) -> LiveResult> { - let mut command = Command::new("yt-dlp"); - command - .stdin(Stdio::null()) - .arg("--dump-single-json") - .arg("--skip-download") - .arg("--ignore-errors") - .arg("--extractor-retries") - .arg("0") - .arg("--no-warnings"); - if !process { - command.arg("--no-playlist"); - } - for arg in self.auth_cookie_args() { - command.arg(arg); - } - command.arg(url).kill_on_drop(true); + /// 运行 yt-dlp 提取信息。`flat` 为 true 时用于 /videos 列表页， + /// 以 `--flat-playlist` 惰性枚举条目（对应 Python 版 process=False）； + /// 为 false 时对单个条目做完整提取。 + async fn extract_info(&self, url: &str, flat: bool) -> LiveResult> { + let mut auth_token = effective_auth_token(self.twitch_auth_token.as_deref()); + loop { + let cookie_file = match auth_token.as_deref() { + Some(token) => write_cookie_file(token).await, + None => None, + }; + let mut command = Command::new("yt-dlp"); + command + .stdin(Stdio::null()) + .arg("--dump-single-json") + .arg("--skip-download") + .arg("--ignore-errors") + .arg("--extractor-retries") + .arg("0") + .arg("--no-warnings"); + if flat { + command.arg("--flat-playlist"); + } else { + command.arg("--no-playlist"); + } + if let Some(cookie_file) = &cookie_file { + command.arg("--cookies").arg(cookie_file); + } + command.arg(url).kill_on_drop(true); - let output = command.output().await.map_err(|err| { - LiveError::custom(format!("运行 yt-dlp 获取 Twitch 视频信息失败: {err}")) - })?; - let stderr = String::from_utf8_lossy(&output.stderr); - if !output.status.success() { - if stderr.contains("Unauthorized") && self.twitch_auth_token.is_some() { - return self.extract_info_without_auth(url, process).await; + let output = command.output().await.map_err(|err| { + LiveError::custom(format!("运行 yt-dlp 获取 Twitch 视频信息失败: {err}")) + })?; + if output.status.success() { + return parse_ytdlp_stdout(&output.stdout); + } + let stderr = String::from_utf8_lossy(&output.stderr); + if stderr.contains("Unauthorized") + && let Some(token) = auth_token.take() + { + // 带 token 被拒时记为失效并改用匿名请求重试一次 + mark_auth_token_invalid(&token); + continue; } return Ok(None); } - - parse_ytdlp_stdout(&output.stdout) } - async fn extract_info_without_auth( - &self, - url: &str, - process: bool, - ) -> LiveResult> { - let mut command = Command::new("yt-dlp"); - command - .stdin(Stdio::null()) - .arg("--dump-single-json") - .arg("--skip-download") - .arg("--ignore-errors") - .arg("--extractor-retries") - .arg("0") - .arg("--no-warnings"); - if !process { - command.arg("--no-playlist"); - } - command.arg(url).kill_on_drop(true); - - let output = command.output().await.map_err(|err| { - LiveError::custom(format!("运行 yt-dlp 获取 Twitch 视频信息失败: {err}")) - })?; - if !output.status.success() { - return Ok(None); - } - parse_ytdlp_stdout(&output.stdout) - } - - fn auth_cookie_args(&self) -> Vec { - let Some(auth_token) = self - .twitch_auth_token - .as_deref() - .filter(|value| !value.is_empty()) - else { - return Vec::new(); - }; - vec![ - "--add-header".to_string(), - format!("Cookie: auth-token={auth_token}"), - ] + /// auth_token 可用时生成 Netscape cookie 文件，返回其路径传给 yt-dlp --cookies。 + async fn auth_cookie_file(&self) -> Option { + let auth_token = effective_auth_token(self.twitch_auth_token.as_deref())?; + write_cookie_file(&auth_token).await } async fn archive_ids(&self) -> HashSet { @@ -416,7 +422,7 @@ impl TwitchVideosLive { entry .get("id") .and_then(Value::as_str) - .is_none_or(|id| !archive_ids.contains(id)) + .is_none_or(|id| !archive_contains(archive_ids, id)) }) } else { Some(value) @@ -535,3 +541,129 @@ fn string_field(entry: &Value, keys: &[&str]) -> Option { .map(str::to_string) }) } + +/// 判断条目 id 是否已归档。yt-dlp 新版对 Twitch VOD 记录的归档 id 为 `v<数字>` +/// （flat 条目 id 同为 `v<数字>`），旧版本归档为纯数字，两种形式都视为已归档。 +fn archive_contains(archive_ids: &HashSet, id: &str) -> bool { + if archive_ids.contains(id) { + return true; + } + if let Some(stripped) = id.strip_prefix('v') { + !stripped.is_empty() + && stripped.bytes().all(|byte| byte.is_ascii_digit()) + && archive_ids.contains(stripped) + } else { + !id.is_empty() + && id.bytes().all(|byte| byte.is_ascii_digit()) + && archive_ids.contains(&format!("v{id}")) + } +} + +/// 生成 Netscape 格式 cookie 文件内容（域 .twitch.tv，键 auth-token）， +/// 与 Python 版传给 yt-dlp cookiefile 的内容一致。 +fn netscape_cookie_file_content(auth_token: &str) -> String { + format!( + "# Netscape HTTP Cookie File\n.twitch.tv\tTRUE\t/\tFALSE\t0\tauth-token\t{auth_token}\n" + ) +} + +/// cookie 文件放系统临时目录，路径由 token 哈希决定： +/// 同一 token 复用同一路径，token 变化自然写入新文件。 +fn cookie_file_path(auth_token: &str) -> PathBuf { + let mut hasher = DefaultHasher::new(); + auth_token.hash(&mut hasher); + std::env::temp_dir().join(format!("biliup-twitch-cookies-{:016x}.txt", hasher.finish())) +} + +/// 将 auth_token 写入 Netscape cookie 文件并返回路径；内容未变化时不重写。 +/// 写入失败仅告警并返回 None（降级为匿名请求）。 +async fn write_cookie_file(auth_token: &str) -> Option { + let path = cookie_file_path(auth_token); + let content = netscape_cookie_file_content(auth_token); + if fs::read_to_string(&path).await.ok().as_deref() != Some(content.as_str()) { + if let Err(err) = fs::write(&path, &content).await { + warn!("写入 Twitch cookie 文件失败: {}: {err}", path.display()); + return None; + } + } + Some(path) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn netscape_cookie_file_content_matches_python_format() { + assert_eq!( + netscape_cookie_file_content("abc123"), + "# Netscape HTTP Cookie File\n.twitch.tv\tTRUE\t/\tFALSE\t0\tauth-token\tabc123\n" + ); + } + + #[test] + fn cookie_file_path_is_stable_per_token() { + assert_eq!(cookie_file_path("token-a"), cookie_file_path("token-a")); + assert_ne!(cookie_file_path("token-a"), cookie_file_path("token-b")); + assert!(cookie_file_path("token-a").starts_with(std::env::temp_dir())); + } + + #[test] + fn archive_contains_matches_both_id_forms() { + let archive_ids: HashSet = ["v111", "222", "SomeClipSlug"] + .into_iter() + .map(str::to_string) + .collect(); + // 完全一致 + assert!(archive_contains(&archive_ids, "v111")); + assert!(archive_contains(&archive_ids, "222")); + assert!(archive_contains(&archive_ids, "SomeClipSlug")); + // v 前缀与纯数字互认 + assert!(archive_contains(&archive_ids, "111")); + assert!(archive_contains(&archive_ids, "v222")); + // 未归档 + assert!(!archive_contains(&archive_ids, "v333")); + assert!(!archive_contains(&archive_ids, "333")); + // 非数字 id 不做前缀换算 + assert!(!archive_contains(&archive_ids, "omeClipSlug")); + assert!(!archive_contains(&archive_ids, "v")); + assert!(!archive_contains(&archive_ids, "")); + } + + #[test] + fn write_cookie_file_writes_and_reuses_path() { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .unwrap(); + runtime.block_on(async { + let token = "unit-test-token"; + let path = write_cookie_file(token).await.expect("写入 cookie 文件失败"); + assert_eq!( + fs::read_to_string(&path).await.unwrap(), + netscape_cookie_file_content(token) + ); + // 复用同一路径 + assert_eq!(write_cookie_file(token).await.as_deref(), Some(&*path)); + let _ = fs::remove_file(&path).await; + }); + } + + #[test] + fn invalid_auth_token_is_remembered_and_skipped() { + // 共用全局状态，串行放在同一个测试内 + assert_eq!(effective_auth_token(None), None); + assert_eq!(effective_auth_token(Some("")), None); + assert_eq!( + effective_auth_token(Some("token-x")), + Some("token-x".to_string()) + ); + mark_auth_token_invalid("token-x"); + assert_eq!(effective_auth_token(Some("token-x")), None); + // 更换 token 后恢复可用 + assert_eq!( + effective_auth_token(Some("token-y")), + Some("token-y".to_string()) + ); + } +} diff --git a/crates/biliup/src/downloader/live/wbi.rs b/crates/biliup/src/downloader/live/wbi.rs index 5ae5810..0b9d35d 100644 --- a/crates/biliup/src/downloader/live/wbi.rs +++ b/crates/biliup/src/downloader/live/wbi.rs @@ -1,3 +1,4 @@ +use reqwest::Client; use reqwest::header::HeaderMap; use serde::Deserialize; use std::collections::BTreeMap; @@ -31,19 +32,22 @@ struct WbiImg { sub_url: String, } +#[derive(Default)] +struct WbiState { + key: Option, + last_update: u64, +} + +/// WBI 签名器。key 状态放在 `Arc` 内共享：插件结构体持有一份， +/// 每次 `check_stream` 克隆句柄即可跨轮询复用，使 2 小时更新间隔真正生效。 +#[derive(Clone, Default)] pub struct WbiSigner { - client: reqwest::Client, - key: Arc>>, - last_update: Arc>, + state: Arc>, } impl WbiSigner { - pub fn new(client: reqwest::Client) -> Self { - Self { - client, - key: Arc::new(RwLock::new(None)), - last_update: Arc::new(RwLock::new(0)), - } + pub fn new() -> Self { + Self::default() } fn extract_key(url: &str) -> Option { @@ -62,23 +66,29 @@ impl WbiSigner { .collect() } - async fn update_key(&self, headers: &HeaderMap) -> LiveResult<()> { + fn is_fresh(state: &WbiState, now: u64) -> bool { + state.key.is_some() && now.saturating_sub(state.last_update) < UPDATE_INTERVAL + } + + async fn update_key(&self, client: &Client, headers: &HeaderMap) -> LiveResult<()> { let now = SystemTime::now() .duration_since(UNIX_EPOCH) .map(|duration| duration.as_secs()) .unwrap_or_default(); - { - let last = *self.last_update.read().await; - if now.saturating_sub(last) < UPDATE_INTERVAL && self.key.read().await.is_some() { - return Ok(()); - } + if Self::is_fresh(&*self.state.read().await, now) { + return Ok(()); + } + + // 写锁内复查，避免并发 check_stream 时重复请求 nav + let mut state = self.state.write().await; + if Self::is_fresh(&state, now) { + return Ok(()); } debug!("Updating WBI key..."); - let resp: NavResponse = self - .client + let resp: NavResponse = client .get("https://api.bilibili.com/x/web-interface/nav") .headers(headers.clone()) .timeout(Duration::from_secs(5)) @@ -92,21 +102,25 @@ impl WbiSigner { let sub = Self::extract_key(&resp.data.wbi_img.sub_url) .ok_or_else(|| LiveError::custom("提取 B 站 WBI sub key 失败"))?; - *self.key.write().await = Some(Self::create_mixin_key(&img, &sub)); - *self.last_update.write().await = now; + state.key = Some(Self::create_mixin_key(&img, &sub)); + state.last_update = now; Ok(()) } pub async fn sign( &self, + client: &Client, params: &mut BTreeMap, headers: &HeaderMap, ) -> LiveResult<()> { - self.update_key(headers).await?; + self.update_key(client, headers).await?; - let key = self.key.read().await; - let key = key - .as_ref() + let key = self + .state + .read() + .await + .key + .clone() .ok_or_else(|| LiveError::custom("B 站 WBI key 为空"))?; let ts = SystemTime::now() diff --git a/crates/biliup/src/downloader/live/youtube.rs b/crates/biliup/src/downloader/live/youtube.rs index b91a282..9535adb 100644 --- a/crates/biliup/src/downloader/live/youtube.rs +++ b/crates/biliup/src/downloader/live/youtube.rs @@ -7,6 +7,8 @@ use chrono::Utc; use regex::Regex; use serde_json::{Map, Value}; use std::collections::{HashMap, HashSet}; +use std::future::Future; +use std::pin::Pin; use std::process::Stdio; use tokio::fs; use tokio::process::Command; @@ -70,6 +72,17 @@ struct YoutubeSelection { is_live: bool, } +/// 条目筛选结果，对齐 Python `loop_entries` 的 dict / None / 'stop' 三态。 +#[derive(Debug, PartialEq)] +enum EntryDecision { + /// 命中待下载条目 + Select(Value), + /// 跳过当前条目，继续扫描 + Skip, + /// 条目早于 after_date，列表按时间倒序，无需继续扫描当前列表 + Stop, +} + impl YoutubeLive { fn new(request: LiveRequest) -> Self { let options = request.options.youtube; @@ -92,12 +105,15 @@ impl YoutubeLive { } async fn check_stream(&self) -> LiveResult { - let Some(info) = self.extract_info(&self.url).await? else { + // 列表型 URL 先用 --flat-playlist 轻量枚举（对齐 Python extract_info(process=False)）， + // 命中的条目再单条完整提取 + let Some(info) = self.extract_info(&self.url, true).await? else { return Ok(LiveStatus::Offline); }; let archive_ids = self.archive_ids().await; - let Some(entry) = self.select_entry(&info, &archive_ids) else { - return Ok(LiveStatus::Offline); + let entry = match self.select_entry(&info, &archive_ids).await? { + EntryDecision::Select(entry) => entry, + EntryDecision::Skip | EntryDecision::Stop => return Ok(LiveStatus::Offline), }; let entry = self.resolve_entry(entry).await?; let selection = self.selection(&entry); @@ -144,15 +160,16 @@ impl YoutubeLive { prefer_acodec: self.prefer_acodec.clone(), max_filesize: self.max_videosize.clone(), max_height: self.max_resolution, - download_archive: (!selection.is_live) - .then(|| std::path::PathBuf::from("archive.txt")), + // 直播完成后 yt-dlp 也会写入 archive（对齐 youtube.py:312）， + // 避免直播结束后同一视频被当作回放重复下载 + download_archive: Some(std::path::PathBuf::from("archive.txt")), extra_ytdlp_args: Vec::new(), })), }), }) } - async fn extract_info(&self, url: &str) -> LiveResult> { + async fn extract_info(&self, url: &str, flat: bool) -> LiveResult> { let mut command = Command::new("yt-dlp"); command .stdin(Stdio::null()) @@ -163,6 +180,11 @@ impl YoutubeLive { .arg("0") .arg("--no-warnings"); + if flat { + // 列表页只做轻量枚举，不逐条完整解析；对单视频 URL 无影响 + command.arg("--flat-playlist"); + } + if let Some(cookie_file) = &self.cookie_file { command.arg("--cookies").arg(cookie_file); } @@ -188,85 +210,130 @@ impl YoutubeLive { .collect() } + /// 递归扫描播放列表，返回首个命中的条目（对齐 Python loop_entries）。 + /// Stop 只终止当前列表的扫描，向上层表现为未命中。 fn select_entry<'a>( - &self, + &'a self, value: &'a Value, + archive_ids: &'a HashSet, + ) -> Pin> + Send + 'a>> { + Box::pin(async move { + let Some(object) = value.as_object() else { + return Ok(EntryDecision::Skip); + }; + + if object.get("_type").and_then(Value::as_str) == Some("playlist") { + let entries = object.get("entries").and_then(Value::as_array); + for entry in entries.into_iter().flatten() { + match self.select_entry(entry, archive_ids).await? { + selected @ EntryDecision::Select(_) => return Ok(selected), + EntryDecision::Stop => return Ok(EntryDecision::Skip), + EntryDecision::Skip => {} + } + } + return Ok(EntryDecision::Skip); + } + + self.evaluate_entry(value, object, archive_ids).await + }) + } + + async fn evaluate_entry( + &self, + value: &Value, + object: &Map, archive_ids: &HashSet, - ) -> Option<&'a Value> { - let object = value.as_object()?; - if self.entry_allowed(object, archive_ids) { - return Some(value); + ) -> LiveResult { + if !self.entry_allowed(object, archive_ids) { + return Ok(EntryDecision::Skip); } - object - .get("entries") - .and_then(Value::as_array) - .and_then(|entries| { - entries - .iter() - .find_map(|entry| self.select_entry(entry, archive_ids)) - }) + // 未配置日期范围时无需 upload_date，直接命中 + if self.after_date.is_none() && self.before_date.is_none() { + return Ok(EntryDecision::Select(value.clone())); + } + + // flat 条目缺少 upload_date 时单条完整提取后再判定（对齐 youtube.py:259-272） + let mut candidate = value.clone(); + if candidate.get("upload_date").and_then(Value::as_str).is_none() + && let Some(url) = string_field(&candidate, &["webpage_url", "url"]) + && let Some(resolved) = self.extract_info(&url, true).await? + { + candidate = resolved; + } + + let Some(upload_date) = candidate + .get("upload_date") + .and_then(Value::as_str) + .map(str::to_string) + else { + // 时间必然存在，补全后仍缺失说明条目异常，暂时跳过（对齐 youtube.py:268-270） + return Ok(EntryDecision::Skip); + }; + + Ok(self.date_decision(candidate, &upload_date)) } fn entry_allowed(&self, object: &Map, archive_ids: &HashSet) -> bool { - match object.get("live_status").and_then(Value::as_str) { - Some("is_upcoming") => return false, - Some("is_live") if !self.enable_download_live => return false, - Some("was_live") if self.live_url || !self.enable_download_playback => return false, - _ if self.live_url => return false, - _ => {} - } - - if object.get("_type").and_then(Value::as_str) == Some("playlist") { + let live_status = object.get("live_status").and_then(Value::as_str); + if !self.live_status_allowed(live_status) { return false; } - let is_live = object.get("live_status").and_then(Value::as_str) == Some("is_live"); - if !is_live + // 已在 archive 中的条目视为已下载；直播中的条目例外（对齐 youtube.py:253-257） + if live_status != Some("is_live") && let Some(id) = object.get("id").and_then(Value::as_str) && archive_ids.contains(id) { return false; } - self.date_allowed(object.get("upload_date")) + true } - fn date_allowed(&self, upload_date: Option<&Value>) -> bool { - if self.after_date.is_none() && self.before_date.is_none() { - return true; + /// live_status 准入判定： + /// - is_upcoming 一律跳过； + /// - is_live 由直播下载开关控制； + /// - /live URL 只处理直播中的条目，其余状态一律跳过； + /// - was_live 由回放下载开关控制； + /// - 其他状态（含普通视频）放行。 + fn live_status_allowed(&self, live_status: Option<&str>) -> bool { + match live_status { + Some("is_upcoming") => false, + Some("is_live") => self.enable_download_live, + _ if self.live_url => false, + Some("was_live") => self.enable_download_playback, + _ => true, } + } - let Some(upload_date) = upload_date.and_then(Value::as_str) else { - return false; - }; - + /// 日期范围判定。早于 after_date 时返回 Stop： + /// 列表按时间倒序，后续条目只会更旧（对齐 youtube.py:274-275）。 + fn date_decision(&self, candidate: Value, upload_date: &str) -> EntryDecision { if let Some(after_date) = &self.after_date && upload_date < after_date.as_str() { - return false; + return EntryDecision::Stop; } if let Some(before_date) = &self.before_date && upload_date > before_date.as_str() { - return false; + return EntryDecision::Skip; } - true + EntryDecision::Select(candidate) } - async fn resolve_entry(&self, entry: &Value) -> LiveResult { + async fn resolve_entry(&self, entry: Value) -> LiveResult { if entry.get("_type").and_then(Value::as_str) != Some("url") { - return Ok(entry.clone()); + return Ok(entry); } - let Some(url) = string_field(entry, &["webpage_url", "url"]) else { - return Ok(entry.clone()); + let Some(url) = string_field(&entry, &["webpage_url", "url"]) else { + return Ok(entry); }; - Ok(self - .extract_info(&url) - .await? - .unwrap_or_else(|| entry.clone())) + // 命中的 flat 条目单条完整提取，获得下载所需字段（对齐 youtube.py:292-293） + Ok(self.extract_info(&url, false).await?.unwrap_or(entry)) } fn selection(&self, entry: &Value) -> YoutubeSelection { @@ -359,6 +426,29 @@ fn is_youtube_live_url(url: &str) -> bool { #[cfg(test)] mod tests { use super::*; + use serde_json::json; + + fn youtube_live( + live_url: bool, + enable_download_live: bool, + enable_download_playback: bool, + ) -> YoutubeLive { + YoutubeLive { + url: "https://www.youtube.com/@test".to_string(), + name: "test".to_string(), + cookie_file: None, + live_url, + enable_download_live, + enable_download_playback, + after_date: None, + before_date: None, + prefer_vcodec: None, + prefer_acodec: None, + max_resolution: None, + max_videosize: None, + youtube_danmaku: false, + } + } #[test] fn matches_youtube_urls() { @@ -369,4 +459,113 @@ mod tests { assert!(plugin.matches("https://youtu.be/test")); assert!(!plugin.matches("https://example.com/watch?v=test")); } + + #[test] + fn live_url_allows_ongoing_live() { + // 回归测试：/live URL 直播中且开启直播下载时应放行， + // 原 match 会落入 `_ if self.live_url` 臂被误拒 + let live = youtube_live(true, true, true); + assert!(live.live_status_allowed(Some("is_live"))); + + let entry = json!({"id": "abc", "live_status": "is_live"}); + assert!(live.entry_allowed(entry.as_object().unwrap(), &HashSet::new())); + } + + #[test] + fn live_status_matrix() { + for live_url in [false, true] { + for enable_live in [false, true] { + for enable_playback in [false, true] { + let live = youtube_live(live_url, enable_live, enable_playback); + + // 预告一律跳过 + assert!(!live.live_status_allowed(Some("is_upcoming"))); + // 直播中仅由直播下载开关控制，与 /live URL 无关 + assert_eq!(live.live_status_allowed(Some("is_live")), enable_live); + // 回放：/live URL 一律跳过，否则由回放开关控制 + assert_eq!( + live.live_status_allowed(Some("was_live")), + !live_url && enable_playback + ); + // 普通视频/未知状态：仅 /live URL 跳过 + assert_eq!(live.live_status_allowed(None), !live_url); + assert_eq!(live.live_status_allowed(Some("post_live")), !live_url); + } + } + } + } + + #[test] + fn archived_entry_skipped_unless_live() { + let live = youtube_live(false, true, true); + let archive: HashSet = ["abc".to_string()].into(); + + // 已归档的回放跳过 + let vod = json!({"id": "abc", "live_status": "was_live"}); + assert!(!live.entry_allowed(vod.as_object().unwrap(), &archive)); + + // 直播中的条目即使已在 archive 中也不算已下载 + let living = json!({"id": "abc", "live_status": "is_live"}); + assert!(live.entry_allowed(living.as_object().unwrap(), &archive)); + + // 未归档的回放放行 + let fresh = json!({"id": "xyz", "live_status": "was_live"}); + assert!(live.entry_allowed(fresh.as_object().unwrap(), &archive)); + } + + #[test] + fn date_decision_respects_range() { + let mut live = youtube_live(false, true, true); + live.after_date = Some("20240101".to_string()); + live.before_date = Some("20241231".to_string()); + + assert_eq!( + live.date_decision(json!({}), "20240601"), + EntryDecision::Select(json!({})) + ); + // 早于 after_date：列表按时间倒序，停止扫描 + assert_eq!(live.date_decision(json!({}), "20231231"), EntryDecision::Stop); + // 晚于 before_date：跳过当前条目继续扫描 + assert_eq!(live.date_decision(json!({}), "20250101"), EntryDecision::Skip); + } + + #[tokio::test] + async fn selects_first_allowed_entry_from_flat_playlist() { + let live = youtube_live(false, true, true); + let archive: HashSet = ["done".to_string()].into(); + let info = json!({ + "_type": "playlist", + "entries": [ + {"_type": "url", "id": "up", "url": "https://www.youtube.com/watch?v=up", "live_status": "is_upcoming"}, + {"_type": "url", "id": "done", "url": "https://www.youtube.com/watch?v=done", "live_status": "was_live"}, + {"_type": "url", "id": "vod", "url": "https://www.youtube.com/watch?v=vod", "live_status": "was_live"}, + ] + }); + + match live.select_entry(&info, &archive).await.unwrap() { + EntryDecision::Select(entry) => { + assert_eq!(entry.get("id").and_then(Value::as_str), Some("vod")); + } + other => panic!("应命中 vod 条目，实际为 {other:?}"), + } + } + + #[tokio::test] + async fn stops_scanning_when_older_than_after_date() { + let mut live = youtube_live(false, true, true); + live.after_date = Some("20240101".to_string()); + let info = json!({ + "_type": "playlist", + "entries": [ + {"_type": "url", "id": "old", "url": "https://www.youtube.com/watch?v=old", "upload_date": "20230101"}, + {"_type": "url", "id": "new", "url": "https://www.youtube.com/watch?v=new", "upload_date": "20240601"}, + ] + }); + + // 首个条目早于 after_date 即停止扫描当前列表，后续条目不再命中 + assert_eq!( + live.select_entry(&info, &HashSet::new()).await.unwrap(), + EntryDecision::Skip + ); + } }