rsynko_reqwest/
environment.rs

1use crate::{RuntimeObservation, RuntimeObservationSender, RuntimePause, RuntimeXError};
2use core::convert::Infallible;
3use reqwest::blocking::{Client, RequestBuilder, Response};
4use rsynko_download::*;
5use rsynko_media::*;
6use rsynko_memory::{
7    Artifact, DownloadEvent, DownloadProgress, DownloadSyntax, Extraction, ExtractorKey, Format, FormatPredicate,
8    FormatSelection, InfoRecord, InfoValue, Media, MediaSyntax, YoutubeRequest, interpret_selection, predicate_accepts,
9};
10use rsynko_x::status_id;
11use rsynko_yt::{YoutubeError, YoutubeExtractionExt, YoutubeRequestBytesAlg, youtube_id};
12use std::cell::RefCell;
13use std::ffi::OsString;
14use std::fs::{self, File};
15use std::io::{self, Cursor, Read, Write};
16use std::path::{Path, PathBuf};
17use thiserror::Error;
18
19/// Denotes the exact media bytes of `fixture://single-video`.
20pub const FIXTURE_BYTES: &[u8] = b"rsynko fixture: single-video\n";
21
22const FIXTURE_MEDIA_URL: &str = "fixture-media://single-video/best.mp4";
23
24/// Names the handheld client, which states one muxed representation served without throttling.
25pub const ANDROID_CLIENT: &str = "ANDROID";
26
27/// Names the headset client, which states every adaptive representation.
28pub const VISIONOS_CLIENT: &str = "VISIONOS";
29
30/// States the client the headset catalog is requested as.
31const VISIONOS_USER_AGENT: &str = "Mozilla/5.0 (Macintosh; Intel Mac OS X 15_7_3) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/26.0 Safari/605.1.15";
32
33/// States the client the handheld catalog is requested as.
34const ANDROID_USER_AGENT: &str = "com.google.android.youtube/21.26.364 (Linux; U; Android 11) gzip";
35
36/// Denotes resource retrieval failure in the Reqwest interpreter.
37#[derive(Debug, Error)]
38pub enum RuntimeFetchError {
39    /// Denotes an unknown fixture resource.
40    #[error("unknown fixture resource {0}")]
41    UnknownFixture(String),
42    /// Denotes an HTTP request failure.
43    #[error(transparent)]
44    Http(#[from] reqwest::Error),
45    /// Denotes failure while incrementally reading a resource response.
46    #[error(transparent)]
47    Io(#[from] io::Error),
48}
49
50/// Denotes extraction failure in the assembled runtime.
51#[derive(Debug, Error)]
52pub enum RuntimeExtractionError {
53    /// Denotes Youtube specification failure.
54    #[error("{0}")]
55    Youtube(YoutubeError<RuntimeFetchError, serde_json::Error, Infallible>),
56    /// Denotes failure to read what one tweet carries.
57    #[error("{0}")]
58    X(RuntimeXError),
59    /// Denotes application of an unknown extractor key.
60    #[error("unknown extractor {0:?}")]
61    UnknownExtractor(ExtractorKey),
62}
63
64/// Interprets exact requests, fixture bytes, atomic files, and runtime observations.
65#[derive(Debug)]
66pub struct RuntimeEnvironment {
67    client: Client,
68    events: RefCell<Vec<DownloadEvent>>,
69    progress: RefCell<Vec<DownloadProgress>>,
70    extractors: [ExtractorKey; 3],
71    observer: Option<RuntimeObservationSender>,
72    pause: Option<RuntimePause>,
73}
74
75/// Names this retrieval the way a browser names itself.
76///
77/// What answers these addresses answers pages, and a request stating no agent at all is refused
78/// rather than answered.
79const BROWSER_AGENT: &str = "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/139 Safari/537.36";
80
81impl RuntimeEnvironment {
82    /// Constructs the production environment and its HTTP client.
83    ///
84    /// # Errors
85    ///
86    /// Returns failure to construct the Rustls-backed client.
87    pub fn build() -> Result<Self, reqwest::Error> {
88        // Rustls states the protocol and leaves the cryptography to a provider, which is a choice
89        // this interpreter makes once, here, rather than one the library makes for it.
90        let _installed = rustls::crypto::ring::default_provider().install_default();
91        Ok(Self {
92            client: Client::builder().build()?,
93            events: RefCell::default(),
94            progress: RefCell::default(),
95            extractors: [ExtractorKey::new("fixture"), ExtractorKey::new("youtube"), ExtractorKey::new("x")],
96            observer: None,
97            pause: None,
98        })
99    }
100
101    /// Constructs the production environment with a live observation channel.
102    ///
103    /// # Errors
104    ///
105    /// Returns failure to construct the Rustls-backed client.
106    pub fn build_observed(observer: RuntimeObservationSender) -> Result<Self, reqwest::Error> {
107        let mut environment = Self::build()?;
108        environment.observer = Some(observer);
109        Ok(environment)
110    }
111
112    /// Constructs the production environment with observation and cooperative pause control.
113    ///
114    /// # Errors
115    ///
116    /// Returns failure to construct the Rustls-backed client.
117    pub fn build_pausable(observer: RuntimeObservationSender, pause: RuntimePause) -> Result<Self, reqwest::Error> {
118        let mut environment = Self::build_observed(observer)?;
119        environment.pause = Some(pause);
120        Ok(environment)
121    }
122
123    /// Observes terminal download events in emission order.
124    #[must_use]
125    pub fn events(&self) -> Vec<DownloadEvent> {
126        self.events.borrow().clone()
127    }
128
129    /// Observes byte-progress reports in emission order.
130    #[must_use]
131    pub fn progress(&self) -> Vec<DownloadProgress> {
132        self.progress.borrow().clone()
133    }
134
135    fn request(&self, request: &YoutubeRequest) -> RequestBuilder {
136        match request {
137            // A watch page and the player program it names are both ordinary browser documents.
138            YoutubeRequest::Watch(url) | YoutubeRequest::PlayerProgram(url) => {
139                self.client.get(url).header("user-agent", BROWSER_AGENT)
140            }
141            // The headset client states the full adaptive catalog without attestation and without
142            // a player token, which is what an unauthenticated retrieval can ask for today. It
143            // still states the program it runs, so a response that does guard its representations
144            // is answered rather than refused.
145            YoutubeRequest::Player { id, api_key, claim } => {
146                let (context, agent, name, version) = if claim.client == ANDROID_CLIENT {
147                    (
148                        serde_json::json!({
149                            "clientName": ANDROID_CLIENT,
150                            "clientVersion": "21.26.364",
151                            "androidSdkVersion": 30,
152                            "userAgent": ANDROID_USER_AGENT,
153                            "osName": "Android",
154                            "osVersion": "11",
155                            "visitorData": claim.visitor,
156                            "hl": "en",
157                            "gl": "US",
158                            "timeZone": "UTC",
159                            "utcOffsetMinutes": 0
160                        }),
161                        ANDROID_USER_AGENT,
162                        "3",
163                        "21.26.364",
164                    )
165                } else {
166                    (
167                        serde_json::json!({
168                            "clientName": VISIONOS_CLIENT,
169                            "clientVersion": "1.02",
170                            "deviceMake": "Apple",
171                            "deviceModel": "RealityDevice17,1",
172                            "userAgent": VISIONOS_USER_AGENT,
173                            "osName": "visionOS",
174                            "osVersion": "26.5.23O471",
175                            "visitorData": claim.visitor,
176                            "hl": "en",
177                            "gl": "US",
178                            "timeZone": "UTC",
179                            "utcOffsetMinutes": 0
180                        }),
181                        VISIONOS_USER_AGENT,
182                        "101",
183                        "1.02",
184                    )
185                };
186                let mut request = self
187                    .client
188                    .post(format!("https://www.youtube.com/youtubei/v1/player?key={api_key}&prettyPrint=false"))
189                    .header("content-type", "application/json")
190                    .header("origin", "https://www.youtube.com")
191                    .header("x-youtube-client-name", name)
192                    .header("x-youtube-client-version", version)
193                    .header("user-agent", agent);
194                // The session the watch page issued is what an unauthenticated catalog request is
195                // recognized by; without it the request reads as coming from nobody at all.
196                if let Some(visitor) = &claim.visitor {
197                    request = request.header("x-goog-visitor-id", visitor);
198                }
199                request.json(&serde_json::json!({
200                    "context": { "client": context },
201                    "playbackContext": { "contentPlaybackContext": {
202                        "html5Preference": "HTML5_PREF_WANTS",
203                        "signatureTimestamp": claim.timestamp
204                    }},
205                    "videoId": id,
206                    "contentCheckOk": true,
207                    "racyCheckOk": true
208                }))
209            }
210            YoutubeRequest::Media(url) => self
211                .client
212                .get(url)
213                .header("referer", "https://www.youtube.com/")
214                .header("user-agent", "com.google.android.youtube/21.26.364 (Linux; U; Android 11) gzip"),
215        }
216    }
217
218    /// Asks one address and reads the whole of what it answers.
219    ///
220    /// The agent is stated because a request without one is refused rather than answered: an
221    /// address that answers a page states what it would state to a browser, or nothing.
222    pub(crate) fn fetch_bytes(&self, url: &str) -> Result<Vec<u8>, reqwest::Error> {
223        Ok(self.client.get(url).header("user-agent", BROWSER_AGENT).send()?.error_for_status()?.bytes()?.to_vec())
224    }
225
226    fn execute(&self, request: &YoutubeRequest) -> Result<Response, RuntimeFetchError> {
227        Ok(self.request(request).send()?.error_for_status()?)
228    }
229
230    fn read_stream(&self, stream: &mut RuntimeFetchStream, buffer: &mut [u8]) -> Result<usize, RuntimeFetchError> {
231        if let Some(pause) = &self.pause
232            && !pause.wait_until_running()
233        {
234            return Err(io::Error::new(io::ErrorKind::Interrupted, "download cancelled").into());
235        }
236        Ok(stream.0.read(buffer)?)
237    }
238}
239
240/// Carries an incrementally readable fixture or HTTP resource.
241pub struct RuntimeFetchStream(Box<dyn Read + Send>);
242
243/// Carries a filesystem publication that is not yet visible at its final path.
244pub struct RuntimePublication {
245    file: File,
246    partial: PathBuf,
247    destination: PathBuf,
248}
249
250impl MediaSorts for RuntimeEnvironment {
251    type Value = InfoValue;
252    type Metadata = InfoRecord;
253    type Format = Format;
254    type Artifact = Artifact;
255    type Media = Media;
256    type Extraction = Extraction;
257    type Extractor = ExtractorKey;
258    type Predicate = FormatPredicate;
259    type Selection = FormatSelection;
260    type Output = PathBuf;
261}
262
263impl FormatViewAlg for RuntimeEnvironment {
264    fn format_text<'a>(&self, format: &'a Format, key: &str) -> Option<&'a str> {
265        format.text(key)
266    }
267}
268
269impl YoutubeRequestBytesAlg for RuntimeEnvironment {
270    type Error = RuntimeFetchError;
271
272    fn youtube_request_bytes(&self, request: &YoutubeRequest) -> Result<Vec<u8>, Self::Error> {
273        Ok(self.execute(request)?.bytes()?.to_vec())
274    }
275}
276
277impl FetchStreamAlg for RuntimeEnvironment {
278    type Error = RuntimeFetchError;
279    type Stream = RuntimeFetchStream;
280
281    fn open_fetch(&self, url: &str) -> Result<FetchStream<Self::Stream>, Self::Error> {
282        if url.starts_with("fixture-media://") {
283            if url != FIXTURE_MEDIA_URL {
284                return Err(RuntimeFetchError::UnknownFixture(url.to_owned()));
285            }
286            return Ok(FetchStream::new(
287                RuntimeFetchStream(Box::new(Cursor::new(FIXTURE_BYTES))),
288                u64::try_from(FIXTURE_BYTES.len()).ok(),
289            ));
290        }
291        let response = self.client.get(url).send()?.error_for_status()?;
292        let total = response.content_length();
293        Ok(FetchStream::new(RuntimeFetchStream(Box::new(response)), total))
294    }
295
296    fn read_fetch(&self, stream: &mut Self::Stream, buffer: &mut [u8]) -> Result<usize, Self::Error> {
297        self.read_stream(stream, buffer)
298    }
299}
300
301impl FetchStreamAlg<String> for RuntimeEnvironment {
302    type Error = RuntimeFetchError;
303    type Stream = RuntimeFetchStream;
304
305    fn open_fetch(&self, url: &String) -> Result<FetchStream<Self::Stream>, Self::Error> {
306        <Self as FetchStreamAlg<str>>::open_fetch(self, url.as_str())
307    }
308
309    fn read_fetch(&self, stream: &mut Self::Stream, buffer: &mut [u8]) -> Result<usize, Self::Error> {
310        self.read_stream(stream, buffer)
311    }
312}
313
314impl FormatSourceAlg<String> for RuntimeEnvironment {
315    fn format_source(&self, format: &Format) -> Option<String> {
316        format.source().map(str::to_owned)
317    }
318}
319
320impl FetchStreamAlg<YoutubeRequest> for RuntimeEnvironment {
321    type Error = RuntimeFetchError;
322    type Stream = RuntimeFetchStream;
323
324    fn open_fetch(&self, request: &YoutubeRequest) -> Result<FetchStream<Self::Stream>, Self::Error> {
325        let response = self.execute(request)?;
326        let total = response.content_length();
327        Ok(FetchStream::new(RuntimeFetchStream(Box::new(response)), total))
328    }
329
330    fn read_fetch(&self, stream: &mut Self::Stream, buffer: &mut [u8]) -> Result<usize, Self::Error> {
331        self.read_stream(stream, buffer)
332    }
333}
334
335impl AtomicPublishAlg for RuntimeEnvironment {
336    type Error = io::Error;
337    type Publication = RuntimePublication;
338
339    fn begin_publication(&self, destination: &Path) -> Result<Self::Publication, Self::Error> {
340        if let Some(parent) = destination.parent().filter(|path| !path.as_os_str().is_empty()) {
341            fs::create_dir_all(parent)?;
342        }
343        let partial = partial_path(destination);
344        if partial.exists() {
345            fs::remove_file(&partial)?;
346        }
347        let file = File::create(&partial)?;
348        Ok(RuntimePublication { file, partial, destination: destination.to_owned() })
349    }
350
351    fn write_publication(&self, publication: &mut Self::Publication, bytes: &[u8]) -> Result<(), Self::Error> {
352        publication.file.write_all(bytes)
353    }
354
355    fn commit_publication(&self, publication: Self::Publication) -> Result<(), Self::Error> {
356        publication.file.sync_all()?;
357        let RuntimePublication { file, partial, destination } = publication;
358        drop(file);
359        let result = fs::rename(&partial, destination);
360        if result.is_err() {
361            let _ignored = fs::remove_file(partial);
362        }
363        result
364    }
365
366    fn abort_publication(&self, publication: Self::Publication) {
367        let RuntimePublication { file, partial, .. } = publication;
368        drop(file);
369        let _ignored = fs::remove_file(partial);
370    }
371}
372
373impl DownloadReportAlg for RuntimeEnvironment {
374    type Event = DownloadEvent;
375
376    fn report_download(&self, event: DownloadEvent) {
377        self.events.borrow_mut().push(event.clone());
378        if let Some(observer) = &self.observer {
379            observer.send(RuntimeObservation::terminal(event)).expect("a running download is being read");
380        }
381    }
382}
383
384impl DownloadProgressAlg for RuntimeEnvironment {
385    type Progress = DownloadProgress;
386
387    fn report_progress(&self, progress: DownloadProgress) {
388        self.progress.borrow_mut().push(progress.clone());
389        if let Some(observer) = &self.observer {
390            observer.send(RuntimeObservation::progress(progress)).expect("a running download is being read");
391        }
392    }
393}
394
395impl DownloadObservationAlg for RuntimeEnvironment {
396    type Event = DownloadEvent;
397    type Progress = DownloadProgress;
398
399    fn download_progress(&self, destination: &Path, downloaded: u64, total: Option<u64>) -> Self::Progress {
400        DownloadSyntax.download_progress(destination, downloaded, total)
401    }
402
403    fn download_succeeded(&self, destination: &Path, bytes: u64) -> Self::Event {
404        DownloadSyntax.download_succeeded(destination, bytes)
405    }
406
407    fn download_failed(&self, destination: &Path, message: String) -> Self::Event {
408        DownloadSyntax.download_failed(destination, message)
409    }
410}
411
412impl ExtractionCatalogAlg for RuntimeEnvironment {
413    fn extractor_keys(&self) -> impl Iterator<Item = &ExtractorKey> {
414        self.extractors.iter()
415    }
416
417    fn extractor_accepts(&self, extractor: &ExtractorKey, url: &str) -> bool {
418        match extractor.0.as_str() {
419            "fixture" => url == "fixture://single-video",
420            "youtube" => youtube_id(url).is_some(),
421            "x" => status_id(url).is_some(),
422            _ => false,
423        }
424    }
425}
426
427impl ExtractionApplyAlg for RuntimeEnvironment {
428    type Error = RuntimeExtractionError;
429
430    fn extract_with(&self, extractor: &ExtractorKey, url: &str) -> Result<Extraction, Self::Error> {
431        match extractor.0.as_str() {
432            "fixture" => Ok(fixture_extraction()),
433            "youtube" => self.extract_youtube(url).map_err(RuntimeExtractionError::Youtube),
434            "x" => self.extract_x(url).map_err(RuntimeExtractionError::X),
435            _ => Err(RuntimeExtractionError::UnknownExtractor(extractor.clone())),
436        }
437    }
438}
439
440impl FormatPredicateMatchAlg for RuntimeEnvironment {
441    fn format_matches(&self, predicate: &FormatPredicate, format: &Format) -> bool {
442        predicate_accepts(predicate, format)
443    }
444}
445
446impl FormatSelectionApplyAlg for RuntimeEnvironment {
447    fn select_formats<'a>(&self, formats: &'a [Format], selection: &FormatSelection) -> Option<Vec<&'a Format>> {
448        interpret_selection(self, formats, selection)
449    }
450}
451
452fn fixture_observations() -> InfoRecord {
453    MediaSyntax.metadata([
454        (FORMAT_SOURCE.to_owned(), MediaSyntax.string_metadata(FIXTURE_MEDIA_URL)),
455        (FORMAT_EXTENSION.to_owned(), MediaSyntax.string_metadata("mp4")),
456        (FORMAT_HAS_AUDIO.to_owned(), MediaSyntax.boolean_metadata(true)),
457        (FORMAT_HAS_VIDEO.to_owned(), MediaSyntax.boolean_metadata(true)),
458    ])
459}
460
461fn fixture_extraction() -> Extraction {
462    Extraction::Media(Media::new(
463        "single-video".to_owned(),
464        InfoRecord::default(),
465        vec![Format::new("fixture-best".to_owned(), fixture_observations())],
466    ))
467}
468
469fn partial_path(destination: &Path) -> PathBuf {
470    let mut name: OsString = destination.as_os_str().to_owned();
471    name.push(".part");
472    PathBuf::from(name)
473}
rsynko_reqwest/environment.rs

rsynko_reqwest/
environment.rs
