diff --git a/app/services/ingestion/scholar_processing.py b/app/services/ingestion/scholar_processing.py index f41d904..7867014 100644 --- a/app/services/ingestion/scholar_processing.py +++ b/app/services/ingestion/scholar_processing.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio import logging import random +from collections.abc import Callable, Coroutine from datetime import UTC, datetime from typing import Any @@ -274,6 +275,7 @@ async def _run_first_pass( request_delay_seconds: int, queue_delay_seconds: int, progress: RunProgress, + on_progress: Callable[[int, int], Coroutine[Any, Any, None]] | None = None, ) -> dict[int, int]: first_pass_cstarts: dict[int, int] = {} for index, scholar in enumerate(scholars): @@ -310,6 +312,8 @@ async def _run_first_pass( scholars_remaining=len(scholars) - index - 1, ) return first_pass_cstarts + if on_progress is not None: + await on_progress(1, 0) resume_cstart = outcome.result_entry.get("continuation_cstart") if resume_cstart is not None and int(resume_cstart) > start_cstart: first_pass_cstarts[int(scholar.id)] = int(resume_cstart) @@ -330,6 +334,7 @@ async def _run_depth_pass( auto_queue_continuations: bool, queue_delay_seconds: int, progress: RunProgress, + on_progress: Callable[[int, int], Coroutine[Any, Any, None]] | None = None, ) -> None: for index, scholar in enumerate(scholars): resume_cstart = first_pass_cstarts.get(int(scholar.id)) @@ -367,6 +372,8 @@ async def _run_depth_pass( scholars_remaining=len(scholars) - index - 1, ) break + if on_progress is not None: + await on_progress(0, 1) async def run_scholar_iteration( @@ -387,6 +394,8 @@ async def run_scholar_iteration( auto_queue_continuations: bool, queue_delay_seconds: int, ) -> RunProgress: + from app.services.runs.events import run_events + progress = RunProgress() scholar_kwargs: dict[str, Any] = { "request_delay_seconds": request_delay_seconds, @@ -396,6 +405,21 @@ async def run_scholar_iteration( "rate_limit_backoff_seconds": rate_limit_backoff_seconds, "page_size": page_size, } + + visited = 0 + finished = 0 + total = len(scholars) + + async def _emit(v: int = 0, f: int = 0) -> None: + nonlocal visited, finished + visited += v + finished += f + await run_events.publish( + run.id, + "scholar_progress", + {"visited": visited, "finished": finished, "total": total}, + ) + first_pass_cstarts = await _run_first_pass( db_session, scholars=scholars, @@ -407,8 +431,12 @@ async def run_scholar_iteration( request_delay_seconds=request_delay_seconds, queue_delay_seconds=queue_delay_seconds, progress=progress, + on_progress=_emit, ) remaining_max = max(max_pages_per_scholar - 1, 0) + scholars_finished_in_first_pass = len(scholars) - len(first_pass_cstarts) + if scholars_finished_in_first_pass > 0: + await _emit(f=scholars_finished_in_first_pass) if remaining_max <= 0: return progress await _run_depth_pass( @@ -424,5 +452,6 @@ async def run_scholar_iteration( auto_queue_continuations=auto_queue_continuations, queue_delay_seconds=queue_delay_seconds, progress=progress, + on_progress=_emit, ) return progress diff --git a/frontend/src/pages/DashboardPage.vue b/frontend/src/pages/DashboardPage.vue index fbd0b4c..d5a82ee 100644 --- a/frontend/src/pages/DashboardPage.vue +++ b/frontend/src/pages/DashboardPage.vue @@ -486,9 +486,16 @@ watch(

- Started {{ formatDate(displayedLatestRun.start_dt) }}. Processed - {{ displayedLatestRun.scholar_count }} scholars and discovered - {{ displayedLatestRun.new_publication_count }} new publications. + Started {{ formatDate(displayedLatestRun.start_dt) }}. + +

, + scholarProgress: null as { visited: number; finished: number; total: number } | null, }), getters: { isRunActive(state): boolean { @@ -257,6 +258,7 @@ export const useRunStatusStore = defineStore("runStatus", { } activeStreamRunId = targetRunId; this.livePublications = []; + this.scholarProgress = null; eventSource = new EventSource(`/api/v1/runs/${targetRunId}/stream`); eventSource.addEventListener("publication_discovered", (e) => { try { @@ -312,6 +314,19 @@ export const useRunStatusStore = defineStore("runStatus", { console.error("Failed to parse SSE event", err); } }); + eventSource.addEventListener("scholar_progress", (e) => { + try { + const data = JSON.parse(e.data); + const visited = typeof data.visited === "number" ? Math.max(0, Math.trunc(data.visited)) : null; + const finished = typeof data.finished === "number" ? Math.max(0, Math.trunc(data.finished)) : null; + const total = typeof data.total === "number" ? Math.max(1, Math.trunc(data.total)) : null; + if (visited !== null && finished !== null && total !== null) { + this.scholarProgress = { visited, finished, total }; + } + } catch (err) { + console.error("Failed to parse SSE event", err); + } + }); eventSource.onerror = () => { // Reconnecting is handled automatically by EventSource, // but if it's permanently closed, we could do something here. @@ -491,6 +506,7 @@ export const useRunStatusStore = defineStore("runStatus", { this.lastSyncAt = null; this.safetyState = createDefaultSafetyState(); this.livePublications = []; + this.scholarProgress = null; }, }, });