diff --git a/.claude/settings.json b/.claude/settings.json index a332c01..e801a7c 100644 --- a/.claude/settings.json +++ b/.claude/settings.json @@ -19,7 +19,12 @@ "Bash(git add app/api/routers/runs.py app/services/arxiv/rate_limit.py tests/integration/test_run_lifecycle_consistency.py docs/website/package-lock.json)", "Bash(git push)", "Bash(uvx ruff format --check app/api/routers/runs.py app/services/arxiv/rate_limit.py tests/integration/test_run_lifecycle_consistency.py)", - "Bash(test -f docs/website/package-lock.json)" + "Bash(test -f docs/website/package-lock.json)", + "Bash(gh run list --branch openalex-size-reduction-scholar-hydration --limit 3 --json databaseId,status,conclusion,name,event)", + "Bash(gh run list --limit 10 --json databaseId,status,conclusion,name,event,headBranch)", + "Bash(gh run view 22554585841 --log-failed)", + "Bash(gh run rerun 22554585841 --failed)", + "Bash(bash scripts/premerge.sh)" ], "additionalDirectories": [ "/home/jv/src/personal/playground/scholar_scraper/scholar-scraper/frontend/src" diff --git a/.env.example b/.env.example index 17e0363..9e4a0a9 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,7 @@ DATABASE_POOL_MODE=auto DATABASE_POOL_SIZE=5 DATABASE_POOL_MAX_OVERFLOW=10 DATABASE_POOL_TIMEOUT_SECONDS=30 +DATABASE_RESERVED_API_CONNECTIONS=3 # ------------------------------ # Frontend Dev Overrides diff --git a/.gitignore b/.gitignore index 68a3f87..7c06050 100644 --- a/.gitignore +++ b/.gitignore @@ -18,3 +18,5 @@ frontend/dist/ frontend/.vite/ docs/website/node_modules/ docs/website/build/ +docs/website/.docusaurus/ +.claude/ diff --git a/app/api/routers/admin_dbops.py b/app/api/routers/admin_dbops.py index 0f4d1f6..048be70 100644 --- a/app/api/routers/admin_dbops.py +++ b/app/api/routers/admin_dbops.py @@ -5,7 +5,7 @@ import logging from fastapi import APIRouter, Depends, Path, Query, Request from sqlalchemy.ext.asyncio import AsyncSession -from app.api.deps import get_api_admin_user +from app.api.deps import get_api_admin_user, get_api_current_user from app.api.errors import ApiException from app.api.responses import success_payload from app.api.schemas import ( @@ -57,7 +57,7 @@ def _requested_by_value(*, payload, admin_user: User) -> str: return from_payload or admin_user.email -def _serialize_pdf_queue_item(item) -> dict[str, object]: +def _serialize_pdf_queue_item(item, *, is_admin: bool = False) -> dict[str, object]: return { "publication_id": item.publication_id, "title": item.title, @@ -69,7 +69,7 @@ def _serialize_pdf_queue_item(item) -> dict[str, object]: "last_failure_detail": item.last_failure_detail, "last_source": item.last_source, "requested_by_user_id": item.requested_by_user_id, - "requested_by_email": item.requested_by_email, + "requested_by_email": item.requested_by_email if is_admin else None, "queued_at": item.queued_at, "last_attempt_at": item.last_attempt_at, "resolved_at": item.resolved_at, @@ -185,7 +185,7 @@ async def get_pdf_queue( offset: int | None = Query(default=None, ge=0), status: str | None = Query(default=None), db_session: AsyncSession = Depends(get_db_session), - admin_user: User = Depends(get_api_admin_user), + current_user: User = Depends(get_api_current_user), ): normalized_status = (status or "").strip().lower() or None resolved_page, resolved_limit, resolved_offset = _resolve_pdf_queue_paging( @@ -204,7 +204,7 @@ async def get_pdf_queue( logger, "info", "api.admin.db.pdf_queue_listed", - admin_user_id=int(admin_user.id), + user_id=int(current_user.id), page=int(resolved_page), page_size=int(resolved_limit), offset=int(resolved_offset), @@ -215,7 +215,7 @@ async def get_pdf_queue( return success_payload( request, data={ - "items": [_serialize_pdf_queue_item(item) for item in queue_page.items], + "items": [_serialize_pdf_queue_item(item, is_admin=current_user.is_admin) for item in queue_page.items], **_pdf_queue_page_data( total_count=queue_page.total_count, page=resolved_page, diff --git a/app/api/routers/runs.py b/app/api/routers/runs.py index d074bb1..87dfa0a 100644 --- a/app/api/routers/runs.py +++ b/app/api/routers/runs.py @@ -221,11 +221,11 @@ def _spawn_background_execution( user_settings: Any, idempotency_key: str | None, ) -> None: - from app.db.session import get_session_factory + from app.db.background_session import background_session task = asyncio.create_task( ingest_service.execute_run( - session_factory=get_session_factory(), + session_factory=background_session, run_id=run.id, user_id=current_user.id, scholars=scholars, diff --git a/app/api/routers/scholars.py b/app/api/routers/scholars.py index 7095471..2a0c276 100644 --- a/app/api/routers/scholars.py +++ b/app/api/routers/scholars.py @@ -23,6 +23,9 @@ from app.api.schemas import ( DataImportEnvelope, DataImportRequest, MessageEnvelope, + ScholarBulkCountEnvelope, + ScholarBulkIdsRequest, + ScholarBulkToggleRequest, ScholarCreateRequest, ScholarEnvelope, ScholarImageUrlUpdateRequest, @@ -42,6 +45,22 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/scholars", tags=["api-scholars"]) +def _parse_ids_param(ids: str | None) -> list[int] | None: + if not ids: + return None + parts = [p.strip() for p in ids.split(",") if p.strip()] + if not parts: + return None + try: + return [int(p) for p in parts] + except ValueError as exc: + raise ApiException( + status_code=400, + code="invalid_ids_param", + message="The 'ids' parameter must be a comma-separated list of integers.", + ) from exc + + @router.get( "", response_model=ScholarsListEnvelope, @@ -69,16 +88,81 @@ async def list_scholars( ) async def export_scholars_and_publications( request: Request, + ids: str | None = Query(None, description="Comma-separated scholar profile IDs to export"), db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_api_current_user), ): + scholar_profile_ids = _parse_ids_param(ids) data = await import_export_service.export_user_data( db_session, user_id=current_user.id, + scholar_profile_ids=scholar_profile_ids, ) return success_payload(request, data=data) +@router.post( + "/bulk-delete", + response_model=ScholarBulkCountEnvelope, +) +async def bulk_delete_scholars( + payload: ScholarBulkIdsRequest, + request: Request, + db_session: AsyncSession = Depends(get_db_session), + current_user: User = Depends(get_api_current_user), +): + try: + deleted_count = await scholar_service.bulk_delete_scholars( + db_session, + user_id=current_user.id, + scholar_profile_ids=payload.scholar_profile_ids, + upload_dir=settings.scholar_image_upload_dir, + ) + except scholar_service.ScholarServiceError as exc: + raise ApiException( + status_code=409, + code="scholar_bulk_delete_failed", + message=str(exc), + ) from exc + structured_log( + logger, + "info", + "scholars.bulk_delete", + user_id=current_user.id, + requested_ids=payload.scholar_profile_ids, + deleted_count=deleted_count, + ) + return success_payload(request, data={"deleted_count": deleted_count, "updated_count": 0}) + + +@router.post( + "/bulk-toggle", + response_model=ScholarBulkCountEnvelope, +) +async def bulk_toggle_scholars( + payload: ScholarBulkToggleRequest, + request: Request, + db_session: AsyncSession = Depends(get_db_session), + current_user: User = Depends(get_api_current_user), +): + updated_count = await scholar_service.bulk_toggle_scholars( + db_session, + user_id=current_user.id, + scholar_profile_ids=payload.scholar_profile_ids, + is_enabled=payload.is_enabled, + ) + structured_log( + logger, + "info", + "scholars.bulk_toggle", + user_id=current_user.id, + requested_ids=payload.scholar_profile_ids, + is_enabled=payload.is_enabled, + updated_count=updated_count, + ) + return success_payload(request, data={"deleted_count": 0, "updated_count": updated_count}) + + @router.post( "/import", response_model=DataImportEnvelope, @@ -144,18 +228,17 @@ async def create_scholar( message=str(exc), ) from exc structured_log(logger, "info", "api.scholars.created", user_id=current_user.id, scholar_profile_id=created.id) - did_queue_initial_scrape = await enqueue_initial_scrape_job_for_scholar( + await enqueue_initial_scrape_job_for_scholar( db_session, profile=created, user_id=current_user.id, ) - if not did_queue_initial_scrape: - created = await hydrate_scholar_metadata_if_needed( - db_session, - profile=created, - source=source, - user_id=current_user.id, - ) + created = await hydrate_scholar_metadata_if_needed( + db_session, + profile=created, + source=source, + user_id=current_user.id, + ) return success_payload( request, @@ -262,11 +345,18 @@ async def delete_scholar( code="scholar_not_found", message="Scholar not found.", ) - await scholar_service.delete_scholar( - db_session, - profile=profile, - upload_dir=settings.scholar_image_upload_dir, - ) + try: + await scholar_service.delete_scholar( + db_session, + profile=profile, + upload_dir=settings.scholar_image_upload_dir, + ) + except scholar_service.ScholarServiceError as exc: + raise ApiException( + status_code=409, + code="scholar_delete_failed", + message=str(exc), + ) from exc structured_log( logger, "info", "api.scholars.deleted", user_id=current_user.id, scholar_profile_id=scholar_profile_id ) diff --git a/app/api/schemas/scholars.py b/app/api/schemas/scholars.py index dda8329..bc1035d 100644 --- a/app/api/schemas/scholars.py +++ b/app/api/schemas/scholars.py @@ -2,7 +2,7 @@ from __future__ import annotations from datetime import datetime -from pydantic import BaseModel, ConfigDict, Field +from pydantic import BaseModel, ConfigDict, Field, model_validator from app.api.schemas.common import ApiMeta @@ -126,13 +126,49 @@ class DataExportEnvelope(BaseModel): model_config = ConfigDict(extra="forbid") +class ScholarBulkIdsRequest(BaseModel): + scholar_profile_ids: list[int] = Field(..., min_length=1, max_length=500) + + model_config = ConfigDict(extra="forbid") + + +class ScholarBulkToggleRequest(BaseModel): + scholar_profile_ids: list[int] = Field(..., min_length=1, max_length=500) + is_enabled: bool + + model_config = ConfigDict(extra="forbid") + + +class ScholarBulkCountData(BaseModel): + deleted_count: int = 0 + updated_count: int = 0 + + model_config = ConfigDict(extra="forbid") + + +class ScholarBulkCountEnvelope(BaseModel): + data: ScholarBulkCountData + meta: ApiMeta + + model_config = ConfigDict(extra="forbid") + + class DataImportRequest(BaseModel): schema_version: int | None = None + exported_at: str | None = None scholars: list[ScholarExportItemData] = Field(default_factory=list) publications: list[PublicationExportItemData] = Field(default_factory=list) model_config = ConfigDict(extra="forbid") + @model_validator(mode="before") + @classmethod + def unwrap_export_envelope(cls, values: dict) -> dict: + """Accept the full export envelope format (with data/meta wrapper).""" + if isinstance(values, dict) and "data" in values and isinstance(values["data"], dict): + return values["data"] + return values + class DataImportResultData(BaseModel): scholars_created: int diff --git a/app/db/background_session.py b/app/db/background_session.py new file mode 100644 index 0000000..745f155 --- /dev/null +++ b/app/db/background_session.py @@ -0,0 +1,57 @@ +"""Semaphore-gated DB sessions for background tasks. + +Background tasks (ingestion, PDF resolution, scheduler) acquire a semaphore +permit before checking out a database connection. The semaphore cap is +``pool_size + max_overflow - reserved_for_api``, which guarantees that at +least *reserved_for_api* connections remain available for API request +handlers at all times. +""" + +from __future__ import annotations + +import asyncio +import logging +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.session import get_session_factory +from app.logging_utils import structured_log +from app.settings import settings + +logger = logging.getLogger(__name__) + +_semaphore: asyncio.Semaphore | None = None + + +def _build_semaphore() -> asyncio.Semaphore: + pool_capacity = max(1, settings.database_pool_size) + max(0, settings.database_pool_max_overflow) + reserved = max(0, settings.database_reserved_api_connections) + limit = max(1, pool_capacity - reserved) + structured_log( + logger, + "info", + "db.background_semaphore_initialized", + pool_capacity=pool_capacity, + reserved_for_api=reserved, + background_limit=limit, + ) + return asyncio.Semaphore(limit) + + +def get_background_semaphore() -> asyncio.Semaphore: + global _semaphore + if _semaphore is None: + _semaphore = _build_semaphore() + return _semaphore + + +@asynccontextmanager +async def background_session() -> AsyncIterator[AsyncSession]: + """Yield a DB session after acquiring the background semaphore.""" + semaphore = get_background_semaphore() + async with semaphore: + session_factory = get_session_factory() + async with session_factory() as session: + yield session diff --git a/app/services/ingestion/enrichment.py b/app/services/ingestion/enrichment.py index 1b66f58..181ff41 100644 --- a/app/services/ingestion/enrichment.py +++ b/app/services/ingestion/enrichment.py @@ -37,6 +37,40 @@ def _sanitize_titles(publications: list) -> list[str]: return titles +# OpenAlex (and most HTTP servers) struggle with very long query strings. +# Non-ASCII characters expand significantly when percent-encoded in URLs, +# so a fixed batch count is unreliable. Instead we cap the combined +# byte-length of the pipe-joined title filter value to stay well within +# safe URL limits (~4 KiB of filter text ≈ ~6 KiB when percent-encoded). +_MAX_TITLE_FILTER_BYTES = 4_000 + + +def _chunk_titles_by_url_length( + titles: list[str], + max_bytes: int = _MAX_TITLE_FILTER_BYTES, +) -> list[list[str]]: + """Split *titles* into chunks whose pipe-joined UTF-8 size stays under *max_bytes*.""" + chunks: list[list[str]] = [] + current_chunk: list[str] = [] + current_bytes = 0 + + for title in titles: + title_bytes = len(title.encode("utf-8")) + # Account for the pipe separator between titles + separator = 3 if current_chunk else 0 # len("|".encode) but url-encoded as %7C = 3 + if current_chunk and current_bytes + separator + title_bytes > max_bytes: + chunks.append(current_chunk) + current_chunk = [title] + current_bytes = title_bytes + else: + current_chunk.append(title) + current_bytes += separator + title_bytes + + if current_chunk: + chunks.append(current_chunk) + return chunks + + class EnrichmentRunner: """Post-run OpenAlex enrichment logic. @@ -156,21 +190,34 @@ class EnrichmentRunner: resolved_key = openalex_api_key or settings.openalex_api_key client = OpenAlexClient(api_key=resolved_key, mailto=settings.crossref_api_mailto) - batch_size = 25 now = datetime.now(UTC) arxiv_lookup_allowed = True - for i in range(0, len(publications), batch_size): + all_titles = _sanitize_titles(publications) + title_chunks = _chunk_titles_by_url_length(all_titles) + + # Build a mapping from sanitized title → publications for batch association + title_to_pubs: dict[str, list[Publication]] = {} + for p in publications: + raw = getattr(p, "title_raw", None) or getattr(p, "title", None) + if not raw or not raw.strip(): + continue + safe = " ".join(re.sub(r"[^\w\s]", " ", raw).split()) + if safe: + title_to_pubs.setdefault(safe, []).append(p) + + for title_chunk in title_chunks: if await self.run_is_canceled(db_session, run_id=run_id): structured_log(logger, "info", "ingestion.enrichment_aborted", run_id=run_id) return - batch = publications[i : i + batch_size] - titles = _sanitize_titles(batch) - if not titles: + batch = [] + for t in title_chunk: + batch.extend(title_to_pubs.get(t, [])) + if not batch: continue try: openalex_works = await client.get_works_by_filter( - {"title.search": "|".join(titles)}, limit=batch_size * 3 + {"title.search": "|".join(title_chunk)}, limit=len(title_chunk) * 3 ) except OpenAlexBudgetExhaustedError: structured_log(logger, "warning", "ingestion.openalex_budget_exhausted", run_id=run_id) @@ -289,19 +336,37 @@ class EnrichmentRunner: client = OpenAlexClient(api_key=settings.openalex_api_key, mailto=settings.crossref_api_mailto) - batch_size = 25 + all_titles = _sanitize_titles(publications) + title_chunks = _chunk_titles_by_url_length(all_titles) + + # Build title → publication mapping for batch association + title_to_pubs: dict[str, list[PublicationCandidate]] = {} + for p in publications: + raw = getattr(p, "title", None) + if not raw or not raw.strip(): + continue + safe = " ".join(re.sub(r"[^\w\s]", " ", raw).split()) + if safe: + title_to_pubs.setdefault(safe, []).append(p) + + # Collect publications that had no sanitizable title — pass through as-is + pubs_with_titles = set() + for titles in title_to_pubs.values(): + for p in titles: + pubs_with_titles.add(id(p)) + enriched: list[PublicationCandidate] = [] - for i in range(0, len(publications), batch_size): - batch = publications[i : i + batch_size] - titles = _sanitize_titles(batch) - if not titles: - enriched.extend(batch) + for title_chunk in title_chunks: + batch = [] + for t in title_chunk: + batch.extend(title_to_pubs.get(t, [])) + if not batch: continue try: openalex_works = await client.get_works_by_filter( - {"title.search": "|".join(titles)}, limit=batch_size * 3 + {"title.search": "|".join(title_chunk)}, limit=len(title_chunk) * 3 ) except Exception as e: structured_log( @@ -335,4 +400,10 @@ class EnrichmentRunner: enriched.append(new_p) else: enriched.append(p) + + # Append publications that had no sanitizable title (pass-through) + for p in publications: + if id(p) not in pubs_with_titles: + enriched.append(p) + return enriched diff --git a/app/services/ingestion/queue_runner.py b/app/services/ingestion/queue_runner.py index 2220f58..9f7bd1b 100644 --- a/app/services/ingestion/queue_runner.py +++ b/app/services/ingestion/queue_runner.py @@ -5,8 +5,8 @@ from datetime import UTC, datetime from sqlalchemy import select +from app.db.background_session import background_session from app.db.models import QueueItemStatus, RunTriggerType, ScholarProfile, UserSetting -from app.db.session import get_session_factory from app.logging_utils import structured_log from app.services.ingestion import queue as queue_service from app.services.ingestion.application import ( @@ -57,8 +57,7 @@ class QueueJobRunner: async def drain_continuation_queue(self) -> None: now = datetime.now(UTC) - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: jobs = await queue_service.list_due_jobs( session, now=now, @@ -73,8 +72,7 @@ class QueueJobRunner: ) -> bool: if job.attempt_count < self._continuation_max_attempts: return False - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: dropped = await queue_service.mark_dropped( session, job_id=job.id, @@ -98,8 +96,7 @@ class QueueJobRunner: self, job: queue_service.ContinuationQueueJob, ) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: queue_item = await queue_service.mark_retrying(session, job_id=job.id) await session.commit() if queue_item is None: @@ -110,8 +107,7 @@ class QueueJobRunner: self, job: queue_service.ContinuationQueueJob, ) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: scholar_result = await session.execute( select(ScholarProfile.id).where( ScholarProfile.user_id == job.user_id, @@ -140,8 +136,7 @@ class QueueJobRunner: return False async def _reschedule_queue_job_lock_active(self, job: queue_service.ContinuationQueueJob) -> None: - session_factory = get_session_factory() - async with session_factory() as recovery_session: + async with background_session() as recovery_session: await queue_service.reschedule_job( recovery_session, job_id=job.id, @@ -167,8 +162,7 @@ class QueueJobRunner: self._tick_seconds, int(exc.safety_state.get("cooldown_remaining_seconds") or 0), ) - session_factory = get_session_factory() - async with session_factory() as recovery_session: + async with background_session() as recovery_session: await queue_service.reschedule_job( recovery_session, job_id=job.id, @@ -193,8 +187,7 @@ class QueueJobRunner: *, exc: Exception, ) -> None: - session_factory = get_session_factory() - async with session_factory() as recovery_session: + async with background_session() as recovery_session: queue_item = await queue_service.increment_attempt_count(recovery_session, job_id=job.id) if queue_item is None: await recovery_session.commit() @@ -238,8 +231,7 @@ class QueueJobRunner: ) async def _load_request_delay_for_user(self, user_id: int, *, floor: int) -> int: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: result = await session.execute( select(UserSetting.request_delay_seconds).where(UserSetting.user_id == user_id) ) @@ -253,8 +245,7 @@ class QueueJobRunner: request_delay_floor: int, ): request_delay_seconds = await self._load_request_delay_for_user(job.user_id, floor=request_delay_floor) - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: ingestion = ScholarIngestionService(source=self._source) try: return await ingestion.run_for_user( @@ -366,8 +357,7 @@ class QueueJobRunner: ) async def _finalize_queue_job_after_run(self, job: queue_service.ContinuationQueueJob, run_summary) -> None: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: if int(run_summary.failed_count) <= 0: await self._finalize_successful_queue_job(session, job, run_summary) else: diff --git a/app/services/ingestion/scheduler.py b/app/services/ingestion/scheduler.py index bf4e9e7..e44cb45 100644 --- a/app/services/ingestion/scheduler.py +++ b/app/services/ingestion/scheduler.py @@ -8,13 +8,13 @@ from typing import Any from sqlalchemy import select +from app.db.background_session import background_session from app.db.models import ( CrawlRun, RunTriggerType, User, UserSetting, ) -from app.db.session import get_session_factory from app.logging_utils import structured_log from app.services.ingestion.application import ( RunAlreadyInProgressError, @@ -148,8 +148,7 @@ class SchedulerService: await self._run_candidate(candidate) async def _load_candidate_rows(self) -> list[Any]: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: result = await session.execute( select( UserSetting.user_id, @@ -204,8 +203,7 @@ class SchedulerService: return candidates async def _is_due(self, candidate: _AutoRunCandidate, *, now: datetime) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: result = await session.execute( select(CrawlRun.start_dt) .where( @@ -227,8 +225,7 @@ class SchedulerService: *, candidate: _AutoRunCandidate, ): - session_factory = get_session_factory() - async with session_factory() as session: + async with background_session() as session: ingestion = ScholarIngestionService(source=self._source) try: return await ingestion.run_for_user( @@ -289,9 +286,12 @@ class SchedulerService: async def _drain_pdf_queue(self) -> None: from app.services.publications.pdf_queue import drain_ready_jobs + from app.services.publications.pdf_queue_resolution import is_budget_cooldown_active - session_factory = get_session_factory() - async with session_factory() as session: + if is_budget_cooldown_active(): + return + + async with background_session() as session: try: processed = await drain_ready_jobs( session, diff --git a/app/services/ingestion/scholar_processing.py b/app/services/ingestion/scholar_processing.py index f41d904..1735915 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,22 @@ 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}, + ) + + await _emit() first_pass_cstarts = await _run_first_pass( db_session, scholars=scholars, @@ -407,8 +432,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 +453,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/app/services/portability/exporting.py b/app/services/portability/exporting.py index 0589498..a11eaef 100644 --- a/app/services/portability/exporting.py +++ b/app/services/portability/exporting.py @@ -56,11 +56,14 @@ async def export_user_data( db_session: AsyncSession, *, user_id: int, + scholar_profile_ids: list[int] | None = None, ) -> dict[str, Any]: - scholars_result = await db_session.execute( - select(ScholarProfile).where(ScholarProfile.user_id == user_id).order_by(ScholarProfile.id.asc()) - ) - publication_result = await db_session.execute( + scholar_query = select(ScholarProfile).where(ScholarProfile.user_id == user_id) + if scholar_profile_ids: + scholar_query = scholar_query.where(ScholarProfile.id.in_(scholar_profile_ids)) + scholars_result = await db_session.execute(scholar_query.order_by(ScholarProfile.id.asc())) + + pub_query = ( select( ScholarProfile.scholar_id, Publication.cluster_id, @@ -77,8 +80,13 @@ async def export_user_data( .join(ScholarPublication, ScholarPublication.scholar_profile_id == ScholarProfile.id) .join(Publication, Publication.id == ScholarPublication.publication_id) .where(ScholarProfile.user_id == user_id) - .order_by(ScholarPublication.created_at.desc(), Publication.id.desc()) ) + if scholar_profile_ids: + pub_query = pub_query.where(ScholarProfile.id.in_(scholar_profile_ids)) + publication_result = await db_session.execute( + pub_query.order_by(ScholarPublication.created_at.desc(), Publication.id.desc()) + ) + scholars = [_serialize_export_scholar(profile) for profile in scholars_result.scalars().all()] publications = [_serialize_export_publication(row) for row in publication_result.all()] return { diff --git a/app/services/publications/pdf_queue_resolution.py b/app/services/publications/pdf_queue_resolution.py index ec4422e..ac71edd 100644 --- a/app/services/publications/pdf_queue_resolution.py +++ b/app/services/publications/pdf_queue_resolution.py @@ -2,9 +2,10 @@ from __future__ import annotations import asyncio import logging +from datetime import UTC, datetime, timedelta +from app.db.background_session import background_session from app.db.models import Publication, PublicationPdfJob -from app.db.session import get_session_factory from app.logging_utils import structured_log from app.services.publication_identifiers import application as identifier_service from app.services.publications.pdf_queue_common import ( @@ -29,8 +30,20 @@ PDF_EVENT_ATTEMPT_STARTED = "attempt_started" PDF_EVENT_RESOLVED = "resolved" PDF_EVENT_FAILED = "failed" +_BUDGET_COOLDOWN_MINUTES = 15 + logger = logging.getLogger(__name__) _scheduled_tasks: set[asyncio.Task[None]] = set() +_budget_cooldown_until: datetime | None = None + + +def is_budget_cooldown_active() -> bool: + return _budget_cooldown_until is not None and datetime.now(UTC) < _budget_cooldown_until + + +def _enter_budget_cooldown() -> None: + global _budget_cooldown_until + _budget_cooldown_until = datetime.now(UTC) + timedelta(minutes=_BUDGET_COOLDOWN_MINUTES) async def _mark_attempt_started( @@ -38,8 +51,7 @@ async def _mark_attempt_started( publication_id: int, user_id: int, ) -> None: - session_factory = get_session_factory() - async with session_factory() as db_session: + async with background_session() as db_session: job = await db_session.get(PublicationPdfJob, publication_id) if job is None: job = queued_job(publication_id=publication_id, user_id=user_id) @@ -125,8 +137,7 @@ async def _persist_outcome( user_id: int, outcome: OaResolutionOutcome, ) -> None: - session_factory = get_session_factory() - async with session_factory() as db_session: + async with background_session() as db_session: publication = await db_session.get(Publication, publication_id) job = await db_session.get(PublicationPdfJob, publication_id) if publication is None or job is None: @@ -207,8 +218,7 @@ async def _run_resolution_task( openalex_api_key: str | None = None try: - session_factory = get_session_factory() - async with session_factory() as key_session: + async with background_session() as key_session: user_settings = await user_settings_service.get_or_create_settings(key_session, user_id=user_id) openalex_api_key = getattr(user_settings, "openalex_api_key", None) or settings.openalex_api_key except Exception: @@ -233,11 +243,13 @@ async def _run_resolution_task( detail="arXiv temporarily disabled for remaining batch after rate limit", ) except OpenAlexBudgetExhaustedError: + _enter_budget_cooldown() structured_log( logger, "warning", "pdf_queue.budget_exhausted", detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted", + cooldown_minutes=_BUDGET_COOLDOWN_MINUTES, ) break except Exception: diff --git a/app/services/scholars/application.py b/app/services/scholars/application.py index 894c371..3496732 100644 --- a/app/services/scholars/application.py +++ b/app/services/scholars/application.py @@ -1,6 +1,7 @@ from __future__ import annotations import os +from typing import Any from uuid import uuid4 from sqlalchemy.exc import IntegrityError @@ -10,7 +11,11 @@ from app.db.models import ScholarProfile from app.services.scholar.parser import ScholarParserError, parse_profile_page from app.services.scholar.source import ScholarSource from app.services.scholars.author_search import search_author_candidates -from app.services.scholars.constants import ALLOWED_IMAGE_UPLOAD_CONTENT_TYPES +from app.services.scholars.constants import ( + ALLOWED_IMAGE_UPLOAD_CONTENT_TYPES, + SEARCH_COOLDOWN_REASON, + SEARCH_DISABLED_REASON, +) from app.services.scholars.exceptions import ScholarServiceError from app.services.scholars.uploads import ( _ensure_upload_root, @@ -23,8 +28,66 @@ from app.services.scholars.validators import ( validate_scholar_id, ) + +async def bulk_delete_scholars( + db_session: AsyncSession, + *, + user_id: int, + scholar_profile_ids: list[int], + upload_dir: str | None = None, +) -> int: + from sqlalchemy import select + + result = await db_session.execute( + select(ScholarProfile).where( + ScholarProfile.id.in_(scholar_profile_ids), + ScholarProfile.user_id == user_id, + ) + ) + profiles = list(result.scalars().all()) + if not profiles: + return 0 + if upload_dir: + upload_root = _ensure_upload_root(upload_dir, create=True) + for profile in profiles: + _safe_remove_upload(upload_root, profile.profile_image_upload_path) + for profile in profiles: + await db_session.delete(profile) + try: + await db_session.commit() + except IntegrityError as exc: + await db_session.rollback() + raise ScholarServiceError("Unable to bulk-delete scholars due to a database constraint.") from exc + return len(profiles) + + +async def bulk_toggle_scholars( + db_session: AsyncSession, + *, + user_id: int, + scholar_profile_ids: list[int], + is_enabled: bool, +) -> int: + from sqlalchemy import CursorResult, update + + cursor: CursorResult[Any] = await db_session.execute( # type: ignore[assignment] + update(ScholarProfile) + .where( + ScholarProfile.id.in_(scholar_profile_ids), + ScholarProfile.user_id == user_id, + ) + .values(is_enabled=is_enabled) + ) + await db_session.commit() + return int(cursor.rowcount or 0) + + __all__ = [ + "SEARCH_COOLDOWN_REASON", + "SEARCH_DISABLED_REASON", "ScholarServiceError", + "bulk_delete_scholars", + "bulk_toggle_scholars", "clear_profile_image_customization", "create_scholar_for_user", "delete_scholar", @@ -119,7 +182,11 @@ async def delete_scholar( _safe_remove_upload(upload_root, profile.profile_image_upload_path) await db_session.delete(profile) - await db_session.commit() + try: + await db_session.commit() + except IntegrityError as exc: + await db_session.rollback() + raise ScholarServiceError("Unable to delete scholar due to a database constraint.") from exc async def hydrate_profile_metadata( diff --git a/app/settings.py b/app/settings.py index c9098a1..40596d1 100644 --- a/app/settings.py +++ b/app/settings.py @@ -275,6 +275,8 @@ class Settings: crossref_max_lookups_per_request: int = _env_int("CROSSREF_MAX_LOOKUPS_PER_REQUEST", 8) openalex_api_key: str | None = os.getenv("OPENALEX_API_KEY") + database_reserved_api_connections: int = _env_int("DATABASE_RESERVED_API_CONNECTIONS", 3) + crossref_api_token: str | None = os.getenv("CROSSREF_API_TOKEN") crossref_api_mailto: str | None = os.getenv("CROSSREF_API_MAILTO") diff --git a/frontend/src/components/patterns/ScrapeSafetyBadge.test.ts b/frontend/src/components/patterns/ScrapeSafetyBadge.test.ts new file mode 100644 index 0000000..8d17969 --- /dev/null +++ b/frontend/src/components/patterns/ScrapeSafetyBadge.test.ts @@ -0,0 +1,61 @@ +// @vitest-environment happy-dom +import { describe, expect, it } from "vitest"; +import { mount } from "@vue/test-utils"; +import ScrapeSafetyBadge from "./ScrapeSafetyBadge.vue"; +import { createDefaultSafetyState, type ScrapeSafetyState } from "@/features/safety"; + +function buildState(overrides: Partial = {}): ScrapeSafetyState { + return { ...createDefaultSafetyState(), ...overrides }; +} + +describe("ScrapeSafetyBadge", () => { + it("shows ready tooltip when cooldown is inactive", () => { + const wrapper = mount(ScrapeSafetyBadge, { + props: { state: buildState({ cooldown_active: false }) }, + }); + expect(wrapper.text()).toContain("Safety ready"); + const hint = wrapper.findComponent({ name: "AppHelpHint" }); + expect(hint.exists()).toBe(true); + expect(hint.props("text")).toContain("No active cooldown"); + }); + + it("shows cooldown tooltip with reason and action when active", () => { + const wrapper = mount(ScrapeSafetyBadge, { + props: { + state: buildState({ + cooldown_active: true, + cooldown_reason: "blocked_failure_threshold_exceeded", + cooldown_reason_label: "Too many blocked requests", + recommended_action: "Wait for cooldown to expire", + cooldown_remaining_seconds: 120, + }), + }, + }); + expect(wrapper.text()).toContain("Safety cooldown"); + const hint = wrapper.findComponent({ name: "AppHelpHint" }); + expect(hint.exists()).toBe(true); + const text = hint.props("text") as string; + expect(text).toContain("Google Scholar rate-limits"); + expect(text).toContain("Too many blocked requests"); + expect(text).toContain("Wait for cooldown to expire"); + }); + + it("shows cooldown tooltip without optional fields", () => { + const wrapper = mount(ScrapeSafetyBadge, { + props: { + state: buildState({ + cooldown_active: true, + cooldown_reason: "some_reason", + cooldown_reason_label: null, + recommended_action: null, + cooldown_remaining_seconds: 60, + }), + }, + }); + const hint = wrapper.findComponent({ name: "AppHelpHint" }); + const text = hint.props("text") as string; + expect(text).toContain("Google Scholar rate-limits"); + expect(text).not.toContain("Why:"); + expect(text).not.toContain("Action:"); + }); +}); diff --git a/frontend/src/components/patterns/ScrapeSafetyBadge.vue b/frontend/src/components/patterns/ScrapeSafetyBadge.vue index b40c1c9..845fb2f 100644 --- a/frontend/src/components/patterns/ScrapeSafetyBadge.vue +++ b/frontend/src/components/patterns/ScrapeSafetyBadge.vue @@ -1,6 +1,7 @@ diff --git a/frontend/src/components/patterns/ScrapeSafetyBanner.vue b/frontend/src/components/patterns/ScrapeSafetyBanner.vue index 300b05a..6ea343d 100644 --- a/frontend/src/components/patterns/ScrapeSafetyBanner.vue +++ b/frontend/src/components/patterns/ScrapeSafetyBanner.vue @@ -2,6 +2,7 @@ import { computed, onBeforeUnmount, onMounted, ref } from "vue"; import AppAlert from "@/components/ui/AppAlert.vue"; +import AppHelpHint from "@/components/ui/AppHelpHint.vue"; import { formatCooldownCountdown, type ScrapeSafetyState, @@ -79,6 +80,10 @@ const actionText = computed(() => { return props.safetyState.recommended_action; }); +const bannerHintText = computed(() => { + return "Google Scholar rate-limits automated requests. The cooldown pauses scraping to avoid your IP being blocked."; +}); + onMounted(() => { timer = setInterval(() => { now.value = Date.now(); @@ -95,7 +100,12 @@ onBeforeUnmount(() => {