diff --git a/backend/app/sync/autotag.py b/backend/app/sync/autotag.py index e10db9e..09cce68 100644 --- a/backend/app/sync/autotag.py +++ b/backend/app/sync/autotag.py @@ -13,6 +13,7 @@ from sqlalchemy.orm import Session log = logging.getLogger("subfeed.autotag") +from app import progress from app.config import settings from app.models import Channel, ChannelTag, Tag, Video @@ -270,13 +271,14 @@ def run_autotag_all(db: Session, only_missing: bool = False) -> dict: ) channels = db.execute(query).scalars().all() tagged = 0 - for channel in channels: + for i, channel in enumerate(channels, 1): try: apply_channel_autotags(db, channel) tagged += 1 except Exception: db.rollback() log.exception("Auto-tagging failed for channel %s", channel.id) + progress.report(i, len(channels), "autotag") removed = _cleanup_orphan_system_tags(db) return {"channels_tagged": tagged, "orphan_tags_removed": removed} diff --git a/backend/app/sync/playlists.py b/backend/app/sync/playlists.py index bb60489..f21ef4f 100644 --- a/backend/app/sync/playlists.py +++ b/backend/app/sync/playlists.py @@ -12,7 +12,7 @@ from datetime import datetime, timezone from sqlalchemy import delete, select from sqlalchemy.orm import Session -from app import quota +from app import progress, quota from app.auth import has_read_scope, has_write_scope from app.models import Channel, OAuthToken, Playlist, PlaylistItem, User, Video from app.sync.videos import apply_video_details, parse_dt @@ -382,8 +382,9 @@ def sync_all_playlists(db: Session) -> dict: .all() ) total = 0 - for user in users: + for i, user in enumerate(users, 1): if not has_read_scope(user): + progress.report(i, len(users), "playlist_sync") continue try: with quota.attribute(user.id, "playlist_sync"): @@ -394,4 +395,5 @@ def sync_all_playlists(db: Session) -> dict: except Exception: db.rollback() log.exception("Playlist sync crashed for user %s", user.id) + progress.report(i, len(users), "playlist_sync") return {"users": len(users), "playlists_synced": total} diff --git a/backend/app/sync/runner.py b/backend/app/sync/runner.py index 2a1a239..ed76ebc 100644 --- a/backend/app/sync/runner.py +++ b/backend/app/sync/runner.py @@ -5,6 +5,7 @@ user" (any user with a stored refresh token) unless a YOUTUBE_API_KEY is configu public reads. """ import logging +from concurrent.futures import ThreadPoolExecutor, as_completed from sqlalchemy import select from sqlalchemy.orm import Session @@ -16,15 +17,19 @@ log = logging.getLogger("subfeed.sync") from app.models import Channel, OAuthToken, Subscription, User from app.sync.subscriptions import import_subscriptions from app.sync.videos import ( + apply_rss_feed, backfill_channel_deep, backfill_channel_recent, enrich_pending, - poll_rss_channel, reconcile_full_history, refresh_live, run_shorts_classification, ) from app.youtube.client import YouTubeClient +from app.youtube.rss import fetch_channel_feed + +# RSS feeds are free and network-bound, so fetch many at once; DB writes stay single-threaded. +RSS_POLL_WORKERS = 16 def get_service_user(db: Session) -> User | None: @@ -43,13 +48,30 @@ def get_service_user(db: Session) -> User | None: def run_rss_poll(db: Session, channels: list[Channel] | None = None) -> int: if channels is None: channels = db.execute(select(Channel)).scalars().all() + total = len(channels) new = 0 - for channel in channels: - try: - new += poll_rss_channel(db, channel) - except Exception: - db.rollback() - log.exception("RSS poll failed for channel %s", channel.id) + done = 0 + # Fetch feeds concurrently (network-bound, no DB), then apply inserts on this thread as + # each fetch returns — a SQLAlchemy session isn't thread-safe. Pass the channel id (a + # plain str) to the workers; the ORM instance is touched only here on the main thread. + futures = {} + with ThreadPoolExecutor(max_workers=RSS_POLL_WORKERS) as pool: + for channel in channels: + futures[pool.submit(fetch_channel_feed, channel.id)] = channel + for fut in as_completed(futures): + channel = futures[fut] + try: + entries = fut.result() + except Exception: + log.exception("RSS fetch failed for channel %s", channel.id) + entries = [] + try: + new += apply_rss_feed(db, channel, entries) + except Exception: + db.rollback() + log.exception("RSS apply failed for channel %s", channel.id) + done += 1 + progress.report(done, total, "rss_poll") return new @@ -90,12 +112,13 @@ def run_subscription_resync(db: Session) -> dict: .scalars() .all() ) - for user in users: + for i, user in enumerate(users, 1): try: import_subscriptions(db, user) except Exception: db.rollback() log.exception("Subscription resync failed for user %s", user.id) + progress.report(i, len(users), "subscriptions") return {"users": len(users)} diff --git a/backend/app/sync/videos.py b/backend/app/sync/videos.py index 0292390..4b64c51 100644 --- a/backend/app/sync/videos.py +++ b/backend/app/sync/videos.py @@ -67,8 +67,10 @@ def _insert_stubs(db: Session, rows: list[dict]) -> int: # --- RSS (free) --- -def poll_rss_channel(db: Session, channel: Channel) -> int: - entries = fetch_channel_feed(channel.id) +def apply_rss_feed(db: Session, channel: Channel, entries: list[dict]) -> int: + """Insert new video stubs from an already-fetched feed and stamp the channel's last RSS + time. DB-only — must run on the session's own thread; the network fetch + (fetch_channel_feed) is kept separate so callers like run_rss_poll can parallelise it.""" inserted = _insert_stubs(db, entries) channel.last_rss_at = _now() db.add(channel) @@ -76,6 +78,10 @@ def poll_rss_channel(db: Session, channel: Channel) -> int: return inserted +def poll_rss_channel(db: Session, channel: Channel) -> int: + return apply_rss_feed(db, channel, fetch_channel_feed(channel.id)) + + # --- Backfill (uploads playlist; costs quota) --- def _stub_from_playlist_item(item: dict, channel_id: str) -> dict | None: content = item.get("contentDetails", {}) diff --git a/frontend/src/components/Scheduler.tsx b/frontend/src/components/Scheduler.tsx index 549916e..5a2556d 100644 --- a/frontend/src/components/Scheduler.tsx +++ b/frontend/src/components/Scheduler.tsx @@ -21,6 +21,8 @@ import { notify } from "../lib/notifications"; import Tooltip from "./Tooltip"; const POLL_MS = 4000; +// While any job is running, poll faster so short scheduled runs aren't missed between ticks. +const FAST_POLL_MS = 1500; // Seconds until an ISO instant (negative = past). function secsUntil(iso: string | null): number | null { @@ -73,19 +75,25 @@ function StatusLegend() { ); } -function JobProgress({ p }: { p: NonNullable }) { +// Shown for ANY running job. With reported counts it's a determinate bar; for a job that +// runs but doesn't report progress (or hasn't yet) it falls back to a generic "working" +// label and an indeterminate sliver — so every active run is visible, not just the ones +// that report numbers. +function JobProgress({ p }: { p: SchedulerJob["progress"] }) { const { t } = useTranslation(); - const phase = p.phase ? t(`scheduler.phase.${p.phase}`, p.phase) : null; + const phase = p?.phase ? t(`scheduler.phase.${p.phase}`, p.phase) : t("scheduler.phase.working"); const pct = - p.total && p.total > 0 ? Math.min(100, Math.round((p.current / p.total) * 100)) : null; + p?.total && p.total > 0 ? Math.min(100, Math.round((p.current / p.total) * 100)) : null; return (
{phase} - {p.total != null + {p?.total != null ? `${p.current.toLocaleString()} / ${p.total.toLocaleString()}` - : p.current.toLocaleString()} + : p + ? p.current.toLocaleString() + : ""}
@@ -184,7 +192,7 @@ function JobRow({ · {job.last_result} ) : null}
- {job.running && job.progress && } + {job.running && }
{job.running ? ( @@ -298,8 +306,10 @@ function Stat({ export default function Scheduler() { const { t } = useTranslation(); const qc = useQueryClient(); + // Poll faster while any job is running so live progress is smooth, then ease back when + // idle. Derived from the freshest data by react-query itself (see useLiveQuery). const q = useLiveQuery(["scheduler"], api.schedulerStatus, { - intervalMs: POLL_MS, + intervalMs: (d) => (d?.jobs?.some((j) => j.running) ? FAST_POLL_MS : POLL_MS), }); const data = q.data; diff --git a/frontend/src/i18n/locales/de/scheduler.json b/frontend/src/i18n/locales/de/scheduler.json index 8c30032..d1a451f 100644 --- a/frontend/src/i18n/locales/de/scheduler.json +++ b/frontend/src/i18n/locales/de/scheduler.json @@ -38,7 +38,12 @@ "enrich": "Videos anreichern", "backfill_recent": "Neue Uploads nachladen", "backfill_deep": "Ganze Historie nachladen", - "probe": "Shorts klassifizieren" + "probe": "Shorts klassifizieren", + "rss_poll": "Auf neue Uploads prüfen", + "subscriptions": "Abos synchronisieren", + "autotag": "Kanäle automatisch taggen", + "playlist_sync": "Playlists synchronisieren", + "working": "Arbeitet…" }, "maintenance": { "title": "Wartung", diff --git a/frontend/src/i18n/locales/en/scheduler.json b/frontend/src/i18n/locales/en/scheduler.json index b33bcf5..f9b0073 100644 --- a/frontend/src/i18n/locales/en/scheduler.json +++ b/frontend/src/i18n/locales/en/scheduler.json @@ -38,7 +38,12 @@ "enrich": "Enriching videos", "backfill_recent": "Backfilling recent uploads", "backfill_deep": "Backfilling full history", - "probe": "Classifying Shorts" + "probe": "Classifying Shorts", + "rss_poll": "Checking for new uploads", + "subscriptions": "Syncing subscriptions", + "autotag": "Auto-tagging channels", + "playlist_sync": "Syncing playlists", + "working": "Working…" }, "maintenance": { "title": "Maintenance", diff --git a/frontend/src/i18n/locales/hu/scheduler.json b/frontend/src/i18n/locales/hu/scheduler.json index a3a1b9a..7a8bc36 100644 --- a/frontend/src/i18n/locales/hu/scheduler.json +++ b/frontend/src/i18n/locales/hu/scheduler.json @@ -38,7 +38,12 @@ "enrich": "Videók dúsítása", "backfill_recent": "Friss feltöltések behúzása", "backfill_deep": "Teljes előzmény behúzása", - "probe": "Shorts besorolás" + "probe": "Shorts besorolás", + "rss_poll": "Új feltöltések keresése", + "subscriptions": "Feliratkozások szinkronizálása", + "autotag": "Csatornák automatikus címkézése", + "playlist_sync": "Lejátszási listák szinkronizálása", + "working": "Dolgozik…" }, "maintenance": { "title": "Karbantartás", diff --git a/frontend/src/lib/useLiveQuery.ts b/frontend/src/lib/useLiveQuery.ts index 121045b..9571337 100644 --- a/frontend/src/lib/useLiveQuery.ts +++ b/frontend/src/lib/useLiveQuery.ts @@ -8,14 +8,22 @@ import { useQuery, type QueryKey } from "@tanstack/react-query"; export function useLiveQuery( key: QueryKey, queryFn: () => Promise, - opts: { intervalMs?: number; enabled?: boolean } = {} + // intervalMs may be a function of the latest data, so the poll cadence can adapt to it + // (e.g. poll faster while a job is running). react-query re-evaluates it after each fetch + // against fresh data — the right place for this, vs. a React state toggle that can lag or + // stall the live updates. + opts: { intervalMs?: number | ((data: T | undefined) => number); enabled?: boolean } = {} ) { const { intervalMs = 4000, enabled = true } = opts; return useQuery({ queryKey: key, queryFn, enabled, - refetchInterval: enabled ? intervalMs : false, + refetchInterval: !enabled + ? false + : typeof intervalMs === "function" + ? (query) => intervalMs(query.state.data as T | undefined) + : intervalMs, refetchIntervalInBackground: false, staleTime: 0, });