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

refactor: migrate all remaining logging to structured_log, remove boilerplate helpers

Browse source 
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Justin Visser 2026-02-26 21:59:34 +01:00
parent 19ca53186c
commit db0ac6a228
27 changed files with 361 additions and 913 deletions
Download patch file Download diff file Expand all files Collapse all files

47
app/services/domains/arxiv/client.py
View file

@ -5,6 +5,7 @@ import logging
import httpx
from app.logging_utils import structured_log
from app.services.domains.arxiv.cache import (
build_query_fingerprint,
get_cached_feed,
@ -102,17 +103,9 @@ class ArxivClient:
if self._cache_enabled:
cached = await get_cached_feed(query_fingerprint=query_fingerprint)
if cached is not None:
_log_cache_event(
event_name="arxiv.cache_hit",
query_fingerprint=query_fingerprint,
source_path=source_path,
)
structured_log(logger, "info", "arxiv.cache_hit", query_fingerprint=query_fingerprint, source_path=source_path)
return cached
_log_cache_event(
event_name="arxiv.cache_miss",
query_fingerprint=query_fingerprint,
source_path=source_path,
)
structured_log(logger, "info", "arxiv.cache_miss", query_fingerprint=query_fingerprint, source_path=source_path)
return await run_with_inflight_dedupe(
query_fingerprint=query_fingerprint,
fetch_feed=lambda: self._fetch_live_feed(
@ -222,9 +215,10 @@ async def _request_arxiv_feed(
source_path = _source_path_from_params(params)
cooldown_status = await get_arxiv_cooldown_status()
if cooldown_status.is_active:
_log_request_skipped_for_cooldown(
structured_log(
logger, "warning", "arxiv.request_skipped_cooldown",
source_path=source_path,
cooldown_remaining_seconds=cooldown_status.remaining_seconds,
cooldown_remaining_seconds=float(cooldown_status.remaining_seconds),
)
raise ArxivRateLimitError(
f"arXiv global cooldown active ({cooldown_status.remaining_seconds:.0f}s remaining)"
@ -280,32 +274,3 @@ def _source_path_from_params(params: dict[str, object]) -> str:
return ARXIV_SOURCE_PATH_UNKNOWN
def _log_cache_event(
*,
event_name: str,
query_fingerprint: str,
source_path: str,
) -> None:
logger.info(
event_name,
extra={
"event": event_name,
"query_fingerprint": query_fingerprint,
"source_path": source_path,
},
)
def _log_request_skipped_for_cooldown(
*,
source_path: str,
cooldown_remaining_seconds: float,
) -> None:
logger.warning(
"arxiv.request_skipped_cooldown",
extra={
"event": "arxiv.request_skipped_cooldown",
"source_path": source_path,
"cooldown_remaining_seconds": float(cooldown_remaining_seconds),
},
)

5
app/services/domains/arxiv/gateway.py
View file

@ -5,6 +5,7 @@ import re
from typing import TYPE_CHECKING, Protocol
import unicodedata
from app.logging_utils import structured_log
from app.services.domains.arxiv.client import ArxivClient
from app.services.domains.arxiv.errors import ArxivRateLimitError
from app.services.domains.arxiv.types import ArxivFeed
@ -112,7 +113,7 @@ class HttpArxivGateway:
except ArxivRateLimitError:
raise
except Exception as exc:
logger.debug("arxiv.query_failed", extra={"event": "arxiv.query_failed", "error": str(exc)})
structured_log(logger, "debug", "arxiv.query_failed", error=str(exc))
return None
@ -136,6 +137,6 @@ def _author_surname(scholar_label: str | None) -> str | None:
def _first_discovered_id(result: ArxivFeed) -> str | None:
for entry in result.entries:
if entry.arxiv_id:
logger.debug("arxiv.id_discovered", extra={"event": "arxiv.id_discovered", "arxiv_id": entry.arxiv_id})
structured_log(logger, "debug", "arxiv.id_discovered", arxiv_id=entry.arxiv_id)
return entry.arxiv_id
return None

77
app/services/domains/arxiv/rate_limit.py
View file

@ -12,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.db.models import ArxivRuntimeState
from app.db.session import get_session_factory
from app.logging_utils import structured_log
from app.services.domains.arxiv.constants import (
ARXIV_RATE_LIMIT_LOCK_KEY,
ARXIV_RATE_LIMIT_LOCK_NAMESPACE,
@ -83,14 +84,12 @@ async def _run_serialized_fetch(
response_status=int(response.status_code),
source_path=source_path,
)
_log_request_completed(
response_status=int(response.status_code),
structured_log(
logger, "info", "arxiv.request_completed",
status_code=int(response.status_code),
wait_seconds=wait_seconds,
cooldown_remaining_seconds=_cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=datetime.now(timezone.utc)),
source_path=source_path,
cooldown_remaining_seconds=_cooldown_remaining_seconds(
runtime_state.cooldown_until,
now_utc=datetime.now(timezone.utc),
),
)
return response, hit_rate_limit
@ -128,18 +127,10 @@ async def _wait_for_allowed_slot_or_raise(
now_utc = datetime.now(timezone.utc)
cooldown_seconds = _cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=now_utc)
if cooldown_seconds > 0:
_log_request_scheduled(
wait_seconds=0.0,
source_path=source_path,
cooldown_remaining_seconds=cooldown_seconds,
)
structured_log(logger, "info", "arxiv.request_scheduled", wait_seconds=0.0, source_path=source_path, cooldown_remaining_seconds=cooldown_seconds)
raise ArxivRateLimitError(f"arXiv global cooldown active ({cooldown_seconds:.0f}s remaining)")
wait_seconds = _next_allowed_wait_seconds(runtime_state.next_allowed_at, now_utc=now_utc)
_log_request_scheduled(
wait_seconds=wait_seconds,
source_path=source_path,
cooldown_remaining_seconds=0.0,
)
structured_log(logger, "info", "arxiv.request_scheduled", wait_seconds=wait_seconds, source_path=source_path, cooldown_remaining_seconds=0.0)
if wait_seconds > 0:
await asyncio.sleep(wait_seconds)
return wait_seconds
@ -156,9 +147,10 @@ def _record_post_response_state(
if response_status == 429:
cooldown_seconds = _cooldown_seconds()
runtime_state.cooldown_until = now_utc + timedelta(seconds=cooldown_seconds)
_log_cooldown_activated(
source_path=source_path,
structured_log(
logger, "warning", "arxiv.cooldown_activated",
cooldown_remaining_seconds=cooldown_seconds,
source_path=source_path,
)
return True
if _cooldown_remaining_seconds(runtime_state.cooldown_until, now_utc=now_utc) <= 0:
@ -196,52 +188,3 @@ def _cooldown_seconds() -> float:
return max(float(settings.arxiv_rate_limit_cooldown_seconds), 0.0)
def _log_request_scheduled(
*,
wait_seconds: float,
source_path: str,
cooldown_remaining_seconds: float,
) -> None:
logger.info(
"arxiv.request_scheduled",
extra={
"event": "arxiv.request_scheduled",
"wait_seconds": float(wait_seconds),
"cooldown_remaining_seconds": float(cooldown_remaining_seconds),
"source_path": source_path,
},
)
def _log_request_completed(
*,
response_status: int,
wait_seconds: float,
source_path: str,
cooldown_remaining_seconds: float,
) -> None:
logger.info(
"arxiv.request_completed",
extra={
"event": "arxiv.request_completed",
"status_code": int(response_status),
"wait_seconds": float(wait_seconds),
"cooldown_remaining_seconds": float(cooldown_remaining_seconds),
"source_path": source_path,
},
)
def _log_cooldown_activated(
*,
source_path: str,
cooldown_remaining_seconds: float,
) -> None:
logger.warning(
"arxiv.cooldown_activated",
extra={
"event": "arxiv.cooldown_activated",
"cooldown_remaining_seconds": float(cooldown_remaining_seconds),
"source_path": source_path,
},
)

3
app/services/domains/crossref/application.py
View file

@ -10,6 +10,7 @@ from typing import TYPE_CHECKING
from crossref.restful import Etiquette, Works
from app.services.domains.doi.normalize import normalize_doi
from app.logging_utils import structured_log
from app.settings import settings
if TYPE_CHECKING:
@ -377,6 +378,6 @@ async def discover_doi_for_publication(
author_surname=author_surname,
)
if doi:
logger.debug("crossref.doi_discovered", extra={"event": "crossref.doi_discovered"})
structured_log(logger, "debug", "crossref.doi_discovered")
return doi
return None

227
app/services/domains/ingestion/scheduler.py
View file

@ -23,6 +23,7 @@ from app.services.domains.ingestion.application import (
RunBlockedBySafetyPolicyError,
ScholarIngestionService,
)
from app.logging_utils import structured_log
from app.services.domains.settings import application as user_settings_service
from app.services.domains.scholar.source import LiveScholarSource
from app.settings import settings
@ -89,31 +90,23 @@ class SchedulerService:
async def start(self) -> None:
if not self._enabled:
logger.info(
"scheduler.disabled",
extra={
"event": "scheduler.disabled",
},
)
structured_log(logger, "info", "scheduler.disabled")
return
if self._task is not None:
return
self._task = asyncio.create_task(self._run_loop(), name="scholarr-scheduler")
logger.info(
"scheduler.started",
extra={
"event": "scheduler.started",
"tick_seconds": self._tick_seconds,
"network_error_retries": self._network_error_retries,
"retry_backoff_seconds": self._retry_backoff_seconds,
"max_pages_per_scholar": self._max_pages_per_scholar,
"page_size": self._page_size,
"continuation_queue_enabled": self._continuation_queue_enabled,
"continuation_base_delay_seconds": self._continuation_base_delay_seconds,
"continuation_max_delay_seconds": self._continuation_max_delay_seconds,
"continuation_max_attempts": self._continuation_max_attempts,
"queue_batch_size": self._queue_batch_size,
},
structured_log(
logger, "info", "scheduler.started",
tick_seconds=self._tick_seconds,
network_error_retries=self._network_error_retries,
retry_backoff_seconds=self._retry_backoff_seconds,
max_pages_per_scholar=self._max_pages_per_scholar,
page_size=self._page_size,
continuation_queue_enabled=self._continuation_queue_enabled,
continuation_base_delay_seconds=self._continuation_base_delay_seconds,
continuation_max_delay_seconds=self._continuation_max_delay_seconds,
continuation_max_attempts=self._continuation_max_attempts,
queue_batch_size=self._queue_batch_size,
)
async def stop(self) -> None:
@ -126,7 +119,7 @@ class SchedulerService:
pass
finally:
self._task = None
logger.info("scheduler.stopped", extra={"event": "scheduler.stopped"})
structured_log(logger, "info", "scheduler.stopped")
async def _run_loop(self) -> None:
while True:
@ -137,9 +130,7 @@ class SchedulerService:
except Exception:
logger.exception(
"scheduler.tick_failed",
extra={
"event": "scheduler.tick_failed",
},
extra={},
)
await asyncio.sleep(float(self._tick_seconds))
@ -181,17 +172,12 @@ class SchedulerService:
if cooldown_until is not None and cooldown_until.tzinfo is None:
cooldown_until = cooldown_until.replace(tzinfo=timezone.utc)
if cooldown_until is not None and cooldown_until > now_utc:
logger.info(
"scheduler.run_skipped_safety_cooldown_precheck",
extra={
"event": "scheduler.run_skipped_safety_cooldown_precheck",
"user_id": int(user_id),
"reason": cooldown_reason,
"cooldown_until": cooldown_until,
"cooldown_remaining_seconds": int((cooldown_until - now_utc).total_seconds()),
"metric_name": "scheduler_run_skipped_safety_cooldown_total",
"metric_value": 1,
},
structured_log(
logger, "info", "scheduler.run_skipped_safety_cooldown_precheck",
user_id=int(user_id),
reason=cooldown_reason,
cooldown_until=cooldown_until,
cooldown_remaining_seconds=int((cooldown_until - now_utc).total_seconds()),
)
return None
return _AutoRunCandidate(
@ -261,42 +247,34 @@ class SchedulerService:
)
except RunAlreadyInProgressError:
await session.rollback()
logger.info("scheduler.run_skipped_locked", extra={"event": "scheduler.run_skipped_locked", "user_id": candidate.user_id})
structured_log(logger, "info", "scheduler.run_skipped_locked", user_id=candidate.user_id)
return None
except RunBlockedBySafetyPolicyError as exc:
await session.rollback()
logger.info(
"scheduler.run_skipped_safety_cooldown",
extra={
"event": "scheduler.run_skipped_safety_cooldown",
"user_id": candidate.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": "scheduler_run_skipped_safety_cooldown_total",
"metric_value": 1,
},
structured_log(
logger, "info", "scheduler.run_skipped_safety_cooldown",
user_id=candidate.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"),
)
return None
except Exception:
await session.rollback()
logger.exception("scheduler.run_failed", extra={"event": "scheduler.run_failed", "user_id": candidate.user_id})
logger.exception("scheduler.run_failed", extra={"user_id": candidate.user_id})
return None
async def _run_candidate(self, candidate: _AutoRunCandidate) -> None:
run_summary = await self._run_candidate_ingestion(candidate=candidate)
if run_summary is None:
return
logger.info(
"scheduler.run_completed",
extra={
"event": "scheduler.run_completed",
"user_id": candidate.user_id,
"run_id": run_summary.crawl_run_id,
"status": run_summary.status.value,
"scholar_count": run_summary.scholar_count,
"new_publication_count": run_summary.new_publication_count,
},
structured_log(
logger, "info", "scheduler.run_completed",
user_id=candidate.user_id,
run_id=run_summary.crawl_run_id,
status=run_summary.status.value,
scholar_count=run_summary.scholar_count,
new_publication_count=run_summary.new_publication_count,
)
async def _drain_continuation_queue(self) -> None:
@ -326,16 +304,13 @@ class SchedulerService:
)
await session.commit()
if dropped is not None:
logger.warning(
"scheduler.queue_item_dropped_max_attempts",
extra={
"event": "scheduler.queue_item_dropped_max_attempts",
"queue_item_id": job.id,
"user_id": job.user_id,
"scholar_profile_id": job.scholar_profile_id,
"attempt_count": job.attempt_count,
"max_attempts": self._continuation_max_attempts,
},
structured_log(
logger, "warning", "scheduler.queue_item_dropped_max_attempts",
queue_item_id=job.id,
user_id=job.user_id,
scholar_profile_id=job.scholar_profile_id,
attempt_count=job.attempt_count,
max_attempts=self._continuation_max_attempts,
)
return True
@ -376,14 +351,11 @@ class SchedulerService:
)
await session.commit()
if dropped is not None:
logger.info(
"scheduler.queue_item_dropped_scholar_unavailable",
extra={
"event": "scheduler.queue_item_dropped_scholar_unavailable",
"queue_item_id": job.id,
"user_id": job.user_id,
"scholar_profile_id": job.scholar_profile_id,
},
structured_log(
logger, "info", "scheduler.queue_item_dropped_scholar_unavailable",
queue_item_id=job.id,
user_id=job.user_id,
scholar_profile_id=job.scholar_profile_id,
)
return False
@ -398,9 +370,9 @@ class SchedulerService:
error="run_already_in_progress",
)
await recovery_session.commit()
logger.info(
"scheduler.queue_item_deferred_lock",
extra={"event": "scheduler.queue_item_deferred_lock", "queue_item_id": job.id, "user_id": job.user_id},
structured_log(
logger, "info", "scheduler.queue_item_deferred_lock",
queue_item_id=job.id, user_id=job.user_id,
)
async def _reschedule_queue_job_safety_cooldown(
@ -422,17 +394,12 @@ class SchedulerService:
error=str(exc.message),
)
await recovery_session.commit()
logger.info(
"scheduler.queue_item_deferred_safety_cooldown",
extra={
"event": "scheduler.queue_item_deferred_safety_cooldown",
"queue_item_id": job.id,
"user_id": job.user_id,
"reason": exc.safety_state.get("cooldown_reason"),
"cooldown_remaining_seconds": cooldown_remaining_seconds,
"metric_name": "scheduler_queue_item_deferred_safety_cooldown_total",
"metric_value": 1,
},
structured_log(
logger, "info", "scheduler.queue_item_deferred_safety_cooldown",
queue_item_id=job.id,
user_id=job.user_id,
reason=exc.safety_state.get("cooldown_reason"),
cooldown_remaining_seconds=cooldown_remaining_seconds,
)
async def _reschedule_queue_job_after_exception(
@ -455,9 +422,9 @@ class SchedulerService:
error=str(exc),
)
await recovery_session.commit()
logger.warning(
"scheduler.queue_item_dropped_after_exception",
extra={"event": "scheduler.queue_item_dropped_after_exception", "queue_item_id": job.id, "user_id": job.user_id, "attempt_count": queue_item.attempt_count},
structured_log(
logger, "warning", "scheduler.queue_item_dropped_after_exception",
queue_item_id=job.id, user_id=job.user_id, attempt_count=queue_item.attempt_count,
)
return
delay_seconds = queue_service.compute_backoff_seconds(
@ -473,7 +440,7 @@ class SchedulerService:
error=str(exc),
)
await recovery_session.commit()
logger.exception("scheduler.queue_item_run_failed", extra={"event": "scheduler.queue_item_run_failed", "queue_item_id": job.id, "user_id": job.user_id})
logger.exception("scheduler.queue_item_run_failed", extra={"queue_item_id": job.id, "user_id": job.user_id})
async def _run_ingestion_for_queue_job(
self,
@ -514,28 +481,6 @@ class SchedulerService:
await self._reschedule_queue_job_after_exception(job, exc=exc)
return None
@staticmethod
def _log_queue_item_resolved(
*,
event_name: str,
job: queue_service.ContinuationQueueJob,
run_summary,
attempt_count: int | None = None,
delay_seconds: int | None = None,
) -> None:
payload = {
"event": event_name,
"queue_item_id": job.id,
"user_id": job.user_id,
"run_id": run_summary.crawl_run_id,
"status": run_summary.status.value,
}
if attempt_count is not None:
payload["attempt_count"] = attempt_count
if delay_seconds is not None:
payload["delay_seconds"] = delay_seconds
logger.info(event_name, extra=payload)
async def _finalize_queue_job_after_run(self, job: queue_service.ContinuationQueueJob, run_summary) -> None:
session_factory = get_session_factory()
async with session_factory() as session:
@ -543,35 +488,45 @@ class SchedulerService:
queue_item = await queue_service.reset_attempt_count(session, job_id=job.id)
await session.commit()
if queue_item is None:
self._log_queue_item_resolved(event_name="scheduler.queue_item_resolved", job=job, run_summary=run_summary)
structured_log(
logger, "info", "scheduler.queue_item_resolved",
queue_item_id=job.id, user_id=job.user_id,
run_id=run_summary.crawl_run_id, status=run_summary.status.value,
)
return
self._log_queue_item_resolved(
event_name="scheduler.queue_item_progressed",
job=job,
run_summary=run_summary,
structured_log(
logger, "info", "scheduler.queue_item_progressed",
queue_item_id=job.id, user_id=job.user_id,
run_id=run_summary.crawl_run_id, status=run_summary.status.value,
attempt_count=int(queue_item.attempt_count),
)
return
queue_item = await queue_service.increment_attempt_count(session, job_id=job.id)
if queue_item is None:
await session.commit()
self._log_queue_item_resolved(event_name="scheduler.queue_item_resolved", job=job, run_summary=run_summary)
structured_log(
logger, "info", "scheduler.queue_item_resolved",
queue_item_id=job.id, user_id=job.user_id,
run_id=run_summary.crawl_run_id, status=run_summary.status.value,
)
return
if int(queue_item.attempt_count) >= self._continuation_max_attempts:
await queue_service.mark_dropped(session, job_id=job.id, reason="max_attempts_after_run")
await session.commit()
logger.warning(
"scheduler.queue_item_dropped_max_attempts_after_run",
extra={"event": "scheduler.queue_item_dropped_max_attempts_after_run", "queue_item_id": job.id, "user_id": job.user_id, "attempt_count": queue_item.attempt_count, "run_id": run_summary.crawl_run_id, "status": run_summary.status.value},
structured_log(
logger, "warning", "scheduler.queue_item_dropped_max_attempts_after_run",
queue_item_id=job.id, user_id=job.user_id,
attempt_count=queue_item.attempt_count,
run_id=run_summary.crawl_run_id, status=run_summary.status.value,
)
return
delay_seconds = queue_service.compute_backoff_seconds(base_seconds=self._continuation_base_delay_seconds, attempt_count=int(queue_item.attempt_count), max_seconds=self._continuation_max_delay_seconds)
await queue_service.reschedule_job(session, job_id=job.id, delay_seconds=delay_seconds, reason=queue_item.reason, error=queue_item.last_error)
await session.commit()
self._log_queue_item_resolved(
event_name="scheduler.queue_item_rescheduled_failed",
job=job,
run_summary=run_summary,
structured_log(
logger, "info", "scheduler.queue_item_rescheduled_failed",
queue_item_id=job.id, user_id=job.user_id,
run_id=run_summary.crawl_run_id, status=run_summary.status.value,
attempt_count=int(queue_item.attempt_count),
delay_seconds=delay_seconds,
)
@ -600,14 +555,12 @@ class SchedulerService:
max_attempts=settings.pdf_auto_retry_max_attempts,
)
if processed > 0:
logger.info("scheduler.pdf_queue_drain_completed", extra={
"event": "scheduler.pdf_queue_drain_completed",
"processed_count": processed,
})
structured_log(
logger, "info", "scheduler.pdf_queue_drain_completed",
processed_count=processed,
)
except Exception:
logger.exception("scheduler.pdf_queue_drain_failed", extra={
"event": "scheduler.pdf_queue_drain_failed",
})
logger.exception("scheduler.pdf_queue_drain_failed", extra={})
async def _load_request_delay_for_user(
self,

4
app/services/domains/openalex/client.py
View file

@ -84,7 +84,7 @@ class OpenAlexClient:
)
raise OpenAlexRateLimitError("Rate limit exceeded fetching OpenAlex work by DOI")
if response.status_code >= 400:
logger.warning("OpenAlex API error: %s %s", response.status_code, response.text)
logger.warning("OpenAlex API error: %s %s", response.status_code, response.text[:500])
raise OpenAlexClientError(f"API Error {response.status_code}")
data = response.json()
@ -134,7 +134,7 @@ class OpenAlexClient:
)
raise OpenAlexRateLimitError("Rate limit exceeded fetching OpenAlex works list")
if response.status_code >= 400:
logger.warning("OpenAlex API error (filters=%s): %s %s", filters, response.status_code, response.text)
logger.warning("OpenAlex API error (filters=%s): %s %s", filters, response.status_code, response.text[:500])
raise OpenAlexClientError(f"API Error {response.status_code}")
data = response.json()

10
app/services/domains/publications/dedup.py
View file

@ -10,6 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import aliased
from app.db.models import Publication, PublicationIdentifier, ScholarPublication
from app.logging_utils import structured_log
from app.services.domains.ingestion.fingerprints import (
canonical_title_text_for_dedup,
canonical_title_tokens_for_dedup,
@ -104,14 +105,7 @@ async def merge_duplicate_publication(
await _migrate_scholar_links(db_session, winner_id=winner_id, dup_id=dup_id)
await _migrate_identifiers(db_session, winner_id=winner_id, dup_id=dup_id)
await db_session.execute(delete(Publication).where(Publication.id == dup_id))
logger.info(
"publications.identifier_merge",
extra={
"event": "publications.identifier_merge",
"winner_id": winner_id,
"dup_id": dup_id,
},
)
structured_log(logger, "info", "publications.identifier_merge", winner_id=winner_id, dup_id=dup_id)
async def _load_publication(

20
app/services/domains/publications/enrichment.py
View file

@ -9,6 +9,7 @@ from app.services.domains.publications.pdf_queue import (
enqueue_retry_pdf_job,
overlay_pdf_job_state,
)
from app.logging_utils import structured_log
from app.services.domains.publications.types import PublicationListItem
logger = logging.getLogger(__name__)
@ -29,14 +30,7 @@ async def schedule_missing_pdf_enrichment_for_user(
rows=items,
max_items=max_items,
)
logger.info(
"publications.enrichment.scheduled",
extra={
"event": "publications.enrichment.scheduled",
"user_id": user_id,
"publication_count": len(queued_ids),
},
)
structured_log(logger, "info", "publications.enrichment.scheduled", user_id=user_id, publication_count=len(queued_ids))
return len(queued_ids)
@ -53,15 +47,7 @@ async def schedule_retry_pdf_enrichment_for_row(
request_email=request_email,
row=item,
)
logger.info(
"publications.enrichment.retry_scheduled",
extra={
"event": "publications.enrichment.retry_scheduled",
"user_id": user_id,
"publication_id": item.publication_id,
"queued": queued,
},
)
structured_log(logger, "info", "publications.enrichment.retry_scheduled", user_id=user_id, publication_id=item.publication_id, queued=queued)
return queued

29
app/services/domains/publications/pdf_queue.py
View file

@ -27,6 +27,7 @@ from app.services.domains.unpaywall.application import (
FAILURE_RESOLUTION_EXCEPTION,
OaResolutionOutcome,
)
from app.logging_utils import structured_log
from app.settings import settings
PDF_STATUS_UNTRACKED = "untracked"
@ -320,10 +321,7 @@ def _drop_finished_task(task: asyncio.Task[None]) -> None:
try:
task.result()
except Exception:
logger.exception(
"publications.pdf_queue.task_failed",
extra={"event": "publications.pdf_queue.task_failed"},
)
logger.exception("publications.pdf_queue.task_failed")
async def _mark_attempt_started(
@ -474,14 +472,7 @@ async def _resolve_publication_row(
# Propagate upward so the batch loop can stop immediately.
raise
except Exception as exc: # pragma: no cover - defensive network boundary
logger.warning(
"publications.pdf_queue.resolve_failed",
extra={
"event": "publications.pdf_queue.resolve_failed",
"publication_id": row.publication_id,
"error": str(exc),
},
)
structured_log(logger, "warning", "publications.pdf_queue.resolve_failed", publication_id=row.publication_id, error=str(exc))
outcome = _failed_outcome(row=row)
arxiv_rate_limited = False
await _persist_outcome(
@ -523,19 +514,9 @@ async def _run_resolution_task(
)
if arxiv_rate_limited and arxiv_lookup_allowed:
arxiv_lookup_allowed = False
logger.warning(
"publications.pdf_queue.arxiv_batch_disabled",
extra={
"event": "publications.pdf_queue.arxiv_batch_disabled",
"detail": "arXiv temporarily disabled for remaining batch after rate limit",
},
)
structured_log(logger, "warning", "publications.pdf_queue.arxiv_batch_disabled", detail="arXiv temporarily disabled for remaining batch after rate limit")
except OpenAlexBudgetExhaustedError:
logger.warning(
"publications.pdf_queue.budget_exhausted",
extra={"event": "publications.pdf_queue.budget_exhausted",
"detail": "Stopping PDF resolution batch — OpenAlex daily budget exhausted"},
)
structured_log(logger, "warning", "publications.pdf_queue.budget_exhausted", detail="Stopping PDF resolution batch — OpenAlex daily budget exhausted")
break

33
app/services/domains/publications/pdf_resolution_pipeline.py
View file

@ -9,6 +9,7 @@ from app.services.domains.arxiv.guards import arxiv_skip_reason_for_item
from app.services.domains.openalex.client import OpenAlexBudgetExhaustedError
from app.services.domains.publications.types import PublicationListItem
from app.services.domains.unpaywall.application import OaResolutionOutcome, resolve_publication_oa_outcomes
from app.logging_utils import structured_log
from app.settings import settings
logger = logging.getLogger(__name__)
@ -44,13 +45,7 @@ async def resolve_publication_pdf_outcome_for_row(
except ArxivRateLimitError:
arxiv_rate_limited = True
arxiv_outcome = None
logger.warning(
"publications.pdf_resolution.arxiv_rate_limited",
extra={
"event": "publications.pdf_resolution.arxiv_rate_limited",
"publication_id": int(row.publication_id),
},
)
structured_log(logger, "warning", "publications.pdf_resolution.arxiv_rate_limited", publication_id=int(row.publication_id))
if arxiv_outcome and arxiv_outcome.pdf_url:
return PipelineOutcome(arxiv_outcome, None, arxiv_rate_limited=arxiv_rate_limited)
@ -99,10 +94,7 @@ async def _openalex_outcome(
# Re-raise so the caller's batch loop can stop hitting the API.
raise
except Exception as exc:
logger.warning(
"publications.pdf_resolution.openalex_failed",
extra={"event": "publications.pdf_resolution.openalex_failed", "error": str(exc)},
)
structured_log(logger, "warning", "publications.pdf_resolution.openalex_failed", error=str(exc))
return None
@ -115,12 +107,12 @@ async def _arxiv_outcome(
from app.services.domains.arxiv.application import discover_arxiv_id_for_publication
if not allow_lookup:
_log_arxiv_skip(row=row, skip_reason="batch_arxiv_cooldown_active")
structured_log(logger, "info", "publications.pdf_resolution.arxiv_skipped", publication_id=int(row.publication_id), skip_reason="batch_arxiv_cooldown_active")
return None
skip_reason = arxiv_skip_reason_for_item(item=row)
if skip_reason is not None:
_log_arxiv_skip(row=row, skip_reason=skip_reason)
structured_log(logger, "info", "publications.pdf_resolution.arxiv_skipped", publication_id=int(row.publication_id), skip_reason=skip_reason)
return None
try:
@ -138,23 +130,10 @@ async def _arxiv_outcome(
except ArxivRateLimitError:
raise # propagate so orchestration can switch to non-arXiv fallback
except Exception as exc:
logger.warning(
"publications.pdf_resolution.arxiv_failed",
extra={"event": "publications.pdf_resolution.arxiv_failed", "error": str(exc)},
)
structured_log(logger, "warning", "publications.pdf_resolution.arxiv_failed", error=str(exc))
return None
def _log_arxiv_skip(*, row: PublicationListItem, skip_reason: str) -> None:
logger.info(
"publications.pdf_resolution.arxiv_skipped",
extra={
"event": "publications.pdf_resolution.arxiv_skipped",
"publication_id": int(row.publication_id),
"skip_reason": skip_reason,
},
)
async def _oa_outcome(
*,

66
app/services/domains/scholar/source.py
View file

@ -9,6 +9,7 @@ from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from app.logging_utils import structured_log
from app.services.domains.scholar import rate_limit as scholar_rate_limit
from app.settings import settings
@ -100,15 +101,12 @@ class LiveScholarSource:
cstart=cstart,
pagesize=pagesize,
)
logger.debug(
"scholar_source.fetch_started",
extra={
"event": "scholar_source.fetch_started",
"scholar_id": scholar_id,
"requested_url": requested_url,
"cstart": cstart,
"pagesize": pagesize,
},
structured_log(
logger, "debug", "scholar_source.fetch_started",
scholar_id=scholar_id,
requested_url=requested_url,
cstart=cstart,
pagesize=pagesize,
)
return await self._fetch_with_global_throttle(requested_url)
@ -122,24 +120,18 @@ class LiveScholarSource:
query=query,
start=start,
)
logger.debug(
"scholar_source.search_fetch_started",
extra={
"event": "scholar_source.search_fetch_started",
"query": query,
"requested_url": requested_url,
"start": start,
},
structured_log(
logger, "debug", "scholar_source.search_fetch_started",
query=query,
requested_url=requested_url,
start=start,
)
return await self._fetch_with_global_throttle(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,
},
structured_log(
logger, "debug", "scholar_source.publication_fetch_started",
requested_url=publication_url,
)
return await self._fetch_with_global_throttle(publication_url)
@ -169,9 +161,9 @@ class LiveScholarSource:
@staticmethod
def _network_error_result(requested_url: str, exc: URLError) -> FetchResult:
logger.warning(
"scholar_source.fetch_network_error",
extra={"event": "scholar_source.fetch_network_error", "requested_url": requested_url},
structured_log(
logger, "warning", "scholar_source.fetch_network_error",
requested_url=requested_url,
)
return FetchResult(
requested_url=requested_url,
@ -183,13 +175,10 @@ class LiveScholarSource:
@staticmethod
def _http_error_result(requested_url: str, exc: HTTPError) -> FetchResult:
logger.warning(
"scholar_source.fetch_http_error",
extra={
"event": "scholar_source.fetch_http_error",
"requested_url": requested_url,
"status_code": exc.code,
},
structured_log(
logger, "warning", "scholar_source.fetch_http_error",
requested_url=requested_url,
status_code=exc.code,
)
return FetchResult(
requested_url=requested_url,
@ -203,13 +192,10 @@ class LiveScholarSource:
def _success_result(requested_url: str, response) -> FetchResult:
body = response.read().decode("utf-8", errors="replace")
status_code = getattr(response, "status", 200)
logger.debug(
"scholar_source.fetch_succeeded",
extra={
"event": "scholar_source.fetch_succeeded",
"requested_url": requested_url,
"status_code": status_code,
},
structured_log(
logger, "debug", "scholar_source.fetch_succeeded",
requested_url=requested_url,
status_code=status_code,
)
return FetchResult(
requested_url=requested_url,

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

@ -41,6 +41,7 @@ from app.services.domains.scholars.constants import (
SEARCH_COOLDOWN_REASON,
SEARCH_DISABLED_REASON,
)
from app.logging_utils import structured_log
from app.services.domains.scholars.exceptions import ScholarServiceError
from app.services.domains.scholars.search_hints import (
_merge_warnings,
@ -430,9 +431,9 @@ def _normalize_author_search_inputs(query: str, limit: int) -> tuple[str, int, s
def _disabled_search_result(*, normalized_query: str, bounded_limit: int) -> ParsedAuthorSearchPage:
logger.warning(
"scholar_search.disabled_by_configuration",
extra={"event": "scholar_search.disabled_by_configuration", "query": normalized_query},
structured_log(
logger, "warning", "scholar_search.disabled_by_configuration",
query=normalized_query,
)
return _policy_blocked_author_search_result(
reason=SEARCH_DISABLED_REASON,
@ -456,9 +457,9 @@ def _normalize_runtime_cooldown_state(
updated = True
if now_utc < cooldown_until:
return updated
logger.info(
"scholar_search.cooldown_expired",
extra={"event": "scholar_search.cooldown_expired", "cooldown_until_utc": cooldown_until.isoformat()},
structured_log(
logger, "info", "scholar_search.cooldown_expired",
cooldown_until_utc=cooldown_until.isoformat(),
)
runtime_state.cooldown_until = None
runtime_state.cooldown_rejection_count = 0
@ -492,15 +493,12 @@ def _emit_cooldown_threshold_alert(
return True
if bool(runtime_state.cooldown_alert_emitted):
return True
logger.error(
"scholar_search.cooldown_rejection_threshold_exceeded",
extra={
"event": "scholar_search.cooldown_rejection_threshold_exceeded",
"query": normalized_query,
"cooldown_rejection_count": int(runtime_state.cooldown_rejection_count),
"threshold": threshold,
"cooldown_until_utc": runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None,
},
structured_log(
logger, "error", "scholar_search.cooldown_rejection_threshold_exceeded",
query=normalized_query,
cooldown_rejection_count=int(runtime_state.cooldown_rejection_count),
threshold=threshold,
cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None,
)
runtime_state.cooldown_alert_emitted = True
return True
@ -519,14 +517,11 @@ def _cooldown_block_result(
normalized_query=normalized_query,
cooldown_rejection_alert_threshold=cooldown_rejection_alert_threshold,
)
logger.warning(
"scholar_search.cooldown_active",
extra={
"event": "scholar_search.cooldown_active",
"query": normalized_query,
"cooldown_remaining_seconds": cooldown_remaining_seconds,
"cooldown_until_utc": runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None,
},
structured_log(
logger, "warning", "scholar_search.cooldown_active",
query=normalized_query,
cooldown_remaining_seconds=cooldown_remaining_seconds,
cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None,
)
return _policy_blocked_author_search_result(
reason=SEARCH_COOLDOWN_REASON,
@ -553,14 +548,11 @@ async def _cache_hit_result(
)
if cached is None:
return None
logger.info(
"scholar_search.cache_hit",
extra={
"event": "scholar_search.cache_hit",
"query": normalized_query,
"state": cached.state.value,
"state_reason": cached.state_reason,
},
structured_log(
logger, "info", "scholar_search.cache_hit",
query=normalized_query,
state=cached.state.value,
state_reason=cached.state_reason,
)
state_reason_override = SEARCH_CACHED_BLOCK_REASON if _is_author_search_block_state(cached) else None
return _trim_author_search_result(
@ -610,9 +602,10 @@ async def _wait_for_author_search_throttle(
)
if sleep_seconds <= 0.0:
return updated
logger.info(
"scholar_search.throttle_wait",
extra={"event": "scholar_search.throttle_wait", "query": normalized_query, "sleep_seconds": round(sleep_seconds, 3)},
structured_log(
logger, "info", "scholar_search.throttle_wait",
query=normalized_query,
sleep_seconds=round(sleep_seconds, 3),
)
await asyncio.sleep(sleep_seconds)
return True
@ -665,16 +658,13 @@ def _with_retry_warnings(
threshold = max(1, int(retry_alert_threshold))
if retry_scheduled_count < threshold:
return merged
logger.warning(
"scholar_search.retry_threshold_exceeded",
extra={
"event": "scholar_search.retry_threshold_exceeded",
"query": normalized_query,
"retry_scheduled_count": retry_scheduled_count,
"threshold": threshold,
"final_state": merged.state.value,
"final_state_reason": merged.state_reason,
},
structured_log(
logger, "warning", "scholar_search.retry_threshold_exceeded",
query=normalized_query,
retry_scheduled_count=retry_scheduled_count,
threshold=threshold,
final_state=merged.state.value,
final_state_reason=merged.state_reason,
)
return replace(
merged,
@ -697,14 +687,11 @@ def _apply_block_circuit_breaker(
runtime_state.consecutive_blocked_count = 0
return merged_parsed
runtime_state.consecutive_blocked_count = int(runtime_state.consecutive_blocked_count) + 1
logger.warning(
"scholar_search.block_detected",
extra={
"event": "scholar_search.block_detected",
"query": normalized_query,
"state_reason": merged_parsed.state_reason,
"consecutive_blocked_count": int(runtime_state.consecutive_blocked_count),
},
structured_log(
logger, "warning", "scholar_search.block_detected",
query=normalized_query,
state_reason=merged_parsed.state_reason,
consecutive_blocked_count=int(runtime_state.consecutive_blocked_count),
)
if int(runtime_state.consecutive_blocked_count) < max(1, int(cooldown_block_threshold)):
return merged_parsed
@ -712,13 +699,10 @@ def _apply_block_circuit_breaker(
runtime_state.consecutive_blocked_count = 0
runtime_state.cooldown_rejection_count = 0
runtime_state.cooldown_alert_emitted = False
logger.error(
"scholar_search.cooldown_activated",
extra={
"event": "scholar_search.cooldown_activated",
"query": normalized_query,
"cooldown_until_utc": runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None,
},
structured_log(
logger, "error", "scholar_search.cooldown_activated",
query=normalized_query,
cooldown_until_utc=runtime_state.cooldown_until.isoformat() if runtime_state.cooldown_until else None,
)
return replace(
merged_parsed,

41
app/services/domains/unpaywall/application.py
View file

@ -13,6 +13,7 @@ from app.services.domains.unpaywall.pdf_discovery import (
resolve_pdf_from_landing_page,
)
from app.services.domains.unpaywall.rate_limit import wait_for_unpaywall_slot
from app.logging_utils import structured_log
from app.settings import settings
if TYPE_CHECKING:
@ -157,10 +158,7 @@ async def _resolved_pdf_url_from_payload(
for page_url in _crawl_targets(payload=payload, pdf_candidates=pdf_candidates)[:3]:
discovered = await resolve_pdf_from_landing_page(client, page_url=page_url)
if discovered:
logger.info(
"unpaywall.pdf_discovered_from_landing",
extra={"event": "unpaywall.pdf_discovered_from_landing", "landing_url": page_url},
)
structured_log(logger, "info", "unpaywall.pdf_discovered_from_landing", landing_url=page_url)
return discovered
return None
@ -191,27 +189,6 @@ def _email_for_request(request_email: str | None) -> str | None:
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,
@ -370,14 +347,7 @@ async def _safe_outcome_for_item(
allow_crossref=allow_crossref,
)
except Exception as exc: # pragma: no cover - defensive network boundary
logger.warning(
"unpaywall.resolve_item_failed",
extra={
"event": "unpaywall.resolve_item_failed",
"publication_id": item.publication_id,
"error": str(exc),
},
)
structured_log(logger, "warning", "unpaywall.resolve_item_failed", publication_id=item.publication_id, error=str(exc))
return _outcome_with_failure(
item=item,
failure_reason=FAILURE_RESOLUTION_EXCEPTION,
@ -439,12 +409,13 @@ async def resolve_publication_oa_outcomes(
targets=targets,
email=email,
)
_log_resolution_summary(
structured_log(
logger, "info", "unpaywall.resolve_completed",
publication_count=len(items),
doi_input_count=_doi_input_count(items),
search_attempt_count=_search_attempt_count(targets=targets),
resolved_pdf_count=_resolved_pdf_count(outcomes),
email=email,
email_domain=email.split("@", 1)[-1] if "@" in email else None,
)
return outcomes