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

Feat/decomposition #19

Merged
JustinZeus merged 38 commits from feat/decomposition into main 2026-03-01 22:55:33 +01:00
Conversation 1 Commits 38 Files changed 327 +195 -385
1 changed files with 195 additions and 385 deletions
Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 19ca53186c - Show all commits

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

@ -66,6 +66,7 @@ from app.services.domains.scholar.parser import (
parse_profile_page, parse_profile_page,
) )
from app.services.domains.scholar.source import FetchResult, ScholarSource from app.services.domains.scholar.source import FetchResult, ScholarSource
from app.logging_utils import structured_log
from app.settings import settings from app.settings import settings
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -119,28 +120,6 @@ class ScholarIngestionService:
) )
return max(policy_minimum, _int_or_default(value, policy_minimum)) return max(policy_minimum, _int_or_default(value, policy_minimum))
@staticmethod
def _log_request_delay_coercion(
*,
user_id: int,
requested_request_delay_seconds: int,
effective_request_delay_seconds: int,
) -> None:
logger.warning(
"ingestion.request_delay_coerced_to_policy_floor",
extra={
"event": "ingestion.request_delay_coerced_to_policy_floor",
"user_id": user_id,
"requested_request_delay_seconds": requested_request_delay_seconds,
"effective_request_delay_seconds": effective_request_delay_seconds,
"policy_minimum_request_delay_seconds": user_settings_service.resolve_request_delay_minimum(
settings.ingestion_min_request_delay_seconds
),
"metric_name": "ingestion_request_delay_coerced_total",
"metric_value": 1,
},
)
async def _load_user_settings_for_run( async def _load_user_settings_for_run(
self, self,
db_session: AsyncSession, db_session: AsyncSession,
@ -173,16 +152,11 @@ class ScholarIngestionService:
if run_safety_service.clear_expired_cooldown(user_settings, now_utc=now_utc): if run_safety_service.clear_expired_cooldown(user_settings, now_utc=now_utc):
await db_session.commit() await db_session.commit()
await db_session.refresh(user_settings) await db_session.refresh(user_settings)
logger.info( structured_log(
"ingestion.safety_cooldown_cleared", logger, "info", "ingestion.safety_cooldown_cleared",
extra={ user_id=user_id,
"event": "ingestion.safety_cooldown_cleared", reason=previous.get("cooldown_reason"),
"user_id": user_id, cooldown_until=previous.get("cooldown_until"),
"reason": previous.get("cooldown_reason"),
"cooldown_until": previous.get("cooldown_until"),
"metric_name": "ingestion_safety_cooldown_cleared_total",
"metric_value": 1,
},
) )
now_utc = datetime.now(timezone.utc) now_utc = datetime.now(timezone.utc)
if run_safety_service.is_cooldown_active(user_settings, now_utc=now_utc): if run_safety_service.is_cooldown_active(user_settings, now_utc=now_utc):
@ -208,19 +182,14 @@ class ScholarIngestionService:
now_utc=now_utc, now_utc=now_utc,
) )
await db_session.commit() await db_session.commit()
logger.warning( structured_log(
"ingestion.safety_policy_blocked_run_start", logger, "warning", "ingestion.safety_policy_blocked_run_start",
extra={ user_id=user_id,
"event": "ingestion.safety_policy_blocked_run_start", trigger_type=trigger_type.value,
"user_id": user_id, reason=safety_state.get("cooldown_reason"),
"trigger_type": trigger_type.value, cooldown_until=safety_state.get("cooldown_until"),
"reason": safety_state.get("cooldown_reason"), cooldown_remaining_seconds=safety_state.get("cooldown_remaining_seconds"),
"cooldown_until": safety_state.get("cooldown_until"), blocked_start_count=((safety_state.get("counters") or {}).get("blocked_start_count")),
"cooldown_remaining_seconds": safety_state.get("cooldown_remaining_seconds"),
"blocked_start_count": ((safety_state.get("counters") or {}).get("blocked_start_count")),
"metric_name": "ingestion_safety_run_start_blocked_total",
"metric_value": 1,
},
) )
raise RunBlockedBySafetyPolicyError( raise RunBlockedBySafetyPolicyError(
code="scrape_cooldown_active", code="scrape_cooldown_active",
@ -306,43 +275,6 @@ class ScholarIngestionService:
error_log={}, error_log={},
) )
@staticmethod
def _log_run_started(
*,
user_id: int,
trigger_type: RunTriggerType,
scholar_count: int,
filtered: bool,
request_delay_seconds: int,
network_error_retries: int,
retry_backoff_seconds: float,
max_pages_per_scholar: int,
page_size: int,
idempotency_key: str | None,
alert_blocked_failure_threshold: int,
alert_network_failure_threshold: int,
alert_retry_scheduled_threshold: int,
) -> None:
logger.info(
"ingestion.run_started",
extra={
"event": "ingestion.run_started",
"user_id": user_id,
"trigger_type": trigger_type.value,
"scholar_count": scholar_count,
"is_filtered_run": filtered,
"request_delay_seconds": request_delay_seconds,
"network_error_retries": network_error_retries,
"retry_backoff_seconds": retry_backoff_seconds,
"max_pages_per_scholar": max_pages_per_scholar,
"page_size": page_size,
"idempotency_key": idempotency_key,
"alert_blocked_failure_threshold": alert_blocked_failure_threshold,
"alert_network_failure_threshold": alert_network_failure_threshold,
"alert_retry_scheduled_threshold": alert_retry_scheduled_threshold,
},
)
@staticmethod @staticmethod
async def _wait_between_scholars(*, index: int, request_delay_seconds: int) -> None: async def _wait_between_scholars(*, index: int, request_delay_seconds: int) -> None:
if index <= 0 or request_delay_seconds <= 0: if index <= 0 or request_delay_seconds <= 0:
@ -379,35 +311,6 @@ class ScholarIngestionService:
scholar.profile_image_url = first_page.profile_image_url scholar.profile_image_url = first_page.profile_image_url
scholar.last_initial_page_checked_at = run_dt scholar.last_initial_page_checked_at = run_dt
@staticmethod
def _log_scholar_parsed(
*,
user_id: int,
run_id: int,
scholar: ScholarProfile,
paged_parse_result: PagedParseResult,
) -> None:
parsed_page = paged_parse_result.parsed_page
logger.info(
"ingestion.scholar_parsed",
extra={
"event": "ingestion.scholar_parsed",
"user_id": user_id,
"crawl_run_id": run_id,
"scholar_profile_id": scholar.id,
"scholar_id": scholar.scholar_id,
"state": parsed_page.state.value,
"publication_count": len(paged_parse_result.publications),
"has_show_more_button": parsed_page.has_show_more_button,
"pages_fetched": paged_parse_result.pages_fetched,
"pages_attempted": paged_parse_result.pages_attempted,
"has_more_remaining": paged_parse_result.has_more_remaining,
"pagination_truncated_reason": paged_parse_result.pagination_truncated_reason,
"warning_count": len(parsed_page.warnings),
"skipped_no_change": paged_parse_result.skipped_no_change,
},
)
@staticmethod @staticmethod
def _build_result_entry( def _build_result_entry(
*, *,
@ -594,7 +497,6 @@ class ScholarIngestionService:
logger.exception( logger.exception(
"ingestion.scholar_failed", "ingestion.scholar_failed",
extra={ extra={
"event": "ingestion.scholar_failed",
"crawl_run_id": run.id, "crawl_run_id": run.id,
"scholar_profile_id": scholar.id, "scholar_profile_id": scholar.id,
"scholar_id": scholar.scholar_id, "scholar_id": scholar.scholar_id,
@ -618,16 +520,13 @@ class ScholarIngestionService:
attempt_log=paged_parse_result.attempt_log, attempt_log=paged_parse_result.attempt_log,
page_logs=paged_parse_result.page_logs, page_logs=paged_parse_result.page_logs,
) )
logger.warning( structured_log(
"ingestion.scholar_parse_failed", logger, "warning", "ingestion.scholar_parse_failed",
extra={ scholar_profile_id=scholar.id,
"event": "ingestion.scholar_parse_failed", scholar_id=scholar.scholar_id,
"scholar_profile_id": scholar.id, state=paged_parse_result.parsed_page.state.value,
"scholar_id": scholar.scholar_id, state_reason=paged_parse_result.parsed_page.state_reason,
"state": paged_parse_result.parsed_page.state.value, status_code=paged_parse_result.fetch_result.status_code,
"state_reason": paged_parse_result.parsed_page.state_reason,
"status_code": paged_parse_result.fetch_result.status_code,
},
) )
return ScholarProcessingOutcome(result_entry, 0, 1, 0, 0) return ScholarProcessingOutcome(result_entry, 0, 1, 0, 0)
@ -797,7 +696,23 @@ class ScholarIngestionService:
) )
self._assert_valid_paged_parse_result(scholar_id=scholar.scholar_id, paged_parse_result=paged_parse_result) self._assert_valid_paged_parse_result(scholar_id=scholar.scholar_id, paged_parse_result=paged_parse_result)
self._apply_first_page_profile_metadata(scholar=scholar, paged_parse_result=paged_parse_result, run_dt=run_dt) self._apply_first_page_profile_metadata(scholar=scholar, paged_parse_result=paged_parse_result, run_dt=run_dt)
self._log_scholar_parsed(user_id=user_id, run_id=run.id, scholar=scholar, paged_parse_result=paged_parse_result) parsed_page = paged_parse_result.parsed_page
structured_log(
logger, "info", "ingestion.scholar_parsed",
user_id=user_id,
crawl_run_id=run.id,
scholar_profile_id=scholar.id,
scholar_id=scholar.scholar_id,
state=parsed_page.state.value,
publication_count=len(paged_parse_result.publications),
has_show_more_button=parsed_page.has_show_more_button,
pages_fetched=paged_parse_result.pages_fetched,
pages_attempted=paged_parse_result.pages_attempted,
has_more_remaining=paged_parse_result.has_more_remaining,
pagination_truncated_reason=paged_parse_result.pagination_truncated_reason,
warning_count=len(parsed_page.warnings),
skipped_no_change=paged_parse_result.skipped_no_change,
)
result_entry = self._build_result_entry( result_entry = self._build_result_entry(
scholar=scholar, scholar=scholar,
start_cstart=start_cstart, start_cstart=start_cstart,
@ -918,7 +833,6 @@ class ScholarIngestionService:
logger.exception( logger.exception(
"ingestion.scholar_unexpected_failure", "ingestion.scholar_unexpected_failure",
extra={ extra={
"event": "ingestion.scholar_unexpected_failure",
"crawl_run_id": run.id, "crawl_run_id": run.id,
"scholar_profile_id": scholar.id, "scholar_profile_id": scholar.id,
"scholar_id": scholar.scholar_id, "scholar_id": scholar.scholar_id,
@ -1009,54 +923,6 @@ class ScholarIngestionService:
alert_flags=alert_flags, alert_flags=alert_flags,
) )
@staticmethod
def _log_alert_thresholds(
*,
user_id: int,
run_id: int,
failure_summary: RunFailureSummary,
alert_summary: RunAlertSummary,
) -> None:
if alert_summary.alert_flags["blocked_failure_threshold_exceeded"]:
logger.warning(
"ingestion.alert_blocked_failure_threshold_exceeded",
extra={
"event": "ingestion.alert_blocked_failure_threshold_exceeded",
"user_id": user_id,
"crawl_run_id": run_id,
"blocked_failure_count": alert_summary.blocked_failure_count,
"threshold": alert_summary.blocked_failure_threshold,
"metric_name": "ingestion_blocked_failure_threshold_exceeded_total",
"metric_value": 1,
},
)
if alert_summary.alert_flags["network_failure_threshold_exceeded"]:
logger.warning(
"ingestion.alert_network_failure_threshold_exceeded",
extra={
"event": "ingestion.alert_network_failure_threshold_exceeded",
"user_id": user_id,
"crawl_run_id": run_id,
"network_failure_count": alert_summary.network_failure_count,
"threshold": alert_summary.network_failure_threshold,
"metric_name": "ingestion_network_failure_threshold_exceeded_total",
"metric_value": 1,
},
)
if alert_summary.alert_flags["retry_scheduled_threshold_exceeded"]:
logger.warning(
"ingestion.alert_retry_scheduled_threshold_exceeded",
extra={
"event": "ingestion.alert_retry_scheduled_threshold_exceeded",
"user_id": user_id,
"crawl_run_id": run_id,
"retries_scheduled_count": failure_summary.retries_scheduled_count,
"threshold": alert_summary.retry_scheduled_threshold,
"metric_name": "ingestion_retry_scheduled_threshold_exceeded_total",
"metric_value": 1,
},
)
@staticmethod @staticmethod
def _apply_safety_outcome( def _apply_safety_outcome(
*, *,
@ -1080,56 +946,27 @@ class ScholarIngestionService:
network_cooldown_seconds=settings.ingestion_safety_cooldown_network_seconds, network_cooldown_seconds=settings.ingestion_safety_cooldown_network_seconds,
now_utc=datetime.now(timezone.utc), now_utc=datetime.now(timezone.utc),
) )
ScholarIngestionService._log_safety_transition(
user_id=user_id,
run_id=int(run.id),
alert_summary=alert_summary,
pre_apply_state=pre_apply_state,
safety_state=safety_state,
cooldown_trigger_reason=cooldown_trigger_reason,
)
@staticmethod
def _log_safety_transition(
*,
user_id: int,
run_id: int,
alert_summary: RunAlertSummary,
pre_apply_state: dict[str, Any],
safety_state: dict[str, Any],
cooldown_trigger_reason: str | None,
) -> None:
if cooldown_trigger_reason is not None: if cooldown_trigger_reason is not None:
logger.warning( structured_log(
"ingestion.safety_cooldown_entered", logger, "warning", "ingestion.safety_cooldown_entered",
extra={ user_id=user_id,
"event": "ingestion.safety_cooldown_entered", crawl_run_id=int(run.id),
"user_id": user_id, reason=cooldown_trigger_reason,
"crawl_run_id": run_id, blocked_failure_count=alert_summary.blocked_failure_count,
"reason": cooldown_trigger_reason, network_failure_count=alert_summary.network_failure_count,
"blocked_failure_count": alert_summary.blocked_failure_count, blocked_failure_threshold=alert_summary.blocked_failure_threshold,
"network_failure_count": alert_summary.network_failure_count, network_failure_threshold=alert_summary.network_failure_threshold,
"blocked_failure_threshold": alert_summary.blocked_failure_threshold, cooldown_until=safety_state.get("cooldown_until"),
"network_failure_threshold": alert_summary.network_failure_threshold, cooldown_remaining_seconds=safety_state.get("cooldown_remaining_seconds"),
"cooldown_until": safety_state.get("cooldown_until"), safety_counters=safety_state.get("counters", {}),
"cooldown_remaining_seconds": safety_state.get("cooldown_remaining_seconds"),
"safety_counters": safety_state.get("counters", {}),
"metric_name": "ingestion_safety_cooldown_entered_total",
"metric_value": 1,
},
) )
elif pre_apply_state.get("cooldown_active") and not safety_state.get("cooldown_active"): elif pre_apply_state.get("cooldown_active") and not safety_state.get("cooldown_active"):
logger.info( structured_log(
"ingestion.safety_cooldown_cleared", logger, "info", "ingestion.safety_cooldown_cleared",
extra={ user_id=user_id,
"event": "ingestion.safety_cooldown_cleared", crawl_run_id=int(run.id),
"user_id": user_id, reason=pre_apply_state.get("cooldown_reason"),
"crawl_run_id": run_id, cooldown_until=pre_apply_state.get("cooldown_until"),
"reason": pre_apply_state.get("cooldown_reason"),
"cooldown_until": pre_apply_state.get("cooldown_until"),
"metric_name": "ingestion_safety_cooldown_cleared_total",
"metric_value": 1,
},
) )
def _finalize_run_record( def _finalize_run_record(
@ -1174,35 +1011,6 @@ class ScholarIngestionService:
"meta": {"idempotency_key": idempotency_key} if idempotency_key else {}, "meta": {"idempotency_key": idempotency_key} if idempotency_key else {},
} }
@staticmethod
def _log_run_completed(
*,
user_id: int,
run: CrawlRun,
scholars: list[ScholarProfile],
progress: RunProgress,
alert_summary: RunAlertSummary,
failure_summary: RunFailureSummary,
) -> None:
logger.info(
"ingestion.run_completed",
extra={
"event": "ingestion.run_completed",
"user_id": user_id,
"crawl_run_id": run.id,
"status": run.status.value,
"scholar_count": len(scholars),
"succeeded_count": progress.succeeded_count,
"failed_count": progress.failed_count,
"partial_count": progress.partial_count,
"new_publication_count": run.new_pub_count,
"blocked_failure_count": alert_summary.blocked_failure_count,
"network_failure_count": alert_summary.network_failure_count,
"retries_scheduled_count": failure_summary.retries_scheduled_count,
"alert_flags": alert_summary.alert_flags,
},
)
@staticmethod @staticmethod
def _paging_kwargs( def _paging_kwargs(
*, *,
@ -1307,11 +1115,12 @@ class ScholarIngestionService:
alert_network_failure_threshold: int, alert_network_failure_threshold: int,
alert_retry_scheduled_threshold: int, alert_retry_scheduled_threshold: int,
) -> CrawlRun: ) -> CrawlRun:
self._log_run_started( structured_log(
logger, "info", "ingestion.run_started",
user_id=user_id, user_id=user_id,
trigger_type=trigger_type, trigger_type=trigger_type.value,
scholar_count=len(scholars), scholar_count=len(scholars),
filtered=filtered, is_filtered_run=filtered,
request_delay_seconds=request_delay_seconds, request_delay_seconds=request_delay_seconds,
network_error_retries=network_error_retries, network_error_retries=network_error_retries,
retry_backoff_seconds=retry_backoff_seconds, retry_backoff_seconds=retry_backoff_seconds,
@ -1367,9 +1176,9 @@ class ScholarIngestionService:
for index, scholar in enumerate(scholars): for index, scholar in enumerate(scholars):
await db_session.refresh(run) await db_session.refresh(run)
if run.status == RunStatus.CANCELED: if run.status == RunStatus.CANCELED:
logger.info( structured_log(
"ingestion.run_canceled", logger, "info", "ingestion.run_canceled",
extra={"event": "ingestion.run_canceled", "run_id": run.id, "user_id": user_id}, run_id=run.id, user_id=user_id,
) )
return progress return progress
await self._wait_between_scholars(index=index, request_delay_seconds=request_delay_seconds) await self._wait_between_scholars(index=index, request_delay_seconds=request_delay_seconds)
@ -1407,9 +1216,9 @@ class ScholarIngestionService:
continue # first page failed, had no continuation, or was skipped continue # first page failed, had no continuation, or was skipped
await db_session.refresh(run) await db_session.refresh(run)
if run.status == RunStatus.CANCELED: if run.status == RunStatus.CANCELED:
logger.info( structured_log(
"ingestion.run_canceled", logger, "info", "ingestion.run_canceled",
extra={"event": "ingestion.run_canceled", "run_id": run.id, "user_id": user_id}, run_id=run.id, user_id=user_id,
) )
break break
await self._wait_between_scholars(index=index, request_delay_seconds=request_delay_seconds) await self._wait_between_scholars(index=index, request_delay_seconds=request_delay_seconds)
@ -1452,12 +1261,30 @@ class ScholarIngestionService:
alert_network_failure_threshold=alert_network_failure_threshold, alert_network_failure_threshold=alert_network_failure_threshold,
alert_retry_scheduled_threshold=alert_retry_scheduled_threshold, alert_retry_scheduled_threshold=alert_retry_scheduled_threshold,
) )
self._log_alert_thresholds( if alert_summary.alert_flags["blocked_failure_threshold_exceeded"]:
user_id=user_id, structured_log(
run_id=int(run.id), logger, "warning", "ingestion.alert_blocked_failure_threshold_exceeded",
failure_summary=failure_summary, user_id=user_id,
alert_summary=alert_summary, crawl_run_id=int(run.id),
) blocked_failure_count=alert_summary.blocked_failure_count,
threshold=alert_summary.blocked_failure_threshold,
)
if alert_summary.alert_flags["network_failure_threshold_exceeded"]:
structured_log(
logger, "warning", "ingestion.alert_network_failure_threshold_exceeded",
user_id=user_id,
crawl_run_id=int(run.id),
network_failure_count=alert_summary.network_failure_count,
threshold=alert_summary.network_failure_threshold,
)
if alert_summary.alert_flags["retry_scheduled_threshold_exceeded"]:
structured_log(
logger, "warning", "ingestion.alert_retry_scheduled_threshold_exceeded",
user_id=user_id,
crawl_run_id=int(run.id),
retries_scheduled_count=failure_summary.retries_scheduled_count,
threshold=alert_summary.retry_scheduled_threshold,
)
self._apply_safety_outcome(user_settings=user_settings, run=run, user_id=user_id, alert_summary=alert_summary) self._apply_safety_outcome(user_settings=user_settings, run=run, user_id=user_id, alert_summary=alert_summary)
self._finalize_run_record( self._finalize_run_record(
run=run, run=run,
@ -1491,10 +1318,14 @@ class ScholarIngestionService:
) -> tuple[CrawlRun, list[ScholarProfile], dict[int, int]]: ) -> tuple[CrawlRun, list[ScholarProfile], dict[int, int]]:
effective_delay = self._effective_request_delay_seconds(request_delay_seconds) effective_delay = self._effective_request_delay_seconds(request_delay_seconds)
if effective_delay != _int_or_default(request_delay_seconds, effective_delay): if effective_delay != _int_or_default(request_delay_seconds, effective_delay):
self._log_request_delay_coercion( structured_log(
logger, "warning", "ingestion.request_delay_coerced_to_policy_floor",
user_id=user_id, user_id=user_id,
requested_request_delay_seconds=_int_or_default(request_delay_seconds, 0), requested_request_delay_seconds=_int_or_default(request_delay_seconds, 0),
effective_request_delay_seconds=effective_delay, effective_request_delay_seconds=effective_delay,
policy_minimum_request_delay_seconds=user_settings_service.resolve_request_delay_minimum(
settings.ingestion_min_request_delay_seconds
),
) )
paging_kwargs = self._paging_kwargs( paging_kwargs = self._paging_kwargs(
@ -1595,7 +1426,21 @@ class ScholarIngestionService:
if intended_final_status not in (RunStatus.CANCELED,): if intended_final_status not in (RunStatus.CANCELED,):
run.status = RunStatus.RESOLVING run.status = RunStatus.RESOLVING
await db_session.commit() await db_session.commit()
self._log_run_completed(user_id=user_id, run=run, scholars=attached_scholars, progress=progress, alert_summary=alert_summary, failure_summary=failure_summary) structured_log(
logger, "info", "ingestion.run_completed",
user_id=user_id,
crawl_run_id=run.id,
status=run.status.value,
scholar_count=len(attached_scholars),
succeeded_count=progress.succeeded_count,
failed_count=progress.failed_count,
partial_count=progress.partial_count,
new_publication_count=run.new_pub_count,
blocked_failure_count=alert_summary.blocked_failure_count,
network_failure_count=alert_summary.network_failure_count,
retries_scheduled_count=failure_summary.retries_scheduled_count,
alert_flags=alert_summary.alert_flags,
)
# Fire-and-forget enrichment in a separate background task # Fire-and-forget enrichment in a separate background task
if intended_final_status not in (RunStatus.CANCELED,): if intended_final_status not in (RunStatus.CANCELED,):
@ -1609,7 +1454,7 @@ class ScholarIngestionService:
) )
except Exception as exc: except Exception as exc:
await db_session.rollback() await db_session.rollback()
logger.exception("ingestion.background_run_failed", extra={"event": "ingestion.background_run_failed", "run_id": run_id, "user_id": user_id}) logger.exception("ingestion.background_run_failed", extra={"run_id": run_id, "user_id": user_id})
await self._fail_run_in_background(session_factory, run_id, exc) await self._fail_run_in_background(session_factory, run_id, exc)
async def _prepare_execute_run( async def _prepare_execute_run(
@ -1782,7 +1627,21 @@ class ScholarIngestionService:
run.status = intended_final_status run.status = intended_final_status
await db_session.commit() await db_session.commit()
self._log_run_completed(user_id=user_id, run=run, scholars=scholars, progress=progress, alert_summary=alert_summary, failure_summary=failure_summary) structured_log(
logger, "info", "ingestion.run_completed",
user_id=user_id,
crawl_run_id=run.id,
status=run.status.value,
scholar_count=len(scholars),
succeeded_count=progress.succeeded_count,
failed_count=progress.failed_count,
partial_count=progress.partial_count,
new_publication_count=run.new_pub_count,
blocked_failure_count=alert_summary.blocked_failure_count,
network_failure_count=alert_summary.network_failure_count,
retries_scheduled_count=failure_summary.retries_scheduled_count,
alert_flags=alert_summary.alert_flags,
)
return self._run_execution_summary(run=run, scholars=scholars, progress=progress) return self._run_execution_summary(run=run, scholars=scholars, progress=progress)
async def _fetch_profile_page( async def _fetch_profile_page(
@ -1816,7 +1675,6 @@ class ScholarIngestionService:
logger.exception( logger.exception(
"ingestion.fetch_unexpected_error", "ingestion.fetch_unexpected_error",
extra={ extra={
"event": "ingestion.fetch_unexpected_error",
"scholar_id": scholar_id, "scholar_id": scholar_id,
"cstart": cstart, "cstart": cstart,
"page_size": page_size, "page_size": page_size,
@ -1833,23 +1691,6 @@ class ScholarIngestionService:
error=str(exc), error=str(exc),
) )
@staticmethod
def _attempt_log_entry(
*,
attempt: int,
cstart: int,
fetch_result: FetchResult,
parsed_page: ParsedProfilePage,
) -> dict[str, Any]:
return {
"attempt": attempt,
"cstart": cstart,
"state": parsed_page.state.value,
"state_reason": parsed_page.state_reason,
"status_code": fetch_result.status_code,
"fetch_error": fetch_result.error,
}
@staticmethod @staticmethod
def _should_retry_page_fetch( def _should_retry_page_fetch(
*, *,
@ -1883,16 +1724,13 @@ class ScholarIngestionService:
sleep_seconds = retry_backoff_seconds * (2 ** (network_attempt_count - 1)) sleep_seconds = retry_backoff_seconds * (2 ** (network_attempt_count - 1))
attempt_label = network_attempt_count attempt_label = network_attempt_count
logger.warning( structured_log(
"ingestion.scholar_retry_scheduled", logger, "warning", "ingestion.scholar_retry_scheduled",
extra={ scholar_id=scholar_id,
"event": "ingestion.scholar_retry_scheduled", cstart=cstart,
"scholar_id": scholar_id, attempt_count=attempt_label,
"cstart": cstart, sleep_seconds=sleep_seconds,
"attempt_count": attempt_label, state_reason=state_reason,
"sleep_seconds": sleep_seconds,
"state_reason": state_reason,
},
) )
if sleep_seconds > 0: if sleep_seconds > 0:
await asyncio.sleep(sleep_seconds) await asyncio.sleep(sleep_seconds)
@ -1931,14 +1769,14 @@ class ScholarIngestionService:
else: else:
total_attempts = network_attempts + rate_limit_attempts + 1 total_attempts = network_attempts + rate_limit_attempts + 1
attempt_log.append( attempt_log.append({
self._attempt_log_entry( "attempt": total_attempts,
attempt=total_attempts, "cstart": cstart,
cstart=cstart, "state": parsed_page.state.value,
fetch_result=fetch_result, "state_reason": parsed_page.state_reason,
parsed_page=parsed_page, "status_code": fetch_result.status_code,
) "fetch_error": fetch_result.error,
) })
if not self._should_retry_page_fetch( if not self._should_retry_page_fetch(
parsed_page=parsed_page, parsed_page=parsed_page,
@ -1995,28 +1833,6 @@ class ScholarIngestionService:
articles_range=None, articles_range=None,
) )
@staticmethod
def _page_log_entry(
*,
page_number: int,
cstart: int,
fetch_result: FetchResult,
parsed_page: ParsedProfilePage,
attempt_count: int,
) -> dict[str, Any]:
return {
"page": page_number,
"cstart": cstart,
"state": parsed_page.state.value,
"state_reason": parsed_page.state_reason,
"status_code": fetch_result.status_code,
"publication_count": len(parsed_page.publications),
"articles_range": parsed_page.articles_range,
"has_show_more_button": parsed_page.has_show_more_button,
"warning_codes": parsed_page.warnings,
"attempt_count": attempt_count,
}
@staticmethod @staticmethod
def _should_skip_no_change( def _should_skip_no_change(
*, *,
@ -2179,15 +1995,18 @@ class ScholarIngestionService:
) -> None: ) -> None:
state.pages_attempted += 1 state.pages_attempted += 1
state.attempt_log.extend(page_attempt_log) state.attempt_log.extend(page_attempt_log)
state.page_logs.append( state.page_logs.append({
ScholarIngestionService._page_log_entry( "page": state.pages_attempted,
page_number=state.pages_attempted, "cstart": state.current_cstart,
cstart=state.current_cstart, "state": parsed_page.state.value,
fetch_result=fetch_result, "state_reason": parsed_page.state_reason,
parsed_page=parsed_page, "status_code": fetch_result.status_code,
attempt_count=len(page_attempt_log), "publication_count": len(parsed_page.publications),
) "articles_range": parsed_page.articles_range,
) "has_show_more_button": parsed_page.has_show_more_button,
"warning_codes": parsed_page.warnings,
"attempt_count": len(page_attempt_log),
})
state.fetch_result = fetch_result state.fetch_result = fetch_result
state.parsed_page = parsed_page state.parsed_page = parsed_page
@ -2233,15 +2052,18 @@ class ScholarIngestionService:
) )
first_page_fingerprint_sha256 = build_initial_page_fingerprint(parsed_page) first_page_fingerprint_sha256 = build_initial_page_fingerprint(parsed_page)
attempt_log = list(first_attempt_log) attempt_log = list(first_attempt_log)
page_logs = [ page_logs = [{
self._page_log_entry( "page": 1,
page_number=1, "cstart": start_cstart,
cstart=start_cstart, "state": parsed_page.state.value,
fetch_result=fetch_result, "state_reason": parsed_page.state_reason,
parsed_page=parsed_page, "status_code": fetch_result.status_code,
attempt_count=len(first_attempt_log), "publication_count": len(parsed_page.publications),
) "articles_range": parsed_page.articles_range,
] "has_show_more_button": parsed_page.has_show_more_button,
"warning_codes": parsed_page.warnings,
"attempt_count": len(first_attempt_log),
}]
return fetch_result, parsed_page, first_page_fingerprint_sha256, attempt_log, page_logs return fetch_result, parsed_page, first_page_fingerprint_sha256, attempt_log, page_logs
async def _paginate_loop( async def _paginate_loop(
@ -2275,9 +2097,9 @@ class ScholarIngestionService:
while state.parsed_page.has_show_more_button: while state.parsed_page.has_show_more_button:
await db_session.refresh(run) await db_session.refresh(run)
if run.status == RunStatus.CANCELED: if run.status == RunStatus.CANCELED:
logger.info( structured_log(
"ingestion.pagination_canceled", logger, "info", "ingestion.pagination_canceled",
extra={"event": "ingestion.pagination_canceled", "run_id": run.id}, run_id=run.id,
) )
self._set_truncated_state( self._set_truncated_state(
state=state, state=state,
@ -2532,22 +2354,22 @@ class ScholarIngestionService:
{"title.search": "|".join(titles)}, limit=batch_size * 3 {"title.search": "|".join(titles)}, limit=batch_size * 3
) )
except OpenAlexBudgetExhaustedError: except OpenAlexBudgetExhaustedError:
logger.warning( structured_log(
"ingestion.openalex_budget_exhausted", logger, "warning", "ingestion.openalex_budget_exhausted",
extra={"event": "ingestion.openalex_budget_exhausted", "run_id": run_id}, run_id=run_id,
) )
break # Stop all enrichment — budget won't reset until midnight UTC break # Stop all enrichment — budget won't reset until midnight UTC
except OpenAlexRateLimitError: except OpenAlexRateLimitError:
logger.warning( structured_log(
"ingestion.openalex_rate_limited", logger, "warning", "ingestion.openalex_rate_limited",
extra={"event": "ingestion.openalex_rate_limited", "run_id": run_id}, run_id=run_id,
) )
await asyncio.sleep(60) await asyncio.sleep(60)
continue continue
except Exception as e: except Exception as e:
logger.warning( structured_log(
"ingestion.openalex_enrichment_failed", logger, "warning", "ingestion.openalex_enrichment_failed",
extra={"event": "ingestion.openalex_enrichment_failed", "error": str(e), "run_id": run_id}, error=str(e), run_id=run_id,
) )
continue continue
@ -2583,13 +2405,10 @@ class ScholarIngestionService:
merge_count = await sweep_identifier_duplicates(db_session) merge_count = await sweep_identifier_duplicates(db_session)
if merge_count: if merge_count:
logger.info( structured_log(
"ingestion.identifier_dedup_sweep", logger, "info", "ingestion.identifier_dedup_sweep",
extra={ merged_count=merge_count,
"event": "ingestion.identifier_dedup_sweep", run_id=run_id,
"merged_count": merge_count,
"run_id": run_id,
},
) )
async def _discover_identifiers_for_enrichment( async def _discover_identifiers_for_enrichment(
@ -2624,14 +2443,11 @@ class ScholarIngestionService:
) )
return True return True
except ArxivRateLimitError: except ArxivRateLimitError:
logger.warning( structured_log(
"ingestion.arxiv_rate_limited", logger, "warning", "ingestion.arxiv_rate_limited",
extra={ run_id=run_id,
"event": "ingestion.arxiv_rate_limited", publication_id=int(publication.id),
"run_id": run_id, detail="arXiv temporarily disabled for remaining enrichment pass",
"publication_id": int(publication.id),
"detail": "arXiv temporarily disabled for remaining enrichment pass",
},
) )
await identifier_service.sync_identifiers_for_publication_fields( await identifier_service.sync_identifiers_for_publication_fields(
db_session, db_session,
@ -2780,14 +2596,11 @@ class ScholarIngestionService:
db_session.add(link) db_session.add(link)
discovered_count += 1 discovered_count += 1
logger.debug( structured_log(
"ingestion.publication_discovered", logger, "debug", "ingestion.publication_discovered",
extra={ scholar_profile_id=scholar.id,
"event": "ingestion.publication_discovered", publication_id=publication.id,
"scholar_profile_id": scholar.id, crawl_run_id=run.id,
"publication_id": publication.id,
"crawl_run_id": run.id,
},
) )
await self._commit_discovered_publication( await self._commit_discovered_publication(
@ -2907,13 +2720,10 @@ class ScholarIngestionService:
) )
db_session.add(publication) db_session.add(publication)
await db_session.flush() await db_session.flush()
logger.debug( structured_log(
"ingestion.publication_created", logger, "debug", "ingestion.publication_created",
extra={ publication_id=publication.id,
"event": "ingestion.publication_created", cluster_id=publication.cluster_id,
"publication_id": publication.id,
"cluster_id": publication.cluster_id,
},
) )
return publication return publication