"""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}