peter
955c8c3445
Deep review of the debt burn: - `refusal_reason` told the user to reconnect on ANY token-endpoint failure — a 5xx, a 429, a misconfigured client — while the grant stayed and the account still read as connected, with no connect button. Only `invalid_grant` (the one that drops the grant) means reconnect now. - `pins.reorder` writes only the rows that change, so two overlapping reorders each skipped rows that matched in their own stale snapshot and interleaved into two pins on one slot. It locks the user's pins first; the second waits and deals from the first's result. - Adding a whitelist entry committed the invite and the activation it implies separately, and its audit row was written before a new invite had an id (target NULL). One transaction, flushed for the id. - The mutation convention test skipped async routes and dicts returned by name: request-access, purge-discovery and the subscription sync answered without `ok`. They do now, and the test covers both. The helper-commit check looked only at `_`-names; it reads every non-route function, which listed two deliberate committers (upsert_pending_invite, ensure_welcome) — now allowlisted with their reasons.
557 lines
24 KiB
Python
557 lines
24 KiB
Python
"""Thin synchronous YouTube Data API v3 client bound to a user's OAuth token.
|
|
|
|
Each list call costs 1 quota unit and is recorded via the central quota guard.
|
|
Public reads (channels/videos/playlistItems) use the configured API key when available
|
|
so they don't depend on a specific user's token; subscriptions.list?mine=true always
|
|
uses OAuth.
|
|
"""
|
|
|
|
import logging
|
|
from collections.abc import Iterator
|
|
from datetime import datetime, timedelta, timezone
|
|
from typing import Literal
|
|
|
|
import httpx
|
|
|
|
from app import logs, quota, sysconfig
|
|
from app.models import User
|
|
from app.security import decrypt, decrypt_optional, encrypt
|
|
|
|
log = logs.get_logger(__name__)
|
|
GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
|
|
API_BASE = "https://www.googleapis.com/youtube/v3"
|
|
|
|
|
|
class YouTubeError(Exception):
|
|
"""An upstream YouTube Data API failure.
|
|
|
|
`reason` is the API's own error reason code (`quotaExceeded`, `subscriptionDuplicate`,
|
|
`videoNotFound`, …) when the body carried one. Branch on THAT, never on the message: the
|
|
message is prose, it changes under us, and it is deliberately the poorer of the two here.
|
|
|
|
`path` is the API endpoint that ANSWERED with the failure (`playlistItems`); empty when no
|
|
endpoint did — the request never got through (`transport`) or the token step refused first."""
|
|
|
|
def __init__(self, message: str, *, reason: str = "", path: str = ""):
|
|
super().__init__(message)
|
|
self.reason = reason
|
|
self.path = path
|
|
|
|
|
|
def _error_reason(resp: httpx.Response) -> str:
|
|
"""The `reason` YouTube puts on an API error. That is the whole diagnostic value of an error
|
|
body; the `message` beside it is prose, and upstream prose can quote the request back — a
|
|
search term, a playlist title."""
|
|
try:
|
|
error = resp.json().get("error")
|
|
# The OAuth token endpoint's shape: `{"error": "invalid_grant", "error_description": …}`.
|
|
if isinstance(error, str):
|
|
return error or "<none>"
|
|
errors = (error or {}).get("errors") or []
|
|
reasons = ",".join(e.get("reason", "") for e in errors if e.get("reason"))
|
|
# silent-ok: the sentinel it returns goes straight into the caller's warning line, which
|
|
# is the report. (Any non-JSON body — an HTML error page from a proxy.)
|
|
except Exception: # noqa: BLE001
|
|
return "<unparseable>"
|
|
return reasons or "<none>"
|
|
|
|
|
|
def _upstream_failure(
|
|
method: str, path: str, resp: httpx.Response, expected: frozenset[str] = frozenset()
|
|
) -> YouTubeError:
|
|
"""Report one upstream failure and build the error to raise. Every YouTube response that is
|
|
not a success goes through here — the Data API's reads, writes and deletes and the OAuth
|
|
token refresh — so what may appear in a log line cannot drift apart between them, and the
|
|
error a caller raises carries the reason code rather than the body, because callers put it in
|
|
a log line (`routes/channels.py`) and in a 422 answered to the user. (A request that got no
|
|
response at all is `_send`'s, with the reason `transport`.) The body is logged only under
|
|
the requesting user's own Teljes consent: it is the request quoted back."""
|
|
reason = _error_reason(resp)
|
|
# A reason the caller named as an answer rather than a failure (`channelNotFound` on a `mine`
|
|
# read: the account simply has no channel) is still raised, but it is not a warning — it came
|
|
# back every six hours for an account that will never have one.
|
|
log.log(
|
|
logging.DEBUG if reason in expected else logging.WARNING,
|
|
"YouTube %s %s -> %s (%s)",
|
|
method,
|
|
path,
|
|
resp.status_code,
|
|
reason,
|
|
extra=logs.event(
|
|
"youtube.api.error",
|
|
method=method,
|
|
path=path,
|
|
status=resp.status_code,
|
|
reason=reason,
|
|
),
|
|
)
|
|
if logs.consented_trace():
|
|
log.log(
|
|
logs.TRACE,
|
|
"the upstream body behind it",
|
|
# log-pii-ok: only under this user's own Teljes consent, which is what it is for
|
|
extra=logs.event("youtube.api.body", body=resp.text[:500]),
|
|
)
|
|
return YouTubeError(
|
|
f"{method} {path} -> {resp.status_code}: {reason}", reason=reason, path=path
|
|
)
|
|
|
|
|
|
# The `mine` reads' answer for a Google account that has no YouTube channel.
|
|
_NO_CHANNEL = frozenset({"channelNotFound"})
|
|
|
|
# The reasons that mean the project's quota is spent: every further call today gets the same answer.
|
|
_QUOTA_REASONS = frozenset({"quotaExceeded", "dailyLimitExceeded"})
|
|
|
|
|
|
def is_quota_refusal(exc: YouTubeError) -> bool:
|
|
"""Whether the API refused for want of quota. `reason` can list several codes, comma-joined."""
|
|
return any(r in _QUOTA_REASONS for r in exc.reason.split(","))
|
|
|
|
|
|
# The token endpoint's answer for a revoked or expired grant — the one refusal a reconnect fixes.
|
|
_DEAD_GRANT = "invalid_grant"
|
|
|
|
# What a refused YouTube action means to the person who asked for it — the three things they can
|
|
# act on differently: wait for tomorrow's budget, reconnect the account, or try again later.
|
|
YouTubeRefusal = Literal["quota", "reconnect", "unavailable"]
|
|
|
|
|
|
def refusal_reason(exc: YouTubeError) -> YouTubeRefusal:
|
|
"""Sort a failure by what the user can do about it. A dead grant (Google refused the refresh
|
|
with `invalid_grant`, which also drops the grant — see `_drop_dead_grant`) or no usable token at
|
|
all (raised before any endpoint answered, so with no path and no transport reason) means the
|
|
account must be reconnected; retrying cannot help. Any other token-endpoint failure — a 5xx, a
|
|
429, a misconfigured client — leaves the grant as it was: reconnecting would not fix it, and
|
|
the account would still read as connected, with no connect button to press."""
|
|
if is_quota_refusal(exc):
|
|
return "quota"
|
|
if exc.path == "oauth2/token":
|
|
return "reconnect" if exc.reason == _DEAD_GRANT else "unavailable"
|
|
if not exc.path and exc.reason != "transport":
|
|
return "reconnect"
|
|
return "unavailable"
|
|
|
|
|
|
def _chunks(items: list, size: int) -> Iterator[list]:
|
|
for i in range(0, len(items), size):
|
|
yield items[i : i + size]
|
|
|
|
|
|
def best_thumbnail(thumbnails: dict | None) -> str | None:
|
|
if not thumbnails:
|
|
return None
|
|
for key in ("maxres", "standard", "high", "medium", "default"):
|
|
if key in thumbnails and thumbnails[key].get("url"):
|
|
return thumbnails[key]["url"]
|
|
return None
|
|
|
|
|
|
class YouTubeClient:
|
|
def __init__(self, db, user: User):
|
|
self.db = db
|
|
self.user = user
|
|
# Optionally route all YouTube traffic through a fixed-IP egress proxy so an
|
|
# IP-restricted API key keeps working from a host with a dynamic public IP.
|
|
proxy = sysconfig.get_str(db, "youtube_api_proxy") or None
|
|
self._http = httpx.Client(timeout=30.0, proxy=proxy)
|
|
|
|
def __enter__(self) -> "YouTubeClient":
|
|
return self
|
|
|
|
def __exit__(self, *exc) -> None:
|
|
self._http.close()
|
|
|
|
# --- transport ---
|
|
def _send(self, method: str, url: str, **kwargs) -> httpx.Response:
|
|
"""Issue an HTTP request, mapping any transport/network failure (egress proxy down,
|
|
DNS, timeout, connection reset) to YouTubeError. Every caller already handles
|
|
YouTubeError — degrade gracefully, log, and move on — so wrapping here keeps a
|
|
transient network fault (e.g. the fixed-IP egress proxy being unreachable) from
|
|
escaping as a raw httpx error and turning into an uncaught 500. HTTP status errors
|
|
are NOT touched here (we don't use raise_for_status): each call site inspects
|
|
resp.status_code itself and hands the response to `_upstream_failure`."""
|
|
try:
|
|
return self._http.request(method, url, **kwargs)
|
|
except httpx.HTTPError as exc:
|
|
raise YouTubeError(
|
|
f"YouTube API unreachable ({exc.__class__.__name__}: {exc})", reason="transport"
|
|
) from exc
|
|
|
|
# --- auth ---
|
|
def _access_token(self) -> str:
|
|
tok = self.user.token
|
|
if tok is None:
|
|
raise YouTubeError("User has no stored OAuth token")
|
|
now = datetime.now(timezone.utc)
|
|
cached = decrypt_optional(tok.access_token)
|
|
if cached and tok.expiry and tok.expiry > now + timedelta(seconds=60):
|
|
return cached
|
|
try:
|
|
refresh = decrypt(tok.refresh_token_enc)
|
|
except Exception as exc: # noqa: BLE001 — Fernet InvalidToken (TOKEN_ENCRYPTION_KEY changed)
|
|
# A key mismatch must degrade like any other token problem (caller handles YouTubeError),
|
|
# not surface as a raw 500 on every scheduler job / connect check. (B-core-2)
|
|
raise YouTubeError("Stored OAuth token is unreadable — reconnect your account") from exc
|
|
if not refresh:
|
|
raise YouTubeError("No refresh token; user must re-authenticate")
|
|
resp = self._send(
|
|
"POST",
|
|
GOOGLE_TOKEN_URL,
|
|
data={
|
|
"client_id": sysconfig.get_str(self.db, "google_client_id"),
|
|
"client_secret": sysconfig.get_str(self.db, "google_client_secret"),
|
|
"refresh_token": refresh,
|
|
"grant_type": "refresh_token",
|
|
},
|
|
)
|
|
if resp.status_code != 200:
|
|
exc = _upstream_failure("POST", "oauth2/token", resp)
|
|
if exc.reason == _DEAD_GRANT:
|
|
self._drop_dead_grant(tok)
|
|
raise exc
|
|
data = resp.json()
|
|
access = data["access_token"]
|
|
tok.access_token = encrypt(access)
|
|
tok.expiry = now + timedelta(seconds=int(data.get("expires_in", 3600)))
|
|
self.db.add(tok)
|
|
self.db.commit()
|
|
log.info("Refreshed access token for user %s", self.user.id)
|
|
return access
|
|
|
|
def _drop_dead_grant(self, tok) -> None:
|
|
"""Google answered the refresh with `invalid_grant`: the grant was revoked or expired, and no
|
|
retry will revive it. Forget it, so the account reads as NOT connected — Settings and the
|
|
onboarding wizard then offer the connect button again (a fresh consent issues a new refresh
|
|
token). Kept as "granted", the UI showed a live connection with no way to renew it while
|
|
every call failed. Committed like the refresh above: the caller rolls back on the error."""
|
|
tok.scopes = None
|
|
tok.access_token = None
|
|
tok.refresh_token_enc = None
|
|
tok.expiry = None
|
|
self.db.add(tok)
|
|
self.db.commit()
|
|
# INFO: the refusal itself is already the warning (`youtube.api.error`, invalid_grant).
|
|
log.info(
|
|
"cleared the dead YouTube grant of user %s",
|
|
self.user.id,
|
|
extra=logs.event("youtube.grant.cleared"),
|
|
)
|
|
|
|
# --- core request ---
|
|
def _get(
|
|
self,
|
|
path: str,
|
|
params: dict,
|
|
cost: int = 1,
|
|
allow_key: bool = True,
|
|
expected: frozenset[str] = frozenset(),
|
|
) -> dict:
|
|
p = dict(params)
|
|
headers = {}
|
|
api_key = sysconfig.get_str(self.db, "youtube_api_key")
|
|
if allow_key and api_key:
|
|
p["key"] = api_key
|
|
else:
|
|
headers["Authorization"] = f"Bearer {self._access_token()}"
|
|
resp = self._send("GET", f"{API_BASE}/{path}", params=p, headers=headers)
|
|
quota.record_usage(cost)
|
|
if resp.status_code != 200:
|
|
raise _upstream_failure("GET", path, resp, expected)
|
|
return resp.json()
|
|
|
|
# --- endpoints ---
|
|
def iter_subscriptions(self) -> Iterator[dict]:
|
|
"""Yield the authenticated user's subscriptions (requires OAuth)."""
|
|
page_token = None
|
|
while True:
|
|
params = {
|
|
"part": "snippet",
|
|
"mine": "true",
|
|
"maxResults": 50,
|
|
"order": "alphabetical",
|
|
}
|
|
if page_token:
|
|
params["pageToken"] = page_token
|
|
data = self._get("subscriptions", params, cost=1, allow_key=False)
|
|
for item in data.get("items", []):
|
|
snippet = item.get("snippet", {})
|
|
resource = snippet.get("resourceId", {})
|
|
yield {
|
|
"channel_id": resource.get("channelId"),
|
|
"title": snippet.get("title"),
|
|
"description": snippet.get("description"),
|
|
"thumbnail_url": best_thumbnail(snippet.get("thumbnails")),
|
|
"yt_subscription_id": item.get("id"),
|
|
}
|
|
page_token = data.get("nextPageToken")
|
|
if not page_token:
|
|
break
|
|
|
|
def iter_my_playlists(self) -> Iterator[dict]:
|
|
"""Yield the authenticated user's own playlists (OAuth, so private ones are
|
|
included). Note: YouTube's special Watch Later / History playlists are NOT
|
|
returned by the Data API and cannot be synced."""
|
|
page_token = None
|
|
while True:
|
|
params = {"part": "snippet,contentDetails", "mine": "true", "maxResults": 50}
|
|
if page_token:
|
|
params["pageToken"] = page_token
|
|
try:
|
|
data = self._get("playlists", params, cost=1, allow_key=False, expected=_NO_CHANNEL)
|
|
except YouTubeError as exc:
|
|
# A Google account without a YouTube channel has no playlists, and YouTube says
|
|
# so as a 404. That is an empty list, not a failed sync.
|
|
if exc.reason in _NO_CHANNEL:
|
|
return
|
|
raise
|
|
for item in data.get("items", []):
|
|
sn = item.get("snippet", {})
|
|
yield {
|
|
"id": item.get("id"),
|
|
"title": sn.get("title"),
|
|
"item_count": item.get("contentDetails", {}).get("itemCount"),
|
|
"thumbnail_url": best_thumbnail(sn.get("thumbnails")),
|
|
}
|
|
page_token = data.get("nextPageToken")
|
|
if not page_token:
|
|
break
|
|
|
|
def _iter_playlist_items(self, playlist_id: str) -> Iterator[dict]:
|
|
"""Paginate one of the user's playlists' items (playlistItems, contentDetails part;
|
|
OAuth so private/unlisted work), yielding each raw item in playlist order."""
|
|
page_token = None
|
|
while True:
|
|
params = {
|
|
"part": "contentDetails",
|
|
"playlistId": playlist_id,
|
|
"maxResults": 50,
|
|
}
|
|
if page_token:
|
|
params["pageToken"] = page_token
|
|
data = self._get("playlistItems", params, cost=1, allow_key=False)
|
|
yield from data.get("items", [])
|
|
page_token = data.get("nextPageToken")
|
|
if not page_token:
|
|
break
|
|
|
|
def iter_my_playlist_video_ids(self, playlist_id: str) -> Iterator[str]:
|
|
"""Yield the video ids of one of the user's own playlists, in playlist order
|
|
(OAuth, so private/unlisted playlists work)."""
|
|
for item in self._iter_playlist_items(playlist_id):
|
|
vid = item.get("contentDetails", {}).get("videoId")
|
|
if vid:
|
|
yield vid
|
|
|
|
def get_my_channel_id(self) -> str | None:
|
|
"""The authenticated user's own channel id (channels.list?mine=true). OAuth only;
|
|
1 quota unit. Returns None if the account has no channel."""
|
|
data = self._get(
|
|
"channels", {"part": "id", "mine": "true", "maxResults": 1}, allow_key=False
|
|
)
|
|
items = data.get("items", [])
|
|
return items[0].get("id") if items else None
|
|
|
|
def get_channels(self, channel_ids: list[str]) -> list[dict]:
|
|
items: list[dict] = []
|
|
for batch in _chunks(channel_ids, 50):
|
|
data = self._get(
|
|
"channels",
|
|
{
|
|
"part": "snippet,contentDetails,statistics,topicDetails,brandingSettings",
|
|
"id": ",".join(batch),
|
|
"maxResults": 50,
|
|
},
|
|
)
|
|
items.extend(data.get("items", []))
|
|
return items
|
|
|
|
def get_playlist_items(self, playlist_id: str, page_token: str | None = None) -> dict:
|
|
"""One page (up to 50) of a playlist's items, most-recent-first for uploads
|
|
playlists. Returns the raw response (items + nextPageToken)."""
|
|
params = {
|
|
"part": "contentDetails,snippet",
|
|
"playlistId": playlist_id,
|
|
"maxResults": 50,
|
|
}
|
|
if page_token:
|
|
params["pageToken"] = page_token
|
|
return self._get("playlistItems", params)
|
|
|
|
def delete_subscription(self, subscription_id: str) -> None:
|
|
"""Unsubscribe on YouTube. Requires the write scope and the user's OAuth token
|
|
(never the public API key). subscriptions.delete costs 50 quota units."""
|
|
resp = self._send(
|
|
"DELETE",
|
|
f"{API_BASE}/subscriptions",
|
|
params={"id": subscription_id},
|
|
headers={"Authorization": f"Bearer {self._access_token()}"},
|
|
)
|
|
quota.record_usage(50)
|
|
if resp.status_code not in (200, 204):
|
|
raise _upstream_failure("DELETE", "subscriptions", resp)
|
|
|
|
def insert_subscription(self, channel_id: str) -> str:
|
|
"""Subscribe to a channel on YouTube. Requires the write scope and the user's
|
|
OAuth token (never the public API key). subscriptions.insert costs 50 quota
|
|
units. Returns the new subscription resource id — store it as
|
|
`yt_subscription_id` so the channel can later be unsubscribed."""
|
|
data = self._write(
|
|
"POST",
|
|
"subscriptions",
|
|
params={"part": "snippet"},
|
|
json={"snippet": {"resourceId": {"kind": "youtube#channel", "channelId": channel_id}}},
|
|
)
|
|
return data.get("id", "")
|
|
|
|
def iter_playlist_items_with_ids(self, playlist_id: str) -> Iterator[dict]:
|
|
"""Yield {item_id, video_id} for each item of one of the user's playlists, in
|
|
playlist order. `item_id` is the playlistItem resource id, needed to delete or
|
|
reposition the item via the write API. OAuth (private playlists work)."""
|
|
for item in self._iter_playlist_items(playlist_id):
|
|
vid = item.get("contentDetails", {}).get("videoId")
|
|
if vid:
|
|
yield {"item_id": item.get("id"), "video_id": vid}
|
|
|
|
# --- write endpoints (OAuth only, never the API key; each costs 50 quota units) ---
|
|
def _write(self, method: str, path: str, *, params: dict, json: dict | None = None) -> dict:
|
|
headers = {"Authorization": f"Bearer {self._access_token()}"}
|
|
resp = self._send(method, f"{API_BASE}/{path}", params=params, json=json, headers=headers)
|
|
quota.record_usage(50)
|
|
if resp.status_code not in (200, 204):
|
|
raise _upstream_failure(method, path, resp)
|
|
return resp.json() if resp.content else {}
|
|
|
|
def get_playlist_snippet(self, playlist_id: str) -> dict | None:
|
|
"""The snippet (title/description) of one playlist, or None if it no longer
|
|
exists. 1 unit (OAuth, so private playlists work)."""
|
|
data = self._get(
|
|
"playlists",
|
|
{"part": "snippet", "id": playlist_id, "maxResults": 1},
|
|
cost=1,
|
|
allow_key=False,
|
|
)
|
|
items = data.get("items", [])
|
|
return items[0].get("snippet") if items else None
|
|
|
|
def update_playlist_title(self, playlist_id: str, title: str) -> None:
|
|
"""Rename a playlist on YouTube. 50 units."""
|
|
self._write(
|
|
"PUT",
|
|
"playlists",
|
|
params={"part": "snippet"},
|
|
json={"id": playlist_id, "snippet": {"title": title[:150]}},
|
|
)
|
|
|
|
def create_playlist(self, title: str, description: str = "", privacy: str = "private") -> str:
|
|
"""Create a new YouTube playlist; returns its id. 50 units."""
|
|
data = self._write(
|
|
"POST",
|
|
"playlists",
|
|
params={"part": "snippet,status"},
|
|
json={
|
|
"snippet": {"title": title[:150], "description": description[:5000]},
|
|
"status": {"privacyStatus": privacy},
|
|
},
|
|
)
|
|
return data["id"]
|
|
|
|
def add_playlist_item(
|
|
self, playlist_id: str, video_id: str, position: int | None = None
|
|
) -> str:
|
|
"""Append (or insert at `position`) a video into a playlist; returns the new
|
|
playlistItem id. 50 units."""
|
|
snippet = {
|
|
"playlistId": playlist_id,
|
|
"resourceId": {"kind": "youtube#video", "videoId": video_id},
|
|
}
|
|
if position is not None:
|
|
snippet["position"] = position
|
|
data = self._write(
|
|
"POST", "playlistItems", params={"part": "snippet"}, json={"snippet": snippet}
|
|
)
|
|
return data["id"]
|
|
|
|
def move_playlist_item(
|
|
self, item_id: str, playlist_id: str, video_id: str, position: int
|
|
) -> None:
|
|
"""Reposition an existing playlistItem to absolute 0-based `position`. 50 units."""
|
|
self._write(
|
|
"PUT",
|
|
"playlistItems",
|
|
params={"part": "snippet"},
|
|
json={
|
|
"id": item_id,
|
|
"snippet": {
|
|
"playlistId": playlist_id,
|
|
"position": position,
|
|
"resourceId": {"kind": "youtube#video", "videoId": video_id},
|
|
},
|
|
},
|
|
)
|
|
|
|
def delete_playlist_item(self, item_id: str) -> None:
|
|
"""Remove a playlistItem from its playlist. 50 units."""
|
|
self._write("DELETE", "playlistItems", params={"id": item_id})
|
|
|
|
def delete_playlist(self, playlist_id: str) -> None:
|
|
"""Delete a whole playlist on YouTube. 50 units."""
|
|
self._write("DELETE", "playlists", params={"id": playlist_id})
|
|
|
|
def search_videos(
|
|
self, q: str, page_token: str | None = None, order: str = "relevance"
|
|
) -> dict:
|
|
"""Live YouTube search (search.list) for embeddable videos. **100 quota units per
|
|
call** — by far the most expensive read, so callers must gate it (per-user cap +
|
|
budget pre-check). Public read: prefers the API key over OAuth.
|
|
|
|
Returns one page (up to 50) as ``{"items": [...], "next_page_token": str|None}``
|
|
where each item is a flat dict ready for stub insertion. `liveBroadcastContent` is
|
|
surfaced so the caller can drop currently-live/upcoming results (search can't filter
|
|
those out for us). Snippet carries no duration/stats — the caller enriches via
|
|
videos.list afterwards."""
|
|
params = {
|
|
"part": "snippet",
|
|
"q": q,
|
|
"type": "video",
|
|
"maxResults": 50,
|
|
"order": order,
|
|
"videoEmbeddable": "true",
|
|
"safeSearch": "none",
|
|
}
|
|
if page_token:
|
|
params["pageToken"] = page_token
|
|
data = self._get("search", params, cost=100)
|
|
items: list[dict] = []
|
|
for it in data.get("items", []):
|
|
vid = it.get("id", {}).get("videoId")
|
|
if not vid:
|
|
continue
|
|
sn = it.get("snippet", {})
|
|
items.append(
|
|
{
|
|
"id": vid,
|
|
"channel_id": sn.get("channelId"),
|
|
"channel_title": sn.get("channelTitle"),
|
|
"title": sn.get("title"),
|
|
"published_at": sn.get("publishedAt"),
|
|
"thumbnail_url": best_thumbnail(sn.get("thumbnails")),
|
|
"live_broadcast": sn.get("liveBroadcastContent") or "none",
|
|
}
|
|
)
|
|
return {"items": items, "next_page_token": data.get("nextPageToken")}
|
|
|
|
def get_videos(self, video_ids: list[str]) -> list[dict]:
|
|
items: list[dict] = []
|
|
for batch in _chunks(video_ids, 50):
|
|
data = self._get(
|
|
"videos",
|
|
{
|
|
"part": "snippet,contentDetails,statistics,topicDetails,liveStreamingDetails,status",
|
|
"id": ",".join(batch),
|
|
"maxResults": 50,
|
|
},
|
|
)
|
|
items.extend(data.get("items", []))
|
|
return items
|