siftlode/backend/app/sync/subscriptions.py

"""Import the authenticated user's YouTube subscriptions and channel metadata."""
import logging
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

log = logging.getLogger("siftlode.sync")


def _to_int(value) -> int | None:
    try:
        return int(value) if value is not None else None
    except (TypeError, ValueError):
        return None


def _parse_dt(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


def _external_links(branding: dict) -> list | None:
    """Channel "About" links. The API exposes a few in brandingSettings.channel.profileColor/
    /... varies; the durable spot is brandingSettings.channel.links is gone post-2023, so we
    read what's present (settings/links) defensively and keep [{title, url}] pairs."""
    channel = branding.get("channel", {})
    links = channel.get("links") or []
    out = [
        {"title": item.get("title"), "url": item.get("url")}
        for item in links
        if isinstance(item, dict) and item.get("url")
    ]
    return out or 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", {})
        branding = item.get("brandingSettings", {})

        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.total_view_count = _to_int(stats.get("viewCount"))
        channel.published_at = _parse_dt(snippet.get("publishedAt"))
        channel.banner_url = branding.get("image", {}).get("bannerExternalUrl")
        channel.external_links = _external_links(branding)
        channel.topic_categories = topics.get("topicCategories")
        channel.keywords = branding.get("channel", {}).get("keywords")
        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()

    result = {
        "subscriptions": len(fetched),
        "channels_new": new_channels,
        "channels_detailed": detailed,
        "removed_stale": removed,
    }
    log.info("Subscription import (user %s): %s", user.id, result)
    return result