1"""YouTube Shorts client — channel Shorts tab, paginated through InnerTube.
2
3Verified against live pages on 2026-08-06 from a datacenter IP:
4
5* ``GET youtube.com/@handle/shorts`` returns the full first page with **no proxy and no TLS
6 impersonation** — the data sits in the inline ``var ytInitialData = {...};`` blob, so there is
7 nothing to render and no browser to pay for.
8* Each Shorts entry is a ``shortsLockupViewModel``: 48 per page, carrying the videoId, title and
9 a human view count ("101M views"). Google renamed this node from the older
10 ``reelItemRenderer``; scrapers still looking for the old name find nothing on a page that looks
11 perfectly fine, which is the quiet way this breaks.
12* The other two listing types use DIFFERENT nodes, measured on the same day: a channel's
13 ``/videos`` tab is 31x ``lockupViewModel`` (NOT ``videoRenderer`` — that is the old layout), and
14 ``/results?search_query=`` is ``videoRenderer`` plus ``shortsLockupViewModel`` mixed together.
15 One page shape does not imply the next, so each type declares its own node.
16* Pagination is InnerTube: the page also ships ``INNERTUBE_API_KEY`` and
17 ``INNERTUBE_CLIENT_VERSION``, and ytInitialData carries a ``continuationCommand`` token. POSTing
18 that token to ``/youtubei/v1/browse`` returns the next 48 plus the next token. Verified: page 2
19 returned 48 more and another token.
20
21The client is deliberately paced. YouTube does not answer with a clean 429 the way Google Trends
22does — it degrades, serving a consent/challenge page or an empty payload, which is far harder to
23detect than an error code. Slow and boring is the cheaper failure mode.
24"""
25
26from __future__ import annotations
27
28import gzip
29import http.cookiejar
30import json
31import logging
32import random
33import re
34import time
35import urllib.error
36import urllib.parse
37import urllib.request
38
39BASE = "https://www.youtube.com"
40UA = (
41 "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
42 "(KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36"
43)
44
45MIN_GAP = 1.0
46BACKOFF = (5.0, 15.0)
47PAGE_SIZE_HINT = 48
48
49log = logging.getLogger("youtube-shorts")
50
51
52class ShortsError(RuntimeError):
53 """Fetch or parse failed in a way retrying will not fix."""
54
55
56class ShortsBlocked(ShortsError):
57 """YouTube stopped answering with data — rotate the IP rather than waiting."""
58
59
60class ShortsUnavailable(ShortsError):
61 """A structure we depend on is gone (YouTube renamed or removed it)."""
62
63
64def _find_all(node, key, out):
65 """Collect every value stored under `key`, at any depth.
66
67 YouTube moves nodes between wrappers constantly (tabs, shelves, view models), so walking for
68 the leaf we want survives layout churn that a fixed path would not.
69 """
70 if isinstance(node, dict):
71 for k, v in node.items():
72 if k == key:
73 out.append(v)
74 else:
75 _find_all(v, key, out)
76 elif isinstance(node, list):
77 for v in node:
78 _find_all(v, key, out)
79 return out
80
81
82
83
84LISTING_NODES = {
85 "shorts": ("shortsLockupViewModel",),
86 "videos": ("lockupViewModel",),
87
88
89
90 "search": ("videoRenderer",),
91}
92
93
94
95
96LISTING_ENDPOINT = {"shorts": "browse", "videos": "browse", "search": "search"}
97
98
99def channel_url(channel: str) -> str:
100 """Accept a handle, a bare name, a /channel/UC… id or a full URL, return the Shorts tab."""
101 c = (channel or "").strip()
102 if not c:
103 raise ShortsError("empty channel")
104 if c.startswith("http://") or c.startswith("https://"):
105 url = c.split("?")[0].rstrip("/")
106
107 for tab in (
108 "/shorts",
109 "/videos",
110 "/streams",
111 "/featured",
112 "/community",
113 "/playlists",
114 ):
115 if url.endswith(tab):
116 url = url[: -len(tab)]
117 return url + "/shorts"
118 if c.startswith("UC") and len(c) >= 20:
119 return "%s/channel/%s/shorts" % (BASE, c)
120 return "%s/@%s/shorts" % (BASE, c.lstrip("@"))
121
122
123def listing_url(target: str, kind: str) -> str:
124 """URL for a listing. `kind` is shorts | videos | search."""
125 if kind == "search":
126 return "%s/results?search_query=%s" % (BASE, urllib.parse.quote(target))
127 url = channel_url(target)
128 if kind == "videos":
129 return url[: -len("/shorts")] + "/videos"
130 return url
131
132
133class ShortsClient:
134 """One session against YouTube, optionally through one proxy.
135
136 Bound to a single egress IP for its life: the consent cookies YouTube sets belong with the IP
137 that got them. To rotate, build a new client (see `new_client` in main.py).
138 """
139
140 def __init__(
141 self,
142 min_gap: float = MIN_GAP,
143 proxy_url: str | None = None,
144 hl: str = "en",
145 gl: str = "US",
146 ):
147 handlers = [urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())]
148 if proxy_url:
149 handlers.append(
150 urllib.request.ProxyHandler({"http": proxy_url, "https": proxy_url})
151 )
152 self._op = urllib.request.build_opener(*handlers)
153 self.proxy_url = proxy_url
154 self.min_gap = float(min_gap)
155 self.hl = hl
156 self.gl = gl
157 self._last = 0.0
158 self.requests_made = 0
159 self._api_key = None
160 self._client_version = None
161
162
163 def _pace(self) -> None:
164 wait = self.min_gap - (time.monotonic() - self._last)
165 if wait > 0:
166 time.sleep(wait + random.uniform(0, 0.3))
167 self._last = time.monotonic()
168
169 def _raw(self, url: str, data: bytes | None = None) -> str:
170 headers = {
171 "User-Agent": UA,
172 "Accept": "*/*",
173 "Accept-Language": "%s,en;q=0.9" % self.hl,
174 "Accept-Encoding": "gzip",
175 }
176 if data is not None:
177 headers["Content-Type"] = "application/json"
178 resp = self._op.open(
179 urllib.request.Request(url, data=data, headers=headers), timeout=45
180 )
181 blob = resp.read()
182 if resp.headers.get("Content-Encoding") == "gzip":
183 blob = gzip.decompress(blob)
184 return blob.decode("utf-8", "replace")
185
186 def _fetch(self, url: str, data: bytes | None = None) -> str:
187 for attempt, pause in enumerate((None,) + BACKOFF):
188 if pause:
189 log.warning(
190 "YouTube refused the request — waiting %.0fs then retrying", pause
191 )
192 time.sleep(pause)
193 self._pace()
194 try:
195 self.requests_made += 1
196 return self._raw(url, data)
197 except urllib.error.HTTPError as e:
198 if e.code in (429, 503):
199 continue
200 if e.code == 404:
201 raise ShortsUnavailable("channel not found: %s" % url)
202 raise ShortsError("HTTP %s for %s" % (e.code, url))
203 raise ShortsBlocked("blocked after %d attempts: %s" % (len(BACKOFF) + 1, url))
204
205
206 def first_page(self, target: str, kind: str = "shorts") -> tuple[list, str | None, dict]:
207 """(rows, continuation_token, info) for a channel tab or a search query."""
208 url = listing_url(target, kind)
209 html = self._fetch(url)
210
211 m = re.search(r"var ytInitialData = (\{.*?\});</script>", html, re.S)
212 if not m:
213
214
215 raise ShortsBlocked(
216 "no ytInitialData on %s — served a consent/challenge page instead" % url
217 )
218 data = json.loads(m.group(1))
219
220
221 k = re.search(r'"INNERTUBE_API_KEY":"([^"]+)"', html)
222 v = re.search(r'"INNERTUBE_CLIENT_VERSION":"([^"]+)"', html)
223 if k:
224 self._api_key = k.group(1)
225 if v:
226 self._client_version = v.group(1)
227
228 shorts = self._parse_shorts(data, kind)
229 if not shorts and self._looks_like_a_channel(data):
230
231
232 log.info("%s has no Shorts", url)
233 return shorts, self._continuation(data), self._channel_info(data)
234
235 @staticmethod
236 def _looks_like_a_channel(data: dict) -> bool:
237 return bool(
238 _find_all(data, "c4TabbedHeaderRenderer", [])
239 or _find_all(data, "pageHeaderRenderer", [])
240 )
241
242
243 def next_page(self, token: str, kind: str = "shorts") -> tuple[list, str | None]:
244 if not (self._api_key and self._client_version):
245 raise ShortsError(
246 "call first_page() before paginating — no InnerTube credentials yet"
247 )
248 body = json.dumps(
249 {
250 "context": {
251 "client": {
252 "clientName": "WEB",
253 "clientVersion": self._client_version,
254 "hl": self.hl,
255 "gl": self.gl,
256 }
257 },
258 "continuation": token,
259 }
260 ).encode()
261 raw = self._fetch(
262 "%s/youtubei/v1/%s?key=%s&prettyPrint=false"
263 % (BASE, LISTING_ENDPOINT.get(kind, "browse"), self._api_key),
264 data=body,
265 )
266 data = json.loads(raw)
267 return self._parse_shorts(data, kind), self._continuation(data)
268
269 @staticmethod
270 def _continuation(data: dict) -> str | None:
271 for cmd in _find_all(data, "continuationCommand", []):
272 tok = (cmd or {}).get("token")
273 if tok:
274 return tok
275 return None
276
277
278 @classmethod
279 def _parse_shorts(cls, data: dict, kind: str = "shorts") -> list:
280 """Rows for one listing type. Each node shape gets its own reader because YouTube gives
281 them genuinely different fields — a search hit has a channel and a publish date that a
282 channel listing does not, and pretending otherwise would silently drop them."""
283 rows, seen = [], set()
284 for node in LISTING_NODES.get(kind, LISTING_NODES["shorts"]):
285 for vm in _find_all(data, node, []):
286 row = (cls._one_short(vm) if node == "shortsLockupViewModel"
287 else cls._one_lockup(vm) if node == "lockupViewModel"
288 else cls._one_video_renderer(vm))
289 vid = row.get("videoId")
290 ctype = row.pop("contentType", None)
291
292
293
294
295 if ctype and "VIDEO" not in str(ctype).upper():
296 continue
297 if not row.get("title"):
298 continue
299
300
301 if vid and vid not in seen:
302 seen.add(vid)
303 rows.append(row)
304 return rows
305
306 @staticmethod
307 def _one_lockup(vm: dict) -> dict:
308 """`lockupViewModel` — the current channel /videos and mixed-shelf shape."""
309 vid = vm.get("contentId")
310 meta = ((vm.get("metadata") or {}).get("lockupMetadataViewModel") or {})
311 title = ((meta.get("title") or {}).get("content"))
312 rows_meta = (((meta.get("metadata") or {}).get("contentMetadataViewModel") or {})
313 .get("metadataRows") or [])
314 bits = []
315 for r in rows_meta:
316 for part in (r.get("metadataParts") or []):
317 t = ((part.get("text") or {}).get("content"))
318 if t:
319 bits.append(t)
320 views = next((b for b in bits if "view" in b.lower()), None)
321 published = next((b for b in bits if "ago" in b.lower()), None)
322 thumbs = (((vm.get("contentImage") or {}).get("thumbnailViewModel") or {})
323 .get("image") or {}).get("sources") or []
324 return {
325 "videoId": vid,
326 "contentType": vm.get("contentType"),
327 "title": title,
328 "url": ("https://www.youtube.com/watch?v=%s" % vid) if vid else None,
329 "viewCountText": views,
330 "viewCount": _parse_views(views),
331 "publishedText": published,
332 "thumbnail": (thumbs[-1].get("url") if thumbs else None),
333 "metadataParts": bits or None,
334 }
335
336 @staticmethod
337 def _one_video_renderer(vm: dict) -> dict:
338 """`videoRenderer` — the search-results shape, which alone carries the channel name."""
339 def txt(node):
340 if not isinstance(node, dict):
341 return None
342 if node.get("simpleText"):
343 return node["simpleText"]
344 runs = node.get("runs") or []
345 return "".join(r.get("text", "") for r in runs) or None
346 vid = vm.get("videoId")
347 thumbs = ((vm.get("thumbnail") or {}).get("thumbnails")) or []
348 owner = (vm.get("ownerText") or vm.get("longBylineText") or {})
349 return {
350 "videoId": vid,
351 "title": txt(vm.get("title")),
352 "url": ("https://www.youtube.com/watch?v=%s" % vid) if vid else None,
353 "viewCountText": txt(vm.get("viewCountText")),
354 "viewCount": _parse_views(txt(vm.get("viewCountText"))),
355 "publishedText": txt(vm.get("publishedTimeText")),
356 "durationText": txt(vm.get("lengthText")),
357 "channelName": txt(owner),
358 "description": txt(vm.get("detailedMetadataSnippets", [{}])[0].get("snippetText"))
359 if vm.get("detailedMetadataSnippets") else None,
360 "thumbnail": (thumbs[-1].get("url") if thumbs else None),
361 }
362
363 @staticmethod
364 def _one_short(vm: dict) -> dict:
365 overlay = vm.get("overlayMetadata") or {}
366 title = (overlay.get("primaryText") or {}).get("content")
367 views_text = (overlay.get("secondaryText") or {}).get("content")
368 tap = (vm.get("onTap") or {}).get("innertubeCommand") or {}
369 vid = (tap.get("reelWatchEndpoint") or {}).get("videoId")
370 thumbs = ((vm.get("thumbnail") or {}).get("sources")) or []
371 return {
372 "videoId": vid,
373 "title": title,
374 "url": ("https://www.youtube.com/shorts/%s" % vid) if vid else None,
375 "viewCountText": views_text,
376 "viewCount": _parse_views(views_text),
377 "thumbnail": (thumbs[-1].get("url") if thumbs else None),
378 "accessibilityText": ((vm.get("accessibilityText")) or None),
379 }
380
381 @staticmethod
382 def _channel_info(data: dict) -> dict:
383 for h in _find_all(data, "pageHeaderRenderer", []):
384 return {"channelName": h.get("pageTitle")}
385 for h in _find_all(data, "c4TabbedHeaderRenderer", []):
386 return {
387 "channelName": h.get("title"),
388 "subscriberCountText": (h.get("subscriberCountText") or {}).get(
389 "simpleText"
390 ),
391 }
392 return {}
393
394
395_SUFFIX = {"K": 1_000, "M": 1_000_000, "B": 1_000_000_000}
396
397
398def _parse_views(text) -> int | None:
399 """ "1.2B views" -> 1200000000. Returns None rather than 0 when unparseable: 0 would be a
400 lie a caller could quietly aggregate, whereas None is visibly missing."""
401 if not text:
402 return None
403
404
405 if "no view" in str(text).lower():
406 return 0
407 m = re.search(r"([\d.,]+)\s*([KMB])?", str(text).replace(",", ""))
408 if not m:
409 return None
410 try:
411 n = float(m.group(1))
412 except ValueError:
413 return None
414 return int(n * _SUFFIX.get((m.group(2) or "").upper(), 1))
