justinzeus/scholarr
1
0
Fork 0
Code Issues 1 Pull requests 18 Projects Releases Packages Wiki Activity Actions

harden domain architecture and add lazy Crossref+Unpaywall PDF enrichment with publications workspace refactor

Browse source
This commit is contained in:
Justin Visser 2026-02-20 22:56:52 +01:00
parent a527e7a535
commit 01454162bb
62 changed files with 2562 additions and 688 deletions
Download patch file Download diff file Expand all files Collapse all files

2
app/api/routers/admin.py
View file

@ -20,7 +20,7 @@ from app.auth.deps import get_auth_service
from app.auth.service import AuthService
from app.db.models import User
from app.db.session import get_db_session
from app.services import users as user_service
from app.services.domains.users import application as user_service
logger = logging.getLogger(__name__)

89
app/api/routers/auth.py
View file

@ -24,13 +24,55 @@ from app.auth.session import set_session_user
from app.db.models import User
from app.db.session import get_db_session
from app.security.csrf import ensure_csrf_token
from app.services import users as user_service
from app.services.domains.users import application as user_service
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/auth", tags=["api-auth"])
def _login_limiter_key_and_email(request: Request, payload: LoginRequest) -> tuple[str, str]:
return auth_runtime.login_rate_limit_key(request, payload.email), payload.email.strip().lower()
def _raise_rate_limited(normalized_email: str, retry_after_seconds: int) -> None:
logger.warning(
"api.auth.login_rate_limited",
extra={
"event": "api.auth.login_rate_limited",
"email": normalized_email,
"retry_after_seconds": retry_after_seconds,
},
)
raise ApiException(
status_code=429,
code="rate_limited",
message="Too many login attempts. Please try again later.",
details={"retry_after_seconds": retry_after_seconds},
)
def _serialize_user_payload(user: User) -> dict[str, object]:
return {
"id": int(user.id),
"email": user.email,
"is_admin": bool(user.is_admin),
"is_active": bool(user.is_active),
}
def _raise_invalid_credentials(*, normalized_email: str) -> None:
logger.info(
"api.auth.login_failed",
extra={"event": "api.auth.login_failed", "email": normalized_email},
)
raise ApiException(
status_code=401,
code="invalid_credentials",
message="Invalid email or password.",
)
@router.post(
"/login",
response_model=LoginEnvelope,
@ -42,24 +84,10 @@ async def login(
auth_service: AuthService = Depends(get_auth_service),
rate_limiter: SlidingWindowRateLimiter = Depends(get_login_rate_limiter),
):
limiter_key = auth_runtime.login_rate_limit_key(request, payload.email)
limiter_key, normalized_email = _login_limiter_key_and_email(request, payload)
decision = rate_limiter.check(limiter_key)
normalized_email = payload.email.strip().lower()
if not decision.allowed:
logger.warning(
"api.auth.login_rate_limited",
extra={
"event": "api.auth.login_rate_limited",
"email": normalized_email,
"retry_after_seconds": decision.retry_after_seconds,
},
)
raise ApiException(
status_code=429,
code="rate_limited",
message="Too many login attempts. Please try again later.",
details={"retry_after_seconds": decision.retry_after_seconds},
)
_raise_rate_limited(normalized_email, int(decision.retry_after_seconds))
user = await auth_service.authenticate_user(
db_session,
@ -68,18 +96,7 @@ async def login(
)
if user is None:
rate_limiter.record_failure(limiter_key)
logger.info(
"api.auth.login_failed",
extra={
"event": "api.auth.login_failed",
"email": normalized_email,
},
)
raise ApiException(
status_code=401,
code="invalid_credentials",
message="Invalid email or password.",
)
_raise_invalid_credentials(normalized_email=normalized_email)
rate_limiter.reset(limiter_key)
set_session_user(
@ -101,12 +118,7 @@ async def login(
data={
"authenticated": True,
"csrf_token": ensure_csrf_token(request),
"user": {
"id": int(user.id),
"email": user.email,
"is_admin": bool(user.is_admin),
"is_active": bool(user.is_active),
},
"user": _serialize_user_payload(user),
},
)
@ -124,12 +136,7 @@ async def get_current_session(
data={
"authenticated": True,
"csrf_token": ensure_csrf_token(request),
"user": {
"id": int(current_user.id),
"email": current_user.email,
"is_admin": bool(current_user.is_admin),
"is_active": bool(current_user.is_active),
},
"user": _serialize_user_payload(current_user),
},
)

213
app/api/routers/publications.py
View file

@ -3,7 +3,7 @@ from __future__ import annotations
import logging
from typing import Literal
from fastapi import APIRouter, Depends, Query, Request
from fastapi import APIRouter, Depends, Path, Query, Request
from sqlalchemy.ext.asyncio import AsyncSession
from app.api.deps import get_api_current_user
@ -14,17 +14,104 @@ from app.api.schemas import (
MarkSelectedReadEnvelope,
MarkSelectedReadRequest,
PublicationsListEnvelope,
RetryPublicationPdfEnvelope,
RetryPublicationPdfRequest,
)
from app.db.models import User
from app.db.session import get_db_session
from app.services import publications as publication_service
from app.services import scholars as scholar_service
from app.services.domains.publications import application as publication_service
from app.services.domains.scholars import application as scholar_service
from app.settings import settings
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/publications", tags=["api-publications"])
async def _require_selected_profile(
db_session: AsyncSession,
*,
user_id: int,
selected_scholar_id: int | None,
) -> None:
if selected_scholar_id is None:
return
selected_profile = await scholar_service.get_user_scholar_by_id(
db_session,
user_id=user_id,
scholar_profile_id=selected_scholar_id,
)
if selected_profile is None:
raise ApiException(
status_code=404,
code="scholar_not_found",
message="Scholar filter not found.",
)
def _serialize_publication_item(item) -> dict[str, object]:
return {
"publication_id": item.publication_id,
"scholar_profile_id": item.scholar_profile_id,
"scholar_label": item.scholar_label,
"title": item.title,
"year": item.year,
"citation_count": item.citation_count,
"venue_text": item.venue_text,
"pub_url": item.pub_url,
"doi": item.doi,
"pdf_url": item.pdf_url,
"is_read": item.is_read,
"first_seen_at": item.first_seen_at,
"is_new_in_latest_run": item.is_new_in_latest_run,
}
async def _publication_counts(
db_session: AsyncSession,
*,
user_id: int,
selected_scholar_id: int | None,
) -> tuple[int, int, int]:
unread_count = await publication_service.count_unread_for_user(
db_session,
user_id=user_id,
scholar_profile_id=selected_scholar_id,
)
latest_count = await publication_service.count_latest_for_user(
db_session,
user_id=user_id,
scholar_profile_id=selected_scholar_id,
)
total_count = await publication_service.count_for_user(
db_session,
user_id=user_id,
mode=publication_service.MODE_ALL,
scholar_profile_id=selected_scholar_id,
)
return unread_count, latest_count, total_count
def _publications_list_data(
*,
mode: str,
selected_scholar_id: int | None,
unread_count: int,
latest_count: int,
total_count: int,
publications: list,
) -> dict[str, object]:
return {
"mode": mode,
"selected_scholar_profile_id": selected_scholar_id,
"unread_count": unread_count,
"latest_count": latest_count,
"new_count": latest_count,
"total_count": total_count,
"publications": [_serialize_publication_item(item) for item in publications],
}
@router.get(
"",
response_model=PublicationsListEnvelope,
@ -39,18 +126,11 @@ async def list_publications(
):
resolved_mode = publication_service.resolve_publication_view_mode(mode)
selected_scholar_id = scholar_profile_id
if selected_scholar_id is not None:
selected_profile = await scholar_service.get_user_scholar_by_id(
db_session,
user_id=current_user.id,
scholar_profile_id=selected_scholar_id,
)
if selected_profile is None:
raise ApiException(
status_code=404,
code="scholar_not_found",
message="Scholar filter not found.",
)
await _require_selected_profile(
db_session,
user_id=current_user.id,
selected_scholar_id=selected_scholar_id,
)
publications = await publication_service.list_for_user(
db_session,
@ -59,50 +139,26 @@ async def list_publications(
scholar_profile_id=selected_scholar_id,
limit=limit,
)
unread_count = await publication_service.count_unread_for_user(
await publication_service.schedule_missing_pdf_enrichment_for_user(
user_id=current_user.id,
request_email=current_user.email,
items=publications,
max_items=settings.unpaywall_max_items_per_request,
)
unread_count, latest_count, total_count = await _publication_counts(
db_session,
user_id=current_user.id,
scholar_profile_id=selected_scholar_id,
selected_scholar_id=selected_scholar_id,
)
latest_count = await publication_service.count_latest_for_user(
db_session,
user_id=current_user.id,
scholar_profile_id=selected_scholar_id,
)
total_count = await publication_service.count_for_user(
db_session,
user_id=current_user.id,
mode=publication_service.MODE_ALL,
scholar_profile_id=selected_scholar_id,
)
return success_payload(
request,
data={
"mode": resolved_mode,
"selected_scholar_profile_id": selected_scholar_id,
"unread_count": unread_count,
"latest_count": latest_count,
"new_count": latest_count,
"total_count": total_count,
"publications": [
{
"publication_id": item.publication_id,
"scholar_profile_id": item.scholar_profile_id,
"scholar_label": item.scholar_label,
"title": item.title,
"year": item.year,
"citation_count": item.citation_count,
"venue_text": item.venue_text,
"pub_url": item.pub_url,
"pdf_url": item.pdf_url,
"is_read": item.is_read,
"first_seen_at": item.first_seen_at,
"is_new_in_latest_run": item.is_new_in_latest_run,
}
for item in publications
],
},
data = _publications_list_data(
mode=resolved_mode,
selected_scholar_id=selected_scholar_id,
unread_count=unread_count,
latest_count=latest_count,
total_count=total_count,
publications=publications,
)
return success_payload(request, data=data)
@router.post(
@ -173,3 +229,50 @@ async def mark_selected_publications_read(
"updated_count": updated_count,
},
)
@router.post(
"/{publication_id}/retry-pdf",
response_model=RetryPublicationPdfEnvelope,
)
async def retry_publication_pdf(
payload: RetryPublicationPdfRequest,
request: Request,
publication_id: int = Path(ge=1),
db_session: AsyncSession = Depends(get_db_session),
current_user: User = Depends(get_api_current_user),
):
publication = await publication_service.retry_pdf_for_user(
db_session,
user_id=current_user.id,
scholar_profile_id=payload.scholar_profile_id,
publication_id=publication_id,
unpaywall_email=current_user.email,
)
if publication is None:
raise ApiException(
status_code=404,
code="publication_not_found",
message="Publication not found.",
)
resolved_pdf = bool(publication.pdf_url)
logger.info(
"api.publications.retry_pdf",
extra={
"event": "api.publications.retry_pdf",
"user_id": current_user.id,
"scholar_profile_id": payload.scholar_profile_id,
"publication_id": publication_id,
"resolved_pdf": resolved_pdf,
"has_doi": bool(publication.doi),
},
)
message = "Open-access PDF link resolved." if resolved_pdf else "No open-access PDF link found."
return success_payload(
request,
data={
"message": message,
"resolved_pdf": resolved_pdf,
"publication": _serialize_publication_item(publication),
},
)

415
app/api/routers/runs.py
View file

@ -21,10 +21,10 @@ from app.api.schemas import (
)
from app.db.models import RunStatus, RunTriggerType, User
from app.db.session import get_db_session
from app.services import ingestion as ingestion_service
from app.services import run_safety as run_safety_service
from app.services import runs as run_service
from app.services import user_settings as user_settings_service
from app.services.domains.ingestion import application as ingestion_service
from app.services.domains.ingestion import safety as run_safety_service
from app.services.domains.runs import application as run_service
from app.services.domains.settings import application as user_settings_service
from app.settings import settings
from app.api.runtime_deps import get_ingestion_service
@ -294,10 +294,228 @@ async def _load_safety_state(
"metric_name": "api_runs_safety_cooldown_cleared_total",
"metric_value": 1,
},
)
)
return run_safety_service.get_safety_state_payload(user_settings)
def _raise_manual_runs_disabled(*, user_id: int, safety_state: dict[str, Any]) -> None:
logger.warning(
"api.runs.manual_blocked_policy",
extra={
"event": "api.runs.manual_blocked_policy",
"user_id": user_id,
"policy": {"manual_run_allowed": False},
"safety_state": safety_state,
"metric_name": "api_runs_manual_blocked_policy_total",
"metric_value": 1,
},
)
raise ApiException(
status_code=403,
code="manual_runs_disabled",
message="Manual checks are disabled by server policy.",
details={
"policy": {"manual_run_allowed": False},
"safety_state": safety_state,
},
)
async def _reused_manual_run_payload(
db_session: AsyncSession,
*,
request: Request,
user_id: int,
idempotency_key: str | None,
safety_state: dict[str, Any],
) -> dict[str, Any] | None:
if idempotency_key is None:
return None
previous_run = await run_service.get_manual_run_by_idempotency_key(
db_session,
user_id=user_id,
idempotency_key=idempotency_key,
)
if previous_run is None:
return None
if previous_run.status == RunStatus.RUNNING:
raise ApiException(
status_code=409,
code="run_in_progress",
message="A run with this idempotency key is still in progress.",
details={"run_id": int(previous_run.id), "idempotency_key": idempotency_key},
)
return success_payload(
request,
data=_manual_run_payload_from_run(
previous_run,
idempotency_key=idempotency_key,
reused_existing_run=True,
safety_state=safety_state,
),
)
async def _run_ingestion_for_manual(
db_session: AsyncSession,
*,
ingest_service: ingestion_service.ScholarIngestionService,
user_id: int,
idempotency_key: str | None,
):
user_settings = await user_settings_service.get_or_create_settings(
db_session,
user_id=user_id,
)
return await ingest_service.run_for_user(
db_session,
user_id=user_id,
trigger_type=RunTriggerType.MANUAL,
request_delay_seconds=user_settings.request_delay_seconds,
network_error_retries=settings.ingestion_network_error_retries,
retry_backoff_seconds=settings.ingestion_retry_backoff_seconds,
max_pages_per_scholar=settings.ingestion_max_pages_per_scholar,
page_size=settings.ingestion_page_size,
auto_queue_continuations=settings.ingestion_continuation_queue_enabled,
queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds,
idempotency_key=idempotency_key,
alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold,
alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold,
alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold,
)
async def _recover_integrity_error(
db_session: AsyncSession,
*,
request: Request,
user_id: int,
idempotency_key: str | None,
original_exc: IntegrityError,
) -> dict[str, Any]:
if idempotency_key is None:
logger.exception(
"api.runs.manual_integrity_error",
extra={"event": "api.runs.manual_integrity_error", "user_id": user_id},
)
raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc
existing_run = await run_service.get_manual_run_by_idempotency_key(
db_session,
user_id=user_id,
idempotency_key=idempotency_key,
)
if existing_run is None:
logger.exception(
"api.runs.manual_integrity_error",
extra={"event": "api.runs.manual_integrity_error", "user_id": user_id},
)
raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc
if existing_run.status == RunStatus.RUNNING:
raise ApiException(
status_code=409,
code="run_in_progress",
message="A run with this idempotency key is still in progress.",
details={"run_id": int(existing_run.id), "idempotency_key": idempotency_key},
) from original_exc
return success_payload(
request,
data=_manual_run_payload_from_run(
existing_run,
idempotency_key=idempotency_key,
reused_existing_run=True,
safety_state=await _load_safety_state(db_session, user_id=user_id),
),
)
async def _execute_manual_run(
db_session: AsyncSession,
*,
request: Request,
ingest_service: ingestion_service.ScholarIngestionService,
user_id: int,
idempotency_key: str | None,
):
try:
return await _run_ingestion_for_manual(
db_session,
ingest_service=ingest_service,
user_id=user_id,
idempotency_key=idempotency_key,
)
except ingestion_service.RunAlreadyInProgressError as exc:
await db_session.rollback()
raise ApiException(
status_code=409,
code="run_in_progress",
message="A run is already in progress for this account.",
) from exc
except ingestion_service.RunBlockedBySafetyPolicyError as exc:
await db_session.rollback()
_raise_manual_blocked_safety(exc=exc, user_id=user_id)
except IntegrityError as exc:
await db_session.rollback()
return await _recover_integrity_error(
db_session,
request=request,
user_id=user_id,
idempotency_key=idempotency_key,
original_exc=exc,
)
except Exception as exc:
await db_session.rollback()
_raise_manual_failed(exc=exc, user_id=user_id)
def _raise_manual_blocked_safety(*, exc, user_id: int) -> None:
logger.info(
"api.runs.manual_blocked_safety",
extra={
"event": "api.runs.manual_blocked_safety",
"user_id": user_id,
"reason": exc.safety_state.get("cooldown_reason"),
"cooldown_until": exc.safety_state.get("cooldown_until"),
"cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"),
"metric_name": "api_runs_manual_blocked_safety_total",
"metric_value": 1,
},
)
raise ApiException(
status_code=429,
code=exc.code,
message=exc.message,
details={"safety_state": exc.safety_state},
) from exc
def _raise_manual_failed(*, exc: Exception, user_id: int) -> None:
logger.exception(
"api.runs.manual_failed",
extra={"event": "api.runs.manual_failed", "user_id": user_id},
)
raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from exc
def _manual_run_success_payload(
*,
run_summary,
idempotency_key: str | None,
safety_state: dict[str, Any],
) -> dict[str, Any]:
return {
"run_id": run_summary.crawl_run_id,
"status": run_summary.status.value,
"scholar_count": run_summary.scholar_count,
"succeeded_count": run_summary.succeeded_count,
"failed_count": run_summary.failed_count,
"partial_count": run_summary.partial_count,
"new_publication_count": run_summary.new_publication_count,
"reused_existing_run": False,
"idempotency_key": idempotency_key,
"safety_state": safety_state,
}
@router.get(
"",
response_model=RunsListEnvelope,
@ -383,164 +601,28 @@ async def run_manual(
user_id=current_user.id,
)
if not settings.ingestion_manual_run_allowed:
logger.warning(
"api.runs.manual_blocked_policy",
extra={
"event": "api.runs.manual_blocked_policy",
"user_id": current_user.id,
"policy": {"manual_run_allowed": False},
"safety_state": safety_state,
"metric_name": "api_runs_manual_blocked_policy_total",
"metric_value": 1,
},
)
raise ApiException(
status_code=403,
code="manual_runs_disabled",
message="Manual checks are disabled by server policy.",
details={
"policy": {
"manual_run_allowed": False,
},
"safety_state": safety_state,
},
)
_raise_manual_runs_disabled(user_id=current_user.id, safety_state=safety_state)
idempotency_key = _normalize_idempotency_key(request.headers.get(IDEMPOTENCY_HEADER))
if idempotency_key is not None:
previous_run = await run_service.get_manual_run_by_idempotency_key(
db_session,
user_id=current_user.id,
idempotency_key=idempotency_key,
)
if previous_run is not None:
if previous_run.status == RunStatus.RUNNING:
raise ApiException(
status_code=409,
code="run_in_progress",
message="A run with this idempotency key is still in progress.",
details={
"run_id": int(previous_run.id),
"idempotency_key": idempotency_key,
},
)
return success_payload(
request,
data=_manual_run_payload_from_run(
previous_run,
idempotency_key=idempotency_key,
reused_existing_run=True,
safety_state=safety_state,
),
)
user_settings = await user_settings_service.get_or_create_settings(
reused_payload = await _reused_manual_run_payload(
db_session,
request=request,
user_id=current_user.id,
idempotency_key=idempotency_key,
safety_state=safety_state,
)
try:
run_summary = await ingest_service.run_for_user(
db_session,
user_id=current_user.id,
trigger_type=RunTriggerType.MANUAL,
request_delay_seconds=user_settings.request_delay_seconds,
network_error_retries=settings.ingestion_network_error_retries,
retry_backoff_seconds=settings.ingestion_retry_backoff_seconds,
max_pages_per_scholar=settings.ingestion_max_pages_per_scholar,
page_size=settings.ingestion_page_size,
auto_queue_continuations=settings.ingestion_continuation_queue_enabled,
queue_delay_seconds=settings.ingestion_continuation_base_delay_seconds,
idempotency_key=idempotency_key,
alert_blocked_failure_threshold=settings.ingestion_alert_blocked_failure_threshold,
alert_network_failure_threshold=settings.ingestion_alert_network_failure_threshold,
alert_retry_scheduled_threshold=settings.ingestion_alert_retry_scheduled_threshold,
)
except ingestion_service.RunAlreadyInProgressError as exc:
await db_session.rollback()
raise ApiException(
status_code=409,
code="run_in_progress",
message="A run is already in progress for this account.",
) from exc
except ingestion_service.RunBlockedBySafetyPolicyError as exc:
await db_session.rollback()
logger.info(
"api.runs.manual_blocked_safety",
extra={
"event": "api.runs.manual_blocked_safety",
"user_id": current_user.id,
"reason": exc.safety_state.get("cooldown_reason"),
"cooldown_until": exc.safety_state.get("cooldown_until"),
"cooldown_remaining_seconds": exc.safety_state.get("cooldown_remaining_seconds"),
"metric_name": "api_runs_manual_blocked_safety_total",
"metric_value": 1,
},
)
raise ApiException(
status_code=429,
code=exc.code,
message=exc.message,
details={
"safety_state": exc.safety_state,
},
) from exc
except IntegrityError as exc:
await db_session.rollback()
if idempotency_key is not None:
existing_run = await run_service.get_manual_run_by_idempotency_key(
db_session,
user_id=current_user.id,
idempotency_key=idempotency_key,
)
if existing_run is not None:
if existing_run.status == RunStatus.RUNNING:
raise ApiException(
status_code=409,
code="run_in_progress",
message="A run with this idempotency key is still in progress.",
details={
"run_id": int(existing_run.id),
"idempotency_key": idempotency_key,
},
) from exc
return success_payload(
request,
data=_manual_run_payload_from_run(
existing_run,
idempotency_key=idempotency_key,
reused_existing_run=True,
safety_state=await _load_safety_state(
db_session,
user_id=current_user.id,
),
),
)
logger.exception(
"api.runs.manual_integrity_error",
extra={
"event": "api.runs.manual_integrity_error",
"user_id": current_user.id,
},
)
raise ApiException(
status_code=500,
code="manual_run_failed",
message="Manual run failed.",
) from exc
except Exception as exc:
await db_session.rollback()
logger.exception(
"api.runs.manual_failed",
extra={
"event": "api.runs.manual_failed",
"user_id": current_user.id,
},
)
raise ApiException(
status_code=500,
code="manual_run_failed",
message="Manual run failed.",
) from exc
if reused_payload is not None:
return reused_payload
run_summary = await _execute_manual_run(
db_session,
request=request,
ingest_service=ingest_service,
user_id=current_user.id,
idempotency_key=idempotency_key,
)
if isinstance(run_summary, dict):
return run_summary
current_safety_state = await _load_safety_state(
db_session,
@ -548,18 +630,11 @@ async def run_manual(
)
return success_payload(
request,
data={
"run_id": run_summary.crawl_run_id,
"status": run_summary.status.value,
"scholar_count": run_summary.scholar_count,
"succeeded_count": run_summary.succeeded_count,
"failed_count": run_summary.failed_count,
"partial_count": run_summary.partial_count,
"new_publication_count": run_summary.new_publication_count,
"reused_existing_run": False,
"idempotency_key": idempotency_key,
"safety_state": current_safety_state,
},
data=_manual_run_success_payload(
run_summary=run_summary,
idempotency_key=idempotency_key,
safety_state=current_safety_state,
),
)

186
app/api/routers/scholars.py
View file

@ -25,9 +25,9 @@ from app.api.schemas import (
)
from app.db.models import User
from app.db.session import get_db_session
from app.services import import_export as import_export_service
from app.services import scholars as scholar_service
from app.services.scholar_source import ScholarSource
from app.services.domains.portability import application as import_export_service
from app.services.domains.scholars import application as scholar_service
from app.services.domains.scholar.source import ScholarSource
from app.settings import settings
logger = logging.getLogger(__name__)
@ -64,6 +64,107 @@ def _serialize_scholar(profile) -> dict[str, object]:
}
async def _hydrate_scholar_metadata_if_needed(
db_session: AsyncSession,
*,
profile,
source: ScholarSource,
user_id: int,
):
try:
if not profile.profile_image_url or not (profile.display_name or "").strip():
return await asyncio.wait_for(
scholar_service.hydrate_profile_metadata(
db_session,
profile=profile,
source=source,
),
timeout=5.0,
)
except Exception:
logger.warning(
"api.scholars.create_metadata_hydration_failed",
extra={
"event": "api.scholars.create_metadata_hydration_failed",
"user_id": user_id,
"scholar_profile_id": profile.id,
},
)
return profile
def _search_kwargs() -> dict[str, object]:
return {
"network_error_retries": settings.ingestion_network_error_retries,
"retry_backoff_seconds": settings.ingestion_retry_backoff_seconds,
"search_enabled": settings.scholar_name_search_enabled,
"cache_ttl_seconds": settings.scholar_name_search_cache_ttl_seconds,
"blocked_cache_ttl_seconds": settings.scholar_name_search_blocked_cache_ttl_seconds,
"cache_max_entries": settings.scholar_name_search_cache_max_entries,
"min_interval_seconds": settings.scholar_name_search_min_interval_seconds,
"interval_jitter_seconds": settings.scholar_name_search_interval_jitter_seconds,
"cooldown_block_threshold": settings.scholar_name_search_cooldown_block_threshold,
"cooldown_seconds": settings.scholar_name_search_cooldown_seconds,
"retry_alert_threshold": settings.scholar_name_search_alert_retry_count_threshold,
"cooldown_rejection_alert_threshold": (
settings.scholar_name_search_alert_cooldown_rejections_threshold
),
}
def _search_response_data(query: str, parsed) -> dict[str, object]:
return {
"query": query.strip(),
"state": parsed.state.value,
"state_reason": parsed.state_reason,
"action_hint": scholar_service.scrape_state_hint(
state=parsed.state,
state_reason=parsed.state_reason,
),
"candidates": [
{
"scholar_id": item.scholar_id,
"display_name": item.display_name,
"affiliation": item.affiliation,
"email_domain": item.email_domain,
"cited_by_count": item.cited_by_count,
"interests": item.interests,
"profile_url": item.profile_url,
"profile_image_url": item.profile_image_url,
}
for item in parsed.candidates
],
"warnings": parsed.warnings,
}
async def _read_uploaded_image(image: UploadFile) -> bytes:
try:
return await image.read()
finally:
await image.close()
async def _require_user_profile(
db_session: AsyncSession,
*,
user_id: int,
scholar_profile_id: int,
):
profile = await scholar_service.get_user_scholar_by_id(
db_session,
user_id=user_id,
scholar_profile_id=scholar_profile_id,
)
if profile is None:
raise ApiException(
status_code=404,
code="scholar_not_found",
message="Scholar not found.",
)
return profile
@router.get(
"",
response_model=ScholarsListEnvelope,
@ -181,25 +282,12 @@ async def create_scholar(
"scholar_profile_id": created.id,
},
)
try:
if not created.profile_image_url or not (created.display_name or "").strip():
created = await asyncio.wait_for(
scholar_service.hydrate_profile_metadata(
db_session,
profile=created,
source=source,
),
timeout=5.0,
)
except Exception:
logger.warning(
"api.scholars.create_metadata_hydration_failed",
extra={
"event": "api.scholars.create_metadata_hydration_failed",
"user_id": current_user.id,
"scholar_profile_id": created.id,
},
)
created = await _hydrate_scholar_metadata_if_needed(
db_session,
profile=created,
source=source,
user_id=current_user.id,
)
return success_payload(
request,
@ -225,20 +313,7 @@ async def search_scholars(
db_session=db_session,
query=query,
limit=limit,
network_error_retries=settings.ingestion_network_error_retries,
retry_backoff_seconds=settings.ingestion_retry_backoff_seconds,
search_enabled=settings.scholar_name_search_enabled,
cache_ttl_seconds=settings.scholar_name_search_cache_ttl_seconds,
blocked_cache_ttl_seconds=settings.scholar_name_search_blocked_cache_ttl_seconds,
cache_max_entries=settings.scholar_name_search_cache_max_entries,
min_interval_seconds=settings.scholar_name_search_min_interval_seconds,
interval_jitter_seconds=settings.scholar_name_search_interval_jitter_seconds,
cooldown_block_threshold=settings.scholar_name_search_cooldown_block_threshold,
cooldown_seconds=settings.scholar_name_search_cooldown_seconds,
retry_alert_threshold=settings.scholar_name_search_alert_retry_count_threshold,
cooldown_rejection_alert_threshold=(
settings.scholar_name_search_alert_cooldown_rejections_threshold
),
**_search_kwargs(),
)
except scholar_service.ScholarServiceError as exc:
raise ApiException(
@ -259,29 +334,7 @@ async def search_scholars(
)
return success_payload(
request,
data={
"query": query.strip(),
"state": parsed.state.value,
"state_reason": parsed.state_reason,
"action_hint": scholar_service.scrape_state_hint(
state=parsed.state,
state_reason=parsed.state_reason,
),
"candidates": [
{
"scholar_id": item.scholar_id,
"display_name": item.display_name,
"affiliation": item.affiliation,
"email_domain": item.email_domain,
"cited_by_count": item.cited_by_count,
"interests": item.interests,
"profile_url": item.profile_url,
"profile_image_url": item.profile_image_url,
}
for item in parsed.candidates
],
"warnings": parsed.warnings,
},
data=_search_response_data(query, parsed),
)
@ -424,20 +477,14 @@ async def upload_scholar_image(
db_session: AsyncSession = Depends(get_db_session),
current_user: User = Depends(get_api_current_user),
):
profile = await scholar_service.get_user_scholar_by_id(
profile = await _require_user_profile(
db_session,
user_id=current_user.id,
scholar_profile_id=scholar_profile_id,
)
if profile is None:
raise ApiException(
status_code=404,
code="scholar_not_found",
message="Scholar not found.",
)
image_bytes = await _read_uploaded_image(image)
try:
image_bytes = await image.read()
updated = await scholar_service.set_profile_image_upload(
db_session,
profile=profile,
@ -452,9 +499,8 @@ async def upload_scholar_image(
code="invalid_scholar_image",
message=str(exc),
) from exc
finally:
await image.close()
image_size = len(image_bytes)
logger.info(
"api.scholars.image_uploaded",
extra={
@ -462,7 +508,7 @@ async def upload_scholar_image(
"user_id": current_user.id,
"scholar_profile_id": updated.id,
"content_type": image.content_type,
"size_bytes": len(image_bytes),
"size_bytes": image_size,
},
)
return success_payload(

159
app/api/routers/settings.py
View file

@ -11,8 +11,8 @@ from app.api.responses import success_payload
from app.api.schemas import SettingsEnvelope, SettingsUpdateRequest
from app.db.models import User
from app.db.session import get_db_session
from app.services import run_safety as run_safety_service
from app.services import user_settings as user_settings_service
from app.services.domains.ingestion import safety as run_safety_service
from app.services.domains.settings import application as user_settings_service
from app.settings import settings as settings_module
logger = logging.getLogger(__name__)
@ -21,12 +21,7 @@ router = APIRouter(prefix="/settings", tags=["api-settings"])
def _serialize_settings(user_settings) -> dict[str, object]:
min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum(
settings_module.ingestion_min_run_interval_minutes
)
min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum(
settings_module.ingestion_min_request_delay_seconds
)
min_run_interval_minutes, min_request_delay_seconds = _minimum_policy()
return {
"auto_run_enabled": bool(user_settings.auto_run_enabled) and settings_module.ingestion_automation_allowed,
"run_interval_minutes": int(user_settings.run_interval_minutes),
@ -46,6 +41,76 @@ def _serialize_settings(user_settings) -> dict[str, object]:
}
def _minimum_policy() -> tuple[int, int]:
min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum(
settings_module.ingestion_min_run_interval_minutes
)
min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum(
settings_module.ingestion_min_request_delay_seconds
)
return min_run_interval_minutes, min_request_delay_seconds
def _parse_settings_payload(payload: SettingsUpdateRequest, user_settings) -> tuple[int, int, list[str]]:
min_run_interval_minutes, min_request_delay_seconds = _minimum_policy()
parsed_interval = user_settings_service.parse_run_interval_minutes(
str(payload.run_interval_minutes),
minimum=min_run_interval_minutes,
)
parsed_delay = user_settings_service.parse_request_delay_seconds(
str(payload.request_delay_seconds),
minimum=min_request_delay_seconds,
)
parsed_nav_visible_pages = user_settings_service.parse_nav_visible_pages(
payload.nav_visible_pages
if payload.nav_visible_pages is not None
else list(user_settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES)
)
return parsed_interval, parsed_delay, parsed_nav_visible_pages
async def _clear_expired_cooldown_with_log(
db_session: AsyncSession,
*,
user_id: int,
user_settings,
) -> None:
previous_safety_state = run_safety_service.get_safety_event_context(user_settings)
if not run_safety_service.clear_expired_cooldown(user_settings):
return
await db_session.commit()
await db_session.refresh(user_settings)
logger.info(
"api.settings.safety_cooldown_cleared",
extra={
"event": "api.settings.safety_cooldown_cleared",
"user_id": user_id,
"reason": previous_safety_state.get("cooldown_reason"),
"cooldown_until": previous_safety_state.get("cooldown_until"),
"metric_name": "api_settings_safety_cooldown_cleared_total",
"metric_value": 1,
},
)
def _effective_auto_run_enabled(payload: SettingsUpdateRequest) -> bool:
return bool(payload.auto_run_enabled) and settings_module.ingestion_automation_allowed
def _log_settings_update(*, user_id: int, updated) -> None:
logger.info(
"api.settings.updated",
extra={
"event": "api.settings.updated",
"user_id": user_id,
"auto_run_enabled": updated.auto_run_enabled,
"run_interval_minutes": updated.run_interval_minutes,
"request_delay_seconds": updated.request_delay_seconds,
"nav_visible_pages": updated.nav_visible_pages,
},
)
@router.get(
"",
response_model=SettingsEnvelope,
@ -59,21 +124,11 @@ async def get_settings(
db_session,
user_id=current_user.id,
)
previous_safety_state = run_safety_service.get_safety_event_context(user_settings)
if run_safety_service.clear_expired_cooldown(user_settings):
await db_session.commit()
await db_session.refresh(user_settings)
logger.info(
"api.settings.safety_cooldown_cleared",
extra={
"event": "api.settings.safety_cooldown_cleared",
"user_id": current_user.id,
"reason": previous_safety_state.get("cooldown_reason"),
"cooldown_until": previous_safety_state.get("cooldown_until"),
"metric_name": "api_settings_safety_cooldown_cleared_total",
"metric_value": 1,
},
)
await _clear_expired_cooldown_with_log(
db_session,
user_id=current_user.id,
user_settings=user_settings,
)
return success_payload(
request,
data=_serialize_settings(user_settings),
@ -95,26 +150,10 @@ async def update_settings(
user_id=current_user.id,
)
min_run_interval_minutes = user_settings_service.resolve_run_interval_minimum(
settings_module.ingestion_min_run_interval_minutes
)
min_request_delay_seconds = user_settings_service.resolve_request_delay_minimum(
settings_module.ingestion_min_request_delay_seconds
)
try:
parsed_interval = user_settings_service.parse_run_interval_minutes(
str(payload.run_interval_minutes),
minimum=min_run_interval_minutes,
)
parsed_delay = user_settings_service.parse_request_delay_seconds(
str(payload.request_delay_seconds),
minimum=min_request_delay_seconds,
)
parsed_nav_visible_pages = user_settings_service.parse_nav_visible_pages(
payload.nav_visible_pages
if payload.nav_visible_pages is not None
else list(user_settings.nav_visible_pages or user_settings_service.DEFAULT_NAV_VISIBLE_PAGES)
parsed_interval, parsed_delay, parsed_nav_visible_pages = _parse_settings_payload(
payload,
user_settings,
)
except user_settings_service.UserSettingsServiceError as exc:
raise ApiException(
@ -123,44 +162,20 @@ async def update_settings(
message=str(exc),
) from exc
effective_auto_run_enabled = (
bool(payload.auto_run_enabled) and settings_module.ingestion_automation_allowed
)
updated = await user_settings_service.update_settings(
db_session,
settings=user_settings,
auto_run_enabled=effective_auto_run_enabled,
auto_run_enabled=_effective_auto_run_enabled(payload),
run_interval_minutes=parsed_interval,
request_delay_seconds=parsed_delay,
nav_visible_pages=parsed_nav_visible_pages,
)
previous_safety_state = run_safety_service.get_safety_event_context(updated)
if run_safety_service.clear_expired_cooldown(updated):
await db_session.commit()
await db_session.refresh(updated)
logger.info(
"api.settings.safety_cooldown_cleared",
extra={
"event": "api.settings.safety_cooldown_cleared",
"user_id": current_user.id,
"reason": previous_safety_state.get("cooldown_reason"),
"cooldown_until": previous_safety_state.get("cooldown_until"),
"metric_name": "api_settings_safety_cooldown_cleared_total",
"metric_value": 1,
},
)
logger.info(
"api.settings.updated",
extra={
"event": "api.settings.updated",
"user_id": current_user.id,
"auto_run_enabled": updated.auto_run_enabled,
"run_interval_minutes": updated.run_interval_minutes,
"request_delay_seconds": updated.request_delay_seconds,
"nav_visible_pages": updated.nav_visible_pages,
},
await _clear_expired_cooldown_with_log(
db_session,
user_id=current_user.id,
user_settings=updated,
)
_log_settings_update(user_id=current_user.id, updated=updated)
return success_payload(
request,
data=_serialize_settings(updated),

23
app/api/schemas.py
View file

@ -205,6 +205,7 @@ class PublicationExportItemData(BaseModel):
author_text: str | None = None
venue_text: str | None = None
pub_url: str | None = None
doi: str | None = None
pdf_url: str | None = None
is_read: bool = False
@ -573,6 +574,7 @@ class PublicationItemData(BaseModel):
citation_count: int
venue_text: str | None
pub_url: str | None
doi: str | None
pdf_url: str | None
is_read: bool
first_seen_at: datetime
@ -640,3 +642,24 @@ class MarkSelectedReadEnvelope(BaseModel):
meta: ApiMeta
model_config = ConfigDict(extra="forbid")
class RetryPublicationPdfRequest(BaseModel):
scholar_profile_id: int = Field(ge=1)
model_config = ConfigDict(extra="forbid")
class RetryPublicationPdfData(BaseModel):
message: str
resolved_pdf: bool
publication: PublicationItemData
model_config = ConfigDict(extra="forbid")
class RetryPublicationPdfEnvelope(BaseModel):
data: RetryPublicationPdfData
meta: ApiMeta
model_config = ConfigDict(extra="forbid")

2
app/auth/runtime.py
View file

@ -8,7 +8,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.auth.session import clear_session_user, get_session_user, set_session_user
from app.db.models import User
from app.security.csrf import CSRF_SESSION_KEY
from app.services import users as user_service
from app.services.domains.users import application as user_service
logger = logging.getLogger(__name__)

1
app/db/models.py
View file

@ -227,6 +227,7 @@ class Publication(Base):
author_text: Mapped[str | None] = mapped_column(Text)
venue_text: Mapped[str | None] = mapped_column(Text)
pub_url: Mapped[str | None] = mapped_column(Text)
doi: Mapped[str | None] = mapped_column(String(255))
pdf_url: Mapped[str | None] = mapped_column(Text)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), nullable=False, server_default=func.now()

3
app/services/continuation_queue.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.ingestion.queue import *

5
app/services/domains/crossref/__init__.py Normal file
View file

@ -0,0 +1,5 @@
from __future__ import annotations
from app.services.domains.crossref.application import discover_doi_for_publication
__all__ = ["discover_doi_for_publication"]

259
app/services/domains/crossref/application.py Normal file
View file

@ -0,0 +1,259 @@
from __future__ import annotations
import asyncio
import logging
import re
import threading
import time
from typing import TYPE_CHECKING
from crossref.restful import Etiquette, Works
from app.services.domains.doi.normalize import normalize_doi
from app.settings import settings
if TYPE_CHECKING:
from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem
TOKEN_RE = re.compile(r"[a-z0-9]+")
NON_ALNUM_RE = re.compile(r"[^a-z0-9\\s]+")
STOP_WORDS = {"the", "and", "for", "with", "from", "method", "study", "analysis"}
_RATE_LOCK = threading.Lock()
_LAST_REQUEST_AT = 0.0
logger = logging.getLogger(__name__)
def _rate_limit_wait(min_interval_seconds: float) -> None:
global _LAST_REQUEST_AT
interval = max(float(min_interval_seconds), 0.0)
with _RATE_LOCK:
elapsed = time.monotonic() - _LAST_REQUEST_AT
remaining = interval - elapsed
if remaining > 0:
time.sleep(remaining)
_LAST_REQUEST_AT = time.monotonic()
def _normalized_tokens(value: str) -> list[str]:
lowered = value.lower().replace("’", "'").replace("“", "\"").replace("”", "\"")
lowered = NON_ALNUM_RE.sub(" ", lowered)
return [token for token in TOKEN_RE.findall(lowered) if len(token) >= 3]
def _normalized_query(value: str) -> str:
tokens = [token for token in _normalized_tokens(value) if token not in STOP_WORDS]
if len(tokens) < 3:
tokens = _normalized_tokens(value)
if len(tokens) < 3:
return ""
return " ".join(tokens[:12]).strip()
def _query_author(value: str) -> str | None:
tokens = [token for token in value.strip().split() if token]
if len(tokens) < 2:
return None
return " ".join(tokens[:2])[:64]
def _author_surname(value: str) -> str | None:
tokens = [token for token in value.strip().split() if token]
if not tokens:
return None
return NON_ALNUM_RE.sub("", tokens[-1].lower()) or None
def _query_filters(year: int | None) -> list[tuple[str, str] | None]:
if year is None:
return [None]
return [
(f"{year - 1}-01-01", f"{year + 1}-12-31"),
(f"{year}-01-01", f"{year}-12-31"),
None,
]
def _candidate_title(item: dict) -> str:
titles = item.get("title")
if isinstance(titles, list) and titles:
return str(titles[0] or "")
return str(item.get("title") or "")
def _title_match_score(source: str, candidate: str) -> float:
source_tokens = {token for token in _normalized_tokens(source) if len(token) >= 3}
candidate_tokens = {token for token in _normalized_tokens(candidate) if len(token) >= 3}
if not source_tokens or not candidate_tokens:
return 0.0
return len(source_tokens & candidate_tokens) / float(len(source_tokens))
def _candidate_year(item: dict) -> int | None:
issued = item.get("issued")
if not isinstance(issued, dict):
return None
date_parts = issued.get("date-parts")
if not isinstance(date_parts, list) or not date_parts:
return None
first = date_parts[0]
if not isinstance(first, list) or not first:
return None
try:
return int(first[0])
except (TypeError, ValueError):
return None
def _candidate_author_match(item: dict, surname: str | None) -> bool:
if not surname:
return True
authors = item.get("author")
if not isinstance(authors, list):
return False
for author in authors:
if not isinstance(author, dict):
continue
family = NON_ALNUM_RE.sub("", str(author.get("family") or "").lower())
if family and family == surname:
return True
return False
def _candidate_rank(*, title: str, year: int | None, item: dict) -> tuple[float, str | None]:
doi = normalize_doi(str(item.get("DOI") or ""))
if doi is None:
return 0.0, None
score = _title_match_score(title, _candidate_title(item))
candidate_year = _candidate_year(item)
if year is not None and candidate_year is not None:
if abs(year - candidate_year) > 1:
return 0.0, None
score += 0.1
return score, doi
def _best_candidate_doi(
*,
title: str,
year: int | None,
items: list[dict],
author_surname: str | None,
) -> str | None:
best_score = 0.0
best_doi: str | None = None
best_year: int | None = None
for item in items:
if not isinstance(item, dict):
continue
if not _candidate_author_match(item, author_surname):
continue
score, doi = _candidate_rank(title=title, year=year, item=item)
candidate_year = _candidate_year(item)
if doi is None or score < 0.75:
continue
if score > best_score:
best_score = score
best_doi = doi
best_year = candidate_year
continue
if abs(score - best_score) > 0.02:
continue
if best_year is None or candidate_year is None:
continue
if candidate_year < best_year:
best_doi = doi
best_year = candidate_year
return best_doi
def _works_client(email: str | None) -> Works:
if email:
etiquette = Etiquette(settings.app_name, "0.1.0", "https://scholarr.local", email)
return Works(etiquette=etiquette)
return Works()
def _fetch_items_sync(
*,
query: str,
author: str | None,
date_range: tuple[str, str] | None,
max_rows: int,
email: str | None,
min_interval_seconds: float,
) -> list[dict]:
_rate_limit_wait(min_interval_seconds)
works = _works_client(email)
params = {"bibliographic": query}
if author:
params["author"] = author
request = works.query(**params)
if date_range is not None:
from_date, until_date = date_range
request = request.filter(from_pub_date=from_date, until_pub_date=until_date)
request = request.select(["DOI", "title", "issued", "score", "author"])
items: list[dict] = []
for entry in request:
if isinstance(entry, dict):
items.append(entry)
if len(items) >= max(max_rows, 1):
break
return items
async def _fetch_items(
*,
query: str,
author: str | None,
date_range: tuple[str, str] | None,
max_rows: int,
email: str | None,
) -> list[dict]:
timeout = max(float(settings.crossref_timeout_seconds), 0.5)
try:
return await asyncio.wait_for(
asyncio.to_thread(
_fetch_items_sync,
query=query,
author=author,
date_range=date_range,
max_rows=max_rows,
email=email,
min_interval_seconds=settings.crossref_min_interval_seconds,
),
timeout=timeout,
)
except Exception:
return []
async def discover_doi_for_publication(
*,
item: PublicationListItem | UnreadPublicationItem,
max_rows: int = 10,
email: str | None = None,
) -> str | None:
title = (item.title or "").strip()
query = _normalized_query(title)
if not query:
return None
author = _query_author(item.scholar_label)
author_surname = _author_surname(item.scholar_label)
for date_range in _query_filters(item.year):
items = await _fetch_items(
query=query,
author=author,
date_range=date_range,
max_rows=max_rows,
email=email,
)
doi = _best_candidate_doi(
title=title,
year=item.year,
items=items,
author_surname=author_surname,
)
if doi:
logger.debug("crossref.doi_discovered", extra={"event": "crossref.doi_discovered"})
return doi
return None

23
app/services/domains/doi/normalize.py Normal file
View file

@ -0,0 +1,23 @@
from __future__ import annotations
import re
from urllib.parse import unquote
DOI_RE = re.compile(r"10\.\d{4,9}/[-._;()/:A-Z0-9]+", re.I)
def normalize_doi(value: str | None) -> str | None:
if not value:
return None
match = DOI_RE.search(unquote(value))
if not match:
return None
return match.group(0).rstrip(" .;,)").lower()
def first_doi_from_texts(*values: str | None) -> str | None:
for value in values:
doi = normalize_doi(value)
if doi:
return doi
return None

44
app/services/domains/ingestion/application.py
View file

@ -28,6 +28,7 @@ from app.services.domains.ingestion.constants import (
RESUMABLE_PARTIAL_REASONS,
RUN_LOCK_NAMESPACE,
)
from app.services.domains.doi.normalize import first_doi_from_texts
from app.services.domains.ingestion.fingerprints import (
_build_body_excerpt,
_dedupe_publication_candidates,
@ -55,6 +56,7 @@ from app.services.domains.scholar.parser import (
ParseState,
ParsedProfilePage,
PublicationCandidate,
ScholarParserError,
parse_profile_page,
)
from app.services.domains.scholar.source import FetchResult, ScholarSource
@ -1431,7 +1433,7 @@ class ScholarIngestionService:
cstart=cstart,
page_size=page_size,
)
parsed_page = parse_profile_page(fetch_result)
parsed_page = self._parse_profile_page_or_layout_error(fetch_result=fetch_result)
attempt_log.append(
self._attempt_log_entry(
attempt=attempt_index + 1,
@ -1458,6 +1460,38 @@ class ScholarIngestionService:
raise RuntimeError("Fetch-and-parse retry loop produced no result.")
return fetch_result, parsed_page, attempt_log
def _parse_profile_page_or_layout_error(
self,
*,
fetch_result: FetchResult,
) -> ParsedProfilePage:
try:
return parse_profile_page(fetch_result)
except ScholarParserError as exc:
return self._parsed_page_from_parser_error(
fetch_result=fetch_result,
code=exc.code,
)
@staticmethod
def _parsed_page_from_parser_error(
*,
fetch_result: FetchResult,
code: str,
) -> ParsedProfilePage:
return ParsedProfilePage(
state=ParseState.LAYOUT_CHANGED,
state_reason=code,
profile_name=None,
profile_image_url=None,
publications=[],
marker_counts={},
warnings=[code],
has_show_more_button=False,
has_operation_error_banner=False,
articles_range=None,
)
@staticmethod
def _page_log_entry(
*,
@ -1942,7 +1976,8 @@ class ScholarIngestionService:
author_text=candidate.authors_text,
venue_text=candidate.venue_text,
pub_url=build_publication_url(candidate.title_url),
pdf_url=build_publication_url(candidate.pdf_url),
doi=first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title),
pdf_url=None,
)
db_session.add(publication)
await db_session.flush()
@ -1976,8 +2011,9 @@ class ScholarIngestionService:
publication.venue_text = candidate.venue_text
if candidate.title_url:
publication.pub_url = build_publication_url(candidate.title_url)
if candidate.pdf_url:
publication.pdf_url = build_publication_url(candidate.pdf_url)
local_doi = first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title)
if local_doi and not publication.doi:
publication.doi = local_doi
async def _resolve_publication(
self,

3
app/services/domains/portability/exporting.py
View file

@ -34,6 +34,7 @@ def _serialize_export_publication(row: tuple[Any, ...]) -> dict[str, Any]:
author_text,
venue_text,
pub_url,
doi,
pdf_url,
is_read,
) = row
@ -47,6 +48,7 @@ def _serialize_export_publication(row: tuple[Any, ...]) -> dict[str, Any]:
"author_text": author_text,
"venue_text": venue_text,
"pub_url": pub_url,
"doi": doi,
"pdf_url": pdf_url,
"is_read": bool(is_read),
}
@ -73,6 +75,7 @@ async def export_user_data(
Publication.author_text,
Publication.venue_text,
Publication.pub_url,
Publication.doi,
Publication.pdf_url,
ScholarPublication.is_read,
)

10
app/services/domains/portability/publication_import.py
View file

@ -7,6 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import Publication, ScholarProfile, ScholarPublication
from app.services.domains.ingestion.application import build_publication_url, normalize_title
from app.services.domains.doi.normalize import normalize_doi
from app.services.domains.portability.normalize import (
_normalize_citation_count,
_normalize_optional_text,
@ -70,6 +71,7 @@ def _apply_imported_publication_values(
author_text: str | None,
venue_text: str | None,
pub_url: str | None,
doi: str | None,
pdf_url: str | None,
cluster_id: str | None,
) -> bool:
@ -96,6 +98,9 @@ def _apply_imported_publication_values(
if pub_url and publication.pub_url != pub_url:
publication.pub_url = pub_url
updated = True
if doi and publication.doi != doi:
publication.doi = doi
updated = True
if pdf_url and publication.pdf_url != pdf_url:
publication.pdf_url = pdf_url
updated = True
@ -112,6 +117,7 @@ def _new_publication(
author_text: str | None,
venue_text: str | None,
pub_url: str | None,
doi: str | None,
pdf_url: str | None,
) -> Publication:
return Publication(
@ -124,6 +130,7 @@ def _new_publication(
author_text=author_text,
venue_text=venue_text,
pub_url=pub_url,
doi=doi,
pdf_url=pdf_url,
)
@ -226,6 +233,7 @@ def _build_imported_publication_input(
venue_text=venue_text,
cluster_id=_normalize_optional_text(item.get("cluster_id")),
pub_url=build_publication_url(_normalize_optional_text(item.get("pub_url"))),
doi=normalize_doi(_normalize_optional_text(item.get("doi"))),
pdf_url=build_publication_url(_normalize_optional_text(item.get("pdf_url"))),
fingerprint=_resolve_fingerprint(
title=title,
@ -277,6 +285,7 @@ async def _create_import_publication(
author_text=payload.author_text,
venue_text=payload.venue_text,
pub_url=payload.pub_url,
doi=payload.doi,
pdf_url=payload.pdf_url,
)
db_session.add(publication)
@ -297,6 +306,7 @@ def _update_import_publication(
author_text=payload.author_text,
venue_text=payload.venue_text,
pub_url=payload.pub_url,
doi=payload.doi,
pdf_url=payload.pdf_url,
cluster_id=payload.cluster_id,
)

1
app/services/domains/portability/types.py
View file

@ -19,6 +19,7 @@ class ImportedPublicationInput:
venue_text: str | None
cluster_id: str | None
pub_url: str | None
doi: str | None
pdf_url: str | None
fingerprint: str
is_read: bool

8
app/services/domains/publications/application.py
View file

@ -8,8 +8,12 @@ from app.services.domains.publications.counts import (
from app.services.domains.publications.listing import (
list_for_user,
list_new_for_latest_run_for_user,
retry_pdf_for_user,
list_unread_for_user,
)
from app.services.domains.publications.enrichment import (
schedule_missing_pdf_enrichment_for_user,
)
from app.services.domains.publications.modes import (
MODE_ALL,
MODE_LATEST,
@ -20,6 +24,7 @@ from app.services.domains.publications.modes import (
)
from app.services.domains.publications.queries import (
get_latest_completed_run_id_for_user,
get_publication_item_for_user,
publications_query,
)
from app.services.domains.publications.read_state import (
@ -39,9 +44,12 @@ __all__ = [
"resolve_mode",
"get_latest_completed_run_id_for_user",
"publications_query",
"get_publication_item_for_user",
"list_for_user",
"list_unread_for_user",
"list_new_for_latest_run_for_user",
"retry_pdf_for_user",
"schedule_missing_pdf_enrichment_for_user",
"count_for_user",
"count_unread_for_user",
"count_latest_for_user",

135
app/services/domains/publications/enrichment.py Normal file
View file

@ -0,0 +1,135 @@
from __future__ import annotations
import asyncio
import logging
import time
from app.db.session import get_session_factory
from app.services.domains.publications.listing import (
missing_pdf_items,
resolve_and_persist_oa_metadata,
)
from app.services.domains.publications.types import PublicationListItem
from app.settings import settings
logger = logging.getLogger(__name__)
_enrichment_lock = asyncio.Lock()
_inflight_publications: set[tuple[int, int]] = set()
_recent_attempt_seconds: dict[tuple[int, int], float] = {}
_scheduled_tasks: set[asyncio.Task[None]] = set()
def _cooldown_seconds() -> float:
return max(float(settings.unpaywall_retry_cooldown_seconds), 1.0)
def _prune_recent_attempts(now_seconds: float, *, cooldown_seconds: float) -> None:
expiry = cooldown_seconds * 3
stale_keys = [
key for key, attempted_seconds in _recent_attempt_seconds.items()
if now_seconds - attempted_seconds >= expiry
]
for key in stale_keys:
_recent_attempt_seconds.pop(key, None)
async def _claim_items(
*,
user_id: int,
items: list[PublicationListItem],
max_items: int,
) -> list[PublicationListItem]:
candidates = missing_pdf_items(items, limit=max_items)
if not candidates:
return []
now_seconds = time.monotonic()
cooldown_seconds = _cooldown_seconds()
claimed: list[PublicationListItem] = []
async with _enrichment_lock:
_prune_recent_attempts(now_seconds, cooldown_seconds=cooldown_seconds)
for item in candidates:
key = (user_id, item.publication_id)
attempted_seconds = _recent_attempt_seconds.get(key)
if key in _inflight_publications:
continue
if attempted_seconds is not None and now_seconds - attempted_seconds < cooldown_seconds:
continue
_inflight_publications.add(key)
_recent_attempt_seconds[key] = now_seconds
claimed.append(item)
return claimed
async def _release_claims(*, user_id: int, publication_ids: list[int]) -> None:
async with _enrichment_lock:
for publication_id in publication_ids:
_inflight_publications.discard((user_id, publication_id))
def _on_task_done(task: asyncio.Task[None]) -> None:
_scheduled_tasks.discard(task)
try:
task.result()
except Exception:
logger.exception(
"publications.enrichment.task_failed",
extra={"event": "publications.enrichment.task_failed"},
)
async def _run_enrichment(
*,
user_id: int,
request_email: str | None,
items: list[PublicationListItem],
) -> None:
publication_ids = [item.publication_id for item in items]
try:
session_factory = get_session_factory()
async with session_factory() as db_session:
await resolve_and_persist_oa_metadata(
db_session,
rows=items,
unpaywall_email=request_email,
)
finally:
await _release_claims(user_id=user_id, publication_ids=publication_ids)
logger.info(
"publications.enrichment.completed",
extra={
"event": "publications.enrichment.completed",
"user_id": user_id,
"publication_count": len(items),
},
)
async def schedule_missing_pdf_enrichment_for_user(
*,
user_id: int,
request_email: str | None,
items: list[PublicationListItem],
max_items: int,
) -> int:
claimed_items = await _claim_items(user_id=user_id, items=items, max_items=max_items)
if not claimed_items:
return 0
task = asyncio.create_task(
_run_enrichment(
user_id=user_id,
request_email=request_email,
items=claimed_items,
)
)
_scheduled_tasks.add(task)
task.add_done_callback(_on_task_done)
logger.info(
"publications.enrichment.scheduled",
extra={
"event": "publications.enrichment.scheduled",
"user_id": user_id,
"publication_count": len(claimed_items),
},
)
return len(claimed_items)

146
app/services/domains/publications/listing.py
View file

@ -10,11 +10,100 @@ from app.services.domains.publications.modes import (
)
from app.services.domains.publications.queries import (
get_latest_completed_run_id_for_user,
get_publication_item_for_user,
publication_list_item_from_row,
publications_query,
unread_item_from_row,
)
from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem
from app.services.domains.unpaywall.application import resolve_publication_oa_metadata
from sqlalchemy import update
from app.db.models import Publication
def _with_oa_overrides(
rows: list[PublicationListItem],
oa_data: dict[int, tuple[str | None, str | None]],
) -> list[PublicationListItem]:
return [
PublicationListItem(
publication_id=row.publication_id,
scholar_profile_id=row.scholar_profile_id,
scholar_label=row.scholar_label,
title=row.title,
year=row.year,
citation_count=row.citation_count,
venue_text=row.venue_text,
pub_url=row.pub_url,
doi=(oa_data.get(row.publication_id) or (None, None))[0] or row.doi,
pdf_url=(oa_data.get(row.publication_id) or (None, None))[1] or row.pdf_url,
is_read=row.is_read,
first_seen_at=row.first_seen_at,
is_new_in_latest_run=row.is_new_in_latest_run,
)
for row in rows
]
def _resolved_fields(
*,
row: PublicationListItem,
resolved: tuple[str | None, str | None] | None,
) -> tuple[str | None, str | None]:
if resolved is None:
return row.doi, row.pdf_url
resolved_doi, resolved_pdf = resolved
return resolved_doi or row.doi, resolved_pdf or row.pdf_url
async def _persist_resolved_metadata(
db_session: AsyncSession,
*,
rows: list[PublicationListItem],
oa_data: dict[int, tuple[str | None, str | None]],
) -> None:
by_id = {row.publication_id: row for row in rows}
updates: list[tuple[int, str | None, str | None]] = []
for publication_id, resolved in oa_data.items():
existing = by_id.get(publication_id)
if existing is None:
continue
next_doi, next_pdf = _resolved_fields(row=existing, resolved=resolved)
if next_doi == existing.doi and next_pdf == existing.pdf_url:
continue
updates.append((publication_id, next_doi, next_pdf))
for publication_id, doi, pdf_url in updates:
await db_session.execute(
update(Publication)
.where(Publication.id == publication_id)
.values(doi=doi, pdf_url=pdf_url)
)
if updates:
await db_session.commit()
def missing_pdf_items(
rows: list[PublicationListItem],
*,
limit: int,
) -> list[PublicationListItem]:
bounded_limit = max(0, int(limit))
if bounded_limit == 0:
return []
return [row for row in rows if not row.pdf_url][:bounded_limit]
async def resolve_and_persist_oa_metadata(
db_session: AsyncSession,
*,
rows: list[PublicationListItem],
unpaywall_email: str | None = None,
) -> dict[int, tuple[str | None, str | None]]:
if not rows:
return {}
oa_data = await resolve_publication_oa_metadata(rows, request_email=unpaywall_email)
await _persist_resolved_metadata(db_session, rows=rows, oa_data=oa_data)
return oa_data
async def list_for_user(
@ -36,10 +125,35 @@ async def list_for_user(
limit=limit,
)
)
return [
rows = [
publication_list_item_from_row(row, latest_run_id=latest_run_id)
for row in result.all()
]
return rows
async def retry_pdf_for_user(
db_session: AsyncSession,
*,
user_id: int,
scholar_profile_id: int,
publication_id: int,
unpaywall_email: str | None = None,
) -> PublicationListItem | None:
item = await get_publication_item_for_user(
db_session,
user_id=user_id,
scholar_profile_id=scholar_profile_id,
publication_id=publication_id,
)
if item is None:
return None
oa_data = await resolve_and_persist_oa_metadata(
db_session,
rows=[item],
unpaywall_email=unpaywall_email,
)
return _with_oa_overrides([item], oa_data)[0]
async def list_unread_for_user(
@ -73,17 +187,19 @@ async def list_new_for_latest_run_for_user(
scholar_profile_id=None,
limit=limit,
)
return [
UnreadPublicationItem(
publication_id=row.publication_id,
scholar_profile_id=row.scholar_profile_id,
scholar_label=row.scholar_label,
title=row.title,
year=row.year,
citation_count=row.citation_count,
venue_text=row.venue_text,
pub_url=row.pub_url,
pdf_url=row.pdf_url,
)
for row in rows
]
return [_to_unread_item(row) for row in rows]
def _to_unread_item(row: PublicationListItem) -> UnreadPublicationItem:
return UnreadPublicationItem(
publication_id=row.publication_id,
scholar_profile_id=row.scholar_profile_id,
scholar_label=row.scholar_label,
title=row.title,
year=row.year,
citation_count=row.citation_count,
venue_text=row.venue_text,
pub_url=row.pub_url,
doi=row.doi,
pdf_url=row.pdf_url,
)

60
app/services/domains/publications/queries.py
View file

@ -50,6 +50,7 @@ def publications_query(
Publication.citation_count,
Publication.venue_text,
Publication.pub_url,
Publication.doi,
Publication.pdf_url,
ScholarPublication.is_read,
ScholarPublication.first_seen_run_id,
@ -72,6 +73,61 @@ def publications_query(
return stmt
def publication_query_for_user(
*,
user_id: int,
scholar_profile_id: int,
publication_id: int,
) -> Select[tuple]:
return (
select(
Publication.id,
ScholarProfile.id,
ScholarProfile.display_name,
ScholarProfile.scholar_id,
Publication.title_raw,
Publication.year,
Publication.citation_count,
Publication.venue_text,
Publication.pub_url,
Publication.doi,
Publication.pdf_url,
ScholarPublication.is_read,
ScholarPublication.first_seen_run_id,
ScholarPublication.created_at,
)
.join(ScholarPublication, ScholarPublication.publication_id == Publication.id)
.join(ScholarProfile, ScholarProfile.id == ScholarPublication.scholar_profile_id)
.where(
ScholarProfile.user_id == user_id,
ScholarProfile.id == scholar_profile_id,
Publication.id == publication_id,
)
.limit(1)
)
async def get_publication_item_for_user(
db_session: AsyncSession,
*,
user_id: int,
scholar_profile_id: int,
publication_id: int,
) -> PublicationListItem | None:
latest_run_id = await get_latest_completed_run_id_for_user(db_session, user_id=user_id)
result = await db_session.execute(
publication_query_for_user(
user_id=user_id,
scholar_profile_id=scholar_profile_id,
publication_id=publication_id,
)
)
row = result.one_or_none()
if row is None:
return None
return publication_list_item_from_row(row, latest_run_id=latest_run_id)
def publication_list_item_from_row(
row: tuple,
*,
@ -87,6 +143,7 @@ def publication_list_item_from_row(
citation_count,
venue_text,
pub_url,
doi,
pdf_url,
is_read,
first_seen_run_id,
@ -101,6 +158,7 @@ def publication_list_item_from_row(
citation_count=_normalized_citation_count(citation_count),
venue_text=venue_text,
pub_url=pub_url,
doi=doi,
pdf_url=pdf_url,
is_read=bool(is_read),
first_seen_at=created_at,
@ -121,6 +179,7 @@ def unread_item_from_row(row: tuple) -> UnreadPublicationItem:
citation_count,
venue_text,
pub_url,
doi,
pdf_url,
_is_read,
_first_seen_run_id,
@ -135,5 +194,6 @@ def unread_item_from_row(row: tuple) -> UnreadPublicationItem:
citation_count=_normalized_citation_count(citation_count),
venue_text=venue_text,
pub_url=pub_url,
doi=doi,
pdf_url=pdf_url,
)

2
app/services/domains/publications/types.py
View file

@ -14,6 +14,7 @@ class PublicationListItem:
citation_count: int
venue_text: str | None
pub_url: str | None
doi: str | None
pdf_url: str | None
is_read: bool
first_seen_at: datetime
@ -30,4 +31,5 @@ class UnreadPublicationItem:
citation_count: int
venue_text: str | None
pub_url: str | None
doi: str | None
pdf_url: str | None

65
app/services/domains/scholar/parser.py
View file

@ -11,14 +11,12 @@ from app.services.domains.scholar.parser_types import (
ParsedAuthorSearchPage,
ParsedProfilePage,
PublicationCandidate,
ScholarDomInvariantError,
ScholarMalformedDataError,
ScholarParserError,
ScholarSearchCandidate,
)
from app.services.domains.scholar.parser_utils import (
attr_class,
attr_href,
attr_src,
build_absolute_scholar_url,
normalize_space,
strip_tags,
)
from app.services.domains.scholar.profile_rows import (
@ -44,6 +42,55 @@ from app.services.domains.scholar.state_detection import (
)
def _raise_dom_error(code: str, message: str) -> None:
raise ScholarDomInvariantError(code=code, message=message)
def _assert_profile_dom_invariants(
*,
fetch_result: FetchResult,
marker_counts: dict[str, int],
publications: list[PublicationCandidate],
warnings: list[str],
has_show_more_button_flag: bool,
articles_range: str | None,
) -> None:
if fetch_result.status_code is None:
return
final_url = (fetch_result.final_url or "").lower()
if "accounts.google.com" in final_url or "sorry/index" in final_url:
return
if any(code.startswith("layout_") for code in warnings):
reason = next(code for code in warnings if code.startswith("layout_"))
_raise_dom_error(reason, f"Detected layout warning: {reason}")
if has_show_more_button_flag and not articles_range:
_raise_dom_error(
"layout_show_more_without_articles_range",
"Show-more control exists without an articles range marker.",
)
if marker_counts.get("gsc_a_tr", 0) > 0 and marker_counts.get("gsc_a_at", 0) <= 0:
_raise_dom_error(
"layout_missing_publication_title_anchor",
"Publication rows were present but title anchors were absent.",
)
if not publications:
has_profile_markers = marker_counts.get("gsc_prf_in", 0) > 0
has_table_markers = marker_counts.get("gsc_a_tr", 0) > 0 or marker_counts.get("gsc_a_at", 0) > 0
if not has_profile_markers and not has_table_markers:
_raise_dom_error("layout_markers_missing", "Expected scholar profile markers were absent.")
for publication in publications:
if not publication.title.strip():
raise ScholarMalformedDataError(
code="malformed_publication_title",
message="Encountered a publication candidate with an empty title.",
)
if publication.citation_count is not None and int(publication.citation_count) < 0:
raise ScholarMalformedDataError(
code="malformed_publication_negative_citations",
message="Encountered a publication candidate with negative citations.",
)
def parse_profile_page(fetch_result: FetchResult) -> ParsedProfilePage:
publications, warnings = parse_publications(fetch_result.body)
marker_counts = count_markers(fetch_result.body)
@ -59,6 +106,14 @@ def parse_profile_page(fetch_result: FetchResult) -> ParsedProfilePage:
warnings.append("operation_error_banner_present")
warnings = sorted(set(warnings))
_assert_profile_dom_invariants(
fetch_result=fetch_result,
marker_counts=marker_counts,
publications=publications,
warnings=warnings,
has_show_more_button_flag=show_more,
articles_range=articles_range,
)
state, state_reason = detect_state(
fetch_result,

16
app/services/domains/scholar/parser_types.py
View file

@ -12,6 +12,22 @@ class ParseState(StrEnum):
NETWORK_ERROR = "network_error"
class ScholarParserError(RuntimeError):
code: str
def __init__(self, *, code: str, message: str) -> None:
super().__init__(message)
self.code = code
class ScholarDomInvariantError(ScholarParserError):
pass
class ScholarMalformedDataError(ScholarParserError):
pass
@dataclass(frozen=True)
class PublicationCandidate:
title: str

56
app/services/domains/scholar/profile_rows.py
View file

@ -2,13 +2,10 @@ from __future__ import annotations
import re
from html.parser import HTMLParser
from typing import Any
from urllib.parse import parse_qs, urlparse
from app.services.domains.scholar.parser_constants import (
MARKER_KEYS,
PROFILE_ROW_DIRECT_LABEL_TOKENS,
PROFILE_ROW_PARSER_DIRECT_MARKERS,
SHOW_MORE_BUTTON_RE,
)
from app.services.domains.scholar.parser_types import PublicationCandidate
@ -26,7 +23,6 @@ class ScholarRowParser(HTMLParser):
def __init__(self) -> None:
super().__init__(convert_charrefs=True)
self.title_href: str | None = None
self.direct_download_href: str | None = None
self.title_parts: list[str] = []
self.citation_parts: list[str] = []
self.year_parts: list[str] = []
@ -35,14 +31,7 @@ class ScholarRowParser(HTMLParser):
self._title_depth = 0
self._citation_depth = 0
self._year_depth = 0
self._gray_stack: list[dict[str, Any]] = []
self._direct_marker_depth = 0
self._aux_link_stack: list[dict[str, Any]] = []
@staticmethod
def _contains_direct_marker(classes: str) -> bool:
lowered = classes.lower()
return any(marker in lowered for marker in PROFILE_ROW_PARSER_DIRECT_MARKERS)
self._gray_stack: list[dict[str, object]] = []
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
if self._title_depth > 0:
@ -53,14 +42,8 @@ class ScholarRowParser(HTMLParser):
self._year_depth += 1
if self._gray_stack:
self._gray_stack[-1]["depth"] += 1
if self._direct_marker_depth > 0:
self._direct_marker_depth += 1
if self._aux_link_stack:
self._aux_link_stack[-1]["depth"] += 1
classes = attr_class(attrs)
if tag in {"div", "span"} and self._contains_direct_marker(classes):
self._direct_marker_depth = 1
if tag == "a" and "gsc_a_at" in classes:
self._title_depth = 1
@ -79,17 +62,6 @@ class ScholarRowParser(HTMLParser):
self._gray_stack.append({"depth": 1, "parts": []})
return
if tag == "a":
href = attr_href(attrs)
self._aux_link_stack.append(
{
"depth": 1,
"classes": classes,
"href": href,
"parts": [],
}
)
def handle_data(self, data: str) -> None:
if self._title_depth > 0:
self.title_parts.append(data)
@ -99,23 +71,6 @@ class ScholarRowParser(HTMLParser):
self.year_parts.append(data)
if self._gray_stack:
self._gray_stack[-1]["parts"].append(data)
if self._aux_link_stack:
self._aux_link_stack[-1]["parts"].append(data)
def _capture_direct_download_href(self, link_info: dict[str, Any]) -> None:
if self.direct_download_href:
return
href = link_info.get("href")
if not href:
return
classes = str(link_info.get("classes") or "")
label = normalize_space("".join(link_info.get("parts") or [])).lower()
if "gs_ggsd" in classes or "gs_ggs" in classes or "gs_ggsa" in classes:
self.direct_download_href = href
return
label_match = any(token in label for token in PROFILE_ROW_DIRECT_LABEL_TOKENS)
if self._direct_marker_depth > 0 and label_match:
self.direct_download_href = href
def handle_endtag(self, tag: str) -> None:
if self._title_depth > 0:
@ -131,13 +86,6 @@ class ScholarRowParser(HTMLParser):
if text_value:
self.gray_texts.append(text_value)
self._gray_stack.pop()
if self._aux_link_stack:
self._aux_link_stack[-1]["depth"] -= 1
if self._aux_link_stack[-1]["depth"] == 0:
link_info = self._aux_link_stack.pop()
self._capture_direct_download_href(link_info)
if self._direct_marker_depth > 0:
self._direct_marker_depth -= 1
def extract_rows(html: str) -> list[str]:
@ -223,7 +171,7 @@ def _parse_publication_row(row_html: str) -> tuple[PublicationCandidate | None,
citation_count=citation_count,
authors_text=authors_text,
venue_text=venue_text,
pdf_url=build_absolute_scholar_url(parser.direct_download_href),
pdf_url=None,
),
warnings,
)

17
app/services/domains/scholars/application.py
View file

@ -18,6 +18,7 @@ from app.db.models import AuthorSearchCacheEntry, AuthorSearchRuntimeState, Scho
from app.services.domains.scholar.parser import (
ParseState,
ParsedAuthorSearchPage,
ScholarParserError,
parse_author_search_page,
parse_profile_page,
)
@ -635,7 +636,16 @@ async def _fetch_author_search_with_retries(
retry_scheduled_count = 0
for attempt_index in range(max_attempts):
fetch_result = await source.fetch_author_search_html(normalized_query, start=0)
parsed = parse_author_search_page(fetch_result)
try:
parsed = parse_author_search_page(fetch_result)
except ScholarParserError as exc:
parsed = ParsedAuthorSearchPage(
state=ParseState.LAYOUT_CHANGED,
state_reason=exc.code,
candidates=[],
marker_counts={},
warnings=[exc.code],
)
if parsed.state != ParseState.NETWORK_ERROR or attempt_index >= max_attempts - 1:
break
retry_warnings.append("network_retry_scheduled_for_author_search")
@ -873,7 +883,10 @@ async def hydrate_profile_metadata(
source: ScholarSource,
) -> ScholarProfile:
fetch_result = await source.fetch_profile_html(profile.scholar_id)
parsed_page = parse_profile_page(fetch_result)
try:
parsed_page = parse_profile_page(fetch_result)
except ScholarParserError:
return profile
if parsed_page.profile_name and not (profile.display_name or "").strip():
profile.display_name = parsed_page.profile_name

5
app/services/domains/unpaywall/__init__.py Normal file
View file

@ -0,0 +1,5 @@
from __future__ import annotations
from app.services.domains.unpaywall.application import resolve_publication_pdf_urls
__all__ = ["resolve_publication_pdf_urls"]

218
app/services/domains/unpaywall/application.py Normal file
View file

@ -0,0 +1,218 @@
from __future__ import annotations
import logging
import re
from urllib.parse import unquote
from app.services.domains.crossref.application import discover_doi_for_publication
from app.services.domains.publications.types import PublicationListItem, UnreadPublicationItem
from app.services.domains.doi.normalize import normalize_doi
from app.settings import settings
DOI_PATTERN = re.compile(r"10\.\d{4,9}/[-._;()/:A-Z0-9]+", re.I)
DOI_PREFIX_RE = re.compile(r"\bdoi\s*[:=]\s*(10\.\d{4,9}/[-._;()/:A-Z0-9]+)", re.I)
DOI_URL_RE = re.compile(r"(?:https?://)?(?:dx\.)?doi\.org/(10\.\d{4,9}/[-._;()/:A-Z0-9]+)", re.I)
UNPAYWALL_URL_TEMPLATE = "https://api.unpaywall.org/v2/{doi}"
logger = logging.getLogger(__name__)
def _extract_doi_candidate(text: str | None) -> str | None:
if not text:
return None
decoded = unquote(text)
match = DOI_PATTERN.search(decoded)
if not match:
return None
return match.group(0).rstrip(" .;,)")
def _extract_explicit_doi(text: str | None) -> str | None:
if not text:
return None
decoded = unquote(text)
url_match = DOI_URL_RE.search(decoded)
if url_match:
return normalize_doi(url_match.group(1))
prefix_match = DOI_PREFIX_RE.search(decoded)
if prefix_match:
return normalize_doi(prefix_match.group(1))
return None
def _publication_doi(item: PublicationListItem | UnreadPublicationItem) -> str | None:
stored = normalize_doi(item.doi)
if stored:
in_metadata = any(
normalize_doi(_extract_explicit_doi(value)) == stored
for value in (item.pub_url, item.venue_text)
)
if in_metadata:
return stored
pub_url_doi = _extract_doi_candidate(item.pub_url)
if pub_url_doi:
return normalize_doi(pub_url_doi)
return (
_extract_explicit_doi(item.pub_url)
or _extract_explicit_doi(item.venue_text)
)
def _payload_pdf_url(payload: dict) -> str | None:
best = payload.get("best_oa_location")
if isinstance(best, dict):
pdf_url = best.get("url_for_pdf")
if isinstance(pdf_url, str) and pdf_url.strip():
return pdf_url.strip()
locations = payload.get("oa_locations")
if not isinstance(locations, list):
return None
for location in locations:
if not isinstance(location, dict):
continue
pdf_url = location.get("url_for_pdf")
if isinstance(pdf_url, str) and pdf_url.strip():
return pdf_url.strip()
return None
async def _fetch_unpaywall_payload_by_doi(
*,
client,
doi: str,
email: str,
) -> dict | None:
response = await client.get(
UNPAYWALL_URL_TEMPLATE.format(doi=doi),
params={"email": email},
)
if response.status_code != 200:
return None
payload = response.json()
if not isinstance(payload, dict):
return None
return payload
def _email_for_request(request_email: str | None) -> str | None:
email = (request_email or "").strip() or settings.unpaywall_email.strip()
return email or None
def _log_resolution_summary(
*,
publication_count: int,
doi_input_count: int,
search_attempt_count: int,
resolved_pdf_count: int,
email: str,
) -> None:
logger.info(
"unpaywall.resolve_completed",
extra={
"event": "unpaywall.resolve_completed",
"publication_count": publication_count,
"doi_input_count": doi_input_count,
"search_attempt_count": search_attempt_count,
"resolved_pdf_count": resolved_pdf_count,
"email_domain": email.split("@", 1)[-1] if "@" in email else None,
},
)
async def _resolve_item_payload(
*,
client,
item: PublicationListItem,
email: str,
allow_crossref: bool,
) -> tuple[dict | None, bool]:
doi = _publication_doi(item)
payload: dict | None = None
if doi:
payload = await _fetch_unpaywall_payload_by_doi(client=client, doi=doi, email=email)
if payload is not None and _payload_pdf_url(payload):
return payload, False
if not allow_crossref or not settings.crossref_enabled:
return payload, False
crossref_doi = await discover_doi_for_publication(
item=item,
max_rows=settings.crossref_max_rows,
email=email,
)
if crossref_doi is None or crossref_doi == doi:
return payload, crossref_doi is not None
crossref_payload = await _fetch_unpaywall_payload_by_doi(
client=client,
doi=crossref_doi,
email=email,
)
if crossref_payload is not None:
return crossref_payload, True
return payload, True
def _doi_and_pdf_from_payload(payload: dict) -> tuple[str | None, str | None]:
doi = normalize_doi(str(payload.get("doi") or ""))
return doi, _payload_pdf_url(payload)
def _resolution_targets(items: list[PublicationListItem]) -> list[PublicationListItem]:
return [item for item in items if not item.pdf_url]
def _crossref_budget_value() -> int:
return max(int(settings.crossref_max_lookups_per_request), 0)
async def resolve_publication_oa_metadata(
items: list[PublicationListItem],
*,
request_email: str | None = None,
) -> dict[int, tuple[str | None, str | None]]:
if not settings.unpaywall_enabled:
return {}
email = _email_for_request(request_email)
if email is None:
logger.debug("unpaywall.resolve_skipped_missing_email")
return {}
import httpx
timeout_seconds = max(float(settings.unpaywall_timeout_seconds), 0.5)
resolved: dict[int, tuple[str | None, str | None]] = {}
crossref_budget = _crossref_budget_value()
crossref_lookups = 0
targets = _resolution_targets(items)[: max(int(settings.unpaywall_max_items_per_request), 0)]
async with httpx.AsyncClient(timeout=timeout_seconds) as client:
for item in targets:
allow_crossref = crossref_budget > 0 and crossref_lookups < crossref_budget
payload, used_crossref = await _resolve_item_payload(
client=client,
item=item,
email=email,
allow_crossref=allow_crossref,
)
if used_crossref:
crossref_lookups += 1
if not isinstance(payload, dict):
continue
resolved[item.publication_id] = _doi_and_pdf_from_payload(payload)
resolved_count = sum(1 for _doi, pdf in resolved.values() if pdf)
doi_input_count = len([item for item in items if _publication_doi(item)])
target_doi_count = len([item for item in targets if _publication_doi(item)])
_log_resolution_summary(
publication_count=len(items),
doi_input_count=doi_input_count,
search_attempt_count=max(0, len(targets) - target_doi_count),
resolved_pdf_count=resolved_count,
email=email,
)
return resolved
async def resolve_publication_pdf_urls(
items: list[PublicationListItem],
*,
request_email: str | None = None,
) -> dict[int, str | None]:
resolved = await resolve_publication_oa_metadata(items, request_email=request_email)
return {publication_id: pdf for publication_id, (_doi, pdf) in resolved.items()}

3
app/services/import_export.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.portability.application import *

3
app/services/ingestion.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.ingestion.application import *

3
app/services/publications.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.publications.application import *

3
app/services/run_safety.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.ingestion.safety import *

3
app/services/runs.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.runs.application import *

3
app/services/scheduler.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.ingestion.scheduler import *

3
app/services/scholar_parser.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.scholar.parser import *

4
app/services/scholar_source.py
View file

@ -1,4 +0,0 @@
from __future__ import annotations
from app.services.domains.scholar.source import *
from app.services.domains.scholar.source import _build_profile_url

3
app/services/scholars.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.scholars.application import *

3
app/services/user_settings.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.settings.application import *

3
app/services/users.py
View file

@ -1,3 +0,0 @@
from __future__ import annotations
from app.services.domains.users.application import *

10
app/settings.py
View file

@ -229,6 +229,16 @@ class Settings:
"SCHOLAR_NAME_SEARCH_ALERT_COOLDOWN_REJECTIONS_THRESHOLD",
3,
)
unpaywall_enabled: bool = _env_bool("UNPAYWALL_ENABLED", True)
unpaywall_email: str = _env_str("UNPAYWALL_EMAIL", "")
unpaywall_timeout_seconds: float = _env_float("UNPAYWALL_TIMEOUT_SECONDS", 4.0)
unpaywall_max_items_per_request: int = _env_int("UNPAYWALL_MAX_ITEMS_PER_REQUEST", 20)
unpaywall_retry_cooldown_seconds: int = _env_int("UNPAYWALL_RETRY_COOLDOWN_SECONDS", 1800)
crossref_enabled: bool = _env_bool("CROSSREF_ENABLED", True)
crossref_max_rows: int = _env_int("CROSSREF_MAX_ROWS", 10)
crossref_timeout_seconds: float = _env_float("CROSSREF_TIMEOUT_SECONDS", 8.0)
crossref_min_interval_seconds: float = _env_float("CROSSREF_MIN_INTERVAL_SECONDS", 0.6)
crossref_max_lookups_per_request: int = _env_int("CROSSREF_MAX_LOOKUPS_PER_REQUEST", 8)
settings = Settings()