1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
from typing import Any, Optional
from ytmusicapi.navigation import nav
CONTINUATION_TOKEN = ["continuationItemRenderer", "continuationEndpoint", "continuationCommand", "token"]
CONTINUATION_ITEMS = ["onResponseReceivedActions", 0, "appendContinuationItemsAction", "continuationItems"]
def get_continuation_token(results: list[dict[str, Any]]) -> Optional[str]:
return nav(results[-1], CONTINUATION_TOKEN, True)
def get_continuations_2025(results, limit, request_func, parse_func):
items = []
continuation_token = get_continuation_token(results["contents"])
while continuation_token and (limit is None or len(items) < limit):
response = request_func({"continuation": continuation_token})
continuation_items = nav(response, CONTINUATION_ITEMS, True)
if not continuation_items:
break
contents = parse_func(continuation_items)
if len(contents) == 0:
break
items.extend(contents)
continuation_token = get_continuation_token(continuation_items)
return items
def get_continuations(
results, continuation_type, limit, request_func, parse_func, ctoken_path="", reloadable=False
):
items = []
while "continuations" in results and (limit is None or len(items) < limit):
additionalParams = (
get_reloadable_continuation_params(results)
if reloadable
else get_continuation_params(results, ctoken_path)
)
response = request_func(additionalParams)
if "continuationContents" in response:
results = response["continuationContents"][continuation_type]
else:
break
contents = get_continuation_contents(results, parse_func)
if len(contents) == 0:
break
items.extend(contents)
return items
def get_validated_continuations(
results, continuation_type, limit, per_page, request_func, parse_func, ctoken_path=""
):
items = []
while "continuations" in results and len(items) < limit:
additionalParams = get_continuation_params(results, ctoken_path)
wrapped_parse_func = lambda raw_response: get_parsed_continuation_items(
raw_response, parse_func, continuation_type
)
validate_func = lambda parsed: validate_response(parsed, per_page, limit, len(items))
response = resend_request_until_parsed_response_is_valid(
request_func, additionalParams, wrapped_parse_func, validate_func, 3
)
results = response["results"]
items.extend(response["parsed"])
return items
def get_parsed_continuation_items(response, parse_func, continuation_type):
results = response["continuationContents"][continuation_type]
return {"results": results, "parsed": get_continuation_contents(results, parse_func)}
def get_continuation_params(results, ctoken_path=""):
ctoken = nav(results, ["continuations", 0, "next" + ctoken_path + "ContinuationData", "continuation"])
return get_continuation_string(ctoken)
def get_reloadable_continuation_params(results):
ctoken = nav(results, ["continuations", 0, "reloadContinuationData", "continuation"])
return get_continuation_string(ctoken)
def get_continuation_string(ctoken):
return "&ctoken=" + ctoken + "&continuation=" + ctoken
def get_continuation_contents(continuation, parse_func):
for term in ["contents", "items"]:
if term in continuation:
return parse_func(continuation[term])
return []
def resend_request_until_parsed_response_is_valid(
request_func, request_additional_params, parse_func, validate_func, max_retries
):
response = request_func(request_additional_params)
parsed_object = parse_func(response)
retry_counter = 0
while not validate_func(parsed_object) and retry_counter < max_retries:
response = request_func(request_additional_params)
attempt = parse_func(response)
if len(attempt["parsed"]) > len(parsed_object["parsed"]):
parsed_object = attempt
retry_counter += 1
return parsed_object
def validate_response(response, per_page, limit, current_count):
remaining_items_count = limit - current_count
expected_items_count = min(per_page, remaining_items_count)
# response is invalid, if it has less items then minimal expected count
return len(response["parsed"]) >= expected_items_count
|
