c4aecf9f77
Add a daily maintenance/validation job that detects videos which can't be played anywhere and retires them safely. Two phases: re-check already-flagged videos (recover if available again, else hard-delete once the grace period elapses, cascading to states/playlist items), and a rolling re-validation of the least-recently-checked currently-available videos that flags newly-unplayable ones (hidden from the feed immediately via unavailable_since). Detection is ~free: a video missing from the videos.list response is deleted-or-private; an `upcoming` premiere >2 days past its scheduled start that never went live is abandoned. A still-live broadcast is kept (legit 24/7 stream). Enrichment now also fetches part=status to populate the status columns. Grace is 7 days for removed videos, none for abandoned. Before deleting, affected users get one batched notification (never per-video). Interval is admin-tunable via the Scheduler dashboard; batch size and grace are config. Quota-attributed to the system and bounded by the same backfill reserve as the other jobs.
336 lines
13 KiB
Python
336 lines
13 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
|
|
|
|
import httpx
|
|
|
|
from app import quota
|
|
from app.config import settings
|
|
from app.models import User
|
|
from app.security import decrypt
|
|
|
|
log = logging.getLogger("subfeed.youtube")
|
|
|
|
GOOGLE_TOKEN_URL = "https://oauth2.googleapis.com/token"
|
|
API_BASE = "https://www.googleapis.com/youtube/v3"
|
|
|
|
|
|
class YouTubeError(Exception):
|
|
pass
|
|
|
|
|
|
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
|
|
self._http = httpx.Client(timeout=30.0)
|
|
|
|
def __enter__(self) -> "YouTubeClient":
|
|
return self
|
|
|
|
def __exit__(self, *exc) -> None:
|
|
self._http.close()
|
|
|
|
# --- 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)
|
|
if tok.access_token and tok.expiry and tok.expiry > now + timedelta(seconds=60):
|
|
return tok.access_token
|
|
refresh = decrypt(tok.refresh_token_enc)
|
|
if not refresh:
|
|
raise YouTubeError("No refresh token; user must re-authenticate")
|
|
resp = self._http.post(
|
|
GOOGLE_TOKEN_URL,
|
|
data={
|
|
"client_id": settings.google_client_id,
|
|
"client_secret": settings.google_client_secret,
|
|
"refresh_token": refresh,
|
|
"grant_type": "refresh_token",
|
|
},
|
|
)
|
|
if resp.status_code != 200:
|
|
raise YouTubeError(f"Token refresh failed: {resp.status_code} {resp.text[:200]}")
|
|
data = resp.json()
|
|
tok.access_token = data["access_token"]
|
|
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 tok.access_token
|
|
|
|
# --- core request ---
|
|
def _get(self, path: str, params: dict, cost: int = 1, allow_key: bool = True) -> dict:
|
|
p = dict(params)
|
|
headers = {}
|
|
if allow_key and settings.youtube_api_key:
|
|
p["key"] = settings.youtube_api_key
|
|
else:
|
|
headers["Authorization"] = f"Bearer {self._access_token()}"
|
|
resp = self._http.get(f"{API_BASE}/{path}", params=p, headers=headers)
|
|
quota.record_usage(self.db, cost)
|
|
if resp.status_code != 200:
|
|
log.warning("YouTube API %s -> %s: %s", path, resp.status_code, resp.text[:200])
|
|
raise YouTubeError(f"GET {path} -> {resp.status_code}: {resp.text[:300]}")
|
|
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
|
|
data = self._get("playlists", params, cost=1, allow_key=False)
|
|
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_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)."""
|
|
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)
|
|
for item in data.get("items", []):
|
|
vid = item.get("contentDetails", {}).get("videoId")
|
|
if vid:
|
|
yield vid
|
|
page_token = data.get("nextPageToken")
|
|
if not page_token:
|
|
break
|
|
|
|
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",
|
|
"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._http.delete(
|
|
f"{API_BASE}/subscriptions",
|
|
params={"id": subscription_id},
|
|
headers={"Authorization": f"Bearer {self._access_token()}"},
|
|
)
|
|
quota.record_usage(self.db, 50)
|
|
if resp.status_code not in (200, 204):
|
|
log.warning(
|
|
"YouTube subscriptions.delete -> %s: %s",
|
|
resp.status_code,
|
|
resp.text[:200],
|
|
)
|
|
raise YouTubeError(
|
|
f"DELETE subscriptions -> {resp.status_code}: {resp.text[:300]}"
|
|
)
|
|
|
|
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)."""
|
|
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)
|
|
for item in data.get("items", []):
|
|
vid = item.get("contentDetails", {}).get("videoId")
|
|
if vid:
|
|
yield {"item_id": item.get("id"), "video_id": vid}
|
|
page_token = data.get("nextPageToken")
|
|
if not page_token:
|
|
break
|
|
|
|
# --- 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._http.request(
|
|
method, f"{API_BASE}/{path}", params=params, json=json, headers=headers
|
|
)
|
|
quota.record_usage(self.db, 50)
|
|
if resp.status_code not in (200, 204):
|
|
log.warning("YouTube %s %s -> %s: %s", method, path, resp.status_code, resp.text[:200])
|
|
raise YouTubeError(f"{method} {path} -> {resp.status_code}: {resp.text[:300]}")
|
|
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 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
|