From 01454162bb7d78c07a4c774e741f751c6037d3e7 Mon Sep 17 00:00:00 2001 From: Justin Visser Date: Fri, 20 Feb 2026 22:56:52 +0100 Subject: [PATCH] harden domain architecture and add lazy Crossref+Unpaywall PDF enrichment with publications workspace refactor --- .env.example | 10 + README.md | 31 +- agents.md | 10 +- .../20260220_0011_add_publication_doi.py | 50 +++ app/api/routers/admin.py | 2 +- app/api/routers/auth.py | 89 ++-- app/api/routers/publications.py | 213 ++++++--- app/api/routers/runs.py | 415 +++++++++++------- app/api/routers/scholars.py | 186 +++++--- app/api/routers/settings.py | 159 ++++--- app/api/schemas.py | 23 + app/auth/runtime.py | 2 +- app/db/models.py | 1 + app/services/continuation_queue.py | 3 - app/services/domains/crossref/__init__.py | 5 + app/services/domains/crossref/application.py | 259 +++++++++++ app/services/domains/doi/normalize.py | 23 + app/services/domains/ingestion/application.py | 44 +- app/services/domains/portability/exporting.py | 3 + .../domains/portability/publication_import.py | 10 + app/services/domains/portability/types.py | 1 + .../domains/publications/application.py | 8 + .../domains/publications/enrichment.py | 135 ++++++ app/services/domains/publications/listing.py | 146 +++++- app/services/domains/publications/queries.py | 60 +++ app/services/domains/publications/types.py | 2 + app/services/domains/scholar/parser.py | 65 ++- app/services/domains/scholar/parser_types.py | 16 + app/services/domains/scholar/profile_rows.py | 56 +-- app/services/domains/scholars/application.py | 17 +- app/services/domains/unpaywall/__init__.py | 5 + app/services/domains/unpaywall/application.py | 218 +++++++++ app/services/import_export.py | 3 - app/services/ingestion.py | 3 - app/services/publications.py | 3 - app/services/run_safety.py | 3 - app/services/runs.py | 3 - app/services/scheduler.py | 3 - app/services/scholar_parser.py | 3 - app/services/scholar_source.py | 4 - app/services/scholars.py | 3 - app/services/user_settings.py | 3 - app/services/users.py | 3 - app/settings.py | 10 + frontend/src/components/layout/AppNav.vue | 2 +- frontend/src/features/publications/index.ts | 21 + frontend/src/features/scholars/index.ts | 1 + frontend/src/pages/PublicationsPage.vue | 338 ++++++++------ frontend/src/pages/ScholarsPage.vue | 1 + pyproject.toml | 3 +- tests/integration/test_api_v1.py | 180 +++++++- tests/integration/test_fixture_probe_runs.py | 4 +- tests/integration/test_migrations.py | 2 +- tests/unit/test_crossref_lookup.py | 75 ++++ tests/unit/test_doi_normalize.py | 17 + tests/unit/test_run_safety.py | 2 +- tests/unit/test_runs_summary.py | 2 +- tests/unit/test_scholar_parser.py | 26 +- tests/unit/test_scholar_search_safety.py | 6 +- tests/unit/test_scholar_source.py | 2 +- tests/unit/test_user_settings.py | 2 +- uv.lock | 255 ++++++++++- 62 files changed, 2562 insertions(+), 688 deletions(-) create mode 100644 alembic/versions/20260220_0011_add_publication_doi.py delete mode 100644 app/services/continuation_queue.py create mode 100644 app/services/domains/crossref/__init__.py create mode 100644 app/services/domains/crossref/application.py create mode 100644 app/services/domains/doi/normalize.py create mode 100644 app/services/domains/publications/enrichment.py create mode 100644 app/services/domains/unpaywall/__init__.py create mode 100644 app/services/domains/unpaywall/application.py delete mode 100644 app/services/import_export.py delete mode 100644 app/services/ingestion.py delete mode 100644 app/services/publications.py delete mode 100644 app/services/run_safety.py delete mode 100644 app/services/runs.py delete mode 100644 app/services/scheduler.py delete mode 100644 app/services/scholar_parser.py delete mode 100644 app/services/scholar_source.py delete mode 100644 app/services/scholars.py delete mode 100644 app/services/user_settings.py delete mode 100644 app/services/users.py create mode 100644 tests/unit/test_crossref_lookup.py create mode 100644 tests/unit/test_doi_normalize.py diff --git a/.env.example b/.env.example index ec10203..dc38b6b 100644 --- a/.env.example +++ b/.env.example @@ -76,6 +76,16 @@ SCHOLAR_NAME_SEARCH_COOLDOWN_BLOCK_THRESHOLD=1 SCHOLAR_NAME_SEARCH_COOLDOWN_SECONDS=1800 SCHOLAR_NAME_SEARCH_ALERT_RETRY_COUNT_THRESHOLD=2 SCHOLAR_NAME_SEARCH_ALERT_COOLDOWN_REJECTIONS_THRESHOLD=3 +UNPAYWALL_ENABLED=1 +UNPAYWALL_EMAIL= +UNPAYWALL_TIMEOUT_SECONDS=4.0 +UNPAYWALL_MAX_ITEMS_PER_REQUEST=20 +UNPAYWALL_RETRY_COOLDOWN_SECONDS=1800 +CROSSREF_ENABLED=1 +CROSSREF_MAX_ROWS=10 +CROSSREF_TIMEOUT_SECONDS=8.0 +CROSSREF_MIN_INTERVAL_SECONDS=0.6 +CROSSREF_MAX_LOOKUPS_PER_REQUEST=8 BOOTSTRAP_ADMIN_ON_START=0 BOOTSTRAP_ADMIN_EMAIL= diff --git a/README.md b/README.md index f4f3fca..3ec9573 100644 --- a/README.md +++ b/README.md @@ -78,15 +78,18 @@ docker compose -f docker-compose.yml -f docker-compose.dev.yml down - Scholar tracking is user-scoped: each account can track the same Scholar ID independently. - Publications are shared/global records deduplicated by Scholar cluster ID and normalized fingerprint. - Per-account visibility and read state is stored on scholar-publication links, not on the global publication row. -- Publication records include both canonical Scholar detail URLs (`pub_url`) and direct download targets (`pdf_url`) when available. +- Publication records include canonical Scholar detail URLs (`pub_url`), normalized DOI (`doi`), and resolved OA PDF links (`pdf_url`) when available. ## Backend Architecture (Refactored) -- Compatibility facades remain in `app/services/*.py`, but canonical service logic now lives in `app/services/domains/*`. +- Canonical service logic lives in `app/services/domains/*`; root-level `app/services/*.py` business-logic modules are removed. - `app/services/domains/ingestion/*`: run orchestration (`application.py`), continuation queue (`queue.py`), scrape safety cooldown policy (`safety.py`), scheduler/queue draining (`scheduler.py`), plus shared constants/types/fingerprint helpers. - `app/services/domains/scholar/*`: fail-fast Google Scholar parsing and source access. Parser flow is modularized across row extractors, state detection, constants, and parser types. - `app/services/domains/scholars/*`: scholar CRUD, name-search safety/caching, profile-image validation/upload handling, and shared scholar validation helpers. -- `app/services/domains/publications/*`: publication listing/read-state query modules for dashboard/publication UIs (`pub_url` + `pdf_url`). +- `app/services/domains/publications/*`: publication listing/read-state query modules, lazy/background OA enrichment scheduling (non-blocking for list responses), and per-publication PDF retry support. +- `app/services/domains/doi/*`: DOI normalization helpers used by ingestion and enrichment. +- `app/services/domains/crossref/*`: DOI discovery fallback using Crossref metadata queries with bounded/paced request behavior. +- `app/services/domains/unpaywall/*`: OA resolution by DOI (best OA location + PDF URL extraction), using per-user email for API calls. - `app/services/domains/runs/*`: run history summaries plus continuation queue status/retry/drop/clear operations. - `app/services/domains/portability/*`: import/export of tracked scholars and publication-link state while preserving global publication deduplication. - `app/services/domains/settings/*` and `app/services/domains/users/*`: user settings and user management services. @@ -225,6 +228,21 @@ Notes: | `SCHOLAR_NAME_SEARCH_ALERT_RETRY_COUNT_THRESHOLD` | `2` | integer >= 1 | deploy, dev | Emit retry-threshold observability warning when name-search retry count reaches this value. | | `SCHOLAR_NAME_SEARCH_ALERT_COOLDOWN_REJECTIONS_THRESHOLD` | `3` | integer >= 1 | deploy, dev | Emit cooldown-threshold observability alert after this many requests are rejected during active cooldown. | +### Open-Access Enrichment (Crossref + Unpaywall) + +| Variable | Default | Options | Scope | Description | +| --- | --- | --- | --- | --- | +| `UNPAYWALL_ENABLED` | `1` | boolean | deploy, dev | Enable Unpaywall OA resolution. | +| `UNPAYWALL_EMAIL` | empty | valid email | deploy, dev | Fallback email if per-user email is unavailable. | +| `UNPAYWALL_TIMEOUT_SECONDS` | `4.0` | float >= 0.5 | deploy, dev | Timeout for Unpaywall requests. | +| `UNPAYWALL_MAX_ITEMS_PER_REQUEST` | `20` | integer >= 0 | deploy, dev | Max publications queued per lazy enrichment pass. | +| `UNPAYWALL_RETRY_COOLDOWN_SECONDS` | `1800` | integer >= 1 | deploy, dev | Per-user/per-publication cooldown before re-scheduling unresolved OA lookups. | +| `CROSSREF_ENABLED` | `1` | boolean | deploy, dev | Enable Crossref DOI discovery fallback when DOI is missing/insufficient. | +| `CROSSREF_MAX_ROWS` | `10` | integer >= 1 | deploy, dev | Max Crossref rows evaluated per lookup. | +| `CROSSREF_TIMEOUT_SECONDS` | `8.0` | float >= 0.5 | deploy, dev | Timeout budget per Crossref lookup task. | +| `CROSSREF_MIN_INTERVAL_SECONDS` | `0.6` | float >= 0 | deploy, dev | Minimum pacing interval between Crossref requests. | +| `CROSSREF_MAX_LOOKUPS_PER_REQUEST` | `8` | integer >= 0 | deploy, dev | Max Crossref DOI lookups performed during one lazy pass or manual retry. | + ### Scrape Safety Operations - Structured safety events are emitted for: @@ -290,7 +308,12 @@ Scheduled fixture probes run in GitHub Actions via `.github/workflows/scheduled- - `GET /api/v1/publications` supports `mode=all|unread|latest` (plus temporary alias `mode=new`). - `unread` = actionable read-state (`is_read=false`). - `latest` = discovery-state (`first seen in the latest completed check`). -- Publications include `pub_url` and `pdf_url` in API responses. `pdf_url` is a direct-download target extracted from Scholar result-side links when present. +- Publications include `pub_url`, `doi`, and `pdf_url` in API responses. +- OA enrichment flow is DOI-first: + - existing DOI (or explicit DOI in source metadata) -> Unpaywall lookup, + - Crossref DOI discovery fallback (paced + bounded), + - Unpaywall DOI lookup for OA status and `pdf_url`. +- `POST /api/v1/publications/{publication_id}/retry-pdf` retries enrichment for a single publication in current user scope. - Response counters: - `unread_count`: unread publications in current scope. - `latest_count`: publications discovered in latest completed check. diff --git a/agents.md b/agents.md index b7bd7e3..18abc7c 100644 --- a/agents.md +++ b/agents.md @@ -27,10 +27,12 @@ These limits prevent IP bans and are not to be optimized away. * **Frontend:** TypeScript, Vue 3, Vite. * **Infrastructure:** Multi-stage Docker. -## 5. Refactored Service Boundaries (Current) -* **Compatibility rule:** `app/services/*.py` files are import façades only. Keep them thin and route all real logic to domain modules. -* **`app/services/domains/scholar/*`:** Parser contract is fail-fast. Layout drift must emit explicit `layout_*` reasons/warnings, never silent partial success. +## 5. Domain Service Boundaries +* **Strict Modularity:** Flat files in the `app/services/` root are strictly prohibited. All business logic and routing must reside exclusively within `app/services/domains/`. +* **`app/services/domains/scholar/*`:** Parser contract is fail-fast. Layout drift must emit explicit exceptions and `layout_*` reasons/warnings. Never allow silent partial success. * **`app/services/domains/ingestion/application.py`:** Orchestrates ingestion runs; validate parser outputs before persistence; enforce publication candidate constraints before upsert. -* **`app/services/domains/publications/*`:** Publication list/read-state query layer; include both `pub_url` and `pdf_url` for UI consumption. +* **`app/services/domains/publications/*`:** Publication list/read-state query layer. Includes `doi` + `pdf_url` fields for UI consumption, enforces non-blocking lazy OA enrichment scheduling on list reads, and exposes per-publication PDF retry behavior. +* **`app/services/domains/crossref/*`:** DOI discovery fallback module. Must use bounded/paced lookups to avoid burst traffic and 429 responses. +* **`app/services/domains/unpaywall/*`:** OA resolver by DOI only (best OA location + PDF URL extraction). Do not use Google Scholar for PDF resolution to avoid N+1 scrape amplification. * **`app/services/domains/portability/*`:** Handles JSON import/export for user-scoped scholars and scholar-publication link state while preserving global publication dedup rules. * **`app/services/domains/ingestion/scheduler.py`:** Owns automatic runs and continuation queue retries/drops; do not bypass safety gate or cooldown logic. diff --git a/alembic/versions/20260220_0011_add_publication_doi.py b/alembic/versions/20260220_0011_add_publication_doi.py new file mode 100644 index 0000000..bb9a813 --- /dev/null +++ b/alembic/versions/20260220_0011_add_publication_doi.py @@ -0,0 +1,50 @@ +"""Add DOI column to publications. + +Revision ID: 20260220_0011 +Revises: 20260219_0010 +Create Date: 2026-02-20 13:35:00.000000 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = "20260220_0011" +down_revision: str | Sequence[str] | None = "20260219_0010" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def _has_column(inspector: sa.Inspector, table_name: str, column_name: str) -> bool: + for column in inspector.get_columns(table_name): + if str(column.get("name")) == column_name: + return True + return False + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + table_names = set(inspector.get_table_names()) + if "publications" not in table_names: + return + if not _has_column(inspector, "publications", "doi"): + op.add_column("publications", sa.Column("doi", sa.String(length=255), nullable=True)) + index_names = {index["name"] for index in inspector.get_indexes("publications")} + if "ix_publications_doi" not in index_names: + op.create_index("ix_publications_doi", "publications", ["doi"], unique=False) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + table_names = set(inspector.get_table_names()) + if "publications" not in table_names: + return + index_names = {index["name"] for index in inspector.get_indexes("publications")} + if "ix_publications_doi" in index_names: + op.drop_index("ix_publications_doi", table_name="publications") + if _has_column(inspector, "publications", "doi"): + op.drop_column("publications", "doi") diff --git a/app/api/routers/admin.py b/app/api/routers/admin.py index dbf00bf..9844c3e 100644 --- a/app/api/routers/admin.py +++ b/app/api/routers/admin.py @@ -20,7 +20,7 @@ from app.auth.deps import get_auth_service from app.auth.service import AuthService from app.db.models import User from app.db.session import get_db_session -from app.services import users as user_service +from app.services.domains.users import application as user_service logger = logging.getLogger(__name__) diff --git a/app/api/routers/auth.py b/app/api/routers/auth.py index 1a76ced..8f8da17 100644 --- a/app/api/routers/auth.py +++ b/app/api/routers/auth.py @@ -24,13 +24,55 @@ from app.auth.session import set_session_user from app.db.models import User from app.db.session import get_db_session from app.security.csrf import ensure_csrf_token -from app.services import users as user_service +from app.services.domains.users import application as user_service logger = logging.getLogger(__name__) router = APIRouter(prefix="/auth", tags=["api-auth"]) +def _login_limiter_key_and_email(request: Request, payload: LoginRequest) -> tuple[str, str]: + return auth_runtime.login_rate_limit_key(request, payload.email), payload.email.strip().lower() + + +def _raise_rate_limited(normalized_email: str, retry_after_seconds: int) -> None: + logger.warning( + "api.auth.login_rate_limited", + extra={ + "event": "api.auth.login_rate_limited", + "email": normalized_email, + "retry_after_seconds": retry_after_seconds, + }, + ) + raise ApiException( + status_code=429, + code="rate_limited", + message="Too many login attempts. Please try again later.", + details={"retry_after_seconds": retry_after_seconds}, + ) + + +def _serialize_user_payload(user: User) -> dict[str, object]: + return { + "id": int(user.id), + "email": user.email, + "is_admin": bool(user.is_admin), + "is_active": bool(user.is_active), + } + + +def _raise_invalid_credentials(*, normalized_email: str) -> None: + logger.info( + "api.auth.login_failed", + extra={"event": "api.auth.login_failed", "email": normalized_email}, + ) + raise ApiException( + status_code=401, + code="invalid_credentials", + message="Invalid email or password.", + ) + + @router.post( "/login", response_model=LoginEnvelope, @@ -42,24 +84,10 @@ async def login( auth_service: AuthService = Depends(get_auth_service), rate_limiter: SlidingWindowRateLimiter = Depends(get_login_rate_limiter), ): - limiter_key = auth_runtime.login_rate_limit_key(request, payload.email) + limiter_key, normalized_email = _login_limiter_key_and_email(request, payload) decision = rate_limiter.check(limiter_key) - normalized_email = payload.email.strip().lower() if not decision.allowed: - logger.warning( - "api.auth.login_rate_limited", - extra={ - "event": "api.auth.login_rate_limited", - "email": normalized_email, - "retry_after_seconds": decision.retry_after_seconds, - }, - ) - raise ApiException( - status_code=429, - code="rate_limited", - message="Too many login attempts. Please try again later.", - details={"retry_after_seconds": decision.retry_after_seconds}, - ) + _raise_rate_limited(normalized_email, int(decision.retry_after_seconds)) user = await auth_service.authenticate_user( db_session, @@ -68,18 +96,7 @@ async def login( ) if user is None: rate_limiter.record_failure(limiter_key) - logger.info( - "api.auth.login_failed", - extra={ - "event": "api.auth.login_failed", - "email": normalized_email, - }, - ) - raise ApiException( - status_code=401, - code="invalid_credentials", - message="Invalid email or password.", - ) + _raise_invalid_credentials(normalized_email=normalized_email) rate_limiter.reset(limiter_key) set_session_user( @@ -101,12 +118,7 @@ async def login( data={ "authenticated": True, "csrf_token": ensure_csrf_token(request), - "user": { - "id": int(user.id), - "email": user.email, - "is_admin": bool(user.is_admin), - "is_active": bool(user.is_active), - }, + "user": _serialize_user_payload(user), }, ) @@ -124,12 +136,7 @@ async def get_current_session( data={ "authenticated": True, "csrf_token": ensure_csrf_token(request), - "user": { - "id": int(current_user.id), - "email": current_user.email, - "is_admin": bool(current_user.is_admin), - "is_active": bool(current_user.is_active), - }, + "user": _serialize_user_payload(current_user), }, ) diff --git a/app/api/routers/publications.py b/app/api/routers/publications.py index 53a8109..998cdea 100644 --- a/app/api/routers/publications.py +++ b/app/api/routers/publications.py @@ -3,7 +3,7 @@ from __future__ import annotations import logging from typing import Literal -from fastapi import APIRouter, Depends, Query, Request +from fastapi import APIRouter, Depends, Path, Query, Request from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_api_current_user @@ -14,17 +14,104 @@ from app.api.schemas import ( MarkSelectedReadEnvelope, MarkSelectedReadRequest, PublicationsListEnvelope, + RetryPublicationPdfEnvelope, + RetryPublicationPdfRequest, ) from app.db.models import User from app.db.session import get_db_session -from app.services import publications as publication_service -from app.services import scholars as scholar_service +from app.services.domains.publications import application as publication_service +from app.services.domains.scholars import application as scholar_service +from app.settings import settings logger = logging.getLogger(__name__) router = APIRouter(prefix="/publications", tags=["api-publications"]) +async def _require_selected_profile( + db_session: AsyncSession, + *, + user_id: int, + selected_scholar_id: int | None, +) -> None: + if selected_scholar_id is None: + return + selected_profile = await scholar_service.get_user_scholar_by_id( + db_session, + user_id=user_id, + scholar_profile_id=selected_scholar_id, + ) + if selected_profile is None: + raise ApiException( + status_code=404, + code="scholar_not_found", + message="Scholar filter not found.", + ) + + +def _serialize_publication_item(item) -> dict[str, object]: + return { + "publication_id": item.publication_id, + "scholar_profile_id": item.scholar_profile_id, + "scholar_label": item.scholar_label, + "title": item.title, + "year": item.year, + "citation_count": item.citation_count, + "venue_text": item.venue_text, + "pub_url": item.pub_url, + "doi": item.doi, + "pdf_url": item.pdf_url, + "is_read": item.is_read, + "first_seen_at": item.first_seen_at, + "is_new_in_latest_run": item.is_new_in_latest_run, + } + + +async def _publication_counts( + db_session: AsyncSession, + *, + user_id: int, + selected_scholar_id: int | None, +) -> tuple[int, int, int]: + unread_count = await publication_service.count_unread_for_user( + db_session, + user_id=user_id, + scholar_profile_id=selected_scholar_id, + ) + latest_count = await publication_service.count_latest_for_user( + db_session, + user_id=user_id, + scholar_profile_id=selected_scholar_id, + ) + total_count = await publication_service.count_for_user( + db_session, + user_id=user_id, + mode=publication_service.MODE_ALL, + scholar_profile_id=selected_scholar_id, + ) + return unread_count, latest_count, total_count + + +def _publications_list_data( + *, + mode: str, + selected_scholar_id: int | None, + unread_count: int, + latest_count: int, + total_count: int, + publications: list, +) -> dict[str, object]: + return { + "mode": mode, + "selected_scholar_profile_id": selected_scholar_id, + "unread_count": unread_count, + "latest_count": latest_count, + "new_count": latest_count, + "total_count": total_count, + "publications": [_serialize_publication_item(item) for item in publications], + } + + @router.get( "", response_model=PublicationsListEnvelope, @@ -39,18 +126,11 @@ async def list_publications( ): resolved_mode = publication_service.resolve_publication_view_mode(mode) selected_scholar_id = scholar_profile_id - if selected_scholar_id is not None: - selected_profile = await scholar_service.get_user_scholar_by_id( - db_session, - user_id=current_user.id, - scholar_profile_id=selected_scholar_id, - ) - if selected_profile is None: - raise ApiException( - status_code=404, - code="scholar_not_found", - message="Scholar filter not found.", - ) + await _require_selected_profile( + db_session, + user_id=current_user.id, + selected_scholar_id=selected_scholar_id, + ) publications = await publication_service.list_for_user( db_session, @@ -59,50 +139,26 @@ async def list_publications( scholar_profile_id=selected_scholar_id, limit=limit, ) - unread_count = await publication_service.count_unread_for_user( + await publication_service.schedule_missing_pdf_enrichment_for_user( + user_id=current_user.id, + request_email=current_user.email, + items=publications, + max_items=settings.unpaywall_max_items_per_request, + ) + unread_count, latest_count, total_count = await _publication_counts( db_session, user_id=current_user.id, - scholar_profile_id=selected_scholar_id, + selected_scholar_id=selected_scholar_id, ) - latest_count = await publication_service.count_latest_for_user( - db_session, - user_id=current_user.id, - scholar_profile_id=selected_scholar_id, - ) - total_count = await publication_service.count_for_user( - db_session, - user_id=current_user.id, - mode=publication_service.MODE_ALL, - scholar_profile_id=selected_scholar_id, - ) - return success_payload( - request, - data={ - "mode": resolved_mode, - "selected_scholar_profile_id": selected_scholar_id, - "unread_count": unread_count, - "latest_count": latest_count, - "new_count": latest_count, - "total_count": total_count, - "publications": [ - { - "publication_id": item.publication_id, - "scholar_profile_id": item.scholar_profile_id, - "scholar_label": item.scholar_label, - "title": item.title, - "year": item.year, - "citation_count": item.citation_count, - "venue_text": item.venue_text, - "pub_url": item.pub_url, - "pdf_url": item.pdf_url, - "is_read": item.is_read, - "first_seen_at": item.first_seen_at, - "is_new_in_latest_run": item.is_new_in_latest_run, - } - for item in publications - ], - }, + data = _publications_list_data( + mode=resolved_mode, + selected_scholar_id=selected_scholar_id, + unread_count=unread_count, + latest_count=latest_count, + total_count=total_count, + publications=publications, ) + return success_payload(request, data=data) @router.post( @@ -173,3 +229,50 @@ async def mark_selected_publications_read( "updated_count": updated_count, }, ) + + +@router.post( + "/{publication_id}/retry-pdf", + response_model=RetryPublicationPdfEnvelope, +) +async def retry_publication_pdf( + payload: RetryPublicationPdfRequest, + request: Request, + publication_id: int = Path(ge=1), + db_session: AsyncSession = Depends(get_db_session), + current_user: User = Depends(get_api_current_user), +): + publication = await publication_service.retry_pdf_for_user( + db_session, + user_id=current_user.id, + scholar_profile_id=payload.scholar_profile_id, + publication_id=publication_id, + unpaywall_email=current_user.email, + ) + if publication is None: + raise ApiException( + status_code=404, + code="publication_not_found", + message="Publication not found.", + ) + resolved_pdf = bool(publication.pdf_url) + logger.info( + "api.publications.retry_pdf", + extra={ + "event": "api.publications.retry_pdf", + "user_id": current_user.id, + "scholar_profile_id": payload.scholar_profile_id, + "publication_id": publication_id, + "resolved_pdf": resolved_pdf, + "has_doi": bool(publication.doi), + }, + ) + message = "Open-access PDF link resolved." if resolved_pdf else "No open-access PDF link found." + return success_payload( + request, + data={ + "message": message, + "resolved_pdf": resolved_pdf, + "publication": _serialize_publication_item(publication), + }, + ) diff --git a/app/api/routers/runs.py b/app/api/routers/runs.py index ae9d803..ddd5290 100644 --- a/app/api/routers/runs.py +++ b/app/api/routers/runs.py @@ -21,10 +21,10 @@ from app.api.schemas import ( ) from app.db.models import RunStatus, RunTriggerType, User from app.db.session import get_db_session -from app.services import ingestion as ingestion_service -from app.services import run_safety as run_safety_service -from app.services import runs as run_service -from app.services import user_settings as user_settings_service +from app.services.domains.ingestion import application as ingestion_service +from app.services.domains.ingestion import safety as run_safety_service +from app.services.domains.runs import application as run_service +from app.services.domains.settings import application as user_settings_service from app.settings import settings from app.api.runtime_deps import get_ingestion_service @@ -294,10 +294,228 @@ async def _load_safety_state( "metric_name": "api_runs_safety_cooldown_cleared_total", "metric_value": 1, }, - ) + ) return run_safety_service.get_safety_state_payload(user_settings) +def _raise_manual_runs_disabled(*, user_id: int, safety_state: dict[str, Any]) -> None: + logger.warning( + "api.runs.manual_blocked_policy", + extra={ + "event": "api.runs.manual_blocked_policy", + "user_id": user_id, + "policy": {"manual_run_allowed": False}, + "safety_state": safety_state, + "metric_name": "api_runs_manual_blocked_policy_total", + "metric_value": 1, + }, + ) + raise ApiException( + status_code=403, + code="manual_runs_disabled", + message="Manual checks are disabled by server policy.", + details={ + "policy": {"manual_run_allowed": False}, + "safety_state": safety_state, + }, + ) + + +async def _reused_manual_run_payload( + db_session: AsyncSession, + *, + request: Request, + user_id: int, + idempotency_key: str | None, + safety_state: dict[str, Any], +) -> dict[str, Any] | None: + if idempotency_key is None: + return None + previous_run = await run_service.get_manual_run_by_idempotency_key( + db_session, + user_id=user_id, + idempotency_key=idempotency_key, + ) + if previous_run is None: + return None + if previous_run.status == RunStatus.RUNNING: + raise ApiException( + status_code=409, + code="run_in_progress", + message="A run with this idempotency key is still in progress.", + details={"run_id": int(previous_run.id), "idempotency_key": idempotency_key}, + ) + return success_payload( + request, + data=_manual_run_payload_from_run( + previous_run, + idempotency_key=idempotency_key, + reused_existing_run=True, + safety_state=safety_state, + ), + ) + + +async def _run_ingestion_for_manual( + db_session: AsyncSession, + *, + ingest_service: ingestion_service.ScholarIngestionService, + user_id: int, + idempotency_key: str | None, +): + user_settings = await user_settings_service.get_or_create_settings( + db_session, + user_id=user_id, + ) + return await ingest_service.run_for_user( + db_session, + user_id=user_id, + trigger_type=RunTriggerType.MANUAL, + request_delay_seconds=user_settings.request_delay_seconds, + network_error_retries=settings.ingestion_network_error_retries, + retry_backoff_seconds=settings.ingestion_retry_backoff_seconds, + max_pages_per_scholar=settings.ingestion_max_pages_per_scholar, + page_size=settings.ingestion_page_size, + auto_queue_continuations=settings.ingestion_continuation_queue_enabled, + queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds, + idempotency_key=idempotency_key, + alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold, + alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold, + alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold, + ) + + +async def _recover_integrity_error( + db_session: AsyncSession, + *, + request: Request, + user_id: int, + idempotency_key: str | None, + original_exc: IntegrityError, +) -> dict[str, Any]: + if idempotency_key is None: + logger.exception( + "api.runs.manual_integrity_error", + extra={"event": "api.runs.manual_integrity_error", "user_id": user_id}, + ) + raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc + existing_run = await run_service.get_manual_run_by_idempotency_key( + db_session, + user_id=user_id, + idempotency_key=idempotency_key, + ) + if existing_run is None: + logger.exception( + "api.runs.manual_integrity_error", + extra={"event": "api.runs.manual_integrity_error", "user_id": user_id}, + ) + raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc + if existing_run.status == RunStatus.RUNNING: + raise ApiException( + status_code=409, + code="run_in_progress", + message="A run with this idempotency key is still in progress.", + details={"run_id": int(existing_run.id), "idempotency_key": idempotency_key}, + ) from original_exc + return success_payload( + request, + data=_manual_run_payload_from_run( + existing_run, + idempotency_key=idempotency_key, + reused_existing_run=True, + safety_state=await _load_safety_state(db_session, user_id=user_id), + ), + ) + + +async def _execute_manual_run( + db_session: AsyncSession, + *, + request: Request, + ingest_service: ingestion_service.ScholarIngestionService, + user_id: int, + idempotency_key: str | None, +): + try: + return await _run_ingestion_for_manual( + db_session, + ingest_service=ingest_service, + user_id=user_id, + idempotency_key=idempotency_key, + ) + except ingestion_service.RunAlreadyInProgressError as exc: + await db_session.rollback() + raise ApiException( + status_code=409, + code="run_in_progress", + message="A run is already in progress for this account.", + ) from exc + except ingestion_service.RunBlockedBySafetyPolicyError as exc: + await db_session.rollback() + _raise_manual_blocked_safety(exc=exc, user_id=user_id) + except IntegrityError as exc: + await db_session.rollback() + return await _recover_integrity_error( + db_session, + request=request, + user_id=user_id, + idempotency_key=idempotency_key, + original_exc=exc, + ) + except Exception as exc: + await db_session.rollback() + _raise_manual_failed(exc=exc, user_id=user_id) + + +def _raise_manual_blocked_safety(*, exc, user_id: int) -> None: + logger.info( + "api.runs.manual_blocked_safety", + extra={ + "event": "api.runs.manual_blocked_safety", + "user_id": user_id, + "reason": exc.safety_state.get("cooldown_reason"), + "cooldown_until": exc.safety_state.get("cooldown_until"), + "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"), + "metric_name": "api_runs_manual_blocked_safety_total", + "metric_value": 1, + }, + ) + raise ApiException( + status_code=429, + code=exc.code, + message=exc.message, + details={"safety_state": exc.safety_state}, + ) from exc + + +def _raise_manual_failed(*, exc: Exception, user_id: int) -> None: + logger.exception( + "api.runs.manual_failed", + extra={"event": "api.runs.manual_failed", "user_id": user_id}, + ) + raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from exc + + +def _manual_run_success_payload( + *, + run_summary, + idempotency_key: str | None, + safety_state: dict[str, Any], +) -> dict[str, Any]: + return { + "run_id": run_summary.crawl_run_id, + "status": run_summary.status.value, + "scholar_count": run_summary.scholar_count, + "succeeded_count": run_summary.succeeded_count, + "failed_count": run_summary.failed_count, + "partial_count": run_summary.partial_count, + "new_publication_count": run_summary.new_publication_count, + "reused_existing_run": False, + "idempotency_key": idempotency_key, + "safety_state": safety_state, + } + + @router.get( "", response_model=RunsListEnvelope, @@ -383,164 +601,28 @@ async def run_manual( user_id=current_user.id, ) if not settings.ingestion_manual_run_allowed: - logger.warning( - "api.runs.manual_blocked_policy", - extra={ - "event": "api.runs.manual_blocked_policy", - "user_id": current_user.id, - "policy": {"manual_run_allowed": False}, - "safety_state": safety_state, - "metric_name": "api_runs_manual_blocked_policy_total", - "metric_value": 1, - }, - ) - raise ApiException( - status_code=403, - code="manual_runs_disabled", - message="Manual checks are disabled by server policy.", - details={ - "policy": { - "manual_run_allowed": False, - }, - "safety_state": safety_state, - }, - ) + _raise_manual_runs_disabled(user_id=current_user.id, safety_state=safety_state) idempotency_key = _normalize_idempotency_key(request.headers.get(IDEMPOTENCY_HEADER)) - if idempotency_key is not None: - previous_run = await run_service.get_manual_run_by_idempotency_key( - db_session, - user_id=current_user.id, - idempotency_key=idempotency_key, - ) - if previous_run is not None: - if previous_run.status == RunStatus.RUNNING: - raise ApiException( - status_code=409, - code="run_in_progress", - message="A run with this idempotency key is still in progress.", - details={ - "run_id": int(previous_run.id), - "idempotency_key": idempotency_key, - }, - ) - return success_payload( - request, - data=_manual_run_payload_from_run( - previous_run, - idempotency_key=idempotency_key, - reused_existing_run=True, - safety_state=safety_state, - ), - ) - - user_settings = await user_settings_service.get_or_create_settings( + reused_payload = await _reused_manual_run_payload( db_session, + request=request, user_id=current_user.id, + idempotency_key=idempotency_key, + safety_state=safety_state, ) - try: - run_summary = await ingest_service.run_for_user( - db_session, - user_id=current_user.id, - trigger_type=RunTriggerType.MANUAL, - request_delay_seconds=user_settings.request_delay_seconds, - network_error_retries=settings.ingestion_network_error_retries, - retry_backoff_seconds=settings.ingestion_retry_backoff_seconds, - max_pages_per_scholar=settings.ingestion_max_pages_per_scholar, - page_size=settings.ingestion_page_size, - auto_queue_continuations=settings.ingestion_continuation_queue_enabled, - queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds, - idempotency_key=idempotency_key, - alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold, - alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold, - alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold, - ) - except ingestion_service.RunAlreadyInProgressError as exc: - await db_session.rollback() - raise ApiException( - status_code=409, - code="run_in_progress", - message="A run is already in progress for this account.", - ) from exc - except ingestion_service.RunBlockedBySafetyPolicyError as exc: - await db_session.rollback() - logger.info( - "api.runs.manual_blocked_safety", - extra={ - "event": "api.runs.manual_blocked_safety", - "user_id": current_user.id, - "reason": exc.safety_state.get("cooldown_reason"), - "cooldown_until": exc.safety_state.get("cooldown_until"), - "cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"), - "metric_name": "api_runs_manual_blocked_safety_total", - "metric_value": 1, - }, - ) - raise ApiException( - status_code=429, - code=exc.code, - message=exc.message, - details={ - "safety_state": exc.safety_state, - }, - ) from exc - except IntegrityError as exc: - await db_session.rollback() - if idempotency_key is not None: - existing_run = await run_service.get_manual_run_by_idempotency_key( - db_session, - user_id=current_user.id, - idempotency_key=idempotency_key, - ) - if existing_run is not None: - if existing_run.status == RunStatus.RUNNING: - raise ApiException( - status_code=409, - code="run_in_progress", - message="A run with this idempotency key is still in progress.", - details={ - "run_id": int(existing_run.id), - "idempotency_key": idempotency_key, - }, - ) from exc - return success_payload( - request, - data=_manual_run_payload_from_run( - existing_run, - idempotency_key=idempotency_key, - reused_existing_run=True, - safety_state=await _load_safety_state( - db_session, - user_id=current_user.id, - ), - ), - ) - logger.exception( - "api.runs.manual_integrity_error", - extra={ - "event": "api.runs.manual_integrity_error", - "user_id": current_user.id, - }, - ) - raise ApiException( - status_code=500, - code="manual_run_failed", - message="Manual run failed.", - ) from exc - except Exception as exc: - await db_session.rollback() - logger.exception( - "api.runs.manual_failed", - extra={ - "event": "api.runs.manual_failed", - "user_id": current_user.id, - }, - ) - raise ApiException( - status_code=500, - code="manual_run_failed", - message="Manual run failed.", - ) from exc + if reused_payload is not None: + return reused_payload + + run_summary = await _execute_manual_run( + db_session, + request=request, + ingest_service=ingest_service, + user_id=current_user.id, + idempotency_key=idempotency_key, + ) + if isinstance(run_summary, dict): + return run_summary current_safety_state = await _load_safety_state( db_session, @@ -548,18 +630,11 @@ async def run_manual( ) return success_payload( request, - data={ - "run_id": run_summary.crawl_run_id, - "status": run_summary.status.value, - "scholar_count": run_summary.scholar_count, - "succeeded_count": run_summary.succeeded_count, - "failed_count": run_summary.failed_count, - "partial_count": run_summary.partial_count, - "new_publication_count": run_summary.new_publication_count, - "reused_existing_run": False, - "idempotency_key": idempotency_key, - "safety_state": current_safety_state, - }, + data=_manual_run_success_payload( + run_summary=run_summary, + idempotency_key=idempotency_key, + safety_state=current_safety_state, + ), ) diff --git a/app/api/routers/scholars.py b/app/api/routers/scholars.py index b6a7e47..7a8038a 100644 --- a/app/api/routers/scholars.py +++ b/app/api/routers/scholars.py @@ -25,9 +25,9 @@ from app.api.schemas import ( ) from app.db.models import User from app.db.session import get_db_session -from app.services import import_export as import_export_service -from app.services import scholars as scholar_service -from app.services.scholar_source import ScholarSource +from app.services.domains.portability import application as import_export_service +from app.services.domains.scholars import application as scholar_service +from app.services.domains.scholar.source import ScholarSource from app.settings import settings logger = logging.getLogger(__name__) @@ -64,6 +64,107 @@ def _serialize_scholar(profile) -> dict[str, object]: } +async def _hydrate_scholar_metadata_if_needed( + db_session: AsyncSession, + *, + profile, + source: ScholarSource, + user_id: int, +): + try: + if not profile.profile_image_url or not (profile.display_name or "").strip(): + return await asyncio.wait_for( + scholar_service.hydrate_profile_metadata( + db_session, + profile=profile, + source=source, + ), + timeout=5.0, + ) + except Exception: + logger.warning( + "api.scholars.create_metadata_hydration_failed", + extra={ + "event": "api.scholars.create_metadata_hydration_failed", + "user_id": user_id, + "scholar_profile_id": profile.id, + }, + ) + return profile + + +def _search_kwargs() -> dict[str, object]: + return { + "network_error_retries": settings.ingestion_network_error_retries, + "retry_backoff_seconds": settings.ingestion_retry_backoff_seconds, + "search_enabled": settings.scholar_name_search_enabled, + "cache_ttl_seconds": settings.scholar_name_search_cache_ttl_seconds, + "blocked_cache_ttl_seconds": settings.scholar_name_search_blocked_cache_ttl_seconds, + "cache_max_entries": settings.scholar_name_search_cache_max_entries, + "min_interval_seconds": settings.scholar_name_search_min_interval_seconds, + "interval_jitter_seconds": settings.scholar_name_search_interval_jitter_seconds, + "cooldown_block_threshold": settings.scholar_name_search_cooldown_block_threshold, + "cooldown_seconds": settings.scholar_name_search_cooldown_seconds, + "retry_alert_threshold": settings.scholar_name_search_alert_retry_count_threshold, + "cooldown_rejection_alert_threshold": ( + settings.scholar_name_search_alert_cooldown_rejections_threshold + ), + } + + +def _search_response_data(query: str, parsed) -> dict[str, object]: + return { + "query": query.strip(), + "state": parsed.state.value, + "state_reason": parsed.state_reason, + "action_hint": scholar_service.scrape_state_hint( + state=parsed.state, + state_reason=parsed.state_reason, + ), + "candidates": [ + { + "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 parsed.candidates + ], + "warnings": parsed.warnings, + } + + +async def _read_uploaded_image(image: UploadFile) -> bytes: + try: + return await image.read() + finally: + await image.close() + + +async def _require_user_profile( + db_session: AsyncSession, + *, + user_id: int, + scholar_profile_id: int, +): + profile = await scholar_service.get_user_scholar_by_id( + db_session, + user_id=user_id, + scholar_profile_id=scholar_profile_id, + ) + if profile is None: + raise ApiException( + status_code=404, + code="scholar_not_found", + message="Scholar not found.", + ) + return profile + + @router.get( "", response_model=ScholarsListEnvelope, @@ -181,25 +282,12 @@ async def create_scholar( "scholar_profile_id": created.id, }, ) - try: - if not created.profile_image_url or not (created.display_name or "").strip(): - created = await asyncio.wait_for( - scholar_service.hydrate_profile_metadata( - db_session, - profile=created, - source=source, - ), - timeout=5.0, - ) - except Exception: - logger.warning( - "api.scholars.create_metadata_hydration_failed", - extra={ - "event": "api.scholars.create_metadata_hydration_failed", - "user_id": current_user.id, - "scholar_profile_id": created.id, - }, - ) + created = await _hydrate_scholar_metadata_if_needed( + db_session, + profile=created, + source=source, + user_id=current_user.id, + ) return success_payload( request, @@ -225,20 +313,7 @@ async def search_scholars( db_session=db_session, query=query, limit=limit, - network_error_retries=settings.ingestion_network_error_retries, - retry_backoff_seconds=settings.ingestion_retry_backoff_seconds, - search_enabled=settings.scholar_name_search_enabled, - cache_ttl_seconds=settings.scholar_name_search_cache_ttl_seconds, - blocked_cache_ttl_seconds=settings.scholar_name_search_blocked_cache_ttl_seconds, - cache_max_entries=settings.scholar_name_search_cache_max_entries, - min_interval_seconds=settings.scholar_name_search_min_interval_seconds, - interval_jitter_seconds=settings.scholar_name_search_interval_jitter_seconds, - cooldown_block_threshold=settings.scholar_name_search_cooldown_block_threshold, - cooldown_seconds=settings.scholar_name_search_cooldown_seconds, - retry_alert_threshold=settings.scholar_name_search_alert_retry_count_threshold, - cooldown_rejection_alert_threshold=( - settings.scholar_name_search_alert_cooldown_rejections_threshold - ), + **_search_kwargs(), ) except scholar_service.ScholarServiceError as exc: raise ApiException( @@ -259,29 +334,7 @@ async def search_scholars( ) return success_payload( request, - data={ - "query": query.strip(), - "state": parsed.state.value, - "state_reason": parsed.state_reason, - "action_hint": scholar_service.scrape_state_hint( - state=parsed.state, - state_reason=parsed.state_reason, - ), - "candidates": [ - { - "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 parsed.candidates - ], - "warnings": parsed.warnings, - }, + data=_search_response_data(query, parsed), ) @@ -424,20 +477,14 @@ async def upload_scholar_image( db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_api_current_user), ): - profile = await scholar_service.get_user_scholar_by_id( + profile = await _require_user_profile( db_session, user_id=current_user.id, scholar_profile_id=scholar_profile_id, ) - if profile is None: - raise ApiException( - status_code=404, - code="scholar_not_found", - message="Scholar not found.", - ) + image_bytes = await _read_uploaded_image(image) try: - image_bytes = await image.read() updated = await scholar_service.set_profile_image_upload( db_session, profile=profile, @@ -452,9 +499,8 @@ async def upload_scholar_image( code="invalid_scholar_image", message=str(exc), ) from exc - finally: - await image.close() + image_size = len(image_bytes) logger.info( "api.scholars.image_uploaded", extra={ @@ -462,7 +508,7 @@ async def upload_scholar_image( "user_id": current_user.id, "scholar_profile_id": updated.id, "content_type": image.content_type, - "size_bytes": len(image_bytes), + "size_bytes": image_size, }, ) return success_payload( diff --git a/app/api/routers/settings.py b/app/api/routers/settings.py index f4b36a5..f5e9cac 100644 --- a/app/api/routers/settings.py +++ b/app/api/routers/settings.py @@ -11,8 +11,8 @@ from app.api.responses import success_payload from app.api.schemas import SettingsEnvelope, SettingsUpdateRequest from app.db.models import User from app.db.session import get_db_session -from app.services import run_safety as run_safety_service -from app.services import user_settings as user_settings_service +from app.services.domains.ingestion import safety as run_safety_service +from app.services.domains.settings import application as user_settings_service from app.settings import settings as settings_module logger = logging.getLogger(__name__) @@ -21,12 +21,7 @@ router = APIRouter(prefix="/settings", tags=["api-settings"]) def _serialize_settings(user_settings) -> dict[str, object]: - min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum( - settings_module.ingestion_min_run_interval_minutes - ) - min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum( - settings_module.ingestion_min_request_delay_seconds - ) + min_run_interval_minutes, min_request_delay_seconds = _minimum_policy() return { "auto_run_enabled": bool(user_settings.auto_run_enabled) and settings_module.ingestion_automation_allowed, "run_interval_minutes": int(user_settings.run_interval_minutes), @@ -46,6 +41,76 @@ def _serialize_settings(user_settings) -> dict[str, object]: } +def _minimum_policy() -> tuple[int, int]: + min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum( + settings_module.ingestion_min_run_interval_minutes + ) + min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum( + settings_module.ingestion_min_request_delay_seconds + ) + return min_run_interval_minutes, min_request_delay_seconds + + +def _parse_settings_payload(payload: SettingsUpdateRequest, user_settings) -> tuple[int, int, list[str]]: + min_run_interval_minutes, min_request_delay_seconds = _minimum_policy() + parsed_interval = user_settings_service.parse_run_interval_minutes( + str(payload.run_interval_minutes), + minimum=min_run_interval_minutes, + ) + parsed_delay = user_settings_service.parse_request_delay_seconds( + str(payload.request_delay_seconds), + minimum=min_request_delay_seconds, + ) + parsed_nav_visible_pages = user_settings_service.parse_nav_visible_pages( + payload.nav_visible_pages + if payload.nav_visible_pages is not None + else list(user_settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES) + ) + return parsed_interval, parsed_delay, parsed_nav_visible_pages + + +async def _clear_expired_cooldown_with_log( + db_session: AsyncSession, + *, + user_id: int, + user_settings, +) -> None: + previous_safety_state = run_safety_service.get_safety_event_context(user_settings) + if not run_safety_service.clear_expired_cooldown(user_settings): + return + await db_session.commit() + await db_session.refresh(user_settings) + logger.info( + "api.settings.safety_cooldown_cleared", + extra={ + "event": "api.settings.safety_cooldown_cleared", + "user_id": user_id, + "reason": previous_safety_state.get("cooldown_reason"), + "cooldown_until": previous_safety_state.get("cooldown_until"), + "metric_name": "api_settings_safety_cooldown_cleared_total", + "metric_value": 1, + }, + ) + + +def _effective_auto_run_enabled(payload: SettingsUpdateRequest) -> bool: + return bool(payload.auto_run_enabled) and settings_module.ingestion_automation_allowed + + +def _log_settings_update(*, user_id: int, updated) -> None: + logger.info( + "api.settings.updated", + extra={ + "event": "api.settings.updated", + "user_id": user_id, + "auto_run_enabled": updated.auto_run_enabled, + "run_interval_minutes": updated.run_interval_minutes, + "request_delay_seconds": updated.request_delay_seconds, + "nav_visible_pages": updated.nav_visible_pages, + }, + ) + + @router.get( "", response_model=SettingsEnvelope, @@ -59,21 +124,11 @@ async def get_settings( db_session, user_id=current_user.id, ) - previous_safety_state = run_safety_service.get_safety_event_context(user_settings) - if run_safety_service.clear_expired_cooldown(user_settings): - await db_session.commit() - await db_session.refresh(user_settings) - logger.info( - "api.settings.safety_cooldown_cleared", - extra={ - "event": "api.settings.safety_cooldown_cleared", - "user_id": current_user.id, - "reason": previous_safety_state.get("cooldown_reason"), - "cooldown_until": previous_safety_state.get("cooldown_until"), - "metric_name": "api_settings_safety_cooldown_cleared_total", - "metric_value": 1, - }, - ) + await _clear_expired_cooldown_with_log( + db_session, + user_id=current_user.id, + user_settings=user_settings, + ) return success_payload( request, data=_serialize_settings(user_settings), @@ -95,26 +150,10 @@ async def update_settings( user_id=current_user.id, ) - min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum( - settings_module.ingestion_min_run_interval_minutes - ) - min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum( - settings_module.ingestion_min_request_delay_seconds - ) - try: - parsed_interval = user_settings_service.parse_run_interval_minutes( - str(payload.run_interval_minutes), - minimum=min_run_interval_minutes, - ) - parsed_delay = user_settings_service.parse_request_delay_seconds( - str(payload.request_delay_seconds), - minimum=min_request_delay_seconds, - ) - parsed_nav_visible_pages = user_settings_service.parse_nav_visible_pages( - payload.nav_visible_pages - if payload.nav_visible_pages is not None - else list(user_settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES) + parsed_interval, parsed_delay, parsed_nav_visible_pages = _parse_settings_payload( + payload, + user_settings, ) except user_settings_service.UserSettingsServiceError as exc: raise ApiException( @@ -123,44 +162,20 @@ async def update_settings( message=str(exc), ) from exc - effective_auto_run_enabled = ( - bool(payload.auto_run_enabled) and settings_module.ingestion_automation_allowed - ) - updated = await user_settings_service.update_settings( db_session, settings=user_settings, - auto_run_enabled=effective_auto_run_enabled, + auto_run_enabled=_effective_auto_run_enabled(payload), run_interval_minutes=parsed_interval, request_delay_seconds=parsed_delay, nav_visible_pages=parsed_nav_visible_pages, ) - previous_safety_state = run_safety_service.get_safety_event_context(updated) - if run_safety_service.clear_expired_cooldown(updated): - await db_session.commit() - await db_session.refresh(updated) - logger.info( - "api.settings.safety_cooldown_cleared", - extra={ - "event": "api.settings.safety_cooldown_cleared", - "user_id": current_user.id, - "reason": previous_safety_state.get("cooldown_reason"), - "cooldown_until": previous_safety_state.get("cooldown_until"), - "metric_name": "api_settings_safety_cooldown_cleared_total", - "metric_value": 1, - }, - ) - logger.info( - "api.settings.updated", - extra={ - "event": "api.settings.updated", - "user_id": current_user.id, - "auto_run_enabled": updated.auto_run_enabled, - "run_interval_minutes": updated.run_interval_minutes, - "request_delay_seconds": updated.request_delay_seconds, - "nav_visible_pages": updated.nav_visible_pages, - }, + await _clear_expired_cooldown_with_log( + db_session, + user_id=current_user.id, + user_settings=updated, ) + _log_settings_update(user_id=current_user.id, updated=updated) return success_payload( request, data=_serialize_settings(updated), diff --git a/app/api/schemas.py b/app/api/schemas.py index c80afa5..9deb89a 100644 --- a/app/api/schemas.py +++ b/app/api/schemas.py @@ -205,6 +205,7 @@ class PublicationExportItemData(BaseModel): author_text: str | None = None venue_text: str | None = None pub_url: str | None = None + doi: str | None = None pdf_url: str | None = None is_read: bool = False @@ -573,6 +574,7 @@ class PublicationItemData(BaseModel): citation_count: int venue_text: str | None pub_url: str | None + doi: str | None pdf_url: str | None is_read: bool first_seen_at: datetime @@ -640,3 +642,24 @@ class MarkSelectedReadEnvelope(BaseModel): meta: ApiMeta model_config = ConfigDict(extra="forbid") + + +class RetryPublicationPdfRequest(BaseModel): + scholar_profile_id: int = Field(ge=1) + + model_config = ConfigDict(extra="forbid") + + +class RetryPublicationPdfData(BaseModel): + message: str + resolved_pdf: bool + publication: PublicationItemData + + model_config = ConfigDict(extra="forbid") + + +class RetryPublicationPdfEnvelope(BaseModel): + data: RetryPublicationPdfData + meta: ApiMeta + + model_config = ConfigDict(extra="forbid") diff --git a/app/auth/runtime.py b/app/auth/runtime.py index 5bb1a74..777e799 100644 --- a/app/auth/runtime.py +++ b/app/auth/runtime.py @@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.auth.session import clear_session_user, get_session_user, set_session_user from app.db.models import User from app.security.csrf import CSRF_SESSION_KEY -from app.services import users as user_service +from app.services.domains.users import application as user_service logger = logging.getLogger(__name__) diff --git a/app/db/models.py b/app/db/models.py index 2d7de3c..e9e4303 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -227,6 +227,7 @@ class Publication(Base): author_text: Mapped[str | None] = mapped_column(Text) venue_text: Mapped[str | None] = mapped_column(Text) pub_url: Mapped[str | None] = mapped_column(Text) + doi: Mapped[str | None] = mapped_column(String(255)) pdf_url: Mapped[str | None] = mapped_column(Text) created_at: Mapped[datetime] = mapped_column( DateTime(timezone=True), nullable=False, server_default=func.now() diff --git a/app/services/continuation_queue.py b/app/services/continuation_queue.py deleted file mode 100644 index a41fa8f..0000000 --- a/app/services/continuation_queue.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.ingestion.queue import * diff --git a/app/services/domains/crossref/__init__.py b/app/services/domains/crossref/__init__.py new file mode 100644 index 0000000..8adb7ba --- /dev/null +++ b/app/services/domains/crossref/__init__.py @@ -0,0 +1,5 @@ +from __future__ import annotations + +from app.services.domains.crossref.application import discover_doi_for_publication + +__all__ = ["discover_doi_for_publication"] diff --git a/app/services/domains/crossref/application.py b/app/services/domains/crossref/application.py new file mode 100644 index 0000000..569c12d --- /dev/null +++ b/app/services/domains/crossref/application.py @@ -0,0 +1,259 @@ +from __future__ import annotations + +import asyncio +import logging +import re +import threading +import time +from typing import TYPE_CHECKING + +from crossref.restful import Etiquette, Works + +from app.services.domains.doi.normalize import normalize_doi +from app.settings import settings + +if TYPE_CHECKING: + from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem + +TOKEN_RE = re.compile(r"[a-z0-9]+") +NON_ALNUM_RE = re.compile(r"[^a-z0-9\\s]+") +STOP_WORDS = {"the", "and", "for", "with", "from", "method", "study", "analysis"} +_RATE_LOCK = threading.Lock() +_LAST_REQUEST_AT = 0.0 +logger = logging.getLogger(__name__) + + +def _rate_limit_wait(min_interval_seconds: float) -> None: + global _LAST_REQUEST_AT + interval = max(float(min_interval_seconds), 0.0) + with _RATE_LOCK: + elapsed = time.monotonic() - _LAST_REQUEST_AT + remaining = interval - elapsed + if remaining > 0: + time.sleep(remaining) + _LAST_REQUEST_AT = time.monotonic() + + +def _normalized_tokens(value: str) -> list[str]: + lowered = value.lower().replace("’", "'").replace("“", "\"").replace("”", "\"") + lowered = NON_ALNUM_RE.sub(" ", lowered) + return [token for token in TOKEN_RE.findall(lowered) if len(token) >= 3] + + +def _normalized_query(value: str) -> str: + tokens = [token for token in _normalized_tokens(value) if token not in STOP_WORDS] + if len(tokens) < 3: + tokens = _normalized_tokens(value) + if len(tokens) < 3: + return "" + return " ".join(tokens[:12]).strip() + + +def _query_author(value: str) -> str | None: + tokens = [token for token in value.strip().split() if token] + if len(tokens) < 2: + return None + return " ".join(tokens[:2])[:64] + + +def _author_surname(value: str) -> str | None: + tokens = [token for token in value.strip().split() if token] + if not tokens: + return None + return NON_ALNUM_RE.sub("", tokens[-1].lower()) or None + + +def _query_filters(year: int | None) -> list[tuple[str, str] | None]: + if year is None: + return [None] + return [ + (f"{year - 1}-01-01", f"{year + 1}-12-31"), + (f"{year}-01-01", f"{year}-12-31"), + None, + ] + + +def _candidate_title(item: dict) -> str: + titles = item.get("title") + if isinstance(titles, list) and titles: + return str(titles[0] or "") + return str(item.get("title") or "") + + +def _title_match_score(source: str, candidate: str) -> float: + source_tokens = {token for token in _normalized_tokens(source) if len(token) >= 3} + candidate_tokens = {token for token in _normalized_tokens(candidate) if len(token) >= 3} + if not source_tokens or not candidate_tokens: + return 0.0 + return len(source_tokens & candidate_tokens) / float(len(source_tokens)) + + +def _candidate_year(item: dict) -> int | None: + issued = item.get("issued") + if not isinstance(issued, dict): + return None + date_parts = issued.get("date-parts") + if not isinstance(date_parts, list) or not date_parts: + return None + first = date_parts[0] + if not isinstance(first, list) or not first: + return None + try: + return int(first[0]) + except (TypeError, ValueError): + return None + + +def _candidate_author_match(item: dict, surname: str | None) -> bool: + if not surname: + return True + authors = item.get("author") + if not isinstance(authors, list): + return False + for author in authors: + if not isinstance(author, dict): + continue + family = NON_ALNUM_RE.sub("", str(author.get("family") or "").lower()) + if family and family == surname: + return True + return False + + +def _candidate_rank(*, title: str, year: int | None, item: dict) -> tuple[float, str | None]: + doi = normalize_doi(str(item.get("DOI") or "")) + if doi is None: + return 0.0, None + score = _title_match_score(title, _candidate_title(item)) + candidate_year = _candidate_year(item) + if year is not None and candidate_year is not None: + if abs(year - candidate_year) > 1: + return 0.0, None + score += 0.1 + return score, doi + + +def _best_candidate_doi( + *, + title: str, + year: int | None, + items: list[dict], + author_surname: str | None, +) -> str | None: + best_score = 0.0 + best_doi: str | None = None + best_year: int | None = None + for item in items: + if not isinstance(item, dict): + continue + if not _candidate_author_match(item, author_surname): + continue + score, doi = _candidate_rank(title=title, year=year, item=item) + candidate_year = _candidate_year(item) + if doi is None or score < 0.75: + continue + if score > best_score: + best_score = score + best_doi = doi + best_year = candidate_year + continue + if abs(score - best_score) > 0.02: + continue + if best_year is None or candidate_year is None: + continue + if candidate_year < best_year: + best_doi = doi + best_year = candidate_year + return best_doi + + +def _works_client(email: str | None) -> Works: + if email: + etiquette = Etiquette(settings.app_name, "0.1.0", "https://scholarr.local", email) + return Works(etiquette=etiquette) + return Works() + + +def _fetch_items_sync( + *, + query: str, + author: str | None, + date_range: tuple[str, str] | None, + max_rows: int, + email: str | None, + min_interval_seconds: float, +) -> list[dict]: + _rate_limit_wait(min_interval_seconds) + works = _works_client(email) + params = {"bibliographic": query} + if author: + params["author"] = author + request = works.query(**params) + if date_range is not None: + from_date, until_date = date_range + request = request.filter(from_pub_date=from_date, until_pub_date=until_date) + request = request.select(["DOI", "title", "issued", "score", "author"]) + items: list[dict] = [] + for entry in request: + if isinstance(entry, dict): + items.append(entry) + if len(items) >= max(max_rows, 1): + break + return items + + +async def _fetch_items( + *, + query: str, + author: str | None, + date_range: tuple[str, str] | None, + max_rows: int, + email: str | None, +) -> list[dict]: + timeout = max(float(settings.crossref_timeout_seconds), 0.5) + try: + return await asyncio.wait_for( + asyncio.to_thread( + _fetch_items_sync, + query=query, + author=author, + date_range=date_range, + max_rows=max_rows, + email=email, + min_interval_seconds=settings.crossref_min_interval_seconds, + ), + timeout=timeout, + ) + except Exception: + return [] + + +async def discover_doi_for_publication( + *, + item: PublicationListItem | UnreadPublicationItem, + max_rows: int = 10, + email: str | None = None, +) -> str | None: + title = (item.title or "").strip() + query = _normalized_query(title) + if not query: + return None + author = _query_author(item.scholar_label) + author_surname = _author_surname(item.scholar_label) + for date_range in _query_filters(item.year): + items = await _fetch_items( + query=query, + author=author, + date_range=date_range, + max_rows=max_rows, + email=email, + ) + doi = _best_candidate_doi( + title=title, + year=item.year, + items=items, + author_surname=author_surname, + ) + if doi: + logger.debug("crossref.doi_discovered", extra={"event": "crossref.doi_discovered"}) + return doi + return None diff --git a/app/services/domains/doi/normalize.py b/app/services/domains/doi/normalize.py new file mode 100644 index 0000000..4719cf3 --- /dev/null +++ b/app/services/domains/doi/normalize.py @@ -0,0 +1,23 @@ +from __future__ import annotations + +import re +from urllib.parse import unquote + +DOI_RE = re.compile(r"10\.\d{4,9}/[-._;()/:A-Z0-9]+", re.I) + + +def normalize_doi(value: str | None) -> str | None: + if not value: + return None + match = DOI_RE.search(unquote(value)) + if not match: + return None + return match.group(0).rstrip(" .;,)").lower() + + +def first_doi_from_texts(*values: str | None) -> str | None: + for value in values: + doi = normalize_doi(value) + if doi: + return doi + return None diff --git a/app/services/domains/ingestion/application.py b/app/services/domains/ingestion/application.py index 5a0fac8..aa2844d 100644 --- a/app/services/domains/ingestion/application.py +++ b/app/services/domains/ingestion/application.py @@ -28,6 +28,7 @@ from app.services.domains.ingestion.constants import ( RESUMABLE_PARTIAL_REASONS, RUN_LOCK_NAMESPACE, ) +from app.services.domains.doi.normalize import first_doi_from_texts from app.services.domains.ingestion.fingerprints import ( _build_body_excerpt, _dedupe_publication_candidates, @@ -55,6 +56,7 @@ from app.services.domains.scholar.parser import ( ParseState, ParsedProfilePage, PublicationCandidate, + ScholarParserError, parse_profile_page, ) from app.services.domains.scholar.source import FetchResult, ScholarSource @@ -1431,7 +1433,7 @@ class ScholarIngestionService: cstart=cstart, page_size=page_size, ) - parsed_page = parse_profile_page(fetch_result) + parsed_page = self._parse_profile_page_or_layout_error(fetch_result=fetch_result) attempt_log.append( self._attempt_log_entry( attempt=attempt_index + 1, @@ -1458,6 +1460,38 @@ class ScholarIngestionService: raise RuntimeError("Fetch-and-parse retry loop produced no result.") return fetch_result, parsed_page, attempt_log + def _parse_profile_page_or_layout_error( + self, + *, + fetch_result: FetchResult, + ) -> ParsedProfilePage: + try: + return parse_profile_page(fetch_result) + except ScholarParserError as exc: + return self._parsed_page_from_parser_error( + fetch_result=fetch_result, + code=exc.code, + ) + + @staticmethod + def _parsed_page_from_parser_error( + *, + fetch_result: FetchResult, + code: str, + ) -> ParsedProfilePage: + return ParsedProfilePage( + state=ParseState.LAYOUT_CHANGED, + state_reason=code, + profile_name=None, + profile_image_url=None, + publications=[], + marker_counts={}, + warnings=[code], + has_show_more_button=False, + has_operation_error_banner=False, + articles_range=None, + ) + @staticmethod def _page_log_entry( *, @@ -1942,7 +1976,8 @@ class ScholarIngestionService: author_text=candidate.authors_text, venue_text=candidate.venue_text, pub_url=build_publication_url(candidate.title_url), - pdf_url=build_publication_url(candidate.pdf_url), + doi=first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title), + pdf_url=None, ) db_session.add(publication) await db_session.flush() @@ -1976,8 +2011,9 @@ class ScholarIngestionService: publication.venue_text = candidate.venue_text if candidate.title_url: publication.pub_url = build_publication_url(candidate.title_url) - if candidate.pdf_url: - publication.pdf_url = build_publication_url(candidate.pdf_url) + local_doi = first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title) + if local_doi and not publication.doi: + publication.doi = local_doi async def _resolve_publication( self, diff --git a/app/services/domains/portability/exporting.py b/app/services/domains/portability/exporting.py index 5c89006..ef43a89 100644 --- a/app/services/domains/portability/exporting.py +++ b/app/services/domains/portability/exporting.py @@ -34,6 +34,7 @@ def _serialize_export_publication(row: tuple[Any, ...]) -> dict[str, Any]: author_text, venue_text, pub_url, + doi, pdf_url, is_read, ) = row @@ -47,6 +48,7 @@ def _serialize_export_publication(row: tuple[Any, ...]) -> dict[str, Any]: "author_text": author_text, "venue_text": venue_text, "pub_url": pub_url, + "doi": doi, "pdf_url": pdf_url, "is_read": bool(is_read), } @@ -73,6 +75,7 @@ async def export_user_data( Publication.author_text, Publication.venue_text, Publication.pub_url, + Publication.doi, Publication.pdf_url, ScholarPublication.is_read, ) diff --git a/app/services/domains/portability/publication_import.py b/app/services/domains/portability/publication_import.py index 62f81ab..daa1d5d 100644 --- a/app/services/domains/portability/publication_import.py +++ b/app/services/domains/portability/publication_import.py @@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import Publication, ScholarProfile, ScholarPublication from app.services.domains.ingestion.application import build_publication_url, normalize_title +from app.services.domains.doi.normalize import normalize_doi from app.services.domains.portability.normalize import ( _normalize_citation_count, _normalize_optional_text, @@ -70,6 +71,7 @@ def _apply_imported_publication_values( author_text: str | None, venue_text: str | None, pub_url: str | None, + doi: str | None, pdf_url: str | None, cluster_id: str | None, ) -> bool: @@ -96,6 +98,9 @@ def _apply_imported_publication_values( if pub_url and publication.pub_url != pub_url: publication.pub_url = pub_url updated = True + if doi and publication.doi != doi: + publication.doi = doi + updated = True if pdf_url and publication.pdf_url != pdf_url: publication.pdf_url = pdf_url updated = True @@ -112,6 +117,7 @@ def _new_publication( author_text: str | None, venue_text: str | None, pub_url: str | None, + doi: str | None, pdf_url: str | None, ) -> Publication: return Publication( @@ -124,6 +130,7 @@ def _new_publication( author_text=author_text, venue_text=venue_text, pub_url=pub_url, + doi=doi, pdf_url=pdf_url, ) @@ -226,6 +233,7 @@ def _build_imported_publication_input( venue_text=venue_text, cluster_id=_normalize_optional_text(item.get("cluster_id")), pub_url=build_publication_url(_normalize_optional_text(item.get("pub_url"))), + doi=normalize_doi(_normalize_optional_text(item.get("doi"))), pdf_url=build_publication_url(_normalize_optional_text(item.get("pdf_url"))), fingerprint=_resolve_fingerprint( title=title, @@ -277,6 +285,7 @@ async def _create_import_publication( author_text=payload.author_text, venue_text=payload.venue_text, pub_url=payload.pub_url, + doi=payload.doi, pdf_url=payload.pdf_url, ) db_session.add(publication) @@ -297,6 +306,7 @@ def _update_import_publication( author_text=payload.author_text, venue_text=payload.venue_text, pub_url=payload.pub_url, + doi=payload.doi, pdf_url=payload.pdf_url, cluster_id=payload.cluster_id, ) diff --git a/app/services/domains/portability/types.py b/app/services/domains/portability/types.py index 8851b13..2a7f6e8 100644 --- a/app/services/domains/portability/types.py +++ b/app/services/domains/portability/types.py @@ -19,6 +19,7 @@ class ImportedPublicationInput: venue_text: str | None cluster_id: str | None pub_url: str | None + doi: str | None pdf_url: str | None fingerprint: str is_read: bool diff --git a/app/services/domains/publications/application.py b/app/services/domains/publications/application.py index 7570de3..8e9e32a 100644 --- a/app/services/domains/publications/application.py +++ b/app/services/domains/publications/application.py @@ -8,8 +8,12 @@ from app.services.domains.publications.counts import ( from app.services.domains.publications.listing import ( list_for_user, list_new_for_latest_run_for_user, + retry_pdf_for_user, list_unread_for_user, ) +from app.services.domains.publications.enrichment import ( + schedule_missing_pdf_enrichment_for_user, +) from app.services.domains.publications.modes import ( MODE_ALL, MODE_LATEST, @@ -20,6 +24,7 @@ from app.services.domains.publications.modes import ( ) from app.services.domains.publications.queries import ( get_latest_completed_run_id_for_user, + get_publication_item_for_user, publications_query, ) from app.services.domains.publications.read_state import ( @@ -39,9 +44,12 @@ __all__ = [ "resolve_mode", "get_latest_completed_run_id_for_user", "publications_query", + "get_publication_item_for_user", "list_for_user", "list_unread_for_user", "list_new_for_latest_run_for_user", + "retry_pdf_for_user", + "schedule_missing_pdf_enrichment_for_user", "count_for_user", "count_unread_for_user", "count_latest_for_user", diff --git a/app/services/domains/publications/enrichment.py b/app/services/domains/publications/enrichment.py new file mode 100644 index 0000000..5c09357 --- /dev/null +++ b/app/services/domains/publications/enrichment.py @@ -0,0 +1,135 @@ +from __future__ import annotations + +import asyncio +import logging +import time + +from app.db.session import get_session_factory +from app.services.domains.publications.listing import ( + missing_pdf_items, + resolve_and_persist_oa_metadata, +) +from app.services.domains.publications.types import PublicationListItem +from app.settings import settings + +logger = logging.getLogger(__name__) + +_enrichment_lock = asyncio.Lock() +_inflight_publications: set[tuple[int, int]] = set() +_recent_attempt_seconds: dict[tuple[int, int], float] = {} +_scheduled_tasks: set[asyncio.Task[None]] = set() + + +def _cooldown_seconds() -> float: + return max(float(settings.unpaywall_retry_cooldown_seconds), 1.0) + + +def _prune_recent_attempts(now_seconds: float, *, cooldown_seconds: float) -> None: + expiry = cooldown_seconds * 3 + stale_keys = [ + key for key, attempted_seconds in _recent_attempt_seconds.items() + if now_seconds - attempted_seconds >= expiry + ] + for key in stale_keys: + _recent_attempt_seconds.pop(key, None) + + +async def _claim_items( + *, + user_id: int, + items: list[PublicationListItem], + max_items: int, +) -> list[PublicationListItem]: + candidates = missing_pdf_items(items, limit=max_items) + if not candidates: + return [] + now_seconds = time.monotonic() + cooldown_seconds = _cooldown_seconds() + claimed: list[PublicationListItem] = [] + async with _enrichment_lock: + _prune_recent_attempts(now_seconds, cooldown_seconds=cooldown_seconds) + for item in candidates: + key = (user_id, item.publication_id) + attempted_seconds = _recent_attempt_seconds.get(key) + if key in _inflight_publications: + continue + if attempted_seconds is not None and now_seconds - attempted_seconds < cooldown_seconds: + continue + _inflight_publications.add(key) + _recent_attempt_seconds[key] = now_seconds + claimed.append(item) + return claimed + + +async def _release_claims(*, user_id: int, publication_ids: list[int]) -> None: + async with _enrichment_lock: + for publication_id in publication_ids: + _inflight_publications.discard((user_id, publication_id)) + + +def _on_task_done(task: asyncio.Task[None]) -> None: + _scheduled_tasks.discard(task) + try: + task.result() + except Exception: + logger.exception( + "publications.enrichment.task_failed", + extra={"event": "publications.enrichment.task_failed"}, + ) + + +async def _run_enrichment( + *, + user_id: int, + request_email: str | None, + items: list[PublicationListItem], +) -> None: + publication_ids = [item.publication_id for item in items] + try: + session_factory = get_session_factory() + async with session_factory() as db_session: + await resolve_and_persist_oa_metadata( + db_session, + rows=items, + unpaywall_email=request_email, + ) + finally: + await _release_claims(user_id=user_id, publication_ids=publication_ids) + logger.info( + "publications.enrichment.completed", + extra={ + "event": "publications.enrichment.completed", + "user_id": user_id, + "publication_count": len(items), + }, + ) + + +async def schedule_missing_pdf_enrichment_for_user( + *, + user_id: int, + request_email: str | None, + items: list[PublicationListItem], + max_items: int, +) -> int: + claimed_items = await _claim_items(user_id=user_id, items=items, max_items=max_items) + if not claimed_items: + return 0 + task = asyncio.create_task( + _run_enrichment( + user_id=user_id, + request_email=request_email, + items=claimed_items, + ) + ) + _scheduled_tasks.add(task) + task.add_done_callback(_on_task_done) + logger.info( + "publications.enrichment.scheduled", + extra={ + "event": "publications.enrichment.scheduled", + "user_id": user_id, + "publication_count": len(claimed_items), + }, + ) + return len(claimed_items) diff --git a/app/services/domains/publications/listing.py b/app/services/domains/publications/listing.py index 751ccd2..69893a9 100644 --- a/app/services/domains/publications/listing.py +++ b/app/services/domains/publications/listing.py @@ -10,11 +10,100 @@ from app.services.domains.publications.modes import ( ) from app.services.domains.publications.queries import ( get_latest_completed_run_id_for_user, + get_publication_item_for_user, publication_list_item_from_row, publications_query, unread_item_from_row, ) from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem +from app.services.domains.unpaywall.application import resolve_publication_oa_metadata +from sqlalchemy import update +from app.db.models import Publication + + +def _with_oa_overrides( + rows: list[PublicationListItem], + oa_data: dict[int, tuple[str | None, str | None]], +) -> list[PublicationListItem]: + return [ + PublicationListItem( + publication_id=row.publication_id, + scholar_profile_id=row.scholar_profile_id, + scholar_label=row.scholar_label, + title=row.title, + year=row.year, + citation_count=row.citation_count, + venue_text=row.venue_text, + pub_url=row.pub_url, + doi=(oa_data.get(row.publication_id) or (None, None))[0] or row.doi, + pdf_url=(oa_data.get(row.publication_id) or (None, None))[1] or row.pdf_url, + is_read=row.is_read, + first_seen_at=row.first_seen_at, + is_new_in_latest_run=row.is_new_in_latest_run, + ) + for row in rows + ] + + +def _resolved_fields( + *, + row: PublicationListItem, + resolved: tuple[str | None, str | None] | None, +) -> tuple[str | None, str | None]: + if resolved is None: + return row.doi, row.pdf_url + resolved_doi, resolved_pdf = resolved + return resolved_doi or row.doi, resolved_pdf or row.pdf_url + + +async def _persist_resolved_metadata( + db_session: AsyncSession, + *, + rows: list[PublicationListItem], + oa_data: dict[int, tuple[str | None, str | None]], +) -> None: + by_id = {row.publication_id: row for row in rows} + updates: list[tuple[int, str | None, str | None]] = [] + for publication_id, resolved in oa_data.items(): + existing = by_id.get(publication_id) + if existing is None: + continue + next_doi, next_pdf = _resolved_fields(row=existing, resolved=resolved) + if next_doi == existing.doi and next_pdf == existing.pdf_url: + continue + updates.append((publication_id, next_doi, next_pdf)) + for publication_id, doi, pdf_url in updates: + await db_session.execute( + update(Publication) + .where(Publication.id == publication_id) + .values(doi=doi, pdf_url=pdf_url) + ) + if updates: + await db_session.commit() + + +def missing_pdf_items( + rows: list[PublicationListItem], + *, + limit: int, +) -> list[PublicationListItem]: + bounded_limit = max(0, int(limit)) + if bounded_limit == 0: + return [] + return [row for row in rows if not row.pdf_url][:bounded_limit] + + +async def resolve_and_persist_oa_metadata( + db_session: AsyncSession, + *, + rows: list[PublicationListItem], + unpaywall_email: str | None = None, +) -> dict[int, tuple[str | None, str | None]]: + if not rows: + return {} + oa_data = await resolve_publication_oa_metadata(rows, request_email=unpaywall_email) + await _persist_resolved_metadata(db_session, rows=rows, oa_data=oa_data) + return oa_data async def list_for_user( @@ -36,10 +125,35 @@ async def list_for_user( limit=limit, ) ) - return [ + rows = [ publication_list_item_from_row(row, latest_run_id=latest_run_id) for row in result.all() ] + return rows + + +async def retry_pdf_for_user( + db_session: AsyncSession, + *, + user_id: int, + scholar_profile_id: int, + publication_id: int, + unpaywall_email: str | None = None, +) -> PublicationListItem | None: + item = await get_publication_item_for_user( + db_session, + user_id=user_id, + scholar_profile_id=scholar_profile_id, + publication_id=publication_id, + ) + if item is None: + return None + oa_data = await resolve_and_persist_oa_metadata( + db_session, + rows=[item], + unpaywall_email=unpaywall_email, + ) + return _with_oa_overrides([item], oa_data)[0] async def list_unread_for_user( @@ -73,17 +187,19 @@ async def list_new_for_latest_run_for_user( scholar_profile_id=None, limit=limit, ) - return [ - UnreadPublicationItem( - publication_id=row.publication_id, - scholar_profile_id=row.scholar_profile_id, - scholar_label=row.scholar_label, - title=row.title, - year=row.year, - citation_count=row.citation_count, - venue_text=row.venue_text, - pub_url=row.pub_url, - pdf_url=row.pdf_url, - ) - for row in rows - ] + return [_to_unread_item(row) for row in rows] + + +def _to_unread_item(row: PublicationListItem) -> UnreadPublicationItem: + return UnreadPublicationItem( + publication_id=row.publication_id, + scholar_profile_id=row.scholar_profile_id, + scholar_label=row.scholar_label, + title=row.title, + year=row.year, + citation_count=row.citation_count, + venue_text=row.venue_text, + pub_url=row.pub_url, + doi=row.doi, + pdf_url=row.pdf_url, + ) diff --git a/app/services/domains/publications/queries.py b/app/services/domains/publications/queries.py index 1c878df..b55d567 100644 --- a/app/services/domains/publications/queries.py +++ b/app/services/domains/publications/queries.py @@ -50,6 +50,7 @@ def publications_query( Publication.citation_count, Publication.venue_text, Publication.pub_url, + Publication.doi, Publication.pdf_url, ScholarPublication.is_read, ScholarPublication.first_seen_run_id, @@ -72,6 +73,61 @@ def publications_query( return stmt +def publication_query_for_user( + *, + user_id: int, + scholar_profile_id: int, + publication_id: int, +) -> Select[tuple]: + return ( + select( + Publication.id, + ScholarProfile.id, + ScholarProfile.display_name, + ScholarProfile.scholar_id, + Publication.title_raw, + Publication.year, + Publication.citation_count, + Publication.venue_text, + Publication.pub_url, + Publication.doi, + Publication.pdf_url, + ScholarPublication.is_read, + ScholarPublication.first_seen_run_id, + ScholarPublication.created_at, + ) + .join(ScholarPublication, ScholarPublication.publication_id == Publication.id) + .join(ScholarProfile, ScholarProfile.id == ScholarPublication.scholar_profile_id) + .where( + ScholarProfile.user_id == user_id, + ScholarProfile.id == scholar_profile_id, + Publication.id == publication_id, + ) + .limit(1) + ) + + +async def get_publication_item_for_user( + db_session: AsyncSession, + *, + user_id: int, + scholar_profile_id: int, + publication_id: int, +) -> PublicationListItem | None: + latest_run_id = await get_latest_completed_run_id_for_user(db_session, user_id=user_id) + result = await db_session.execute( + publication_query_for_user( + user_id=user_id, + scholar_profile_id=scholar_profile_id, + publication_id=publication_id, + ) + ) + row = result.one_or_none() + if row is None: + return None + return publication_list_item_from_row(row, latest_run_id=latest_run_id) + + def publication_list_item_from_row( row: tuple, *, @@ -87,6 +143,7 @@ def publication_list_item_from_row( citation_count, venue_text, pub_url, + doi, pdf_url, is_read, first_seen_run_id, @@ -101,6 +158,7 @@ def publication_list_item_from_row( citation_count=_normalized_citation_count(citation_count), venue_text=venue_text, pub_url=pub_url, + doi=doi, pdf_url=pdf_url, is_read=bool(is_read), first_seen_at=created_at, @@ -121,6 +179,7 @@ def unread_item_from_row(row: tuple) -> UnreadPublicationItem: citation_count, venue_text, pub_url, + doi, pdf_url, _is_read, _first_seen_run_id, @@ -135,5 +194,6 @@ def unread_item_from_row(row: tuple) -> UnreadPublicationItem: citation_count=_normalized_citation_count(citation_count), venue_text=venue_text, pub_url=pub_url, + doi=doi, pdf_url=pdf_url, ) diff --git a/app/services/domains/publications/types.py b/app/services/domains/publications/types.py index d64a812..f3a12c0 100644 --- a/app/services/domains/publications/types.py +++ b/app/services/domains/publications/types.py @@ -14,6 +14,7 @@ class PublicationListItem: citation_count: int venue_text: str | None pub_url: str | None + doi: str | None pdf_url: str | None is_read: bool first_seen_at: datetime @@ -30,4 +31,5 @@ class UnreadPublicationItem: citation_count: int venue_text: str | None pub_url: str | None + doi: str | None pdf_url: str | None diff --git a/app/services/domains/scholar/parser.py b/app/services/domains/scholar/parser.py index 49dc773..060e825 100644 --- a/app/services/domains/scholar/parser.py +++ b/app/services/domains/scholar/parser.py @@ -11,14 +11,12 @@ from app.services.domains.scholar.parser_types import ( ParsedAuthorSearchPage, ParsedProfilePage, PublicationCandidate, + ScholarDomInvariantError, + ScholarMalformedDataError, + ScholarParserError, ScholarSearchCandidate, ) from app.services.domains.scholar.parser_utils import ( - attr_class, - attr_href, - attr_src, - build_absolute_scholar_url, - normalize_space, strip_tags, ) from app.services.domains.scholar.profile_rows import ( @@ -44,6 +42,55 @@ from app.services.domains.scholar.state_detection import ( ) +def _raise_dom_error(code: str, message: str) -> None: + raise ScholarDomInvariantError(code=code, message=message) + + +def _assert_profile_dom_invariants( + *, + fetch_result: FetchResult, + marker_counts: dict[str, int], + publications: list[PublicationCandidate], + warnings: list[str], + has_show_more_button_flag: bool, + articles_range: str | None, +) -> None: + if fetch_result.status_code is None: + return + final_url = (fetch_result.final_url or "").lower() + if "accounts.google.com" in final_url or "sorry/index" in final_url: + return + if any(code.startswith("layout_") for code in warnings): + reason = next(code for code in warnings if code.startswith("layout_")) + _raise_dom_error(reason, f"Detected layout warning: {reason}") + if has_show_more_button_flag and not articles_range: + _raise_dom_error( + "layout_show_more_without_articles_range", + "Show-more control exists without an articles range marker.", + ) + if marker_counts.get("gsc_a_tr", 0) > 0 and marker_counts.get("gsc_a_at", 0) <= 0: + _raise_dom_error( + "layout_missing_publication_title_anchor", + "Publication rows were present but title anchors were absent.", + ) + if not publications: + has_profile_markers = marker_counts.get("gsc_prf_in", 0) > 0 + has_table_markers = marker_counts.get("gsc_a_tr", 0) > 0 or marker_counts.get("gsc_a_at", 0) > 0 + if not has_profile_markers and not has_table_markers: + _raise_dom_error("layout_markers_missing", "Expected scholar profile markers were absent.") + for publication in publications: + if not publication.title.strip(): + raise ScholarMalformedDataError( + code="malformed_publication_title", + message="Encountered a publication candidate with an empty title.", + ) + if publication.citation_count is not None and int(publication.citation_count) < 0: + raise ScholarMalformedDataError( + code="malformed_publication_negative_citations", + message="Encountered a publication candidate with negative citations.", + ) + + def parse_profile_page(fetch_result: FetchResult) -> ParsedProfilePage: publications, warnings = parse_publications(fetch_result.body) marker_counts = count_markers(fetch_result.body) @@ -59,6 +106,14 @@ def parse_profile_page(fetch_result: FetchResult) -> ParsedProfilePage: warnings.append("operation_error_banner_present") warnings = sorted(set(warnings)) + _assert_profile_dom_invariants( + fetch_result=fetch_result, + marker_counts=marker_counts, + publications=publications, + warnings=warnings, + has_show_more_button_flag=show_more, + articles_range=articles_range, + ) state, state_reason = detect_state( fetch_result, diff --git a/app/services/domains/scholar/parser_types.py b/app/services/domains/scholar/parser_types.py index a536389..7b657d8 100644 --- a/app/services/domains/scholar/parser_types.py +++ b/app/services/domains/scholar/parser_types.py @@ -12,6 +12,22 @@ class ParseState(StrEnum): NETWORK_ERROR = "network_error" +class ScholarParserError(RuntimeError): + code: str + + def __init__(self, *, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +class ScholarDomInvariantError(ScholarParserError): + pass + + +class ScholarMalformedDataError(ScholarParserError): + pass + + @dataclass(frozen=True) class PublicationCandidate: title: str diff --git a/app/services/domains/scholar/profile_rows.py b/app/services/domains/scholar/profile_rows.py index 896bcaa..4ae3bc4 100644 --- a/app/services/domains/scholar/profile_rows.py +++ b/app/services/domains/scholar/profile_rows.py @@ -2,13 +2,10 @@ from __future__ import annotations import re from html.parser import HTMLParser -from typing import Any from urllib.parse import parse_qs, urlparse from app.services.domains.scholar.parser_constants import ( MARKER_KEYS, - PROFILE_ROW_DIRECT_LABEL_TOKENS, - PROFILE_ROW_PARSER_DIRECT_MARKERS, SHOW_MORE_BUTTON_RE, ) from app.services.domains.scholar.parser_types import PublicationCandidate @@ -26,7 +23,6 @@ class ScholarRowParser(HTMLParser): def __init__(self) -> None: super().__init__(convert_charrefs=True) self.title_href: str | None = None - self.direct_download_href: str | None = None self.title_parts: list[str] = [] self.citation_parts: list[str] = [] self.year_parts: list[str] = [] @@ -35,14 +31,7 @@ class ScholarRowParser(HTMLParser): self._title_depth = 0 self._citation_depth = 0 self._year_depth = 0 - self._gray_stack: list[dict[str, Any]] = [] - self._direct_marker_depth = 0 - self._aux_link_stack: list[dict[str, Any]] = [] - - @staticmethod - def _contains_direct_marker(classes: str) -> bool: - lowered = classes.lower() - return any(marker in lowered for marker in PROFILE_ROW_PARSER_DIRECT_MARKERS) + self._gray_stack: list[dict[str, object]] = [] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: if self._title_depth > 0: @@ -53,14 +42,8 @@ class ScholarRowParser(HTMLParser): self._year_depth += 1 if self._gray_stack: self._gray_stack[-1]["depth"] += 1 - if self._direct_marker_depth > 0: - self._direct_marker_depth += 1 - if self._aux_link_stack: - self._aux_link_stack[-1]["depth"] += 1 classes = attr_class(attrs) - if tag in {"div", "span"} and self._contains_direct_marker(classes): - self._direct_marker_depth = 1 if tag == "a" and "gsc_a_at" in classes: self._title_depth = 1 @@ -79,17 +62,6 @@ class ScholarRowParser(HTMLParser): self._gray_stack.append({"depth": 1, "parts": []}) return - if tag == "a": - href = attr_href(attrs) - self._aux_link_stack.append( - { - "depth": 1, - "classes": classes, - "href": href, - "parts": [], - } - ) - def handle_data(self, data: str) -> None: if self._title_depth > 0: self.title_parts.append(data) @@ -99,23 +71,6 @@ class ScholarRowParser(HTMLParser): self.year_parts.append(data) if self._gray_stack: self._gray_stack[-1]["parts"].append(data) - if self._aux_link_stack: - self._aux_link_stack[-1]["parts"].append(data) - - def _capture_direct_download_href(self, link_info: dict[str, Any]) -> None: - if self.direct_download_href: - return - href = link_info.get("href") - if not href: - return - classes = str(link_info.get("classes") or "") - label = normalize_space("".join(link_info.get("parts") or [])).lower() - if "gs_ggsd" in classes or "gs_ggs" in classes or "gs_ggsa" in classes: - self.direct_download_href = href - return - label_match = any(token in label for token in PROFILE_ROW_DIRECT_LABEL_TOKENS) - if self._direct_marker_depth > 0 and label_match: - self.direct_download_href = href def handle_endtag(self, tag: str) -> None: if self._title_depth > 0: @@ -131,13 +86,6 @@ class ScholarRowParser(HTMLParser): if text_value: self.gray_texts.append(text_value) self._gray_stack.pop() - if self._aux_link_stack: - self._aux_link_stack[-1]["depth"] -= 1 - if self._aux_link_stack[-1]["depth"] == 0: - link_info = self._aux_link_stack.pop() - self._capture_direct_download_href(link_info) - if self._direct_marker_depth > 0: - self._direct_marker_depth -= 1 def extract_rows(html: str) -> list[str]: @@ -223,7 +171,7 @@ def _parse_publication_row(row_html: str) -> tuple[PublicationCandidate | None, citation_count=citation_count, authors_text=authors_text, venue_text=venue_text, - pdf_url=build_absolute_scholar_url(parser.direct_download_href), + pdf_url=None, ), warnings, ) diff --git a/app/services/domains/scholars/application.py b/app/services/domains/scholars/application.py index 65a3bee..5581bca 100644 --- a/app/services/domains/scholars/application.py +++ b/app/services/domains/scholars/application.py @@ -18,6 +18,7 @@ from app.db.models import AuthorSearchCacheEntry, AuthorSearchRuntimeState, Scho from app.services.domains.scholar.parser import ( ParseState, ParsedAuthorSearchPage, + ScholarParserError, parse_author_search_page, parse_profile_page, ) @@ -635,7 +636,16 @@ async def _fetch_author_search_with_retries( retry_scheduled_count = 0 for attempt_index in range(max_attempts): fetch_result = await source.fetch_author_search_html(normalized_query, start=0) - parsed = parse_author_search_page(fetch_result) + 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") @@ -873,7 +883,10 @@ async def hydrate_profile_metadata( source: ScholarSource, ) -> ScholarProfile: fetch_result = await source.fetch_profile_html(profile.scholar_id) - parsed_page = parse_profile_page(fetch_result) + try: + parsed_page = parse_profile_page(fetch_result) + except ScholarParserError: + return profile if parsed_page.profile_name and not (profile.display_name or "").strip(): profile.display_name = parsed_page.profile_name diff --git a/app/services/domains/unpaywall/__init__.py b/app/services/domains/unpaywall/__init__.py new file mode 100644 index 0000000..762e664 --- /dev/null +++ b/app/services/domains/unpaywall/__init__.py @@ -0,0 +1,5 @@ +from __future__ import annotations + +from app.services.domains.unpaywall.application import resolve_publication_pdf_urls + +__all__ = ["resolve_publication_pdf_urls"] diff --git a/app/services/domains/unpaywall/application.py b/app/services/domains/unpaywall/application.py new file mode 100644 index 0000000..703b138 --- /dev/null +++ b/app/services/domains/unpaywall/application.py @@ -0,0 +1,218 @@ +from __future__ import annotations + +import logging +import re +from urllib.parse import unquote + +from app.services.domains.crossref.application import discover_doi_for_publication +from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem +from app.services.domains.doi.normalize import normalize_doi +from app.settings import settings + +DOI_PATTERN = re.compile(r"10\.\d{4,9}/[-._;()/:A-Z0-9]+", re.I) +DOI_PREFIX_RE = re.compile(r"\bdoi\s*[:=]\s*(10\.\d{4,9}/[-._;()/:A-Z0-9]+)", re.I) +DOI_URL_RE = re.compile(r"(?:https?://)?(?:dx\.)?doi\.org/(10\.\d{4,9}/[-._;()/:A-Z0-9]+)", re.I) +UNPAYWALL_URL_TEMPLATE = "https://api.unpaywall.org/v2/{doi}" +logger = logging.getLogger(__name__) + + +def _extract_doi_candidate(text: str | None) -> str | None: + if not text: + return None + decoded = unquote(text) + match = DOI_PATTERN.search(decoded) + if not match: + return None + return match.group(0).rstrip(" .;,)") + + +def _extract_explicit_doi(text: str | None) -> str | None: + if not text: + return None + decoded = unquote(text) + url_match = DOI_URL_RE.search(decoded) + if url_match: + return normalize_doi(url_match.group(1)) + prefix_match = DOI_PREFIX_RE.search(decoded) + if prefix_match: + return normalize_doi(prefix_match.group(1)) + return None + + +def _publication_doi(item: PublicationListItem | UnreadPublicationItem) -> str | None: + stored = normalize_doi(item.doi) + if stored: + in_metadata = any( + normalize_doi(_extract_explicit_doi(value)) == stored + for value in (item.pub_url, item.venue_text) + ) + if in_metadata: + return stored + pub_url_doi = _extract_doi_candidate(item.pub_url) + if pub_url_doi: + return normalize_doi(pub_url_doi) + return ( + _extract_explicit_doi(item.pub_url) + or _extract_explicit_doi(item.venue_text) + ) + + +def _payload_pdf_url(payload: dict) -> str | None: + best = payload.get("best_oa_location") + if isinstance(best, dict): + pdf_url = best.get("url_for_pdf") + if isinstance(pdf_url, str) and pdf_url.strip(): + return pdf_url.strip() + locations = payload.get("oa_locations") + if not isinstance(locations, list): + return None + for location in locations: + if not isinstance(location, dict): + continue + pdf_url = location.get("url_for_pdf") + if isinstance(pdf_url, str) and pdf_url.strip(): + return pdf_url.strip() + return None + + +async def _fetch_unpaywall_payload_by_doi( + *, + client, + doi: str, + email: str, +) -> dict | None: + response = await client.get( + UNPAYWALL_URL_TEMPLATE.format(doi=doi), + params={"email": email}, + ) + if response.status_code != 200: + return None + payload = response.json() + if not isinstance(payload, dict): + return None + return payload + + +def _email_for_request(request_email: str | None) -> str | None: + email = (request_email or "").strip() or settings.unpaywall_email.strip() + return email or None + + +def _log_resolution_summary( + *, + publication_count: int, + doi_input_count: int, + search_attempt_count: int, + resolved_pdf_count: int, + email: str, +) -> None: + logger.info( + "unpaywall.resolve_completed", + extra={ + "event": "unpaywall.resolve_completed", + "publication_count": publication_count, + "doi_input_count": doi_input_count, + "search_attempt_count": search_attempt_count, + "resolved_pdf_count": resolved_pdf_count, + "email_domain": email.split("@", 1)[-1] if "@" in email else None, + }, + ) + + +async def _resolve_item_payload( + *, + client, + item: PublicationListItem, + email: str, + allow_crossref: bool, +) -> tuple[dict | None, bool]: + doi = _publication_doi(item) + payload: dict | None = None + if doi: + payload = await _fetch_unpaywall_payload_by_doi(client=client, doi=doi, email=email) + if payload is not None and _payload_pdf_url(payload): + return payload, False + if not allow_crossref or not settings.crossref_enabled: + return payload, False + crossref_doi = await discover_doi_for_publication( + item=item, + max_rows=settings.crossref_max_rows, + email=email, + ) + if crossref_doi is None or crossref_doi == doi: + return payload, crossref_doi is not None + crossref_payload = await _fetch_unpaywall_payload_by_doi( + client=client, + doi=crossref_doi, + email=email, + ) + if crossref_payload is not None: + return crossref_payload, True + return payload, True + + +def _doi_and_pdf_from_payload(payload: dict) -> tuple[str | None, str | None]: + doi = normalize_doi(str(payload.get("doi") or "")) + return doi, _payload_pdf_url(payload) + + +def _resolution_targets(items: list[PublicationListItem]) -> list[PublicationListItem]: + return [item for item in items if not item.pdf_url] + + +def _crossref_budget_value() -> int: + return max(int(settings.crossref_max_lookups_per_request), 0) + + +async def resolve_publication_oa_metadata( + items: list[PublicationListItem], + *, + request_email: str | None = None, +) -> dict[int, tuple[str | None, str | None]]: + if not settings.unpaywall_enabled: + return {} + email = _email_for_request(request_email) + if email is None: + logger.debug("unpaywall.resolve_skipped_missing_email") + return {} + import httpx + + timeout_seconds = max(float(settings.unpaywall_timeout_seconds), 0.5) + resolved: dict[int, tuple[str | None, str | None]] = {} + crossref_budget = _crossref_budget_value() + crossref_lookups = 0 + targets = _resolution_targets(items)[: max(int(settings.unpaywall_max_items_per_request), 0)] + async with httpx.AsyncClient(timeout=timeout_seconds) as client: + for item in targets: + allow_crossref = crossref_budget > 0 and crossref_lookups < crossref_budget + payload, used_crossref = await _resolve_item_payload( + client=client, + item=item, + email=email, + allow_crossref=allow_crossref, + ) + if used_crossref: + crossref_lookups += 1 + if not isinstance(payload, dict): + continue + resolved[item.publication_id] = _doi_and_pdf_from_payload(payload) + resolved_count = sum(1 for _doi, pdf in resolved.values() if pdf) + doi_input_count = len([item for item in items if _publication_doi(item)]) + target_doi_count = len([item for item in targets if _publication_doi(item)]) + _log_resolution_summary( + publication_count=len(items), + doi_input_count=doi_input_count, + search_attempt_count=max(0, len(targets) - target_doi_count), + resolved_pdf_count=resolved_count, + email=email, + ) + return resolved + + +async def resolve_publication_pdf_urls( + items: list[PublicationListItem], + *, + request_email: str | None = None, +) -> dict[int, str | None]: + resolved = await resolve_publication_oa_metadata(items, request_email=request_email) + return {publication_id: pdf for publication_id, (_doi, pdf) in resolved.items()} diff --git a/app/services/import_export.py b/app/services/import_export.py deleted file mode 100644 index ba710a6..0000000 --- a/app/services/import_export.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.portability.application import * diff --git a/app/services/ingestion.py b/app/services/ingestion.py deleted file mode 100644 index 2b2bcc9..0000000 --- a/app/services/ingestion.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.ingestion.application import * diff --git a/app/services/publications.py b/app/services/publications.py deleted file mode 100644 index f77b641..0000000 --- a/app/services/publications.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.publications.application import * diff --git a/app/services/run_safety.py b/app/services/run_safety.py deleted file mode 100644 index 1c9329d..0000000 --- a/app/services/run_safety.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.ingestion.safety import * diff --git a/app/services/runs.py b/app/services/runs.py deleted file mode 100644 index e917e47..0000000 --- a/app/services/runs.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.runs.application import * diff --git a/app/services/scheduler.py b/app/services/scheduler.py deleted file mode 100644 index 15b5656..0000000 --- a/app/services/scheduler.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.ingestion.scheduler import * diff --git a/app/services/scholar_parser.py b/app/services/scholar_parser.py deleted file mode 100644 index c31db46..0000000 --- a/app/services/scholar_parser.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.scholar.parser import * diff --git a/app/services/scholar_source.py b/app/services/scholar_source.py deleted file mode 100644 index b4a9147..0000000 --- a/app/services/scholar_source.py +++ /dev/null @@ -1,4 +0,0 @@ -from __future__ import annotations - -from app.services.domains.scholar.source import * -from app.services.domains.scholar.source import _build_profile_url diff --git a/app/services/scholars.py b/app/services/scholars.py deleted file mode 100644 index ebba422..0000000 --- a/app/services/scholars.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.scholars.application import * diff --git a/app/services/user_settings.py b/app/services/user_settings.py deleted file mode 100644 index c3be1d7..0000000 --- a/app/services/user_settings.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.settings.application import * diff --git a/app/services/users.py b/app/services/users.py deleted file mode 100644 index 12c56f6..0000000 --- a/app/services/users.py +++ /dev/null @@ -1,3 +0,0 @@ -from __future__ import annotations - -from app.services.domains.users.application import * diff --git a/app/settings.py b/app/settings.py index 3656d74..82988ac 100644 --- a/app/settings.py +++ b/app/settings.py @@ -229,6 +229,16 @@ class Settings: "SCHOLAR_NAME_SEARCH_ALERT_COOLDOWN_REJECTIONS_THRESHOLD", 3, ) + unpaywall_enabled: bool = _env_bool("UNPAYWALL_ENABLED", True) + unpaywall_email: str = _env_str("UNPAYWALL_EMAIL", "") + unpaywall_timeout_seconds: float = _env_float("UNPAYWALL_TIMEOUT_SECONDS", 4.0) + unpaywall_max_items_per_request: int = _env_int("UNPAYWALL_MAX_ITEMS_PER_REQUEST", 20) + unpaywall_retry_cooldown_seconds: int = _env_int("UNPAYWALL_RETRY_COOLDOWN_SECONDS", 1800) + crossref_enabled: bool = _env_bool("CROSSREF_ENABLED", True) + crossref_max_rows: int = _env_int("CROSSREF_MAX_ROWS", 10) + crossref_timeout_seconds: float = _env_float("CROSSREF_TIMEOUT_SECONDS", 8.0) + crossref_min_interval_seconds: float = _env_float("CROSSREF_MIN_INTERVAL_SECONDS", 0.6) + crossref_max_lookups_per_request: int = _env_int("CROSSREF_MAX_LOOKUPS_PER_REQUEST", 8) settings = Settings() diff --git a/frontend/src/components/layout/AppNav.vue b/frontend/src/components/layout/AppNav.vue index 71205c6..3345074 100644 --- a/frontend/src/components/layout/AppNav.vue +++ b/frontend/src/components/layout/AppNav.vue @@ -81,7 +81,7 @@ function onNavigate(): void { class="min-h-0 min-w-0 overflow-x-hidden border-b border-stroke-subtle bg-surface-nav/70 px-4 py-4 lg:h-full lg:border-b-0 lg:border-r lg:px-5 lg:py-6" >
-