siftlode/backend/app/sync/runner.py

"""Reusable sync routines shared by the scheduler and the manual /api/sync endpoints.

Channel/video data is shared across users, so background work acts through a "service
user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configured for
public reads.
"""
import logging

from sqlalchemy import select
from sqlalchemy.orm import Session

from app import quota
from app.config import settings

log = logging.getLogger("subfeed.sync")
from app.models import Channel, OAuthToken, User
from app.sync.subscriptions import import_subscriptions
from app.sync.videos import (
    backfill_channel_deep,
    backfill_channel_recent,
    enrich_pending,
    poll_rss_channel,
    run_shorts_classification,
)
from app.youtube.client import YouTubeClient


def get_service_user(db: Session) -> User | None:
    return (
        db.execute(
            select(User)
            .join(OAuthToken)
            .where(OAuthToken.refresh_token_enc.is_not(None))
            .order_by(User.id)
        )
        .scalars()
        .first()
    )


def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int:
    if channels is None:
        channels = db.execute(select(Channel)).scalars().all()
    new = 0
    for channel in channels:
        try:
            new += poll_rss_channel(db, channel)
        except Exception:
            db.rollback()
            log.exception("RSS poll failed for channel %s", channel.id)
    return new


def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
    user = get_service_user(db)
    if user is None:
        return 0
    total = 0
    with YouTubeClient(db, user) as yt:
        for _ in range(max_batches):
            if quota.remaining_today(db) <= floor:
                break
            n = enrich_pending(db, yt)
            total += n
            if n == 0:
                break
    return total


def run_shorts(db: Session) -> dict:
    return run_shorts_classification(db)


def run_subscription_resync(db: Session) -> dict:
    """Re-import every user's subscriptions so unsubscribes and new subscriptions on
    YouTube are reflected automatically."""
    users = (
        db.execute(
            select(User)
            .join(OAuthToken)
            .where(OAuthToken.refresh_token_enc.is_not(None))
        )
        .scalars()
        .all()
    )
    for user in users:
        try:
            import_subscriptions(db, user)
        except Exception:
            db.rollback()
            log.exception("Subscription resync failed for user %s", user.id)
    return {"users": len(users)}


def run_recent_backfill(
    db: Session, channels: list[Channel] | None = None, max_channels: int | None = None
) -> dict:
    user = get_service_user(db)
    if user is None:
        return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
    if channels is None:
        channels = (
            db.execute(select(Channel).where(Channel.recent_synced_at.is_(None)))
            .scalars()
            .all()
        )
    if max_channels:
        channels = channels[:max_channels]
    processed = 0
    videos_added = 0
    with YouTubeClient(db, user) as yt:
        for channel in channels:
            if quota.remaining_today(db) <= settings.backfill_quota_reserve:
                break
            try:
                videos_added += backfill_channel_recent(db, yt, channel)
                processed += 1
            except Exception:
                db.rollback()
                log.exception("Recent backfill failed for channel %s", channel.id)
    return {"channels_processed": processed, "videos_added": videos_added}


def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -> dict:
    user = get_service_user(db)
    if user is None:
        return {"channels_processed": 0, "videos_added": 0, "reason": "no service user"}
    channels = (
        db.execute(
            select(Channel).where(
                Channel.backfill_done.is_(False),
                Channel.recent_synced_at.is_not(None),
            )
        )
        .scalars()
        .all()
    )[:max_channels]
    processed = 0
    videos_added = 0
    with YouTubeClient(db, user) as yt:
        for channel in channels:
            if quota.remaining_today(db) <= settings.backfill_quota_reserve:
                break
            try:
                videos_added += backfill_channel_deep(db, yt, channel, max_pages)
                processed += 1
            except Exception:
                db.rollback()
                log.exception("Deep backfill failed for channel %s", channel.id)
    return {"channels_processed": processed, "videos_added": videos_added}