"""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. """ from sqlalchemy import select from sqlalchemy.orm import Session from app import quota from app.config import settings from app.models import Channel, OAuthToken, User 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() 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_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() 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() return {"channels_processed": processed, "videos_added": videos_added}