1pub mod api {
2 use serde_json::Value;
3 use std::sync::Arc;
4 use tokio::sync::RwLock;
5 use tracing::{debug, warn};
6 const TWITCH_GQL: &str = "https://gql.twitch.tv/gql";
7 const TWITCH_URL: &str = "https://www.twitch.tv";
8 const BROWSER_UA: &str =
9 "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0";
10 const METADATA_PAYLOAD: &str = r#"{"operationName":"StreamMetadata","query":"query StreamMetadata($channelLogin: String!) { user(login: $channelLogin) { stream { type } lastBroadcast { title } } }","variables":{"channelLogin":"%s"}}"#;
11 const ACCESS_TOKEN_PAYLOAD: &str = r#"{"operationName":"PlaybackAccessToken_Template","query":"query PlaybackAccessToken_Template($login: String!,$isLive:Boolean!,$vodID:ID!,$isVod:Boolean!,$playerType:String!){streamPlaybackAccessToken(channelName:$login,params:{platform:\"web\",playerBackend:\"mediaplayer\",playerType:$playerType})@include(if:$isLive){value signature __typename}videoPlaybackAccessToken(id:$vodID,params:{platform:\"web\",playerBackend:\"mediaplayer\",playerType:$playerType})@include(if:$isVod){value signature __typename}}","variables":{"isLive":true,"login":"%s","isVod":false,"vodID":"","playerType":"site"}}"#;
12 pub struct TwitchGqlClient {
13 http: Arc<reqwest::Client>,
14 client_id: RwLock<Option<String>>,
15 device_id: RwLock<Option<String>>,
16 }
17 impl TwitchGqlClient {
18 pub fn new(http: Arc<reqwest::Client>, pinned_client_id: Option<String>) -> Self {
19 Self {
20 http,
21 client_id: RwLock::new(pinned_client_id),
22 device_id: RwLock::new(None),
23 }
24 }
25 pub fn is_initialized(&self) -> bool {
26 self.client_id
27 .try_read()
28 .map(|g| g.is_some())
29 .unwrap_or(false)
30 }
31 pub async fn init_request_headers(&self) {
32 if self.client_id.read().await.is_some() {
33 return;
34 }
35 let resp = match self
36 .http
37 .get(TWITCH_URL)
38 .header("Accept", "text/html")
39 .header("User-Agent", BROWSER_UA)
40 .send()
41 .await
42 {
43 Ok(r) => r,
44 Err(e) => {
45 warn!("Twitch: failed to fetch main page: {e}");
46 return;
47 }
48 };
49 let cookie_headers: Vec<String> = resp
50 .headers()
51 .get_all("set-cookie")
52 .iter()
53 .filter_map(|v| v.to_str().ok().map(str::to_owned))
54 .collect();
55 for cookie in &cookie_headers {
56 if cookie.contains("unique_id=")
57 && let Some(id) = extract_between(cookie, "unique_id=", ";")
58 {
59 *self.device_id.write().await = Some(id.trim().to_string());
60 break;
61 }
62 }
63 let body = match resp.text().await {
64 Ok(b) => b,
65 Err(e) => {
66 warn!("Twitch: failed to read main page body: {e}");
67 return;
68 }
69 };
70 if let Some(id) = extract_between(&body, "clientId=\"", "\"") {
71 debug!("Twitch: initialized client_id from main page");
72 *self.client_id.write().await = Some(id.to_string());
73 }
74 }
75 async fn post_raw(&self, body: String) -> Option<Value> {
76 let client_id = self.client_id.read().await.clone()?;
77 let device_id = self.device_id.read().await.clone();
78 let mut req = self
79 .http
80 .post(TWITCH_GQL)
81 .header("Client-ID", client_id)
82 .header("Content-Type", "text/plain;charset=UTF-8")
83 .body(body);
84 if let Some(did) = device_id {
85 req = req.header("X-Device-ID", did);
86 }
87 let resp = match req.send().await {
88 Ok(r) => r,
89 Err(e) => {
90 debug!("Twitch GQL send error: {e}");
91 return None;
92 }
93 };
94 if !resp.status().is_success() {
95 warn!("Twitch GQL HTTP {}", resp.status());
96 return None;
97 }
98 match resp.json::<Value>().await {
99 Ok(v) => Some(v),
100 Err(e) => {
101 debug!("Twitch GQL JSON parse error: {e}");
102 None
103 }
104 }
105 }
106 pub async fn fetch_stream_channel_info(&self, channel: &str) -> Option<Value> {
107 let payload = METADATA_PAYLOAD.replace("%s", channel);
108 let body = match self.post_raw(payload).await {
109 Some(b) => b,
110 None => {
111 debug!("Twitch: stream metadata request failed for '{channel}'");
112 return None;
113 }
114 };
115 if let Some(errors) = body["errors"].as_array() {
116 for e in errors {
117 debug!(
118 "Twitch GQL error: {}",
119 e["message"].as_str().unwrap_or("unknown")
120 );
121 }
122 return None;
123 }
124 Some(body)
125 }
126 pub async fn fetch_access_token(&self, channel: &str) -> Option<(String, String)> {
127 let payload = ACCESS_TOKEN_PAYLOAD.replace("%s", channel);
128 let body = match self.post_raw(payload).await {
129 Some(b) => b,
130 None => {
131 debug!("Twitch: access token request failed for '{channel}'");
132 return None;
133 }
134 };
135 if let Some(errors) = body["errors"].as_array() {
136 for e in errors {
137 debug!(
138 "Twitch access token GQL error: {}",
139 e["message"].as_str().unwrap_or("unknown")
140 );
141 }
142 return None;
143 }
144 let token = &body["data"]["streamPlaybackAccessToken"];
145 let value = match token["value"].as_str() {
146 Some(v) => v.to_string(),
147 None => {
148 debug!("Twitch: access token 'value' missing for '{channel}'");
149 return None;
150 }
151 };
152 let sig = match token["signature"].as_str() {
153 Some(s) => s.to_string(),
154 None => {
155 debug!("Twitch: access token 'signature' missing for '{channel}'");
156 return None;
157 }
158 };
159 Some((value, sig))
160 }
161 pub async fn fetch_text(&self, url: &str) -> Option<String> {
162 let client_id = self.client_id.read().await.clone();
163 let device_id = self.device_id.read().await.clone();
164 let mut req = self.http.get(url);
165 if let Some(id) = client_id {
166 req = req.header("Client-ID", id);
167 }
168 if let Some(did) = device_id {
169 req = req.header("X-Device-ID", did);
170 }
171 match req.send().await {
172 Ok(resp) => match resp.text().await {
173 Ok(t) => Some(t),
174 Err(e) => {
175 debug!("Twitch: failed to read response body from '{url}': {e}");
176 None
177 }
178 },
179 Err(e) => {
180 debug!("Twitch: HTTP GET failed for '{url}': {e}");
181 None
182 }
183 }
184 }
185 }
186 fn extract_between<'a>(src: &'a str, prefix: &str, suffix: &str) -> Option<&'a str> {
187 let start = src.find(prefix)? + prefix.len();
188 let end = src[start..].find(suffix)? + start;
189 Some(&src[start..end])
190 }
191}
192pub mod manager {
193 use super::{api::TwitchGqlClient, track::TwitchTrack};
194 use crate::{
195 config::TwitchConfig,
196 protocol::tracks::{LoadError, LoadResult, Track, TrackInfo},
197 sources::{SourcePlugin, playable_track::BoxedTrack},
198 };
199 use async_trait::async_trait;
200 use regex::Regex;
201 use std::sync::Arc;
202 use tracing::{debug, warn};
203 const STREAM_NAME_REGEX: &str = r"(?i)^https?://(?:www\.|go\.|m\.)?twitch\.tv/([^/]+)$";
204 const TWITCH_DOMAIN_REGEX: &str = r"(?i)^https?://(?:www\.|go\.|m\.)?twitch\.tv/";
205 const TWITCH_IMAGE_PREVIEW_URL: &str =
206 "https://static-cdn.jtvnw.net/previews-ttv/live_user_%s-440x248.jpg";
207 pub struct TwitchSource {
208 gql: Arc<TwitchGqlClient>,
209 proxy: Option<crate::config::HttpProxyConfig>,
210 stream_name_regex: Regex,
211 twitch_domain_regex: Regex,
212 }
213 impl TwitchSource {
214 pub fn new(config: TwitchConfig, client: Arc<reqwest::Client>) -> Self {
215 Self {
216 gql: Arc::new(TwitchGqlClient::new(client, config.client_id)),
217 proxy: config.proxy,
218 stream_name_regex: Regex::new(STREAM_NAME_REGEX).unwrap(),
219 twitch_domain_regex: Regex::new(TWITCH_DOMAIN_REGEX).unwrap(),
220 }
221 }
222 fn get_channel_identifier_from_url(&self, url: &str) -> Option<String> {
223 self.stream_name_regex
224 .captures(url)
225 .and_then(|c| c.get(1))
226 .map(|m| m.as_str().to_lowercase())
227 }
228 async fn ensure_initialized(&self) {
229 if !self.gql.is_initialized() {
230 self.gql.init_request_headers().await;
231 }
232 }
233 async fn get_channel_streams_url(&self, channel: &str) -> Option<String> {
234 let (token, sig) = self.gql.fetch_access_token(channel).await?;
235 Some(format!(
236 "https://usher.ttvnw.net/api/channel/hls/{}.m3u8?token={}&sig={}&allow_source=true&allow_spectre=true&allow_audio_only=true&player_backend=html5&expgroup=regular",
237 channel,
238 urlencoding::encode(&token),
239 urlencoding::encode(&sig),
240 ))
241 }
242 async fn fetch_segment_playlist_url(&self, channel: &str) -> Option<String> {
243 let streams_url = self.get_channel_streams_url(channel).await?;
244 let m3u8 = self.gql.fetch_text(&streams_url).await?;
245 let streams = load_channel_streams_list(&m3u8);
246 if streams.is_empty() {
247 debug!("Twitch: no streams available on channel '{channel}'");
248 return None;
249 }
250 let chosen = streams.last().unwrap();
251 debug!(
252 "Twitch: chose stream with quality {} from url {}",
253 chosen.quality, chosen.url
254 );
255 Some(chosen.url.clone())
256 }
257 }
258 struct ChannelStreamInfo {
259 quality: String,
260 url: String,
261 }
262 fn load_channel_streams_list(m3u8: &str) -> Vec<ChannelStreamInfo> {
263 let lines: Vec<&str> = m3u8.lines().collect();
264 let mut streams = Vec::new();
265 let mut i = 0;
266 while i < lines.len() {
267 let line = lines[i].trim();
268 if line.starts_with("#EXT-X-STREAM-INF:") {
269 let quality = line
270 .split(',')
271 .find_map(|part| {
272 part.trim()
273 .strip_prefix("VIDEO=\"")
274 .and_then(|v| v.strip_suffix('"'))
275 })
276 .unwrap_or("unknown")
277 .to_string();
278 if let Some(url_line) = lines.get(i + 1) {
279 let url = url_line.trim();
280 if !url.is_empty() && !url.starts_with('#') {
281 streams.push(ChannelStreamInfo {
282 quality,
283 url: url.to_string(),
284 });
285 }
286 }
287 }
288 i += 1;
289 }
290 streams
291 }
292 #[async_trait]
293 impl SourcePlugin for TwitchSource {
294 fn name(&self) -> &str {
295 "twitch"
296 }
297 fn can_handle(&self, identifier: &str) -> bool {
298 self.twitch_domain_regex.is_match(identifier)
299 }
300 async fn load(
301 &self,
302 identifier: &str,
303 _routeplanner: Option<Arc<dyn crate::routeplanner::RoutePlanner>>,
304 ) -> LoadResult {
305 let stream_name = match self.get_channel_identifier_from_url(identifier) {
306 Some(n) => n,
307 None => return LoadResult::Empty {},
308 };
309 self.ensure_initialized().await;
310 let channel_info_body = match self.gql.fetch_stream_channel_info(&stream_name).await {
311 Some(b) => b,
312 None => {
313 return LoadResult::Error(LoadError {
314 message: Some(format!(
315 "Loading Twitch channel information failed for '{stream_name}'"
316 )),
317 severity: crate::common::Severity::Suspicious,
318 cause: "GQL request failed".to_string(),
319 cause_stack_trace: None,
320 });
321 }
322 };
323 let channel_info = &channel_info_body["data"]["user"];
324 if channel_info.is_null() || channel_info["stream"]["type"].is_null() {
325 return LoadResult::Empty {};
326 }
327 let title = channel_info["lastBroadcast"]["title"]
328 .as_str()
329 .unwrap_or("")
330 .to_string();
331 let thumbnail = TWITCH_IMAGE_PREVIEW_URL.replace("%s", &stream_name);
332 LoadResult::Track(Track::new(TrackInfo {
333 identifier: stream_name.clone(),
334 is_seekable: false,
335 author: stream_name.clone(),
336 length: 0,
337 is_stream: true,
338 position: 0,
339 title,
340 uri: Some(identifier.to_string()),
341 artwork_url: Some(thumbnail),
342 isrc: None,
343 source_name: "twitch".to_string(),
344 }))
345 }
346 async fn get_track(
347 &self,
348 identifier: &str,
349 routeplanner: Option<Arc<dyn crate::routeplanner::RoutePlanner>>,
350 ) -> Option<BoxedTrack> {
351 let stream_name = self.get_channel_identifier_from_url(identifier)?;
352 self.ensure_initialized().await;
353 let local_addr = routeplanner.and_then(|rp| rp.get_address());
354 let stream_url = match self.fetch_segment_playlist_url(&stream_name).await {
355 Some(u) => u,
356 None => {
357 warn!("Twitch: failed to resolve stream for '{stream_name}'");
358 return None;
359 }
360 };
361 Some(Arc::new(TwitchTrack {
362 stream_url,
363 local_addr,
364 proxy: self.proxy.clone(),
365 }))
366 }
367 fn get_proxy_config(&self) -> Option<crate::config::HttpProxyConfig> {
368 self.proxy.clone()
369 }
370 }
371}
372pub mod track {
373 use crate::{
374 common::types::AudioFormat,
375 config::HttpProxyConfig,
376 sources::{
377 playable_track::{PlayableTrack, ResolvedTrack},
378 youtube::hls::{
379 fetcher::fetch_segment_into, resolver::fetch_text, ts_demux::extract_adts_from_ts,
380 types::Resource, utils::resolve_url,
381 },
382 },
383 };
384 use async_trait::async_trait;
385 use std::{
386 collections::HashSet,
387 io::{self, Read, Seek, SeekFrom},
388 net::IpAddr,
389 };
390 use symphonia::core::io::MediaSource;
391 pub struct TwitchTrack {
392 pub stream_url: String,
393 pub local_addr: Option<IpAddr>,
394 pub proxy: Option<HttpProxyConfig>,
395 }
396 #[async_trait]
397 impl PlayableTrack for TwitchTrack {
398 async fn resolve(&self) -> Result<ResolvedTrack, String> {
399 let handle = tokio::runtime::Handle::current();
400 let (err_tx, _err_rx) = flume::bounded::<String>(1);
401 let reader = Box::new(
402 LiveHlsReader::new(
403 self.stream_url.clone(),
404 self.local_addr,
405 self.proxy.clone(),
406 handle,
407 err_tx,
408 )
409 .await,
410 ) as Box<dyn MediaSource>;
411 Ok(ResolvedTrack::new(reader, Some(AudioFormat::Aac)))
412 }
413 }
414 struct LiveHlsReader {
415 chunk_rx: flume::Receiver<Vec<u8>>,
416 current: Vec<u8>,
417 pos: usize,
418 }
419 impl LiveHlsReader {
420 pub async fn new(
421 manifest_url: String,
422 local_addr: Option<IpAddr>,
423 proxy: Option<HttpProxyConfig>,
424 _handle: tokio::runtime::Handle,
425 err_tx: flume::Sender<String>,
426 ) -> Self {
427 let (chunk_tx, chunk_rx) = flume::bounded::<Vec<u8>>(16);
428 tokio::spawn(async move {
429 let mut builder =
430 reqwest::Client::builder().timeout(std::time::Duration::from_secs(15));
431 if let Some(ip) = local_addr {
432 builder = builder.local_address(ip);
433 }
434 if let Some(ref cfg) = proxy
435 && let Some(ref url) = cfg.url
436 {
437 match reqwest::Proxy::all(url) {
438 Ok(mut p) => {
439 if let (Some(u), Some(pw)) = (&cfg.username, &cfg.password) {
440 p = p.basic_auth(u, pw);
441 }
442 builder = builder.proxy(p);
443 }
444 Err(e) => {
445 tracing::error!("Twitch live HLS: proxy setup failed for {url}: {e}");
446 let _ = err_tx.send(format!("Proxy setup failed: {e}"));
447 return;
448 }
449 }
450 }
451 let client = match builder.build() {
452 Ok(c) => c,
453 Err(e) => {
454 tracing::error!("Twitch live HLS: client build failed: {e}");
455 let _ = err_tx.send(format!("Client build failed: {e}"));
456 return;
457 }
458 };
459 let mut seen: HashSet<String> = HashSet::new();
460 let mut seen_history: std::collections::VecDeque<String> =
461 std::collections::VecDeque::with_capacity(50);
462 loop {
463 let text = match fetch_text(&client, &manifest_url).await {
464 Ok(t) => t,
465 Err(e) => {
466 tracing::warn!("Twitch: live playlist refresh failed: {e}");
467 tokio::time::sleep(std::time::Duration::from_secs(2)).await;
468 continue;
469 }
470 };
471 let (segments, target_duration) = parse_live_playlist(&text, &manifest_url);
472 for seg in segments {
473 if seen.contains(&seg.url) {
474 continue;
475 }
476 let mut raw = Vec::new();
477 if let Err(e) = fetch_segment_into(&client, &seg, &mut raw).await {
478 tracing::warn!("Twitch: segment fetch error: {e}");
479 continue;
480 }
481 let payload = if raw.first() == Some(&0x47) {
482 let adts = extract_adts_from_ts(&raw);
483 if adts.is_empty() {
484 tracing::debug!("Twitch: ADTS extraction failed, skipping segment");
485 continue;
486 }
487 adts
488 } else {
489 raw
490 };
491 if chunk_tx.send(payload).is_err() {
492 return;
493 }
494 if seen.insert(seg.url.clone()) {
495 seen_history.push_back(seg.url);
496 if seen_history.len() > 50
497 && let Some(old) = seen_history.pop_front()
498 {
499 seen.remove(&old);
500 }
501 }
502 }
503 let wait = (target_duration / 2.0).max(1.0);
504 tokio::time::sleep(std::time::Duration::from_secs_f64(wait)).await;
505 }
506 });
507 Self {
508 chunk_rx,
509 current: Vec::new(),
510 pos: 0,
511 }
512 }
513 }
514 fn parse_live_playlist(text: &str, base_url: &str) -> (Vec<Resource>, f64) {
515 let mut segments = Vec::new();
516 let mut target_duration = 6.0f64;
517 let lines: Vec<&str> = text.lines().map(str::trim).collect();
518 let mut i = 0;
519 while i < lines.len() {
520 let line = lines[i];
521 if let Some(rest) = line.strip_prefix("#EXT-X-TARGETDURATION:") {
522 if let Ok(d) = rest.trim().parse::<f64>() {
523 target_duration = d;
524 }
525 } else if line.starts_with("#EXTINF:") {
526 let duration = line
527 .strip_prefix("#EXTINF:")
528 .and_then(|r| r.split(',').next())
529 .and_then(|d| d.trim().parse::<f64>().ok());
530 let mut j = i + 1;
531 while j < lines.len() && lines[j].starts_with('#') {
532 j += 1;
533 }
534 if j < lines.len() && !lines[j].is_empty() {
535 let url = resolve_url(base_url, lines[j]);
536 segments.push(Resource {
537 url,
538 range: None,
539 duration,
540 });
541 }
542 i = j + 1;
543 continue;
544 }
545 i += 1;
546 }
547 (segments, target_duration)
548 }
549 impl Read for LiveHlsReader {
550 fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
551 loop {
552 if self.pos < self.current.len() {
553 let n = buf.len().min(self.current.len() - self.pos);
554 buf[..n].copy_from_slice(&self.current[self.pos..self.pos + n]);
555 self.pos += n;
556 return Ok(n);
557 }
558 match self
559 .chunk_rx
560 .recv_timeout(std::time::Duration::from_millis(500))
561 {
562 Ok(chunk) => {
563 self.current = chunk;
564 self.pos = 0;
565 }
566 Err(flume::RecvTimeoutError::Timeout) => continue,
567 Err(flume::RecvTimeoutError::Disconnected) => return Ok(0),
568 }
569 }
570 }
571 }
572 impl Seek for LiveHlsReader {
573 fn seek(&mut self, _: SeekFrom) -> io::Result<u64> {
574 Err(io::Error::new(
575 io::ErrorKind::Unsupported,
576 "live streams are not seekable",
577 ))
578 }
579 }
580 impl MediaSource for LiveHlsReader {
581 fn is_seekable(&self) -> bool {
582 false
583 }
584 fn byte_len(&self) -> Option<u64> {
585 None
586 }
587 }
588}
589pub use manager::TwitchSource;