79d645c2e2
Add scheduler_settings (per-job interval override, migration 0015) and a
PATCH /api/admin/scheduler/jobs/{id} that persists the override and live-
reschedules the running APScheduler job. Intervals load from the DB (env
defaults as fallback); the snapshot reports the live trigger interval.
110 lines
3.7 KiB
Python
110 lines
3.7 KiB
Python
"""Admin Scheduler dashboard: a live view of what the background scheduler is doing —
|
|
per-job run activity (from the in-process scheduler), the work still queued (DB-derived),
|
|
and the shared quota picture."""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy import and_, func, or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import quota, state
|
|
from app.config import settings
|
|
from app.db import get_db
|
|
from app.models import Channel, Subscription, User, Video
|
|
from app.routes.admin import admin_user
|
|
from app.scheduler import JOB_INTERVALS, MAX_INTERVAL, MIN_INTERVAL, apply_interval, scheduler_snapshot
|
|
from app.sync.runner import estimate_deep_backfill
|
|
|
|
router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"])
|
|
|
|
|
|
@router.patch("/jobs/{job_id}")
|
|
def set_job_interval(
|
|
job_id: str,
|
|
payload: dict,
|
|
_: User = Depends(admin_user),
|
|
db: Session = Depends(get_db),
|
|
) -> dict:
|
|
"""Change how often a scheduler job runs (minutes). Persisted as an override and applied
|
|
to the live scheduler immediately."""
|
|
if job_id not in JOB_INTERVALS:
|
|
raise HTTPException(status_code=404, detail="Unknown job")
|
|
try:
|
|
minutes = int(payload.get("interval_minutes"))
|
|
except (TypeError, ValueError):
|
|
raise HTTPException(status_code=400, detail="interval_minutes must be a number")
|
|
if not (MIN_INTERVAL <= minutes <= MAX_INTERVAL):
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"interval must be between {MIN_INTERVAL} and {MAX_INTERVAL} minutes",
|
|
)
|
|
applied = apply_interval(db, job_id, minutes)
|
|
return {"id": job_id, "interval_minutes": applied}
|
|
|
|
|
|
@router.get("")
|
|
def get_scheduler(
|
|
_: User = Depends(admin_user), db: Session = Depends(get_db)
|
|
) -> dict:
|
|
snap = scheduler_snapshot()
|
|
|
|
# Work still queued (shared catalog, so these are instance-wide, not per-user).
|
|
deep_requested = (
|
|
select(Subscription.id)
|
|
.where(
|
|
Subscription.channel_id == Channel.id,
|
|
Subscription.deep_requested.is_(True),
|
|
)
|
|
.exists()
|
|
)
|
|
deep_channels = (
|
|
db.execute(
|
|
select(Channel).where(Channel.backfill_done.is_(False), deep_requested)
|
|
)
|
|
.scalars()
|
|
.all()
|
|
)
|
|
deep_eta = estimate_deep_backfill(db, deep_channels)
|
|
|
|
queue = {
|
|
"channels_recent_pending": db.scalar(
|
|
select(func.count()).select_from(Channel).where(Channel.recent_synced_at.is_(None))
|
|
)
|
|
or 0,
|
|
"channels_deep_pending": deep_eta["channels_pending"],
|
|
"deep_eta_seconds": deep_eta["eta_seconds"],
|
|
"videos_pending_enrich": db.scalar(
|
|
select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
|
|
)
|
|
or 0,
|
|
"videos_pending_shorts": db.scalar(
|
|
select(func.count())
|
|
.select_from(Video)
|
|
.where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None))
|
|
)
|
|
or 0,
|
|
"videos_live_refresh": db.scalar(
|
|
select(func.count())
|
|
.select_from(Video)
|
|
.where(
|
|
or_(
|
|
Video.live_status.in_(("live", "upcoming")),
|
|
and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)),
|
|
)
|
|
)
|
|
)
|
|
or 0,
|
|
}
|
|
|
|
used = quota.units_used_today(db)
|
|
quota_info = {
|
|
"used_today": used,
|
|
"remaining_today": quota.remaining_today(db),
|
|
"daily_budget": settings.quota_daily_budget,
|
|
"backfill_reserve": settings.backfill_quota_reserve,
|
|
}
|
|
|
|
return {
|
|
**snap,
|
|
"paused": state.is_sync_paused(db),
|
|
"queue": queue,
|
|
"quota": quota_info,
|
|
}
|