Sulkta-Coop/rustypipe
Archived
2
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix: retry on empty continuation responses

Browse source
This commit is contained in:
ThetaDev 2022-10-11 18:49:15 +02:00
parent ef35c48890
commit 562ac2df7e
10 changed files with 142 additions and 88 deletions
Download patch file Download diff file Expand all files Collapse all files

22
src/client/channel.rs
View file

@ -281,12 +281,11 @@ impl MapResponse<Paginator<ChannelVideo>> for response::ChannelCont {
_deobf: Option<&crate::deobfuscate::Deobfuscator>,
) -> Result<MapResult<Paginator<ChannelVideo>>, ExtractionError> {
let mut actions = self.on_response_received_actions;
let res = some_or_bail!(
actions.try_swap_remove(0),
Err(ExtractionError::InvalidData("no received action".into()))
)
.append_continuation_items_action
.continuation_items;
let res = actions
.try_swap_remove(0)
.ok_or(ExtractionError::Retry)?
.append_continuation_items_action
.continuation_items;
Ok(map_videos(res, lang))
}
@ -300,12 +299,11 @@ impl MapResponse<Paginator<ChannelPlaylist>> for response::ChannelCont {
_deobf: Option<&crate::deobfuscate::Deobfuscator>,
) -> Result<MapResult<Paginator<ChannelPlaylist>>, ExtractionError> {
let mut actions = self.on_response_received_actions;
let res = some_or_bail!(
actions.try_swap_remove(0),
Err(ExtractionError::InvalidData("no received action".into()))
)
.append_continuation_items_action
.continuation_items;
let res = actions
.try_swap_remove(0)
.ok_or(ExtractionError::Retry)?
.append_continuation_items_action
.continuation_items;
Ok(map_playlists(res))
}

71
src/client/mod.rs
View file

