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, sysconfig
from app.config import settings
from app.models import User
from app.security import decrypt

log = logging.getLogger("siftlode.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
        # 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): callers inspect resp.status_code
        themselves and raise their own YouTubeError with the response body."""
        try:
            return self._http.request(method, url, **kwargs)
        except httpx.HTTPError as exc:
            raise YouTubeError(
                f"YouTube API unreachable ({exc.__class__.__name__}: {exc})"
            ) 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)
        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._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:
            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 = {}
        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(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_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(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 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)."""
        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._send(
            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 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