diff --git a/backend/alembic/versions/0015_scheduler_settings.py b/backend/alembic/versions/0015_scheduler_settings.py new file mode 100644 index 0000000..5a7f590 --- /dev/null +++ b/backend/alembic/versions/0015_scheduler_settings.py @@ -0,0 +1,30 @@ +"""scheduler job interval overrides + +Revision ID: 0015_scheduler_settings +Revises: 0014_demo_account +Create Date: 2026-06-16 + +Adds scheduler_settings: a per-job run-interval override (minutes), editable from the admin +Scheduler dashboard. Absent row = use the env/config default, so this is purely additive. +""" +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "0015_scheduler_settings" +down_revision: Union[str, None] = "0014_demo_account" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "scheduler_settings", + sa.Column("job_id", sa.String(length=40), primary_key=True), + sa.Column("interval_minutes", sa.Integer(), nullable=False), + ) + + +def downgrade() -> None: + op.drop_table("scheduler_settings") diff --git a/backend/app/models.py b/backend/app/models.py index 55f694c..f99e8ec 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -325,6 +325,17 @@ class AppState(Base): ) +class SchedulerSetting(Base): + """Admin override for a scheduler job's run interval (minutes). One row per job id; + absent = use the env/config default. DB-backed so it's editable from the admin UI at + runtime without a redeploy (env stays a fallback).""" + + __tablename__ = "scheduler_settings" + + job_id: Mapped[str] = mapped_column(String(40), primary_key=True) + interval_minutes: Mapped[int] = mapped_column(Integer) + + class Playlist(Base): """A per-user named, ordered collection of videos from the shared catalog. diff --git a/backend/app/routes/scheduler.py b/backend/app/routes/scheduler.py index effa4c6..11eeeea 100644 --- a/backend/app/routes/scheduler.py +++ b/backend/app/routes/scheduler.py @@ -1,7 +1,7 @@ """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 +from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import and_, func, or_, select from sqlalchemy.orm import Session @@ -10,12 +10,36 @@ 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 scheduler_snapshot +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) diff --git a/backend/app/scheduler.py b/backend/app/scheduler.py index 467910f..656db73 100644 --- a/backend/app/scheduler.py +++ b/backend/app/scheduler.py @@ -5,10 +5,12 @@ import threading from datetime import datetime, timezone from apscheduler.schedulers.background import BackgroundScheduler +from sqlalchemy import select from app import quota from app.config import settings from app.db import SessionLocal +from app.models import SchedulerSetting from app.state import is_sync_paused from app.sync.autotag import run_autotag_all from app.sync.playlists import sync_all_playlists @@ -32,7 +34,8 @@ _scheduler: BackgroundScheduler | None = None _activity: dict[str, dict] = {} _activity_lock = threading.Lock() -# Each interval job's configured period, surfaced so the dashboard can show "every N min". +# Each interval job's env/config default period (minutes). The admin can override any of +# these at runtime (stored in scheduler_settings, applied live); see load_intervals. JOB_INTERVALS: dict[str, int] = { "rss_poll": settings.rss_poll_minutes, "enrich": settings.enrich_interval_minutes, @@ -43,6 +46,38 @@ JOB_INTERVALS: dict[str, int] = { "playlist_sync": settings.playlist_sync_minutes, } +# Sane bounds for an admin-set interval (minutes). +MIN_INTERVAL = 1 +MAX_INTERVAL = 1440 # one day + + +def load_intervals(db) -> dict[str, int]: + """Effective per-job intervals: the env/config defaults overlaid with any admin overrides + saved in scheduler_settings.""" + iv = dict(JOB_INTERVALS) + for row in db.execute(select(SchedulerSetting)).scalars(): + if row.job_id in iv and row.interval_minutes: + iv[row.job_id] = row.interval_minutes + return iv + + +def apply_interval(db, job_id: str, minutes: int) -> int: + """Persist a job's interval override and reschedule the live job immediately (if the + scheduler runs in this process). Returns the applied value.""" + if job_id not in JOB_INTERVALS: + raise KeyError(job_id) + minutes = max(MIN_INTERVAL, min(MAX_INTERVAL, int(minutes))) + row = db.get(SchedulerSetting, job_id) + if row is None: + db.add(SchedulerSetting(job_id=job_id, interval_minutes=minutes)) + else: + row.interval_minutes = minutes + db.commit() + if _scheduler is not None and _scheduler.get_job(job_id) is not None: + _scheduler.reschedule_job(job_id, trigger="interval", minutes=minutes) + logger.info("rescheduled job %s to every %s min", job_id, minutes) + return minutes + def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() @@ -90,19 +125,23 @@ def scheduler_snapshot() -> dict: 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] = {} + live_intervals: dict[str, int] = {} 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 + iv = getattr(job.trigger, "interval", None) + if iv is not None: + live_intervals[job.id] = round(iv.total_seconds() / 60) with _activity_lock: acts = {k: dict(v) for k, v in _activity.items()} jobs = [] - for job_id, interval in JOB_INTERVALS.items(): + for job_id, default_interval in JOB_INTERVALS.items(): a = acts.get(job_id, {}) jobs.append( { "id": job_id, - "interval_minutes": interval, + "interval_minutes": live_intervals.get(job_id, default_interval), "next_run": next_runs.get(job_id), "running": a.get("running", False), "status": a.get("status"), @@ -166,43 +205,24 @@ def start_scheduler() -> None: global _scheduler if not settings.scheduler_enabled or _scheduler is not None: return + # Effective intervals = env defaults overlaid with any admin overrides from the DB. + db = SessionLocal() + try: + iv = load_intervals(db) + finally: + db.close() + fns = { + "rss_poll": _rss_job, + "enrich": _enrich_job, + "backfill": _backfill_job, + "autotag": _autotag_job, + "shorts": _shorts_job, + "subscriptions": _subscriptions_job, + "playlist_sync": _playlist_sync_job, + } 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", - ) + for job_id, fn in fns.items(): + scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id) scheduler.start() _scheduler = scheduler logger.info("scheduler started") diff --git a/frontend/src/components/Scheduler.tsx b/frontend/src/components/Scheduler.tsx index 1226fb7..8ac9928 100644 --- a/frontend/src/components/Scheduler.tsx +++ b/frontend/src/components/Scheduler.tsx @@ -4,12 +4,15 @@ import { useMutation, useQueryClient } from "@tanstack/react-query"; import { Activity, AlertTriangle, + Check, Clock, Database, Gauge, Pause, + Pencil, Play, RadioTower, + X, } from "lucide-react"; import { api, type SchedulerJob, type SchedulerStatus } from "../lib/api"; import { useLiveQuery } from "../lib/useLiveQuery"; @@ -34,32 +37,109 @@ function fmtCountdown(s: number): string { return `${h}h ${m % 60}m`; } -function StatusDot({ job }: { job: SchedulerJob }) { - const color = job.running - ? "bg-accent animate-pulse" - : job.status === "ok" - ? "bg-emerald-500" - : job.status === "error" - ? "bg-red-500" - : job.status === "skipped" - ? "bg-amber-500" - : "bg-border"; - return ; +const DOT_CLASS: Record = { + running: "bg-accent animate-pulse", + ok: "bg-emerald-500", + error: "bg-red-500", + skipped: "bg-amber-500", + idle: "bg-border", +}; + +function statusKey(job: SchedulerJob): keyof typeof DOT_CLASS { + if (job.running) return "running"; + if (job.status === "ok") return "ok"; + if (job.status === "error") return "error"; + if (job.status === "skipped") return "skipped"; + return "idle"; } -function JobRow({ job }: { job: SchedulerJob }) { +function StatusDot({ k, withTooltip = true }: { k: keyof typeof DOT_CLASS; withTooltip?: boolean }) { const { t } = useTranslation(); + const dot = ; + return withTooltip ? {dot} : dot; +} + +function StatusLegend() { + const { t } = useTranslation(); + return ( +
+ {(["running", "ok", "idle", "error", "skipped"] as const).map((k) => ( + + + {t(`scheduler.dot.${k}`)} + + ))} +
+ ); +} + +function JobRow({ + job, + onSave, + saving, +}: { + job: SchedulerJob; + onSave: (minutes: number) => void; + saving: boolean; +}) { + const { t } = useTranslation(); + const [editing, setEditing] = useState(false); + const [val, setVal] = useState(String(job.interval_minutes)); const next = secsUntil(job.next_run); + useEffect(() => { + if (!editing) setVal(String(job.interval_minutes)); + }, [job.interval_minutes, editing]); + + function save() { + const m = parseInt(val, 10); + if (Number.isFinite(m) && m >= 1 && m <= 1440 && m !== job.interval_minutes) onSave(m); + setEditing(false); + } + return (
- +
-
- {t(`scheduler.jobs.${job.id}`, job.id)} - - {" "} - · {t("scheduler.everyMin", { count: job.interval_minutes })} - +
+ + + {t(`scheduler.jobs.${job.id}`, job.id)} + + + {editing ? ( + + setVal(e.target.value)} + onKeyDown={(e) => { + if (e.key === "Enter") save(); + if (e.key === "Escape") setEditing(false); + }} + className="w-16 bg-card border border-border rounded px-1.5 py-0.5 text-xs outline-none focus:border-accent" + /> + {t("scheduler.minutes")} + + + + ) : ( + + + + )}
{job.running @@ -140,6 +220,13 @@ export default function Scheduler() { onError: () => notify({ level: "error", message: t("scheduler.toggleFailed") }), }); + const intervalMut = useMutation({ + mutationFn: (v: { jobId: string; minutes: number }) => + api.updateSchedulerJob(v.jobId, v.minutes), + onSuccess: () => qc.invalidateQueries({ queryKey: ["scheduler"] }), + onError: () => notify({ level: "error", message: t("scheduler.intervalFailed") }), + }); + if (q.isLoading && !data) return
{t("scheduler.loading")}
; if (!data) @@ -195,9 +282,15 @@ export default function Scheduler() { {/* Jobs */}
{t("scheduler.jobsTitle")}
+
{data.jobs.map((job) => ( - + intervalMut.mutate({ jobId: job.id, minutes })} + /> ))}
diff --git a/frontend/src/i18n/locales/de/scheduler.json b/frontend/src/i18n/locales/de/scheduler.json index e54aa38..293c752 100644 --- a/frontend/src/i18n/locales/de/scheduler.json +++ b/frontend/src/i18n/locales/de/scheduler.json @@ -22,6 +22,25 @@ "quotaUsed": "{{used}} / {{budget}} Einheiten heute verbraucht", "quotaRemaining": "{{formatted}} übrig", "quotaNote": "Gemeinsames YouTube-Data-API-Budget; Reset um Mitternacht US-Pazifik. Backfill tritt zurück, um Spielraum für interaktive Nutzung zu lassen.", + "minutes": "Min", + "editInterval": "Klicken, um das Intervall zu ändern", + "intervalFailed": "Intervall konnte nicht geändert werden", + "dot": { + "running": "läuft gerade", + "ok": "letzter Lauf OK", + "idle": "untätig (in dieser Sitzung noch nicht gelaufen)", + "error": "letzter Lauf fehlgeschlagen", + "skipped": "übersprungen (Sync pausiert)" + }, + "jobDesc": { + "rss_poll": "Prüft den RSS-Feed jedes Kanals auf neue Uploads (kostenlos, kein Kontingent). Fällt er aus, erscheinen neue Videos erst später, wenn ein Backfill sie erfasst — der Feed hinkt YouTube hinterher.", + "enrich": "Holt Videodetails (Dauer, Aufrufe, Live-/Shorts-Flags) und prüft Live-Videos erneut, bis sie enden. Fällt er aus, bleiben Videos ohne Dauer/Statistik und beendete Streams hängen als „live“ fest.", + "backfill": "Holt neue Uploads und dann den gesamten Verlauf vorgemerkter Kanäle, innerhalb des gemeinsamen Kontingents. Fällt er aus, werden ältere Videos und der volle Verlauf nicht ergänzt.", + "autotag": "Erkennt Sprache/Themen jedes Kanals und vergibt automatische Tags. Fällt er aus, bekommen neue Kanäle keine Auto-Tags, sodass Sprach-/Themenfilter sie verfehlen.", + "shorts": "Prüft youtube.com/shorts, um Shorts zu markieren. Fällt er aus, werden Shorts im Feed nicht von normalen Videos getrennt.", + "subscriptions": "Importiert die YouTube-Abos jedes Nutzers neu. Fällt er aus, werden auf YouTube hinzugefügte oder entfernte Abos hier nicht übernommen.", + "playlist_sync": "Spiegelt die YouTube-Wiedergabelisten jedes Nutzers in die App. Fällt er aus, erscheinen Änderungen an deinen YouTube-Listen lokal nicht." + }, "jobs": { "rss_poll": "RSS-Abfrage (neue Uploads)", "enrich": "Anreicherung + Live-Aktualisierung", diff --git a/frontend/src/i18n/locales/en/scheduler.json b/frontend/src/i18n/locales/en/scheduler.json index 60bff5c..e0db313 100644 --- a/frontend/src/i18n/locales/en/scheduler.json +++ b/frontend/src/i18n/locales/en/scheduler.json @@ -22,6 +22,25 @@ "quotaUsed": "{{used}} / {{budget}} units used today", "quotaRemaining": "{{formatted}} left", "quotaNote": "Shared YouTube Data API budget; resets at midnight US Pacific. Backfill yields to leave headroom for interactive use.", + "minutes": "min", + "editInterval": "Click to change how often this runs", + "intervalFailed": "Couldn't change the interval", + "dot": { + "running": "running now", + "ok": "last run OK", + "idle": "idle (not run yet this session)", + "error": "last run failed", + "skipped": "skipped (sync paused)" + }, + "jobDesc": { + "rss_poll": "Checks each channel's RSS feed for new uploads (free, no quota). If it stops, new videos only appear later when a backfill pass catches them, so your feed lags behind YouTube.", + "enrich": "Fetches video details (duration, views, live/Shorts flags) and re-checks live videos until they end. If it stops, videos stay without a duration/stats and ended livestreams stay stuck as “live”.", + "backfill": "Pulls in recent uploads, then the full back-catalogue for opted-in channels, within the shared quota. If it stops, older videos and full history don't fill in.", + "autotag": "Detects each channel's language/topics and applies automatic tags. If it stops, new channels get no auto tags, so language/topic filters miss them.", + "shorts": "Probes youtube.com/shorts to mark which videos are Shorts. If it stops, Shorts aren't separated from normal videos in the feed.", + "subscriptions": "Re-imports every user's YouTube subscriptions. If it stops, channels subscribed to or dropped on YouTube aren't reflected here.", + "playlist_sync": "Mirrors each user's YouTube playlists into the app. If it stops, changes to your YouTube playlists don't show up locally." + }, "jobs": { "rss_poll": "RSS poll (new uploads)", "enrich": "Enrichment + live refresh", diff --git a/frontend/src/i18n/locales/hu/scheduler.json b/frontend/src/i18n/locales/hu/scheduler.json index 485bf0e..bfedaf9 100644 --- a/frontend/src/i18n/locales/hu/scheduler.json +++ b/frontend/src/i18n/locales/hu/scheduler.json @@ -22,6 +22,25 @@ "quotaUsed": "{{used}} / {{budget}} egység ma felhasználva", "quotaRemaining": "{{formatted}} maradt", "quotaNote": "Közös YouTube Data API-keret; éjfélkor (US Pacific) nullázódik. A backfill hátrébb sorol, hogy maradjon hely az interaktív használatnak.", + "minutes": "perc", + "editInterval": "Kattints a gyakoriság módosításához", + "intervalFailed": "Nem sikerült módosítani a gyakoriságot", + "dot": { + "running": "most fut", + "ok": "utolsó futás OK", + "idle": "tétlen (ebben a munkamenetben még nem futott)", + "error": "utolsó futás hibázott", + "skipped": "kihagyva (szinkron szüneteltetve)" + }, + "jobDesc": { + "rss_poll": "Az egyes csatornák RSS-feedjét nézi új feltöltésekért (ingyenes, nem fogyaszt kvótát). Ha leáll, az új videók csak később jelennek meg, amikor egy backfill behúzza őket — így a hírfolyam lemarad a YouTube mögött.", + "enrich": "Lekéri a videók adatait (hossz, megtekintés, élő/Shorts jelzők), és frissíti az élő videókat, amíg véget nem érnek. Ha leáll, a videók hossz/statisztika nélkül maradnak, a véget ért adások pedig „élő” állapotban ragadnak.", + "backfill": "Behúzza a friss feltöltéseket, majd a kijelölt csatornák teljes előzményét, a közös kvótán belül. Ha leáll, a régebbi videók és a teljes előzmény nem töltődik fel.", + "autotag": "Felismeri a csatornák nyelvét/témáit, és automatikus címkéket ad. Ha leáll, az új csatornák címke nélkül maradnak, így a nyelv/téma szűrők kihagyják őket.", + "shorts": "A youtube.com/shorts próbával jelöli, mely videók Shorts-ok. Ha leáll, a Shorts-ok nem különülnek el a normál videóktól a hírfolyamban.", + "subscriptions": "Újraimportálja minden felhasználó YouTube-feliratkozásait. Ha leáll, a YouTube-on felvett vagy törölt feliratkozások nem tükröződnek itt.", + "playlist_sync": "Tükrözi a felhasználók YouTube lejátszási listáit az appba. Ha leáll, a YouTube-listák változásai nem jelennek meg lokálisan." + }, "jobs": { "rss_poll": "RSS-lekérdezés (új feltöltések)", "enrich": "Adatdúsítás + élő frissítés", diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 960bbe1..0615f19 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -434,6 +434,11 @@ export const api = { myUsage: (): Promise => req("/api/quota/my-usage"), adminQuota: (days = 30): Promise => req(`/api/quota/admin?days=${days}`), schedulerStatus: (): Promise => req("/api/admin/scheduler"), + updateSchedulerJob: (jobId: string, intervalMinutes: number): Promise<{ id: string; interval_minutes: number }> => + req(`/api/admin/scheduler/jobs/${jobId}`, { + method: "PATCH", + body: JSON.stringify({ interval_minutes: intervalMinutes }), + }), // --- onboarding / admin --- requestAccess: (email: string): Promise<{ status: string }> =>