@ -169,7 +169,8 @@ struct RustyPipeRef {
http: Client,
storage: Option<Box<dyn CacheStorage>>,
reporter: Option<Box<dyn Reporter>>,
n_retries: u32,
n_http_retries: u32,
n_query_retries: u32,
consent_cookie: String,
cache: CacheHolder,
default_opts: RustyPipeOpts,
@ -186,7 +187,8 @@ struct RustyPipeOpts {
pub struct RustyPipeBuilder {
storage: Option<Box<dyn CacheStorage>>,
reporter: Option<Box<dyn Reporter>>,
n_retries: u32,
n_http_retries: u32,
n_query_retries: u32,
user_agent: String,
default_opts: RustyPipeOpts,
}
@ -277,7 +279,8 @@ impl RustyPipeBuilder {
default_opts: RustyPipeOpts::default(),
storage: Some(Box::new(FileStorage::default())),
reporter: Some(Box::new(FileReporter::default())),
n_retries: 3,
n_http_retries: 3,
n_query_retries: 2,
user_agent: DEFAULT_UA.to_owned(),
}
}
@ -312,7 +315,8 @@ impl RustyPipeBuilder {
http,
storage: self.storage,
reporter: self.reporter,
n_retries: self.n_retries,
n_http_retries: self.n_http_retries,
n_query_retries: self.n_query_retries,
consent_cookie: format!(
"{}={}{}",
CONSENT_COOKIE,
@ -367,8 +371,18 @@ impl RustyPipeBuilder {
/// random jitter to be less predictable).
///
/// **Default value**: 3
pub fn n_retries(mut self, n_retries: u32) -> Self {
self.n_retries = n_retries;
pub fn n_http_retries(mut self, n_retries: u32) -> Self {
self.n_http_retries = n_retries;
self
}
/// Set the number of retries for YouTube API queries.
///
/// If a YouTube API requests returns invalid data, the request is repeated.
///
/// **Default value**: 2
pub fn n_query_retries(mut self, n_retries: u32) -> Self {
self.n_http_retries = n_retries;
self
}
@ -458,7 +472,7 @@ impl RustyPipe {
request: Request,
) -> core::result::Result<Response, reqwest::Error> {
let mut last_res = None;
for n in 0..self.inner.n_retries {
for n in 0..self.inner.n_http_retries {
let res = self.inner.http.execute(request.try_clone().unwrap()).await;
let emsg = match &res {
Ok(response) => {
@ -939,6 +953,44 @@ impl RustyPipeQuery {
endpoint: &str,
body: &B,
deobf: Option<&Deobfuscator>,
) -> Result<M> {
for n in 0..self.client.inner.n_query_retries.saturating_sub(1) {
let res = self
._try_execute_request_deobf::<R, M, B>(ctype, operation, id, endpoint, body, deobf)
.await;
let emsg = match res {
Ok(res) => return Ok(res),
Err(error) => match &error {
Error::Extraction(e) => match e {
ExtractionError::Deserialization(_)
| ExtractionError::InvalidData(_)
| ExtractionError::WrongResult(_)
| ExtractionError::Retry => e.to_string(),
_ => return Err(error),
},
_ => return Err(error),
},
};
warn!("{} retry attempt #{}. Error: {}.", operation, n, emsg);
}
self._try_execute_request_deobf::<R, M, B>(ctype, operation, id, endpoint, body, deobf)
.await
}
/// Single try of `execute_request_deobf`
async fn _try_execute_request_deobf<
R: DeserializeOwned + MapResponse<M> + Debug,
M,
B: Serialize + ?Sized,
>(
&self,
ctype: ClientType,
operation: &str,
id: &str,
endpoint: &str,
body: &B,
deobf: Option<&Deobfuscator>,
) -> Result<M> {
let request = self
.request_builder(ctype, endpoint)
@ -949,7 +1001,7 @@ impl RustyPipeQuery {
let request_url = request.url().to_string();
let request_headers = request.headers().to_owned();
let response = self.client.inner.http.execute(request).await?;
let response = self.client.http_request(request).await?;
let status = response.status();
let resp_str = response.text().await?;
@ -1013,7 +1065,8 @@ impl RustyPipeQuery {
ExtractionError::VideoUnavailable(_, _)
| ExtractionError::VideoAgeRestricted
| ExtractionError::ContentUnavailable(_)
| ExtractionError::NoData => (),
| ExtractionError::NoData
| ExtractionError::Retry => (),
_ => create_report(Level::ERR, Some(e.to_string()), Vec::new()),
}
Err(e.into())

7
src/client/playlist.rs
View file

@ -193,12 +193,7 @@ impl MapResponse<Paginator<PlaylistVideo>> for response::PlaylistCont {
_deobf: Option<&Deobfuscator>,
) -> Result<MapResult<Paginator<PlaylistVideo>>, ExtractionError> {
let mut actions = self.on_response_received_actions;
let action = some_or_bail!(
actions.try_swap_remove(0),
Err(ExtractionError::InvalidData(
"no continuation action".into()
))
);
let action = actions.try_swap_remove(0).ok_or(ExtractionError::Retry)?;
let (items, ctoken) =
map_playlist_items(action.append_continuation_items_action.continuation_items.c);

2
src/client/response/channel.rs
View file

@ -25,6 +25,8 @@ pub struct Channel {
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ChannelCont {
#[serde(default)]
#[serde_as(as = "VecSkipError<_>")]
pub on_response_received_actions: Vec<OnResponseReceivedAction>,
}

1
src/client/response/playlist.rs
View file

@ -19,6 +19,7 @@ pub struct Playlist {
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PlaylistCont {
#[serde(default)]
#[serde_as(as = "VecSkipError<_>")]
pub on_response_received_actions: Vec<OnResponseReceivedAction>,
}

1
src/client/response/search.rs
View file

@ -25,6 +25,7 @@ pub struct Search {
pub struct SearchCont {
#[serde_as(as = "Option<JsonString>")]
pub estimated_results: Option<u64>,
#[serde_as(as = "VecSkipError<_>")]
pub on_response_received_commands: Vec<SearchContCommand>,
}

12
src/client/response/video_details.rs
View file

@ -11,7 +11,8 @@ use crate::serializer::{
};
use super::{
ContinuationEndpoint, ContinuationItemRenderer, Icon, Thumbnails, VideoListItem, VideoOwner,
ContinuationEndpoint, ContinuationItemRenderer, Icon, MusicContinuation, Thumbnails,
VideoListItem, VideoOwner,
};
/*
@ -282,6 +283,8 @@ pub struct RecommendationResults {
/// Can be `None` for age-restricted videos
#[serde_as(as = "Option<VecLogError<_>>")]
pub results: Option<MapResult<Vec<VideoListItem>>>,
#[serde_as(as = "Option<VecSkipError<_>>")]
pub continuations: Option<Vec<MusicContinuation>>,
}
/// The engagement panels are displayed below the video and contain chapter markers
@ -418,9 +421,12 @@ pub struct CommentItemSectionHeaderMenuItem {
*/
/// Video recommendations continuation response
#[serde_as]
#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct VideoRecommendations {
#[serde(default)]
#[serde_as(as = "VecSkipError<_>")]
pub on_response_received_endpoints: Vec<RecommendationsContItem>,
}
@ -459,8 +465,8 @@ pub struct VideoComments {
/// - Comment replies: appendContinuationItemsAction
/// - n*commentRenderer, continuationItemRenderer:
/// replies + continuation
#[serde_as(as = "VecLogError<_>")]
pub on_response_received_endpoints: MapResult<Vec<CommentsContItem>>,
#[serde_as(as = "Option<VecLogError<_>>")]
pub on_response_received_endpoints: Option<MapResult<Vec<CommentsContItem>>>,
}
/// Video comments continuation

102
src/client/video_details.rs
View file

@ -251,7 +251,7 @@ impl MapResponse<VideoDetails> for response::VideoDetails {
.secondary_results
.and_then(|sr| {
sr.secondary_results.results.map(|r| {
let mut res = map_recommendations(r, lang);
let mut res = map_recommendations(r, sr.secondary_results.continuations, lang);
warnings.append(&mut res.warnings);
res.c
})
@ -342,15 +342,11 @@ impl MapResponse<Paginator<RecommendedVideo>> for response::VideoRecommendations
_deobf: Option<&crate::deobfuscate::Deobfuscator>,
) -> Result<MapResult<Paginator<RecommendedVideo>>, ExtractionError> {
let mut endpoints = self.on_response_received_endpoints;
let cont = some_or_bail!(
endpoints.try_swap_remove(0),
Err(ExtractionError::InvalidData(
"no continuation endpoint".into()
))
);
let cont = endpoints.try_swap_remove(0).ok_or(ExtractionError::Retry)?;
Ok(map_recommendations(
cont.append_continuation_items_action.continuation_items,
None,
lang,
))
}
@ -363,57 +359,54 @@ impl MapResponse<Paginator<Comment>> for response::VideoComments {
lang: Language,
_deobf: Option<&crate::deobfuscate::Deobfuscator>,
) -> Result<MapResult<Paginator<Comment>>, ExtractionError> {
let mut warnings = self.on_response_received_endpoints.warnings;
let received_endpoints = self
.on_response_received_endpoints
.ok_or(ExtractionError::Retry)?;
let mut warnings = received_endpoints.warnings;
let mut comments = Vec::new();
let mut comment_count = None;
let mut ctoken = None;
self.on_response_received_endpoints
.c
.into_iter()
.for_each(|citem| {
let mut items = citem.append_continuation_items_action.continuation_items;
warnings.append(&mut items.warnings);
items.c.into_iter().for_each(|item| match item {
response::video_details::CommentListItem::CommentThreadRenderer {
comment,
replies,
received_endpoints.c.into_iter().for_each(|citem| {
let mut items = citem.append_continuation_items_action.continuation_items;
warnings.append(&mut items.warnings);
items.c.into_iter().for_each(|item| match item {
response::video_details::CommentListItem::CommentThreadRenderer {
comment,
replies,
rendering_priority,
} => {
let mut res = map_comment(
comment.comment_renderer,
Some(replies),
rendering_priority,
} => {
let mut res = map_comment(
comment.comment_renderer,
Some(replies),
rendering_priority,
lang,
);
comments.push(res.c);
warnings.append(&mut res.warnings)
}
response::video_details::CommentListItem::CommentRenderer(comment) => {
let mut res = map_comment(
comment,
None,
response::video_details::CommentPriority::RenderingPriorityUnknown,
lang,
);
comments.push(res.c);
warnings.append(&mut res.warnings)
}
response::video_details::CommentListItem::ContinuationItemRenderer {
continuation_endpoint,
} => {
ctoken = Some(continuation_endpoint.continuation_command.token);
}
response::video_details::CommentListItem::CommentsHeaderRenderer {
count_text,
} => {
comment_count = count_text.and_then(|txt| {
util::parse_numeric_or_warn::<u64>(&txt, &mut warnings)
});
}
});
lang,
);
comments.push(res.c);
warnings.append(&mut res.warnings)
}
response::video_details::CommentListItem::CommentRenderer(comment) => {
let mut res = map_comment(
comment,
None,
response::video_details::CommentPriority::RenderingPriorityUnknown,
lang,
);
comments.push(res.c);
warnings.append(&mut res.warnings)
}
response::video_details::CommentListItem::ContinuationItemRenderer {
continuation_endpoint,
} => {
ctoken = Some(continuation_endpoint.continuation_command.token);
}
response::video_details::CommentListItem::CommentsHeaderRenderer { count_text } => {
comment_count = count_text
.and_then(|txt| util::parse_numeric_or_warn::<u64>(&txt, &mut warnings));
}
});
});
Ok(MapResult {
c: Paginator::new(comment_count, comments, ctoken),
@ -424,6 +417,7 @@ impl MapResponse<Paginator<Comment>> for response::VideoComments {
fn map_recommendations(
r: MapResult<Vec<response::VideoListItem>>,
continuations: Option<Vec<response::MusicContinuation>>,
lang: Language,
) -> MapResult<Paginator<RecommendedVideo>> {
let mut warnings = r.warnings;
@ -475,6 +469,12 @@ fn map_recommendations(
})
.collect::<Vec<_>>();
if let Some(continuations) = continuations {
continuations.into_iter().for_each(|c| {
ctoken = Some(c.next_continuation_data.continuation);
})
};
MapResult {
c: Paginator::new(None, items, ctoken),
warnings,

2
src/error.rs
View file

@ -87,6 +87,8 @@ pub enum ExtractionError {
WrongResult(String),
#[error("Warnings during deserialization/mapping")]
DeserializationWarnings,
#[error("Got no data from YouTube, attempt retry")]
Retry,
}
/// Internal error

10
tests/youtube.rs
View file

@ -19,7 +19,7 @@ use rustypipe::param::{
#[case::tv_html5_embed(ClientType::TvHtml5Embed)]
#[case::android(ClientType::Android)]
#[case::ios(ClientType::Ios)]
#[test_log::test(tokio::test)]
#[tokio::test]
async fn get_player(#[case] client_type: ClientType) {
let rp = RustyPipe::builder().strict().build();
let player_data = rp.query().player("n4tK7LYFxI0", client_type).await.unwrap();
@ -179,7 +179,7 @@ async fn get_playlist(
assert!(!playlist.thumbnail.is_empty());
}
#[test_log::test(tokio::test)]
#[tokio::test]
async fn playlist_cont() {
let rp = RustyPipe::builder().strict().build();
let mut playlist = rp
@ -197,7 +197,7 @@ async fn playlist_cont() {
assert!(playlist.videos.count.unwrap() > 100);
}
#[test_log::test(tokio::test)]
#[tokio::test]
async fn playlist_cont2() {
let rp = RustyPipe::builder().strict().build();
let mut playlist = rp
@ -311,7 +311,6 @@ async fn get_video_details_music() {
assert!(!details.is_live);
assert!(!details.is_ccommons);
assert!(!details.recommended.items.is_empty());
assert!(!details.recommended.is_exhausted());
// Comments are disabled for this video
@ -369,7 +368,6 @@ async fn get_video_details_ccommons() {
assert!(!details.is_live);
assert!(details.is_ccommons);
assert!(!details.recommended.items.is_empty());
assert!(!details.recommended.is_exhausted());
assert!(
@ -506,7 +504,6 @@ async fn get_video_details_chapters() {
]
"###);
assert!(!details.recommended.items.is_empty());
assert!(!details.recommended.is_exhausted());
assert!(
@ -566,7 +563,6 @@ async fn get_video_details_live() {
assert!(details.is_live);
assert!(!details.is_ccommons);
assert!(!details.recommended.items.is_empty());
assert!(!details.recommended.is_exhausted());
// No comments because livestream