siftlode/backend/app/sync/subscriptions.py

"""Import the authenticated user's YouTube subscriptions and channel metadata."""

from collections.abc import Collection, Iterable
from datetime import datetime, timedelta, timezone

from sqlalchemy import or_, select
from sqlalchemy.orm import Session

from app import logs
from app.models import Channel, Subscription, User
from app.utils import normalize_description
from app.youtube.client import YouTubeClient, YouTubeError, best_thumbnail

log = logs.get_logger(__name__)


def _to_int(value) -> int | None:
    try:
        return int(value) if value is not None else None
    # silent-ok: an upstream count we cannot read is stored as no count, which the card renders as
    # absent rather than as zero. Per-channel over a whole subscription list, so it must stay quiet.
    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"))
    # silent-ok: as `_to_int` above — an unreadable upstream date becomes NULL, and every caller
    # already treats a missing date as unknown.
    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


# How old a channel's metadata may get before it is fetched again. Banners, avatars, descriptions and
# counts change on YouTube, and an old banner URL stops resolving at all — without a refresh a channel
# kept whatever it had the day it was first seen. A week keeps counts roughly current at almost no
# cost: `channels.list` is 1 quota unit per 50 channels, and only the stale ones are asked for.
DETAILS_MAX_AGE = timedelta(days=7)


def details_stale(channel: Channel, now: datetime) -> bool:
    """Whether `channel`'s metadata is missing or older than DETAILS_MAX_AGE (see `details_stale_clause`)."""
    synced = channel.details_synced_at
    return synced is None or synced < now - DETAILS_MAX_AGE


def details_stale_clause(now: datetime):
    """`details_stale` as a WHERE clause, so the batch import and a single channel agree on "stale"."""
    return or_(
        Channel.details_synced_at.is_(None),
        Channel.details_synced_at < now - DETAILS_MAX_AGE,
    )


def apply_channel_details(db: Session, items: list[dict], asked: Iterable[str]) -> None:
    """Write `channels.list`'s answer onto the rows. `asked` is every id the call asked for: one
    YouTube did not return (terminated, deleted) is stamped as synced too — otherwise it stays
    stale forever and every view / import asks for it again, spending quota on a known miss."""
    now = datetime.now(timezone.utc)
    returned = {item["id"] for item in items}
    for cid in set(asked) - returned:
        missing = db.get(Channel, cid)
        if missing is not None:
            missing.details_synced_at = now
    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 = normalize_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") or {}).get("keywords")
        channel.details_synced_at = now


def should_prune(fetched_count: int, existing_count: int) -> bool:
    """Whether an import may delete the local subscriptions missing from this fetch.

    NEVER on an empty fetch that contradicts local state: the API can answer 200 with no items (a
    transient blip, a scope/token edge), and pruning against that wipes every subscription the user
    has — taking each row's `deep_requested` flag with it — in a single run. Someone who genuinely
    unsubscribed from everything simply keeps their rows until a sync that returns something.

    Pure so it can be tested without a database; its callers are `import_subscriptions` and, through
    `vanished`, every mirror that drops what the remote no longer lists."""
    return fetched_count > 0 or existing_count == 0


def vanished[K, V](existing: dict[K, V], listed: Collection[K]) -> list[V]:
    """The local mirrors whose key the remote listing no longer has — none at all when the listing
    is an empty answer that contradicts them (`should_prune`). The YouTube playlist mirrors use it:
    without it, one blip dropped every mirror."""
    if not should_prune(len(listed), len(existing)):
        return []
    return [row for key, row in existing.items() if key not in listed]


def blank_listing(listed: Collection, count: object) -> bool:
    """The same rule one level down: no members from a container whose own listing counts some (a
    playlist's `itemCount`) is a blip, not a list emptied. Replacing or merging on it deleted every
    mirrored item and reset the merge baseline."""
    return not listed and isinstance(count, int) and count > 0


def refuse_blank_listing(listed: Collection, count: object) -> None:
    """`blank_listing` for a user-started pull or push: refused as retryable ("try again")."""
    if blank_listing(listed, count):
        raise YouTubeError(
            "YouTube listed no items for a playlist it counts as non-empty",
            reason="emptyAnswer",
            path="playlistItems",
        )


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 whose details are missing or stale (`DETAILS_MAX_AGE`)."""
    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=normalize_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 — but only when this fetch is trustworthy
        # (see `should_prune`).
        removed = 0
        existing = (
            db.execute(select(Subscription).where(Subscription.user_id == user.id)).scalars().all()
        )
        prune_skipped = not should_prune(len(fetched), len(existing))
        if prune_skipped:
            log.warning(
                "Subscription import (user %s): YouTube returned 0 subscriptions while %d exist "
                "locally — skipping the prune (treating it as a transient empty response)",
                user.id,
                len(existing),
            )
        else:
            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, or had them too long ago.
        need_details = (
            db.execute(
                select(Channel).where(
                    Channel.id.in_(fetched.keys()),
                    details_stale_clause(datetime.now(timezone.utc)),
                )
            )
            .scalars()
            .all()
        )
        detailed = 0
        if need_details:
            ids = [c.id for c in need_details]
            items = yt.get_channels(ids)
            apply_channel_details(db, items, ids)
            detailed = len(items)
        db.commit()

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