"""Background scheduler: free RSS detection, enrichment of new videos, and quota-aware recent-first / deep backfill. Runs inside the API process (single worker).""" import logging from apscheduler.schedulers.background import BackgroundScheduler from app.config import settings from app.db import SessionLocal from app.sync.autotag import run_autotag_all from app.sync.runner import ( run_deep_backfill, run_enrich, run_recent_backfill, run_rss_poll, run_shorts, run_subscription_resync, ) logger = logging.getLogger("subfeed.scheduler") _scheduler: BackgroundScheduler | None = None def _job(name: str, fn) -> None: db = SessionLocal() try: result = fn(db) logger.info("job %s -> %s", name, result) except Exception: db.rollback() logger.exception("job %s failed", name) finally: db.close() def _rss_job() -> None: _job("rss_poll", lambda db: run_rss_poll(db)) def _enrich_job() -> None: _job("enrich", lambda db: run_enrich(db)) def _backfill_job() -> None: # Recent-first for not-yet-synced channels, then deep backfill for the rest. def work(db): recent = run_recent_backfill(db, max_channels=25) deep = run_deep_backfill(db, max_channels=10) return {"recent": recent, "deep": deep} _job("backfill", work) def _autotag_job() -> None: _job("autotag", lambda db: run_autotag_all(db, only_missing=True)) def _shorts_job() -> None: _job("shorts", run_shorts) def _subscriptions_job() -> None: _job("subscriptions", run_subscription_resync) def start_scheduler() -> None: global _scheduler if not settings.scheduler_enabled or _scheduler is not None: return scheduler = BackgroundScheduler(timezone="UTC") scheduler.add_job( _rss_job, "interval", minutes=settings.rss_poll_minutes, id="rss_poll" ) scheduler.add_job( _enrich_job, "interval", minutes=settings.enrich_interval_minutes, id="enrich" ) scheduler.add_job( _backfill_job, "interval", minutes=settings.backfill_interval_minutes, id="backfill", ) scheduler.add_job( _autotag_job, "interval", minutes=settings.autotag_interval_minutes, id="autotag", ) scheduler.add_job( _shorts_job, "interval", minutes=settings.shorts_probe_interval_minutes, id="shorts", ) scheduler.add_job( _subscriptions_job, "interval", minutes=settings.subscriptions_resync_minutes, id="subscriptions", ) scheduler.start() _scheduler = scheduler logger.info("scheduler started") def shutdown_scheduler() -> None: global _scheduler if _scheduler is not None: _scheduler.shutdown(wait=False) _scheduler = None