diff --git a/agents.md b/agents.md
index 07b62d2..da59b76 100644
--- a/agents.md
+++ b/agents.md
@@ -7,6 +7,7 @@ Adhere strictly to these constraints.
* **DRY (Don't Repeat Yourself):** Abstract repetitive logic immediately. No duplicate boilerplate for database queries, API responses, or error handling.
* **Negative Space Programming:** Utilize explicit assertions and constraints to define invalid states. Fail fast and early. Do not allow silent failures or cascading malformed data, especially in DOM parsing.
* **Cyclomatic Complexity:** Flatten logic. Use early returns and guard clauses instead of deep nesting.
+no magic numbers
## 2. Domain Architecture & Data Model
* **Data Isolation:** Scholar tracking is **user-scoped**. Validate mapping/join tables; never assume global links between users and Scholar IDs.
@@ -29,13 +30,6 @@ These limits prevent IP bans and are not to be optimized away.
## 5. Domain Service Boundaries
* **Strict Modularity:** Flat files in the `app/services/` root are strictly prohibited. All business logic and routing must reside exclusively within `app/services/domains/`.
-* **`app/services/domains/scholar/*`:** Parser contract is fail-fast. Layout drift must emit explicit exceptions and `layout_*` reasons/warnings. Never allow silent partial success.
-* **`app/services/domains/ingestion/application.py`:** Orchestrates ingestion runs; validate parser outputs before persistence; enforce publication candidate constraints before upsert.
-* **`app/services/domains/publications/*`:** Publication list/read-state query layer. Includes `doi` + `pdf_url` fields for UI consumption, enforces non-blocking lazy OA enrichment scheduling on list reads, and exposes per-publication PDF retry behavior.
-* **`app/services/domains/crossref/*`:** DOI discovery fallback module. Must use bounded/paced lookups to avoid burst traffic and 429 responses.
-* **`app/services/domains/unpaywall/*`:** OA resolver by DOI only (best OA location + PDF URL extraction). Do not use Google Scholar for PDF resolution to avoid N+1 scrape amplification.
-* **`app/services/domains/portability/*`:** Handles JSON import/export for user-scoped scholars and scholar-publication link state while preserving global publication dedup rules.
-* **`app/services/domains/ingestion/scheduler.py`:** Owns automatic runs and continuation queue retries/drops; do not bypass safety gate or cooldown logic.
## 6. UI rules
diff --git a/alembic/versions/20260222_0016_add_publication_identifiers.py b/alembic/versions/20260222_0016_add_publication_identifiers.py
new file mode 100644
index 0000000..cd824a2
--- /dev/null
+++ b/alembic/versions/20260222_0016_add_publication_identifiers.py
@@ -0,0 +1,104 @@
+"""Add publication identifiers table.
+
+Revision ID: 20260222_0016
+Revises: 20260221_0015
+Create Date: 2026-02-22 12:45:00.000000
+"""
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "20260222_0016"
+down_revision: str | Sequence[str] | None = "20260221_0015"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _table_names() -> set[str]:
+ bind = op.get_bind()
+ inspector = sa.inspect(bind)
+ return set(inspector.get_table_names())
+
+
+def _create_publication_identifiers_table() -> None:
+ op.create_table(
+ "publication_identifiers",
+ sa.Column("id", sa.Integer(), nullable=False),
+ sa.Column("publication_id", sa.Integer(), nullable=False),
+ sa.Column("kind", sa.String(length=32), nullable=False),
+ sa.Column("value_raw", sa.Text(), nullable=False),
+ sa.Column("value_normalized", sa.String(length=255), nullable=False),
+ sa.Column("source", sa.String(length=64), nullable=False),
+ sa.Column(
+ "confidence_score",
+ sa.Float(),
+ nullable=False,
+ server_default=sa.text("0"),
+ ),
+ sa.Column("evidence_url", sa.Text(), 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.CheckConstraint(
+ "confidence_score >= 0 AND confidence_score <= 1",
+ name="publication_identifiers_confidence_score_range",
+ ),
+ sa.ForeignKeyConstraint(
+ ["publication_id"],
+ ["publications.id"],
+ ondelete="CASCADE",
+ ),
+ sa.PrimaryKeyConstraint("id", name=op.f("pk_publication_identifiers")),
+ sa.UniqueConstraint(
+ "publication_id",
+ "kind",
+ "value_normalized",
+ name="uq_publication_identifiers_publication_kind_value",
+ ),
+ )
+
+
+def _create_publication_identifiers_indexes() -> None:
+ op.create_index(
+ "ix_publication_identifiers_kind_value",
+ "publication_identifiers",
+ ["kind", "value_normalized"],
+ unique=False,
+ )
+ op.create_index(
+ "ix_publication_identifiers_publication_id",
+ "publication_identifiers",
+ ["publication_id"],
+ unique=False,
+ )
+
+
+def _drop_publication_identifiers_indexes() -> None:
+ op.drop_index("ix_publication_identifiers_publication_id", table_name="publication_identifiers")
+ op.drop_index("ix_publication_identifiers_kind_value", table_name="publication_identifiers")
+
+
+def upgrade() -> None:
+ if "publication_identifiers" in _table_names():
+ return
+ _create_publication_identifiers_table()
+ _create_publication_identifiers_indexes()
+
+
+def downgrade() -> None:
+ if "publication_identifiers" not in _table_names():
+ return
+ _drop_publication_identifiers_indexes()
+ op.drop_table("publication_identifiers")
diff --git a/app/api/routers/admin_dbops.py b/app/api/routers/admin_dbops.py
index 16271a0..cd6d924 100644
--- a/app/api/routers/admin_dbops.py
+++ b/app/api/routers/admin_dbops.py
@@ -58,6 +58,7 @@ def _serialize_pdf_queue_item(item) -> dict[str, object]:
"publication_id": item.publication_id,
"title": item.title,
"doi": item.doi,
+ "display_identifier": _serialize_display_identifier(item.display_identifier),
"pdf_url": item.pdf_url,
"status": item.status,
"attempt_count": item.attempt_count,
@@ -73,6 +74,18 @@ def _serialize_pdf_queue_item(item) -> dict[str, object]:
}
+def _serialize_display_identifier(value) -> dict[str, object] | None:
+ if value is None:
+ return None
+ return {
+ "kind": value.kind,
+ "value": value.value,
+ "label": value.label,
+ "url": value.url,
+ "confidence_score": float(value.confidence_score),
+ }
+
+
def _requeue_response_state(*, queued: bool) -> tuple[str, str]:
if queued:
return "queued", "PDF lookup queued."
diff --git a/app/api/routers/publications.py b/app/api/routers/publications.py
index 5055c6c..ad8fc55 100644
--- a/app/api/routers/publications.py
+++ b/app/api/routers/publications.py
@@ -21,6 +21,7 @@ from app.api.schemas import (
)
from app.db.models import User
from app.db.session import get_db_session
+from app.services.domains.publication_identifiers import application as identifier_service
from app.services.domains.publications import application as publication_service
from app.services.domains.scholars import application as scholar_service
from app.settings import settings
@@ -62,6 +63,7 @@ def _serialize_publication_item(item) -> dict[str, object]:
"venue_text": item.venue_text,
"pub_url": item.pub_url,
"doi": item.doi,
+ "display_identifier": _serialize_display_identifier(item.display_identifier),
"pdf_url": item.pdf_url,
"pdf_status": item.pdf_status,
"pdf_attempt_count": item.pdf_attempt_count,
@@ -74,6 +76,18 @@ def _serialize_publication_item(item) -> dict[str, object]:
}
+def _serialize_display_identifier(value) -> dict[str, object] | None:
+ if value is None:
+ return None
+ return {
+ "kind": value.kind,
+ "value": value.value,
+ "label": value.label,
+ "url": value.url,
+ "confidence_score": float(value.confidence_score),
+ }
+
+
async def _publication_counts(
db_session: AsyncSession,
*,
@@ -280,7 +294,12 @@ async def _favorite_publication_state(
db_session,
items=[publication],
)
- return hydrated[0] if hydrated else publication
+ current = hydrated[0] if hydrated else publication
+ identifiers = await identifier_service.overlay_publication_items_with_display_identifiers(
+ db_session,
+ items=[current],
+ )
+ return identifiers[0] if identifiers else current
def _log_retry_pdf_result(
diff --git a/app/api/schemas.py b/app/api/schemas.py
index 3ae8876..112f672 100644
--- a/app/api/schemas.py
+++ b/app/api/schemas.py
@@ -581,10 +581,21 @@ class AdminDbRepairJobsEnvelope(BaseModel):
model_config = ConfigDict(extra="forbid")
+class DisplayIdentifierData(BaseModel):
+ kind: str
+ value: str
+ label: str
+ url: str | None
+ confidence_score: float = Field(ge=0.0, le=1.0)
+
+ model_config = ConfigDict(extra="forbid")
+
+
class AdminPdfQueueItemData(BaseModel):
publication_id: int
title: str
doi: str | None
+ display_identifier: DisplayIdentifierData | None = None
pdf_url: str | None
status: str
attempt_count: int
@@ -744,6 +755,7 @@ class PublicationItemData(BaseModel):
venue_text: str | None
pub_url: str | None
doi: str | None
+ display_identifier: DisplayIdentifierData | None = None
pdf_url: str | None
pdf_status: str = "untracked"
pdf_attempt_count: int = 0
diff --git a/app/db/models.py b/app/db/models.py
index 43b44c2..1f0b250 100644
--- a/app/db/models.py
+++ b/app/db/models.py
@@ -8,6 +8,7 @@ from sqlalchemy import (
CheckConstraint,
DateTime,
Enum,
+ Float,
ForeignKey,
Index,
Integer,
@@ -237,6 +238,53 @@ class Publication(Base):
)
+class PublicationIdentifier(Base):
+ __tablename__ = "publication_identifiers"
+ __table_args__ = (
+ UniqueConstraint(
+ "publication_id",
+ "kind",
+ "value_normalized",
+ name="uq_publication_identifiers_publication_kind_value",
+ ),
+ CheckConstraint(
+ "confidence_score >= 0 AND confidence_score <= 1",
+ name="publication_identifiers_confidence_score_range",
+ ),
+ Index(
+ "ix_publication_identifiers_kind_value",
+ "kind",
+ "value_normalized",
+ ),
+ Index(
+ "ix_publication_identifiers_publication_id",
+ "publication_id",
+ ),
+ )
+
+ id: Mapped[int] = mapped_column(primary_key=True)
+ publication_id: Mapped[int] = mapped_column(
+ ForeignKey("publications.id", ondelete="CASCADE"),
+ nullable=False,
+ )
+ kind: Mapped[str] = mapped_column(String(32), nullable=False)
+ value_raw: Mapped[str] = mapped_column(Text, nullable=False)
+ value_normalized: Mapped[str] = mapped_column(String(255), nullable=False)
+ source: Mapped[str] = mapped_column(String(64), nullable=False)
+ confidence_score: Mapped[float] = mapped_column(
+ Float,
+ nullable=False,
+ server_default=text("0"),
+ )
+ evidence_url: Mapped[str | None] = mapped_column(Text)
+ 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 PublicationPdfJob(Base):
__tablename__ = "publication_pdf_jobs"
__table_args__ = (
diff --git a/app/services/domains/crossref/application.py b/app/services/domains/crossref/application.py
index 569c12d..f0e1311 100644
--- a/app/services/domains/crossref/application.py
+++ b/app/services/domains/crossref/application.py
@@ -21,6 +21,8 @@ STOP_WORDS = {"the", "and", "for", "with", "from", "method", "study", "analysis"
_RATE_LOCK = threading.Lock()
_LAST_REQUEST_AT = 0.0
logger = logging.getLogger(__name__)
+STRICT_TITLE_MATCH_THRESHOLD = 0.75
+RELAXED_TITLE_MATCH_THRESHOLD = 0.85
def _rate_limit_wait(min_interval_seconds: float) -> None:
@@ -132,7 +134,42 @@ def _candidate_rank(*, title: str, year: int | None, item: dict) -> tuple[float,
return score, doi
-def _best_candidate_doi(
+def _year_delta(source_year: int | None, candidate_year: int | None) -> int | None:
+ if source_year is None or candidate_year is None:
+ return None
+ return abs(int(source_year) - int(candidate_year))
+
+
+def _candidate_rank_relaxed(
+ *,
+ title: str,
+ year: int | None,
+ item: dict,
+ author_surname: str | None,
+) -> tuple[float, str | None]:
+ doi = normalize_doi(str(item.get("DOI") or ""))
+ if doi is None:
+ return 0.0, None
+ score = _title_match_score(title, _candidate_title(item))
+ if score <= 0:
+ return 0.0, None
+ candidate_year = _candidate_year(item)
+ delta = _year_delta(year, candidate_year)
+ if delta is not None:
+ if delta <= 1:
+ score += 0.05
+ elif delta <= 3:
+ score += 0.0
+ elif delta <= 5:
+ score -= 0.03
+ else:
+ score -= 0.08
+ if _candidate_author_match(item, author_surname):
+ score += 0.03
+ return score, doi
+
+
+def _best_candidate_doi_strict(
*,
title: str,
year: int | None,
@@ -149,7 +186,7 @@ def _best_candidate_doi(
continue
score, doi = _candidate_rank(title=title, year=year, item=item)
candidate_year = _candidate_year(item)
- if doi is None or score < 0.75:
+ if doi is None or score < STRICT_TITLE_MATCH_THRESHOLD:
continue
if score > best_score:
best_score = score
@@ -166,6 +203,92 @@ def _best_candidate_doi(
return best_doi
+def _best_candidate_doi_relaxed(
+ *,
+ title: str,
+ year: int | None,
+ items: list[dict],
+ author_surname: str | None,
+) -> str | None:
+ best_score = 0.0
+ best_doi: str | None = None
+ best_author_match = False
+ best_delta: int | None = None
+ best_year: int | None = None
+ for item in items:
+ if not isinstance(item, dict):
+ continue
+ score, doi = _candidate_rank_relaxed(
+ title=title,
+ year=year,
+ item=item,
+ author_surname=author_surname,
+ )
+ if doi is None or score < RELAXED_TITLE_MATCH_THRESHOLD:
+ continue
+ candidate_year = _candidate_year(item)
+ candidate_author_match = _candidate_author_match(item, author_surname)
+ candidate_delta = _year_delta(year, candidate_year)
+ if score > best_score:
+ best_score = score
+ best_doi = doi
+ best_author_match = candidate_author_match
+ best_delta = candidate_delta
+ best_year = candidate_year
+ continue
+ if abs(score - best_score) > 0.02:
+ continue
+ if candidate_author_match and not best_author_match:
+ best_doi = doi
+ best_author_match = True
+ best_delta = candidate_delta
+ best_year = candidate_year
+ continue
+ if best_delta is None and candidate_delta is not None:
+ best_doi = doi
+ best_author_match = candidate_author_match
+ best_delta = candidate_delta
+ best_year = candidate_year
+ continue
+ if best_delta is not None and candidate_delta is not None and candidate_delta < best_delta:
+ best_doi = doi
+ best_author_match = candidate_author_match
+ best_delta = candidate_delta
+ best_year = candidate_year
+ continue
+ if best_year is None or candidate_year is None:
+ continue
+ if candidate_year < best_year:
+ best_doi = doi
+ best_author_match = candidate_author_match
+ best_delta = candidate_delta
+ best_year = candidate_year
+ return best_doi
+
+
+def _best_candidate_doi(
+ *,
+ title: str,
+ year: int | None,
+ items: list[dict],
+ author_surname: str | None,
+) -> str | None:
+ strict_match = _best_candidate_doi_strict(
+ title=title,
+ year=year,
+ items=items,
+ author_surname=author_surname,
+ )
+ if strict_match:
+ return strict_match
+ return _best_candidate_doi_relaxed(
+ title=title,
+ year=year,
+ items=items,
+ author_surname=author_surname,
+ )
+
+
def _works_client(email: str | None) -> Works:
if email:
etiquette = Etiquette(settings.app_name, "0.1.0", "https://scholarr.local", email)
diff --git a/app/services/domains/ingestion/application.py b/app/services/domains/ingestion/application.py
index 54a7e25..4cb493d 100644
--- a/app/services/domains/ingestion/application.py
+++ b/app/services/domains/ingestion/application.py
@@ -29,6 +29,7 @@ from app.services.domains.ingestion.constants import (
RUN_LOCK_NAMESPACE,
)
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 (
_build_body_excerpt,
_dedupe_publication_candidates,
@@ -2071,15 +2072,24 @@ class ScholarIngestionService:
fingerprint_publication=fingerprint_publication,
)
if publication is None:
- return await self._create_publication(
+ created = await self._create_publication(
db_session,
candidate=candidate,
fingerprint=fingerprint,
)
+ await identifier_service.sync_identifiers_for_publication_fields(
+ db_session,
+ publication=created,
+ )
+ return created
self._update_existing_publication(
publication=publication,
candidate=candidate,
)
+ await identifier_service.sync_identifiers_for_publication_fields(
+ db_session,
+ publication=publication,
+ )
return publication
def _resolve_run_status(
diff --git a/app/services/domains/publication_identifiers/__init__.py b/app/services/domains/publication_identifiers/__init__.py
new file mode 100644
index 0000000..3a35c66
--- /dev/null
+++ b/app/services/domains/publication_identifiers/__init__.py
@@ -0,0 +1,17 @@
+from app.services.domains.publication_identifiers.application import (
+ DisplayIdentifier,
+ derive_display_identifier_from_values,
+ overlay_pdf_queue_items_with_display_identifiers,
+ overlay_publication_items_with_display_identifiers,
+ sync_identifiers_for_publication_fields,
+ sync_identifiers_for_publication_resolution,
+)
+
+__all__ = [
+ "DisplayIdentifier",
+ "derive_display_identifier_from_values",
+ "overlay_pdf_queue_items_with_display_identifiers",
+ "overlay_publication_items_with_display_identifiers",
+ "sync_identifiers_for_publication_fields",
+ "sync_identifiers_for_publication_resolution",
+]
diff --git a/app/services/domains/publication_identifiers/application.py b/app/services/domains/publication_identifiers/application.py
new file mode 100644
index 0000000..d34e757
--- /dev/null
+++ b/app/services/domains/publication_identifiers/application.py
@@ -0,0 +1,351 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from typing import TYPE_CHECKING
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models import Publication, PublicationIdentifier
+from app.services.domains.doi.normalize import normalize_doi
+from app.services.domains.publication_identifiers.normalize import (
+ normalize_arxiv_id,
+ normalize_pmcid,
+ normalize_pmid,
+)
+from app.services.domains.publication_identifiers.types import (
+ DisplayIdentifier,
+ IdentifierCandidate,
+ IdentifierKind,
+)
+
+if TYPE_CHECKING:
+ from app.services.domains.publications.pdf_queue import PdfQueueListItem
+ from app.services.domains.publications.types import PublicationListItem
+
+CONFIDENCE_HIGH = 0.98
+CONFIDENCE_MEDIUM = 0.9
+CONFIDENCE_LOW = 0.6
+CONFIDENCE_FALLBACK = 0.4
+PRIORITY_DOI = 400
+PRIORITY_ARXIV = 300
+PRIORITY_PMCID = 200
+PRIORITY_PMID = 100
+
+
+def derive_display_identifier_from_values(
+ *,
+ doi: str | None,
+ pub_url: str | None = None,
+ pdf_url: str | None = None,
+) -> DisplayIdentifier | None:
+ candidates = _fallback_candidates_from_values(doi=doi, pub_url=pub_url, pdf_url=pdf_url)
+ return _best_display_identifier(candidates)
+
+
+def _fallback_candidates_from_values(
+ *,
+ doi: str | None,
+ pub_url: str | None,
+ pdf_url: str | None,
+) -> list[IdentifierCandidate]:
+ values = [value for value in [pub_url, pdf_url] if value]
+ candidates = []
+ if doi:
+ normalized_doi = normalize_doi(doi)
+ if normalized_doi:
+ candidates.append(_candidate(IdentifierKind.DOI, doi, normalized_doi, "legacy_doi", CONFIDENCE_HIGH, pub_url))
+ candidates.extend(_url_identifier_candidates(values=values, source="legacy_urls"))
+ return _dedup_candidates(candidates)
+
+
+def _url_identifier_candidates(*, values: list[str], source: str) -> list[IdentifierCandidate]:
+ candidates: list[IdentifierCandidate] = []
+ for value in values:
+ candidates.extend(_url_candidates_for_value(value=value, source=source))
+ return candidates
+
+
+def _url_candidates_for_value(*, value: str, source: str) -> list[IdentifierCandidate]:
+ candidates: list[IdentifierCandidate] = []
+ arxiv = normalize_arxiv_id(value)
+ if arxiv:
+ candidates.append(_candidate(IdentifierKind.ARXIV, value, arxiv, source, CONFIDENCE_MEDIUM, value))
+ pmcid = normalize_pmcid(value)
+ if pmcid:
+ candidates.append(_candidate(IdentifierKind.PMCID, value, pmcid, source, CONFIDENCE_LOW, value))
+ pmid = normalize_pmid(value)
+ if pmid:
+ candidates.append(_candidate(IdentifierKind.PMID, value, pmid, source, CONFIDENCE_FALLBACK, value))
+ return candidates
+
+
+def _candidate(
+ kind: IdentifierKind,
+ value_raw: str,
+ value_normalized: str,
+ source: str,
+ confidence_score: float,
+ evidence_url: str | None,
+) -> IdentifierCandidate:
+ return IdentifierCandidate(
+ kind=kind,
+ value_raw=value_raw,
+ value_normalized=value_normalized,
+ source=source,
+ confidence_score=float(confidence_score),
+ evidence_url=evidence_url,
+ )
+
+
+def _dedup_candidates(candidates: list[IdentifierCandidate]) -> list[IdentifierCandidate]:
+ deduped: dict[tuple[str, str], IdentifierCandidate] = {}
+ for candidate in candidates:
+ key = (candidate.kind.value, candidate.value_normalized)
+ current = deduped.get(key)
+ if current is None or candidate.confidence_score > current.confidence_score:
+ deduped[key] = candidate
+ return list(deduped.values())
+
+
+async def sync_identifiers_for_publication_fields(
+ db_session: AsyncSession,
+ *,
+ publication: Publication,
+) -> None:
+ candidates = _publication_field_candidates(publication)
+ await _upsert_publication_candidates(db_session, publication_id=int(publication.id), candidates=candidates)
+
+
+def _publication_field_candidates(publication: Publication) -> list[IdentifierCandidate]:
+ return _fallback_candidates_from_values(
+ doi=publication.doi,
+ pub_url=publication.pub_url,
+ pdf_url=publication.pdf_url,
+ )
+
+
+async def sync_identifiers_for_publication_resolution(
+ db_session: AsyncSession,
+ *,
+ publication: Publication,
+ source: str | None,
+) -> None:
+ candidates = _publication_field_candidates(publication)
+ rewritten = [_candidate_with_source(candidate, source=source) for candidate in candidates]
+ await _upsert_publication_candidates(db_session, publication_id=int(publication.id), candidates=rewritten)
+
+
+def _candidate_with_source(candidate: IdentifierCandidate, *, source: str | None) -> IdentifierCandidate:
+ if not source:
+ return candidate
+ return IdentifierCandidate(
+ kind=candidate.kind,
+ value_raw=candidate.value_raw,
+ value_normalized=candidate.value_normalized,
+ source=source,
+ confidence_score=candidate.confidence_score,
+ evidence_url=candidate.evidence_url,
+ )
+
+
+async def _upsert_publication_candidates(
+ db_session: AsyncSession,
+ *,
+ publication_id: int,
+ candidates: list[IdentifierCandidate],
+) -> None:
+ for candidate in _dedup_candidates(candidates):
+ await _upsert_publication_candidate(db_session, publication_id=publication_id, candidate=candidate)
+
+
+async def _upsert_publication_candidate(
+ db_session: AsyncSession,
+ *,
+ publication_id: int,
+ candidate: IdentifierCandidate,
+) -> None:
+ existing = await _existing_identifier(
+ db_session,
+ publication_id=publication_id,
+ kind=candidate.kind.value,
+ value_normalized=candidate.value_normalized,
+ )
+ if existing is None:
+ db_session.add(_new_identifier_row(publication_id=publication_id, candidate=candidate))
+ return
+ _merge_identifier_row(existing, candidate=candidate)
+
+
+async def _existing_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 _new_identifier_row(
+ *,
+ publication_id: int,
+ candidate: IdentifierCandidate,
+) -> PublicationIdentifier:
+ return PublicationIdentifier(
+ publication_id=publication_id,
+ kind=candidate.kind.value,
+ value_raw=candidate.value_raw,
+ value_normalized=candidate.value_normalized,
+ source=candidate.source,
+ confidence_score=candidate.confidence_score,
+ evidence_url=candidate.evidence_url,
+ )
+
+
+def _merge_identifier_row(existing: PublicationIdentifier, *, candidate: IdentifierCandidate) -> None:
+ if candidate.confidence_score >= float(existing.confidence_score):
+ existing.value_raw = candidate.value_raw
+ existing.source = candidate.source
+ existing.confidence_score = candidate.confidence_score
+ if candidate.evidence_url:
+ existing.evidence_url = candidate.evidence_url
+
+
+async def overlay_publication_items_with_display_identifiers(
+ db_session: AsyncSession,
+ *,
+ items: list[PublicationListItem],
+) -> list[PublicationListItem]:
+ if not items:
+ return []
+ mapping = await _display_identifier_map(db_session, publication_ids=[item.publication_id for item in items])
+ return [_overlay_publication_item(item, mapping.get(item.publication_id)) for item in items]
+
+
+def _overlay_publication_item(
+ item: PublicationListItem,
+ display_identifier: DisplayIdentifier | None,
+) -> PublicationListItem:
+ fallback = display_identifier or derive_display_identifier_from_values(doi=item.doi, pub_url=item.pub_url, pdf_url=item.pdf_url)
+ return replace(item, display_identifier=fallback)
+
+
+async def overlay_pdf_queue_items_with_display_identifiers(
+ db_session: AsyncSession,
+ *,
+ items: list[PdfQueueListItem],
+) -> list[PdfQueueListItem]:
+ if not items:
+ return []
+ mapping = await _display_identifier_map(db_session, publication_ids=[item.publication_id for item in items])
+ return [_overlay_queue_item(item, mapping.get(item.publication_id)) for item in items]
+
+
+def _overlay_queue_item(
+ item: PdfQueueListItem,
+ display_identifier: DisplayIdentifier | None,
+) -> PdfQueueListItem:
+ fallback = display_identifier or derive_display_identifier_from_values(doi=item.doi, pdf_url=item.pdf_url)
+ return replace(item, display_identifier=fallback)
+
+
+async def _display_identifier_map(
+ db_session: AsyncSession,
+ *,
+ publication_ids: list[int],
+) -> dict[int, DisplayIdentifier]:
+ normalized_ids = sorted({int(value) for value in publication_ids if int(value) > 0})
+ if not normalized_ids:
+ return {}
+ result = await db_session.execute(
+ select(PublicationIdentifier).where(PublicationIdentifier.publication_id.in_(normalized_ids))
+ )
+ rows = list(result.scalars().all())
+ return _best_display_identifier_map(rows)
+
+
+def _best_display_identifier_map(rows: list[PublicationIdentifier]) -> dict[int, DisplayIdentifier]:
+ grouped: dict[int, list[IdentifierCandidate]] = {}
+ for row in rows:
+ grouped.setdefault(int(row.publication_id), []).append(_candidate_from_row(row))
+ return {
+ publication_id: display
+ for publication_id, display in (
+ (publication_id, _best_display_identifier(candidates))
+ for publication_id, candidates in grouped.items()
+ )
+ if display is not None
+ }
+
+
+def _candidate_from_row(row: PublicationIdentifier) -> IdentifierCandidate:
+ return IdentifierCandidate(
+ kind=IdentifierKind(str(row.kind)),
+ value_raw=str(row.value_raw),
+ value_normalized=str(row.value_normalized),
+ source=str(row.source),
+ confidence_score=float(row.confidence_score),
+ evidence_url=row.evidence_url,
+ )
+
+
+def _best_display_identifier(candidates: list[IdentifierCandidate]) -> DisplayIdentifier | None:
+ if not candidates:
+ return None
+ ordered = sorted(candidates, key=_display_sort_key, reverse=True)
+ return _display_identifier_from_candidate(ordered[0])
+
+
+def _display_sort_key(candidate: IdentifierCandidate) -> tuple[int, float]:
+ return (_kind_priority(candidate.kind), float(candidate.confidence_score))
+
+
+def _kind_priority(kind: IdentifierKind) -> int:
+ if kind == IdentifierKind.DOI:
+ return PRIORITY_DOI
+ if kind == IdentifierKind.ARXIV:
+ return PRIORITY_ARXIV
+ if kind == IdentifierKind.PMCID:
+ return PRIORITY_PMCID
+ return PRIORITY_PMID
+
+
+def _display_identifier_from_candidate(candidate: IdentifierCandidate) -> DisplayIdentifier:
+ value = candidate.value_normalized
+ return DisplayIdentifier(
+ kind=candidate.kind.value,
+ value=value,
+ label=_display_label(candidate.kind, value),
+ url=_identifier_url(candidate.kind, value),
+ confidence_score=float(candidate.confidence_score),
+ )
+
+
+def _display_label(kind: IdentifierKind, value: str) -> str:
+ if kind == IdentifierKind.DOI:
+ return f"DOI: {value}"
+ if kind == IdentifierKind.ARXIV:
+ return f"arXiv: {value}"
+ if kind == IdentifierKind.PMCID:
+ return f"PMCID: {value}"
+ return f"PMID: {value}"
+
+
+def _identifier_url(kind: IdentifierKind, value: str) -> str | None:
+ if kind == IdentifierKind.DOI:
+ return f"https://doi.org/{value}"
+ if kind == IdentifierKind.ARXIV:
+ return f"https://arxiv.org/abs/{value}"
+ if kind == IdentifierKind.PMCID:
+ return f"https://pmc.ncbi.nlm.nih.gov/articles/{value}/"
+ if kind == IdentifierKind.PMID:
+ return f"https://pubmed.ncbi.nlm.nih.gov/{value}/"
+ return None
diff --git a/app/services/domains/publication_identifiers/normalize.py b/app/services/domains/publication_identifiers/normalize.py
new file mode 100644
index 0000000..50036be
--- /dev/null
+++ b/app/services/domains/publication_identifiers/normalize.py
@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import re
+from urllib.parse import urlparse
+
+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_PATH_RE = re.compile(r"^/(?:abs|pdf)/([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+)/?$")
+
+
+def normalize_identifier(kind: IdentifierKind, value: str | None) -> str | None:
+ if kind == IdentifierKind.DOI:
+ return normalize_doi(value)
+ if kind == IdentifierKind.ARXIV:
+ return normalize_arxiv_id(value)
+ if kind == IdentifierKind.PMCID:
+ return normalize_pmcid(value)
+ if kind == IdentifierKind.PMID:
+ return normalize_pmid(value)
+ return None
+
+
+def normalize_arxiv_id(value: str | None) -> str | None:
+ text = (value or "").strip()
+ if not text:
+ return None
+ parsed = urlparse(text)
+ if parsed.scheme in {"http", "https"} and "arxiv.org" in parsed.netloc.lower():
+ return _arxiv_from_path(parsed.path)
+ match = ARXIV_ABS_RE.search(text)
+ if not match:
+ return None
+ version = (match.group(2) or "").lower()
+ return f"{match.group(1).lower()}{version}"
+
+
+def _arxiv_from_path(path: str) -> str | None:
+ match = ARXIV_PATH_RE.match(path or "")
+ if not match:
+ return None
+ version = (match.group(2) or "").lower()
+ return f"{match.group(1).lower()}{version}"
+
+
+def normalize_pmcid(value: str | None) -> str | None:
+ text = (value or "").strip()
+ if not text:
+ return None
+ parsed = urlparse(text)
+ if parsed.scheme in {"http", "https"} and "ncbi.nlm.nih.gov" in parsed.netloc.lower():
+ return _first_match(PMCID_RE, parsed.path)
+ return _first_match(PMCID_RE, text)
+
+
+def normalize_pmid(value: str | None) -> str | None:
+ text = (value or "").strip()
+ if not text:
+ return None
+ parsed = urlparse(text)
+ if parsed.scheme in {"http", "https"} and "pubmed.ncbi.nlm.nih.gov" in parsed.netloc.lower():
+ match = PUBMED_PATH_RE.match(parsed.path or "")
+ if not match:
+ return None
+ return match.group(1)
+ return None
+
+
+def _first_match(pattern: re.Pattern[str], value: str) -> str | None:
+ match = pattern.search(value)
+ if not match:
+ return None
+ return match.group(1).upper()
diff --git a/app/services/domains/publication_identifiers/types.py b/app/services/domains/publication_identifiers/types.py
new file mode 100644
index 0000000..34b20d0
--- /dev/null
+++ b/app/services/domains/publication_identifiers/types.py
@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class IdentifierKind(StrEnum):
+ DOI = "doi"
+ ARXIV = "arxiv"
+ PMCID = "pmcid"
+ PMID = "pmid"
+
+
+@dataclass(frozen=True)
+class DisplayIdentifier:
+ kind: str
+ value: str
+ label: str
+ url: str | None
+ confidence_score: float
+
+
+@dataclass(frozen=True)
+class IdentifierCandidate:
+ kind: IdentifierKind
+ value_raw: str
+ value_normalized: str
+ source: str
+ confidence_score: float
+ evidence_url: str | None
diff --git a/app/services/domains/publications/listing.py b/app/services/domains/publications/listing.py
index f4f518b..0d09d4c 100644
--- a/app/services/domains/publications/listing.py
+++ b/app/services/domains/publications/listing.py
@@ -2,6 +2,7 @@ from __future__ import annotations
from sqlalchemy.ext.asyncio import AsyncSession
+from app.services.domains.publication_identifiers import application as identifier_service
from app.services.domains.publications.modes import (
MODE_ALL,
MODE_UNREAD,
@@ -44,7 +45,10 @@ async def list_for_user(
publication_list_item_from_row(row, latest_run_id=latest_run_id)
for row in result.all()
]
- return rows
+ return await identifier_service.overlay_publication_items_with_display_identifiers(
+ db_session,
+ items=rows,
+ )
async def retry_pdf_for_user(
@@ -54,12 +58,19 @@ async def retry_pdf_for_user(
scholar_profile_id: int,
publication_id: int,
) -> PublicationListItem | None:
- return await get_publication_item_for_user(
+ item = await get_publication_item_for_user(
db_session,
user_id=user_id,
scholar_profile_id=scholar_profile_id,
publication_id=publication_id,
)
+ if item is None:
+ return None
+ hydrated = await identifier_service.overlay_publication_items_with_display_identifiers(
+ db_session,
+ items=[item],
+ )
+ return hydrated[0] if hydrated else item
async def list_unread_for_user(
diff --git a/app/services/domains/publications/pdf_queue.py b/app/services/domains/publications/pdf_queue.py
index b420d46..3972f7d 100644
--- a/app/services/domains/publications/pdf_queue.py
+++ b/app/services/domains/publications/pdf_queue.py
@@ -17,11 +17,16 @@ from app.db.models import (
User,
)
from app.db.session import get_session_factory
+from app.services.domains.publication_identifiers import application as identifier_service
+from app.services.domains.publication_identifiers.types import DisplayIdentifier
+from app.services.domains.publications.pdf_resolution_pipeline import (
+ PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE,
+ resolve_publication_pdf_outcome_for_row,
+)
from app.services.domains.publications.types import PublicationListItem
from app.services.domains.unpaywall.application import (
FAILURE_RESOLUTION_EXCEPTION,
OaResolutionOutcome,
- resolve_publication_oa_outcomes,
)
from app.settings import settings
@@ -57,6 +62,7 @@ class PdfQueueListItem:
last_attempt_at: datetime | None
resolved_at: datetime | None
updated_at: datetime
+ display_identifier: DisplayIdentifier | None = None
@dataclass(frozen=True)
@@ -118,6 +124,7 @@ def _item_from_row_and_job(
pdf_attempt_count=int(job.attempt_count) if job is not None else 0,
pdf_failure_reason=job.last_failure_reason if job is not None else None,
pdf_failure_detail=job.last_failure_detail if job is not None else None,
+ display_identifier=row.display_identifier,
)
@@ -366,8 +373,11 @@ async def _fetch_outcome_for_row(
row: PublicationListItem,
request_email: str | None,
) -> OaResolutionOutcome:
- outcomes = await resolve_publication_oa_outcomes([row], request_email=request_email)
- outcome = outcomes.get(row.publication_id)
+ pipeline_result = await resolve_publication_pdf_outcome_for_row(
+ row=row,
+ request_email=request_email,
+ )
+ outcome = pipeline_result.outcome
if outcome is not None:
return outcome
return _failed_outcome(row=row)
@@ -417,6 +427,11 @@ async def _persist_outcome(
if publication is None or job is None:
return
_apply_publication_update(publication, doi=outcome.doi, pdf_url=outcome.pdf_url)
+ await identifier_service.sync_identifiers_for_publication_resolution(
+ db_session,
+ publication=publication,
+ source=outcome.source,
+ )
_apply_job_outcome(job, outcome=outcome)
event_type, status = _result_event(outcome)
db_session.add(
@@ -794,6 +809,18 @@ def _queue_item_from_row(row: tuple) -> PdfQueueListItem:
)
+async def _hydrated_queue_items(
+ db_session: AsyncSession,
+ *,
+ rows: list[tuple],
+) -> list[PdfQueueListItem]:
+ items = [_queue_item_from_row(row) for row in rows]
+ return await identifier_service.overlay_pdf_queue_items_with_display_identifiers(
+ db_session,
+ items=items,
+ )
+
+
async def list_pdf_queue_items(
db_session: AsyncSession,
*,
@@ -811,7 +838,7 @@ async def list_pdf_queue_items(
offset=bounded_offset,
)
)
- return [_queue_item_from_row(row) for row in result.all()]
+ return await _hydrated_queue_items(db_session, rows=list(result.all()))
if normalized_status is None:
result = await db_session.execute(
_all_queue_select(
@@ -819,7 +846,7 @@ async def list_pdf_queue_items(
offset=bounded_offset,
)
)
- return [_queue_item_from_row(row) for row in result.all()]
+ return await _hydrated_queue_items(db_session, rows=list(result.all()))
result = await db_session.execute(
_tracked_queue_select(
limit=bounded_limit,
@@ -827,7 +854,7 @@ async def list_pdf_queue_items(
status=normalized_status,
)
)
- return [_queue_item_from_row(row) for row in result.all()]
+ return await _hydrated_queue_items(db_session, rows=list(result.all()))
async def count_pdf_queue_items(
diff --git a/app/services/domains/publications/pdf_resolution_pipeline.py b/app/services/domains/publications/pdf_resolution_pipeline.py
new file mode 100644
index 0000000..bc41257
--- /dev/null
+++ b/app/services/domains/publications/pdf_resolution_pipeline.py
@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import logging
+from urllib.parse import urlparse
+
+import httpx
+
+from app.services.domains.publications.types import PublicationListItem
+from app.services.domains.scholar.publication_pdf import (
+ ScholarPublicationLinkCandidate,
+ ScholarPublicationLinkCandidates,
+ fetch_link_candidates_from_scholar_publication_page,
+)
+from app.services.domains.unpaywall import pdf_discovery as pdf_discovery_service
+from app.services.domains.unpaywall.application import OaResolutionOutcome, resolve_publication_oa_outcomes
+from app.settings import settings
+
+logger = logging.getLogger(__name__)
+
+PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE = "scholar_publication_page"
+PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE_UNLABELED = "scholar_publication_page_unlabeled_fallback"
+PDF_PATH_TOKEN = "/pdf/"
+HTTP_TIMEOUT_FLOOR_SECONDS = 0.5
+
+
+@dataclass(frozen=True)
+class PipelineOutcome:
+ outcome: OaResolutionOutcome | None
+ scholar_candidates: ScholarPublicationLinkCandidates | None
+
+
+async def resolve_publication_pdf_outcome_for_row(
+ *,
+ row: PublicationListItem,
+ request_email: str | None,
+) -> PipelineOutcome:
+ candidates = await _safe_scholar_candidates(row.pub_url)
+ labeled = _labeled_candidate(candidates)
+ if labeled is not None:
+ return PipelineOutcome(_scholar_outcome(row=row, candidate=labeled), candidates)
+ oa_outcome = await _oa_outcome(row=row, request_email=request_email)
+ if _oa_has_pdf(oa_outcome):
+ return PipelineOutcome(oa_outcome, candidates)
+ unlabeled = _unlabeled_candidate(candidates)
+ if unlabeled is None:
+ return PipelineOutcome(oa_outcome, candidates)
+ fallback_outcome = await _unlabeled_fallback_outcome(row=row, candidate=unlabeled)
+ if fallback_outcome is not None:
+ return PipelineOutcome(fallback_outcome, candidates)
+ return PipelineOutcome(oa_outcome, candidates)
+
+
+async def _safe_scholar_candidates(pub_url: str | None) -> ScholarPublicationLinkCandidates | None:
+ try:
+ return await fetch_link_candidates_from_scholar_publication_page(pub_url)
+ except Exception as exc: # pragma: no cover - defensive boundary
+ logger.warning(
+ "publications.pdf_resolution.scholar_candidates_failed",
+ extra={"event": "publications.pdf_resolution.scholar_candidates_failed", "error": str(exc)},
+ )
+ return None
+
+
+def _labeled_candidate(
+ candidates: ScholarPublicationLinkCandidates | None,
+) -> ScholarPublicationLinkCandidate | None:
+ if candidates is None:
+ return None
+ return candidates.labeled_candidate
+
+
+def _unlabeled_candidate(
+ candidates: ScholarPublicationLinkCandidates | None,
+) -> ScholarPublicationLinkCandidate | None:
+ if candidates is None:
+ return None
+ return candidates.fallback_candidate
+
+
+def _scholar_outcome(
+ *,
+ row: PublicationListItem,
+ candidate: ScholarPublicationLinkCandidate,
+) -> OaResolutionOutcome:
+ source = (
+ PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE
+ if candidate.label_present
+ else PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE_UNLABELED
+ )
+ return OaResolutionOutcome(
+ publication_id=row.publication_id,
+ doi=row.doi,
+ pdf_url=candidate.url,
+ failure_reason=None,
+ source=source,
+ used_crossref=False,
+ )
+
+
+async def _oa_outcome(
+ *,
+ row: PublicationListItem,
+ request_email: str | None,
+) -> OaResolutionOutcome | None:
+ outcomes = await resolve_publication_oa_outcomes([row], request_email=request_email)
+ return outcomes.get(row.publication_id)
+
+
+def _oa_has_pdf(outcome: OaResolutionOutcome | None) -> bool:
+ return bool(outcome and outcome.pdf_url)
+
+
+async def _unlabeled_fallback_outcome(
+ *,
+ row: PublicationListItem,
+ candidate: ScholarPublicationLinkCandidate,
+) -> OaResolutionOutcome | None:
+ pdf_url = await _validated_pdf_url(candidate.url)
+ if pdf_url is None:
+ return None
+ return _scholar_outcome(row=row, candidate=ScholarPublicationLinkCandidate(
+ url=pdf_url,
+ confidence_score=candidate.confidence_score,
+ label_present=False,
+ reason=candidate.reason,
+ ))
+
+
+async def _validated_pdf_url(candidate_url: str) -> str | None:
+ if _looks_direct_pdf(candidate_url):
+ return candidate_url
+ timeout_seconds = _discovery_timeout_seconds()
+ async with httpx.AsyncClient(timeout=timeout_seconds) as client:
+ if await pdf_discovery_service._candidate_is_pdf(client, candidate_url=candidate_url):
+ return candidate_url
+ return await pdf_discovery_service.resolve_pdf_from_landing_page(client, page_url=candidate_url)
+
+
+def _looks_direct_pdf(url: str | None) -> bool:
+ if pdf_discovery_service.looks_like_pdf_url(url):
+ return True
+ if not isinstance(url, str):
+ return False
+ path = (urlparse(url).path or "").lower()
+ return PDF_PATH_TOKEN in path
+
+
+def _discovery_timeout_seconds() -> float:
+ return max(float(settings.unpaywall_timeout_seconds), HTTP_TIMEOUT_FLOOR_SECONDS)
diff --git a/app/services/domains/publications/types.py b/app/services/domains/publications/types.py
index d1d4bad..3bde82d 100644
--- a/app/services/domains/publications/types.py
+++ b/app/services/domains/publications/types.py
@@ -3,6 +3,8 @@ from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime
+from app.services.domains.publication_identifiers.types import DisplayIdentifier
+
@dataclass(frozen=True)
class PublicationListItem:
@@ -24,6 +26,7 @@ class PublicationListItem:
pdf_attempt_count: int = 0
pdf_failure_reason: str | None = None
pdf_failure_detail: str | None = None
+ display_identifier: DisplayIdentifier | None = None
@dataclass(frozen=True)
diff --git a/app/services/domains/scholar/publication_pdf.py b/app/services/domains/scholar/publication_pdf.py
new file mode 100644
index 0000000..6f8b6c1
--- /dev/null
+++ b/app/services/domains/scholar/publication_pdf.py
@@ -0,0 +1,232 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from html.parser import HTMLParser
+from urllib.parse import parse_qs, urlparse
+
+from app.services.domains.scholar.parser_types import ScholarDomInvariantError
+from app.services.domains.scholar.parser_utils import attr_href, normalize_space
+from app.services.domains.scholar.rate_limit import wait_for_scholar_slot
+from app.services.domains.scholar.source import LiveScholarSource
+from app.services.domains.settings.application import resolve_request_delay_minimum
+from app.settings import settings
+
+CONTAINER_ID = "gsc_oci_title_gg"
+PDF_LABEL_TOKEN = "[pdf]"
+SCHOLAR_PDF_LABELED_CONFIDENCE = 0.98
+SCHOLAR_PDF_UNLABELED_CONFIDENCE = 0.2
+ALLOWED_URL_SCHEMES = frozenset({"http", "https"})
+
+
+@dataclass(frozen=True)
+class ScholarPublicationLinkCandidate:
+ url: str
+ confidence_score: float
+ label_present: bool
+ reason: str
+
+
+@dataclass(frozen=True)
+class ScholarPublicationLinkCandidates:
+ container_seen: bool
+ labeled_candidate: ScholarPublicationLinkCandidate | None
+ fallback_candidate: ScholarPublicationLinkCandidate | None
+ warnings: tuple[str, ...] = ()
+
+
+@dataclass(frozen=True)
+class _ParsedAnchor:
+ href: str | None
+ text: str
+
+
+class _ScholarPublicationPdfParser(HTMLParser):
+ def __init__(self) -> None:
+ super().__init__(convert_charrefs=True)
+ self.container_seen = False
+ self.anchors: list[_ParsedAnchor] = []
+ self._container_depth = 0
+ self._anchor_depth = 0
+ self._anchor_href: str | None = None
+ self._anchor_parts: list[str] = []
+
+ def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+ self._increment_depths()
+ if self._starts_container(tag, attrs):
+ self.container_seen = True
+ self._container_depth = 1
+ return
+ if self._container_depth <= 0 or tag != "a":
+ return
+ if self._anchor_depth > 0:
+ return
+ self._anchor_href = attr_href(attrs)
+ self._anchor_parts = []
+ self._anchor_depth = 1
+
+ def handle_data(self, data: str) -> None:
+ if self._anchor_depth > 0:
+ self._anchor_parts.append(data)
+
+ def handle_endtag(self, tag: str) -> None:
+ if self._anchor_depth > 0:
+ self._anchor_depth -= 1
+ if self._anchor_depth == 0:
+ self._finish_anchor()
+ if self._container_depth > 0:
+ self._container_depth -= 1
+
+ def _increment_depths(self) -> None:
+ if self._container_depth > 0:
+ self._container_depth += 1
+ if self._anchor_depth > 0:
+ self._anchor_depth += 1
+
+ def _starts_container(self, tag: str, attrs: list[tuple[str, str | None]]) -> bool:
+ if tag != "div":
+ return False
+ attrs_map = {name.lower(): (value or "") for name, value in attrs}
+ return attrs_map.get("id") == CONTAINER_ID
+
+ def _finish_anchor(self) -> None:
+ self.anchors.append(
+ _ParsedAnchor(
+ href=self._anchor_href,
+ text=normalize_space("".join(self._anchor_parts)),
+ )
+ )
+ self._anchor_href = None
+ self._anchor_parts = []
+
+
+def is_scholar_publication_detail_url(url: str | None) -> bool:
+ if not isinstance(url, str) or not url.strip():
+ return False
+ parsed = urlparse(url)
+ if parsed.scheme not in ALLOWED_URL_SCHEMES:
+ return False
+ if parsed.netloc.lower() != "scholar.google.com":
+ return False
+ query = parse_qs(parsed.query)
+ return _has_view_citation_params(query)
+
+
+def _has_view_citation_params(query: dict[str, list[str]]) -> bool:
+ view_op = (query.get("view_op") or [""])[0]
+ citation = (query.get("citation_for_view") or [""])[0].strip()
+ return view_op == "view_citation" and bool(citation)
+
+
+def extract_link_candidates_from_publication_detail_html(html: str) -> ScholarPublicationLinkCandidates:
+ parser = _parsed_publication_detail(html)
+ if not parser.container_seen:
+ return ScholarPublicationLinkCandidates(False, None, None)
+ anchors = _validated_container_anchors(parser.anchors)
+ labeled = _select_labeled_candidate(anchors)
+ fallback = _select_fallback_candidate(anchors, labeled=labeled)
+ warnings = _candidate_warnings(labeled=labeled, fallback=fallback)
+ return ScholarPublicationLinkCandidates(True, labeled, fallback, warnings)
+
+
+def _parsed_publication_detail(html: str) -> _ScholarPublicationPdfParser:
+ parser = _ScholarPublicationPdfParser()
+ parser.feed(html)
+ parser.close()
+ return parser
+
+
+def _validated_container_anchors(anchors: list[_ParsedAnchor]) -> list[_ParsedAnchor]:
+ if not anchors:
+ raise ScholarDomInvariantError(
+ code="layout_publication_link_container_missing_anchor",
+ message="Scholar publication link container was present without an anchor.",
+ )
+ validated: list[_ParsedAnchor] = []
+ for anchor in anchors:
+ validated.append(_validated_anchor(anchor))
+ return validated
+
+
+def _validated_anchor(anchor: _ParsedAnchor) -> _ParsedAnchor:
+ href = (anchor.href or "").strip()
+ if not href:
+ raise ScholarDomInvariantError(
+ code="layout_publication_link_missing_href",
+ message="Scholar publication link container anchor was missing href.",
+ )
+ parsed = urlparse(href)
+ if parsed.scheme not in ALLOWED_URL_SCHEMES:
+ raise ScholarDomInvariantError(
+ code="layout_publication_link_invalid_scheme",
+ message="Scholar publication link used a non-http URL.",
+ )
+ return _ParsedAnchor(href=href, text=anchor.text)
+
+
+def _select_labeled_candidate(anchors: list[_ParsedAnchor]) -> ScholarPublicationLinkCandidate | None:
+ for anchor in anchors:
+ if PDF_LABEL_TOKEN in anchor.text.lower():
+ return ScholarPublicationLinkCandidate(
+ url=str(anchor.href),
+ confidence_score=SCHOLAR_PDF_LABELED_CONFIDENCE,
+ label_present=True,
+ reason="scholar_link_labeled_pdf",
+ )
+ return None
+
+
+def _select_fallback_candidate(
+ anchors: list[_ParsedAnchor],
+ *,
+ labeled: ScholarPublicationLinkCandidate | None,
+) -> ScholarPublicationLinkCandidate | None:
+ for anchor in anchors:
+ if labeled and anchor.href == labeled.url:
+ continue
+ return ScholarPublicationLinkCandidate(
+ url=str(anchor.href),
+ confidence_score=SCHOLAR_PDF_UNLABELED_CONFIDENCE,
+ label_present=False,
+ reason="scholar_link_unlabeled_fallback",
+ )
+ if labeled is None and anchors:
+ anchor = anchors[0]
+ return ScholarPublicationLinkCandidate(
+ url=str(anchor.href),
+ confidence_score=SCHOLAR_PDF_UNLABELED_CONFIDENCE,
+ label_present=False,
+ reason="scholar_link_unlabeled_fallback",
+ )
+ return None
+
+
+def _candidate_warnings(
+ *,
+ labeled: ScholarPublicationLinkCandidate | None,
+ fallback: ScholarPublicationLinkCandidate | None,
+) -> tuple[str, ...]:
+ warnings: list[str] = []
+ if labeled is None and fallback is not None:
+ warnings.append("scholar_publication_link_unlabeled_only")
+ return tuple(warnings)
+
+
+def _scholar_request_delay_seconds() -> int:
+ return resolve_request_delay_minimum(settings.ingestion_min_request_delay_seconds)
+
+
+def _fetch_succeeded(fetch_result) -> bool:
+ return int(fetch_result.status_code or 0) == 200 and not fetch_result.error
+
+
+async def fetch_link_candidates_from_scholar_publication_page(
+ publication_url: str | None,
+) -> ScholarPublicationLinkCandidates | None:
+ if not is_scholar_publication_detail_url(publication_url):
+ return None
+ await wait_for_scholar_slot(min_interval_seconds=float(_scholar_request_delay_seconds()))
+ source = LiveScholarSource()
+ fetch_result = await source.fetch_publication_html(str(publication_url))
+ if not _fetch_succeeded(fetch_result):
+ return None
+ return extract_link_candidates_from_publication_detail_html(fetch_result.body)
diff --git a/app/services/domains/scholar/rate_limit.py b/app/services/domains/scholar/rate_limit.py
new file mode 100644
index 0000000..947c91f
--- /dev/null
+++ b/app/services/domains/scholar/rate_limit.py
@@ -0,0 +1,18 @@
+from __future__ import annotations
+
+import asyncio
+import time
+
+_REQUEST_LOCK = asyncio.Lock()
+_LAST_REQUEST_AT = 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)
+ async with _REQUEST_LOCK:
+ elapsed = time.monotonic() - _LAST_REQUEST_AT
+ remaining = interval - elapsed
+ if remaining > 0:
+ await asyncio.sleep(remaining)
+ _LAST_REQUEST_AT = time.monotonic()
diff --git a/app/services/domains/scholar/source.py b/app/services/domains/scholar/source.py
index 83da165..92c6702 100644
--- a/app/services/domains/scholar/source.py
+++ b/app/services/domains/scholar/source.py
@@ -123,6 +123,16 @@ class LiveScholarSource:
)
return await asyncio.to_thread(self._fetch_sync, requested_url)
+ async def fetch_publication_html(self, publication_url: str) -> FetchResult:
+ logger.debug(
+ "scholar_source.publication_fetch_started",
+ extra={
+ "event": "scholar_source.publication_fetch_started",
+ "requested_url": publication_url,
+ },
+ )
+ return await asyncio.to_thread(self._fetch_sync, publication_url)
+
def _build_request(self, requested_url: str) -> Request:
return Request(
requested_url,
diff --git a/app/services/domains/unpaywall/application.py b/app/services/domains/unpaywall/application.py
index 44c7560..a658818 100644
--- a/app/services/domains/unpaywall/application.py
+++ b/app/services/domains/unpaywall/application.py
@@ -64,20 +64,16 @@ def _extract_explicit_doi(text: str | None) -> str | None:
def _publication_doi(item: PublicationListItem | UnreadPublicationItem) -> str | None:
stored = normalize_doi(item.doi)
- if stored:
- in_metadata = any(
- normalize_doi(_extract_explicit_doi(value)) == stored
- for value in (item.pub_url, item.venue_text)
- )
- if in_metadata:
- return stored
- pub_url_doi = _extract_doi_candidate(item.pub_url)
- if pub_url_doi:
- return normalize_doi(pub_url_doi)
- return (
+ explicit_doi = (
_extract_explicit_doi(item.pub_url)
or _extract_explicit_doi(item.venue_text)
)
+ if explicit_doi:
+ return explicit_doi
+ pub_url_doi = _extract_doi_candidate(item.pub_url)
+ if pub_url_doi:
+ return normalize_doi(pub_url_doi)
+ return stored
def _payload_locations(payload: dict) -> list[dict]:
@@ -217,30 +213,30 @@ async def _resolve_item_payload(
item: PublicationListItem,
email: str,
allow_crossref: bool,
-) -> tuple[dict | None, bool]:
+) -> tuple[dict | None, bool, str | None]:
doi = _publication_doi(item)
payload: dict | None = None
if doi:
payload = await _fetch_unpaywall_payload_by_doi(client=client, doi=doi, email=email)
if payload is not None and _has_direct_payload_pdf(payload):
- return payload, False
+ return payload, False, doi
if not allow_crossref or not settings.crossref_enabled:
- return payload, False
+ return payload, False, doi
crossref_doi = await discover_doi_for_publication(
item=item,
max_rows=settings.crossref_max_rows,
email=email,
)
if crossref_doi is None or crossref_doi == doi:
- return payload, crossref_doi is not None
+ return payload, crossref_doi is not None, doi or crossref_doi
crossref_payload = await _fetch_unpaywall_payload_by_doi(
client=client,
doi=crossref_doi,
email=email,
)
if crossref_payload is not None:
- return crossref_payload, True
- return payload, True
+ return crossref_payload, True, crossref_doi
+ return payload, True, crossref_doi
async def _doi_and_pdf_from_payload(
@@ -265,10 +261,11 @@ def _outcome_with_failure(
item: PublicationListItem,
failure_reason: str,
used_crossref: bool,
+ doi_override: str | None = None,
) -> OaResolutionOutcome:
return OaResolutionOutcome(
publication_id=item.publication_id,
- doi=_publication_doi(item),
+ doi=normalize_doi(doi_override) if doi_override is not None else _publication_doi(item),
pdf_url=None,
failure_reason=failure_reason,
source=None,
@@ -323,7 +320,7 @@ async def _resolve_outcome_for_item(
email: str,
allow_crossref: bool,
) -> OaResolutionOutcome:
- payload, used_crossref = await _resolve_item_payload(
+ payload, used_crossref, resolved_doi = await _resolve_item_payload(
client=client,
item=item,
email=email,
@@ -334,6 +331,7 @@ async def _resolve_outcome_for_item(
item=item,
failure_reason=_missing_payload_failure_reason(item, used_crossref=used_crossref),
used_crossref=used_crossref,
+ doi_override=resolved_doi,
)
doi, pdf_url = await _doi_and_pdf_from_payload(payload, client=client)
return _outcome_from_payload(
diff --git a/frontend/src/components/ui/AppHelpHint.vue b/frontend/src/components/ui/AppHelpHint.vue
index 6736158..50d5b52 100644
--- a/frontend/src/components/ui/AppHelpHint.vue
+++ b/frontend/src/components/ui/AppHelpHint.vue
@@ -182,7 +182,7 @@ onBeforeUnmount(() => {
:class="
hasTriggerSlot
? 'cursor-help rounded-sm'
- : 'h-5 w-5 justify-center rounded-full border border-stroke-interactive/80 bg-surface-card-muted/80 text-[11px] font-semibold text-ink-secondary transition hover:bg-action-secondary-hover-bg'
+ : 'h-5 w-5 justify-center rounded-full border border-stroke-interactive/80 bg-surface-card-muted/80 text-xs font-semibold text-ink-secondary transition hover:bg-action-secondary-hover-bg'
"
:aria-label="text"
@mouseenter="openTooltip"
@@ -200,7 +200,7 @@ onBeforeUnmount(() => {
role="tooltip"
:style="tooltipStyle"
:class="sideClass"
- class="pointer-events-none fixed z-[90] w-64 max-w-[calc(100vw-2rem)] rounded-lg border border-stroke-default bg-surface-card/95 px-2.5 py-2 text-xs leading-relaxed text-ink-secondary shadow-lg"
+ class="pointer-events-none fixed z-50 w-auto max-w-xs rounded-lg border border-stroke-default bg-surface-card/95 px-2.5 py-2 text-xs leading-relaxed text-ink-secondary shadow-lg"
>
{{ text }}
diff --git a/frontend/src/features/admin_dbops/index.ts b/frontend/src/features/admin_dbops/index.ts
index e3269e1..2f042f5 100644
--- a/frontend/src/features/admin_dbops/index.ts
+++ b/frontend/src/features/admin_dbops/index.ts
@@ -1,5 +1,13 @@
import { apiRequest } from "@/lib/api/client";
+export interface DisplayIdentifier {
+ kind: string;
+ value: string;
+ label: string;
+ url: string | null;
+ confidence_score: number;
+}
+
export interface AdminDbIntegrityCheck {
name: string;
count: number;
@@ -34,6 +42,7 @@ export interface AdminPdfQueueItem {
publication_id: number;
title: string;
doi: string | null;
+ display_identifier: DisplayIdentifier | null;
pdf_url: string | null;
status: string;
attempt_count: number;
diff --git a/frontend/src/features/publications/index.ts b/frontend/src/features/publications/index.ts
index 2eddfbb..f238a0e 100644
--- a/frontend/src/features/publications/index.ts
+++ b/frontend/src/features/publications/index.ts
@@ -2,6 +2,14 @@ import { apiRequest } from "@/lib/api/client";
export type PublicationMode = "all" | "unread" | "latest";
+export interface DisplayIdentifier {
+ kind: string;
+ value: string;
+ label: string;
+ url: string | null;
+ confidence_score: number;
+}
+
export interface PublicationItem {
publication_id: number;
scholar_profile_id: number;
@@ -12,6 +20,7 @@ export interface PublicationItem {
venue_text: string | null;
pub_url: string | null;
doi: string | null;
+ display_identifier: DisplayIdentifier | null;
pdf_url: string | null;
pdf_status: "untracked" | "queued" | "running" | "resolved" | "failed";
pdf_attempt_count: number;
diff --git a/frontend/src/features/settings/SettingsAdminPanel.vue b/frontend/src/features/settings/SettingsAdminPanel.vue
index 7f5788e..7edf67b 100644
--- a/frontend/src/features/settings/SettingsAdminPanel.vue
+++ b/frontend/src/features/settings/SettingsAdminPanel.vue
@@ -748,8 +748,14 @@ watch(
|
|
diff --git a/frontend/src/pages/PublicationsPage.vue b/frontend/src/pages/PublicationsPage.vue
index fbecad9..beeedc7 100644
--- a/frontend/src/pages/PublicationsPage.vue
+++ b/frontend/src/pages/PublicationsPage.vue
@@ -169,11 +169,15 @@ function publicationPrimaryUrl(item: PublicationItem): string | null {
return item.pub_url || item.pdf_url;
}
-function publicationDoiUrl(item: PublicationItem): string | null {
- if (!item.doi) {
+function publicationIdentifierUrl(item: PublicationItem): string | null {
+ if (!item.display_identifier?.url) {
return null;
}
- return `https://doi.org/${item.doi}`;
+ return item.display_identifier.url;
+}
+
+function publicationIdentifierLabel(item: PublicationItem): string | null {
+ return item.display_identifier?.label ?? null;
}
const selectedScholarName = computed(() => {
@@ -1043,14 +1047,14 @@ watch(
{{ item.title }}
- DOI: {{ item.doi }}
+ {{ publicationIdentifierLabel(item) }}
diff --git a/tests/unit/test_crossref_lookup.py b/tests/unit/test_crossref_lookup.py
index 5d71281..67712fa 100644
--- a/tests/unit/test_crossref_lookup.py
+++ b/tests/unit/test_crossref_lookup.py
@@ -52,7 +52,9 @@ async def test_crossref_discovers_doi_from_best_title_match(monkeypatch: pytest.
@pytest.mark.asyncio
-async def test_crossref_rejects_large_year_mismatch(monkeypatch: pytest.MonkeyPatch) -> None:
+async def test_crossref_relaxed_fallback_allows_large_year_mismatch_for_strong_title_match(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
async def _fake_fetch_items(**_kwargs):
return [
{
@@ -72,4 +74,30 @@ async def test_crossref_rejects_large_year_mismatch(monkeypatch: pytest.MonkeyPa
max_rows=10,
email=None,
)
- assert doi is None
+ assert doi == "10.1000/wrong-year"
+
+
+@pytest.mark.asyncio
+async def test_crossref_relaxed_fallback_allows_author_mismatch_for_strong_title_match(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ async def _fake_fetch_items(**_kwargs):
+ return [
+ {
+ "DOI": "10.1000/author-fallback",
+ "title": ["Induction of Pluripotent Stem Cells from Adult Human Fibroblasts"],
+ "issued": {"date-parts": [[2007]]},
+ "author": [{"family": "SomeoneElse"}],
+ }
+ ]
+
+ monkeypatch.setattr(crossref_app, "_fetch_items", _fake_fetch_items)
+ doi = await crossref_app.discover_doi_for_publication(
+ item=_item(
+ title="Induction of Pluripotent Stem Cells from Adult Human Fibroblasts",
+ year=2007,
+ ),
+ max_rows=10,
+ email=None,
+ )
+ assert doi == "10.1000/author-fallback"
diff --git a/tests/unit/test_publication_identifiers.py b/tests/unit/test_publication_identifiers.py
new file mode 100644
index 0000000..feab56b
--- /dev/null
+++ b/tests/unit/test_publication_identifiers.py
@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from app.services.domains.publication_identifiers import application as identifier_service
+
+
+def test_derive_display_identifier_prefers_doi_over_arxiv() -> None:
+ display = identifier_service.derive_display_identifier_from_values(
+ doi="10.1000/example",
+ pub_url="https://arxiv.org/abs/1504.08025",
+ pdf_url=None,
+ )
+ assert display is not None
+ assert display.kind == "doi"
+ assert display.value == "10.1000/example"
+ assert display.url == "https://doi.org/10.1000/example"
+
+
+def test_derive_display_identifier_uses_arxiv_when_doi_missing() -> None:
+ display = identifier_service.derive_display_identifier_from_values(
+ doi=None,
+ pub_url="https://arxiv.org/pdf/1504.08025v2",
+ pdf_url=None,
+ )
+ assert display is not None
+ assert display.kind == "arxiv"
+ assert display.value == "1504.08025v2"
+ assert display.label == "arXiv: 1504.08025v2"
+
+
+def test_derive_display_identifier_uses_pmcid_when_present() -> None:
+ display = identifier_service.derive_display_identifier_from_values(
+ doi=None,
+ pub_url=None,
+ pdf_url="https://pmc.ncbi.nlm.nih.gov/articles/PMC2175868/pdf/file.pdf",
+ )
+ assert display is not None
+ assert display.kind == "pmcid"
+ assert display.value == "PMC2175868"
diff --git a/tests/unit/test_publication_pdf_queue_policy.py b/tests/unit/test_publication_pdf_queue_policy.py
index 3d4fe7f..0cc3a7c 100644
--- a/tests/unit/test_publication_pdf_queue_policy.py
+++ b/tests/unit/test_publication_pdf_queue_policy.py
@@ -1,11 +1,14 @@
from __future__ import annotations
from datetime import datetime, timedelta, timezone
+from types import SimpleNamespace
import pytest
from app.db.models import PublicationPdfJob
from app.services.domains.publications import pdf_queue
+from app.services.domains.publications.pdf_resolution_pipeline import PipelineOutcome
+from app.services.domains.unpaywall.application import OaResolutionOutcome
def _job(
@@ -22,6 +25,25 @@ def _job(
)
+def _row(*, pub_url: str | None = "https://scholar.google.com/citations?view_op=view_citation&citation_for_view=abc:xyz") -> SimpleNamespace:
+ return SimpleNamespace(
+ publication_id=1,
+ scholar_profile_id=1,
+ scholar_label="Ada Lovelace",
+ title="A paper",
+ year=2024,
+ citation_count=0,
+ venue_text=None,
+ pub_url=pub_url,
+ doi=None,
+ pdf_url=None,
+ is_read=False,
+ is_favorite=False,
+ first_seen_at=datetime(2026, 2, 22, 12, 0, tzinfo=timezone.utc),
+ is_new_in_latest_run=True,
+ )
+
+
def test_pdf_queue_auto_enqueue_blocks_recent_attempt(monkeypatch: pytest.MonkeyPatch) -> None:
now = datetime(2026, 2, 21, 12, 0, tzinfo=timezone.utc)
monkeypatch.setattr(pdf_queue, "_utcnow", lambda: now)
@@ -107,3 +129,44 @@ def test_pdf_queue_manual_requeue_still_blocks_when_inflight() -> None:
)
assert pdf_queue._can_enqueue_job(running, force_retry=True) is False
assert pdf_queue._can_enqueue_job(queued, force_retry=True) is False
+
+
+@pytest.mark.asyncio
+async def test_fetch_outcome_for_row_uses_pipeline_outcome(monkeypatch: pytest.MonkeyPatch) -> None:
+ async def _fake_pipeline(*, row, request_email=None):
+ assert request_email == "user@example.com"
+ return PipelineOutcome(
+ outcome=OaResolutionOutcome(
+ publication_id=row.publication_id,
+ doi=None,
+ pdf_url="https://arxiv.org/pdf/1703.06103",
+ failure_reason=None,
+ source=pdf_queue.PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE,
+ used_crossref=False,
+ ),
+ scholar_candidates=None,
+ )
+
+ monkeypatch.setattr(pdf_queue, "resolve_publication_pdf_outcome_for_row", _fake_pipeline)
+
+ outcome = await pdf_queue._fetch_outcome_for_row(row=_row(), request_email="user@example.com")
+
+ assert outcome.pdf_url == "https://arxiv.org/pdf/1703.06103"
+ assert outcome.source == pdf_queue.PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE
+ assert outcome.used_crossref is False
+
+
+@pytest.mark.asyncio
+async def test_fetch_outcome_for_row_returns_failed_outcome_when_pipeline_returns_none(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ async def _fake_pipeline(*, row, request_email=None):
+ assert request_email == "user@example.com"
+ return PipelineOutcome(outcome=None, scholar_candidates=None)
+
+ monkeypatch.setattr(pdf_queue, "resolve_publication_pdf_outcome_for_row", _fake_pipeline)
+
+ outcome = await pdf_queue._fetch_outcome_for_row(row=_row(), request_email="user@example.com")
+
+ assert outcome.pdf_url is None
+ assert outcome.failure_reason == pdf_queue.FAILURE_RESOLUTION_EXCEPTION
diff --git a/tests/unit/test_publication_pdf_resolution_pipeline.py b/tests/unit/test_publication_pdf_resolution_pipeline.py
new file mode 100644
index 0000000..bc7bb5c
--- /dev/null
+++ b/tests/unit/test_publication_pdf_resolution_pipeline.py
@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from app.services.domains.publications import pdf_resolution_pipeline as pipeline
+from app.services.domains.unpaywall.application import OaResolutionOutcome
+
+
+@dataclass(frozen=True)
+class _Candidate:
+ url: str
+ confidence_score: float
+ label_present: bool
+ reason: str
+
+
+@dataclass(frozen=True)
+class _Candidates:
+ container_seen: bool
+ labeled_candidate: _Candidate | None
+ fallback_candidate: _Candidate | None
+ warnings: tuple[str, ...] = ()
+
+
+def _row(*, doi: str | None = None) -> SimpleNamespace:
+ return SimpleNamespace(
+ publication_id=1,
+ scholar_profile_id=1,
+ scholar_label="Ada Lovelace",
+ title="A paper",
+ year=2024,
+ citation_count=0,
+ venue_text=None,
+ pub_url="https://scholar.google.com/citations?view_op=view_citation&citation_for_view=abc:xyz",
+ doi=doi,
+ pdf_url=None,
+ is_read=False,
+ is_favorite=False,
+ first_seen_at=datetime(2026, 2, 22, 12, 0, tzinfo=timezone.utc),
+ is_new_in_latest_run=True,
+ )
+
+
+def _oa_outcome(*, pdf_url: str | None, source: str = "unpaywall") -> OaResolutionOutcome:
+ return OaResolutionOutcome(
+ publication_id=1,
+ doi="10.1000/example",
+ pdf_url=pdf_url,
+ failure_reason=None if pdf_url else "no_pdf_found",
+ source=source,
+ used_crossref=False,
+ )
+
+
+@pytest.mark.asyncio
+async def test_pipeline_prefers_labeled_scholar_candidate_before_oa(monkeypatch: pytest.MonkeyPatch) -> None:
+ async def _fake_candidates(_url):
+ return _Candidates(
+ container_seen=True,
+ labeled_candidate=_Candidate(
+ url="https://arxiv.org/pdf/1703.06103",
+ confidence_score=0.98,
+ label_present=True,
+ reason="scholar_link_labeled_pdf",
+ ),
+ fallback_candidate=None,
+ )
+
+ async def _fail_oa(*, row, request_email):
+ raise AssertionError(f"OA should not run when labeled Scholar candidate exists: {row.publication_id} {request_email}")
+
+ monkeypatch.setattr(pipeline, "fetch_link_candidates_from_scholar_publication_page", _fake_candidates)
+ monkeypatch.setattr(pipeline, "_oa_outcome", _fail_oa)
+
+ result = await pipeline.resolve_publication_pdf_outcome_for_row(row=_row(), request_email="user@example.com")
+
+ assert result.outcome is not None
+ assert result.outcome.pdf_url == "https://arxiv.org/pdf/1703.06103"
+ assert result.outcome.source == pipeline.PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE
+
+
+@pytest.mark.asyncio
+async def test_pipeline_uses_oa_result_before_unlabeled_fallback(monkeypatch: pytest.MonkeyPatch) -> None:
+ async def _fake_candidates(_url):
+ return _Candidates(
+ container_seen=True,
+ labeled_candidate=None,
+ fallback_candidate=_Candidate(
+ url="https://example.org/download/42",
+ confidence_score=0.2,
+ label_present=False,
+ reason="scholar_link_unlabeled_fallback",
+ ),
+ )
+
+ async def _fake_oa(*, row, request_email):
+ assert request_email == "user@example.com"
+ return _oa_outcome(pdf_url="https://oa.example.org/found.pdf")
+
+ async def _fail_fallback(*, row, candidate):
+ raise AssertionError(f"Unlabeled fallback should not run when OA returns PDF: {row.publication_id} {candidate.url}")
+
+ monkeypatch.setattr(pipeline, "fetch_link_candidates_from_scholar_publication_page", _fake_candidates)
+ monkeypatch.setattr(pipeline, "_oa_outcome", _fake_oa)
+ monkeypatch.setattr(pipeline, "_unlabeled_fallback_outcome", _fail_fallback)
+
+ result = await pipeline.resolve_publication_pdf_outcome_for_row(row=_row(), request_email="user@example.com")
+
+ assert result.outcome is not None
+ assert result.outcome.pdf_url == "https://oa.example.org/found.pdf"
+ assert result.outcome.source == "unpaywall"
+
+
+@pytest.mark.asyncio
+async def test_pipeline_uses_unlabeled_fallback_after_oa_failure(monkeypatch: pytest.MonkeyPatch) -> None:
+ fallback_candidate = _Candidate(
+ url="https://example.org/download/42",
+ confidence_score=0.2,
+ label_present=False,
+ reason="scholar_link_unlabeled_fallback",
+ )
+
+ async def _fake_candidates(_url):
+ return _Candidates(container_seen=True, labeled_candidate=None, fallback_candidate=fallback_candidate)
+
+ async def _fake_oa(*, row, request_email):
+ assert request_email == "user@example.com"
+ return _oa_outcome(pdf_url=None)
+
+ async def _fake_fallback(*, row, candidate):
+ assert candidate == fallback_candidate
+ return OaResolutionOutcome(
+ publication_id=row.publication_id,
+ doi=row.doi,
+ pdf_url="https://example.org/fallback.pdf",
+ failure_reason=None,
+ source=pipeline.PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE_UNLABELED,
+ used_crossref=False,
+ )
+
+ monkeypatch.setattr(pipeline, "fetch_link_candidates_from_scholar_publication_page", _fake_candidates)
+ monkeypatch.setattr(pipeline, "_oa_outcome", _fake_oa)
+ monkeypatch.setattr(pipeline, "_unlabeled_fallback_outcome", _fake_fallback)
+
+ result = await pipeline.resolve_publication_pdf_outcome_for_row(row=_row(), request_email="user@example.com")
+
+ assert result.outcome is not None
+ assert result.outcome.pdf_url == "https://example.org/fallback.pdf"
+ assert result.outcome.source == pipeline.PDF_SOURCE_SCHOLAR_PUBLICATION_PAGE_UNLABELED
diff --git a/tests/unit/test_scholar_publication_pdf.py b/tests/unit/test_scholar_publication_pdf.py
new file mode 100644
index 0000000..7382b77
--- /dev/null
+++ b/tests/unit/test_scholar_publication_pdf.py
@@ -0,0 +1,76 @@
+from __future__ import annotations
+
+import pytest
+
+from app.services.domains.scholar.parser_types import ScholarDomInvariantError
+from app.services.domains.scholar.publication_pdf import (
+ extract_link_candidates_from_publication_detail_html,
+ is_scholar_publication_detail_url,
+)
+
+
+def test_extract_link_candidates_from_publication_detail_html_reads_gsc_oci_pdf_link() -> None:
+ html = """
+
+
+
+ """
+ candidates = extract_link_candidates_from_publication_detail_html(html)
+ assert candidates.labeled_candidate is not None
+ assert candidates.labeled_candidate.url == "https://arxiv.org/pdf/1703.06103"
+
+
+def test_extract_link_candidates_from_publication_detail_html_returns_no_candidates_when_container_missing() -> None:
+ html = "No PDF section
"
+ candidates = extract_link_candidates_from_publication_detail_html(html)
+ assert candidates.container_seen is False
+ assert candidates.labeled_candidate is None
+ assert candidates.fallback_candidate is None
+
+
+def test_extract_pdf_url_from_publication_detail_html_fails_fast_on_malformed_pdf_container() -> None:
+ html = """
+
+
+
+ """
+ with pytest.raises(ScholarDomInvariantError) as exc:
+ extract_link_candidates_from_publication_detail_html(html)
+ assert exc.value.code == "layout_publication_link_missing_href"
+
+
+def test_extract_link_candidates_from_publication_detail_html_keeps_unlabeled_fallback() -> None:
+ html = """
+
+
+
+ """
+ candidates = extract_link_candidates_from_publication_detail_html(html)
+ assert candidates.container_seen is True
+ assert candidates.labeled_candidate is None
+ assert candidates.fallback_candidate is not None
+ assert candidates.fallback_candidate.url == "https://example.org/download?id=42"
+ assert candidates.fallback_candidate.label_present is False
+ assert "scholar_publication_link_unlabeled_only" in candidates.warnings
+ assert candidates.labeled_candidate is None
+
+
+def test_is_scholar_publication_detail_url_matches_view_citation_links() -> None:
+ assert is_scholar_publication_detail_url(
+ "https://scholar.google.com/citations?view_op=view_citation&hl=en&user=8200InoAAAAJ&citation_for_view=8200InoAAAAJ:gsN89kCJA0AC"
+ ) is True
+ assert is_scholar_publication_detail_url("https://example.org/paper") is False
diff --git a/tests/unit/test_unpaywall_resolution.py b/tests/unit/test_unpaywall_resolution.py
index 5cc34ce..651b33d 100644
--- a/tests/unit/test_unpaywall_resolution.py
+++ b/tests/unit/test_unpaywall_resolution.py
@@ -1,5 +1,6 @@
from __future__ import annotations
+from dataclasses import replace
from datetime import datetime, timezone
import pytest
@@ -38,6 +39,26 @@ def _item(publication_id: int) -> PublicationListItem:
)
+def test_publication_doi_uses_stored_value_when_metadata_has_no_doi() -> None:
+ item = replace(
+ _item(99),
+ pub_url="https://scholar.google.com/citations?view_op=view_citation&citation_for_view=abc:123",
+ venue_text="Cell 130 (5), 2007",
+ doi="10.1016/j.cell.2007.11.019",
+ )
+ assert unpaywall_app._publication_doi(item) == "10.1016/j.cell.2007.11.019"
+
+
+def test_publication_doi_prefers_explicit_metadata_doi_over_stored_value() -> None:
+ item = replace(
+ _item(100),
+ pub_url="https://doi.org/10.2000/fresh-value",
+ venue_text="Cell",
+ doi="10.1000/stale-value",
+ )
+ assert unpaywall_app._publication_doi(item) == "10.2000/fresh-value"
+
+
@pytest.mark.asyncio
async def test_unpaywall_resolve_prefers_direct_pdf_without_landing_crawl(monkeypatch: pytest.MonkeyPatch) -> None:
payload = {
@@ -49,7 +70,7 @@ async def test_unpaywall_resolve_prefers_direct_pdf_without_landing_crawl(monkey
}
async def _fake_resolve_item_payload(**_kwargs):
- return payload, False
+ return payload, False, "10.1016/j.cell.2007.11.019"
async def _fail_crawl(_client, *, page_url: str):
raise AssertionError(f"unexpected landing crawl: {page_url}")
@@ -75,7 +96,7 @@ async def test_unpaywall_resolve_crawls_landing_page_when_pdf_url_is_not_direct(
crawled_pages: list[str] = []
async def _fake_resolve_item_payload(**_kwargs):
- return payload, False
+ return payload, False, "10.1016/j.cell.2007.11.019"
async def _fake_crawl(_client, *, page_url: str):
crawled_pages.append(page_url)
@@ -87,3 +108,22 @@ async def test_unpaywall_resolve_crawls_landing_page_when_pdf_url_is_not_direct(
resolved = await unpaywall_app.resolve_publication_oa_metadata([_item(2)], request_email="user@example.com")
assert resolved == {2: ("10.1016/j.cell.2007.11.019", "https://oa.example.org/files/paper-42.pdf")}
assert "https://oa.example.org/landing/42" in crawled_pages
+
+
+@pytest.mark.asyncio
+async def test_unpaywall_preserves_crossref_doi_when_unpaywall_has_no_record(
+ monkeypatch: pytest.MonkeyPatch,
+) -> None:
+ async def _fake_resolve_item_payload(**_kwargs):
+ return None, True, "10.2000/crossref-only"
+
+ monkeypatch.setattr(unpaywall_app, "_resolve_item_payload", _fake_resolve_item_payload)
+ monkeypatch.setattr("httpx.AsyncClient", _DummyAsyncClient)
+
+ outcomes = await unpaywall_app.resolve_publication_oa_outcomes([_item(3)], request_email="user@example.com")
+
+ outcome = outcomes[3]
+ assert outcome.doi == "10.2000/crossref-only"
+ assert outcome.pdf_url is None
+ assert outcome.failure_reason == unpaywall_app.FAILURE_NO_RECORD
+ assert outcome.used_crossref is True