From 03c36e1da7cfefe16ff3e1bd2ac040a336a2e863 Mon Sep 17 00:00:00 2001 From: ForgQi <34411314+ForgQi@users.noreply.github.com> Date: Tue, 14 Jul 2026 21:33:23 +0800 Subject: [PATCH] Add batch live-check support + fixes Introduce batch live-check infrastructure and implementations. Add BatchCheckRequest and LivePlugin::supports_batch_check/batch_check API; Monitor uses BatchLiveCache and batch_verdict to prefilter rooms (platform-level checks + caching and fast rotation). RoomsActor gains PlatformUrls. Implement batch_check for Picarto (explore pagination) and Twitch (GQL batching with auth retry). Other fixes: Douyu auth-failure detection and key invalidation + retry, Inke optional response handling, streamlink runtime tweaks for yt-dlp/Twitch ffmpeg, small test additions and import refactors. --- crates/biliup-cli/src/server/core/live.rs | 38 +++++- crates/biliup-cli/src/server/core/monitor.rs | 135 ++++++++++++++++++- crates/biliup/src/downloader/live/douyu.rs | 93 +++++++++++-- crates/biliup/src/downloader/live/inke.rs | 15 ++- crates/biliup/src/downloader/live/mod.rs | 23 ++++ crates/biliup/src/downloader/live/picarto.rs | 77 ++++++++++- crates/biliup/src/downloader/live/twitch.rs | 106 ++++++++++++++- 7 files changed, 457 insertions(+), 30 deletions(-) diff --git a/crates/biliup-cli/src/server/core/live.rs b/crates/biliup-cli/src/server/core/live.rs index 009ac3a..412eb28 100644 --- a/crates/biliup-cli/src/server/core/live.rs +++ b/crates/biliup-cli/src/server/core/live.rs @@ -9,10 +9,10 @@ use crate::server::core::downloader::{DownloaderRuntime, DownloaderType, RustDan use crate::server::infrastructure::context::Worker; use crate::server::infrastructure::models::StreamerInfo; use biliup::downloader::live::{ - BilibiliOptions, CcOptions, DanmakuSource, DouyinOptions, DouyuOptions, DownloaderHint, - HuyaOptions, KilakilaOptions, KuaishouOptions, LiveCredentials, LiveOptions, LiveRequest, - LiveStream, RuntimeOptions, StreamlinkOptions, StreamlinkPlatform, TwitcastingOptions, - TwitchOptions, YoutubeOptions, YtDlpBackend, YtDlpOptions, + BatchCheckRequest, BilibiliOptions, CcOptions, DanmakuSource, DouyinOptions, DouyuOptions, + DownloaderHint, HuyaOptions, KilakilaOptions, KuaishouOptions, LiveCredentials, LiveOptions, + LiveRequest, LiveStream, RuntimeOptions, StreamlinkOptions, StreamlinkPlatform, + TwitcastingOptions, TwitchOptions, YoutubeOptions, YtDlpBackend, YtDlpOptions, }; use danmaku_client::{PlatformContext, RecorderConfig}; use std::path::PathBuf; @@ -29,6 +29,17 @@ pub fn live_request(worker: &Worker) -> LiveRequest { } } +/// 以某个 worker 的客户端与配置为基础，构造同平台的批量检测请求。 +pub fn batch_check_request(worker: &Worker, urls: Vec) -> BatchCheckRequest { + let config = worker.get_config(); + BatchCheckRequest { + client: worker.client.client.clone(), + urls, + options: live_options(&config), + credentials: live_credentials(&config), + } +} + fn live_options(config: &Config) -> LiveOptions { LiveOptions { bilibili: BilibiliOptions { @@ -170,6 +181,20 @@ pub fn downloader_runtime( match downloader_type { DownloaderType::Streamlink => streamlink_runtime(stream), + // Twitch 指定 ffmpeg 时也走 streamlink：Python 版 ffmpeg 消费的是 + // streamlink --player-external-http 的代理输出（twitch.py:114-144）， + // 去广告与 OAuth 鉴权都由 streamlink 完成，直连 usher 直链会丢掉这两项 + DownloaderType::Ffmpeg + if matches!( + stream.runtime_options.as_ref(), + Some(RuntimeOptions::Streamlink(StreamlinkOptions { + platform: StreamlinkPlatform::Twitch { .. }, + .. + })) + ) => + { + streamlink_runtime(stream) + } DownloaderType::YtDlp | DownloaderType::Ytarchive => ytdlp_runtime(stream, downloader_type), _ => DownloaderRuntime::from_type(downloader_type), } @@ -181,6 +206,11 @@ fn streamlink_runtime(stream: &LiveStream) -> DownloaderRuntime { url.clone().unwrap_or_else(|| stream.raw_stream_url.clone()), streamlink_platform(platform), ), + // yt-dlp 型来源（如 YouTube）交给 streamlink 时，传入网页地址让其自行提取， + // 而非已解析的 manifest 直链（对齐 youtube.py:96-101） + Some(RuntimeOptions::YtDlp(options)) if !options.webpage_url.is_empty() => { + (options.webpage_url.clone(), Platform::Generic) + } _ => (stream.raw_stream_url.clone(), Platform::Generic), }; let downloader = diff --git a/crates/biliup-cli/src/server/core/monitor.rs b/crates/biliup-cli/src/server/core/monitor.rs index 5b2f166..996936a 100644 --- a/crates/biliup-cli/src/server/core/monitor.rs +++ b/crates/biliup-cli/src/server/core/monitor.rs @@ -1,6 +1,6 @@ use crate::server::common::download::start_download_workflow; use crate::server::common::upload::UploaderMessage; -use crate::server::core::live::{live_request, streamer_info}; +use crate::server::core::live::{batch_check_request, live_request, streamer_info}; use crate::server::infrastructure::connection_pool::ConnectionPool; use crate::server::infrastructure::context::{Context, Stage, Worker, WorkerStatus}; use crate::server::infrastructure::models::StreamerInfo; @@ -9,13 +9,35 @@ use biliup::downloader::live::{LivePlugin, LiveStatus}; use ormlite::Model; use ormlite::model::ModelBuilder; use std::collections::hash_map::Entry; -use std::collections::{HashMap, VecDeque}; +use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::{Arc, RwLock}; -use std::time::Duration; +use std::time::{Duration, Instant}; use tokio::sync::{OwnedSemaphorePermit, Semaphore, oneshot}; use tokio::task::JoinHandle; use tracing::{debug, error, info, trace, warn}; +/// 批量检测平台的开播缓存：一次批量请求的结果在一个检测周期内复用， +/// 供逐间循环快速判定某房间是否开播，避免对未开播房间做逐间请求。 +#[derive(Debug)] +struct BatchLiveCache { + live_urls: HashSet, + refreshed_at: Instant, +} + +/// 批量检测对单个房间的判定结果。 +enum BatchVerdict { + /// 批量结果显示开播，继续逐间完整检测以获取流信息 + Live, + /// 批量结果显示未开播，直接放回队列 + Offline, + /// 平台不支持批量检测或本轮批量请求失败，回退逐间检测 + Fallback, +} + +/// 批量结果为未开播时的快速轮换间隔：缓存命中无网络开销， +/// 短暂停即可在一个检测周期内扫完整个队列（网络频率仍由缓存 TTL 限制）。 +const BATCH_ROTATE_SLEEP: Duration = Duration::from_secs(1); + /// 房间处理器 /// 管理多个直播间的状态和操作 #[derive(Debug)] @@ -31,6 +53,8 @@ pub struct Monitor { /// 许可由下载任务持有到录制结束，pool1_size 的唯一限流语义在这里表达。 download_slots: Arc, monitors: RwLock>>, + /// 各批量检测平台的开播缓存（platform_name -> 最近一次批量结果）。 + batch_live: RwLock>, } impl Drop for Monitor { @@ -70,6 +94,7 @@ impl Monitor { uploader, download_slots, monitors: Default::default(), + batch_live: Default::default(), } } @@ -93,6 +118,21 @@ impl Monitor { .await; let url = room.get_streamer().url.clone(); let interval = room.get_config().event_loop_interval; + // 批量检测平台：先用一个检测周期内共享的批量结果快速判定是否开播。 + // 未开播直接跳过（不占用下载槽位、不做逐间请求），并以较短间隔轮换到 + // 下一个房间，使整条队列在一个检测周期内扫完；批量请求本身由缓存 TTL 限流。 + if plugin.supports_batch_check() { + match self.batch_verdict(platform_name, &plugin, &room, interval).await { + BatchVerdict::Offline => { + self.wake_waker(room.id()).await; + debug!(url = url, "批量检测未开播"); + tokio::time::sleep(BATCH_ROTATE_SLEEP).await; + continue; + } + // Live / Fallback 都继续走下面的逐间完整检测 + BatchVerdict::Live | BatchVerdict::Fallback => {} + } + } let Some(download_permit) = self.try_acquire_download_slot(&room).await else { self.wake_waker(room.id()).await; tokio::time::sleep(Duration::from_secs(interval)).await; @@ -163,6 +203,70 @@ impl Monitor { } } + /// 用批量检测结果判定单个房间是否开播。 + /// 缓存超过一个检测周期即刷新：以当前房间的客户端/配置为该平台所有房间发起一次批量请求。 + async fn batch_verdict( + self: &Arc, + platform_name: &str, + plugin: &Arc, + room: &Arc, + interval: u64, + ) -> BatchVerdict { + let url = room.get_streamer().url.clone(); + + // 命中未过期缓存直接判定 + if let Some(cache) = self.batch_live.read().unwrap().get(platform_name) + && cache.refreshed_at.elapsed() < Duration::from_secs(interval) + { + return if cache.live_urls.contains(&url) { + BatchVerdict::Live + } else { + BatchVerdict::Offline + }; + } + + // 缓存过期或缺失：为该平台所有房间发起一次批量检测 + let urls = self.platform_urls(platform_name).await; + if urls.is_empty() { + return BatchVerdict::Fallback; + } + let request = batch_check_request(room, urls); + match plugin.batch_check(request).await { + Ok(live_urls) => { + let live_urls: HashSet = live_urls.into_iter().collect(); + let is_live = live_urls.contains(&url); + self.batch_live.write().unwrap().insert( + platform_name.to_string(), + BatchLiveCache { + live_urls, + refreshed_at: Instant::now(), + }, + ); + if is_live { + BatchVerdict::Live + } else { + BatchVerdict::Offline + } + } + Err(e) => { + // 批量请求失败时回退逐间检测，避免整平台漏检 + warn!(platform = platform_name, e = ?e, "批量检测失败，回退逐间检测"); + BatchVerdict::Fallback + } + } + } + + /// 获取某平台当前队列中所有房间的 URL（用于批量检测）。 + async fn platform_urls(self: &Arc, platform_name: &str) -> Vec { + let (send, recv) = oneshot::channel(); + let msg = ActorMessage::PlatformUrls { + respond_to: send, + platform_name: platform_name.to_owned(), + }; + let _ = self.sender.send(msg).await; + recv.await.unwrap_or_default() + } + /// 添加工作器到房间列表 /// /// # 参数 @@ -350,6 +454,11 @@ enum ActorMessage { respond_to: oneshot::Sender>>, platform_name: String, }, + /// 获取某平台所有房间的 URL（用于批量检测） + PlatformUrls { + respond_to: oneshot::Sender>, + platform_name: String, + }, /// 添加工作器 Add( oneshot::Sender>>, @@ -425,6 +534,13 @@ impl RoomsActor { // 如果使用`select!`宏取消等待响应，可能会发生这种情况 let _ = respond_to.send(self.next(&platform_name)); } + ActorMessage::PlatformUrls { + respond_to, + platform_name, + } => { + // `let _ =` 忽略发送时的任何错误 + let _ = respond_to.send(self.platform_urls(&platform_name)); + } ActorMessage::Add(respond_to, worker) => { let plugin = self.add(worker); let _ = respond_to.send(plugin); @@ -510,6 +626,19 @@ impl RoomsActor { Some(arc) } + /// 获取某平台所有房间的 URL（含正在检测、已弹出队列的房间）。 + /// 以 all_workers 为源按插件归属过滤，保证覆盖整平台而非仅队列内房间。 + fn platform_urls(&self, platform_name: &str) -> Vec { + self.all_workers + .iter() + .filter(|worker| { + self.matches(&worker.live_streamer.url) + .is_some_and(|plugin| plugin.name() == platform_name) + }) + .map(|worker| worker.live_streamer.url.clone()) + .collect() + } + /// 放回工作队列 fn push_back(&mut self, id: i64) -> Option> { // 在总数组中找不到，说明该房间已被移除我们也不放回 diff --git a/crates/biliup/src/downloader/live/douyu.rs b/crates/biliup/src/downloader/live/douyu.rs index 39cc4d4..9d96e37 100644 --- a/crates/biliup/src/downloader/live/douyu.rs +++ b/crates/biliup/src/downloader/live/douyu.rs @@ -306,10 +306,14 @@ impl<'a> DouyuLive<'a> { async fn get_web_play_info(&self, room_id: &str) -> LiveResult { let mut cdn = self.douyu_cdn.clone(); let mut last_error = None; + // 鉴权失败时最多刷新一次加密密钥后重试（对齐 rust-srec get_play_info_fallback）： + // 缓存密钥可能在本地 expire_at 之前就被服务端判为失效，此时需强制刷新。 + let mut key_refreshed = false; - for _ in 0..2 { + // scdn 规避最多重试 2 次；叠加一次鉴权刷新，循环上界取 3 + for _ in 0..3 { match self.request_web_play_info(room_id, &cdn).await { - Ok(play_info) => { + Ok(PlayOutcome::Ok(play_info)) => { if play_info.rtmp_cdn.starts_with("scdn") && let Some(next_cdn) = play_info .cdns_with_name @@ -322,6 +326,15 @@ impl<'a> DouyuLive<'a> { } return Ok(play_info); } + Ok(PlayOutcome::AuthFailed) => { + if !key_refreshed { + self.invalidate_key().await; + key_refreshed = true; + continue; + } + last_error = Some(LiveError::custom("斗鱼播放信息鉴权失败")); + break; + } Err(err) => last_error = Some(err), } } @@ -329,7 +342,12 @@ impl<'a> DouyuLive<'a> { Err(last_error.unwrap_or_else(|| LiveError::custom("获取斗鱼播放信息失败"))) } - async fn request_web_play_info(&self, room_id: &str, cdn: &str) -> LiveResult { + /// 清空缓存的加密密钥，强制下次 update_key 重新向 getEncryption 请求。 + async fn invalidate_key(&self) { + *self.encrypt_key_cache.lock().await = None; + } + + async fn request_web_play_info(&self, room_id: &str, cdn: &str) -> LiveResult { let now = unix_now()?; let (encrypt_key, user_agent) = self.update_key().await?; let auth = sign_stream(&encrypt_key, room_id, now); @@ -350,7 +368,7 @@ impl<'a> DouyuLive<'a> { ("auth", auth), ]; - let rsp: PlayResponse = self + let rsp = self .client .post(format!( "https://{DOUYU_WEB_DOMAIN}/lapi/live/getH5PlayV1/{room_id}" @@ -362,17 +380,36 @@ impl<'a> DouyuLive<'a> { .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}")) })?; + let status = rsp.status(); + let body = rsp.text().await.map_err(|err| { + LiveError::custom(format!("读取斗鱼播放信息失败 room_id: {room_id}: {err}")) + })?; + + // 鉴权失败：HTTP 403 或响应体含「鉴权失败」，由调用方刷新密钥后重试 + // （对齐 rust-srec is_douyu_auth_failed） + if is_douyu_auth_failed(status.as_u16(), &body) { + return Ok(PlayOutcome::AuthFailed); + } + + let rsp: PlayResponse = serde_json::from_str(&body).map_err(|err| { + LiveError::custom(format!("解析斗鱼播放信息失败 room_id: {room_id}: {err}")) + })?; + if rsp.error == 0 && let Some(data) = rsp.data { - return Ok(data); + return Ok(PlayOutcome::Ok(data)); + } + + // JSON 形式返回的鉴权失败同样触发刷新重试 + if rsp + .msg + .as_deref() + .is_some_and(|msg| msg.contains("鉴权失败")) + { + return Ok(PlayOutcome::AuthFailed); } if rsp.error == -5 { @@ -532,6 +569,27 @@ fn unix_now() -> LiveResult { .as_secs()) } +/// getH5PlayV1 请求结果：正常拿到播放信息，或需刷新密钥重试的鉴权失败。 +enum PlayOutcome { + Ok(PlayInfo), + AuthFailed, +} + +/// 判定 getH5PlayV1 是否鉴权失败：HTTP 403，或响应体（去空白、去引号后）含「鉴权失败」 +/// （对齐 rust-srec is_douyu_auth_failed / normalize_douyu_error_body）。 +fn is_douyu_auth_failed(status: u16, body: &str) -> bool { + if status == 403 { + return true; + } + let normalized: String = body + .trim() + .trim_matches('"') + .chars() + .filter(|ch| !ch.is_whitespace()) + .collect(); + normalized.contains("鉴权失败") +} + /// 生成随机 Chrome UA（对应 biliup/plugins/__init__.py 的 random_user_agent()， /// 桌面端格式，Chrome 大版本随机 100~120） fn random_chrome_user_agent() -> String { @@ -831,8 +889,7 @@ mod tests { } #[test] - fn white_encrypt_key_missing_expire_at_defaults_to_zero() { - let key: WhiteEncryptKey = serde_json::from_str( + 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(); @@ -892,4 +949,16 @@ mod tests { assert_eq!(live.resolve_room_id().await.unwrap(), "123456"); } + + #[test] + fn is_douyu_auth_failed_detects_403_and_message() { + // HTTP 403 一律视为鉴权失败 + assert!(is_douyu_auth_failed(403, "")); + // 响应体含「鉴权失败」（含空白/引号包裹）也算 + assert!(is_douyu_auth_failed(200, r#" "鉴 权 失 败" "#)); + assert!(is_douyu_auth_failed(200, "鉴权失败")); + // 正常响应不算 + assert!(!is_douyu_auth_failed(200, r#"{"error":0,"data":{}}"#)); + assert!(!is_douyu_auth_failed(200, "")); + } } diff --git a/crates/biliup/src/downloader/live/inke.rs b/crates/biliup/src/downloader/live/inke.rs index 82eac17..6f66381 100644 --- a/crates/biliup/src/downloader/live/inke.rs +++ b/crates/biliup/src/downloader/live/inke.rs @@ -70,11 +70,13 @@ impl InkeLive { .await .map_err(|err| LiveError::custom(format!("解析映客直播间信息失败: {err}")))?; - if response.error_code != 0 || !response.data.status { + let Some(data) = response.data.filter(|_| response.error_code == 0) else { + return Ok(LiveStatus::Offline); + }; + if !data.status { return Ok(LiveStatus::Offline); } - let raw_stream_url = response - .data + let raw_stream_url = data .live_addr .first() .map(|addr| addr.stream_addr.clone()) @@ -85,9 +87,9 @@ impl InkeLive { stream: Box::new(LiveStream { name: self.name.clone(), url: self.url.clone(), - title: response.data.title.unwrap_or(uid), + title: data.title.unwrap_or(uid), date: Utc::now(), - live_cover_url: response.data.image.unwrap_or_default(), + live_cover_url: data.image.unwrap_or_default(), suffix: media_ext_from_url(&raw_stream_url).unwrap_or_else(|| "flv".to_string()), raw_stream_url, platform: "inke".to_string(), @@ -111,7 +113,8 @@ impl InkeLive { #[derive(Deserialize)] struct InkeResponse { error_code: i32, - data: InkeData, + #[serde(default)] + data: Option, } #[derive(Deserialize)] diff --git a/crates/biliup/src/downloader/live/mod.rs b/crates/biliup/src/downloader/live/mod.rs index 4e854ec..64495fb 100644 --- a/crates/biliup/src/downloader/live/mod.rs +++ b/crates/biliup/src/downloader/live/mod.rs @@ -75,6 +75,29 @@ pub trait LivePlugin: Send + Sync { fn name(&self) -> &'static str; fn matches(&self, url: &str) -> bool; async fn check_stream(&self, request: LiveRequest) -> LiveResult; + + /// 是否支持批量直播检测。默认 false。 + /// 支持的平台可用一次请求判定多个直播间的开播状态（对齐 Python `BatchCheck`）， + /// 监控侧据此先批量过滤，未开播的房间跳过逐间检测。 + fn supports_batch_check(&self) -> bool { + false + } + + /// 批量检测：给定同平台的一批 URL，返回其中正在直播的 URL 子集。 + /// 默认返回空（不支持批量检测的平台不应被调用到）。 + async fn batch_check(&self, request: BatchCheckRequest) -> LiveResult> { + let _ = request; + Ok(Vec::new()) + } +} + +/// 批量检测请求。承载同平台待检测的 URL 列表与所需的客户端 / 凭据 / 选项。 +#[derive(Debug, Clone)] +pub struct BatchCheckRequest { + pub client: Client, + pub urls: Vec, + pub options: LiveOptions, + pub credentials: LiveCredentials, } #[derive(Debug, Clone)] diff --git a/crates/biliup/src/downloader/live/picarto.rs b/crates/biliup/src/downloader/live/picarto.rs index 79178f5..4fa22aa 100644 --- a/crates/biliup/src/downloader/live/picarto.rs +++ b/crates/biliup/src/downloader/live/picarto.rs @@ -1,6 +1,6 @@ use super::{ - DownloaderHint, LiveError, LivePlugin, LiveRequest, LiveResult, LiveStatus, LiveStream, - media_ext_from_url, + BatchCheckRequest, DownloaderHint, LiveError, LivePlugin, LiveRequest, LiveResult, LiveStatus, + LiveStream, media_ext_from_url, }; use async_trait::async_trait; use chrono::Utc; @@ -10,6 +10,8 @@ use std::collections::HashMap; const PICARTO_API_CHANNEL: &str = "https://ptvintern.picarto.tv/api/channel/detail"; const PICARTO_HLS_URL: &str = "https://{origin}.picarto.tv/stream/hls/{stream_name}/index.m3u8"; +/// 批量检测的 explore 接口：按观看人数倒序、每页 100 条、含成人内容。 +const PICARTO_API_EXPLORE: &str = "https://ptvintern.picarto.tv/api/explore?first=100&page={page}&filter_params%5Badult%5D=true&order_by%5Bfield%5D=viewers&order_by%5Border%5D=DESC&type=stream"; pub struct Picarto { re: Regex, @@ -42,6 +44,63 @@ impl LivePlugin for Picarto { async fn check_stream(&self, request: LiveRequest) -> LiveResult { PicartoLive::new(request).check_stream().await } + + fn supports_batch_check(&self) -> bool { + true + } + + /// 批量检测：分页拉取 explore 列表（当前正在直播的频道）， + /// 返回其中用户名命中待检测 URL 的那些 URL（对齐 picarto.py:61-78）。 + async fn batch_check(&self, request: BatchCheckRequest) -> LiveResult> { + // 用户名（小写）-> 原始待检测 URL，便于用 explore 结果反查 + let mut wanted: HashMap = HashMap::new(); + for url in &request.urls { + if let Some(name) = self.username_of(url) { + wanted.insert(name.to_lowercase(), url.clone()); + } + } + if wanted.is_empty() { + return Ok(Vec::new()); + } + + let mut live_urls = Vec::new(); + let mut page = 1u32; + loop { + let explore: PicartoExplore = request + .client + .get(PICARTO_API_EXPLORE.replace("{page}", &page.to_string())) + .send() + .await + .map_err(|err| LiveError::custom(format!("获取 Picarto explore 失败: {err}")))? + .json() + .await + .map_err(|err| { + LiveError::custom(format!("解析 Picarto explore 失败: {err}")) + })?; + + for entry in &explore.data { + if let Some(url) = wanted.remove(&entry.name.to_lowercase()) { + live_urls.push(url); + } + } + + if wanted.is_empty() || explore.next_page_url.is_none() { + break; + } + page += 1; + } + + Ok(live_urls) + } +} + +impl Picarto { + fn username_of(&self, url: &str) -> Option { + self.re + .captures(url) + .and_then(|caps| caps.name("id")) + .map(|m| m.as_str().to_string()) + } } struct PicartoLive { @@ -161,3 +220,17 @@ struct PicartoStream { stream_name: String, thumbnail_image: Option, } + +#[derive(Deserialize)] +struct PicartoExplore { + #[serde(default)] + data: Vec, + #[serde(default)] + next_page_url: Option, +} + +#[derive(Deserialize)] +struct PicartoExploreEntry { + #[serde(default)] + name: String, +} diff --git a/crates/biliup/src/downloader/live/twitch.rs b/crates/biliup/src/downloader/live/twitch.rs index 5bcd7c0..a645b62 100644 --- a/crates/biliup/src/downloader/live/twitch.rs +++ b/crates/biliup/src/downloader/live/twitch.rs @@ -1,7 +1,7 @@ use super::{ - DanmakuSource, DownloaderHint, LiveError, LivePlugin, LiveRequest, LiveResult, LiveStatus, - LiveStream, RuntimeOptions, StreamlinkOptions, StreamlinkPlatform, YtDlpOptions, - media_ext_from_url, + BatchCheckRequest, DanmakuSource, DownloaderHint, LiveError, LivePlugin, LiveRequest, + LiveResult, LiveStatus, LiveStream, RuntimeOptions, StreamlinkOptions, StreamlinkPlatform, + YtDlpOptions, media_ext_from_url, }; use async_trait::async_trait; use chrono::Utc; @@ -23,6 +23,8 @@ use tracing::warn; const CLIENT_ID: &str = "kimne78kx3ncx6brgo4mv6wki5h1ko"; const GQL_URL: &str = "https://gql.twitch.tv/gql"; +/// 单次 GQL 批量请求的最大操作数（对齐 twitch.py TwitchUtils.post_gql limit=30） +const GQL_BATCH_LIMIT: usize = 30; /// Twitch 已失效的 auth_token（Twitch 与 TwitchVideos 共享，进程内长期记忆）。 /// 对应 Python 版 TwitchUtils._invalid_auth_token 类级属性。 @@ -118,6 +120,64 @@ impl LivePlugin for Twitch { .check_stream() .await } + + fn supports_batch_check(&self) -> bool { + true + } + + /// 批量检测：一次 GQL 请求最多判定 30 个直播间（对齐 twitch.py:159-184）。 + async fn batch_check(&self, request: BatchCheckRequest) -> LiveResult> { + let auth_token = request.credentials.twitch_cookie.clone(); + let mut live_urls = Vec::new(); + + for chunk in request.urls.chunks(GQL_BATCH_LIMIT) { + // 无法解析出频道名的 URL 保留占位，保证响应下标与 chunk 对齐 + let ops: Vec = chunk + .iter() + .map(|url| match self.batch_channel_name(url) { + Some(channel_name) => json!({ + "query": r#" + query query($login:String!) { + user(login: $login){ + stream { + type + } + } + } + "#, + "variables": { "login": channel_name } + }), + None => Value::Null, + }) + .collect(); + + let responses = + post_gql_batch(&request.client, auth_token.as_deref(), &ops).await?; + + for (index, url) in chunk.iter().enumerate() { + let is_live = responses + .get(index) + .and_then(|resp| resp.data.as_ref()) + .and_then(|data| data.user.as_ref()) + .and_then(|user| user.stream.as_ref()) + .is_some_and(|stream| stream.stream_type.as_deref() == Some("live")); + if is_live { + live_urls.push(url.clone()); + } + } + } + + Ok(live_urls) + } +} + +impl Twitch { + fn batch_channel_name(&self, url: &str) -> Option { + self.re + .captures(url) + .and_then(|caps| caps.name("id")) + .map(|m| m.as_str().to_lowercase()) + } } struct TwitchLive { @@ -521,6 +581,46 @@ fn gql_headers(auth_token: Option<&str>) -> LiveResult { Ok(headers) } +/// 批量 POST GQL 操作数组。带 token 收到 Unauthorized 时记为失效并匿名重试一次 +/// （对齐 twitch.py TwitchUtils.__post_gql 的失效处理）。 +async fn post_gql_batch( + client: &Client, + auth_token: Option<&str>, + ops: &[Value], +) -> LiveResult> { + let mut auth_token = effective_auth_token(auth_token); + loop { + let resp = client + .post(GQL_URL) + .headers(gql_headers(auth_token.as_deref())?) + .json(&ops) + .timeout(Duration::from_secs(15)) + .send() + .await + .map_err(|err| LiveError::custom(format!("请求 Twitch GQL 失败: {err}")))?; + if !resp.status().is_success() { + return Err(LiveError::custom(format!( + "请求 Twitch GQL 错误: {}", + resp.status() + ))); + } + let body: Value = resp + .json() + .await + .map_err(|err| LiveError::custom(format!("解析 Twitch GQL 失败: {err}")))?; + // 鉴权失败时返回的是 {"error": "Unauthorized"} 对象而非数组 + if body.get("error").and_then(Value::as_str) == Some("Unauthorized") { + if let Some(token) = auth_token.take() { + mark_auth_token_invalid(&token); + continue; + } + return Err(LiveError::custom("请求 Twitch GQL 错误: Unauthorized")); + } + return serde_json::from_value(body) + .map_err(|err| LiveError::custom(format!("解析 Twitch GQL 批量响应失败: {err}"))); + } +} + fn parse_ytdlp_stdout(stdout: &[u8]) -> LiveResult> { let stdout = String::from_utf8_lossy(stdout); let stdout = stdout.trim();