02d913f133
Record each scheduler job's run activity in-process (running / last result / error / timestamps) and expose a snapshot. New admin-only GET /api/admin/scheduler returns the per-job activity (with APScheduler next-run times), the DB-derived work still queued (channels/videos pending, deep ETA, live-refresh backlog) and the shared quota picture.
215 lines
7 KiB
Python
215 lines
7 KiB
Python
"""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
|
|
import threading
|
|
from datetime import datetime, timezone
|
|
|
|
from apscheduler.schedulers.background import BackgroundScheduler
|
|
|
|
from app import quota
|
|
from app.config import settings
|
|
from app.db import SessionLocal
|
|
from app.state import is_sync_paused
|
|
from app.sync.autotag import run_autotag_all
|
|
from app.sync.playlists import sync_all_playlists
|
|
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
|
|
|
|
# Per-job run activity, kept in-memory for the admin Scheduler dashboard to read (the
|
|
# scheduler lives in the same single-worker process as the API, so a request can read it
|
|
# directly). Ephemeral by design — it reflects "what the scheduler is doing right now and
|
|
# how the last runs went" since this process started; durable history isn't the goal here.
|
|
_activity: dict[str, dict] = {}
|
|
_activity_lock = threading.Lock()
|
|
|
|
# Each interval job's configured period, surfaced so the dashboard can show "every N min".
|
|
JOB_INTERVALS: dict[str, int] = {
|
|
"rss_poll": settings.rss_poll_minutes,
|
|
"enrich": settings.enrich_interval_minutes,
|
|
"backfill": settings.backfill_interval_minutes,
|
|
"autotag": settings.autotag_interval_minutes,
|
|
"shorts": settings.shorts_probe_interval_minutes,
|
|
"subscriptions": settings.subscriptions_resync_minutes,
|
|
"playlist_sync": settings.playlist_sync_minutes,
|
|
}
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).isoformat()
|
|
|
|
|
|
def _record(name: str, **fields) -> None:
|
|
with _activity_lock:
|
|
_activity.setdefault(name, {}).update(fields)
|
|
|
|
|
|
def _job(name: str, fn) -> None:
|
|
db = SessionLocal()
|
|
_record(name, running=True, last_started=_now_iso(), last_error=None)
|
|
try:
|
|
if is_sync_paused(db):
|
|
logger.info("job %s skipped (sync paused)", name)
|
|
_record(name, running=False, status="skipped", last_finished=_now_iso(),
|
|
last_result="paused")
|
|
return
|
|
result = fn(db)
|
|
logger.info("job %s -> %s", name, result)
|
|
_record(name, running=False, status="ok", last_finished=_now_iso(),
|
|
last_result=_summarize(result))
|
|
except Exception as exc:
|
|
db.rollback()
|
|
logger.exception("job %s failed", name)
|
|
_record(name, running=False, status="error", last_finished=_now_iso(),
|
|
last_error=str(exc) or exc.__class__.__name__)
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def _summarize(result) -> str | None:
|
|
"""Compact one-line rendering of a job's return value for the dashboard."""
|
|
if result is None:
|
|
return None
|
|
if isinstance(result, dict):
|
|
return ", ".join(f"{k}={v}" for k, v in result.items()) or None
|
|
return str(result)
|
|
|
|
|
|
def scheduler_snapshot() -> dict:
|
|
"""What the scheduler is doing, for the admin dashboard. `running_here` is False in
|
|
environments where the scheduler is disabled (e.g. local dev shares the prod-ish DB but
|
|
runs no scheduler), so the UI can say so while still showing DB-derived queue/quota."""
|
|
running_here = _scheduler is not None
|
|
next_runs: dict[str, str | None] = {}
|
|
if _scheduler is not None:
|
|
for job in _scheduler.get_jobs():
|
|
nr = getattr(job, "next_run_time", None)
|
|
next_runs[job.id] = nr.astimezone(timezone.utc).isoformat() if nr else None
|
|
with _activity_lock:
|
|
acts = {k: dict(v) for k, v in _activity.items()}
|
|
jobs = []
|
|
for job_id, interval in JOB_INTERVALS.items():
|
|
a = acts.get(job_id, {})
|
|
jobs.append(
|
|
{
|
|
"id": job_id,
|
|
"interval_minutes": interval,
|
|
"next_run": next_runs.get(job_id),
|
|
"running": a.get("running", False),
|
|
"status": a.get("status"),
|
|
"last_started": a.get("last_started"),
|
|
"last_finished": a.get("last_finished"),
|
|
"last_result": a.get("last_result"),
|
|
"last_error": a.get("last_error"),
|
|
}
|
|
)
|
|
return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}
|
|
|
|
|
|
def _rss_job() -> None:
|
|
_job("rss_poll", lambda db: run_rss_poll(db))
|
|
|
|
|
|
def _enrich_job() -> None:
|
|
def work(db):
|
|
with quota.attribute(None, "enrich"):
|
|
return run_enrich(db)
|
|
|
|
_job("enrich", work)
|
|
|
|
|
|
def _backfill_job() -> None:
|
|
# Recent-first for not-yet-synced channels, then deep backfill for the rest. All
|
|
# background spend is attributed to the system (no actor), split by action.
|
|
def work(db):
|
|
with quota.attribute(None, "backfill_recent"):
|
|
recent = run_recent_backfill(db, max_channels=25)
|
|
with quota.attribute(None, "backfill_deep"):
|
|
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:
|
|
def work(db):
|
|
with quota.attribute(None, "subscription_resync"):
|
|
return run_subscription_resync(db)
|
|
|
|
_job("subscriptions", work)
|
|
|
|
|
|
def _playlist_sync_job() -> None:
|
|
# Mirror each read-scope user's YouTube playlists; per-user quota attribution is
|
|
# handled inside sync_all_playlists.
|
|
_job("playlist_sync", sync_all_playlists)
|
|
|
|
|
|
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.add_job(
|
|
_playlist_sync_job,
|
|
"interval",
|
|
minutes=settings.playlist_sync_minutes,
|
|
id="playlist_sync",
|
|
)
|
|
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
|