Improve downloader reliability and add caching

- 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)
This commit is contained in:
ForgQi committed 2026-07-14 00:32:29 +08:00
1 parent 18c5bf086e
commit 083313257f
17 files changed
+1946 -337

No files matched your search

+3
View File
@@ -122,6 +122,9 @@ pub struct Config {
/// 虎牙是否保留 imgplus 流名
#[serde(default)]
pub huya_imgplus: Option<bool>,
/// 虎牙走小程序 API 获取房间信息
#[serde(default)]
pub huya_mobile_api: Option<bool>,
/// 虎牙编码参数
#[serde(default)]
pub huya_codec: Option<String>,
@@ -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);
}
@@ -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()
+48 -6
View File
@@ -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<Option<CachedLogin>>,
}
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<Option<String>> {
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<LiveStatus> {
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<LiveStatus> {
let cookie = self.login().await?;
async fn check_stream(&self, cookie: Option<&str>) -> LiveResult<LiveStatus> {
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 {
@@ -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 直播间地址错误"))
}
+214 -69
View File
@@ -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<LiveStatus> {
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<Vec<BiliStreamCandidate>> {
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<Option<Vec<BiliStreamCandidate>>> {
let streams = match play_info
protocol: &str,
) -> Option<Vec<BiliStreamCandidate>> {
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<Option<u64>> {
let value: Value = self
/// 获取登录用户 mid。与 Python 一致:nav 请求失败或未登录时返回 None,
/// 由调用方以主播 uid 兜底,不中断整次取流。
async fn login_mid(&self, headers: &HeaderMap) -> Option<u64> {
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<String> {
@@ -544,18 +580,24 @@ impl BilibiliLive {
}
async fn select_stream_url(&self, candidates: &[BiliStreamCandidate]) -> LiveResult<String> {
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<Vec<BiliStreamCandidate>> {
fn parse_master_m3u8(text: &str) -> Option<Vec<BiliStreamCandidate>> {
let mut current_qn = None;
let mut candidates = Vec::new();
let mut groups: BTreeMap<u32, Vec<BiliStreamCandidate>> = 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<u32> {
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<String> {
Regex::new(r#"CODECS="([^"]+)""#)
.unwrap()
.captures(line)
.map(|captures| captures[1].to_string())
}
fn first_variant_uri(text: &str) -> Option<String> {
@@ -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<_>>(),
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/"));
}
}
+60 -22
View File
@@ -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<Option<String>>,
}
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<LiveStatus> {
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<String>,
room_id: Option<String>,
sec_uid: Option<String>,
ttwid_cache: &'a Mutex<Option<String>>,
}
impl DouyinLive {
fn new(request: LiveRequest) -> Self {
impl<'a> DouyinLive<'a> {
fn new(request: LiveRequest, ttwid_cache: &'a Mutex<Option<String>>) -> 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<LiveStatus> {
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<bool> {
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<Value> {
@@ -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<String> {
.map(|captures| captures[1].to_string())
}
async fn fetch_ttwid(client: &Client) -> String {
async fn fetch_ttwid(client: &Client) -> Option<String> {
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 {
+274 -41
View File
@@ -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<Option<CachedEncryptKey>>,
/// 进程级 url -> 真实房间号缓存(对应 Python get_real_rid 的 alru_cache)
real_room_id: RwLock<HashMap<String, String>>,
}
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<LiveStatus> {
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<String>,
encrypt_key_cache: &'a Mutex<Option<CachedEncryptKey>>,
real_room_id_cache: &'a RwLock<HashMap<String, String>>,
}
impl DouyuLive {
fn new(request: LiveRequest) -> Self {
impl<'a> DouyuLive<'a> {
fn new(
request: LiveRequest,
encrypt_key_cache: &'a Mutex<Option<CachedEncryptKey>>,
real_room_id_cache: &'a RwLock<HashMap<String, String>>,
) -> 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<Option<RoomInfo>> {
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<String, reqwest::Error> {
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<bool> {
let data: Value = self
.client
@@ -262,11 +330,8 @@ impl DouyuLive {
}
async fn request_web_play_info(&self, room_id: &str, cdn: &str) -> LiveResult<PlayInfo> {
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<WhiteEncryptKey> {
/// 获取加密密钥及其配套 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<DanmakuSource> {
@@ -440,6 +525,22 @@ impl DouyuLive {
}
}
fn unix_now() -> LiveResult<u64> {
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<Option<CachedEncryptKey>>,
real_room_id_cache: &'a RwLock<HashMap<String, String>>,
) -> 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");
}
}
+380 -40
View File
@@ -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<HashMap<String, String>>,
}
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<LiveStatus> {
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<HashMap<String, String>>,
}
impl HuyaLive {
fn new(request: LiveRequest) -> Self {
impl<'a> HuyaLive<'a> {
fn new(request: LiveRequest, real_rid_cache: &'a RwLock<HashMap<String, String>>) -> 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<LiveStatus> {
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<String> {
let room_id = self
/// 解析房间号:URL 路径本身是数字则直接使用,
/// 否则请求房间页取 TT_ROOM_DATA.profileRoom(结果进程内缓存)
async fn resolve_room_id(&self) -> LiveResult<String> {
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<Option<HuyaRoomProfile>> {
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<Option<HuyaRoomProfile>> {
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<Value> = 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<String> {
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<Vec<(String, String)>> {
async fn build_stream_urls(&self, streams_info: &[Value]) -> LiveResult<Vec<(String, String)>> {
let mut streams = Vec::new();
// 同一房间所有 CDN 的 stream_name 相同,防盗链参数只需计算一次
let mut cached_anticode: Option<String> = 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<String> {
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<String> {
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<Value> {
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<usize> {
None
}
fn build_anticode(stream_name: &str, anti_code: &str) -> LiveResult<String> {
fn build_anticode(
stream_name: &str,
anti_code: &str,
presenter_uid: u64,
) -> LiveResult<String> {
let query = serde_urlencoded::from_str::<HashMap<String, String>>(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<String> {
.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::<u64>(), 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<String> {
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<String> {
.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("&gt;", ">")
}
/// 返回 master m3u8 中第一个 #EXT-X-STREAM-INF 变体的 URI;媒体列表返回 None
fn first_m3u8_variant(playlist: &str) -> Option<String> {
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<String> {
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#"
@@ -0,0 +1,453 @@
//! 虎牙 WUP(getCdnTokenInfoEx) 请求所需的最小 tars 编解码。
//!
//! 只实现该接口用到的子集:TarsV3 uni-packet(`map<string, bytes>` 载荷)、
//! `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<u8>,
}
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<u32> {
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<i64> {
let (_, ty) = self.read_head()?;
self.read_int_body(ty)
}
fn read_int_body(&mut self, ty: u8) -> Option<i64> {
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<String> {
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<Vec<u8>> {
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<String> {
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<Vec<u8>> {
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<u8> {
// 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<string, bytes> { "tReq": <encoded req> }
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<String> {
let data = body.get(4..)?;
let mut packet = TarsInputStream::new(data);
let payload = packet.read_bytes(7)?;
// map<string, bytes>,找 "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<u8> {
(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");
}
}
}
@@ -119,6 +119,7 @@ struct InkeData {
status: bool,
#[serde(default)]
live_addr: Vec<InkeLiveAddr>,
#[serde(alias = "live_name")]
title: Option<String>,
image: Option<String>,
}
@@ -153,6 +153,6 @@ struct KilakilaBody {
#[serde(rename = "hlsPlayUrl")]
hls_play_url: Option<String>,
title: Option<String>,
#[serde(rename = "coverUrl")]
#[serde(rename = "coverUrl", alias = "backPic")]
cover_url: Option<String>,
}
+5
View File
@@ -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,
}
@@ -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 {
+214 -82
View File
@@ -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<Option<String>> = RwLock::new(None);
/// 返回可用的 auth_token:未配置、为空或已被记为失效时返回 None。
fn effective_auth_token(auth_token: Option<&str>) -> Option<String> {
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<T: Serialize>(
&self,
ops: T,
auth_token: Option<&str>,
) -> LiveResult<GqlResponse> {
for retry in 0..2 {
async fn post_gql<T: Serialize>(&self, ops: T) -> LiveResult<GqlResponse> {
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<DanmakuSource> {
@@ -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<Option<Value>> {
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<Option<Value>> {
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<Option<Value>> {
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<String> {
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<PathBuf> {
let auth_token = effective_auth_token(self.twitch_auth_token.as_deref())?;
write_cookie_file(&auth_token).await
}
async fn archive_ids(&self) -> HashSet<String> {
@@ -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<String> {
.map(str::to_string)
})
}
/// 判断条目 id 是否已归档。yt-dlp 新版对 Twitch VOD 记录的归档 id 为 `v<数字>`
/// (flat 条目 id 同为 `v<数字>`),旧版本归档为纯数字,两种形式都视为已归档。
fn archive_contains(archive_ids: &HashSet<String>, 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<PathBuf> {
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<String> = ["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())
);
}
}
+37 -23
View File
@@ -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<String>,
last_update: u64,
}
/// WBI 签名器。key 状态放在 `Arc` 内共享:插件结构体持有一份,
/// 每次 `check_stream` 克隆句柄即可跨轮询复用,使 2 小时更新间隔真正生效。
#[derive(Clone, Default)]
pub struct WbiSigner {
client: reqwest::Client,
key: Arc<RwLock<Option<String>>>,
last_update: Arc<RwLock<u64>>,
state: Arc<RwLock<WbiState>>,
}
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<String> {
@@ -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<String, String>,
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()
+248 -49
View File
@@ -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<LiveStatus> {
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<Option<Value>> {
async fn extract_info(&self, url: &str, flat: bool) -> LiveResult<Option<Value>> {
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<String>,
) -> Pin<Box<dyn Future<Output = LiveResult<EntryDecision>> + 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<String, Value>,
archive_ids: &HashSet<String>,
) -> Option<&'a Value> {
let object = value.as_object()?;
if self.entry_allowed(object, archive_ids) {
return Some(value);
) -> LiveResult<EntryDecision> {
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<String, Value>, archive_ids: &HashSet<String>) -> 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<Value> {
async fn resolve_entry(&self, entry: Value) -> LiveResult<Value> {
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<String> = ["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<String> = ["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
);
}
}