diff --git a/backend/routers/channels.py b/backend/routers/channels.py index 38ce5d0..95b734c 100644 --- a/backend/routers/channels.py +++ b/backend/routers/channels.py @@ -1,4 +1,5 @@ import json +import threading as _threading from datetime import datetime, timedelta from typing import Optional @@ -14,6 +15,9 @@ from ..services import ytdlp router = APIRouter() +_tasks: dict = {} +_tasks_lock = _threading.Lock() + class ChannelOut(BaseModel): id: int @@ -286,6 +290,12 @@ def sync_all_channels( return {"indexing": len(channels)} +@router.get("/tasks") +def get_active_tasks(current_user: User = Depends(get_current_user)): + with _tasks_lock: + return list(_tasks.values()) + + @router.post("/mark-seen", status_code=204) def mark_channels_seen( db: Session = Depends(get_db), @@ -652,11 +662,11 @@ def fetch_popular_videos( ): """Fetch the channel's most popular videos from YouTube and index them.""" channel = _get_channel_or_404(db, channel_id) - background_tasks.add_task(_fetch_popular_task, channel_id, channel.youtube_channel_id) + background_tasks.add_task(_fetch_popular_task, channel_id, channel.youtube_channel_id, channel.name or "") return {"detail": "Fetching popular videos"} -def _fetch_popular_task(channel_id: int, youtube_channel_id: str): +def _fetch_popular_task(channel_id: int, youtube_channel_id: str, channel_name: str = ""): """Half-and-half popular fetch. Phase 1 (fast): flat-playlist crawl of the full channel → store any @@ -742,15 +752,32 @@ def _fetch_popular_task(channel_id: int, youtube_channel_id: str): if not video_ids: return - with ThreadPoolExecutor(max_workers=3) as pool: - futures = {pool.submit(ytdlp.fetch_video_metadata, vid): vid for vid in video_ids} - results = {} - for future in as_completed(futures): - vid = futures[future] - try: - results[vid] = future.result() - except Exception: - pass + task_id = f"popular-{channel_id}" + with _tasks_lock: + _tasks[task_id] = { + "id": task_id, + "label": f"Popular fetch — {channel_name}" if channel_name else "Popular fetch", + "total": len(video_ids), + "done": 0, + "started_at": datetime.utcnow().isoformat(), + } + + results = {} + try: + with ThreadPoolExecutor(max_workers=3) as pool: + futures = {pool.submit(ytdlp.fetch_video_metadata, vid): vid for vid in video_ids} + for future in as_completed(futures): + vid = futures[future] + try: + results[vid] = future.result() + except Exception: + pass + with _tasks_lock: + if task_id in _tasks: + _tasks[task_id]["done"] += 1 + finally: + with _tasks_lock: + _tasks.pop(task_id, None) db = SessionLocal() try: diff --git a/frontend/src/api/index.js b/frontend/src/api/index.js index dc126c3..0d74a2e 100644 --- a/frontend/src/api/index.js +++ b/frontend/src/api/index.js @@ -64,6 +64,7 @@ export const renameChannelGroup = (id, name) => api.patch(`/channels/groups/${id export const addChannelToGroup = (groupId, channelId) => api.post(`/channels/groups/${groupId}/channels/${channelId}`); export const removeChannelFromGroup = (groupId, channelId) => api.delete(`/channels/groups/${groupId}/channels/${channelId}`); export const bulkChannelAction = (channel_ids, action) => api.post("/channels/bulk-action", { channel_ids, action }); +export const getActiveTasks = () => api.get("/channels/tasks"); // Videos export const homeFeed = (page = 0, limit = 25, mode = "ranked", duration = "") => diff --git a/frontend/src/pages/Downloads.jsx b/frontend/src/pages/Downloads.jsx index 0d39b36..6e66777 100644 --- a/frontend/src/pages/Downloads.jsx +++ b/frontend/src/pages/Downloads.jsx @@ -1,7 +1,7 @@ import { useState, useMemo } from "react"; import { useNavigate } from "react-router-dom"; import { useQuery, useMutation, useQueryClient } from "@tanstack/react-query"; -import { getDownloads, deleteDownload, deleteAllDownloads, restoreDownload } from "../api"; +import { getDownloads, deleteDownload, deleteAllDownloads, restoreDownload, getActiveTasks } from "../api"; import SortPicker from "../components/SortPicker"; const HISTORY_SORTS = [ @@ -66,6 +66,12 @@ export default function DownloadsPage() { }, }); + const { data: activeTasks = [] } = useQuery({ + queryKey: ["active-tasks"], + queryFn: () => getActiveTasks().then((r) => r.data), + refetchInterval: (query) => (query.state.data?.length > 0 ? 2000 : 5000), + }); + const clearAllMut = useMutation({ mutationFn: deleteAllDownloads, onSuccess: () => { @@ -127,6 +133,26 @@ export default function DownloadsPage() { )} + {activeTasks.length > 0 && ( + + Background Tasks + + {activeTasks.map((task) => { + const pct = task.total > 0 ? (task.done / task.total) * 100 : 0; + return ( + + + {task.label} + {task.done}/{task.total} + + + + ); + })} + + + )} + {active.length > 0 && ( Active
{task.label}