diff --git a/.env.example b/.env.example index 326e7d3..e2180ba 100644 --- a/.env.example +++ b/.env.example @@ -79,12 +79,15 @@ LOG_REDACT_FIELDS= SCHEDULER_ENABLED=1 SCHEDULER_TICK_SECONDS=60 SCHEDULER_QUEUE_BATCH_SIZE=10 +SCHEDULER_PDF_QUEUE_BATCH_SIZE=15 INGESTION_AUTOMATION_ALLOWED=1 INGESTION_MANUAL_RUN_ALLOWED=1 INGESTION_MIN_RUN_INTERVAL_MINUTES=15 INGESTION_MIN_REQUEST_DELAY_SECONDS=2 INGESTION_NETWORK_ERROR_RETRIES=1 INGESTION_RETRY_BACKOFF_SECONDS=1.0 +INGESTION_RATE_LIMIT_RETRIES=3 +INGESTION_RATE_LIMIT_BACKOFF_SECONDS=30.0 INGESTION_MAX_PAGES_PER_SCHOLAR=30 INGESTION_PAGE_SIZE=100 INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD=1 @@ -125,6 +128,14 @@ UNPAYWALL_RETRY_COOLDOWN_SECONDS=1800 UNPAYWALL_PDF_DISCOVERY_ENABLED=1 UNPAYWALL_PDF_DISCOVERY_MAX_CANDIDATES=5 UNPAYWALL_PDF_DISCOVERY_MAX_HTML_BYTES=500000 +ARXIV_ENABLED=1 +ARXIV_TIMEOUT_SECONDS=3.0 +ARXIV_MIN_INTERVAL_SECONDS=4.0 +ARXIV_RATE_LIMIT_COOLDOWN_SECONDS=60.0 +ARXIV_DEFAULT_MAX_RESULTS=3 +ARXIV_CACHE_TTL_SECONDS=900 +ARXIV_CACHE_MAX_ENTRIES=512 +ARXIV_MAILTO= PDF_AUTO_RETRY_INTERVAL_SECONDS=86400 PDF_AUTO_RETRY_FIRST_INTERVAL_SECONDS=3600 PDF_AUTO_RETRY_MAX_ATTEMPTS=3 @@ -133,6 +144,9 @@ CROSSREF_MAX_ROWS=10 CROSSREF_TIMEOUT_SECONDS=8.0 CROSSREF_MIN_INTERVAL_SECONDS=0.6 CROSSREF_MAX_LOOKUPS_PER_REQUEST=8 +OPENALEX_API_KEY= +CROSSREF_API_TOKEN= +CROSSREF_API_MAILTO= # ------------------------------ # Startup Bootstrap + DB Wait diff --git a/alembic/versions/20260226_0023_add_arxiv_runtime_state.py b/alembic/versions/20260226_0023_add_arxiv_runtime_state.py new file mode 100644 index 0000000..b75cd0e --- /dev/null +++ b/alembic/versions/20260226_0023_add_arxiv_runtime_state.py @@ -0,0 +1,55 @@ +"""Add shared arXiv runtime state table. + +Revision ID: 20260226_0023 +Revises: 20260225_0022 +Create Date: 2026-02-26 12:40:00.000000 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision: str = "20260226_0023" +down_revision: str | Sequence[str] | None = "20260225_0022" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + table_names = set(inspector.get_table_names()) + + if "arxiv_runtime_state" in table_names: + return + + op.create_table( + "arxiv_runtime_state", + sa.Column("state_key", sa.String(length=64), nullable=False), + sa.Column("next_allowed_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("cooldown_until", sa.DateTime(timezone=True), nullable=True), + sa.Column( + "created_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + ), + sa.PrimaryKeyConstraint("state_key", name=op.f("pk_arxiv_runtime_state")), + ) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + table_names = set(inspector.get_table_names()) + if "arxiv_runtime_state" in table_names: + op.drop_table("arxiv_runtime_state") diff --git a/alembic/versions/20260226_0024_add_arxiv_query_cache_entries.py b/alembic/versions/20260226_0024_add_arxiv_query_cache_entries.py new file mode 100644 index 0000000..fe2c0bc --- /dev/null +++ b/alembic/versions/20260226_0024_add_arxiv_query_cache_entries.py @@ -0,0 +1,70 @@ +"""Add arXiv query cache table. + +Revision ID: 20260226_0024 +Revises: 20260226_0023 +Create Date: 2026-02-26 14:20:00.000000 +""" + +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +# revision identifiers, used by Alembic. +revision: str = "20260226_0024" +down_revision: str | Sequence[str] | None = "20260226_0023" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + table_names = set(inspector.get_table_names()) + if "arxiv_query_cache_entries" in table_names: + return + + op.create_table( + "arxiv_query_cache_entries", + sa.Column("query_fingerprint", sa.String(length=64), nullable=False), + sa.Column("payload", postgresql.JSONB(astext_type=sa.Text()), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False), + sa.Column( + "cached_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + ), + sa.Column( + "updated_at", + sa.DateTime(timezone=True), + nullable=False, + server_default=sa.func.now(), + ), + sa.PrimaryKeyConstraint( + "query_fingerprint", + name=op.f("pk_arxiv_query_cache_entries"), + ), + ) + op.create_index( + "ix_arxiv_query_cache_expires_at", + "arxiv_query_cache_entries", + ["expires_at"], + unique=False, + ) + op.create_index( + "ix_arxiv_query_cache_cached_at", + "arxiv_query_cache_entries", + ["cached_at"], + unique=False, + ) + + +def downgrade() -> None: + bind = op.get_bind() + inspector = sa.inspect(bind) + table_names = set(inspector.get_table_names()) + if "arxiv_query_cache_entries" in table_names: + op.drop_table("arxiv_query_cache_entries") diff --git a/app/api/routers/admin_dbops.py b/app/api/routers/admin_dbops.py index b3c1fdf..be23092 100644 --- a/app/api/routers/admin_dbops.py +++ b/app/api/routers/admin_dbops.py @@ -14,6 +14,8 @@ from app.api.schemas import ( AdminPdfQueueRequeueEnvelope, AdminPdfQueueEnvelope, AdminDbRepairJobsEnvelope, + AdminRepairPublicationNearDuplicatesEnvelope, + AdminRepairPublicationNearDuplicatesRequest, AdminRepairPublicationLinksEnvelope, AdminRepairPublicationLinksRequest, ) @@ -22,6 +24,7 @@ from app.db.session import get_db_session from app.services.domains.dbops import ( collect_integrity_report, list_repair_jobs, + run_publication_near_duplicate_repair, run_publication_link_repair, ) from app.services.domains.publications import application as publication_service @@ -48,7 +51,7 @@ def _serialize_repair_job(job: DataRepairJob) -> dict[str, object]: } -def _requested_by_value(*, payload: AdminRepairPublicationLinksRequest, admin_user: User) -> str: +def _requested_by_value(*, payload, admin_user: User) -> str: from_payload = (payload.requested_by or "").strip() return from_payload or admin_user.email @@ -349,6 +352,47 @@ async def trigger_publication_link_repair( return success_payload(request, data=result) +@router.post( + "/repairs/publication-near-duplicates", + response_model=AdminRepairPublicationNearDuplicatesEnvelope, +) +async def trigger_publication_near_duplicate_repair( + payload: AdminRepairPublicationNearDuplicatesRequest, + request: Request, + db_session: AsyncSession = Depends(get_db_session), + admin_user: User = Depends(get_api_admin_user), +): + try: + result = await run_publication_near_duplicate_repair( + db_session, + dry_run=bool(payload.dry_run), + similarity_threshold=float(payload.similarity_threshold), + min_shared_tokens=int(payload.min_shared_tokens), + max_year_delta=int(payload.max_year_delta), + max_clusters=int(payload.max_clusters), + selected_cluster_keys=list(payload.selected_cluster_keys), + requested_by=_requested_by_value(payload=payload, admin_user=admin_user), + ) + except ValueError as exc: + raise ApiException( + status_code=400, + code="invalid_near_duplicate_repair_request", + message=str(exc), + ) from exc + logger.info( + "api.admin.db.publication_near_duplicate_repair_triggered", + extra={ + "event": "api.admin.db.publication_near_duplicate_repair_triggered", + "admin_user_id": int(admin_user.id), + "dry_run": bool(payload.dry_run), + "selected_cluster_count": len(payload.selected_cluster_keys), + "job_id": int(result["job_id"]), + "status": result["status"], + }, + ) + return success_payload(request, data=result) + + DROP_PUBLICATIONS_CONFIRMATION = "DROP ALL PUBLICATIONS" diff --git a/app/api/routers/publications.py b/app/api/routers/publications.py index b4a616e..7de2f12 100644 --- a/app/api/routers/publications.py +++ b/app/api/routers/publications.py @@ -94,6 +94,7 @@ async def _publication_counts( user_id: int, selected_scholar_id: int | None, favorite_only: bool, + search: str | None, snapshot_before: datetime | None, ) -> tuple[int, int, int, int]: unread_count = await publication_service.count_unread_for_user( @@ -122,6 +123,7 @@ async def _publication_counts( mode=publication_service.MODE_ALL, scholar_profile_id=selected_scholar_id, favorite_only=favorite_only, + search=search, snapshot_before=snapshot_before, ) return unread_count, favorites_count, latest_count, total_count @@ -372,7 +374,7 @@ async def list_publications( favorite_only: bool = Query(default=False), scholar_profile_id: int | None = Query(default=None, ge=1), search: str | None = Query(default=None, min_length=1, max_length=200), - sort_by: Literal["first_seen", "title", "year", "citations", "scholar"] = Query(default="first_seen"), + sort_by: Literal["first_seen", "title", "year", "citations", "scholar", "pdf_status"] = Query(default="first_seen"), sort_dir: Literal["asc", "desc"] = Query(default="desc"), page: int = Query(default=1, ge=1), page_size: int = Query(default=100, ge=1, le=500), @@ -408,6 +410,7 @@ async def list_publications( user_id=current_user.id, selected_scholar_id=selected_scholar_id, favorite_only=favorite_only, + search=normalized_search, snapshot_before=snapshot_before, ) data = _publications_list_data( diff --git a/app/api/routers/scholars.py b/app/api/routers/scholars.py index f4be187..2dc0e2f 100644 --- a/app/api/routers/scholars.py +++ b/app/api/routers/scholars.py @@ -23,7 +23,9 @@ from app.api.schemas import ( ) from app.db.models import User from app.db.session import get_db_session +from app.services.domains.ingestion import queue as ingestion_queue_service from app.services.domains.portability import application as import_export_service +from app.services.domains.scholar import rate_limit as scholar_rate_limit from app.services.domains.scholars import application as scholar_service from app.services.domains.scholar.source import ScholarSource from app.settings import settings @@ -31,6 +33,73 @@ from app.settings import settings logger = logging.getLogger(__name__) router = APIRouter(prefix="/scholars", tags=["api-scholars"]) +CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS = 5.0 +INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS = 0 +INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON = "scholar_added_initial_scrape" + + +def _needs_metadata_hydration(profile) -> bool: + if not profile.profile_image_url: + return True + return not (profile.display_name or "").strip() + + +def _is_create_hydration_rate_limited() -> tuple[bool, float]: + remaining_seconds = scholar_rate_limit.remaining_scholar_slot_seconds( + min_interval_seconds=float(settings.ingestion_min_request_delay_seconds), + ) + return remaining_seconds > 0, remaining_seconds + + +def _auto_enqueue_new_scholar_enabled() -> bool: + if not settings.scheduler_enabled: + return False + if not settings.ingestion_automation_allowed: + return False + return bool(settings.ingestion_continuation_queue_enabled) + + +async def _enqueue_initial_scrape_job_for_scholar( + db_session: AsyncSession, + *, + profile, + user_id: int, +) -> bool: + if not _auto_enqueue_new_scholar_enabled(): + return False + try: + await ingestion_queue_service.upsert_job( + db_session, + user_id=user_id, + scholar_profile_id=int(profile.id), + resume_cstart=0, + reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, + run_id=None, + delay_seconds=INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS, + ) + await db_session.commit() + except Exception: + await db_session.rollback() + logger.warning( + "api.scholars.initial_scrape_enqueue_failed", + extra={ + "event": "api.scholars.initial_scrape_enqueue_failed", + "user_id": user_id, + "scholar_profile_id": profile.id, + }, + ) + return False + + logger.info( + "api.scholars.initial_scrape_enqueued", + extra={ + "event": "api.scholars.initial_scrape_enqueued", + "user_id": user_id, + "scholar_profile_id": profile.id, + "reason": INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, + }, + ) + return True def _uploaded_image_media_path(scholar_profile_id: int) -> str: @@ -69,16 +138,42 @@ async def _hydrate_scholar_metadata_if_needed( source: ScholarSource, user_id: int, ): + if not _needs_metadata_hydration(profile): + return profile + + should_skip, remaining_seconds = _is_create_hydration_rate_limited() + if should_skip: + logger.info( + "api.scholars.create_metadata_hydration_skipped", + extra={ + "event": "api.scholars.create_metadata_hydration_skipped", + "reason": "scholar_request_throttle_active", + "user_id": user_id, + "scholar_profile_id": profile.id, + "retry_after_seconds": round(remaining_seconds, 3), + }, + ) + return profile + 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, - ) + return await asyncio.wait_for( + scholar_service.hydrate_profile_metadata( + db_session, + profile=profile, + source=source, + ), + timeout=CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS, + ) + except TimeoutError: + logger.info( + "api.scholars.create_metadata_hydration_skipped", + extra={ + "event": "api.scholars.create_metadata_hydration_skipped", + "reason": "create_timeout", + "user_id": user_id, + "scholar_profile_id": profile.id, + }, + ) except Exception: logger.warning( "api.scholars.create_metadata_hydration_failed", @@ -280,12 +375,18 @@ async def create_scholar( "scholar_profile_id": created.id, }, ) - created = await _hydrate_scholar_metadata_if_needed( + did_queue_initial_scrape = await _enqueue_initial_scrape_job_for_scholar( db_session, profile=created, - source=source, user_id=current_user.id, ) + if not did_queue_initial_scrape: + created = await _hydrate_scholar_metadata_if_needed( + db_session, + profile=created, + source=source, + user_id=current_user.id, + ) return success_payload( request, @@ -554,4 +655,3 @@ async def clear_scholar_image_customization( request, data=_serialize_scholar(updated), ) - diff --git a/app/api/schemas.py b/app/api/schemas.py index 07ccedf..e3b9ac7 100644 --- a/app/api/schemas.py +++ b/app/api/schemas.py @@ -703,6 +703,70 @@ class AdminRepairPublicationLinksEnvelope(BaseModel): model_config = ConfigDict(extra="forbid") +class AdminNearDuplicateClusterMemberData(BaseModel): + publication_id: int + title: str + year: int | None + citation_count: int + + model_config = ConfigDict(extra="forbid") + + +class AdminNearDuplicateClusterData(BaseModel): + cluster_key: str + winner_publication_id: int + member_count: int + similarity_score: float = Field(ge=0.0, le=1.0) + members: list[AdminNearDuplicateClusterMemberData] = Field(default_factory=list) + + model_config = ConfigDict(extra="forbid") + + +class AdminRepairPublicationNearDuplicatesRequest(BaseModel): + dry_run: bool = True + similarity_threshold: float = Field(default=0.78, ge=0.5, le=1.0) + min_shared_tokens: int = Field(default=3, ge=1, le=8) + max_year_delta: int = Field(default=1, ge=0, le=5) + max_clusters: int = Field(default=25, ge=1, le=200) + selected_cluster_keys: list[str] = Field(default_factory=list, max_length=200) + requested_by: str | None = None + confirmation_text: str | None = None + + model_config = ConfigDict(extra="forbid") + + @model_validator(mode="after") + def validate_apply_mode(self) -> "AdminRepairPublicationNearDuplicatesRequest": + if self.dry_run: + return self + if not self.selected_cluster_keys: + raise ValueError("selected_cluster_keys is required when dry_run=false.") + expected = "MERGE SELECTED DUPLICATES" + provided = (self.confirmation_text or "").strip() + if provided != expected: + raise ValueError( + "confirmation_text must equal 'MERGE SELECTED DUPLICATES' " + "when applying near-duplicate merges." + ) + return self + + +class AdminRepairPublicationNearDuplicatesResultData(BaseModel): + job_id: int + status: str + scope: dict[str, Any] = Field(default_factory=dict) + summary: dict[str, Any] = Field(default_factory=dict) + clusters: list[AdminNearDuplicateClusterData] = Field(default_factory=list) + + model_config = ConfigDict(extra="forbid") + + +class AdminRepairPublicationNearDuplicatesEnvelope(BaseModel): + data: AdminRepairPublicationNearDuplicatesResultData + meta: ApiMeta + + model_config = ConfigDict(extra="forbid") + + class SettingsPolicyData(BaseModel): min_run_interval_minutes: int min_request_delay_seconds: int diff --git a/app/db/models.py b/app/db/models.py index c687547..6a51811 100644 --- a/app/db/models.py +++ b/app/db/models.py @@ -540,6 +540,44 @@ class AuthorSearchRuntimeState(Base): ) +class ArxivRuntimeState(Base): + __tablename__ = "arxiv_runtime_state" + + state_key: Mapped[str] = mapped_column(String(64), primary_key=True) + next_allowed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + cooldown_until: Mapped[datetime | None] = mapped_column(DateTime(timezone=True)) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + +class ArxivQueryCacheEntry(Base): + __tablename__ = "arxiv_query_cache_entries" + __table_args__ = ( + Index("ix_arxiv_query_cache_expires_at", "expires_at"), + Index("ix_arxiv_query_cache_cached_at", "cached_at"), + ) + + query_fingerprint: Mapped[str] = mapped_column(String(64), primary_key=True) + payload: Mapped[dict] = mapped_column( + JSONB, + nullable=False, + ) + expires_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), + nullable=False, + ) + cached_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, server_default=func.now() + ) + + class AuthorSearchCacheEntry(Base): __tablename__ = "author_search_cache_entries" __table_args__ = ( diff --git a/app/services/domains/arxiv/application.py b/app/services/domains/arxiv/application.py index 57b1473..8b99096 100644 --- a/app/services/domains/arxiv/application.py +++ b/app/services/domains/arxiv/application.py @@ -1,110 +1,29 @@ from __future__ import annotations -import asyncio -import logging from typing import TYPE_CHECKING -import xml.etree.ElementTree as ET - -import httpx - -from app.services.domains.publication_identifiers.normalize import normalize_arxiv_id -from app.settings import settings +from app.services.domains.arxiv.gateway import ( + build_arxiv_query, + get_arxiv_gateway, +) +from app.services.domains.arxiv.errors import ArxivRateLimitError if TYPE_CHECKING: from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem -logger = logging.getLogger(__name__) - -# arXiv API terms: max 1 request per 3 seconds, single connection at a time. -_ARXIV_LOCK = asyncio.Lock() -_ARXIV_MIN_INTERVAL_SECONDS = 4.0 -# Global cooldown: when arXiv returns 429, all batches back off for this long. -_ARXIV_RATE_LIMIT_COOLDOWN_SECONDS = 60.0 -_arxiv_rate_limited_until: float = 0.0 # asyncio monotonic time - - -class ArxivRateLimitError(Exception): - """arXiv returned 429 — stop the batch to avoid hammering.""" - pass - def _build_arxiv_query(title: str, author_surname: str | None) -> str | None: - parts = [] - if title: - clean_title = title.replace('"', '').replace("'", "") - parts.append(f'ti:"{clean_title}"') - if author_surname: - parts.append(f'au:"{author_surname}"') - if not parts: - return None - return " AND ".join(parts) + return build_arxiv_query(title, author_surname) async def discover_arxiv_id_for_publication( *, item: PublicationListItem | UnreadPublicationItem, request_email: str | None = None, - timeout_seconds: float = 3.0, + timeout_seconds: float | None = None, ) -> str | None: - title = (item.title or "").strip() - if not title: - return None - - author_surname = None - if item.scholar_label: - tokens = [t for t in item.scholar_label.strip().split() if t] - if tokens: - author_surname = tokens[-1].lower() - - query = _build_arxiv_query(title, author_surname) - if not query: - return None - - url = "https://export.arxiv.org/api/query" - params = {"search_query": query, "start": 0, "max_results": 3} - headers = { - "User-Agent": ( - f"scholar-scraper/1.0 " - f"(mailto:{request_email or settings.crossref_api_mailto or 'unknown@example.com'})" - ) - } - - try: - async with _ARXIV_LOCK: - global _arxiv_rate_limited_until - now = asyncio.get_running_loop().time() - if now < _arxiv_rate_limited_until: - remaining = _arxiv_rate_limited_until - now - raise ArxivRateLimitError(f"arXiv global cooldown active ({remaining:.0f}s remaining)") - - async with httpx.AsyncClient(timeout=timeout_seconds, follow_redirects=True, headers=headers) as client: - response = await client.get(url, params=params) - - if response.status_code == 429: - _arxiv_rate_limited_until = asyncio.get_running_loop().time() + _ARXIV_RATE_LIMIT_COOLDOWN_SECONDS - raise ArxivRateLimitError("arXiv rate limit hit (429) — stopping batch") - - await asyncio.sleep(_ARXIV_MIN_INTERVAL_SECONDS) - - response.raise_for_status() - - root = ET.fromstring(response.text) - namespace = {"atom": "http://www.w3.org/2005/Atom"} - - for entry in root.findall("atom:entry", namespace): - id_elem = entry.find("atom:id", namespace) - if id_elem is not None and id_elem.text: - candidate = str(id_elem.text) - if "/abs/" in candidate: - candidate = candidate.split("/abs/")[-1] - normalized = normalize_arxiv_id(candidate) - if normalized: - logger.debug("arxiv.id_discovered", extra={"event": "arxiv.id_discovered", "arxiv_id": normalized}) - return normalized - - except ArxivRateLimitError: - raise # propagate so the batch loop can stop - except Exception as exc: - logger.debug(f"Failed to query arXiv API: {exc}") - - return None + gateway = get_arxiv_gateway() + return await gateway.discover_arxiv_id_for_publication( + item=item, + request_email=request_email, + timeout_seconds=timeout_seconds, + ) diff --git a/app/services/domains/arxiv/cache.py b/app/services/domains/arxiv/cache.py new file mode 100644 index 0000000..de16416 --- /dev/null +++ b/app/services/domains/arxiv/cache.py @@ -0,0 +1,332 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable, Mapping +from dataclasses import asdict +from datetime import datetime, timedelta, timezone +import hashlib +import json +from typing import Any + +from sqlalchemy import delete, func, select + +from app.db.models import ArxivQueryCacheEntry +from app.db.session import get_session_factory +from app.services.domains.arxiv.constants import ARXIV_CACHE_FINGERPRINT_VERSION +from app.services.domains.arxiv.types import ArxivEntry, ArxivFeed, ArxivOpenSearchMeta + +_INFLIGHT_LOCK = asyncio.Lock() +_INFLIGHT_FEEDS: dict[str, asyncio.Future[ArxivFeed]] = {} + + +def build_query_fingerprint(*, params: Mapping[str, object]) -> str: + canonical = _canonical_cache_payload(params=params) + encoded = json.dumps(canonical, sort_keys=True, separators=(",", ":"), ensure_ascii=True) + payload = f"{ARXIV_CACHE_FINGERPRINT_VERSION}:{encoded}" + return hashlib.sha256(payload.encode("utf-8")).hexdigest() + + +async def get_cached_feed( + *, + query_fingerprint: str, + now_utc: datetime | None = None, +) -> ArxivFeed | None: + timestamp = _as_utc(now_utc) + session_factory = get_session_factory() + async with session_factory() as db_session: + async with db_session.begin(): + result = await db_session.execute( + select(ArxivQueryCacheEntry).where( + ArxivQueryCacheEntry.query_fingerprint == query_fingerprint + ) + ) + entry = result.scalar_one_or_none() + return await _validate_cached_entry(db_session, entry=entry, now_utc=timestamp) + + +async def set_cached_feed( + *, + query_fingerprint: str, + feed: ArxivFeed, + ttl_seconds: float, + max_entries: int, + now_utc: datetime | None = None, +) -> None: + timestamp = _as_utc(now_utc) + session_factory = get_session_factory() + async with session_factory() as db_session: + async with db_session.begin(): + await _write_cached_entry( + db_session, + query_fingerprint=query_fingerprint, + feed=feed, + ttl_seconds=ttl_seconds, + max_entries=max_entries, + now_utc=timestamp, + ) + + +async def run_with_inflight_dedupe( + *, + query_fingerprint: str, + fetch_feed: Callable[[], Awaitable[ArxivFeed]], +) -> ArxivFeed: + future, is_owner = await _reserve_inflight_future(query_fingerprint=query_fingerprint) + if not is_owner: + return await asyncio.shield(future) + try: + result = await fetch_feed() + except Exception as exc: + _complete_future(future, error=exc) + raise + finally: + await _release_inflight_future(query_fingerprint=query_fingerprint, future=future) + _complete_future(future, result=result) + return result + + +def _canonical_cache_payload(*, params: Mapping[str, object]) -> dict[str, object]: + payload: dict[str, object] = {} + for key in sorted(params.keys()): + payload[str(key)] = _normalize_param_value(str(key), params[key]) + return payload + + +def _normalize_param_value(key: str, value: object) -> object: + if key == "search_query": + return _normalize_search_query(str(value or "")) + if key == "id_list": + return _normalize_id_list(str(value or "")) + if isinstance(value, str): + return " ".join(value.strip().split()) + if isinstance(value, (int, float, bool)) or value is None: + return value + return str(value).strip() + + +def _normalize_search_query(value: str) -> str: + return " ".join(value.strip().lower().split()) + + +def _normalize_id_list(value: str) -> str: + normalized = [item.strip().lower() for item in value.split(",") if item.strip()] + return ",".join(sorted(normalized)) + + +async def _validate_cached_entry( + db_session, + *, + entry: ArxivQueryCacheEntry | None, + now_utc: datetime, +) -> ArxivFeed | None: + if entry is None: + return None + if _as_utc(entry.expires_at) <= now_utc: + await db_session.delete(entry) + return None + parsed = _deserialize_feed(entry.payload) + if parsed is None: + await db_session.delete(entry) + return None + return parsed + + +async def _write_cached_entry( + db_session, + *, + query_fingerprint: str, + feed: ArxivFeed, + ttl_seconds: float, + max_entries: int, + now_utc: datetime, +) -> None: + ttl = max(float(ttl_seconds), 0.0) + result = await db_session.execute( + select(ArxivQueryCacheEntry).where( + ArxivQueryCacheEntry.query_fingerprint == query_fingerprint + ) + ) + existing = result.scalar_one_or_none() + if ttl <= 0.0: + if existing is not None: + await db_session.delete(existing) + return + expires_at = now_utc + timedelta(seconds=ttl) + payload = _serialize_feed(feed) + if existing is None: + db_session.add( + ArxivQueryCacheEntry( + query_fingerprint=query_fingerprint, + payload=payload, + expires_at=expires_at, + cached_at=now_utc, + updated_at=now_utc, + ) + ) + else: + existing.payload = payload + existing.expires_at = expires_at + existing.cached_at = now_utc + existing.updated_at = now_utc + await _prune_cache_entries(db_session, now_utc=now_utc, max_entries=max_entries) + + +async def _prune_cache_entries( + db_session, + *, + now_utc: datetime, + max_entries: int, +) -> None: + await db_session.execute( + delete(ArxivQueryCacheEntry).where(ArxivQueryCacheEntry.expires_at <= now_utc) + ) + bounded_max_entries = int(max_entries) + if bounded_max_entries <= 0: + return + count_result = await db_session.execute( + select(func.count()).select_from(ArxivQueryCacheEntry) + ) + entry_count = int(count_result.scalar_one() or 0) + overflow = max(0, entry_count - bounded_max_entries) + if overflow <= 0: + return + stale_result = await db_session.execute( + select(ArxivQueryCacheEntry.query_fingerprint) + .order_by(ArxivQueryCacheEntry.cached_at.asc()) + .limit(overflow) + ) + stale_keys = [str(row[0]) for row in stale_result.all()] + if stale_keys: + await db_session.execute( + delete(ArxivQueryCacheEntry).where( + ArxivQueryCacheEntry.query_fingerprint.in_(stale_keys) + ) + ) + + +def _serialize_feed(feed: ArxivFeed) -> dict[str, Any]: + return asdict(feed) + + +def _deserialize_feed(payload: object) -> ArxivFeed | None: + if not isinstance(payload, dict): + return None + entries_payload = payload.get("entries") + opensearch_payload = payload.get("opensearch") + if not isinstance(entries_payload, list): + return None + entries: list[ArxivEntry] = [] + for value in entries_payload: + entry = _deserialize_entry(value) + if entry is None: + return None + entries.append(entry) + opensearch = _deserialize_opensearch(opensearch_payload) + if opensearch is None: + return None + return ArxivFeed(entries=entries, opensearch=opensearch) + + +def _deserialize_entry(value: object) -> ArxivEntry | None: + if not isinstance(value, dict): + return None + try: + return ArxivEntry( + entry_id_url=str(value["entry_id_url"]), + arxiv_id=_as_optional_string(value.get("arxiv_id")), + title=str(value["title"]), + summary=str(value["summary"]), + published=_as_optional_string(value.get("published")), + updated=_as_optional_string(value.get("updated")), + authors=_as_string_list(value.get("authors")), + links=_as_string_list(value.get("links")), + categories=_as_string_list(value.get("categories")), + primary_category=_as_optional_string(value.get("primary_category")), + ) + except KeyError: + return None + + +def _deserialize_opensearch(value: object) -> ArxivOpenSearchMeta | None: + if not isinstance(value, dict): + return None + try: + return ArxivOpenSearchMeta( + total_results=int(value.get("total_results", 0)), + start_index=int(value.get("start_index", 0)), + items_per_page=int(value.get("items_per_page", 0)), + ) + except (TypeError, ValueError): + return None + + +def _as_optional_string(value: object) -> str | None: + if value is None: + return None + normalized = str(value).strip() + return normalized or None + + +def _as_string_list(value: object) -> list[str]: + if not isinstance(value, list): + return [] + return [str(item) for item in value] + + +def _as_utc(value: datetime | None) -> datetime: + if value is None: + return datetime.now(timezone.utc) + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + +async def _reserve_inflight_future( + *, + query_fingerprint: str, +) -> tuple[asyncio.Future[ArxivFeed], bool]: + async with _INFLIGHT_LOCK: + existing = _INFLIGHT_FEEDS.get(query_fingerprint) + if existing is not None: + return existing, False + loop = asyncio.get_running_loop() + created = loop.create_future() + created.add_done_callback(_consume_unretrieved_future_exception) + _INFLIGHT_FEEDS[query_fingerprint] = created + return created, True + + +async def _release_inflight_future( + *, + query_fingerprint: str, + future: asyncio.Future[ArxivFeed], +) -> None: + async with _INFLIGHT_LOCK: + current = _INFLIGHT_FEEDS.get(query_fingerprint) + if current is future: + _INFLIGHT_FEEDS.pop(query_fingerprint, None) + + +def _complete_future( + future: asyncio.Future[ArxivFeed], + *, + result: ArxivFeed | None = None, + error: Exception | None = None, +) -> None: + if future.done(): + return + if error is not None: + future.set_exception(error) + return + if result is None: + raise RuntimeError("in-flight future completion requires result or error") + future.set_result(result) + + +def _consume_unretrieved_future_exception(future: asyncio.Future[ArxivFeed]) -> None: + if future.cancelled(): + return + try: + _ = future.exception() + except Exception: + return diff --git a/app/services/domains/arxiv/client.py b/app/services/domains/arxiv/client.py new file mode 100644 index 0000000..278ca0f --- /dev/null +++ b/app/services/domains/arxiv/client.py @@ -0,0 +1,311 @@ +from __future__ import annotations + +from collections.abc import Awaitable, Callable +import logging + +import httpx + +from app.services.domains.arxiv.cache import ( + build_query_fingerprint, + get_cached_feed, + run_with_inflight_dedupe, + set_cached_feed, +) +from app.services.domains.arxiv.constants import ( + ARXIV_SOURCE_PATH_LOOKUP_IDS, + ARXIV_SOURCE_PATH_SEARCH, + ARXIV_SOURCE_PATH_UNKNOWN, +) +from app.services.domains.arxiv.errors import ArxivClientValidationError, ArxivRateLimitError +from app.services.domains.arxiv.parser import parse_arxiv_feed +from app.services.domains.arxiv.rate_limit import get_arxiv_cooldown_status, run_with_global_arxiv_limit +from app.services.domains.arxiv.types import ArxivFeed +from app.settings import settings + +_ARXIV_API_URL = "https://export.arxiv.org/api/query" +_ARXIV_QUERY_START = 0 +_ARXIV_MAX_RESULTS_LIMIT = 30_000 +_ARXIV_SORT_BY_ALLOWED = {"relevance", "lastUpdatedDate", "submittedDate"} +_ARXIV_SORT_ORDER_ALLOWED = {"ascending", "descending"} +_FALLBACK_CONTACT_EMAIL = "unknown@example.com" + +ArxivRequestFn = Callable[..., Awaitable[httpx.Response]] +logger = logging.getLogger(__name__) + + +class ArxivClient: + def __init__( + self, + *, + request_fn: ArxivRequestFn | None = None, + cache_enabled: bool | None = None, + ) -> None: + self._request_fn = request_fn or _request_arxiv_feed + self._cache_enabled = _resolve_cache_enabled( + cache_enabled=cache_enabled, + request_fn=request_fn, + ) + self._cache_ttl_seconds = _cache_ttl_seconds() + self._cache_max_entries = _cache_max_entries() + + async def search( + self, + *, + query: str, + start: int = _ARXIV_QUERY_START, + max_results: int | None = None, + sort_by: str | None = None, + sort_order: str | None = None, + request_email: str | None = None, + timeout_seconds: float | None = None, + ) -> ArxivFeed: + params = _search_params( + query=query, + start=start, + max_results=max_results, + sort_by=sort_by, + sort_order=sort_order, + ) + return await self._fetch_feed( + params=params, + request_email=request_email, + timeout_seconds=timeout_seconds, + source_path=ARXIV_SOURCE_PATH_SEARCH, + ) + + async def lookup_ids( + self, + *, + id_list: list[str], + start: int = _ARXIV_QUERY_START, + max_results: int | None = None, + request_email: str | None = None, + timeout_seconds: float | None = None, + ) -> ArxivFeed: + params = _lookup_params(id_list=id_list, start=start, max_results=max_results) + return await self._fetch_feed( + params=params, + request_email=request_email, + timeout_seconds=timeout_seconds, + source_path=ARXIV_SOURCE_PATH_LOOKUP_IDS, + ) + + async def _fetch_feed( + self, + *, + params: dict[str, object], + request_email: str | None, + timeout_seconds: float | None, + source_path: str, + ) -> ArxivFeed: + query_fingerprint = build_query_fingerprint(params=params) + if self._cache_enabled: + cached = await get_cached_feed(query_fingerprint=query_fingerprint) + if cached is not None: + _log_cache_event( + event_name="arxiv.cache_hit", + query_fingerprint=query_fingerprint, + source_path=source_path, + ) + return cached + _log_cache_event( + event_name="arxiv.cache_miss", + query_fingerprint=query_fingerprint, + source_path=source_path, + ) + return await run_with_inflight_dedupe( + query_fingerprint=query_fingerprint, + fetch_feed=lambda: self._fetch_live_feed( + params=params, + request_email=request_email, + timeout_seconds=timeout_seconds, + query_fingerprint=query_fingerprint, + ), + ) + + async def _fetch_live_feed( + self, + *, + params: dict[str, object], + request_email: str | None, + timeout_seconds: float | None, + query_fingerprint: str, + ) -> ArxivFeed: + response = await self._request_fn( + params=params, + request_email=request_email, + timeout_seconds=timeout_seconds, + ) + response.raise_for_status() + feed = parse_arxiv_feed(response.text) + if self._cache_enabled: + await set_cached_feed( + query_fingerprint=query_fingerprint, + feed=feed, + ttl_seconds=self._cache_ttl_seconds, + max_entries=self._cache_max_entries, + ) + return feed + + +def _search_params( + *, + query: str, + start: int, + max_results: int | None, + sort_by: str | None, + sort_order: str | None, +) -> dict[str, object]: + clean_query = query.strip() + if not clean_query: + raise ArxivClientValidationError("search query must not be empty") + params: dict[str, object] = { + "search_query": clean_query, + "start": _validate_start(start), + "max_results": _validate_max_results(max_results), + } + if sort_by is not None: + params["sortBy"] = _validate_sort_by(sort_by) + if sort_order is not None: + params["sortOrder"] = _validate_sort_order(sort_order) + return params + + +def _lookup_params(*, id_list: list[str], start: int, max_results: int | None) -> dict[str, object]: + normalized_ids = [value.strip() for value in id_list if value and value.strip()] + if not normalized_ids: + raise ArxivClientValidationError("id_list must include at least one id") + return { + "id_list": ",".join(normalized_ids), + "start": _validate_start(start), + "max_results": _validate_max_results(max_results), + } + + +def _validate_start(value: int) -> int: + start = int(value) + if start < 0: + raise ArxivClientValidationError("start must be >= 0") + return start + + +def _validate_max_results(value: int | None) -> int: + if value is None: + default_value = int(settings.arxiv_default_max_results) + return max(default_value, 1) + parsed = int(value) + if parsed < 1: + raise ArxivClientValidationError("max_results must be >= 1") + if parsed > _ARXIV_MAX_RESULTS_LIMIT: + raise ArxivClientValidationError(f"max_results must be <= {_ARXIV_MAX_RESULTS_LIMIT}") + return parsed + + +def _validate_sort_by(value: str) -> str: + if value not in _ARXIV_SORT_BY_ALLOWED: + raise ArxivClientValidationError(f"sort_by must be one of: {sorted(_ARXIV_SORT_BY_ALLOWED)!r}") + return value + + +def _validate_sort_order(value: str) -> str: + if value not in _ARXIV_SORT_ORDER_ALLOWED: + raise ArxivClientValidationError(f"sort_order must be one of: {sorted(_ARXIV_SORT_ORDER_ALLOWED)!r}") + return value + + +async def _request_arxiv_feed( + *, + params: dict[str, object], + request_email: str | None, + timeout_seconds: float | None, +) -> httpx.Response: + source_path = _source_path_from_params(params) + cooldown_status = await get_arxiv_cooldown_status() + if cooldown_status.is_active: + _log_request_skipped_for_cooldown( + source_path=source_path, + cooldown_remaining_seconds=cooldown_status.remaining_seconds, + ) + raise ArxivRateLimitError( + f"arXiv global cooldown active ({cooldown_status.remaining_seconds:.0f}s remaining)" + ) + + async def _fetch() -> httpx.Response: + timeout_value = _timeout_seconds(timeout_seconds) + headers = {"User-Agent": f"scholar-scraper/1.0 (mailto:{_contact_email(request_email)})"} + async with httpx.AsyncClient(timeout=timeout_value, follow_redirects=True, headers=headers) as client: + return await client.get(_ARXIV_API_URL, params=params) + + return await run_with_global_arxiv_limit( + fetch=_fetch, + source_path=source_path, + ) + + +def _timeout_seconds(timeout_seconds: float | None) -> float: + if timeout_seconds is not None: + return max(float(timeout_seconds), 0.5) + return max(float(settings.arxiv_timeout_seconds), 0.5) + + +def _contact_email(request_email: str | None) -> str: + return request_email or settings.arxiv_mailto or settings.crossref_api_mailto or _FALLBACK_CONTACT_EMAIL + + +def _resolve_cache_enabled( + *, + cache_enabled: bool | None, + request_fn: ArxivRequestFn | None, +) -> bool: + if cache_enabled is not None: + return bool(cache_enabled) + if request_fn is not None: + return False + return _cache_ttl_seconds() > 0.0 + + +def _cache_ttl_seconds() -> float: + return max(float(settings.arxiv_cache_ttl_seconds), 0.0) + + +def _cache_max_entries() -> int: + return max(int(settings.arxiv_cache_max_entries), 0) + + +def _source_path_from_params(params: dict[str, object]) -> str: + if "search_query" in params: + return ARXIV_SOURCE_PATH_SEARCH + if "id_list" in params: + return ARXIV_SOURCE_PATH_LOOKUP_IDS + return ARXIV_SOURCE_PATH_UNKNOWN + + +def _log_cache_event( + *, + event_name: str, + query_fingerprint: str, + source_path: str, +) -> None: + logger.info( + event_name, + extra={ + "event": event_name, + "query_fingerprint": query_fingerprint, + "source_path": source_path, + }, + ) + + +def _log_request_skipped_for_cooldown( + *, + source_path: str, + cooldown_remaining_seconds: float, +) -> None: + logger.warning( + "arxiv.request_skipped_cooldown", + extra={ + "event": "arxiv.request_skipped_cooldown", + "source_path": source_path, + "cooldown_remaining_seconds": float(cooldown_remaining_seconds), + }, + ) diff --git a/app/services/domains/arxiv/constants.py b/app/services/domains/arxiv/constants.py new file mode 100644 index 0000000..eab9293 --- /dev/null +++ b/app/services/domains/arxiv/constants.py @@ -0,0 +1,13 @@ +from __future__ import annotations + +ARXIV_RUNTIME_STATE_KEY = "global" +ARXIV_RATE_LIMIT_LOCK_NAMESPACE = 91_100 +ARXIV_RATE_LIMIT_LOCK_KEY = 1 +ARXIV_SOURCE_PATH_SEARCH = "search" +ARXIV_SOURCE_PATH_LOOKUP_IDS = "lookup_ids" +ARXIV_SOURCE_PATH_UNKNOWN = "unknown" +ARXIV_CACHE_FINGERPRINT_VERSION = "v1" +ARXIV_TITLE_TOKEN_MIN_LENGTH = 3 +ARXIV_TITLE_MIN_TOKENS = 3 +ARXIV_TITLE_MIN_ALPHA_TOKENS = 2 +ARXIV_STRONG_IDENTIFIER_CONFIDENCE = 0.9 diff --git a/app/services/domains/arxiv/errors.py b/app/services/domains/arxiv/errors.py new file mode 100644 index 0000000..3561f61 --- /dev/null +++ b/app/services/domains/arxiv/errors.py @@ -0,0 +1,13 @@ +from __future__ import annotations + + +class ArxivRateLimitError(Exception): + """arXiv returned 429 or cooldown is active.""" + + +class ArxivClientValidationError(ValueError): + """arXiv client inputs are invalid.""" + + +class ArxivParseError(ValueError): + """arXiv API payload could not be parsed.""" diff --git a/app/services/domains/arxiv/gateway.py b/app/services/domains/arxiv/gateway.py new file mode 100644 index 0000000..6782025 --- /dev/null +++ b/app/services/domains/arxiv/gateway.py @@ -0,0 +1,141 @@ +from __future__ import annotations + +import logging +import re +from typing import TYPE_CHECKING, Protocol +import unicodedata + +from app.services.domains.arxiv.client import ArxivClient +from app.services.domains.arxiv.errors import ArxivRateLimitError +from app.services.domains.arxiv.types import ArxivFeed +from app.settings import settings + +if TYPE_CHECKING: + from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem + +logger = logging.getLogger(__name__) + +_default_gateway: ArxivGateway | None = None +_MOJIBAKE_HINT_RE = re.compile(r"[ÃÂâ]") +_NON_ALNUM_RE = re.compile(r"[^\w\s]+", re.UNICODE) +_WHITESPACE_RE = re.compile(r"\s+") + + +class ArxivGateway(Protocol): + async def discover_arxiv_id_for_publication( + self, + *, + item: PublicationListItem | UnreadPublicationItem, + request_email: str | None = None, + timeout_seconds: float | None = None, + max_results: int | None = None, + ) -> str | None: ... + + +def build_arxiv_query(title: str, author_surname: str | None) -> str | None: + parts: list[str] = [] + if title: + clean_title = _normalize_query_title(title) + if clean_title: + parts.append(f'ti:"{clean_title}"') + if author_surname: + clean_author = _normalize_query_title(author_surname) + if clean_author: + parts.append(f'au:"{clean_author}"') + if not parts: + return None + return " AND ".join(parts) + + +def _normalize_query_title(value: str) -> str: + repaired = _repair_mojibake(value.strip()) + normalized = unicodedata.normalize("NFKC", repaired) + stripped = _NON_ALNUM_RE.sub(" ", _MOJIBAKE_HINT_RE.sub(" ", normalized)) + return _WHITESPACE_RE.sub(" ", stripped).strip() + + +def _repair_mojibake(value: str) -> str: + if not value or not _MOJIBAKE_HINT_RE.search(value): + return value + try: + repaired = value.encode("latin1").decode("utf-8") + except UnicodeError: + return value + return repaired if _mojibake_score(repaired) < _mojibake_score(value) else value + + +def _mojibake_score(value: str) -> int: + return len(_MOJIBAKE_HINT_RE.findall(value)) + + +def get_arxiv_gateway() -> ArxivGateway: + global _default_gateway + if _default_gateway is None: + _default_gateway = HttpArxivGateway() + return _default_gateway + + +def set_arxiv_gateway(gateway: ArxivGateway | None) -> ArxivGateway | None: + global _default_gateway + previous = _default_gateway + _default_gateway = gateway + return previous + + +class HttpArxivGateway: + def __init__(self, *, client: ArxivClient | None = None) -> None: + self._client = client or ArxivClient() + + async def discover_arxiv_id_for_publication( + self, + *, + item: PublicationListItem | UnreadPublicationItem, + request_email: str | None = None, + timeout_seconds: float | None = None, + max_results: int | None = None, + ) -> str | None: + if not settings.arxiv_enabled: + return None + query = _query_for_item(item) + if query is None: + return None + + try: + result = await self._client.search( + query=query, + start=0, + request_email=request_email, + timeout_seconds=timeout_seconds, + max_results=max_results, + ) + return _first_discovered_id(result) + except ArxivRateLimitError: + raise + except Exception as exc: + logger.debug("arxiv.query_failed", extra={"event": "arxiv.query_failed", "error": str(exc)}) + return None + + +def _query_for_item(item: PublicationListItem | UnreadPublicationItem) -> str | None: + title = (item.title or "").strip() + if not title: + return None + author_surname = _author_surname(item.scholar_label) + return build_arxiv_query(title, author_surname) + + +def _author_surname(scholar_label: str | None) -> str | None: + if not scholar_label: + return None + tokens = [token for token in scholar_label.strip().split() if token] + if not tokens: + return None + return tokens[-1].lower() + + +def _first_discovered_id(result: ArxivFeed) -> str | None: + for entry in result.entries: + if entry.arxiv_id: + logger.debug("arxiv.id_discovered", extra={"event": "arxiv.id_discovered", "arxiv_id": entry.arxiv_id}) + return entry.arxiv_id + return None diff --git a/app/services/domains/arxiv/guards.py b/app/services/domains/arxiv/guards.py new file mode 100644 index 0000000..9f7a35f --- /dev/null +++ b/app/services/domains/arxiv/guards.py @@ -0,0 +1,86 @@ +from __future__ import annotations + +import re +from typing import TYPE_CHECKING + +from app.services.domains.arxiv.constants import ( + ARXIV_STRONG_IDENTIFIER_CONFIDENCE, + ARXIV_TITLE_MIN_ALPHA_TOKENS, + ARXIV_TITLE_MIN_TOKENS, + ARXIV_TITLE_TOKEN_MIN_LENGTH, +) +from app.services.domains.doi.normalize import normalize_doi +from app.services.domains.publication_identifiers.normalize import normalize_arxiv_id + +if TYPE_CHECKING: + from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem + +_TITLE_TOKEN_RE = re.compile(r"[a-z0-9]+") + + +def arxiv_skip_reason_for_item( + *, + item: PublicationListItem | UnreadPublicationItem, + has_strong_doi: bool = False, + has_existing_arxiv: bool = False, +) -> str | None: + if has_existing_arxiv or _has_arxiv_identifier_evidence(item): + return "arxiv_identifier_present" + if has_strong_doi or _has_strong_doi_evidence(item): + return "strong_doi_present" + if not _title_passes_quality_guard(item.title): + return "title_quality_below_threshold" + return None + + +def _has_arxiv_identifier_evidence(item: PublicationListItem | UnreadPublicationItem) -> bool: + if _display_identifier_matches(item, expected_kind="arxiv"): + return True + return _has_normalized_identifier(item, normalizer=normalize_arxiv_id) + + +def _has_strong_doi_evidence(item: PublicationListItem | UnreadPublicationItem) -> bool: + if _display_identifier_matches(item, expected_kind="doi"): + return True + return _has_normalized_identifier(item, normalizer=normalize_doi) + + +def _display_identifier_matches( + item: PublicationListItem | UnreadPublicationItem, + *, + expected_kind: str, +) -> bool: + display = getattr(item, "display_identifier", None) + if display is None: + return False + if str(display.kind).lower() != expected_kind: + return False + return float(display.confidence_score) >= ARXIV_STRONG_IDENTIFIER_CONFIDENCE + + +def _has_normalized_identifier( + item: PublicationListItem | UnreadPublicationItem, + *, + normalizer, +) -> bool: + if normalizer(item.pub_url): + return True + return normalizer(item.pdf_url) is not None + + +def _title_passes_quality_guard(title: str | None) -> bool: + tokens = _normalized_tokens(title or "") + if len(tokens) < ARXIV_TITLE_MIN_TOKENS: + return False + alpha_tokens = [token for token in tokens if _is_alpha_token(token)] + return len(alpha_tokens) >= ARXIV_TITLE_MIN_ALPHA_TOKENS + + +def _normalized_tokens(value: str) -> list[str]: + return [token for token in _TITLE_TOKEN_RE.findall(value.lower()) if token] + + +def _is_alpha_token(token: str) -> bool: + if len(token) < ARXIV_TITLE_TOKEN_MIN_LENGTH: + return False + return any(char.isalpha() for char in token) diff --git a/app/services/domains/arxiv/parser.py b/app/services/domains/arxiv/parser.py new file mode 100644 index 0000000..bc1d7ee --- /dev/null +++ b/app/services/domains/arxiv/parser.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import xml.etree.ElementTree as ET + +from app.services.domains.arxiv.errors import ArxivParseError +from app.services.domains.arxiv.types import ArxivEntry, ArxivFeed, ArxivOpenSearchMeta +from app.services.domains.publication_identifiers.normalize import normalize_arxiv_id + +_NAMESPACES = { + "atom": "http://www.w3.org/2005/Atom", + "opensearch": "http://a9.com/-/spec/opensearch/1.1/", + "arxiv": "http://arxiv.org/schemas/atom", +} + + +def parse_arxiv_feed(payload: str) -> ArxivFeed: + root = _parse_xml_root(payload) + opensearch = ArxivOpenSearchMeta( + total_results=_opensearch_int(root, "opensearch:totalResults"), + start_index=_opensearch_int(root, "opensearch:startIndex"), + items_per_page=_opensearch_int(root, "opensearch:itemsPerPage"), + ) + entries = [_parse_entry(entry_elem) for entry_elem in root.findall("atom:entry", _NAMESPACES)] + return ArxivFeed(entries=entries, opensearch=opensearch) + + +def _parse_xml_root(payload: str) -> ET.Element: + try: + return ET.fromstring(payload) + except ET.ParseError as exc: + raise ArxivParseError(f"Invalid arXiv XML payload: {exc}") from exc + + +def _opensearch_int(root: ET.Element, path: str) -> int: + text = _optional_text(root, path) + if text is None: + return 0 + try: + return int(text.strip()) + except ValueError as exc: + raise ArxivParseError(f"Invalid integer value at {path}: {text!r}") from exc + + +def _parse_entry(entry_elem: ET.Element) -> ArxivEntry: + entry_id_url = _required_text(entry_elem, "atom:id").strip() + arxiv_id = normalize_arxiv_id(entry_id_url) + title = _required_text(entry_elem, "atom:title").strip() + summary = (_optional_text(entry_elem, "atom:summary") or "").strip() + published = _optional_text(entry_elem, "atom:published") + updated = _optional_text(entry_elem, "atom:updated") + return ArxivEntry( + entry_id_url=entry_id_url, + arxiv_id=arxiv_id, + title=title, + summary=summary, + published=published, + updated=updated, + authors=_authors(entry_elem), + links=_links(entry_elem), + categories=_categories(entry_elem), + primary_category=_primary_category(entry_elem), + ) + + +def _required_text(elem: ET.Element, path: str) -> str: + text = _optional_text(elem, path) + if text is None or not text.strip(): + raise ArxivParseError(f"Missing required field: {path}") + return text + + +def _optional_text(elem: ET.Element, path: str) -> str | None: + node = elem.find(path, _NAMESPACES) + if node is None or node.text is None: + return None + return str(node.text) + + +def _authors(entry_elem: ET.Element) -> list[str]: + authors: list[str] = [] + for author in entry_elem.findall("atom:author", _NAMESPACES): + name = _optional_text(author, "atom:name") + if name: + authors.append(name.strip()) + return authors + + +def _links(entry_elem: ET.Element) -> list[str]: + values: list[str] = [] + for link in entry_elem.findall("atom:link", _NAMESPACES): + href = str(link.attrib.get("href") or "").strip() + if href: + values.append(href) + return values + + +def _categories(entry_elem: ET.Element) -> list[str]: + values: list[str] = [] + for cat in entry_elem.findall("atom:category", _NAMESPACES): + term = str(cat.attrib.get("term") or "").strip() + if term: + values.append(term) + return values + + +def _primary_category(entry_elem: ET.Element) -> str | None: + node = entry_elem.find("arxiv:primary_category", _NAMESPACES) + if node is None: + return None + value = str(node.attrib.get("term") or "").strip() + return value or None diff --git a/app/services/domains/arxiv/rate_limit.py b/app/services/domains/arxiv/rate_limit.py new file mode 100644 index 0000000..6506035 --- /dev/null +++ b/app/services/domains/arxiv/rate_limit.py @@ -0,0 +1,247 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Awaitable, Callable +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +import logging + +import httpx +from sqlalchemy import select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import ArxivRuntimeState +from app.db.session import get_session_factory +from app.services.domains.arxiv.constants import ( + ARXIV_RATE_LIMIT_LOCK_KEY, + ARXIV_RATE_LIMIT_LOCK_NAMESPACE, + ARXIV_RUNTIME_STATE_KEY, + ARXIV_SOURCE_PATH_UNKNOWN, +) +from app.services.domains.arxiv.errors import ArxivRateLimitError +from app.settings import settings + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class ArxivCooldownStatus: + is_active: bool + remaining_seconds: float + cooldown_until: datetime | None + + +async def run_with_global_arxiv_limit( + *, + fetch: Callable[[], Awaitable[httpx.Response]], + source_path: str = ARXIV_SOURCE_PATH_UNKNOWN, +) -> httpx.Response: + response, hit_rate_limit = await _run_serialized_fetch( + fetch=fetch, + source_path=source_path, + ) + if hit_rate_limit: + raise ArxivRateLimitError("arXiv rate limit hit (429) — stopping batch") + return response + + +async def get_arxiv_cooldown_status(*, now_utc: datetime | None = None) -> ArxivCooldownStatus: + timestamp = _normalize_datetime(now_utc) or datetime.now(timezone.utc) + session_factory = get_session_factory() + async with session_factory() as db_session: + result = await db_session.execute( + select(ArxivRuntimeState.cooldown_until).where( + ArxivRuntimeState.state_key == ARXIV_RUNTIME_STATE_KEY + ) + ) + cooldown_until = _normalize_datetime(result.scalar_one_or_none()) + remaining_seconds = _cooldown_remaining_seconds(cooldown_until, now_utc=timestamp) + return ArxivCooldownStatus( + is_active=remaining_seconds > 0.0, + remaining_seconds=float(remaining_seconds), + cooldown_until=cooldown_until, + ) + + +async def _run_serialized_fetch( + *, + fetch: Callable[[], Awaitable[httpx.Response]], + source_path: str, +) -> tuple[httpx.Response, bool]: + session_factory = get_session_factory() + async with session_factory() as db_session: + async with db_session.begin(): + await _acquire_arxiv_lock(db_session) + runtime_state = await _load_runtime_state_for_update(db_session) + wait_seconds = await _wait_for_allowed_slot_or_raise( + runtime_state, + source_path=source_path, + ) + response = await fetch() + hit_rate_limit = _record_post_response_state( + runtime_state, + response_status=int(response.status_code), + source_path=source_path, + ) + _log_request_completed( + response_status=int(response.status_code), + wait_seconds=wait_seconds, + source_path=source_path, + cooldown_remaining_seconds=_cooldown_remaining_seconds( + runtime_state.cooldown_until, + now_utc=datetime.now(timezone.utc), + ), + ) + return response, hit_rate_limit + + +async def _acquire_arxiv_lock(db_session: AsyncSession) -> None: + await db_session.execute( + text("SELECT pg_advisory_xact_lock(:namespace, :lock_key)"), + { + "namespace": ARXIV_RATE_LIMIT_LOCK_NAMESPACE, + "lock_key": ARXIV_RATE_LIMIT_LOCK_KEY, + }, + ) + + +async def _load_runtime_state_for_update(db_session: AsyncSession) -> ArxivRuntimeState: + result = await db_session.execute( + select(ArxivRuntimeState) + .where(ArxivRuntimeState.state_key == ARXIV_RUNTIME_STATE_KEY) + .with_for_update() + ) + state = result.scalar_one_or_none() + if state is not None: + return state + state = ArxivRuntimeState(state_key=ARXIV_RUNTIME_STATE_KEY) + db_session.add(state) + await db_session.flush() + return state + + +async def _wait_for_allowed_slot_or_raise( + runtime_state: ArxivRuntimeState, + *, + source_path: str, +) -> float: + now_utc = datetime.now(timezone.utc) + cooldown_seconds = _cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=now_utc) + if cooldown_seconds > 0: + _log_request_scheduled( + wait_seconds=0.0, + source_path=source_path, + cooldown_remaining_seconds=cooldown_seconds, + ) + raise ArxivRateLimitError(f"arXiv global cooldown active ({cooldown_seconds:.0f}s remaining)") + wait_seconds = _next_allowed_wait_seconds(runtime_state.next_allowed_at, now_utc=now_utc) + _log_request_scheduled( + wait_seconds=wait_seconds, + source_path=source_path, + cooldown_remaining_seconds=0.0, + ) + if wait_seconds > 0: + await asyncio.sleep(wait_seconds) + return wait_seconds + + +def _record_post_response_state( + runtime_state: ArxivRuntimeState, + *, + response_status: int, + source_path: str, +) -> bool: + now_utc = datetime.now(timezone.utc) + runtime_state.next_allowed_at = now_utc + timedelta(seconds=_min_interval_seconds()) + if response_status == 429: + cooldown_seconds = _cooldown_seconds() + runtime_state.cooldown_until = now_utc + timedelta(seconds=cooldown_seconds) + _log_cooldown_activated( + source_path=source_path, + cooldown_remaining_seconds=cooldown_seconds, + ) + return True + if _cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=now_utc) <= 0: + runtime_state.cooldown_until = None + return False + + +def _cooldown_remaining_seconds(cooldown_until: datetime | None, *, now_utc: datetime) -> float: + bounded = _normalize_datetime(cooldown_until) + if bounded is None: + return 0.0 + return max((bounded - now_utc).total_seconds(), 0.0) + + +def _next_allowed_wait_seconds(next_allowed_at: datetime | None, *, now_utc: datetime) -> float: + bounded = _normalize_datetime(next_allowed_at) + if bounded is None: + return 0.0 + return max((bounded - now_utc).total_seconds(), 0.0) + + +def _normalize_datetime(value: datetime | None) -> datetime | None: + if value is None: + return None + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + + +def _min_interval_seconds() -> float: + return max(float(settings.arxiv_min_interval_seconds), 0.0) + + +def _cooldown_seconds() -> float: + return max(float(settings.arxiv_rate_limit_cooldown_seconds), 0.0) + + +def _log_request_scheduled( + *, + wait_seconds: float, + source_path: str, + cooldown_remaining_seconds: float, +) -> None: + logger.info( + "arxiv.request_scheduled", + extra={ + "event": "arxiv.request_scheduled", + "wait_seconds": float(wait_seconds), + "cooldown_remaining_seconds": float(cooldown_remaining_seconds), + "source_path": source_path, + }, + ) + + +def _log_request_completed( + *, + response_status: int, + wait_seconds: float, + source_path: str, + cooldown_remaining_seconds: float, +) -> None: + logger.info( + "arxiv.request_completed", + extra={ + "event": "arxiv.request_completed", + "status_code": int(response_status), + "wait_seconds": float(wait_seconds), + "cooldown_remaining_seconds": float(cooldown_remaining_seconds), + "source_path": source_path, + }, + ) + + +def _log_cooldown_activated( + *, + source_path: str, + cooldown_remaining_seconds: float, +) -> None: + logger.warning( + "arxiv.cooldown_activated", + extra={ + "event": "arxiv.cooldown_activated", + "cooldown_remaining_seconds": float(cooldown_remaining_seconds), + "source_path": source_path, + }, + ) diff --git a/app/services/domains/arxiv/types.py b/app/services/domains/arxiv/types.py new file mode 100644 index 0000000..574ed93 --- /dev/null +++ b/app/services/domains/arxiv/types.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Literal + +ArxivSortBy = Literal["relevance", "lastUpdatedDate", "submittedDate"] +ArxivSortOrder = Literal["ascending", "descending"] + + +@dataclass(frozen=True) +class ArxivOpenSearchMeta: + total_results: int = 0 + start_index: int = 0 + items_per_page: int = 0 + + +@dataclass(frozen=True) +class ArxivEntry: + entry_id_url: str + arxiv_id: str | None + title: str + summary: str + published: str | None + updated: str | None + authors: list[str] = field(default_factory=list) + links: list[str] = field(default_factory=list) + categories: list[str] = field(default_factory=list) + primary_category: str | None = None + + +@dataclass(frozen=True) +class ArxivFeed: + entries: list[ArxivEntry] = field(default_factory=list) + opensearch: ArxivOpenSearchMeta = field(default_factory=ArxivOpenSearchMeta) diff --git a/app/services/domains/dbops/__init__.py b/app/services/domains/dbops/__init__.py index f5b02e4..cdc5177 100644 --- a/app/services/domains/dbops/__init__.py +++ b/app/services/domains/dbops/__init__.py @@ -1,5 +1,13 @@ from app.services.domains.dbops.application import run_publication_link_repair +from app.services.domains.dbops.near_duplicate_repair import ( + run_publication_near_duplicate_repair, +) from app.services.domains.dbops.integrity import collect_integrity_report from app.services.domains.dbops.query import list_repair_jobs -__all__ = ["collect_integrity_report", "list_repair_jobs", "run_publication_link_repair"] +__all__ = [ + "collect_integrity_report", + "list_repair_jobs", + "run_publication_link_repair", + "run_publication_near_duplicate_repair", +] diff --git a/app/services/domains/dbops/near_duplicate_repair.py b/app/services/domains/dbops/near_duplicate_repair.py new file mode 100644 index 0000000..ab86c70 --- /dev/null +++ b/app/services/domains/dbops/near_duplicate_repair.py @@ -0,0 +1,223 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import DataRepairJob +from app.services.domains.publications import dedup as dedup_service + +REPAIR_STATUS_PLANNED = "planned" +REPAIR_STATUS_RUNNING = "running" +REPAIR_STATUS_COMPLETED = "completed" +REPAIR_STATUS_FAILED = "failed" +NEAR_DUP_JOB_NAME = "repair_publication_near_duplicates" +NEAR_DUP_DEFAULT_MAX_CLUSTERS = 25 + + +def _utcnow() -> datetime: + return datetime.now(timezone.utc) + + +def _normalized_cluster_keys(values: list[str] | None) -> list[str]: + if not values: + return [] + seen: set[str] = set() + normalized: list[str] = [] + for value in values: + key = str(value or "").strip().lower() + if not key or key in seen: + continue + seen.add(key) + normalized.append(key) + return normalized + + +def _normalized_max_clusters(value: int) -> int: + return max(1, min(int(value), 200)) + + +def _scope_payload( + *, + similarity_threshold: float, + min_shared_tokens: int, + max_year_delta: int, + max_clusters: int, + selected_cluster_keys: list[str], +) -> dict[str, Any]: + return { + "similarity_threshold": float(similarity_threshold), + "min_shared_tokens": int(min_shared_tokens), + "max_year_delta": int(max_year_delta), + "max_clusters": int(max_clusters), + "selected_cluster_keys": selected_cluster_keys, + } + + +async def _create_job( + db_session: AsyncSession, + *, + requested_by: str | None, + scope: dict[str, Any], + dry_run: bool, +) -> DataRepairJob: + job = DataRepairJob( + job_name=NEAR_DUP_JOB_NAME, + requested_by=(requested_by or "").strip() or None, + scope=scope, + dry_run=bool(dry_run), + status=REPAIR_STATUS_PLANNED, + summary={}, + ) + db_session.add(job) + await db_session.flush() + job.status = REPAIR_STATUS_RUNNING + job.started_at = _utcnow() + return job + + +def _selected_clusters( + *, + clusters: list[dedup_service.NearDuplicateCluster], + selected_cluster_keys: list[str], +) -> tuple[list[dedup_service.NearDuplicateCluster], list[str]]: + if not selected_cluster_keys: + return [], [] + by_key = {cluster.cluster_key.lower(): cluster for cluster in clusters} + selected: list[dedup_service.NearDuplicateCluster] = [] + missing: list[str] = [] + for key in selected_cluster_keys: + cluster = by_key.get(key) + if cluster is None: + missing.append(key) + continue + selected.append(cluster) + return selected, missing + + +async def _merge_selected_clusters( + db_session: AsyncSession, + *, + selected_clusters: list[dedup_service.NearDuplicateCluster], +) -> int: + merged_publications = 0 + for cluster in selected_clusters: + merged_publications += await dedup_service.merge_near_duplicate_cluster( + db_session, + cluster=cluster, + ) + return merged_publications + + +def _clusters_payload( + *, + clusters: list[dedup_service.NearDuplicateCluster], + max_clusters: int, +) -> list[dict[str, object]]: + preview = clusters[:max_clusters] + return [dedup_service.near_duplicate_cluster_payload(cluster) for cluster in preview] + + +def _summary_payload( + *, + dry_run: bool, + cluster_count: int, + selected_count: int, + missing_count: int, + merged_publications: int, + max_clusters: int, +) -> dict[str, Any]: + return { + "dry_run": bool(dry_run), + "candidate_cluster_count": int(cluster_count), + "selected_cluster_count": int(selected_count), + "missing_selected_cluster_count": int(missing_count), + "merged_publications": int(merged_publications), + "preview_cluster_count": int(min(cluster_count, max_clusters)), + } + + +async def _complete_job( + db_session: AsyncSession, + *, + job: DataRepairJob, + scope: dict[str, Any], + summary: dict[str, Any], + clusters: list[dict[str, object]], +) -> dict[str, Any]: + job.status = REPAIR_STATUS_COMPLETED + job.finished_at = _utcnow() + job.summary = summary + await db_session.commit() + return { + "job_id": int(job.id), + "status": job.status, + "scope": scope, + "summary": summary, + "clusters": clusters, + } + + +async def _fail_job(db_session: AsyncSession, *, job: DataRepairJob, error: Exception) -> None: + await db_session.rollback() + job.status = REPAIR_STATUS_FAILED + job.error_text = str(error) + job.finished_at = _utcnow() + db_session.add(job) + await db_session.commit() + + +async def run_publication_near_duplicate_repair( + db_session: AsyncSession, + *, + dry_run: bool = True, + similarity_threshold: float = dedup_service.NEAR_DUP_DEFAULT_SIMILARITY_THRESHOLD, + min_shared_tokens: int = dedup_service.NEAR_DUP_DEFAULT_MIN_SHARED_TOKENS, + max_year_delta: int = dedup_service.NEAR_DUP_DEFAULT_MAX_YEAR_DELTA, + max_clusters: int = NEAR_DUP_DEFAULT_MAX_CLUSTERS, + selected_cluster_keys: list[str] | None = None, + requested_by: str | None = None, +) -> dict[str, Any]: + normalized_keys = _normalized_cluster_keys(selected_cluster_keys) + bounded_clusters = _normalized_max_clusters(max_clusters) + scope = _scope_payload( + similarity_threshold=similarity_threshold, + min_shared_tokens=min_shared_tokens, + max_year_delta=max_year_delta, + max_clusters=bounded_clusters, + selected_cluster_keys=normalized_keys, + ) + job = await _create_job(db_session, requested_by=requested_by, scope=scope, dry_run=dry_run) + try: + clusters = await dedup_service.find_near_duplicate_clusters( + db_session, + similarity_threshold=similarity_threshold, + min_shared_tokens=min_shared_tokens, + max_year_delta=max_year_delta, + ) + selected, missing = _selected_clusters(clusters=clusters, selected_cluster_keys=normalized_keys) + merged_publications = 0 + if not dry_run: + if not selected: + raise ValueError("No selected near-duplicate clusters matched current data.") + merged_publications = await _merge_selected_clusters(db_session, selected_clusters=selected) + preview = _clusters_payload(clusters=clusters, max_clusters=bounded_clusters) + summary = _summary_payload( + dry_run=dry_run, + cluster_count=len(clusters), + selected_count=len(selected), + missing_count=len(missing), + merged_publications=merged_publications, + max_clusters=bounded_clusters, + ) + return await _complete_job( + db_session, + job=job, + scope=scope, + summary=summary, + clusters=preview, + ) + except Exception as exc: + await _fail_job(db_session, job=job, error=exc) + raise diff --git a/app/services/domains/ingestion/application.py b/app/services/domains/ingestion/application.py index a313a6f..cedd165 100644 --- a/app/services/domains/ingestion/application.py +++ b/app/services/domains/ingestion/application.py @@ -30,6 +30,7 @@ from app.services.domains.ingestion.constants import ( RESUMABLE_PARTIAL_REASONS, RUN_LOCK_NAMESPACE, ) +from app.services.domains.arxiv.errors import ArxivRateLimitError from app.services.domains.doi.normalize import first_doi_from_texts from app.services.domains.publication_identifiers import application as identifier_service from app.services.domains.ingestion.fingerprints import ( @@ -830,17 +831,79 @@ class ScholarIngestionService: result_entry=result_entry, ) + @staticmethod + def _result_counters(result_entry: dict[str, Any]) -> tuple[int, int, int]: + outcome = str(result_entry.get("outcome", "")).strip().lower() + if outcome == "success": + return 1, 0, 0 + if outcome == "partial": + return 1, 0, 1 + if outcome == "failed": + return 0, 1, 0 + raise RuntimeError(f"Unexpected scholar outcome label: {outcome!r}") + + @staticmethod + def _find_scholar_result_index( + *, + scholar_results: list[dict[str, Any]], + scholar_profile_id: int, + ) -> int | None: + for index, result_entry in enumerate(scholar_results): + current_scholar_id = _int_or_default(result_entry.get("scholar_profile_id"), 0) + if current_scholar_id == scholar_profile_id: + return index + return None + + @staticmethod + def _adjust_progress_counts( + *, + progress: RunProgress, + succeeded_delta: int, + failed_delta: int, + partial_delta: int, + ) -> None: + progress.succeeded_count += succeeded_delta + progress.failed_count += failed_delta + progress.partial_count += partial_delta + if progress.succeeded_count < 0 or progress.failed_count < 0 or progress.partial_count < 0: + raise RuntimeError("RunProgress counters entered invalid negative state.") + @staticmethod def _apply_outcome_to_progress( *, progress: RunProgress, - run: CrawlRun, outcome: ScholarProcessingOutcome, ) -> None: - progress.succeeded_count += outcome.succeeded_count_delta - progress.failed_count += outcome.failed_count_delta - progress.partial_count += outcome.partial_count_delta - progress.scholar_results.append(outcome.result_entry) + scholar_profile_id = _int_or_default(outcome.result_entry.get("scholar_profile_id"), 0) + if scholar_profile_id <= 0: + raise RuntimeError("Scholar outcome missing valid scholar_profile_id.") + prior_index = ScholarIngestionService._find_scholar_result_index( + scholar_results=progress.scholar_results, + scholar_profile_id=scholar_profile_id, + ) + next_succeeded, next_failed, next_partial = ScholarIngestionService._result_counters( + outcome.result_entry + ) + if prior_index is None: + progress.scholar_results.append(outcome.result_entry) + ScholarIngestionService._adjust_progress_counts( + progress=progress, + succeeded_delta=next_succeeded, + failed_delta=next_failed, + partial_delta=next_partial, + ) + return + previous_entry = progress.scholar_results[prior_index] + prev_succeeded, prev_failed, prev_partial = ScholarIngestionService._result_counters( + previous_entry + ) + progress.scholar_results[prior_index] = outcome.result_entry + ScholarIngestionService._adjust_progress_counts( + progress=progress, + succeeded_delta=next_succeeded - prev_succeeded, + failed_delta=next_failed - prev_failed, + partial_delta=next_partial - prev_partial, + ) def _unexpected_scholar_exception_outcome( self, @@ -1327,7 +1390,7 @@ class ScholarIngestionService: auto_queue_continuations=False, queue_delay_seconds=queue_delay_seconds, ) - self._apply_outcome_to_progress(progress=progress, run=run, outcome=outcome) + self._apply_outcome_to_progress(progress=progress, outcome=outcome) # Track where to resume from for the deep pass resume_cstart = outcome.result_entry.get("continuation_cstart") if resume_cstart is not None and int(resume_cstart) > start_cstart: @@ -1366,7 +1429,7 @@ class ScholarIngestionService: auto_queue_continuations=auto_queue_continuations, queue_delay_seconds=queue_delay_seconds, ) - self._apply_outcome_to_progress(progress=progress, run=run, outcome=outcome) + self._apply_outcome_to_progress(progress=progress, outcome=outcome) return progress def _complete_run_for_user( @@ -2448,6 +2511,7 @@ class ScholarIngestionService: resolved_key = openalex_api_key or settings.openalex_api_key client = OpenAlexClient(api_key=resolved_key, mailto=settings.crossref_api_mailto) batch_size = 25 + arxiv_lookup_allowed = True for i in range(0, len(publications), batch_size): if await self._run_is_canceled(db_session, run_id=run_id): @@ -2494,12 +2558,11 @@ class ScholarIngestionService: return p.openalex_last_attempt_at = now - - # Perform identifier discovery (moved from synchronous ingest loop) - await identifier_service.discover_and_sync_identifiers_for_publication( + arxiv_lookup_allowed = await self._discover_identifiers_for_enrichment( db_session, publication=p, - scholar_label=p.author_text or "", + run_id=run_id, + allow_arxiv_lookup=arxiv_lookup_allowed, ) match = find_best_match( @@ -2529,6 +2592,86 @@ class ScholarIngestionService: }, ) + async def _discover_identifiers_for_enrichment( + self, + db_session: AsyncSession, + *, + publication: Publication, + run_id: int, + allow_arxiv_lookup: bool, + ) -> bool: + if not allow_arxiv_lookup: + await identifier_service.sync_identifiers_for_publication_fields( + db_session, + publication=publication, + ) + await self._publish_identifier_update_event( + db_session, + run_id=run_id, + publication_id=int(publication.id), + ) + return False + try: + await identifier_service.discover_and_sync_identifiers_for_publication( + db_session, + publication=publication, + scholar_label=publication.author_text or "", + ) + await self._publish_identifier_update_event( + db_session, + run_id=run_id, + publication_id=int(publication.id), + ) + return True + except ArxivRateLimitError: + logger.warning( + "ingestion.arxiv_rate_limited", + extra={ + "event": "ingestion.arxiv_rate_limited", + "run_id": run_id, + "publication_id": int(publication.id), + "detail": "arXiv temporarily disabled for remaining enrichment pass", + }, + ) + await identifier_service.sync_identifiers_for_publication_fields( + db_session, + publication=publication, + ) + await self._publish_identifier_update_event( + db_session, + run_id=run_id, + publication_id=int(publication.id), + ) + return False + + async def _publish_identifier_update_event( + self, + db_session: AsyncSession, + *, + run_id: int, + publication_id: int, + ) -> None: + display = await identifier_service.display_identifier_for_publication_id( + db_session, + publication_id=publication_id, + ) + if display is None: + return + await run_events.publish( + run_id=run_id, + event_type="identifier_updated", + data={ + "publication_id": int(publication_id), + "display_identifier": { + "kind": display.kind, + "value": display.value, + "label": display.label, + "url": display.url, + "confidence_score": float(display.confidence_score), + }, + }, + ) + async def _enrich_publications_with_openalex( self, scholar: ScholarProfile, diff --git a/app/services/domains/ingestion/fingerprints.py b/app/services/domains/ingestion/fingerprints.py index 30745b0..9f67f0f 100644 --- a/app/services/domains/ingestion/fingerprints.py +++ b/app/services/domains/ingestion/fingerprints.py @@ -4,6 +4,7 @@ import hashlib import json import re from typing import Any +import unicodedata from urllib.parse import urljoin from app.services.domains.ingestion.constants import ( @@ -24,37 +25,177 @@ _NOISE_PREPRINT_RE = re.compile( re.IGNORECASE, ) _NOISE_TRAILING_YEAR_RE = re.compile(r"\s*[,(]\s*\d{4}\s*[),]?\s*$") +_NOISE_TRAILING_MONTH_YEAR_RE = re.compile( + r"\s*[,(]\s*(?:jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec)[a-z]*\.?\s+\d{4}\s*[),]?\s*$", + re.IGNORECASE, +) +_NOISE_TRAILING_PUBLICATION_TYPE_RE = re.compile( + r"[,.\s]+(?:conference\s+paper|journal\s+article)\s*$", + re.IGNORECASE, +) +_NOISE_IN_PROCEEDINGS_SUFFIX_RE = re.compile(r"\s+in:\s+proceedings\b.*$", re.IGNORECASE) # Strips ". Capitalised sentence" appended as venue: ". Comput. Sci…", ". Journal of…" _NOISE_VENUE_SENTENCE_RE = re.compile(r"(?<=\w{3})\.\s+[A-Z][a-z].*$") +_MOJIBAKE_HINT_RE = re.compile(r"[ÃÂâ]") +_MOJIBAKE_CHAR_RE = re.compile(r"[Ó”€™]") +_METADATA_ORDINAL_RE = re.compile(r"^\d+(st|nd|rd|th)$") +_NOISE_LEADING_DATE_PREFIX_RE = re.compile( + r"^(?:jan|feb|mar|apr|may|jun|jul|aug|sep|sept|oct|nov|dec)[a-z]*\s+\d{1,2}(?:\s*[-–]\s*\d{1,2})?\)?[,\.\s:;-]+", + re.IGNORECASE, +) +_NOISE_LEADING_AUTHOR_FRAGMENT_RE = re.compile(r"^(?:and|&)\s+[a-z.\s]{1,40}:\s*", re.IGNORECASE) +_METADATA_SEPARATORS = (" - ", " — ", ",", ";", ". ") +_VENUE_HINT_TOKENS = { + "aaai", + "conference", + "conf", + "cvpr", + "eccv", + "iclr", + "icml", + "journal", + "nips", + "neurips", + "proceedings", + "proc", + "symposium", + "workshop", +} +_PUBLICATION_TYPE_TOKENS = {"conference", "paper", "journal", "article"} +_MIN_METADATA_HINT_TOKENS = 2 +_MIN_METADATA_CONTEXT_TOKENS = 4 _CANONICAL_DEDUP_THRESHOLD = 0.82 def normalize_title(value: str) -> str: - lowered = value.lower() + lowered = _normalized_text(value).lower() return TITLE_ALNUM_RE.sub("", lowered) def canonical_title_for_dedup(title: str) -> str: """Strip Scholar-specific noise suffixes then normalize for dedup comparison.""" - t = title.strip() - t = _NOISE_DOI_RE.sub("", t) - t = _NOISE_ARXIV_RE.sub("", t) - t = _NOISE_PREPRINT_RE.sub("", t) - t = _NOISE_TRAILING_YEAR_RE.sub("", t) - t = _NOISE_VENUE_SENTENCE_RE.sub("", t) - return normalize_title(t.strip()) + return normalize_title(_canonical_title_text(title)) + + +def canonical_title_text_for_dedup(title: str) -> str: + """Noise-stripped lowercase title with spaces preserved for token-level matching.""" + return _stripped_title_for_canonical(title) + + +def canonical_title_tokens_for_dedup(title: str) -> set[str]: + """Word tokens of the noise-stripped title.""" + return _canonical_title_tokens(title) def _stripped_title_for_canonical(title: str) -> str: """Apply noise-stripping and lowercase but PRESERVE spaces (for later tokenization).""" - t = title.strip() + t = _canonical_title_text(title) + return t.lower().strip() + + +def _canonical_title_text(title: str) -> str: + t = _normalized_text(title) + t = _strip_noise_suffixes(t) + t = _strip_venue_metadata_suffixes(t) + return _NOISE_VENUE_SENTENCE_RE.sub("", t).strip() + + +def _strip_noise_suffixes(value: str) -> str: + t = _strip_leading_noise_prefixes(value.strip()) t = _NOISE_DOI_RE.sub("", t) t = _NOISE_ARXIV_RE.sub("", t) t = _NOISE_PREPRINT_RE.sub("", t) t = _NOISE_TRAILING_YEAR_RE.sub("", t) - t = _NOISE_VENUE_SENTENCE_RE.sub("", t) - return t.lower().strip() + t = _NOISE_TRAILING_MONTH_YEAR_RE.sub("", t) + t = _NOISE_TRAILING_PUBLICATION_TYPE_RE.sub("", t) + t = _NOISE_IN_PROCEEDINGS_SUFFIX_RE.sub("", t) + return t.strip() + + +def _strip_venue_metadata_suffixes(value: str) -> str: + stripped = value.strip() + while True: + cut_index = _metadata_cut_index(stripped) + if cut_index is None: + return stripped + stripped = stripped[:cut_index].strip() + + +def _metadata_cut_index(value: str) -> int | None: + candidates: list[int] = [] + for candidate in _METADATA_SEPARATORS: + start = 0 + while True: + index = value.find(candidate, start) + if index <= 0: + break + suffix = value[index + len(candidate) :].strip() + if suffix and _looks_like_venue_metadata(suffix): + candidates.append(index) + start = index + len(candidate) + if not candidates: + return None + return min(candidates) + + +def _looks_like_venue_metadata(value: str) -> bool: + tokens = WORD_RE.findall(value.lower()) + if len(tokens) < _MIN_METADATA_HINT_TOKENS: + return False + has_hint = any(_is_venue_hint_token(token) for token in tokens) + if not has_hint: + return False + has_year = any(_is_year_token(token) for token in tokens) + has_ordinal = any(_METADATA_ORDINAL_RE.match(token) for token in tokens) + publication_type_only = all(token in _PUBLICATION_TYPE_TOKENS for token in tokens) + return has_year or has_ordinal or publication_type_only or len(tokens) >= _MIN_METADATA_CONTEXT_TOKENS + + +def _strip_leading_noise_prefixes(value: str) -> str: + stripped = value + while True: + next_value = _NOISE_LEADING_DATE_PREFIX_RE.sub("", stripped).strip() + next_value = _NOISE_LEADING_AUTHOR_FRAGMENT_RE.sub("", next_value).strip() + if next_value == stripped: + return stripped + stripped = next_value + + +def _is_venue_hint_token(token: str) -> bool: + if token in _VENUE_HINT_TOKENS: + return True + return token.startswith("conf") or token.startswith("proceed") + + +def _is_year_token(token: str) -> bool: + if len(token) != 4 or not token.isdigit(): + return False + year = int(token) + return 1900 <= year <= 2100 + + +def _normalized_text(value: str) -> str: + repaired = _repair_mojibake(value.strip()) + normalized = unicodedata.normalize("NFKC", repaired) + cleaned = _MOJIBAKE_CHAR_RE.sub(" ", normalized) + return SPACE_RE.sub(" ", cleaned).strip() + + +def _repair_mojibake(value: str) -> str: + if not value or not _MOJIBAKE_HINT_RE.search(value): + return value + try: + repaired = value.encode("latin1").decode("utf-8") + except UnicodeError: + return value + if _mojibake_score(repaired) < _mojibake_score(value): + return repaired + return value + + +def _mojibake_score(value: str) -> int: + return len(_MOJIBAKE_HINT_RE.findall(value)) def _canonical_title_tokens(title: str) -> set[str]: diff --git a/app/services/domains/publication_identifiers/__init__.py b/app/services/domains/publication_identifiers/__init__.py index 3a35c66..9331dc0 100644 --- a/app/services/domains/publication_identifiers/__init__.py +++ b/app/services/domains/publication_identifiers/__init__.py @@ -1,5 +1,6 @@ from app.services.domains.publication_identifiers.application import ( DisplayIdentifier, + display_identifier_for_publication_id, derive_display_identifier_from_values, overlay_pdf_queue_items_with_display_identifiers, overlay_publication_items_with_display_identifiers, @@ -9,6 +10,7 @@ from app.services.domains.publication_identifiers.application import ( __all__ = [ "DisplayIdentifier", + "display_identifier_for_publication_id", "derive_display_identifier_from_values", "overlay_pdf_queue_items_with_display_identifiers", "overlay_publication_items_with_display_identifiers", diff --git a/app/services/domains/publication_identifiers/application.py b/app/services/domains/publication_identifiers/application.py index ca57d57..31f506c 100644 --- a/app/services/domains/publication_identifiers/application.py +++ b/app/services/domains/publication_identifiers/application.py @@ -7,6 +7,7 @@ from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import Publication, PublicationIdentifier +from app.services.domains.arxiv.guards import arxiv_skip_reason_for_item from app.services.domains.doi.normalize import normalize_doi from app.services.domains.publication_identifiers.normalize import ( normalize_arxiv_id, @@ -124,19 +125,45 @@ async def discover_and_sync_identifiers_for_publication( scholar_label: str, ) -> None: await sync_identifiers_for_publication_fields(db_session, publication=publication) - - existing_doi = await _existing_identifier_by_kind( + + publication_id = int(publication.id) + if await _has_confident_identifier( db_session, - publication_id=int(publication.id), + publication_id=publication_id, kind=IdentifierKind.DOI.value, - ) - if existing_doi is not None: + confidence_floor=0.0, + ): return - from app.services.domains.crossref import application as crossref_service + item = _identifier_lookup_item(publication=publication, scholar_label=scholar_label) + has_strong_doi = await _discover_crossref_doi( + db_session, + publication_id=publication_id, + item=item, + ) + existing_arxiv = await _existing_identifier_by_kind( + db_session, + publication_id=publication_id, + kind=IdentifierKind.ARXIV.value, + ) + skip_reason = arxiv_skip_reason_for_item( + item=item, + has_strong_doi=has_strong_doi, + has_existing_arxiv=existing_arxiv is not None, + ) + if skip_reason is not None: + return + await _discover_arxiv_identifier(db_session, publication_id=publication_id, item=item) + + +def _identifier_lookup_item( + *, + publication: Publication, + scholar_label: str, +) -> UnreadPublicationItem: from app.services.domains.publications.types import UnreadPublicationItem - - item = UnreadPublicationItem( + + return UnreadPublicationItem( publication_id=int(publication.id), scholar_profile_id=0, scholar_label=scholar_label, @@ -147,41 +174,78 @@ async def discover_and_sync_identifiers_for_publication( pub_url=publication.pub_url, pdf_url=publication.pdf_url, ) - - discovered_doi = await crossref_service.discover_doi_for_publication(item=item) - if discovered_doi: - normalized_doi = normalize_doi(discovered_doi) - if normalized_doi: - candidate = _candidate( - IdentifierKind.DOI, - discovered_doi, - normalized_doi, - "crossref_api", - CONFIDENCE_MEDIUM, - None, - ) - await _upsert_publication_candidate(db_session, publication_id=int(publication.id), candidate=candidate) - existing_arxiv = await _existing_identifier_by_kind( - db_session, - publication_id=int(publication.id), - kind=IdentifierKind.ARXIV.value, + +async def _discover_crossref_doi( + db_session: AsyncSession, + *, + publication_id: int, + item: UnreadPublicationItem, +) -> bool: + from app.services.domains.crossref import application as crossref_service + + discovered_doi = await crossref_service.discover_doi_for_publication(item=item) + normalized_doi = normalize_doi(discovered_doi) + if normalized_doi is None: + return False + candidate = _candidate( + IdentifierKind.DOI, + discovered_doi, + normalized_doi, + "crossref_api", + CONFIDENCE_MEDIUM, + None, ) - if existing_arxiv is None: - from app.services.domains.arxiv import application as arxiv_service - discovered_arxiv = await arxiv_service.discover_arxiv_id_for_publication(item=item) - if discovered_arxiv: - normalized_arxiv = normalize_arxiv_id(discovered_arxiv) - if normalized_arxiv: - candidate = _candidate( - IdentifierKind.ARXIV, - discovered_arxiv, - normalized_arxiv, - "arxiv_api", - CONFIDENCE_MEDIUM, - None, - ) - await _upsert_publication_candidate(db_session, publication_id=int(publication.id), candidate=candidate) + await _upsert_publication_candidate( + db_session, + publication_id=publication_id, + candidate=candidate, + ) + return candidate.confidence_score >= CONFIDENCE_MEDIUM + + +async def _discover_arxiv_identifier( + db_session: AsyncSession, + *, + publication_id: int, + item: UnreadPublicationItem, +) -> None: + from app.services.domains.arxiv import application as arxiv_service + + discovered_arxiv = await arxiv_service.discover_arxiv_id_for_publication(item=item) + normalized_arxiv = normalize_arxiv_id(discovered_arxiv) + if normalized_arxiv is None: + return + candidate = _candidate( + IdentifierKind.ARXIV, + discovered_arxiv, + normalized_arxiv, + "arxiv_api", + CONFIDENCE_MEDIUM, + None, + ) + await _upsert_publication_candidate( + db_session, + publication_id=publication_id, + candidate=candidate, + ) + + +async def _has_confident_identifier( + db_session: AsyncSession, + *, + publication_id: int, + kind: str, + confidence_floor: float, +) -> bool: + existing = await _existing_identifier_by_kind( + db_session, + publication_id=publication_id, + kind=kind, + ) + if existing is None: + return False + return float(existing.confidence_score) >= float(confidence_floor) def _publication_field_candidates(publication: Publication) -> list[IdentifierCandidate]: @@ -339,6 +403,28 @@ def _overlay_queue_item( return replace(item, display_identifier=fallback) +async def display_identifier_for_publication_id( + db_session: AsyncSession, + *, + publication_id: int, +) -> DisplayIdentifier | None: + normalized_id = int(publication_id) + if normalized_id <= 0: + raise ValueError("publication_id must be positive.") + mapping = await _display_identifier_map(db_session, publication_ids=[normalized_id]) + display = mapping.get(normalized_id) + if display is not None: + return display + publication = await db_session.get(Publication, normalized_id) + if publication is None: + return None + return derive_display_identifier_from_values( + doi=None, + pub_url=publication.pub_url, + pdf_url=publication.pdf_url, + ) + + async def _display_identifier_map( db_session: AsyncSession, *, diff --git a/app/services/domains/publication_identifiers/normalize.py b/app/services/domains/publication_identifiers/normalize.py index 9f7484a..9743092 100644 --- a/app/services/domains/publication_identifiers/normalize.py +++ b/app/services/domains/publication_identifiers/normalize.py @@ -7,6 +7,7 @@ from app.services.domains.doi.normalize import normalize_doi from app.services.domains.publication_identifiers.types import IdentifierKind ARXIV_ABS_RE = re.compile(r"\barxiv:\s*([a-z-]+/\d{7}|\d{4}\.\d{4,5})(v\d+)?\b", re.I) +ARXIV_RAW_RE = re.compile(r"^([a-z-]+/\d{7}|\d{4}\.\d{4,5})(v\d+)?$", re.I) ARXIV_PATH_RE = re.compile(r"^/(?:abs|pdf|html|ps|format)/([a-z-]+/\d{7}|\d{4}\.\d{4,5})(v\d+)?(?:\.pdf)?/?$", re.I) PMCID_RE = re.compile(r"\b(PMC\d+)\b", re.I) PUBMED_PATH_RE = re.compile(r"^/(\d+)/?$") @@ -31,6 +32,10 @@ def normalize_arxiv_id(value: str | None) -> str | None: parsed = urlparse(text) if parsed.scheme in {"http", "https"} and "arxiv.org" in parsed.netloc.lower(): return _arxiv_from_path(parsed.path) + raw_match = ARXIV_RAW_RE.match(text) + if raw_match: + version = (raw_match.group(2) or "").lower() + return f"{raw_match.group(1).lower()}{version}" match = ARXIV_ABS_RE.search(text) if not match: return None diff --git a/app/services/domains/publications/counts.py b/app/services/domains/publications/counts.py index 51a524d..700043a 100644 --- a/app/services/domains/publications/counts.py +++ b/app/services/domains/publications/counts.py @@ -5,7 +5,7 @@ from datetime import datetime from sqlalchemy import distinct, func, select from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import ScholarProfile, ScholarPublication +from app.db.models import Publication, ScholarProfile, ScholarPublication from app.services.domains.publications.modes import ( MODE_ALL, MODE_LATEST, @@ -22,6 +22,7 @@ async def count_for_user( mode: str = MODE_ALL, scholar_profile_id: int | None = None, favorite_only: bool = False, + search: str | None = None, snapshot_before: datetime | None = None, ) -> int: resolved_mode = resolve_publication_view_mode(mode) @@ -30,8 +31,10 @@ async def count_for_user( select(func.count(distinct(ScholarPublication.publication_id))) .select_from(ScholarPublication) .join(ScholarProfile, ScholarProfile.id == ScholarPublication.scholar_profile_id) + .join(Publication, Publication.id == ScholarPublication.publication_id) .where(ScholarProfile.user_id == user_id) ) + stmt = _apply_search_filter(stmt, search=search) if scholar_profile_id is not None: stmt = stmt.where(ScholarProfile.id == scholar_profile_id) if favorite_only: @@ -48,12 +51,25 @@ async def count_for_user( return int(result.scalar_one() or 0) +def _apply_search_filter(stmt, *, search: str | None): + if not search: + return stmt + safe_search = search.replace("%", r"\%").replace("_", r"\_") + pattern = f"%{safe_search}%" + return stmt.where( + Publication.title_raw.ilike(pattern) + | ScholarProfile.display_name.ilike(pattern) + | Publication.venue_text.ilike(pattern) + ) + + async def count_unread_for_user( db_session: AsyncSession, *, user_id: int, scholar_profile_id: int | None = None, favorite_only: bool = False, + search: str | None = None, snapshot_before: datetime | None = None, ) -> int: return await count_for_user( @@ -62,6 +78,7 @@ async def count_unread_for_user( mode=MODE_UNREAD, scholar_profile_id=scholar_profile_id, favorite_only=favorite_only, + search=search, snapshot_before=snapshot_before, ) @@ -72,6 +89,7 @@ async def count_latest_for_user( user_id: int, scholar_profile_id: int | None = None, favorite_only: bool = False, + search: str | None = None, snapshot_before: datetime | None = None, ) -> int: return await count_for_user( @@ -80,6 +98,7 @@ async def count_latest_for_user( mode=MODE_LATEST, scholar_profile_id=scholar_profile_id, favorite_only=favorite_only, + search=search, snapshot_before=snapshot_before, ) @@ -89,6 +108,7 @@ async def count_favorite_for_user( *, user_id: int, scholar_profile_id: int | None = None, + search: str | None = None, snapshot_before: datetime | None = None, ) -> int: return await count_for_user( @@ -97,5 +117,6 @@ async def count_favorite_for_user( mode=MODE_ALL, scholar_profile_id=scholar_profile_id, favorite_only=True, + search=search, snapshot_before=snapshot_before, ) diff --git a/app/services/domains/publications/dedup.py b/app/services/domains/publications/dedup.py index 8f3c7b7..6615d3b 100644 --- a/app/services/domains/publications/dedup.py +++ b/app/services/domains/publications/dedup.py @@ -1,24 +1,77 @@ from __future__ import annotations +from dataclasses import dataclass +import hashlib import logging +from typing import Iterable from sqlalchemy import delete, select -from sqlalchemy.orm import aliased from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import aliased from app.db.models import Publication, PublicationIdentifier, ScholarPublication +from app.services.domains.ingestion.fingerprints import ( + canonical_title_text_for_dedup, + canonical_title_tokens_for_dedup, + normalize_title, +) logger = logging.getLogger(__name__) +NEAR_DUP_DEFAULT_SIMILARITY_THRESHOLD = 0.78 +NEAR_DUP_DEFAULT_CONTAINMENT_THRESHOLD = 0.92 +NEAR_DUP_DEFAULT_MIN_SHARED_TOKENS = 3 +NEAR_DUP_DEFAULT_MAX_YEAR_DELTA = 1 +NEAR_DUP_MIN_TOKEN_LENGTH = 3 +NEAR_DUP_CLUSTER_KEY_LENGTH = 16 +NEAR_DUP_STOPWORDS = { + "a", + "an", + "and", + "approach", + "for", + "in", + "method", + "of", + "on", + "the", + "to", + "using", + "via", + "with", +} + + +@dataclass(frozen=True) +class NearDuplicateMember: + publication_id: int + title: str + year: int | None + citation_count: int + + +@dataclass(frozen=True) +class NearDuplicateCluster: + cluster_key: str + winner_publication_id: int + similarity_score: float + members: tuple[NearDuplicateMember, ...] + + +@dataclass(frozen=True) +class _NearDuplicateCandidate: + publication_id: int + title: str + year: int | None + citation_count: int + canonical_text: str + tokens: frozenset[str] + async def find_identifier_duplicate_pairs( db_session: AsyncSession, ) -> list[tuple[int, int]]: - """Return (winner_id, dup_id) pairs where two publications share the same identifier. - - Winner is always the lower publication_id (earlier-created). Uses the existing - ix_publication_identifiers_kind_value index for the self-join. - """ + """Return (winner_id, dup_id) pairs where two publications share the same identifier.""" pi1 = aliased(PublicationIdentifier, name="pi1") pi2 = aliased(PublicationIdentifier, name="pi2") rows = await db_session.execute( @@ -40,11 +93,17 @@ async def merge_duplicate_publication( winner_id: int, dup_id: int, ) -> None: - """Merge dup_id into winner_id: migrate scholar links, then delete the dup.""" + """Merge dup_id into winner_id: migrate metadata/links/identifiers, then delete dup.""" + if winner_id == dup_id: + raise ValueError("winner_id and dup_id must differ.") + winner = await _load_publication(db_session, publication_id=winner_id) + dup = await _load_publication(db_session, publication_id=dup_id) + if winner is None or dup is None: + raise ValueError("winner_id and dup_id must both exist.") + _merge_publication_metadata(winner=winner, dup=dup) await _migrate_scholar_links(db_session, winner_id=winner_id, dup_id=dup_id) - await db_session.execute( - delete(Publication).where(Publication.id == dup_id) - ) + await _migrate_identifiers(db_session, winner_id=winner_id, dup_id=dup_id) + await db_session.execute(delete(Publication).where(Publication.id == dup_id)) logger.info( "publications.identifier_merge", extra={ @@ -55,6 +114,45 @@ async def merge_duplicate_publication( ) +async def _load_publication( + db_session: AsyncSession, + *, + publication_id: int, +) -> Publication | None: + result = await db_session.execute( + select(Publication).where(Publication.id == publication_id) + ) + return result.scalar_one_or_none() + + +def _merge_publication_metadata(*, winner: Publication, dup: Publication) -> None: + if winner.year is None and dup.year is not None: + winner.year = dup.year + winner.citation_count = max(int(winner.citation_count or 0), int(dup.citation_count or 0)) + if not winner.author_text and dup.author_text: + winner.author_text = dup.author_text + if not winner.venue_text and dup.venue_text: + winner.venue_text = dup.venue_text + if not winner.pub_url and dup.pub_url: + winner.pub_url = dup.pub_url + if not winner.pdf_url and dup.pdf_url: + winner.pdf_url = dup.pdf_url + if not winner.cluster_id and dup.cluster_id: + winner.cluster_id = dup.cluster_id + if not winner.canonical_title_hash and dup.canonical_title_hash: + winner.canonical_title_hash = dup.canonical_title_hash + winner.title_raw = _preferred_title_text(winner=winner.title_raw, dup=dup.title_raw) + winner.title_normalized = normalize_title(winner.title_raw) + + +def _preferred_title_text(*, winner: str, dup: str) -> str: + winner_score = len(canonical_title_text_for_dedup(winner)) + dup_score = len(canonical_title_text_for_dedup(dup)) + if dup_score > winner_score: + return dup + return winner + + async def _migrate_scholar_links( db_session: AsyncSession, *, @@ -81,17 +179,64 @@ async def _migrate_scholar_links( link.publication_id = winner_id -async def sweep_identifier_duplicates(db_session: AsyncSession) -> int: - """Find publications sharing an identifier and merge duplicates into the winner. +async def _migrate_identifiers( + db_session: AsyncSession, + *, + winner_id: int, + dup_id: int, +) -> None: + result = await db_session.execute( + select(PublicationIdentifier).where(PublicationIdentifier.publication_id == dup_id) + ) + dup_identifiers = result.scalars().all() + for identifier in dup_identifiers: + existing = await _find_identifier( + db_session, + publication_id=winner_id, + kind=identifier.kind, + value_normalized=identifier.value_normalized, + ) + if existing is None: + identifier.publication_id = winner_id + continue + _merge_identifier(existing=existing, dup=identifier) + await db_session.delete(identifier) - Returns the number of duplicate publications removed. - """ + +async def _find_identifier( + db_session: AsyncSession, + *, + publication_id: int, + kind: str, + value_normalized: str, +) -> PublicationIdentifier | None: + result = await db_session.execute( + select(PublicationIdentifier).where( + PublicationIdentifier.publication_id == publication_id, + PublicationIdentifier.kind == kind, + PublicationIdentifier.value_normalized == value_normalized, + ) + ) + return result.scalar_one_or_none() + + +def _merge_identifier(*, existing: PublicationIdentifier, dup: PublicationIdentifier) -> None: + existing.confidence_score = max( + float(existing.confidence_score), + float(dup.confidence_score), + ) + if not existing.evidence_url and dup.evidence_url: + existing.evidence_url = dup.evidence_url + if not existing.value_raw and dup.value_raw: + existing.value_raw = dup.value_raw + + +async def sweep_identifier_duplicates(db_session: AsyncSession) -> int: + """Find publications sharing an identifier and merge duplicates into the winner.""" pairs = await find_identifier_duplicate_pairs(db_session) if not pairs: return 0 - # Deduplicate the pairs — a dup may appear multiple times if it shares - # several identifiers with the winner; process each dup only once. processed_dups: set[int] = set() for winner_id, dup_id in pairs: if dup_id in processed_dups: @@ -101,3 +246,271 @@ async def sweep_identifier_duplicates(db_session: AsyncSession) -> int: await db_session.flush() return len(processed_dups) + + +async def find_near_duplicate_clusters( + db_session: AsyncSession, + *, + similarity_threshold: float = NEAR_DUP_DEFAULT_SIMILARITY_THRESHOLD, + min_shared_tokens: int = NEAR_DUP_DEFAULT_MIN_SHARED_TOKENS, + max_year_delta: int = NEAR_DUP_DEFAULT_MAX_YEAR_DELTA, +) -> list[NearDuplicateCluster]: + candidates = await _load_near_duplicate_candidates(db_session) + if len(candidates) < 2: + return [] + groups = _cluster_candidate_groups( + candidates, + similarity_threshold=similarity_threshold, + min_shared_tokens=min_shared_tokens, + max_year_delta=max_year_delta, + ) + clusters = [_near_duplicate_cluster(group) for group in groups] + return sorted(clusters, key=lambda item: (-len(item.members), item.winner_publication_id)) + + +async def merge_near_duplicate_cluster( + db_session: AsyncSession, + *, + cluster: NearDuplicateCluster, +) -> int: + winner_id = int(cluster.winner_publication_id) + merged = 0 + for member in cluster.members: + if int(member.publication_id) == winner_id: + continue + await merge_duplicate_publication( + db_session, + winner_id=winner_id, + dup_id=int(member.publication_id), + ) + merged += 1 + return merged + + +def near_duplicate_cluster_payload(cluster: NearDuplicateCluster) -> dict[str, object]: + members = [ + { + "publication_id": int(member.publication_id), + "title": member.title, + "year": member.year, + "citation_count": int(member.citation_count), + } + for member in cluster.members + ] + return { + "cluster_key": cluster.cluster_key, + "winner_publication_id": int(cluster.winner_publication_id), + "member_count": len(cluster.members), + "similarity_score": float(cluster.similarity_score), + "members": members, + } + + +async def _load_near_duplicate_candidates( + db_session: AsyncSession, +) -> list[_NearDuplicateCandidate]: + result = await db_session.execute( + select( + Publication.id, + Publication.title_raw, + Publication.year, + Publication.citation_count, + ) + ) + records = [ + _candidate_from_row( + publication_id=int(publication_id), + title=str(title_raw or ""), + year=year, + citation_count=int(citation_count or 0), + ) + for publication_id, title_raw, year, citation_count in result.all() + ] + return [record for record in records if record is not None] + + +def _candidate_from_row( + *, + publication_id: int, + title: str, + year: int | None, + citation_count: int, +) -> _NearDuplicateCandidate | None: + canonical = canonical_title_text_for_dedup(title) + raw_tokens = canonical_title_tokens_for_dedup(title) + tokens = _normalized_tokens(raw_tokens) + if not canonical or not tokens: + return None + return _NearDuplicateCandidate( + publication_id=publication_id, + title=title, + year=year, + citation_count=citation_count, + canonical_text=canonical, + tokens=frozenset(tokens), + ) + + +def _normalized_tokens(tokens: Iterable[str]) -> set[str]: + return { + token + for token in tokens + if len(token) >= NEAR_DUP_MIN_TOKEN_LENGTH and token not in NEAR_DUP_STOPWORDS + } + + +def _cluster_candidate_groups( + candidates: list[_NearDuplicateCandidate], + *, + similarity_threshold: float, + min_shared_tokens: int, + max_year_delta: int, +) -> list[list[_NearDuplicateCandidate]]: + by_id = {candidate.publication_id: candidate for candidate in candidates} + token_index = _candidate_token_index(candidates) + parent = {candidate.publication_id: candidate.publication_id for candidate in candidates} + for candidate in candidates: + peers = _candidate_peer_ids(candidate=candidate, token_index=token_index) + for peer_id in sorted(peers): + if peer_id <= candidate.publication_id: + continue + peer = by_id[peer_id] + if _is_near_duplicate_pair( + candidate, + peer, + similarity_threshold=similarity_threshold, + min_shared_tokens=min_shared_tokens, + max_year_delta=max_year_delta, + ): + _union(parent, candidate.publication_id, peer_id) + return _grouped_candidates(candidates, parent) + + +def _candidate_token_index( + candidates: list[_NearDuplicateCandidate], +) -> dict[str, set[int]]: + index: dict[str, set[int]] = {} + for candidate in candidates: + for token in candidate.tokens: + index.setdefault(token, set()).add(candidate.publication_id) + return index + + +def _candidate_peer_ids( + *, + candidate: _NearDuplicateCandidate, + token_index: dict[str, set[int]], +) -> set[int]: + peers: set[int] = set() + for token in candidate.tokens: + peers.update(token_index.get(token, set())) + peers.discard(candidate.publication_id) + return peers + + +def _is_near_duplicate_pair( + left: _NearDuplicateCandidate, + right: _NearDuplicateCandidate, + *, + similarity_threshold: float, + min_shared_tokens: int, + max_year_delta: int, +) -> bool: + if left.canonical_text == right.canonical_text: + return True + if not _years_compatible(left.year, right.year, max_year_delta=max_year_delta): + return False + shared_tokens = len(left.tokens & right.tokens) + if shared_tokens < min_shared_tokens: + return False + jaccard = _jaccard(left.tokens, right.tokens) + containment = shared_tokens / max(1, min(len(left.tokens), len(right.tokens))) + return jaccard >= similarity_threshold or containment >= NEAR_DUP_DEFAULT_CONTAINMENT_THRESHOLD + + +def _years_compatible(left: int | None, right: int | None, *, max_year_delta: int) -> bool: + if left is None or right is None: + return True + return abs(int(left) - int(right)) <= int(max_year_delta) + + +def _jaccard(left: frozenset[str], right: frozenset[str]) -> float: + if not left or not right: + return 0.0 + return len(left & right) / len(left | right) + + +def _find_root(parent: dict[int, int], value: int) -> int: + root = parent[value] + while root != parent[root]: + root = parent[root] + while value != root: + next_value = parent[value] + parent[value] = root + value = next_value + return root + + +def _union(parent: dict[int, int], left: int, right: int) -> None: + left_root = _find_root(parent, left) + right_root = _find_root(parent, right) + if left_root == right_root: + return + if left_root < right_root: + parent[right_root] = left_root + return + parent[left_root] = right_root + + +def _grouped_candidates( + candidates: list[_NearDuplicateCandidate], + parent: dict[int, int], +) -> list[list[_NearDuplicateCandidate]]: + groups: dict[int, list[_NearDuplicateCandidate]] = {} + for candidate in candidates: + root = _find_root(parent, candidate.publication_id) + groups.setdefault(root, []).append(candidate) + clustered = [members for members in groups.values() if len(members) > 1] + for members in clustered: + members.sort(key=lambda item: item.publication_id) + return clustered + + +def _near_duplicate_cluster(members: list[_NearDuplicateCandidate]) -> NearDuplicateCluster: + winner = _winner_candidate(members) + member_ids = [member.publication_id for member in members] + joined = ",".join(str(publication_id) for publication_id in member_ids) + cluster_key = hashlib.sha256(joined.encode("utf-8")).hexdigest()[:NEAR_DUP_CLUSTER_KEY_LENGTH] + similarity_score = _cluster_similarity_score(members) + return NearDuplicateCluster( + cluster_key=cluster_key, + winner_publication_id=winner.publication_id, + similarity_score=similarity_score, + members=tuple( + NearDuplicateMember( + publication_id=member.publication_id, + title=member.title, + year=member.year, + citation_count=member.citation_count, + ) + for member in members + ), + ) + + +def _winner_candidate(members: list[_NearDuplicateCandidate]) -> _NearDuplicateCandidate: + return min( + members, + key=lambda member: (-int(member.citation_count), member.publication_id), + ) + + +def _cluster_similarity_score(members: list[_NearDuplicateCandidate]) -> float: + best = 0.0 + for index, left in enumerate(members): + for right in members[index + 1 :]: + shared_tokens = len(left.tokens & right.tokens) + jaccard = _jaccard(left.tokens, right.tokens) + containment = shared_tokens / max(1, min(len(left.tokens), len(right.tokens))) + best = max(best, jaccard, containment) + return round(best, 4) diff --git a/app/services/domains/publications/pdf_queue.py b/app/services/domains/publications/pdf_queue.py index 7503138..3cb0de4 100644 --- a/app/services/domains/publications/pdf_queue.py +++ b/app/services/domains/publications/pdf_queue.py @@ -370,16 +370,18 @@ async def _fetch_outcome_for_row( row: PublicationListItem, request_email: str | None, openalex_api_key: str | None = None, -) -> OaResolutionOutcome: + allow_arxiv_lookup: bool = True, +) -> tuple[OaResolutionOutcome, bool]: pipeline_result = await resolve_publication_pdf_outcome_for_row( row=row, request_email=request_email, openalex_api_key=openalex_api_key, + allow_arxiv_lookup=allow_arxiv_lookup, ) outcome = pipeline_result.outcome if outcome is not None: - return outcome - return _failed_outcome(row=row) + return outcome, bool(pipeline_result.arxiv_rate_limited) + return _failed_outcome(row=row), bool(pipeline_result.arxiv_rate_limited) def _apply_publication_update( @@ -450,17 +452,19 @@ async def _resolve_publication_row( request_email: str | None, row: PublicationListItem, openalex_api_key: str | None = None, -) -> None: + allow_arxiv_lookup: bool = True, +) -> bool: from app.services.domains.openalex.client import OpenAlexBudgetExhaustedError - from app.services.domains.arxiv.application import ArxivRateLimitError + await _mark_attempt_started(publication_id=row.publication_id, user_id=user_id) try: - outcome = await _fetch_outcome_for_row( + outcome, arxiv_rate_limited = await _fetch_outcome_for_row( row=row, request_email=request_email, openalex_api_key=openalex_api_key, + allow_arxiv_lookup=allow_arxiv_lookup, ) - except (OpenAlexBudgetExhaustedError, ArxivRateLimitError): + except OpenAlexBudgetExhaustedError: # Persist a terminal outcome so jobs do not remain stuck in "running". await _persist_outcome( publication_id=row.publication_id, @@ -479,11 +483,13 @@ async def _resolve_publication_row( }, ) outcome = _failed_outcome(row=row) + arxiv_rate_limited = False await _persist_outcome( publication_id=row.publication_id, user_id=user_id, outcome=outcome, ) + return bool(arxiv_rate_limited) async def _run_resolution_task( @@ -493,7 +499,6 @@ async def _run_resolution_task( rows: list[PublicationListItem], ) -> None: from app.services.domains.openalex.client import OpenAlexBudgetExhaustedError - from app.services.domains.arxiv.application import ArxivRateLimitError from app.services.domains.settings import application as user_settings_service # Resolve the best available API key: per-user setting → env var fallback. @@ -506,14 +511,25 @@ async def _run_resolution_task( except Exception: openalex_api_key = settings.openalex_api_key + arxiv_lookup_allowed = True for row in rows: try: - await _resolve_publication_row( + arxiv_rate_limited = await _resolve_publication_row( user_id=user_id, request_email=request_email, row=row, openalex_api_key=openalex_api_key, + allow_arxiv_lookup=arxiv_lookup_allowed, ) + if arxiv_rate_limited and arxiv_lookup_allowed: + arxiv_lookup_allowed = False + logger.warning( + "publications.pdf_queue.arxiv_batch_disabled", + extra={ + "event": "publications.pdf_queue.arxiv_batch_disabled", + "detail": "arXiv temporarily disabled for remaining batch after rate limit", + }, + ) except OpenAlexBudgetExhaustedError: logger.warning( "publications.pdf_queue.budget_exhausted", @@ -521,15 +537,6 @@ async def _run_resolution_task( "detail": "Stopping PDF resolution batch — OpenAlex daily budget exhausted"}, ) break - except ArxivRateLimitError: - logger.warning( - "publications.pdf_queue.arxiv_rate_limited", - extra={ - "event": "publications.pdf_queue.arxiv_rate_limited", - "detail": "Stopping PDF resolution batch — arXiv rate limit hit (429)", - }, - ) - break def _schedule_rows( diff --git a/app/services/domains/publications/pdf_resolution_pipeline.py b/app/services/domains/publications/pdf_resolution_pipeline.py index 0dc133c..7ad1f8d 100644 --- a/app/services/domains/publications/pdf_resolution_pipeline.py +++ b/app/services/domains/publications/pdf_resolution_pipeline.py @@ -5,6 +5,7 @@ import logging from typing import Any from app.services.domains.arxiv.application import ArxivRateLimitError +from app.services.domains.arxiv.guards import arxiv_skip_reason_for_item from app.services.domains.openalex.client import OpenAlexBudgetExhaustedError from app.services.domains.publications.types import PublicationListItem from app.services.domains.unpaywall.application import OaResolutionOutcome, resolve_publication_oa_outcomes @@ -17,6 +18,7 @@ logger = logging.getLogger(__name__) class PipelineOutcome: outcome: OaResolutionOutcome | None scholar_candidates: Any | None # Kept for backward compatibility with calling signatures + arxiv_rate_limited: bool = False async def resolve_publication_pdf_outcome_for_row( @@ -24,6 +26,7 @@ async def resolve_publication_pdf_outcome_for_row( row: PublicationListItem, request_email: str | None, openalex_api_key: str | None = None, + allow_arxiv_lookup: bool = True, ) -> PipelineOutcome: # 1. OpenAlex OA — raises OpenAlexBudgetExhaustedError if budget is gone openalex_outcome = await _openalex_outcome(row, request_email=request_email, openalex_api_key=openalex_api_key) @@ -31,13 +34,29 @@ async def resolve_publication_pdf_outcome_for_row( return PipelineOutcome(openalex_outcome, None) # 2. arXiv - arxiv_outcome = await _arxiv_outcome(row, request_email=request_email) + arxiv_rate_limited = False + try: + arxiv_outcome = await _arxiv_outcome( + row, + request_email=request_email, + allow_lookup=allow_arxiv_lookup, + ) + except ArxivRateLimitError: + arxiv_rate_limited = True + arxiv_outcome = None + logger.warning( + "publications.pdf_resolution.arxiv_rate_limited", + extra={ + "event": "publications.pdf_resolution.arxiv_rate_limited", + "publication_id": int(row.publication_id), + }, + ) if arxiv_outcome and arxiv_outcome.pdf_url: - return PipelineOutcome(arxiv_outcome, None) + return PipelineOutcome(arxiv_outcome, None, arxiv_rate_limited=arxiv_rate_limited) # 3. Unpaywall (which falls back to Crossref) oa_outcome = await _oa_outcome(row=row, request_email=request_email) - return PipelineOutcome(oa_outcome, None) + return PipelineOutcome(oa_outcome, None, arxiv_rate_limited=arxiv_rate_limited) async def _openalex_outcome( @@ -87,9 +106,23 @@ async def _openalex_outcome( return None -async def _arxiv_outcome(row: PublicationListItem, request_email: str | None) -> OaResolutionOutcome | None: +async def _arxiv_outcome( + row: PublicationListItem, + *, + request_email: str | None, + allow_lookup: bool = True, +) -> OaResolutionOutcome | None: from app.services.domains.arxiv.application import discover_arxiv_id_for_publication + if not allow_lookup: + _log_arxiv_skip(row=row, skip_reason="batch_arxiv_cooldown_active") + return None + + skip_reason = arxiv_skip_reason_for_item(item=row) + if skip_reason is not None: + _log_arxiv_skip(row=row, skip_reason=skip_reason) + return None + try: arxiv_id = await discover_arxiv_id_for_publication(item=row, request_email=request_email) if arxiv_id: @@ -103,7 +136,7 @@ async def _arxiv_outcome(row: PublicationListItem, request_email: str | None) -> used_crossref=False, ) except ArxivRateLimitError: - raise # propagate so the batch loop can stop + raise # propagate so orchestration can switch to non-arXiv fallback except Exception as exc: logger.warning( "publications.pdf_resolution.arxiv_failed", @@ -112,6 +145,17 @@ async def _arxiv_outcome(row: PublicationListItem, request_email: str | None) -> return None +def _log_arxiv_skip(*, row: PublicationListItem, skip_reason: str) -> None: + logger.info( + "publications.pdf_resolution.arxiv_skipped", + extra={ + "event": "publications.pdf_resolution.arxiv_skipped", + "publication_id": int(row.publication_id), + "skip_reason": skip_reason, + }, + ) + + async def _oa_outcome( *, row: PublicationListItem, diff --git a/app/services/domains/publications/queries.py b/app/services/domains/publications/queries.py index 6f57cab..d642a73 100644 --- a/app/services/domains/publications/queries.py +++ b/app/services/domains/publications/queries.py @@ -2,10 +2,17 @@ from __future__ import annotations from datetime import datetime -from sqlalchemy import Select, func, select +from sqlalchemy import Select, case, func, select from sqlalchemy.ext.asyncio import AsyncSession -from app.db.models import CrawlRun, Publication, RunStatus, ScholarProfile, ScholarPublication +from app.db.models import ( + CrawlRun, + Publication, + PublicationPdfJob, + RunStatus, + ScholarProfile, + ScholarPublication, +) from app.services.domains.publications.modes import MODE_LATEST, MODE_UNREAD from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem @@ -17,6 +24,29 @@ def _normalized_citation_count(value: object) -> int: return 0 +def _pdf_status_sort_rank(): + return case( + (Publication.pdf_url.is_not(None), 4), + (PublicationPdfJob.status == "resolved", 4), + (PublicationPdfJob.status == "running", 3), + (PublicationPdfJob.status == "queued", 2), + (PublicationPdfJob.status == "failed", 0), + else_=1, + ) + + +def _sort_column(sort_by: str): + sort_columns = { + "first_seen": ScholarPublication.created_at, + "title": Publication.title_raw, + "year": Publication.year, + "citations": Publication.citation_count, + "scholar": ScholarProfile.display_name, + "pdf_status": _pdf_status_sort_rank(), + } + return sort_columns.get(sort_by, ScholarPublication.created_at) + + async def get_latest_run_id_for_user( db_session: AsyncSession, *, @@ -55,14 +85,6 @@ def publications_query( sort_dir: str = "desc", snapshot_before: datetime | None = None, ) -> Select[tuple]: - _SORT_COLUMNS = { - "first_seen": ScholarPublication.created_at, - "title": Publication.title_raw, - "year": Publication.year, - "citations": Publication.citation_count, - "scholar": ScholarProfile.display_name, - } - scholar_label = ScholarProfile.display_name stmt = ( select( @@ -83,6 +105,7 @@ def publications_query( ) .join(ScholarPublication, ScholarPublication.publication_id == Publication.id) .join(ScholarProfile, ScholarProfile.id == ScholarPublication.scholar_profile_id) + .outerjoin(PublicationPdfJob, PublicationPdfJob.publication_id == Publication.id) .where(ScholarProfile.user_id == user_id) ) if search: @@ -106,7 +129,7 @@ def publications_query( if snapshot_before is not None: stmt = stmt.where(ScholarPublication.created_at <= snapshot_before) - sort_col = _SORT_COLUMNS.get(sort_by, ScholarPublication.created_at) + sort_col = _sort_column(sort_by) order = sort_col.desc() if sort_dir == "desc" else sort_col.asc() stmt = stmt.order_by(order, Publication.id.desc()) diff --git a/app/services/domains/scholar/rate_limit.py b/app/services/domains/scholar/rate_limit.py index 947c91f..fa708b6 100644 --- a/app/services/domains/scholar/rate_limit.py +++ b/app/services/domains/scholar/rate_limit.py @@ -7,12 +7,28 @@ _REQUEST_LOCK = asyncio.Lock() _LAST_REQUEST_AT = 0.0 +def _normalize_interval_seconds(value: float) -> float: + return max(float(value), 0.0) + + +def remaining_scholar_slot_seconds(*, min_interval_seconds: float) -> float: + interval_seconds = _normalize_interval_seconds(min_interval_seconds) + if interval_seconds <= 0: + return 0.0 + elapsed_seconds = time.monotonic() - _LAST_REQUEST_AT + return max(interval_seconds - elapsed_seconds, 0.0) + + async def wait_for_scholar_slot(*, min_interval_seconds: float) -> None: global _LAST_REQUEST_AT - interval = max(float(min_interval_seconds), 0.0) + interval = _normalize_interval_seconds(min_interval_seconds) async with _REQUEST_LOCK: - elapsed = time.monotonic() - _LAST_REQUEST_AT - remaining = interval - elapsed + remaining = remaining_scholar_slot_seconds(min_interval_seconds=interval) if remaining > 0: await asyncio.sleep(remaining) _LAST_REQUEST_AT = time.monotonic() + + +def reset_scholar_rate_limit_state_for_tests() -> None: + global _LAST_REQUEST_AT + _LAST_REQUEST_AT = 0.0 diff --git a/app/services/domains/scholar/source.py b/app/services/domains/scholar/source.py index 92c6702..cf2ea0a 100644 --- a/app/services/domains/scholar/source.py +++ b/app/services/domains/scholar/source.py @@ -9,6 +9,9 @@ from urllib.error import HTTPError, URLError from urllib.parse import urlencode from urllib.request import Request, urlopen +from app.services.domains.scholar import rate_limit as scholar_rate_limit +from app.settings import settings + SCHOLAR_PROFILE_URL = "https://scholar.google.com/citations" DEFAULT_PAGE_SIZE = 100 @@ -66,9 +69,16 @@ class LiveScholarSource: self, *, timeout_seconds: float = 25.0, + min_interval_seconds: float | None = None, user_agents: list[str] | None = None, ) -> None: self._timeout_seconds = timeout_seconds + configured_interval = ( + float(settings.ingestion_min_request_delay_seconds) + if min_interval_seconds is None + else float(min_interval_seconds) + ) + self._min_interval_seconds = max(configured_interval, 0.0) self._user_agents = user_agents or DEFAULT_USER_AGENTS async def fetch_profile_html(self, scholar_id: str) -> FetchResult: @@ -100,7 +110,7 @@ class LiveScholarSource: "pagesize": pagesize, }, ) - return await asyncio.to_thread(self._fetch_sync, requested_url) + return await self._fetch_with_global_throttle(requested_url) async def fetch_author_search_html( self, @@ -121,7 +131,7 @@ class LiveScholarSource: "start": start, }, ) - return await asyncio.to_thread(self._fetch_sync, requested_url) + return await self._fetch_with_global_throttle(requested_url) async def fetch_publication_html(self, publication_url: str) -> FetchResult: logger.debug( @@ -131,7 +141,13 @@ class LiveScholarSource: "requested_url": publication_url, }, ) - return await asyncio.to_thread(self._fetch_sync, publication_url) + return await self._fetch_with_global_throttle(publication_url) + + async def _fetch_with_global_throttle(self, requested_url: str) -> FetchResult: + await scholar_rate_limit.wait_for_scholar_slot( + min_interval_seconds=self._min_interval_seconds, + ) + return await asyncio.to_thread(self._fetch_sync, requested_url) def _build_request(self, requested_url: str) -> Request: return Request( diff --git a/app/settings.py b/app/settings.py index 15b0c19..51ebcf4 100644 --- a/app/settings.py +++ b/app/settings.py @@ -253,6 +253,14 @@ class Settings: unpaywall_pdf_discovery_enabled: bool = _env_bool("UNPAYWALL_PDF_DISCOVERY_ENABLED", True) unpaywall_pdf_discovery_max_candidates: int = _env_int("UNPAYWALL_PDF_DISCOVERY_MAX_CANDIDATES", 5) unpaywall_pdf_discovery_max_html_bytes: int = _env_int("UNPAYWALL_PDF_DISCOVERY_MAX_HTML_BYTES", 500_000) + arxiv_enabled: bool = _env_bool("ARXIV_ENABLED", True) + arxiv_timeout_seconds: float = _env_float("ARXIV_TIMEOUT_SECONDS", 3.0) + arxiv_min_interval_seconds: float = _env_float("ARXIV_MIN_INTERVAL_SECONDS", 4.0) + arxiv_rate_limit_cooldown_seconds: float = _env_float("ARXIV_RATE_LIMIT_COOLDOWN_SECONDS", 60.0) + arxiv_default_max_results: int = _env_int("ARXIV_DEFAULT_MAX_RESULTS", 3) + arxiv_cache_ttl_seconds: float = _env_float("ARXIV_CACHE_TTL_SECONDS", 900.0) + arxiv_cache_max_entries: int = _env_int("ARXIV_CACHE_MAX_ENTRIES", 512) + arxiv_mailto: str = _env_str("ARXIV_MAILTO", "") 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) diff --git a/docs/developer/architecture.md b/docs/developer/architecture.md index afa6b05..4330830 100644 --- a/docs/developer/architecture.md +++ b/docs/developer/architecture.md @@ -14,6 +14,7 @@ Canonical business logic belongs in `app/services/domains/*`. - `app/services/domains/scholar/*`: fail-fast scholar parsing and source fetch adapters. - `app/services/domains/scholars/*`: scholar CRUD, profile image, and name-search controls. - `app/services/domains/publications/*`: listing/read-state, favorite toggles, enrichment scheduling, and retry paths. +- `app/services/domains/arxiv/*`: typed API client, global DB-backed throttle, query cache, and in-flight request coalescing. - `app/services/domains/crossref/*` + `app/services/domains/unpaywall/*`: DOI/OA enrichment with bounded pacing. - `app/services/domains/runs/*`: run history and continuation queue operations. - `app/services/domains/portability/*`: import/export workflows. @@ -49,3 +50,9 @@ graph TD ### Identifier Engine Philosophy The platform previously treated the `DOI` as a hardcoded 1:1 property of a publication. It now utilizes a decoupled *Identifier Gathering* module (`PublicationIdentifier` table). A single publication can have multiple identifiers (ex. `doi`, `arxiv`, `pmid`, `pmcid`). This creates high resilience when integrating with external APIs, allowing systems like Unpaywall to be fed explicitly with high-confidence DOIs, rather than relying on unstructured search heuristics. + +### arXiv Safety and Efficiency +- arXiv requests are globally serialized via a PostgreSQL advisory lock and shared runtime row (`arxiv_runtime_state`). +- identical request payloads are fingerprinted and cached in `arxiv_query_cache_entries` with TTL + optional max-entry pruning. +- concurrent identical misses are coalesced in-process, so one outbound call serves all waiters. +- structured logs provide auditability for scheduling/completion/cooldown/cache behavior. diff --git a/docs/developer/ingestion.md b/docs/developer/ingestion.md index b906f15..26ee8e7 100644 --- a/docs/developer/ingestion.md +++ b/docs/developer/ingestion.md @@ -24,3 +24,15 @@ Once a publication is built, the `gather_identifiers_for_publication` module iso - `crossref.restful` APIs (Queries by Title and Author strings). These identifiers are accumulated in `publication_identifiers` instead of being bound as hard-coded properties, maximizing matching resilience in the automated Unpaywall PDF acquisition stage. + +## arXiv Request Controls +- **Global throttle state**: arXiv calls share `arxiv_runtime_state` so all workers respect one cooldown/interval clock. +- **Query cache**: identical request parameters map to a stable fingerprint and are stored in `arxiv_query_cache_entries`. +- **In-flight coalescing**: duplicate concurrent misses join one outbound request instead of fan-out. +- **Caller load-shedding**: arXiv lookups are skipped when high-confidence DOI/arXiv evidence already exists, or when title quality is below threshold. + +## arXiv Observability Events +- `arxiv.request_scheduled`: emitted before a gated request; includes `wait_seconds`, `cooldown_remaining_seconds`, `source_path`. +- `arxiv.request_completed`: emitted after response; includes `status_code`, `wait_seconds`, `cooldown_remaining_seconds`, `source_path`. +- `arxiv.cooldown_activated`: emitted when status `429` triggers cooldown. +- `arxiv.cache_hit` / `arxiv.cache_miss`: emitted on query cache lookup with `source_path`. diff --git a/docs/operations/arxiv-runbook.md b/docs/operations/arxiv-runbook.md new file mode 100644 index 0000000..b3183be --- /dev/null +++ b/docs/operations/arxiv-runbook.md @@ -0,0 +1,46 @@ +# arXiv Operations Runbook + +Use this runbook when arXiv lookups are slow, rate-limited, or behaving unexpectedly. + +## Signals To Check +- logs for `arxiv.request_scheduled`, `arxiv.request_completed`, `arxiv.cooldown_activated`, `arxiv.cache_hit`, `arxiv.cache_miss` +- `arxiv_runtime_state` row for cooldown and next-allowed timestamps +- `arxiv_query_cache_entries` size and expiry churn + +## Event Field Guide +- `wait_seconds`: enforced pre-request delay from the global limiter. +- `status_code`: upstream response code from arXiv. +- `cooldown_remaining_seconds`: remaining cooldown when blocked or after 429. +- `source_path`: caller path (`search` or `lookup_ids`). + +## Quick SQL Checks +```sql +SELECT state_key, next_allowed_at, cooldown_until, updated_at +FROM arxiv_runtime_state; +``` + +```sql +SELECT count(*) AS cache_rows, min(expires_at) AS earliest_expiry, max(expires_at) AS latest_expiry +FROM arxiv_query_cache_entries; +``` + +## Common Scenarios +1. Repeated `arxiv.cooldown_activated` events: +- Confirm recent `429` statuses in `arxiv.request_completed`. +- Reduce caller pressure (check new title-quality/identifier guards are active). +- Temporarily raise `ARXIV_RATE_LIMIT_COOLDOWN_SECONDS` if upstream remains strict. + +2. High request latency with few completions: +- Inspect `wait_seconds` in `arxiv.request_scheduled`. +- Verify only one process path is repeatedly hitting arXiv (`source_path`). +- Confirm cache is enabled (`ARXIV_CACHE_TTL_SECONDS > 0`) and effective (`cache_hit` appears). + +3. Low cache effectiveness: +- Validate normalized query behavior and caller churn. +- Increase `ARXIV_CACHE_TTL_SECONDS` for stable workloads. +- Increase `ARXIV_CACHE_MAX_ENTRIES` if heavy eviction is observed. + +## Safe Recovery +1. Pause automated ingestion if rate-limit storms persist. +2. Let cooldown expire naturally; avoid manual burst retries. +3. Resume and monitor event rates before restoring full load. diff --git a/docs/operations/overview.md b/docs/operations/overview.md index 36cae41..f208a7d 100644 --- a/docs/operations/overview.md +++ b/docs/operations/overview.md @@ -3,5 +3,6 @@ Use this section for production operations and incident/runbook workflows. - [Scrape Safety Runbook](./scrape-safety-runbook.md) +- [arXiv Runbook](./arxiv-runbook.md) - [Database Runbook](./database-runbook.md) - [Migration Rollout Checklist](./migration-checklist.md) diff --git a/docs/reference/environment.md b/docs/reference/environment.md index 0dafb9b..75cc192 100644 --- a/docs/reference/environment.md +++ b/docs/reference/environment.md @@ -40,7 +40,7 @@ This document is the source-of-truth audit for what remains configurable versus ### Scheduler + Ingestion Safety -`SCHEDULER_ENABLED`, `SCHEDULER_TICK_SECONDS`, `SCHEDULER_QUEUE_BATCH_SIZE`, `INGESTION_AUTOMATION_ALLOWED`, `INGESTION_MANUAL_RUN_ALLOWED`, `INGESTION_MIN_RUN_INTERVAL_MINUTES`, `INGESTION_MIN_REQUEST_DELAY_SECONDS`, `INGESTION_NETWORK_ERROR_RETRIES`, `INGESTION_RETRY_BACKOFF_SECONDS`, `INGESTION_MAX_PAGES_PER_SCHOLAR`, `INGESTION_PAGE_SIZE`, `INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD`, `INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD`, `INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD`, `INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS`, `INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS`, `INGESTION_CONTINUATION_QUEUE_ENABLED`, `INGESTION_CONTINUATION_BASE_DELAY_SECONDS`, `INGESTION_CONTINUATION_MAX_DELAY_SECONDS`, `INGESTION_CONTINUATION_MAX_ATTEMPTS` +`SCHEDULER_ENABLED`, `SCHEDULER_TICK_SECONDS`, `SCHEDULER_QUEUE_BATCH_SIZE`, `SCHEDULER_PDF_QUEUE_BATCH_SIZE`, `INGESTION_AUTOMATION_ALLOWED`, `INGESTION_MANUAL_RUN_ALLOWED`, `INGESTION_MIN_RUN_INTERVAL_MINUTES`, `INGESTION_MIN_REQUEST_DELAY_SECONDS`, `INGESTION_NETWORK_ERROR_RETRIES`, `INGESTION_RETRY_BACKOFF_SECONDS`, `INGESTION_RATE_LIMIT_RETRIES`, `INGESTION_RATE_LIMIT_BACKOFF_SECONDS`, `INGESTION_MAX_PAGES_PER_SCHOLAR`, `INGESTION_PAGE_SIZE`, `INGESTION_ALERT_BLOCKED_FAILURE_THRESHOLD`, `INGESTION_ALERT_NETWORK_FAILURE_THRESHOLD`, `INGESTION_ALERT_RETRY_SCHEDULED_THRESHOLD`, `INGESTION_SAFETY_COOLDOWN_BLOCKED_SECONDS`, `INGESTION_SAFETY_COOLDOWN_NETWORK_SECONDS`, `INGESTION_CONTINUATION_QUEUE_ENABLED`, `INGESTION_CONTINUATION_BASE_DELAY_SECONDS`, `INGESTION_CONTINUATION_MAX_DELAY_SECONDS`, `INGESTION_CONTINUATION_MAX_ATTEMPTS` ### Scholar Images + Name Search Safety @@ -48,7 +48,7 @@ This document is the source-of-truth audit for what remains configurable versus ### OA Enrichment + PDF Resolution -`UNPAYWALL_ENABLED`, `UNPAYWALL_EMAIL`, `UNPAYWALL_TIMEOUT_SECONDS`, `UNPAYWALL_MIN_INTERVAL_SECONDS`, `UNPAYWALL_MAX_ITEMS_PER_REQUEST`, `UNPAYWALL_RETRY_COOLDOWN_SECONDS`, `UNPAYWALL_PDF_DISCOVERY_ENABLED`, `UNPAYWALL_PDF_DISCOVERY_MAX_CANDIDATES`, `UNPAYWALL_PDF_DISCOVERY_MAX_HTML_BYTES`, `PDF_AUTO_RETRY_INTERVAL_SECONDS`, `PDF_AUTO_RETRY_FIRST_INTERVAL_SECONDS`, `PDF_AUTO_RETRY_MAX_ATTEMPTS`, `CROSSREF_ENABLED`, `CROSSREF_MAX_ROWS`, `CROSSREF_TIMEOUT_SECONDS`, `CROSSREF_MIN_INTERVAL_SECONDS`, `CROSSREF_MAX_LOOKUPS_PER_REQUEST` +`UNPAYWALL_ENABLED`, `UNPAYWALL_EMAIL`, `UNPAYWALL_TIMEOUT_SECONDS`, `UNPAYWALL_MIN_INTERVAL_SECONDS`, `UNPAYWALL_MAX_ITEMS_PER_REQUEST`, `UNPAYWALL_RETRY_COOLDOWN_SECONDS`, `UNPAYWALL_PDF_DISCOVERY_ENABLED`, `UNPAYWALL_PDF_DISCOVERY_MAX_CANDIDATES`, `UNPAYWALL_PDF_DISCOVERY_MAX_HTML_BYTES`, `ARXIV_ENABLED`, `ARXIV_TIMEOUT_SECONDS`, `ARXIV_MIN_INTERVAL_SECONDS`, `ARXIV_RATE_LIMIT_COOLDOWN_SECONDS`, `ARXIV_DEFAULT_MAX_RESULTS`, `ARXIV_CACHE_TTL_SECONDS`, `ARXIV_CACHE_MAX_ENTRIES`, `ARXIV_MAILTO`, `PDF_AUTO_RETRY_INTERVAL_SECONDS`, `PDF_AUTO_RETRY_FIRST_INTERVAL_SECONDS`, `PDF_AUTO_RETRY_MAX_ATTEMPTS`, `CROSSREF_ENABLED`, `CROSSREF_MAX_ROWS`, `CROSSREF_TIMEOUT_SECONDS`, `CROSSREF_MIN_INTERVAL_SECONDS`, `CROSSREF_MAX_LOOKUPS_PER_REQUEST`, `OPENALEX_API_KEY`, `CROSSREF_API_TOKEN`, `CROSSREF_API_MAILTO` ### Startup Bootstrap + DB Wait diff --git a/frontend/src/components/ui/AppBrandMark.vue b/frontend/src/components/ui/AppBrandMark.vue index 0061ca4..f60e405 100644 --- a/frontend/src/components/ui/AppBrandMark.vue +++ b/frontend/src/components/ui/AppBrandMark.vue @@ -12,15 +12,15 @@ const props = withDefaults( const sizeClass = computed(() => { if (props.size === "sm") { - return "h-5 w-5"; + return "h-7 w-7"; } if (props.size === "lg") { - return "h-8 w-8"; + return "h-10 w-10"; } if (props.size === "xl") { - return "h-12 w-12"; + return "h-14 w-14"; } - return "h-6 w-6"; + return "h-7 w-7"; }); const logoMaskStyle: Record = { diff --git a/frontend/src/features/admin_dbops/index.ts b/frontend/src/features/admin_dbops/index.ts index 0b16b3a..aa4a312 100644 --- a/frontend/src/features/admin_dbops/index.ts +++ b/frontend/src/features/admin_dbops/index.ts @@ -95,6 +95,40 @@ export interface TriggerPublicationLinkRepairResult { summary: Record; } +export interface NearDuplicateClusterMember { + publication_id: number; + title: string; + year: number | null; + citation_count: number; +} + +export interface NearDuplicateCluster { + cluster_key: string; + winner_publication_id: number; + member_count: number; + similarity_score: number; + members: NearDuplicateClusterMember[]; +} + +export interface TriggerPublicationNearDuplicateRepairPayload { + dry_run?: boolean; + similarity_threshold?: number; + min_shared_tokens?: number; + max_year_delta?: number; + max_clusters?: number; + selected_cluster_keys?: string[]; + requested_by?: string; + confirmation_text?: string; +} + +export interface TriggerPublicationNearDuplicateRepairResult { + job_id: number; + status: string; + scope: Record; + summary: Record; + clusters: NearDuplicateCluster[]; +} + export async function getAdminDbIntegrityReport(): Promise { const response = await apiRequest("/admin/db/integrity", { method: "GET" }); return response.data; @@ -122,6 +156,19 @@ export async function triggerPublicationLinkRepair( return response.data; } +export async function triggerPublicationNearDuplicateRepair( + payload: TriggerPublicationNearDuplicateRepairPayload, +): Promise { + const response = await apiRequest( + "/admin/db/repairs/publication-near-duplicates", + { + method: "POST", + body: payload, + }, + ); + return response.data; +} + export async function listAdminPdfQueue( page = 1, pageSize = 100, diff --git a/frontend/src/features/publications/index.ts b/frontend/src/features/publications/index.ts index 3f80806..a689ba9 100644 --- a/frontend/src/features/publications/index.ts +++ b/frontend/src/features/publications/index.ts @@ -1,6 +1,13 @@ import { apiRequest } from "@/lib/api/client"; export type PublicationMode = "all" | "unread" | "latest"; +export type PublicationSortBy = + | "first_seen" + | "title" + | "year" + | "citations" + | "scholar" + | "pdf_status"; export interface DisplayIdentifier { kind: string; @@ -54,7 +61,7 @@ export interface PublicationsQuery { favoriteOnly?: boolean; scholarProfileId?: number; search?: string; - sortBy?: string; + sortBy?: PublicationSortBy; sortDir?: "asc" | "desc"; page?: number; pageSize?: number; diff --git a/frontend/src/features/settings/SettingsAdminPanel.vue b/frontend/src/features/settings/SettingsAdminPanel.vue index e5edfb4..9225ba5 100644 --- a/frontend/src/features/settings/SettingsAdminPanel.vue +++ b/frontend/src/features/settings/SettingsAdminPanel.vue @@ -20,11 +20,14 @@ import { listAdminPdfQueue, requeueAdminPdfLookup, requeueAllAdminPdfLookups, + triggerPublicationNearDuplicateRepair, triggerPublicationLinkRepair, type AdminDbIntegrityCheck, type AdminDbIntegrityReport, type AdminDbRepairJob, type AdminPdfQueueItem, + type NearDuplicateCluster, + type TriggerPublicationNearDuplicateRepairResult, type TriggerPublicationLinkRepairResult, } from "@/features/admin_dbops"; import { @@ -43,6 +46,7 @@ const SECTION_PDF = "pdf"; const SCOPE_SINGLE_USER = "single_user"; const SCOPE_ALL_USERS = "all_users"; const APPLY_ALL_USERS_CONFIRM_TEXT = "REPAIR ALL USERS"; +const APPLY_NEAR_DUPLICATES_CONFIRM_TEXT = "MERGE SELECTED DUPLICATES"; const DROP_PUBLICATIONS_CONFIRM_TEXT = "DROP ALL PUBLICATIONS"; type RepairScopeMode = typeof SCOPE_SINGLE_USER | typeof SCOPE_ALL_USERS; @@ -82,6 +86,17 @@ const repairRequestedBy = ref(""); const repairDryRun = ref(true); const repairGcOrphans = ref(false); const repairConfirmationText = ref(""); +const runningNearDuplicateScan = ref(false); +const applyingNearDuplicateRepair = ref(false); +const nearDuplicateRequestedBy = ref(""); +const nearDuplicateSimilarityThreshold = ref("0.78"); +const nearDuplicateMinSharedTokens = ref("3"); +const nearDuplicateMaxYearDelta = ref("1"); +const nearDuplicateMaxClusters = ref("25"); +const nearDuplicateConfirmationText = ref(""); +const nearDuplicateSelectedClusterKeys = ref>(new Set()); +const nearDuplicateClusters = ref([]); +const lastNearDuplicateResult = ref(null); const droppingPublications = ref(false); const dropConfirmationText = ref(""); @@ -105,6 +120,7 @@ const activeUser = computed(() => users.value.find((user) => user.id === activeU const typedConfirmationRequired = computed( () => repairScopeMode.value === SCOPE_ALL_USERS && !repairDryRun.value, ); +const nearDuplicateApplyEnabled = computed(() => nearDuplicateSelectedClusterKeys.value.size > 0); const pdfQueuePageSizeValue = computed(() => { const parsed = Number(pdfQueuePageSize.value); if (!Number.isFinite(parsed)) { @@ -213,6 +229,64 @@ function validateTypedConfirmation(): string { return normalized; } +function parseBoundedNumber( + raw: string, + options: { minimum: number; maximum: number; fallback: number }, +): number { + const { minimum, maximum, fallback } = options; + const parsed = Number(raw.trim()); + if (!Number.isFinite(parsed)) { + return fallback; + } + return Math.max(minimum, Math.min(maximum, parsed)); +} + +function nearDuplicatePayloadBase(): { + similarity_threshold: number; + min_shared_tokens: number; + max_year_delta: number; + max_clusters: number; + requested_by: string | undefined; +} { + return { + similarity_threshold: parseBoundedNumber(nearDuplicateSimilarityThreshold.value, { + minimum: 0.5, + maximum: 1.0, + fallback: 0.78, + }), + min_shared_tokens: Math.trunc( + parseBoundedNumber(nearDuplicateMinSharedTokens.value, { + minimum: 1, + maximum: 8, + fallback: 3, + }), + ), + max_year_delta: Math.trunc( + parseBoundedNumber(nearDuplicateMaxYearDelta.value, { + minimum: 0, + maximum: 5, + fallback: 1, + }), + ), + max_clusters: Math.trunc( + parseBoundedNumber(nearDuplicateMaxClusters.value, { + minimum: 1, + maximum: 200, + fallback: 25, + }), + ), + requested_by: nearDuplicateRequestedBy.value.trim() || undefined, + }; +} + +function selectedNearDuplicateKeys(): string[] { + return [...nearDuplicateSelectedClusterKeys.value].sort((left, right) => left.localeCompare(right)); +} + +function checkboxEventChecked(event: Event): boolean { + return event.target instanceof HTMLInputElement ? event.target.checked : false; +} + function summaryCount(job: AdminDbRepairJob, key: string): string { const value = job.summary[key]; return typeof value === "number" ? String(value) : "n/a"; @@ -384,6 +458,67 @@ async function onRunRepair(): Promise { } } +function onToggleNearDuplicateClusterSelection(clusterKey: string, checked: boolean): void { + const next = new Set(nearDuplicateSelectedClusterKeys.value); + if (checked) { + next.add(clusterKey); + } else { + next.delete(clusterKey); + } + nearDuplicateSelectedClusterKeys.value = next; +} + +async function onRunNearDuplicateScan(): Promise { + runningNearDuplicateScan.value = true; + clearAlerts(); + try { + const result = await triggerPublicationNearDuplicateRepair({ + dry_run: true, + ...nearDuplicatePayloadBase(), + }); + nearDuplicateClusters.value = result.clusters; + nearDuplicateSelectedClusterKeys.value = new Set(); + nearDuplicateConfirmationText.value = ""; + lastNearDuplicateResult.value = result; + successMessage.value = `Near-duplicate scan completed (job #${result.job_id}).`; + await refreshRepairJobs(); + } catch (error) { + assignError(error, "Unable to scan for near-duplicate publications."); + } finally { + runningNearDuplicateScan.value = false; + } +} + +async function onApplyNearDuplicateRepair(): Promise { + applyingNearDuplicateRepair.value = true; + clearAlerts(); + try { + const selectedKeys = selectedNearDuplicateKeys(); + if (selectedKeys.length === 0) { + throw new Error("Select at least one near-duplicate cluster before applying."); + } + if (nearDuplicateConfirmationText.value.trim() !== APPLY_NEAR_DUPLICATES_CONFIRM_TEXT) { + throw new Error(`Type '${APPLY_NEAR_DUPLICATES_CONFIRM_TEXT}' to confirm merge.`); + } + const result = await triggerPublicationNearDuplicateRepair({ + dry_run: false, + selected_cluster_keys: selectedKeys, + confirmation_text: nearDuplicateConfirmationText.value.trim(), + ...nearDuplicatePayloadBase(), + }); + nearDuplicateClusters.value = result.clusters; + nearDuplicateSelectedClusterKeys.value = new Set(); + nearDuplicateConfirmationText.value = ""; + lastNearDuplicateResult.value = result; + successMessage.value = `Merged selected near-duplicate clusters (job #${result.job_id}).`; + await refreshRepairJobs(); + } catch (error) { + assignError(error, "Unable to apply near-duplicate merge."); + } finally { + applyingNearDuplicateRepair.value = false; + } +} + function onScopeModeChange(): void { ensureRepairUserSelected(); } @@ -672,6 +807,98 @@ watch( + +
+

Near-Duplicate Publication Repair

+ +
+ +
+ + + + + +
+ + {{ runningNearDuplicateScan ? "Scanning..." : "Scan near-duplicate clusters" }} + +
+
+ +
+ + + + Select + Cluster + Winner + Members + Similarity + + + + + + + + {{ cluster.cluster_key }} + #{{ cluster.winner_publication_id }} + +
+ + #{{ member.publication_id }} · {{ member.title }} + +
+ + {{ cluster.similarity_score.toFixed(2) }} + + + + +
+ +
+ + {{ applyingNearDuplicateRepair ? "Merging..." : "Merge selected clusters" }} + +
+
+
+ +
+
+ Job #{{ lastNearDuplicateResult.job_id }} + Status: {{ lastNearDuplicateResult.status }} +
+ 
{{ JSON.stringify(lastNearDuplicateResult.summary, null, 2) }}
+
+ +
diff --git a/frontend/src/pages/DashboardPage.vue b/frontend/src/pages/DashboardPage.vue index 90d8513..fce9e02 100644 --- a/frontend/src/pages/DashboardPage.vue +++ b/frontend/src/pages/DashboardPage.vue @@ -1,5 +1,5 @@