From b7015837169e1440b706f2f22940d871c970556c Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Fri, 27 Feb 2026 14:39:12 +0100 Subject: [PATCH] refactor: decompose scholars service and pdf_queue into focused modules Co-Authored-By: Claude Opus 4.6 --- DECOMPOSITION_SLICES.md | 17 +- .../publication_identifiers/application.py | 2 +- app/services/publications/application.py | 4 +- app/services/publications/pdf_queue.py | 638 +------------- .../publications/pdf_queue_queries.py | 395 +++++++++ .../publications/pdf_queue_resolution.py | 311 +++++++ app/services/scholars/application.py | 823 +----------------- app/services/scholars/author_search.py | 580 ++++++++++++ app/services/scholars/author_search_cache.py | 239 +++++ .../unit/test_publication_pdf_queue_policy.py | 26 +- 10 files changed, 1592 insertions(+), 1443 deletions(-) create mode 100644 app/services/publications/pdf_queue_queries.py create mode 100644 app/services/publications/pdf_queue_resolution.py create mode 100644 app/services/scholars/author_search.py create mode 100644 app/services/scholars/author_search_cache.py diff --git a/DECOMPOSITION_SLICES.md b/DECOMPOSITION_SLICES.md index c65cb2a..f14f072 100644 --- a/DECOMPOSITION_SLICES.md +++ b/DECOMPOSITION_SLICES.md @@ -28,7 +28,7 @@ This file contains self-contained decomposition prompts ("slices") to bring the | 1 | Schema split | **DONE** | | 2 | Ingestion: pagination + publication upsert | **DONE** | | 3 | Ingestion: enrichment + scholar processing + run completion | **DONE** | -| 4 | Scholars service + PDF queue | pending | +| 4 | Scholars service + PDF queue | **DONE** | | 5 | Routers + scheduler | pending | --- @@ -132,13 +132,20 @@ Commit message: `refactor: extract enrichment, scholar processing, and run compl --- -## Slice 4: Decompose `app/services/scholars/application.py` (996 lines) and `app/services/publications/pdf_queue.py` (969 lines) +## Slice 4: Decompose `app/services/scholars/application.py` (996 lines) and `app/services/publications/pdf_queue.py` (969 lines) — DONE -**Why:** Two service files nearly 2x the hard ceiling. Each has clear internal seams. +Completed. Extracted four modules from two oversized service files: -**Files to create:** `app/services/scholars/author_search.py`, `app/services/scholars/author_search_cache.py`, `app/services/publications/pdf_queue_queries.py`, `app/services/publications/pdf_queue_resolution.py` +| File | Lines | Contents | +|---|---|---| +| `app/services/scholars/author_search_cache.py` | 239 | Serialize/deserialize cache entries, cache get/set/prune | +| `app/services/scholars/author_search.py` | 580 | Cooldown, throttle, retry, circuit breaker, `search_author_candidates` | +| `app/services/scholars/application.py` | 221 | Scholar CRUD only (list, create, get, toggle, delete, image ops) | +| `app/services/publications/pdf_queue_queries.py` | 395 | SQL builders, row hydration, listing/counting/pagination, retry item builders | +| `app/services/publications/pdf_queue_resolution.py` | 311 | Task execution, outcome persistence, scheduling (`schedule_rows`) | +| `app/services/publications/pdf_queue.py` | 364 | Dataclasses, constants, enqueueing logic, public API | -**Files to modify:** `app/services/scholars/application.py`, `app/services/publications/pdf_queue.py` +Also updated one test file (`test_publication_pdf_queue_policy.py`) to monkeypatch `pdf_queue_resolution` instead of `pdf_queue` for the 4 resolution-related tests, and updated `publications/application.py` and `publication_identifiers/application.py` imports. **Prompt:** diff --git a/app/services/publication_identifiers/application.py b/app/services/publication_identifiers/application.py index 7724f46..560211e 100644 --- a/app/services/publication_identifiers/application.py +++ b/app/services/publication_identifiers/application.py @@ -21,7 +21,7 @@ from app.services.publication_identifiers.types import ( ) if TYPE_CHECKING: - from app.services.publications.pdf_queue import PdfQueueListItem + from app.services.publications.pdf_queue_queries import PdfQueueListItem from app.services.publications.types import PublicationListItem, UnreadPublicationItem CONFIDENCE_HIGH = 0.98 diff --git a/app/services/publications/application.py b/app/services/publications/application.py index 023903c..61cf9ad 100644 --- a/app/services/publications/application.py +++ b/app/services/publications/application.py @@ -24,9 +24,11 @@ from app.services.publications.modes import ( resolve_publication_view_mode, ) from app.services.publications.pdf_queue import ( - count_pdf_queue_items, enqueue_all_missing_pdf_jobs, enqueue_retry_pdf_job_for_publication_id, +) +from app.services.publications.pdf_queue_queries import ( + count_pdf_queue_items, list_pdf_queue_items, list_pdf_queue_page, ) diff --git a/app/services/publications/pdf_queue.py b/app/services/publications/pdf_queue.py index 96ec41b..eef0d61 100644 --- a/app/services/publications/pdf_queue.py +++ b/app/services/publications/pdf_queue.py @@ -1,33 +1,23 @@ from __future__ import annotations -import asyncio import logging from dataclasses import dataclass -from datetime import UTC, datetime, timedelta +from datetime import UTC, datetime -from sqlalchemy import Select, and_, func, literal, or_, select, union_all +from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import ( - Publication, PublicationPdfJob, PublicationPdfJobEvent, - ScholarProfile, - ScholarPublication, User, ) -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.publication_identifiers.types import DisplayIdentifier -from app.services.publications.pdf_resolution_pipeline import ( - resolve_publication_pdf_outcome_for_row, +from app.services.publications.pdf_queue_queries import ( + missing_pdf_candidates, + retry_item_for_publication_id, ) +from app.services.publications.pdf_queue_resolution import schedule_rows from app.services.publications.types import PublicationListItem -from app.services.unpaywall.application import ( - FAILURE_RESOLUTION_EXCEPTION, - OaResolutionOutcome, -) from app.settings import settings PDF_STATUS_UNTRACKED = "untracked" @@ -37,31 +27,8 @@ PDF_STATUS_RESOLVED = "resolved" PDF_STATUS_FAILED = "failed" PDF_EVENT_QUEUED = "queued" -PDF_EVENT_ATTEMPT_STARTED = "attempt_started" -PDF_EVENT_RESOLVED = "resolved" -PDF_EVENT_FAILED = "failed" logger = logging.getLogger(__name__) -_scheduled_tasks: set[asyncio.Task[None]] = set() - - -@dataclass(frozen=True) -class PdfQueueListItem: - publication_id: int - title: str - pdf_url: str | None - status: str - attempt_count: int - last_failure_reason: str | None - last_failure_detail: str | None - last_source: str | None - requested_by_user_id: int | None - requested_by_email: str | None - queued_at: datetime | None - last_attempt_at: datetime | None - resolved_at: datetime | None - updated_at: datetime - display_identifier: DisplayIdentifier | None = None @dataclass(frozen=True) @@ -76,14 +43,6 @@ class PdfBulkQueueResult: queued_count: int -@dataclass(frozen=True) -class PdfQueuePage: - items: list[PdfQueueListItem] - total_count: int - limit: int - offset: int - - def _utcnow() -> datetime: return datetime.now(UTC) @@ -138,14 +97,6 @@ def _queueable_rows( return candidates[:bounded] -def _bounded_limit(limit: int, *, max_value: int = 500) -> int: - return max(1, min(int(limit), max_value)) - - -def _bounded_offset(offset: int) -> int: - return max(int(offset), 0) - - def _auto_retry_interval_seconds() -> int: return max(int(settings.pdf_auto_retry_interval_seconds), 1) @@ -202,18 +153,12 @@ def _event_row( user_id: int | None, event_type: str, status: str | None, - source: str | None = None, - failure_reason: str | None = None, - message: str | None = None, ) -> PublicationPdfJobEvent: return PublicationPdfJobEvent( publication_id=publication_id, user_id=user_id, event_type=event_type, status=status, - source=source, - failure_reason=failure_reason, - message=message, ) @@ -310,247 +255,9 @@ async def _enqueue_rows( return queued -def _register_task(task: asyncio.Task[None]) -> None: - _scheduled_tasks.add(task) - - -def _drop_finished_task(task: asyncio.Task[None]) -> None: - _scheduled_tasks.discard(task) - try: - task.result() - except Exception: - logger.exception("publications.pdf_queue.task_failed") - - -async def _mark_attempt_started( - *, - publication_id: int, - user_id: int, -) -> None: - session_factory = get_session_factory() - async with session_factory() 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) - db_session.add(job) - job.status = PDF_STATUS_RUNNING - job.last_attempt_at = _utcnow() - job.attempt_count = int(job.attempt_count) + 1 - db_session.add( - _event_row( - publication_id=publication_id, - user_id=user_id, - event_type=PDF_EVENT_ATTEMPT_STARTED, - status=PDF_STATUS_RUNNING, - ) - ) - await db_session.commit() - - -def _failed_outcome( - *, - row: PublicationListItem, -) -> OaResolutionOutcome: - return OaResolutionOutcome( - publication_id=row.publication_id, - doi=None, - pdf_url=None, - failure_reason=FAILURE_RESOLUTION_EXCEPTION, - source=None, - used_crossref=False, - ) - - -async def _fetch_outcome_for_row( - *, - row: PublicationListItem, - request_email: str | None, - openalex_api_key: str | None = None, - allow_arxiv_lookup: bool = True, -) -> tuple[OaResolutionOutcome, bool]: - pipeline_result = await resolve_publication_pdf_outcome_for_row( - row=row, - request_email=request_email, - openalex_api_key=openalex_api_key, - allow_arxiv_lookup=allow_arxiv_lookup, - ) - outcome = pipeline_result.outcome - if outcome is not None: - return outcome, bool(pipeline_result.arxiv_rate_limited) - return _failed_outcome(row=row), bool(pipeline_result.arxiv_rate_limited) - - -def _apply_publication_update( - publication: Publication, - *, - pdf_url: str | None, -) -> None: - if pdf_url and publication.pdf_url != pdf_url: - publication.pdf_url = pdf_url - - -def _apply_job_outcome(job: PublicationPdfJob, *, outcome: OaResolutionOutcome) -> None: - job.last_source = outcome.source - if outcome.pdf_url: - job.status = PDF_STATUS_RESOLVED - job.resolved_at = _utcnow() - job.last_failure_reason = None - job.last_failure_detail = None - return - job.status = PDF_STATUS_FAILED - job.last_failure_reason = outcome.failure_reason - job.last_failure_detail = outcome.failure_reason - - -def _result_event(outcome: OaResolutionOutcome) -> tuple[str, str]: - if outcome.pdf_url: - return PDF_EVENT_RESOLVED, PDF_STATUS_RESOLVED - return PDF_EVENT_FAILED, PDF_STATUS_FAILED - - -async def _persist_outcome( - *, - publication_id: int, - user_id: int, - outcome: OaResolutionOutcome, -) -> None: - session_factory = get_session_factory() - async with session_factory() 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: - return - _apply_publication_update(publication, pdf_url=outcome.pdf_url) - await identifier_service.sync_identifiers_for_publication_resolution( - db_session, - publication=publication, - source=outcome.source, - ) - _apply_job_outcome(job, outcome=outcome) - event_type, status = _result_event(outcome) - db_session.add( - _event_row( - publication_id=publication_id, - user_id=user_id, - event_type=event_type, - status=status, - source=outcome.source, - failure_reason=outcome.failure_reason, - message=outcome.failure_reason, - ) - ) - await db_session.commit() - - -async def _resolve_publication_row( - *, - user_id: int, - request_email: str | None, - row: PublicationListItem, - openalex_api_key: str | None = None, - allow_arxiv_lookup: bool = True, -) -> bool: - from app.services.openalex.client import OpenAlexBudgetExhaustedError - - await _mark_attempt_started(publication_id=row.publication_id, user_id=user_id) - try: - outcome, arxiv_rate_limited = await _fetch_outcome_for_row( - row=row, - request_email=request_email, - openalex_api_key=openalex_api_key, - allow_arxiv_lookup=allow_arxiv_lookup, - ) - except OpenAlexBudgetExhaustedError: - # Persist a terminal outcome so jobs do not remain stuck in "running". - await _persist_outcome( - publication_id=row.publication_id, - user_id=user_id, - outcome=_failed_outcome(row=row), - ) - # Propagate upward so the batch loop can stop immediately. - raise - except Exception as exc: # pragma: no cover - defensive network boundary - structured_log( - logger, - "warning", - "publications.pdf_queue.resolve_failed", - publication_id=row.publication_id, - error=str(exc), - ) - outcome = _failed_outcome(row=row) - arxiv_rate_limited = False - await _persist_outcome( - publication_id=row.publication_id, - user_id=user_id, - outcome=outcome, - ) - return bool(arxiv_rate_limited) - - -async def _run_resolution_task( - *, - user_id: int, - request_email: str | None, - rows: list[PublicationListItem], -) -> None: - from app.services.openalex.client import OpenAlexBudgetExhaustedError - from app.services.settings import application as user_settings_service - - # Resolve the best available API key: per-user setting → env var fallback. - openalex_api_key: str | None = None - try: - session_factory = get_session_factory() - async with session_factory() 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: - openalex_api_key = settings.openalex_api_key - - arxiv_lookup_allowed = True - for row in rows: - try: - arxiv_rate_limited = await _resolve_publication_row( - user_id=user_id, - request_email=request_email, - row=row, - openalex_api_key=openalex_api_key, - allow_arxiv_lookup=arxiv_lookup_allowed, - ) - if arxiv_rate_limited and arxiv_lookup_allowed: - arxiv_lookup_allowed = False - structured_log( - logger, - "warning", - "pdf_queue.arxiv_batch_disabled", - detail="arXiv temporarily disabled for remaining batch after rate limit", - ) - except OpenAlexBudgetExhaustedError: - structured_log( - logger, - "warning", - "pdf_queue.budget_exhausted", - detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted", - ) - break - - -def _schedule_rows( - *, - user_id: int, - request_email: str | None, - rows: list[PublicationListItem], -) -> None: - if not rows: - return - task = asyncio.create_task( - _run_resolution_task( - user_id=user_id, - request_email=request_email, - rows=rows, - ) - ) - _register_task(task) - task.add_done_callback(_drop_finished_task) +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- async def enqueue_missing_pdf_jobs( @@ -568,7 +275,7 @@ async def enqueue_missing_pdf_jobs( rows=queueable, force_retry=False, ) - _schedule_rows(user_id=user_id, request_email=request_email, rows=queued_rows) + schedule_rows(user_id=user_id, request_email=request_email, rows=queued_rows) return [row.publication_id for row in queued_rows] @@ -585,69 +292,10 @@ async def enqueue_retry_pdf_job( rows=[row], force_retry=True, ) - _schedule_rows(user_id=user_id, request_email=request_email, rows=queued_rows) + schedule_rows(user_id=user_id, request_email=request_email, rows=queued_rows) return bool(queued_rows) -def _retry_item_label(display_name: str | None, scholar_id: str | None) -> str: - return str(display_name or scholar_id or "unknown") - - -def _retry_item_from_publication( - publication: Publication, - *, - link_row: tuple | None, -) -> PublicationListItem: - if link_row is None: - scholar_profile_id = 0 - scholar_label = "unknown" - is_read = True - first_seen_at = publication.created_at or _utcnow() - else: - scholar_profile_id = int(link_row[0]) - scholar_label = _retry_item_label(link_row[1], link_row[2]) - is_read = bool(link_row[3]) - first_seen_at = link_row[4] or publication.created_at or _utcnow() - return PublicationListItem( - publication_id=int(publication.id), - scholar_profile_id=scholar_profile_id, - scholar_label=scholar_label, - title=publication.title_raw, - year=publication.year, - citation_count=int(publication.citation_count or 0), - venue_text=publication.venue_text, - pub_url=publication.pub_url, - pdf_url=publication.pdf_url, - is_read=is_read, - first_seen_at=first_seen_at, - is_new_in_latest_run=False, - ) - - -async def _retry_item_for_publication_id( - db_session: AsyncSession, - *, - publication_id: int, -) -> PublicationListItem | None: - publication = await db_session.get(Publication, publication_id) - if publication is None: - return None - result = await db_session.execute( - select( - ScholarProfile.id, - ScholarProfile.display_name, - ScholarProfile.scholar_id, - ScholarPublication.is_read, - ScholarPublication.created_at, - ) - .join(ScholarProfile, ScholarProfile.id == ScholarPublication.scholar_profile_id) - .where(ScholarPublication.publication_id == publication_id) - .order_by(ScholarPublication.created_at.asc()) - .limit(1) - ) - return _retry_item_from_publication(publication, link_row=result.one_or_none()) - - async def enqueue_retry_pdf_job_for_publication_id( db_session: AsyncSession, *, @@ -655,7 +303,7 @@ async def enqueue_retry_pdf_job_for_publication_id( request_email: str | None, publication_id: int, ) -> PdfRequeueResult: - row = await _retry_item_for_publication_id( + row = await retry_item_for_publication_id( db_session, publication_id=publication_id, ) @@ -670,54 +318,6 @@ async def enqueue_retry_pdf_job_for_publication_id( return PdfRequeueResult(publication_exists=True, queued=queued) -def _queue_candidate_from_publication(publication: Publication) -> PublicationListItem: - return PublicationListItem( - publication_id=int(publication.id), - scholar_profile_id=0, - scholar_label="", - title=publication.title_raw, - year=publication.year, - citation_count=int(publication.citation_count or 0), - venue_text=publication.venue_text, - pub_url=publication.pub_url, - pdf_url=publication.pdf_url, - is_read=True, - first_seen_at=publication.created_at or _utcnow(), - is_new_in_latest_run=False, - ) - - -async def _missing_pdf_candidates( - db_session: AsyncSession, - *, - limit: int, -) -> list[PublicationListItem]: - bounded_limit = max(1, min(int(limit), 5000)) - now = datetime.now(UTC) - cooldown_threshold = now - timedelta(days=7) - - result = await db_session.execute( - select(Publication) - .outerjoin(PublicationPdfJob, PublicationPdfJob.publication_id == Publication.id) - .where(Publication.pdf_url.is_(None)) - .where( - or_( - PublicationPdfJob.publication_id.is_(None), - and_( - PublicationPdfJob.status.notin_([PDF_STATUS_QUEUED, PDF_STATUS_RUNNING]), - or_( - PublicationPdfJob.last_attempt_at.is_(None), - PublicationPdfJob.last_attempt_at < cooldown_threshold, - ), - ), - ) - ) - .order_by(Publication.updated_at.desc(), Publication.id.desc()) - .limit(bounded_limit) - ) - return [_queue_candidate_from_publication(publication) for publication in result.scalars()] - - async def enqueue_all_missing_pdf_jobs( db_session: AsyncSession, *, @@ -725,230 +325,20 @@ async def enqueue_all_missing_pdf_jobs( request_email: str | None, limit: int = 1000, ) -> PdfBulkQueueResult: - candidates = await _missing_pdf_candidates(db_session, limit=limit) + candidates = await missing_pdf_candidates(db_session, limit=limit) queued_rows = await _enqueue_rows( db_session, user_id=user_id, rows=candidates, force_retry=True, ) - _schedule_rows(user_id=user_id, request_email=request_email, rows=queued_rows) + schedule_rows(user_id=user_id, request_email=request_email, rows=queued_rows) return PdfBulkQueueResult( requested_count=len(candidates), queued_count=len(queued_rows), ) -def _tracked_queue_select_base(*, status: str | None) -> Select[tuple]: - stmt = ( - select( - PublicationPdfJob.publication_id, - Publication.title_raw, - Publication.pdf_url, - PublicationPdfJob.status, - PublicationPdfJob.attempt_count, - PublicationPdfJob.last_failure_reason, - PublicationPdfJob.last_failure_detail, - PublicationPdfJob.last_source, - PublicationPdfJob.last_requested_by_user_id, - User.email, - PublicationPdfJob.queued_at, - PublicationPdfJob.last_attempt_at, - PublicationPdfJob.resolved_at, - PublicationPdfJob.updated_at, - ) - .join(Publication, Publication.id == PublicationPdfJob.publication_id) - .outerjoin(User, User.id == PublicationPdfJob.last_requested_by_user_id) - ) - if status: - stmt = stmt.where(PublicationPdfJob.status == status) - return stmt - - -def _tracked_queue_select(*, limit: int, offset: int, status: str | None) -> Select[tuple]: - return ( - _tracked_queue_select_base(status=status) - .order_by(PublicationPdfJob.updated_at.desc()) - .limit(_bounded_limit(limit)) - .offset(_bounded_offset(offset)) - ) - - -def _untracked_queue_select_base() -> Select[tuple]: - return ( - select( - Publication.id, - Publication.title_raw, - Publication.pdf_url, - literal(PDF_STATUS_UNTRACKED), - literal(0), - literal(None), - literal(None), - literal(None), - literal(None), - literal(None), - literal(None), - literal(None), - literal(None), - Publication.updated_at, - ) - .outerjoin(PublicationPdfJob, PublicationPdfJob.publication_id == Publication.id) - .where(Publication.pdf_url.is_(None)) - .where(PublicationPdfJob.publication_id.is_(None)) - ) - - -def _untracked_queue_select(*, limit: int, offset: int) -> Select[tuple]: - return ( - _untracked_queue_select_base() - .order_by(Publication.updated_at.desc(), Publication.id.desc()) - .limit(_bounded_limit(limit)) - .offset(_bounded_offset(offset)) - ) - - -def _all_queue_select(*, limit: int, offset: int) -> Select[tuple]: - union_stmt = union_all( - _tracked_queue_select_base(status=None), - _untracked_queue_select_base(), - ).subquery() - return ( - select(union_stmt) - .order_by(union_stmt.c.updated_at.desc()) - .limit(_bounded_limit(limit)) - .offset(_bounded_offset(offset)) - ) - - -def _tracked_queue_count_select(*, status: str | None) -> Select[tuple]: - stmt = select(func.count()).select_from(PublicationPdfJob) - if status: - stmt = stmt.where(PublicationPdfJob.status == status) - return stmt - - -def _untracked_queue_count_select() -> Select[tuple]: - return ( - select(func.count()) - .select_from(Publication) - .outerjoin(PublicationPdfJob, PublicationPdfJob.publication_id == Publication.id) - .where(Publication.pdf_url.is_(None)) - .where(PublicationPdfJob.publication_id.is_(None)) - ) - - -def _queue_item_from_row(row: tuple) -> PdfQueueListItem: - return PdfQueueListItem( - publication_id=int(row[0]), - title=str(row[1] or ""), - pdf_url=row[2], - status=str(row[3] or PDF_STATUS_UNTRACKED), - attempt_count=int(row[4] or 0), - last_failure_reason=row[5], - last_failure_detail=row[6], - last_source=row[7], - requested_by_user_id=int(row[8]) if row[8] is not None else None, - requested_by_email=row[9], - queued_at=row[10], - last_attempt_at=row[11], - resolved_at=row[12], - updated_at=row[13], - ) - - -async def _hydrated_queue_items( - db_session: AsyncSession, - *, - rows: list[tuple], -) -> list[PdfQueueListItem]: - items = [_queue_item_from_row(row) for row in rows] - return await identifier_service.overlay_pdf_queue_items_with_display_identifiers( - db_session, - items=items, - ) - - -async def list_pdf_queue_items( - db_session: AsyncSession, - *, - limit: int = 100, - offset: int = 0, - status: str | None = None, -) -> list[PdfQueueListItem]: - bounded_limit = _bounded_limit(limit) - bounded_offset = _bounded_offset(offset) - normalized_status = (status or "").strip().lower() or None - if normalized_status == PDF_STATUS_UNTRACKED: - result = await db_session.execute( - _untracked_queue_select( - limit=bounded_limit, - offset=bounded_offset, - ) - ) - return await _hydrated_queue_items(db_session, rows=list(result.all())) - if normalized_status is None: - result = await db_session.execute( - _all_queue_select( - limit=bounded_limit, - offset=bounded_offset, - ) - ) - return await _hydrated_queue_items(db_session, rows=list(result.all())) - result = await db_session.execute( - _tracked_queue_select( - limit=bounded_limit, - offset=bounded_offset, - status=normalized_status, - ) - ) - return await _hydrated_queue_items(db_session, rows=list(result.all())) - - -async def count_pdf_queue_items( - db_session: AsyncSession, - *, - status: str | None = None, -) -> int: - normalized_status = (status or "").strip().lower() or None - if normalized_status == PDF_STATUS_UNTRACKED: - result = await db_session.execute(_untracked_queue_count_select()) - return int(result.scalar_one() or 0) - tracked_result = await db_session.execute(_tracked_queue_count_select(status=normalized_status)) - tracked_count = int(tracked_result.scalar_one() or 0) - if normalized_status is not None: - return tracked_count - untracked_result = await db_session.execute(_untracked_queue_count_select()) - untracked_count = int(untracked_result.scalar_one() or 0) - return tracked_count + untracked_count - - -async def list_pdf_queue_page( - db_session: AsyncSession, - *, - limit: int = 100, - offset: int = 0, - status: str | None = None, -) -> PdfQueuePage: - bounded_limit = _bounded_limit(limit) - bounded_offset = _bounded_offset(offset) - items = await list_pdf_queue_items( - db_session, - limit=bounded_limit, - offset=bounded_offset, - status=status, - ) - total_count = await count_pdf_queue_items( - db_session, - status=status, - ) - return PdfQueuePage( - items=items, - total_count=total_count, - limit=bounded_limit, - offset=bounded_offset, - ) - - async def drain_ready_jobs( db_session: AsyncSession, *, diff --git a/app/services/publications/pdf_queue_queries.py b/app/services/publications/pdf_queue_queries.py new file mode 100644 index 0000000..d8dad1f --- /dev/null +++ b/app/services/publications/pdf_queue_queries.py @@ -0,0 +1,395 @@ +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta + +from sqlalchemy import Select, and_, func, literal, or_, select, union_all +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import ( + Publication, + PublicationPdfJob, + ScholarProfile, + ScholarPublication, + User, +) +from app.services.publication_identifiers import application as identifier_service +from app.services.publication_identifiers.types import DisplayIdentifier +from app.services.publications.types import PublicationListItem + +PDF_STATUS_UNTRACKED = "untracked" +PDF_STATUS_QUEUED = "queued" +PDF_STATUS_RUNNING = "running" +PDF_STATUS_RESOLVED = "resolved" +PDF_STATUS_FAILED = "failed" + + +def _bounded_limit(limit: int, *, max_value: int = 500) -> int: + return max(1, min(int(limit), max_value)) + + +def _bounded_offset(offset: int) -> int: + return max(int(offset), 0) + + +def _utcnow() -> datetime: + return datetime.now(UTC) + + +def _retry_item_label(display_name: str | None, scholar_id: str | None) -> str: + return str(display_name or scholar_id or "unknown") + + +def _retry_item_from_publication( + publication: Publication, + *, + link_row: tuple | None, +) -> PublicationListItem: + if link_row is None: + scholar_profile_id = 0 + scholar_label = "unknown" + is_read = True + first_seen_at = publication.created_at or _utcnow() + else: + scholar_profile_id = int(link_row[0]) + scholar_label = _retry_item_label(link_row[1], link_row[2]) + is_read = bool(link_row[3]) + first_seen_at = link_row[4] or publication.created_at or _utcnow() + return PublicationListItem( + publication_id=int(publication.id), + scholar_profile_id=scholar_profile_id, + scholar_label=scholar_label, + title=publication.title_raw, + year=publication.year, + citation_count=int(publication.citation_count or 0), + venue_text=publication.venue_text, + pub_url=publication.pub_url, + pdf_url=publication.pdf_url, + is_read=is_read, + first_seen_at=first_seen_at, + is_new_in_latest_run=False, + ) + + +async def retry_item_for_publication_id( + db_session: AsyncSession, + *, + publication_id: int, +) -> PublicationListItem | None: + publication = await db_session.get(Publication, publication_id) + if publication is None: + return None + result = await db_session.execute( + select( + ScholarProfile.id, + ScholarProfile.display_name, + ScholarProfile.scholar_id, + ScholarPublication.is_read, + ScholarPublication.created_at, + ) + .join(ScholarProfile, ScholarProfile.id == ScholarPublication.scholar_profile_id) + .where(ScholarPublication.publication_id == publication_id) + .order_by(ScholarPublication.created_at.asc()) + .limit(1) + ) + return _retry_item_from_publication(publication, link_row=result.one_or_none()) + + +def _queue_candidate_from_publication(publication: Publication) -> PublicationListItem: + return PublicationListItem( + publication_id=int(publication.id), + scholar_profile_id=0, + scholar_label="", + title=publication.title_raw, + year=publication.year, + citation_count=int(publication.citation_count or 0), + venue_text=publication.venue_text, + pub_url=publication.pub_url, + pdf_url=publication.pdf_url, + is_read=True, + first_seen_at=publication.created_at or _utcnow(), + is_new_in_latest_run=False, + ) + + +async def missing_pdf_candidates( + db_session: AsyncSession, + *, + limit: int, +) -> list[PublicationListItem]: + bounded_limit = max(1, min(int(limit), 5000)) + now = datetime.now(UTC) + cooldown_threshold = now - timedelta(days=7) + + result = await db_session.execute( + select(Publication) + .outerjoin(PublicationPdfJob, PublicationPdfJob.publication_id == Publication.id) + .where(Publication.pdf_url.is_(None)) + .where( + or_( + PublicationPdfJob.publication_id.is_(None), + and_( + PublicationPdfJob.status.notin_([PDF_STATUS_QUEUED, PDF_STATUS_RUNNING]), + or_( + PublicationPdfJob.last_attempt_at.is_(None), + PublicationPdfJob.last_attempt_at < cooldown_threshold, + ), + ), + ) + ) + .order_by(Publication.updated_at.desc(), Publication.id.desc()) + .limit(bounded_limit) + ) + return [_queue_candidate_from_publication(publication) for publication in result.scalars()] + + +# --------------------------------------------------------------------------- +# Tracked / untracked queue SQL builders +# --------------------------------------------------------------------------- + + +def _tracked_queue_select_base(*, status: str | None) -> Select[tuple]: + stmt = ( + select( + PublicationPdfJob.publication_id, + Publication.title_raw, + Publication.pdf_url, + PublicationPdfJob.status, + PublicationPdfJob.attempt_count, + PublicationPdfJob.last_failure_reason, + PublicationPdfJob.last_failure_detail, + PublicationPdfJob.last_source, + PublicationPdfJob.last_requested_by_user_id, + User.email, + PublicationPdfJob.queued_at, + PublicationPdfJob.last_attempt_at, + PublicationPdfJob.resolved_at, + PublicationPdfJob.updated_at, + ) + .join(Publication, Publication.id == PublicationPdfJob.publication_id) + .outerjoin(User, User.id == PublicationPdfJob.last_requested_by_user_id) + ) + if status: + stmt = stmt.where(PublicationPdfJob.status == status) + return stmt + + +def _tracked_queue_select(*, limit: int, offset: int, status: str | None) -> Select[tuple]: + return ( + _tracked_queue_select_base(status=status) + .order_by(PublicationPdfJob.updated_at.desc()) + .limit(_bounded_limit(limit)) + .offset(_bounded_offset(offset)) + ) + + +def _untracked_queue_select_base() -> Select[tuple]: + return ( + select( + Publication.id, + Publication.title_raw, + Publication.pdf_url, + literal(PDF_STATUS_UNTRACKED), + literal(0), + literal(None), + literal(None), + literal(None), + literal(None), + literal(None), + literal(None), + literal(None), + literal(None), + Publication.updated_at, + ) + .outerjoin(PublicationPdfJob, PublicationPdfJob.publication_id == Publication.id) + .where(Publication.pdf_url.is_(None)) + .where(PublicationPdfJob.publication_id.is_(None)) + ) + + +def _untracked_queue_select(*, limit: int, offset: int) -> Select[tuple]: + return ( + _untracked_queue_select_base() + .order_by(Publication.updated_at.desc(), Publication.id.desc()) + .limit(_bounded_limit(limit)) + .offset(_bounded_offset(offset)) + ) + + +def _all_queue_select(*, limit: int, offset: int) -> Select[tuple]: + union_stmt = union_all( + _tracked_queue_select_base(status=None), + _untracked_queue_select_base(), + ).subquery() + return ( + select(union_stmt) + .order_by(union_stmt.c.updated_at.desc()) + .limit(_bounded_limit(limit)) + .offset(_bounded_offset(offset)) + ) + + +def _tracked_queue_count_select(*, status: str | None) -> Select[tuple]: + stmt = select(func.count()).select_from(PublicationPdfJob) + if status: + stmt = stmt.where(PublicationPdfJob.status == status) + return stmt + + +def _untracked_queue_count_select() -> Select[tuple]: + return ( + select(func.count()) + .select_from(Publication) + .outerjoin(PublicationPdfJob, PublicationPdfJob.publication_id == Publication.id) + .where(Publication.pdf_url.is_(None)) + .where(PublicationPdfJob.publication_id.is_(None)) + ) + + +# --------------------------------------------------------------------------- +# Row hydration +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class PdfQueueListItem: + publication_id: int + title: str + pdf_url: str | None + status: str + attempt_count: int + last_failure_reason: str | None + last_failure_detail: str | None + last_source: str | None + requested_by_user_id: int | None + requested_by_email: str | None + queued_at: datetime | None + last_attempt_at: datetime | None + resolved_at: datetime | None + updated_at: datetime + display_identifier: DisplayIdentifier | None = None + + +def _queue_item_from_row(row: tuple) -> PdfQueueListItem: + return PdfQueueListItem( + publication_id=int(row[0]), + title=str(row[1] or ""), + pdf_url=row[2], + status=str(row[3] or PDF_STATUS_UNTRACKED), + attempt_count=int(row[4] or 0), + last_failure_reason=row[5], + last_failure_detail=row[6], + last_source=row[7], + requested_by_user_id=int(row[8]) if row[8] is not None else None, + requested_by_email=row[9], + queued_at=row[10], + last_attempt_at=row[11], + resolved_at=row[12], + updated_at=row[13], + ) + + +async def _hydrated_queue_items( + db_session: AsyncSession, + *, + rows: list[tuple], +) -> list[PdfQueueListItem]: + items = [_queue_item_from_row(row) for row in rows] + return await identifier_service.overlay_pdf_queue_items_with_display_identifiers( + db_session, + items=items, + ) + + +# --------------------------------------------------------------------------- +# Public listing / counting +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class PdfQueuePage: + items: list[PdfQueueListItem] + total_count: int + limit: int + offset: int + + +async def list_pdf_queue_items( + db_session: AsyncSession, + *, + limit: int = 100, + offset: int = 0, + status: str | None = None, +) -> list[PdfQueueListItem]: + bounded_limit = _bounded_limit(limit) + bounded_offset = _bounded_offset(offset) + normalized_status = (status or "").strip().lower() or None + if normalized_status == PDF_STATUS_UNTRACKED: + result = await db_session.execute( + _untracked_queue_select( + limit=bounded_limit, + offset=bounded_offset, + ) + ) + return await _hydrated_queue_items(db_session, rows=list(result.all())) + if normalized_status is None: + result = await db_session.execute( + _all_queue_select( + limit=bounded_limit, + offset=bounded_offset, + ) + ) + return await _hydrated_queue_items(db_session, rows=list(result.all())) + result = await db_session.execute( + _tracked_queue_select( + limit=bounded_limit, + offset=bounded_offset, + status=normalized_status, + ) + ) + return await _hydrated_queue_items(db_session, rows=list(result.all())) + + +async def count_pdf_queue_items( + db_session: AsyncSession, + *, + status: str | None = None, +) -> int: + normalized_status = (status or "").strip().lower() or None + if normalized_status == PDF_STATUS_UNTRACKED: + result = await db_session.execute(_untracked_queue_count_select()) + return int(result.scalar_one() or 0) + tracked_result = await db_session.execute(_tracked_queue_count_select(status=normalized_status)) + tracked_count = int(tracked_result.scalar_one() or 0) + if normalized_status is not None: + return tracked_count + untracked_result = await db_session.execute(_untracked_queue_count_select()) + untracked_count = int(untracked_result.scalar_one() or 0) + return tracked_count + untracked_count + + +async def list_pdf_queue_page( + db_session: AsyncSession, + *, + limit: int = 100, + offset: int = 0, + status: str | None = None, +) -> PdfQueuePage: + bounded_limit = _bounded_limit(limit) + bounded_offset = _bounded_offset(offset) + items = await list_pdf_queue_items( + db_session, + limit=bounded_limit, + offset=bounded_offset, + status=status, + ) + total_count = await count_pdf_queue_items( + db_session, + status=status, + ) + return PdfQueuePage( + items=items, + total_count=total_count, + limit=bounded_limit, + offset=bounded_offset, + ) diff --git a/app/services/publications/pdf_queue_resolution.py b/app/services/publications/pdf_queue_resolution.py new file mode 100644 index 0000000..d2d0fea --- /dev/null +++ b/app/services/publications/pdf_queue_resolution.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +import asyncio +import logging + +from app.db.models import Publication, PublicationPdfJob, PublicationPdfJobEvent +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_resolution_pipeline import ( + resolve_publication_pdf_outcome_for_row, +) +from app.services.publications.types import PublicationListItem +from app.services.unpaywall.application import ( + FAILURE_RESOLUTION_EXCEPTION, + OaResolutionOutcome, +) +from app.settings import settings + +PDF_STATUS_QUEUED = "queued" +PDF_STATUS_RUNNING = "running" +PDF_STATUS_RESOLVED = "resolved" +PDF_STATUS_FAILED = "failed" + +PDF_EVENT_ATTEMPT_STARTED = "attempt_started" +PDF_EVENT_RESOLVED = "resolved" +PDF_EVENT_FAILED = "failed" + +logger = logging.getLogger(__name__) +_scheduled_tasks: set[asyncio.Task[None]] = set() + + +def _utcnow(): + from datetime import UTC, datetime + + return datetime.now(UTC) + + +def _event_row( + *, + publication_id: int, + user_id: int | None, + event_type: str, + status: str | None, + source: str | None = None, + failure_reason: str | None = None, + message: str | None = None, +) -> PublicationPdfJobEvent: + return PublicationPdfJobEvent( + publication_id=publication_id, + user_id=user_id, + event_type=event_type, + status=status, + source=source, + failure_reason=failure_reason, + message=message, + ) + + +def _queued_job( + *, + publication_id: int, + user_id: int, +) -> PublicationPdfJob: + now = _utcnow() + return PublicationPdfJob( + publication_id=publication_id, + status=PDF_STATUS_QUEUED, + queued_at=now, + last_requested_by_user_id=user_id, + ) + + +async def _mark_attempt_started( + *, + publication_id: int, + user_id: int, +) -> None: + session_factory = get_session_factory() + async with session_factory() 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) + db_session.add(job) + job.status = PDF_STATUS_RUNNING + job.last_attempt_at = _utcnow() + job.attempt_count = int(job.attempt_count) + 1 + db_session.add( + _event_row( + publication_id=publication_id, + user_id=user_id, + event_type=PDF_EVENT_ATTEMPT_STARTED, + status=PDF_STATUS_RUNNING, + ) + ) + await db_session.commit() + + +def _failed_outcome( + *, + row: PublicationListItem, +) -> OaResolutionOutcome: + return OaResolutionOutcome( + publication_id=row.publication_id, + doi=None, + pdf_url=None, + failure_reason=FAILURE_RESOLUTION_EXCEPTION, + source=None, + used_crossref=False, + ) + + +async def _fetch_outcome_for_row( + *, + row: PublicationListItem, + request_email: str | None, + openalex_api_key: str | None = None, + allow_arxiv_lookup: bool = True, +) -> tuple[OaResolutionOutcome, bool]: + pipeline_result = await resolve_publication_pdf_outcome_for_row( + row=row, + request_email=request_email, + openalex_api_key=openalex_api_key, + allow_arxiv_lookup=allow_arxiv_lookup, + ) + outcome = pipeline_result.outcome + if outcome is not None: + return outcome, bool(pipeline_result.arxiv_rate_limited) + return _failed_outcome(row=row), bool(pipeline_result.arxiv_rate_limited) + + +def _apply_publication_update( + publication: Publication, + *, + pdf_url: str | None, +) -> None: + if pdf_url and publication.pdf_url != pdf_url: + publication.pdf_url = pdf_url + + +def _apply_job_outcome(job: PublicationPdfJob, *, outcome: OaResolutionOutcome) -> None: + job.last_source = outcome.source + if outcome.pdf_url: + job.status = PDF_STATUS_RESOLVED + job.resolved_at = _utcnow() + job.last_failure_reason = None + job.last_failure_detail = None + return + job.status = PDF_STATUS_FAILED + job.last_failure_reason = outcome.failure_reason + job.last_failure_detail = outcome.failure_reason + + +def _result_event(outcome: OaResolutionOutcome) -> tuple[str, str]: + if outcome.pdf_url: + return PDF_EVENT_RESOLVED, PDF_STATUS_RESOLVED + return PDF_EVENT_FAILED, PDF_STATUS_FAILED + + +async def _persist_outcome( + *, + publication_id: int, + user_id: int, + outcome: OaResolutionOutcome, +) -> None: + session_factory = get_session_factory() + async with session_factory() 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: + return + _apply_publication_update(publication, pdf_url=outcome.pdf_url) + await identifier_service.sync_identifiers_for_publication_resolution( + db_session, + publication=publication, + source=outcome.source, + ) + _apply_job_outcome(job, outcome=outcome) + event_type, status = _result_event(outcome) + db_session.add( + _event_row( + publication_id=publication_id, + user_id=user_id, + event_type=event_type, + status=status, + source=outcome.source, + failure_reason=outcome.failure_reason, + message=outcome.failure_reason, + ) + ) + await db_session.commit() + + +async def _resolve_publication_row( + *, + user_id: int, + request_email: str | None, + row: PublicationListItem, + openalex_api_key: str | None = None, + allow_arxiv_lookup: bool = True, +) -> bool: + from app.services.openalex.client import OpenAlexBudgetExhaustedError + + await _mark_attempt_started(publication_id=row.publication_id, user_id=user_id) + try: + outcome, arxiv_rate_limited = await _fetch_outcome_for_row( + row=row, + request_email=request_email, + openalex_api_key=openalex_api_key, + allow_arxiv_lookup=allow_arxiv_lookup, + ) + except OpenAlexBudgetExhaustedError: + await _persist_outcome( + publication_id=row.publication_id, + user_id=user_id, + outcome=_failed_outcome(row=row), + ) + raise + except Exception as exc: # pragma: no cover - defensive network boundary + structured_log( + logger, + "warning", + "publications.pdf_queue.resolve_failed", + publication_id=row.publication_id, + error=str(exc), + ) + outcome = _failed_outcome(row=row) + arxiv_rate_limited = False + await _persist_outcome( + publication_id=row.publication_id, + user_id=user_id, + outcome=outcome, + ) + return bool(arxiv_rate_limited) + + +async def _run_resolution_task( + *, + user_id: int, + request_email: str | None, + rows: list[PublicationListItem], +) -> None: + from app.services.openalex.client import OpenAlexBudgetExhaustedError + from app.services.settings import application as user_settings_service + + openalex_api_key: str | None = None + try: + session_factory = get_session_factory() + async with session_factory() 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: + openalex_api_key = settings.openalex_api_key + + arxiv_lookup_allowed = True + for row in rows: + try: + arxiv_rate_limited = await _resolve_publication_row( + user_id=user_id, + request_email=request_email, + row=row, + openalex_api_key=openalex_api_key, + allow_arxiv_lookup=arxiv_lookup_allowed, + ) + if arxiv_rate_limited and arxiv_lookup_allowed: + arxiv_lookup_allowed = False + structured_log( + logger, + "warning", + "pdf_queue.arxiv_batch_disabled", + detail="arXiv temporarily disabled for remaining batch after rate limit", + ) + except OpenAlexBudgetExhaustedError: + structured_log( + logger, + "warning", + "pdf_queue.budget_exhausted", + detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted", + ) + break + + +def _register_task(task: asyncio.Task[None]) -> None: + _scheduled_tasks.add(task) + + +def _drop_finished_task(task: asyncio.Task[None]) -> None: + _scheduled_tasks.discard(task) + try: + task.result() + except Exception: + logger.exception("publications.pdf_queue.task_failed") + + +def schedule_rows( + *, + user_id: int, + request_email: str | None, + rows: list[PublicationListItem], +) -> None: + if not rows: + return + task = asyncio.create_task( + _run_resolution_task( + user_id=user_id, + request_email=request_email, + rows=rows, + ) + ) + _register_task(task) + task.add_done_callback(_drop_finished_task) diff --git a/app/services/scholars/application.py b/app/services/scholars/application.py index 49f362a..894c371 100644 --- a/app/services/scholars/application.py +++ b/app/services/scholars/application.py @@ -1,52 +1,17 @@ from __future__ import annotations -import asyncio -import logging import os -import random -from dataclasses import replace -from datetime import UTC, datetime, timedelta -from typing import Any from uuid import uuid4 -from sqlalchemy import delete, func, select, text from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import AuthorSearchCacheEntry, AuthorSearchRuntimeState, ScholarProfile -from app.logging_utils import structured_log -from app.services.scholar.parser import ( - ParsedAuthorSearchPage, - ParseState, - ScholarParserError, - parse_author_search_page, - parse_profile_page, -) +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.constants import ( - ALLOWED_IMAGE_UPLOAD_CONTENT_TYPES, - AUTHOR_SEARCH_LOCK_KEY, - AUTHOR_SEARCH_LOCK_NAMESPACE, - AUTHOR_SEARCH_RUNTIME_STATE_KEY, - DEFAULT_AUTHOR_SEARCH_BLOCKED_CACHE_TTL_SECONDS, - DEFAULT_AUTHOR_SEARCH_CACHE_MAX_ENTRIES, - DEFAULT_AUTHOR_SEARCH_COOLDOWN_BLOCK_THRESHOLD, - DEFAULT_AUTHOR_SEARCH_COOLDOWN_REJECTION_ALERT_THRESHOLD, - DEFAULT_AUTHOR_SEARCH_COOLDOWN_SECONDS, - DEFAULT_AUTHOR_SEARCH_INTERVAL_JITTER_SECONDS, - DEFAULT_AUTHOR_SEARCH_MIN_INTERVAL_SECONDS, - DEFAULT_AUTHOR_SEARCH_RETRY_ALERT_THRESHOLD, - MAX_AUTHOR_SEARCH_LIMIT, - SEARCH_CACHED_BLOCK_REASON, - SEARCH_COOLDOWN_REASON, - SEARCH_DISABLED_REASON, -) +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.exceptions import ScholarServiceError -from app.services.scholars.search_hints import ( - _merge_warnings, - _policy_blocked_author_search_result, - _trim_author_search_result, -) from app.services.scholars.uploads import ( _ensure_upload_root, _resolve_upload_path, @@ -58,280 +23,22 @@ from app.services.scholars.validators import ( validate_scholar_id, ) -logger = logging.getLogger(__name__) - - -async def _acquire_author_search_lock(db_session: AsyncSession) -> None: - await db_session.execute( - text("SELECT pg_advisory_xact_lock(:namespace, :lock_key)"), - { - "namespace": AUTHOR_SEARCH_LOCK_NAMESPACE, - "lock_key": AUTHOR_SEARCH_LOCK_KEY, - }, - ) - - -async def _load_runtime_state_for_update( - db_session: AsyncSession, -) -> AuthorSearchRuntimeState: - result = await db_session.execute( - select(AuthorSearchRuntimeState) - .where(AuthorSearchRuntimeState.state_key == AUTHOR_SEARCH_RUNTIME_STATE_KEY) - .with_for_update() - ) - state = result.scalar_one_or_none() - if state is not None: - return state - - state = AuthorSearchRuntimeState(state_key=AUTHOR_SEARCH_RUNTIME_STATE_KEY) - db_session.add(state) - await db_session.flush() - return state - - -def _serialize_parsed_author_search_page(parsed: ParsedAuthorSearchPage) -> dict: - return { - "state": parsed.state.value, - "state_reason": parsed.state_reason, - "marker_counts": {str(key): int(value) for key, value in parsed.marker_counts.items()}, - "warnings": [str(value) for value in parsed.warnings], - "candidates": [ - { - "scholar_id": candidate.scholar_id, - "display_name": candidate.display_name, - "affiliation": candidate.affiliation, - "email_domain": candidate.email_domain, - "cited_by_count": candidate.cited_by_count, - "interests": [str(interest) for interest in candidate.interests], - "profile_url": candidate.profile_url, - "profile_image_url": candidate.profile_image_url, - } - for candidate in parsed.candidates - ], - } - - -def _payload_state(payload: dict[str, object]) -> ParseState | None: - state_raw = str(payload.get("state", "")).strip() - try: - return ParseState(state_raw) - except ValueError: - return None - - -def _payload_marker_counts(payload: dict[str, object]) -> dict[str, int]: - marker_counts_payload = payload.get("marker_counts") - if not isinstance(marker_counts_payload, dict): - return {} - parsed: dict[str, int] = {} - for key, value in marker_counts_payload.items(): - try: - parsed[str(key)] = int(value) - except (TypeError, ValueError): - continue - return parsed - - -def _payload_warnings(payload: dict[str, object]) -> list[str]: - warnings_payload = payload.get("warnings") - if not isinstance(warnings_payload, list): - return [] - return [str(value) for value in warnings_payload if isinstance(value, str)] - - -def _parse_optional_string(value: object) -> str | None: - if value is None: - return None - normalized = str(value).strip() - return normalized or None - - -def _parse_optional_int(value: object) -> int | None: - if isinstance(value, int): - return value - if isinstance(value, str) and value.strip(): - try: - return int(value) - except ValueError: - return None - return None - - -def _normalize_interests(value: object) -> list[str]: - if not isinstance(value, list): - return [] - return [str(item) for item in value if isinstance(item, str)] - - -def _deserialize_candidate_payload(value: object) -> dict[str, object] | None: - if not isinstance(value, dict): - return None - scholar_id = str(value.get("scholar_id", "")).strip() - display_name = str(value.get("display_name", "")).strip() - profile_url = str(value.get("profile_url", "")).strip() - if not scholar_id or not display_name or not profile_url: - return None - return { - "scholar_id": scholar_id, - "display_name": display_name, - "affiliation": _parse_optional_string(value.get("affiliation")), - "email_domain": _parse_optional_string(value.get("email_domain")), - "cited_by_count": _parse_optional_int(value.get("cited_by_count")), - "interests": _normalize_interests(value.get("interests")), - "profile_url": profile_url, - "profile_image_url": _parse_optional_string(value.get("profile_image_url")), - } - - -def _deserialize_candidates(payload: dict[str, object]) -> list[dict[str, Any]]: - candidates_payload = payload.get("candidates") - if not isinstance(candidates_payload, list): - return [] - normalized: list[dict[str, Any]] = [] - for value in candidates_payload: - candidate = _deserialize_candidate_payload(value) - if candidate is not None: - normalized.append(candidate) - return normalized - - -def _deserialize_parsed_author_search_page(payload: object) -> ParsedAuthorSearchPage | None: - if not isinstance(payload, dict): - return None - - state = _payload_state(payload) - if state is None: - return None - - marker_counts = _payload_marker_counts(payload) - warnings = _payload_warnings(payload) - from app.services.scholar.parser import ScholarSearchCandidate - - normalized_candidates = _deserialize_candidates(payload) - - return ParsedAuthorSearchPage( - state=state, - state_reason=str(payload.get("state_reason", "")).strip() or "unknown", - candidates=[ - ScholarSearchCandidate( - scholar_id=item["scholar_id"], - display_name=item["display_name"], - affiliation=item["affiliation"], - email_domain=item["email_domain"], - cited_by_count=item["cited_by_count"], - interests=item["interests"], - profile_url=item["profile_url"], - profile_image_url=item["profile_image_url"], - ) - for item in normalized_candidates - ], - marker_counts=marker_counts, - warnings=warnings, - ) - - -async def _cache_get_author_search_result( - db_session: AsyncSession, - *, - query_key: str, - now_utc: datetime, -) -> ParsedAuthorSearchPage | None: - result = await db_session.execute( - select(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key == query_key) - ) - entry = result.scalar_one_or_none() - if entry is None: - return None - expires_at = entry.expires_at - if expires_at.tzinfo is None: - expires_at = expires_at.replace(tzinfo=UTC) - if expires_at <= now_utc: - await db_session.delete(entry) - return None - parsed = _deserialize_parsed_author_search_page(entry.payload) - if parsed is None: - await db_session.delete(entry) - return None - return parsed - - -async def _cache_set_author_search_result( - db_session: AsyncSession, - *, - query_key: str, - parsed: ParsedAuthorSearchPage, - ttl_seconds: float, - max_entries: int, - now_utc: datetime, -) -> None: - ttl = max(float(ttl_seconds), 0.0) - existing_result = await db_session.execute( - select(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key == query_key) - ) - existing = existing_result.scalar_one_or_none() - - if ttl <= 0.0: - if existing is not None: - await db_session.delete(existing) - return - - expires_at = now_utc + timedelta(seconds=ttl) - payload = _serialize_parsed_author_search_page(parsed) - if existing is None: - db_session.add( - AuthorSearchCacheEntry( - query_key=query_key, - payload=payload, - expires_at=expires_at, - cached_at=now_utc, - updated_at=now_utc, - ) - ) - else: - existing.payload = payload - existing.expires_at = expires_at - existing.cached_at = now_utc - existing.updated_at = now_utc - - await _prune_author_search_cache(db_session, now_utc=now_utc, max_entries=max_entries) - - -async def _prune_author_search_cache( - db_session: AsyncSession, - *, - now_utc: datetime, - max_entries: int, -) -> None: - await db_session.execute(delete(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.expires_at <= now_utc)) - bounded_max_entries = max(1, int(max_entries)) - count_result = await db_session.execute(select(func.count()).select_from(AuthorSearchCacheEntry)) - entry_count = int(count_result.scalar_one() or 0) - overflow = max(0, entry_count - bounded_max_entries) - if overflow <= 0: - return - stale_keys_result = await db_session.execute( - select(AuthorSearchCacheEntry.query_key).order_by(AuthorSearchCacheEntry.cached_at.asc()).limit(overflow) - ) - stale_keys = [str(row[0]) for row in stale_keys_result.all()] - if stale_keys: - await db_session.execute(delete(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key.in_(stale_keys))) - - -def _is_author_search_block_state(parsed: ParsedAuthorSearchPage) -> bool: - return parsed.state == ParseState.BLOCKED_OR_CAPTCHA - - -def _author_search_cooldown_remaining_seconds( - runtime_state: AuthorSearchRuntimeState, - now_utc: datetime, -) -> int: - cooldown_until = runtime_state.cooldown_until - if cooldown_until is None: - return 0 - if cooldown_until.tzinfo is None: - cooldown_until = cooldown_until.replace(tzinfo=UTC) - remaining_seconds = int((cooldown_until - now_utc).total_seconds()) - return max(0, remaining_seconds) +__all__ = [ + "ScholarServiceError", + "clear_profile_image_customization", + "create_scholar_for_user", + "delete_scholar", + "get_user_scholar_by_id", + "hydrate_profile_metadata", + "list_scholars_for_user", + "normalize_display_name", + "normalize_profile_image_url", + "search_author_candidates", + "set_profile_image_override_url", + "set_profile_image_upload", + "toggle_scholar_enabled", + "validate_scholar_id", +] async def list_scholars_for_user( @@ -339,6 +46,8 @@ async def list_scholars_for_user( *, user_id: int, ) -> list[ScholarProfile]: + from sqlalchemy import select + result = await db_session.execute( select(ScholarProfile) .where(ScholarProfile.user_id == user_id) @@ -377,6 +86,8 @@ async def get_user_scholar_by_id( user_id: int, scholar_profile_id: int, ) -> ScholarProfile | None: + from sqlalchemy import select + result = await db_session.execute( select(ScholarProfile).where( ScholarProfile.id == scholar_profile_id, @@ -411,492 +122,6 @@ async def delete_scholar( await db_session.commit() -def _normalize_author_search_inputs(query: str, limit: int) -> tuple[str, int, str]: - normalized_query = query.strip() - if len(normalized_query) < 2: - raise ScholarServiceError("Search query must be at least 2 characters.") - bounded_limit = max(1, min(int(limit), MAX_AUTHOR_SEARCH_LIMIT)) - return normalized_query, bounded_limit, normalized_query.casefold() - - -def _disabled_search_result(*, normalized_query: str, bounded_limit: int) -> ParsedAuthorSearchPage: - structured_log( - logger, - "warning", - "scholar_search.disabled_by_configuration", - query=normalized_query, - ) - return _policy_blocked_author_search_result( - reason=SEARCH_DISABLED_REASON, - warning_codes=["author_search_disabled_by_configuration"], - limit=bounded_limit, - ) - - -def _normalize_runtime_cooldown_state( - runtime_state: AuthorSearchRuntimeState, - *, - now_utc: datetime, -) -> bool: - if runtime_state.cooldown_until is None: - return False - cooldown_until = runtime_state.cooldown_until - updated = False - if cooldown_until.tzinfo is None: - cooldown_until = cooldown_until.replace(tzinfo=UTC) - runtime_state.cooldown_until = cooldown_until - updated = True - if now_utc < cooldown_until: - return updated - structured_log( - logger, - "info", - "scholar_search.cooldown_expired", - cooldown_until_utc=cooldown_until.isoformat(), - ) - runtime_state.cooldown_until = None - runtime_state.cooldown_rejection_count = 0 - runtime_state.cooldown_alert_emitted = False - return True - - -def _cooldown_warning_codes( - *, - runtime_state: AuthorSearchRuntimeState, - cooldown_remaining_seconds: int, -) -> list[str]: - warning_codes = [ - "author_search_cooldown_active", - f"author_search_cooldown_remaining_{cooldown_remaining_seconds}s", - ] - if bool(runtime_state.cooldown_alert_emitted): - warning_codes.append("author_search_cooldown_alert_threshold_exceeded") - return warning_codes - - -def _emit_cooldown_threshold_alert( - *, - runtime_state: AuthorSearchRuntimeState, - normalized_query: str, - cooldown_rejection_alert_threshold: int, -) -> bool: - runtime_state.cooldown_rejection_count = int(runtime_state.cooldown_rejection_count) + 1 - threshold = max(1, int(cooldown_rejection_alert_threshold)) - if int(runtime_state.cooldown_rejection_count) < threshold: - return True - if bool(runtime_state.cooldown_alert_emitted): - return True - structured_log( - logger, - "error", - "scholar_search.cooldown_rejection_threshold_exceeded", - query=normalized_query, - cooldown_rejection_count=int(runtime_state.cooldown_rejection_count), - threshold=threshold, - cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, - ) - runtime_state.cooldown_alert_emitted = True - return True - - -def _cooldown_block_result( - *, - runtime_state: AuthorSearchRuntimeState, - normalized_query: str, - bounded_limit: int, - cooldown_rejection_alert_threshold: int, - cooldown_remaining_seconds: int, -) -> ParsedAuthorSearchPage: - _emit_cooldown_threshold_alert( - runtime_state=runtime_state, - normalized_query=normalized_query, - cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold, - ) - structured_log( - logger, - "warning", - "scholar_search.cooldown_active", - query=normalized_query, - cooldown_remaining_seconds=cooldown_remaining_seconds, - cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, - ) - return _policy_blocked_author_search_result( - reason=SEARCH_COOLDOWN_REASON, - warning_codes=_cooldown_warning_codes( - runtime_state=runtime_state, - cooldown_remaining_seconds=cooldown_remaining_seconds, - ), - limit=bounded_limit, - ) - - -async def _cache_hit_result( - db_session: AsyncSession, - *, - query_key: str, - now_utc: datetime, - normalized_query: str, - bounded_limit: int, -) -> ParsedAuthorSearchPage | None: - cached = await _cache_get_author_search_result( - db_session, - query_key=query_key, - now_utc=now_utc, - ) - if cached is None: - return None - structured_log( - logger, - "info", - "scholar_search.cache_hit", - query=normalized_query, - state=cached.state.value, - state_reason=cached.state_reason, - ) - state_reason_override = SEARCH_CACHED_BLOCK_REASON if _is_author_search_block_state(cached) else None - return _trim_author_search_result( - cached, - limit=bounded_limit, - extra_warnings=["author_search_served_from_cache"], - state_reason_override=state_reason_override, - ) - - -def _throttle_sleep_seconds( - *, - runtime_state: AuthorSearchRuntimeState, - now_utc: datetime, - min_interval_seconds: float, - interval_jitter_seconds: float, -) -> tuple[float, bool]: - updated = False - if runtime_state.last_live_request_at is None: - enforced_wait_seconds = 0.0 - else: - last_live_request_at = runtime_state.last_live_request_at - if last_live_request_at.tzinfo is None: - last_live_request_at = last_live_request_at.replace(tzinfo=UTC) - runtime_state.last_live_request_at = last_live_request_at - updated = True - enforced_wait_seconds = ( - last_live_request_at + timedelta(seconds=max(float(min_interval_seconds), 0.0)) - now_utc - ).total_seconds() - jitter_seconds = random.uniform(0.0, max(float(interval_jitter_seconds), 0.0)) - return max(0.0, float(enforced_wait_seconds)) + jitter_seconds, updated - - -async def _wait_for_author_search_throttle( - *, - runtime_state: AuthorSearchRuntimeState, - normalized_query: str, - now_utc: datetime, - min_interval_seconds: float, - interval_jitter_seconds: float, -) -> bool: - sleep_seconds, updated = _throttle_sleep_seconds( - runtime_state=runtime_state, - now_utc=now_utc, - min_interval_seconds=min_interval_seconds, - interval_jitter_seconds=interval_jitter_seconds, - ) - if sleep_seconds <= 0.0: - return updated - structured_log( - logger, - "info", - "scholar_search.throttle_wait", - query=normalized_query, - sleep_seconds=round(sleep_seconds, 3), - ) - await asyncio.sleep(sleep_seconds) - return True - - -async def _fetch_author_search_with_retries( - *, - source: ScholarSource, - normalized_query: str, - network_error_retries: int, - retry_backoff_seconds: float, -) -> tuple[ParsedAuthorSearchPage, int, list[str]]: - max_attempts = max(1, int(network_error_retries) + 1) - parsed: ParsedAuthorSearchPage | None = None - retry_warnings: list[str] = [] - retry_scheduled_count = 0 - for attempt_index in range(max_attempts): - fetch_result = await source.fetch_author_search_html(normalized_query, start=0) - try: - parsed = parse_author_search_page(fetch_result) - except ScholarParserError as exc: - parsed = ParsedAuthorSearchPage( - state=ParseState.LAYOUT_CHANGED, - state_reason=exc.code, - candidates=[], - marker_counts={}, - warnings=[exc.code], - ) - if parsed.state != ParseState.NETWORK_ERROR or attempt_index >= max_attempts - 1: - break - retry_warnings.append("network_retry_scheduled_for_author_search") - retry_scheduled_count += 1 - retry_sleep_seconds = max(float(retry_backoff_seconds), 0.0) * (2**attempt_index) - if retry_sleep_seconds > 0: - await asyncio.sleep(retry_sleep_seconds) - if parsed is None: - raise ScholarServiceError("Unable to complete scholar author search.") - return parsed, retry_scheduled_count, retry_warnings - - -def _with_retry_warnings( - parsed: ParsedAuthorSearchPage, - *, - retry_warnings: list[str], - retry_scheduled_count: int, - retry_alert_threshold: int, - normalized_query: str, -) -> ParsedAuthorSearchPage: - merged = replace(parsed, warnings=_merge_warnings(parsed.warnings, retry_warnings)) - threshold = max(1, int(retry_alert_threshold)) - if retry_scheduled_count < threshold: - return merged - structured_log( - logger, - "warning", - "scholar_search.retry_threshold_exceeded", - query=normalized_query, - retry_scheduled_count=retry_scheduled_count, - threshold=threshold, - final_state=merged.state.value, - final_state_reason=merged.state_reason, - ) - return replace( - merged, - warnings=_merge_warnings( - merged.warnings, - [f"author_search_retry_threshold_exceeded_{retry_scheduled_count}"], - ), - ) - - -def _apply_block_circuit_breaker( - *, - runtime_state: AuthorSearchRuntimeState, - merged_parsed: ParsedAuthorSearchPage, - cooldown_block_threshold: int, - cooldown_seconds: int, - normalized_query: str, -) -> ParsedAuthorSearchPage: - if not _is_author_search_block_state(merged_parsed): - runtime_state.consecutive_blocked_count = 0 - return merged_parsed - runtime_state.consecutive_blocked_count = int(runtime_state.consecutive_blocked_count) + 1 - structured_log( - logger, - "warning", - "scholar_search.block_detected", - query=normalized_query, - state_reason=merged_parsed.state_reason, - consecutive_blocked_count=int(runtime_state.consecutive_blocked_count), - ) - if int(runtime_state.consecutive_blocked_count) < max(1, int(cooldown_block_threshold)): - return merged_parsed - runtime_state.cooldown_until = datetime.now(UTC) + timedelta(seconds=max(60, int(cooldown_seconds))) - runtime_state.consecutive_blocked_count = 0 - runtime_state.cooldown_rejection_count = 0 - runtime_state.cooldown_alert_emitted = False - structured_log( - logger, - "error", - "scholar_search.cooldown_activated", - query=normalized_query, - cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, - ) - return replace( - merged_parsed, - warnings=_merge_warnings(merged_parsed.warnings, ["author_search_circuit_breaker_armed"]), - ) - - -def _resolve_author_search_cache_ttl_seconds( - *, - merged_parsed: ParsedAuthorSearchPage, - blocked_cache_ttl_seconds: int, - cache_ttl_seconds: int, -) -> int: - if _is_author_search_block_state(merged_parsed): - return min(max(1, int(blocked_cache_ttl_seconds)), max(1, int(cache_ttl_seconds))) - return max(1, int(cache_ttl_seconds)) - - -async def _load_locked_runtime_state( - db_session: AsyncSession, -) -> AuthorSearchRuntimeState: - await _acquire_author_search_lock(db_session) - return await _load_runtime_state_for_update(db_session) - - -async def _cooldown_or_cache_result( - db_session: AsyncSession, - *, - runtime_state: AuthorSearchRuntimeState, - query_key: str, - normalized_query: str, - bounded_limit: int, - cooldown_rejection_alert_threshold: int, -) -> tuple[ParsedAuthorSearchPage | None, bool]: - runtime_state_updated = _normalize_runtime_cooldown_state( - runtime_state, - now_utc=datetime.now(UTC), - ) - cooldown_remaining_seconds = _author_search_cooldown_remaining_seconds( - runtime_state, - datetime.now(UTC), - ) - if cooldown_remaining_seconds > 0: - return ( - _cooldown_block_result( - runtime_state=runtime_state, - normalized_query=normalized_query, - bounded_limit=bounded_limit, - cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold, - cooldown_remaining_seconds=cooldown_remaining_seconds, - ), - True, - ) - cached_result = await _cache_hit_result( - db_session, - query_key=query_key, - now_utc=datetime.now(UTC), - normalized_query=normalized_query, - bounded_limit=bounded_limit, - ) - return cached_result, runtime_state_updated - - -async def _perform_live_author_search( - db_session: AsyncSession, - *, - source: ScholarSource, - runtime_state: AuthorSearchRuntimeState, - normalized_query: str, - query_key: str, - network_error_retries: int, - retry_backoff_seconds: float, - min_interval_seconds: float, - interval_jitter_seconds: float, - retry_alert_threshold: int, - cooldown_block_threshold: int, - cooldown_seconds: int, - blocked_cache_ttl_seconds: int, - cache_ttl_seconds: int, - cache_max_entries: int, -) -> tuple[ParsedAuthorSearchPage, bool]: - await _wait_for_author_search_throttle( - runtime_state=runtime_state, - normalized_query=normalized_query, - now_utc=datetime.now(UTC), - min_interval_seconds=min_interval_seconds, - interval_jitter_seconds=interval_jitter_seconds, - ) - parsed, retry_count, retry_warnings = await _fetch_author_search_with_retries( - source=source, - normalized_query=normalized_query, - network_error_retries=network_error_retries, - retry_backoff_seconds=retry_backoff_seconds, - ) - runtime_state.last_live_request_at = datetime.now(UTC) - merged = _with_retry_warnings( - parsed, - retry_warnings=retry_warnings, - retry_scheduled_count=retry_count, - retry_alert_threshold=retry_alert_threshold, - normalized_query=normalized_query, - ) - merged = _apply_block_circuit_breaker( - runtime_state=runtime_state, - merged_parsed=merged, - cooldown_block_threshold=cooldown_block_threshold, - cooldown_seconds=cooldown_seconds, - normalized_query=normalized_query, - ) - ttl_seconds = _resolve_author_search_cache_ttl_seconds( - merged_parsed=merged, - blocked_cache_ttl_seconds=blocked_cache_ttl_seconds, - cache_ttl_seconds=cache_ttl_seconds, - ) - await _cache_set_author_search_result( - db_session, - query_key=query_key, - parsed=merged, - ttl_seconds=float(ttl_seconds), - max_entries=cache_max_entries, - now_utc=datetime.now(UTC), - ) - return merged, True - - -async def search_author_candidates( - *, - source: ScholarSource, - db_session: AsyncSession, - query: str, - limit: int, - network_error_retries: int = 1, - retry_backoff_seconds: float = 1.0, - search_enabled: bool = True, - cache_ttl_seconds: int = 21_600, - blocked_cache_ttl_seconds: int = DEFAULT_AUTHOR_SEARCH_BLOCKED_CACHE_TTL_SECONDS, - cache_max_entries: int = DEFAULT_AUTHOR_SEARCH_CACHE_MAX_ENTRIES, - min_interval_seconds: float = DEFAULT_AUTHOR_SEARCH_MIN_INTERVAL_SECONDS, - interval_jitter_seconds: float = DEFAULT_AUTHOR_SEARCH_INTERVAL_JITTER_SECONDS, - cooldown_block_threshold: int = DEFAULT_AUTHOR_SEARCH_COOLDOWN_BLOCK_THRESHOLD, - cooldown_seconds: int = DEFAULT_AUTHOR_SEARCH_COOLDOWN_SECONDS, - retry_alert_threshold: int = DEFAULT_AUTHOR_SEARCH_RETRY_ALERT_THRESHOLD, - cooldown_rejection_alert_threshold: int = DEFAULT_AUTHOR_SEARCH_COOLDOWN_REJECTION_ALERT_THRESHOLD, -) -> ParsedAuthorSearchPage: - normalized_query, bounded_limit, query_key = _normalize_author_search_inputs(query, limit) - if not search_enabled: - return _disabled_search_result( - normalized_query=normalized_query, - bounded_limit=bounded_limit, - ) - - runtime_state = await _load_locked_runtime_state(db_session) - early_result, runtime_state_updated = await _cooldown_or_cache_result( - db_session, - runtime_state=runtime_state, - query_key=query_key, - normalized_query=normalized_query, - bounded_limit=bounded_limit, - cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold, - ) - if early_result is not None: - await db_session.commit() - return early_result - - merged_parsed, live_runtime_updated = await _perform_live_author_search( - db_session, - source=source, - runtime_state=runtime_state, - normalized_query=normalized_query, - query_key=query_key, - network_error_retries=network_error_retries, - retry_backoff_seconds=retry_backoff_seconds, - min_interval_seconds=min_interval_seconds, - interval_jitter_seconds=interval_jitter_seconds, - retry_alert_threshold=retry_alert_threshold, - cooldown_block_threshold=cooldown_block_threshold, - cooldown_seconds=cooldown_seconds, - blocked_cache_ttl_seconds=blocked_cache_ttl_seconds, - cache_ttl_seconds=cache_ttl_seconds, - cache_max_entries=cache_max_entries, - ) - runtime_state_updated = runtime_state_updated or live_runtime_updated - if runtime_state_updated: - await db_session.commit() - return _trim_author_search_result(merged_parsed, limit=bounded_limit) - - async def hydrate_profile_metadata( db_session: AsyncSession, *, diff --git a/app/services/scholars/author_search.py b/app/services/scholars/author_search.py new file mode 100644 index 0000000..a9a9a02 --- /dev/null +++ b/app/services/scholars/author_search.py @@ -0,0 +1,580 @@ +from __future__ import annotations + +import asyncio +import logging +import random +from dataclasses import replace +from datetime import UTC, datetime, timedelta + +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import AuthorSearchRuntimeState +from app.logging_utils import structured_log +from app.services.scholar.parser import ( + ParsedAuthorSearchPage, + ParseState, + ScholarParserError, + parse_author_search_page, +) +from app.services.scholar.source import ScholarSource +from app.services.scholars.author_search_cache import ( + cache_get_author_search_result, + cache_set_author_search_result, +) +from app.services.scholars.constants import ( + AUTHOR_SEARCH_LOCK_KEY, + AUTHOR_SEARCH_LOCK_NAMESPACE, + AUTHOR_SEARCH_RUNTIME_STATE_KEY, + DEFAULT_AUTHOR_SEARCH_BLOCKED_CACHE_TTL_SECONDS, + DEFAULT_AUTHOR_SEARCH_CACHE_MAX_ENTRIES, + DEFAULT_AUTHOR_SEARCH_COOLDOWN_BLOCK_THRESHOLD, + DEFAULT_AUTHOR_SEARCH_COOLDOWN_REJECTION_ALERT_THRESHOLD, + DEFAULT_AUTHOR_SEARCH_COOLDOWN_SECONDS, + DEFAULT_AUTHOR_SEARCH_INTERVAL_JITTER_SECONDS, + DEFAULT_AUTHOR_SEARCH_MIN_INTERVAL_SECONDS, + DEFAULT_AUTHOR_SEARCH_RETRY_ALERT_THRESHOLD, + MAX_AUTHOR_SEARCH_LIMIT, + SEARCH_CACHED_BLOCK_REASON, + SEARCH_COOLDOWN_REASON, + SEARCH_DISABLED_REASON, +) +from app.services.scholars.exceptions import ScholarServiceError +from app.services.scholars.search_hints import ( + _merge_warnings, + _policy_blocked_author_search_result, + _trim_author_search_result, +) + +logger = logging.getLogger(__name__) + + +async def _acquire_author_search_lock(db_session: AsyncSession) -> None: + await db_session.execute( + text("SELECT pg_advisory_xact_lock(:namespace, :lock_key)"), + { + "namespace": AUTHOR_SEARCH_LOCK_NAMESPACE, + "lock_key": AUTHOR_SEARCH_LOCK_KEY, + }, + ) + + +async def _load_runtime_state_for_update( + db_session: AsyncSession, +) -> AuthorSearchRuntimeState: + result = await db_session.execute( + select(AuthorSearchRuntimeState) + .where(AuthorSearchRuntimeState.state_key == AUTHOR_SEARCH_RUNTIME_STATE_KEY) + .with_for_update() + ) + state = result.scalar_one_or_none() + if state is not None: + return state + + state = AuthorSearchRuntimeState(state_key=AUTHOR_SEARCH_RUNTIME_STATE_KEY) + db_session.add(state) + await db_session.flush() + return state + + +def _is_author_search_block_state(parsed: ParsedAuthorSearchPage) -> bool: + return parsed.state == ParseState.BLOCKED_OR_CAPTCHA + + +def _author_search_cooldown_remaining_seconds( + runtime_state: AuthorSearchRuntimeState, + now_utc: datetime, +) -> int: + cooldown_until = runtime_state.cooldown_until + if cooldown_until is None: + return 0 + if cooldown_until.tzinfo is None: + cooldown_until = cooldown_until.replace(tzinfo=UTC) + remaining_seconds = int((cooldown_until - now_utc).total_seconds()) + return max(0, remaining_seconds) + + +def _normalize_author_search_inputs(query: str, limit: int) -> tuple[str, int, str]: + normalized_query = query.strip() + if len(normalized_query) < 2: + raise ScholarServiceError("Search query must be at least 2 characters.") + bounded_limit = max(1, min(int(limit), MAX_AUTHOR_SEARCH_LIMIT)) + return normalized_query, bounded_limit, normalized_query.casefold() + + +def _disabled_search_result(*, normalized_query: str, bounded_limit: int) -> ParsedAuthorSearchPage: + structured_log( + logger, + "warning", + "scholar_search.disabled_by_configuration", + query=normalized_query, + ) + return _policy_blocked_author_search_result( + reason=SEARCH_DISABLED_REASON, + warning_codes=["author_search_disabled_by_configuration"], + limit=bounded_limit, + ) + + +def _normalize_runtime_cooldown_state( + runtime_state: AuthorSearchRuntimeState, + *, + now_utc: datetime, +) -> bool: + if runtime_state.cooldown_until is None: + return False + cooldown_until = runtime_state.cooldown_until + updated = False + if cooldown_until.tzinfo is None: + cooldown_until = cooldown_until.replace(tzinfo=UTC) + runtime_state.cooldown_until = cooldown_until + updated = True + if now_utc < cooldown_until: + return updated + structured_log( + logger, + "info", + "scholar_search.cooldown_expired", + cooldown_until_utc=cooldown_until.isoformat(), + ) + runtime_state.cooldown_until = None + runtime_state.cooldown_rejection_count = 0 + runtime_state.cooldown_alert_emitted = False + return True + + +def _cooldown_warning_codes( + *, + runtime_state: AuthorSearchRuntimeState, + cooldown_remaining_seconds: int, +) -> list[str]: + warning_codes = [ + "author_search_cooldown_active", + f"author_search_cooldown_remaining_{cooldown_remaining_seconds}s", + ] + if bool(runtime_state.cooldown_alert_emitted): + warning_codes.append("author_search_cooldown_alert_threshold_exceeded") + return warning_codes + + +def _emit_cooldown_threshold_alert( + *, + runtime_state: AuthorSearchRuntimeState, + normalized_query: str, + cooldown_rejection_alert_threshold: int, +) -> bool: + runtime_state.cooldown_rejection_count = int(runtime_state.cooldown_rejection_count) + 1 + threshold = max(1, int(cooldown_rejection_alert_threshold)) + if int(runtime_state.cooldown_rejection_count) < threshold: + return True + if bool(runtime_state.cooldown_alert_emitted): + return True + structured_log( + logger, + "error", + "scholar_search.cooldown_rejection_threshold_exceeded", + query=normalized_query, + cooldown_rejection_count=int(runtime_state.cooldown_rejection_count), + threshold=threshold, + cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, + ) + runtime_state.cooldown_alert_emitted = True + return True + + +def _cooldown_block_result( + *, + runtime_state: AuthorSearchRuntimeState, + normalized_query: str, + bounded_limit: int, + cooldown_rejection_alert_threshold: int, + cooldown_remaining_seconds: int, +) -> ParsedAuthorSearchPage: + _emit_cooldown_threshold_alert( + runtime_state=runtime_state, + normalized_query=normalized_query, + cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold, + ) + structured_log( + logger, + "warning", + "scholar_search.cooldown_active", + query=normalized_query, + cooldown_remaining_seconds=cooldown_remaining_seconds, + cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, + ) + return _policy_blocked_author_search_result( + reason=SEARCH_COOLDOWN_REASON, + warning_codes=_cooldown_warning_codes( + runtime_state=runtime_state, + cooldown_remaining_seconds=cooldown_remaining_seconds, + ), + limit=bounded_limit, + ) + + +async def _cache_hit_result( + db_session: AsyncSession, + *, + query_key: str, + now_utc: datetime, + normalized_query: str, + bounded_limit: int, +) -> ParsedAuthorSearchPage | None: + cached = await cache_get_author_search_result( + db_session, + query_key=query_key, + now_utc=now_utc, + ) + if cached is None: + return None + structured_log( + logger, + "info", + "scholar_search.cache_hit", + query=normalized_query, + state=cached.state.value, + state_reason=cached.state_reason, + ) + state_reason_override = SEARCH_CACHED_BLOCK_REASON if _is_author_search_block_state(cached) else None + return _trim_author_search_result( + cached, + limit=bounded_limit, + extra_warnings=["author_search_served_from_cache"], + state_reason_override=state_reason_override, + ) + + +def _throttle_sleep_seconds( + *, + runtime_state: AuthorSearchRuntimeState, + now_utc: datetime, + min_interval_seconds: float, + interval_jitter_seconds: float, +) -> tuple[float, bool]: + updated = False + if runtime_state.last_live_request_at is None: + enforced_wait_seconds = 0.0 + else: + last_live_request_at = runtime_state.last_live_request_at + if last_live_request_at.tzinfo is None: + last_live_request_at = last_live_request_at.replace(tzinfo=UTC) + runtime_state.last_live_request_at = last_live_request_at + updated = True + enforced_wait_seconds = ( + last_live_request_at + timedelta(seconds=max(float(min_interval_seconds), 0.0)) - now_utc + ).total_seconds() + jitter_seconds = random.uniform(0.0, max(float(interval_jitter_seconds), 0.0)) + return max(0.0, float(enforced_wait_seconds)) + jitter_seconds, updated + + +async def _wait_for_author_search_throttle( + *, + runtime_state: AuthorSearchRuntimeState, + normalized_query: str, + now_utc: datetime, + min_interval_seconds: float, + interval_jitter_seconds: float, +) -> bool: + sleep_seconds, updated = _throttle_sleep_seconds( + runtime_state=runtime_state, + now_utc=now_utc, + min_interval_seconds=min_interval_seconds, + interval_jitter_seconds=interval_jitter_seconds, + ) + if sleep_seconds <= 0.0: + return updated + structured_log( + logger, + "info", + "scholar_search.throttle_wait", + query=normalized_query, + sleep_seconds=round(sleep_seconds, 3), + ) + await asyncio.sleep(sleep_seconds) + return True + + +async def _fetch_author_search_with_retries( + *, + source: ScholarSource, + normalized_query: str, + network_error_retries: int, + retry_backoff_seconds: float, +) -> tuple[ParsedAuthorSearchPage, int, list[str]]: + max_attempts = max(1, int(network_error_retries) + 1) + parsed: ParsedAuthorSearchPage | None = None + retry_warnings: list[str] = [] + retry_scheduled_count = 0 + for attempt_index in range(max_attempts): + fetch_result = await source.fetch_author_search_html(normalized_query, start=0) + try: + parsed = parse_author_search_page(fetch_result) + except ScholarParserError as exc: + parsed = ParsedAuthorSearchPage( + state=ParseState.LAYOUT_CHANGED, + state_reason=exc.code, + candidates=[], + marker_counts={}, + warnings=[exc.code], + ) + if parsed.state != ParseState.NETWORK_ERROR or attempt_index >= max_attempts - 1: + break + retry_warnings.append("network_retry_scheduled_for_author_search") + retry_scheduled_count += 1 + retry_sleep_seconds = max(float(retry_backoff_seconds), 0.0) * (2**attempt_index) + if retry_sleep_seconds > 0: + await asyncio.sleep(retry_sleep_seconds) + if parsed is None: + raise ScholarServiceError("Unable to complete scholar author search.") + return parsed, retry_scheduled_count, retry_warnings + + +def _with_retry_warnings( + parsed: ParsedAuthorSearchPage, + *, + retry_warnings: list[str], + retry_scheduled_count: int, + retry_alert_threshold: int, + normalized_query: str, +) -> ParsedAuthorSearchPage: + merged = replace(parsed, warnings=_merge_warnings(parsed.warnings, retry_warnings)) + threshold = max(1, int(retry_alert_threshold)) + if retry_scheduled_count < threshold: + return merged + structured_log( + logger, + "warning", + "scholar_search.retry_threshold_exceeded", + query=normalized_query, + retry_scheduled_count=retry_scheduled_count, + threshold=threshold, + final_state=merged.state.value, + final_state_reason=merged.state_reason, + ) + return replace( + merged, + warnings=_merge_warnings( + merged.warnings, + [f"author_search_retry_threshold_exceeded_{retry_scheduled_count}"], + ), + ) + + +def _apply_block_circuit_breaker( + *, + runtime_state: AuthorSearchRuntimeState, + merged_parsed: ParsedAuthorSearchPage, + cooldown_block_threshold: int, + cooldown_seconds: int, + normalized_query: str, +) -> ParsedAuthorSearchPage: + if not _is_author_search_block_state(merged_parsed): + runtime_state.consecutive_blocked_count = 0 + return merged_parsed + runtime_state.consecutive_blocked_count = int(runtime_state.consecutive_blocked_count) + 1 + structured_log( + logger, + "warning", + "scholar_search.block_detected", + query=normalized_query, + state_reason=merged_parsed.state_reason, + consecutive_blocked_count=int(runtime_state.consecutive_blocked_count), + ) + if int(runtime_state.consecutive_blocked_count) < max(1, int(cooldown_block_threshold)): + return merged_parsed + runtime_state.cooldown_until = datetime.now(UTC) + timedelta(seconds=max(60, int(cooldown_seconds))) + runtime_state.consecutive_blocked_count = 0 + runtime_state.cooldown_rejection_count = 0 + runtime_state.cooldown_alert_emitted = False + structured_log( + logger, + "error", + "scholar_search.cooldown_activated", + query=normalized_query, + cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None, + ) + return replace( + merged_parsed, + warnings=_merge_warnings(merged_parsed.warnings, ["author_search_circuit_breaker_armed"]), + ) + + +def _resolve_author_search_cache_ttl_seconds( + *, + merged_parsed: ParsedAuthorSearchPage, + blocked_cache_ttl_seconds: int, + cache_ttl_seconds: int, +) -> int: + if _is_author_search_block_state(merged_parsed): + return min(max(1, int(blocked_cache_ttl_seconds)), max(1, int(cache_ttl_seconds))) + return max(1, int(cache_ttl_seconds)) + + +async def _load_locked_runtime_state( + db_session: AsyncSession, +) -> AuthorSearchRuntimeState: + await _acquire_author_search_lock(db_session) + return await _load_runtime_state_for_update(db_session) + + +async def _cooldown_or_cache_result( + db_session: AsyncSession, + *, + runtime_state: AuthorSearchRuntimeState, + query_key: str, + normalized_query: str, + bounded_limit: int, + cooldown_rejection_alert_threshold: int, +) -> tuple[ParsedAuthorSearchPage | None, bool]: + runtime_state_updated = _normalize_runtime_cooldown_state( + runtime_state, + now_utc=datetime.now(UTC), + ) + cooldown_remaining_seconds = _author_search_cooldown_remaining_seconds( + runtime_state, + datetime.now(UTC), + ) + if cooldown_remaining_seconds > 0: + return ( + _cooldown_block_result( + runtime_state=runtime_state, + normalized_query=normalized_query, + bounded_limit=bounded_limit, + cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold, + cooldown_remaining_seconds=cooldown_remaining_seconds, + ), + True, + ) + cached_result = await _cache_hit_result( + db_session, + query_key=query_key, + now_utc=datetime.now(UTC), + normalized_query=normalized_query, + bounded_limit=bounded_limit, + ) + return cached_result, runtime_state_updated + + +async def _perform_live_author_search( + db_session: AsyncSession, + *, + source: ScholarSource, + runtime_state: AuthorSearchRuntimeState, + normalized_query: str, + query_key: str, + network_error_retries: int, + retry_backoff_seconds: float, + min_interval_seconds: float, + interval_jitter_seconds: float, + retry_alert_threshold: int, + cooldown_block_threshold: int, + cooldown_seconds: int, + blocked_cache_ttl_seconds: int, + cache_ttl_seconds: int, + cache_max_entries: int, +) -> tuple[ParsedAuthorSearchPage, bool]: + await _wait_for_author_search_throttle( + runtime_state=runtime_state, + normalized_query=normalized_query, + now_utc=datetime.now(UTC), + min_interval_seconds=min_interval_seconds, + interval_jitter_seconds=interval_jitter_seconds, + ) + parsed, retry_count, retry_warnings = await _fetch_author_search_with_retries( + source=source, + normalized_query=normalized_query, + network_error_retries=network_error_retries, + retry_backoff_seconds=retry_backoff_seconds, + ) + runtime_state.last_live_request_at = datetime.now(UTC) + merged = _with_retry_warnings( + parsed, + retry_warnings=retry_warnings, + retry_scheduled_count=retry_count, + retry_alert_threshold=retry_alert_threshold, + normalized_query=normalized_query, + ) + merged = _apply_block_circuit_breaker( + runtime_state=runtime_state, + merged_parsed=merged, + cooldown_block_threshold=cooldown_block_threshold, + cooldown_seconds=cooldown_seconds, + normalized_query=normalized_query, + ) + ttl_seconds = _resolve_author_search_cache_ttl_seconds( + merged_parsed=merged, + blocked_cache_ttl_seconds=blocked_cache_ttl_seconds, + cache_ttl_seconds=cache_ttl_seconds, + ) + await cache_set_author_search_result( + db_session, + query_key=query_key, + parsed=merged, + ttl_seconds=float(ttl_seconds), + max_entries=cache_max_entries, + now_utc=datetime.now(UTC), + ) + return merged, True + + +async def search_author_candidates( + *, + source: ScholarSource, + db_session: AsyncSession, + query: str, + limit: int, + network_error_retries: int = 1, + retry_backoff_seconds: float = 1.0, + search_enabled: bool = True, + cache_ttl_seconds: int = 21_600, + blocked_cache_ttl_seconds: int = DEFAULT_AUTHOR_SEARCH_BLOCKED_CACHE_TTL_SECONDS, + cache_max_entries: int = DEFAULT_AUTHOR_SEARCH_CACHE_MAX_ENTRIES, + min_interval_seconds: float = DEFAULT_AUTHOR_SEARCH_MIN_INTERVAL_SECONDS, + interval_jitter_seconds: float = DEFAULT_AUTHOR_SEARCH_INTERVAL_JITTER_SECONDS, + cooldown_block_threshold: int = DEFAULT_AUTHOR_SEARCH_COOLDOWN_BLOCK_THRESHOLD, + cooldown_seconds: int = DEFAULT_AUTHOR_SEARCH_COOLDOWN_SECONDS, + retry_alert_threshold: int = DEFAULT_AUTHOR_SEARCH_RETRY_ALERT_THRESHOLD, + cooldown_rejection_alert_threshold: int = DEFAULT_AUTHOR_SEARCH_COOLDOWN_REJECTION_ALERT_THRESHOLD, +) -> ParsedAuthorSearchPage: + normalized_query, bounded_limit, query_key = _normalize_author_search_inputs(query, limit) + if not search_enabled: + return _disabled_search_result( + normalized_query=normalized_query, + bounded_limit=bounded_limit, + ) + + runtime_state = await _load_locked_runtime_state(db_session) + early_result, runtime_state_updated = await _cooldown_or_cache_result( + db_session, + runtime_state=runtime_state, + query_key=query_key, + normalized_query=normalized_query, + bounded_limit=bounded_limit, + cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold, + ) + if early_result is not None: + await db_session.commit() + return early_result + + merged_parsed, live_runtime_updated = await _perform_live_author_search( + db_session, + source=source, + runtime_state=runtime_state, + normalized_query=normalized_query, + query_key=query_key, + network_error_retries=network_error_retries, + retry_backoff_seconds=retry_backoff_seconds, + min_interval_seconds=min_interval_seconds, + interval_jitter_seconds=interval_jitter_seconds, + retry_alert_threshold=retry_alert_threshold, + cooldown_block_threshold=cooldown_block_threshold, + cooldown_seconds=cooldown_seconds, + blocked_cache_ttl_seconds=blocked_cache_ttl_seconds, + cache_ttl_seconds=cache_ttl_seconds, + cache_max_entries=cache_max_entries, + ) + runtime_state_updated = runtime_state_updated or live_runtime_updated + if runtime_state_updated: + await db_session.commit() + return _trim_author_search_result(merged_parsed, limit=bounded_limit) diff --git a/app/services/scholars/author_search_cache.py b/app/services/scholars/author_search_cache.py new file mode 100644 index 0000000..01be986 --- /dev/null +++ b/app/services/scholars/author_search_cache.py @@ -0,0 +1,239 @@ +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +from sqlalchemy import delete, func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import AuthorSearchCacheEntry +from app.services.scholar.parser import ( + ParsedAuthorSearchPage, + ParseState, + ScholarSearchCandidate, +) + + +def _serialize_parsed_author_search_page(parsed: ParsedAuthorSearchPage) -> dict: + return { + "state": parsed.state.value, + "state_reason": parsed.state_reason, + "marker_counts": {str(key): int(value) for key, value in parsed.marker_counts.items()}, + "warnings": [str(value) for value in parsed.warnings], + "candidates": [ + { + "scholar_id": candidate.scholar_id, + "display_name": candidate.display_name, + "affiliation": candidate.affiliation, + "email_domain": candidate.email_domain, + "cited_by_count": candidate.cited_by_count, + "interests": [str(interest) for interest in candidate.interests], + "profile_url": candidate.profile_url, + "profile_image_url": candidate.profile_image_url, + } + for candidate in parsed.candidates + ], + } + + +def _payload_state(payload: dict[str, object]) -> ParseState | None: + state_raw = str(payload.get("state", "")).strip() + try: + return ParseState(state_raw) + except ValueError: + return None + + +def _payload_marker_counts(payload: dict[str, object]) -> dict[str, int]: + marker_counts_payload = payload.get("marker_counts") + if not isinstance(marker_counts_payload, dict): + return {} + parsed: dict[str, int] = {} + for key, value in marker_counts_payload.items(): + try: + parsed[str(key)] = int(value) + except (TypeError, ValueError): + continue + return parsed + + +def _payload_warnings(payload: dict[str, object]) -> list[str]: + warnings_payload = payload.get("warnings") + if not isinstance(warnings_payload, list): + return [] + return [str(value) for value in warnings_payload if isinstance(value, str)] + + +def _parse_optional_string(value: object) -> str | None: + if value is None: + return None + normalized = str(value).strip() + return normalized or None + + +def _parse_optional_int(value: object) -> int | None: + if isinstance(value, int): + return value + if isinstance(value, str) and value.strip(): + try: + return int(value) + except ValueError: + return None + return None + + +def _normalize_interests(value: object) -> list[str]: + if not isinstance(value, list): + return [] + return [str(item) for item in value if isinstance(item, str)] + + +def _deserialize_candidate_payload(value: object) -> dict[str, object] | None: + if not isinstance(value, dict): + return None + scholar_id = str(value.get("scholar_id", "")).strip() + display_name = str(value.get("display_name", "")).strip() + profile_url = str(value.get("profile_url", "")).strip() + if not scholar_id or not display_name or not profile_url: + return None + return { + "scholar_id": scholar_id, + "display_name": display_name, + "affiliation": _parse_optional_string(value.get("affiliation")), + "email_domain": _parse_optional_string(value.get("email_domain")), + "cited_by_count": _parse_optional_int(value.get("cited_by_count")), + "interests": _normalize_interests(value.get("interests")), + "profile_url": profile_url, + "profile_image_url": _parse_optional_string(value.get("profile_image_url")), + } + + +def _deserialize_candidates(payload: dict[str, object]) -> list[dict[str, object]]: + candidates_payload = payload.get("candidates") + if not isinstance(candidates_payload, list): + return [] + normalized: list[dict[str, object]] = [] + for value in candidates_payload: + candidate = _deserialize_candidate_payload(value) + if candidate is not None: + normalized.append(candidate) + return normalized + + +def _deserialize_parsed_author_search_page(payload: object) -> ParsedAuthorSearchPage | None: + if not isinstance(payload, dict): + return None + + state = _payload_state(payload) + if state is None: + return None + + marker_counts = _payload_marker_counts(payload) + warnings = _payload_warnings(payload) + normalized_candidates = _deserialize_candidates(payload) + + return ParsedAuthorSearchPage( + state=state, + state_reason=str(payload.get("state_reason", "")).strip() or "unknown", + candidates=[ + ScholarSearchCandidate( + scholar_id=item["scholar_id"], + display_name=item["display_name"], + affiliation=item["affiliation"], + email_domain=item["email_domain"], + cited_by_count=item["cited_by_count"], + interests=item["interests"], + profile_url=item["profile_url"], + profile_image_url=item["profile_image_url"], + ) + for item in normalized_candidates + ], + marker_counts=marker_counts, + warnings=warnings, + ) + + +async def cache_get_author_search_result( + db_session: AsyncSession, + *, + query_key: str, + now_utc: datetime, +) -> ParsedAuthorSearchPage | None: + result = await db_session.execute( + select(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key == query_key) + ) + entry = result.scalar_one_or_none() + if entry is None: + return None + expires_at = entry.expires_at + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=UTC) + if expires_at <= now_utc: + await db_session.delete(entry) + return None + parsed = _deserialize_parsed_author_search_page(entry.payload) + if parsed is None: + await db_session.delete(entry) + return None + return parsed + + +async def cache_set_author_search_result( + db_session: AsyncSession, + *, + query_key: str, + parsed: ParsedAuthorSearchPage, + ttl_seconds: float, + max_entries: int, + now_utc: datetime, +) -> None: + ttl = max(float(ttl_seconds), 0.0) + existing_result = await db_session.execute( + select(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key == query_key) + ) + existing = existing_result.scalar_one_or_none() + + if ttl <= 0.0: + if existing is not None: + await db_session.delete(existing) + return + + expires_at = now_utc + timedelta(seconds=ttl) + payload = _serialize_parsed_author_search_page(parsed) + if existing is None: + db_session.add( + AuthorSearchCacheEntry( + query_key=query_key, + payload=payload, + expires_at=expires_at, + cached_at=now_utc, + updated_at=now_utc, + ) + ) + else: + existing.payload = payload + existing.expires_at = expires_at + existing.cached_at = now_utc + existing.updated_at = now_utc + + await prune_author_search_cache(db_session, now_utc=now_utc, max_entries=max_entries) + + +async def prune_author_search_cache( + db_session: AsyncSession, + *, + now_utc: datetime, + max_entries: int, +) -> None: + await db_session.execute(delete(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.expires_at <= now_utc)) + bounded_max_entries = max(1, int(max_entries)) + count_result = await db_session.execute(select(func.count()).select_from(AuthorSearchCacheEntry)) + entry_count = int(count_result.scalar_one() or 0) + overflow = max(0, entry_count - bounded_max_entries) + if overflow <= 0: + return + stale_keys_result = await db_session.execute( + select(AuthorSearchCacheEntry.query_key).order_by(AuthorSearchCacheEntry.cached_at.asc()).limit(overflow) + ) + stale_keys = [str(row[0]) for row in stale_keys_result.all()] + if stale_keys: + await db_session.execute(delete(AuthorSearchCacheEntry).where(AuthorSearchCacheEntry.query_key.in_(stale_keys))) diff --git a/tests/unit/test_publication_pdf_queue_policy.py b/tests/unit/test_publication_pdf_queue_policy.py index b88cb26..ffc78d7 100644 --- a/tests/unit/test_publication_pdf_queue_policy.py +++ b/tests/unit/test_publication_pdf_queue_policy.py @@ -6,7 +6,7 @@ from types import SimpleNamespace import pytest from app.db.models import PublicationPdfJob -from app.services.publications import pdf_queue +from app.services.publications import pdf_queue, pdf_queue_resolution from app.services.publications.pdf_resolution_pipeline import PipelineOutcome from app.services.unpaywall.application import OaResolutionOutcome @@ -150,9 +150,9 @@ async def test_fetch_outcome_for_row_uses_pipeline_outcome(monkeypatch: pytest.M scholar_candidates=None, ) - monkeypatch.setattr(pdf_queue, "resolve_publication_pdf_outcome_for_row", _fake_pipeline) + monkeypatch.setattr(pdf_queue_resolution, "resolve_publication_pdf_outcome_for_row", _fake_pipeline) - outcome, arxiv_rate_limited = await pdf_queue._fetch_outcome_for_row( + outcome, arxiv_rate_limited = await pdf_queue_resolution._fetch_outcome_for_row( row=_row(), request_email="user@example.com", ) @@ -172,15 +172,15 @@ async def test_fetch_outcome_for_row_returns_failed_outcome_when_pipeline_return assert allow_arxiv_lookup is True return PipelineOutcome(outcome=None, scholar_candidates=None, arxiv_rate_limited=True) - monkeypatch.setattr(pdf_queue, "resolve_publication_pdf_outcome_for_row", _fake_pipeline) + monkeypatch.setattr(pdf_queue_resolution, "resolve_publication_pdf_outcome_for_row", _fake_pipeline) - outcome, arxiv_rate_limited = await pdf_queue._fetch_outcome_for_row( + outcome, arxiv_rate_limited = await pdf_queue_resolution._fetch_outcome_for_row( row=_row(), request_email="user@example.com", ) assert outcome.pdf_url is None - assert outcome.failure_reason == pdf_queue.FAILURE_RESOLUTION_EXCEPTION + assert outcome.failure_reason == pdf_queue_resolution.FAILURE_RESOLUTION_EXCEPTION assert arxiv_rate_limited is True @@ -210,11 +210,11 @@ async def test_resolve_publication_row_persists_outcome_and_returns_rate_limit_f async def _capture_persist_outcome(*, publication_id: int, user_id: int, outcome: OaResolutionOutcome) -> None: captured.append((publication_id, user_id, outcome)) - monkeypatch.setattr(pdf_queue, "_mark_attempt_started", _noop_mark_attempt_started) - monkeypatch.setattr(pdf_queue, "_fetch_outcome_for_row", _fake_fetch) - monkeypatch.setattr(pdf_queue, "_persist_outcome", _capture_persist_outcome) + monkeypatch.setattr(pdf_queue_resolution, "_mark_attempt_started", _noop_mark_attempt_started) + monkeypatch.setattr(pdf_queue_resolution, "_fetch_outcome_for_row", _fake_fetch) + monkeypatch.setattr(pdf_queue_resolution, "_persist_outcome", _capture_persist_outcome) - rate_limited = await pdf_queue._resolve_publication_row( + rate_limited = await pdf_queue_resolution._resolve_publication_row( user_id=42, request_email="user@example.com", row=_row(), @@ -252,10 +252,10 @@ async def test_run_resolution_task_disables_arxiv_for_remaining_batch( calls.append((int(row.publication_id), bool(allow_arxiv_lookup))) return row.publication_id == 1 - monkeypatch.setattr(pdf_queue, "get_session_factory", _raise_session_factory_error) - monkeypatch.setattr(pdf_queue, "_resolve_publication_row", _fake_resolve_publication_row) + monkeypatch.setattr(pdf_queue_resolution, "get_session_factory", _raise_session_factory_error) + monkeypatch.setattr(pdf_queue_resolution, "_resolve_publication_row", _fake_resolve_publication_row) - await pdf_queue._run_resolution_task( + await pdf_queue_resolution._run_resolution_task( user_id=42, request_email="user@example.com", rows=[first, second],