siftlode/backend/app/youtube/client.py

"""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 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 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",
                    "id": ",".join(batch),
                    "maxResults": 50,
                },
            )
            items.extend(data.get("items", []))
        return items