From 24b6e0026d97fc75109f59d96d867e36337cfa27 Mon Sep 17 00:00:00 2001 From: npeter83 Date: Thu, 11 Jun 2026 01:22:07 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20M2=20(part=201)=20=E2=80=94=20subscript?= =?UTF-8?q?ion=20import,=20YouTube=20client,=20quota=20guard?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Channel/Subscription/Video/ApiQuotaUsage models + migration 0002 - Synchronous YouTube Data API client with OAuth token refresh and per-call quota accounting (API key preferred for public reads when configured) - Subscription import: upsert channels + subscription links, prune unsubscribed, fetch channel details (uploads playlist, topics, stats, handle, country) - Central quota guard tracking units per US-Pacific day against a daily budget - POST /api/sync/subscriptions and GET /api/sync/status endpoints --- backend/alembic/versions/0002_ingest_core.py | 127 +++++++++++++++++ backend/app/config.py | 10 ++ backend/app/main.py | 3 +- backend/app/models.py | 126 ++++++++++++++++- backend/app/quota.py | 50 +++++++ backend/app/routes/sync.py | 40 ++++++ backend/app/sync/__init__.py | 0 backend/app/sync/subscriptions.py | 115 ++++++++++++++++ backend/app/youtube/__init__.py | 0 backend/app/youtube/client.py | 135 +++++++++++++++++++ 10 files changed, 603 insertions(+), 3 deletions(-) create mode 100644 backend/alembic/versions/0002_ingest_core.py create mode 100644 backend/app/quota.py create mode 100644 backend/app/routes/sync.py create mode 100644 backend/app/sync/__init__.py create mode 100644 backend/app/sync/subscriptions.py create mode 100644 backend/app/youtube/__init__.py create mode 100644 backend/app/youtube/client.py diff --git a/backend/alembic/versions/0002_ingest_core.py b/backend/alembic/versions/0002_ingest_core.py new file mode 100644 index 0000000..5c05bad --- /dev/null +++ b/backend/alembic/versions/0002_ingest_core.py @@ -0,0 +1,127 @@ +"""ingest core: channels, subscriptions, videos, api_quota_usage + +Revision ID: 0002_ingest_core +Revises: 0001_initial +Create Date: 2026-06-11 +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + +revision: str = "0002_ingest_core" +down_revision: Union[str, None] = "0001_initial" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "channels", + sa.Column("id", sa.String(length=32), primary_key=True), + sa.Column("title", sa.String(length=255), nullable=True), + sa.Column("handle", sa.String(length=255), nullable=True), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("thumbnail_url", sa.String(length=1024), nullable=True), + sa.Column("uploads_playlist_id", sa.String(length=64), nullable=True), + sa.Column("subscriber_count", sa.BigInteger(), nullable=True), + sa.Column("video_count", sa.BigInteger(), nullable=True), + sa.Column("topic_categories", sa.JSON(), nullable=True), + sa.Column("default_language", sa.String(length=16), nullable=True), + sa.Column("country", sa.String(length=8), nullable=True), + sa.Column("details_synced_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("recent_synced_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_rss_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("backfill_cursor", sa.String(length=128), nullable=True), + sa.Column("backfill_done", sa.Boolean(), nullable=False, server_default="false"), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + ) + + op.create_table( + "subscriptions", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "user_id", + sa.Integer(), + sa.ForeignKey("users.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "channel_id", + sa.String(length=32), + sa.ForeignKey("channels.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("yt_subscription_id", sa.String(length=128), nullable=True), + sa.Column("priority", sa.Integer(), nullable=False, server_default="0"), + sa.Column("hidden", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("subscribed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + sa.UniqueConstraint("user_id", "channel_id", name="uq_user_channel"), + ) + op.create_index("ix_subscriptions_user_id", "subscriptions", ["user_id"]) + op.create_index("ix_subscriptions_channel_id", "subscriptions", ["channel_id"]) + + op.create_table( + "videos", + sa.Column("id", sa.String(length=16), primary_key=True), + sa.Column( + "channel_id", + sa.String(length=32), + sa.ForeignKey("channels.id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column("title", sa.Text(), nullable=True), + sa.Column("description", sa.Text(), nullable=True), + sa.Column("published_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("thumbnail_url", sa.String(length=1024), nullable=True), + sa.Column("duration_seconds", sa.Integer(), nullable=True), + sa.Column("view_count", sa.BigInteger(), nullable=True), + sa.Column("like_count", sa.BigInteger(), nullable=True), + sa.Column("category_id", sa.Integer(), nullable=True), + sa.Column("topic_categories", sa.JSON(), nullable=True), + sa.Column("default_language", sa.String(length=16), nullable=True), + sa.Column("detected_language", sa.String(length=16), nullable=True), + sa.Column("is_short", sa.Boolean(), nullable=False, server_default="false"), + sa.Column("live_status", sa.String(length=16), nullable=False, server_default="none"), + sa.Column("enriched_at", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + server_default=sa.func.now(), + nullable=False, + ), + ) + op.create_index("ix_videos_channel_id", "videos", ["channel_id"]) + op.create_index("ix_videos_published_at", "videos", ["published_at"]) + op.create_index("ix_videos_is_short", "videos", ["is_short"]) + op.create_index("ix_videos_live_status", "videos", ["live_status"]) + + op.create_table( + "api_quota_usage", + sa.Column("day", sa.Date(), primary_key=True), + sa.Column("units_used", sa.Integer(), nullable=False, server_default="0"), + ) + + +def downgrade() -> None: + op.drop_table("api_quota_usage") + op.drop_index("ix_videos_live_status", table_name="videos") + op.drop_index("ix_videos_is_short", table_name="videos") + op.drop_index("ix_videos_published_at", table_name="videos") + op.drop_index("ix_videos_channel_id", table_name="videos") + op.drop_table("videos") + op.drop_index("ix_subscriptions_channel_id", table_name="subscriptions") + op.drop_index("ix_subscriptions_user_id", table_name="subscriptions") + op.drop_table("subscriptions") + op.drop_table("channels") diff --git a/backend/app/config.py b/backend/app/config.py index 25284b5..43ac6d3 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -24,6 +24,16 @@ class Settings(BaseSettings): # Origin of a separately served frontend dev server (enables CORS). Empty in production. frontend_origin: str = "" + # --- Sync / YouTube Data API --- + # Optional API key for public reads (channels/videos/playlistItems). When set it is + # preferred for shared backfill/enrichment so it doesn't depend on a specific user. + youtube_api_key: str = "" + # Daily quota budget (units). Default leaves headroom under the 10,000/day free limit. + quota_daily_budget: int = 9000 + # Recent-first backfill: how far back to fetch on the first pass per channel. + backfill_recent_max_videos: int = 200 + backfill_recent_max_days: int = 365 + @property def allowed_email_set(self) -> set[str]: return {e.strip().lower() for e in self.allowed_emails.split(",") if e.strip()} diff --git a/backend/app/main.py b/backend/app/main.py index 17122a2..f884cfb 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -8,7 +8,7 @@ from starlette.middleware.sessions import SessionMiddleware from app import auth from app.config import settings -from app.routes import health +from app.routes import health, sync app = FastAPI(title=settings.app_name) @@ -30,6 +30,7 @@ if settings.frontend_origin: app.include_router(health.router) app.include_router(auth.router) +app.include_router(sync.router) STATIC_DIR = Path(__file__).parent / "static" app.mount("/assets", StaticFiles(directory=STATIC_DIR / "assets"), name="assets") diff --git a/backend/app/models.py b/backend/app/models.py index 92e8b91..3087abd 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -1,6 +1,18 @@ -from datetime import datetime +from datetime import date, datetime -from sqlalchemy import DateTime, ForeignKey, String, func +from sqlalchemy import ( + BigInteger, + Boolean, + Date, + DateTime, + ForeignKey, + Integer, + JSON, + String, + Text, + UniqueConstraint, + func, +) from sqlalchemy.orm import Mapped, mapped_column, relationship from app.db import Base @@ -22,6 +34,9 @@ class User(Base): token: Mapped["OAuthToken | None"] = relationship( back_populates="user", uselist=False, cascade="all, delete-orphan" ) + subscriptions: Mapped[list["Subscription"]] = relationship( + back_populates="user", cascade="all, delete-orphan" + ) class OAuthToken(Base): @@ -41,3 +56,110 @@ class OAuthToken(Base): ) user: Mapped["User"] = relationship(back_populates="token") + + +class Channel(Base): + """A YouTube channel. Shared across all users (one channel's videos are the same + for everyone), so its expensive metadata is fetched and stored only once.""" + + __tablename__ = "channels" + + id: Mapped[str] = mapped_column(String(32), primary_key=True) # "UC..." channel id + title: Mapped[str | None] = mapped_column(String(255)) + handle: Mapped[str | None] = mapped_column(String(255)) + description: Mapped[str | None] = mapped_column(Text) + thumbnail_url: Mapped[str | None] = mapped_column(String(1024)) + uploads_playlist_id: Mapped[str | None] = mapped_column(String(64)) + subscriber_count: Mapped[int | None] = mapped_column(BigInteger) + video_count: Mapped[int | None] = mapped_column(BigInteger) + topic_categories: Mapped[list | None] = mapped_column(JSON) + default_language: Mapped[str | None] = mapped_column(String(16)) + country: Mapped[str | None] = mapped_column(String(8)) + + # Sync bookkeeping. + details_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + recent_synced_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + last_rss_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + backfill_cursor: Mapped[str | None] = mapped_column(String(128)) + backfill_done: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false" + ) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + videos: Mapped[list["Video"]] = relationship(back_populates="channel") + subscriptions: Mapped[list["Subscription"]] = relationship(back_populates="channel") + + +class Subscription(Base): + """Per-user link to a channel, with the user's own priority/visibility overrides.""" + + __tablename__ = "subscriptions" + __table_args__ = (UniqueConstraint("user_id", "channel_id", name="uq_user_channel"),) + + id: Mapped[int] = mapped_column(primary_key=True) + user_id: Mapped[int] = mapped_column( + ForeignKey("users.id", ondelete="CASCADE"), index=True + ) + channel_id: Mapped[str] = mapped_column( + ForeignKey("channels.id", ondelete="CASCADE"), index=True + ) + # YouTube's subscription resource id, needed to unsubscribe via the API. + yt_subscription_id: Mapped[str | None] = mapped_column(String(128)) + priority: Mapped[int] = mapped_column(Integer, default=0, server_default="0") + hidden: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") + subscribed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + user: Mapped["User"] = relationship(back_populates="subscriptions") + channel: Mapped["Channel"] = relationship(back_populates="subscriptions") + + +class Video(Base): + """A YouTube video. Shared across users; per-user state lives elsewhere.""" + + __tablename__ = "videos" + + id: Mapped[str] = mapped_column(String(16), primary_key=True) # video id + channel_id: Mapped[str] = mapped_column( + ForeignKey("channels.id", ondelete="CASCADE"), index=True + ) + title: Mapped[str | None] = mapped_column(Text) + description: Mapped[str | None] = mapped_column(Text) + published_at: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), index=True + ) + thumbnail_url: Mapped[str | None] = mapped_column(String(1024)) + duration_seconds: Mapped[int | None] = mapped_column(Integer) + view_count: Mapped[int | None] = mapped_column(BigInteger) + like_count: Mapped[int | None] = mapped_column(BigInteger) + category_id: Mapped[int | None] = mapped_column(Integer) + topic_categories: Mapped[list | None] = mapped_column(JSON) + default_language: Mapped[str | None] = mapped_column(String(16)) + detected_language: Mapped[str | None] = mapped_column(String(16)) + is_short: Mapped[bool] = mapped_column( + Boolean, default=False, server_default="false", index=True + ) + # none | live | upcoming | premiere | was_live + live_status: Mapped[str] = mapped_column( + String(16), default="none", server_default="none", index=True + ) + enriched_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now() + ) + + channel: Mapped["Channel"] = relationship(back_populates="videos") + + +class ApiQuotaUsage(Base): + """Tracks YouTube Data API units spent per Pacific-time day (the quota resets at + midnight Pacific). The whole app shares one daily budget.""" + + __tablename__ = "api_quota_usage" + + day: Mapped[date] = mapped_column(Date, primary_key=True) + units_used: Mapped[int] = mapped_column(Integer, default=0, server_default="0") diff --git a/backend/app/quota.py b/backend/app/quota.py new file mode 100644 index 0000000..b2acbd8 --- /dev/null +++ b/backend/app/quota.py @@ -0,0 +1,50 @@ +"""Central YouTube Data API quota guard. + +The whole app shares one daily budget (the API resets at midnight US Pacific time). +Steady-state work (RSS detection is free; enrichment of new videos is cheap) should +always be allowed; expensive backfill must yield to the remaining budget. +""" +from datetime import datetime +from zoneinfo import ZoneInfo + +from sqlalchemy.dialects.postgresql import insert as pg_insert +from sqlalchemy.orm import Session + +from app.config import settings +from app.models import ApiQuotaUsage + +_PACIFIC = ZoneInfo("America/Los_Angeles") + + +def pacific_today(): + return datetime.now(_PACIFIC).date() + + +def units_used_today(db: Session) -> int: + row = db.get(ApiQuotaUsage, pacific_today()) + return row.units_used if row else 0 + + +def remaining_today(db: Session) -> int: + return max(0, settings.quota_daily_budget - units_used_today(db)) + + +def can_spend(db: Session, units: int) -> bool: + return remaining_today(db) >= units + + +def record_usage(db: Session, units: int) -> None: + """Atomically add `units` to today's counter (upsert).""" + if units <= 0: + return + day = pacific_today() + stmt = ( + pg_insert(ApiQuotaUsage) + .values(day=day, units_used=units) + .on_conflict_do_update( + index_elements=[ApiQuotaUsage.day], + set_={"units_used": ApiQuotaUsage.units_used + units}, + ) + ) + db.execute(stmt) + db.commit() diff --git a/backend/app/routes/sync.py b/backend/app/routes/sync.py new file mode 100644 index 0000000..9f4fb8a --- /dev/null +++ b/backend/app/routes/sync.py @@ -0,0 +1,40 @@ +from fastapi import APIRouter, Depends +from sqlalchemy import func, select +from sqlalchemy.orm import Session + +from app import quota +from app.auth import current_user +from app.db import get_db +from app.models import Channel, Subscription, User, Video +from app.sync.subscriptions import import_subscriptions + +router = APIRouter(prefix="/api/sync", tags=["sync"]) + + +@router.post("/subscriptions") +def sync_subscriptions( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + before = quota.units_used_today(db) + result = import_subscriptions(db, user) + result["quota_used_estimate"] = quota.units_used_today(db) - before + result["quota_remaining_today"] = quota.remaining_today(db) + return result + + +@router.get("/status") +def sync_status( + user: User = Depends(current_user), db: Session = Depends(get_db) +) -> dict: + sub_count = db.scalar( + select(func.count()) + .select_from(Subscription) + .where(Subscription.user_id == user.id) + ) + return { + "subscriptions": sub_count, + "channels_total": db.scalar(select(func.count()).select_from(Channel)), + "videos_total": db.scalar(select(func.count()).select_from(Video)), + "quota_used_today": quota.units_used_today(db), + "quota_remaining_today": quota.remaining_today(db), + } diff --git a/backend/app/sync/__init__.py b/backend/app/sync/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/sync/subscriptions.py b/backend/app/sync/subscriptions.py new file mode 100644 index 0000000..653eba1 --- /dev/null +++ b/backend/app/sync/subscriptions.py @@ -0,0 +1,115 @@ +"""Import the authenticated user's YouTube subscriptions and channel metadata.""" +from datetime import datetime, timezone + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models import Channel, Subscription, User +from app.youtube.client import YouTubeClient, best_thumbnail + + +def _to_int(value) -> int | None: + try: + return int(value) if value is not None else None + except (TypeError, ValueError): + return None + + +def apply_channel_details(db: Session, items: list[dict]) -> None: + now = datetime.now(timezone.utc) + for item in items: + channel = db.get(Channel, item["id"]) + if channel is None: + continue + snippet = item.get("snippet", {}) + content = item.get("contentDetails", {}) + stats = item.get("statistics", {}) + topics = item.get("topicDetails", {}) + + channel.title = snippet.get("title") or channel.title + channel.handle = snippet.get("customUrl") + channel.description = snippet.get("description") + channel.thumbnail_url = ( + best_thumbnail(snippet.get("thumbnails")) or channel.thumbnail_url + ) + channel.default_language = snippet.get("defaultLanguage") + channel.country = snippet.get("country") + channel.uploads_playlist_id = content.get("relatedPlaylists", {}).get("uploads") + channel.subscriber_count = _to_int(stats.get("subscriberCount")) + channel.video_count = _to_int(stats.get("videoCount")) + channel.topic_categories = topics.get("topicCategories") + channel.details_synced_at = now + + +def import_subscriptions(db: Session, user: User) -> dict: + """Pull all subscriptions for `user`, upsert channels + subscription links, prune + channels the user has unsubscribed from on YouTube, and fetch channel details for + any channel we don't have details for yet.""" + fetched: dict[str, dict] = {} + new_channels = 0 + + with YouTubeClient(db, user) as yt: + for sub in yt.iter_subscriptions(): + cid = sub.get("channel_id") + if cid: + fetched[cid] = sub + + for cid, sub in fetched.items(): + channel = db.get(Channel, cid) + if channel is None: + channel = Channel( + id=cid, + title=sub.get("title"), + description=sub.get("description"), + thumbnail_url=sub.get("thumbnail_url"), + ) + db.add(channel) + new_channels += 1 + sub_row = db.execute( + select(Subscription).where( + Subscription.user_id == user.id, Subscription.channel_id == cid + ) + ).scalar_one_or_none() + if sub_row is None: + sub_row = Subscription(user_id=user.id, channel_id=cid) + db.add(sub_row) + sub_row.yt_subscription_id = sub.get("yt_subscription_id") + db.flush() + + # Prune subscriptions removed on YouTube. + removed = 0 + existing = ( + db.execute(select(Subscription).where(Subscription.user_id == user.id)) + .scalars() + .all() + ) + for sub_row in existing: + if sub_row.channel_id not in fetched: + db.delete(sub_row) + removed += 1 + db.commit() + + # Fetch details for channels that don't have them yet. + need_details = ( + db.execute( + select(Channel).where( + Channel.id.in_(fetched.keys()), + Channel.details_synced_at.is_(None), + ) + ) + .scalars() + .all() + ) + detailed = 0 + if need_details: + items = yt.get_channels([c.id for c in need_details]) + apply_channel_details(db, items) + detailed = len(items) + db.commit() + + return { + "subscriptions": len(fetched), + "channels_new": new_channels, + "channels_detailed": detailed, + "removed_stale": removed, + } diff --git a/backend/app/youtube/__init__.py b/backend/app/youtube/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/backend/app/youtube/client.py b/backend/app/youtube/client.py new file mode 100644 index 0000000..b1902a1 --- /dev/null +++ b/backend/app/youtube/client.py @@ -0,0 +1,135 @@ +"""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. +""" +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 + +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() + 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: + 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 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