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

@@ -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<String>) -> 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 =

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<String>,
+    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<Semaphore>,
     monitors: RwLock<HashMap<String, JoinHandle<()>>>,
+    /// 各批量检测平台的开播缓存（platform_name -> 最近一次批量结果）。
+    batch_live: RwLock<HashMap<String, BatchLiveCache>>,
 }
 
 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<Self>,
+        platform_name: &str,
+        plugin: &Arc<dyn LivePlugin + Send + Sync>,
+        room: &Arc<Worker>,
+        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<String> = 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<Self>, platform_name: &str) -> Vec<String> {
+        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<Option<Arc<Worker>>>,
         platform_name: String,
     },
+    /// 获取某平台所有房间的 URL（用于批量检测）
+    PlatformUrls {
+        respond_to: oneshot::Sender<Vec<String>>,
+        platform_name: String,
+    },
     /// 添加工作器
     Add(
         oneshot::Sender<Option<Arc<dyn LivePlugin + Send + Sync>>>,
@@ -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<String> {
+        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<Arc<dyn LivePlugin + Send + Sync>> {
         // 在总数组中找不到，说明该房间已被移除我们也不放回

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<PlayInfo> {
         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<PlayInfo> {
+    /// 清空缓存的加密密钥，强制下次 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<PlayOutcome> {
         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<u64> {
         .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, ""));
+    }
 }

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<InkeData>,
 }
 
 #[derive(Deserialize)]

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<LiveStatus>;
+
+    /// 是否支持批量直播检测。默认 false。
+    /// 支持的平台可用一次请求判定多个直播间的开播状态（对齐 Python `BatchCheck`），
+    /// 监控侧据此先批量过滤，未开播的房间跳过逐间检测。
+    fn supports_batch_check(&self) -> bool {
+        false
+    }
+
+    /// 批量检测：给定同平台的一批 URL，返回其中正在直播的 URL 子集。
+    /// 默认返回空（不支持批量检测的平台不应被调用到）。
+    async fn batch_check(&self, request: BatchCheckRequest) -> LiveResult<Vec<String>> {
+        let _ = request;
+        Ok(Vec::new())
+    }
+}
+
+/// 批量检测请求。承载同平台待检测的 URL 列表与所需的客户端 / 凭据 / 选项。
+#[derive(Debug, Clone)]
+pub struct BatchCheckRequest {
+    pub client: Client,
+    pub urls: Vec<String>,
+    pub options: LiveOptions,
+    pub credentials: LiveCredentials,
 }
 
 #[derive(Debug, Clone)]

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<LiveStatus> {
         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<Vec<String>> {
+        // 用户名（小写）-> 原始待检测 URL，便于用 explore 结果反查
+        let mut wanted: HashMap<String, String> = 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<String> {
+        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<String>,
 }
+
+#[derive(Deserialize)]
+struct PicartoExplore {
+    #[serde(default)]
+    data: Vec<PicartoExploreEntry>,
+    #[serde(default)]
+    next_page_url: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct PicartoExploreEntry {
+    #[serde(default)]
+    name: String,
+}

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<Vec<String>> {
+        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<Value> = 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<String> {
+        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<HeaderMap> {
     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<Vec<GqlResponse>> {
+    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<Option<Value>> {
     let stdout = String::from_utf8_lossy(stdout);
     let stdout = stdout.trim();