diff --git a/DECOMPOSITION_SLICES.md b/DECOMPOSITION_SLICES.md index f14f072..4442463 100644 --- a/DECOMPOSITION_SLICES.md +++ b/DECOMPOSITION_SLICES.md @@ -29,7 +29,7 @@ This file contains self-contained decomposition prompts ("slices") to bring the | 2 | Ingestion: pagination + publication upsert | **DONE** | | 3 | Ingestion: enrichment + scholar processing + run completion | **DONE** | | 4 | Scholars service + PDF queue | **DONE** | -| 5 | Routers + scheduler | pending | +| 5 | Routers + scheduler | **DONE** | --- @@ -177,9 +177,21 @@ Commit message: `refactor: decompose scholars service and pdf_queue into focused --- -## Slice 5: Decompose router files and scheduler +## Slice 5: Decompose router files and scheduler — DONE -**Why:** Three files slightly over the 600-line ceiling. Each needs helper extraction. +Completed. Extracted helpers from three oversized files: + +| File | Lines | Contents | +|---|---|---| +| `app/api/routers/run_serializers.py` | 238 | `serialize_run`, `serialize_queue_item`, `normalize_scholar_result`, `manual_run_payload_from_run`, `manual_run_success_payload`, value coercion helpers, idempotency key validation | +| `app/api/routers/run_manual.py` | 237 | `load_safety_state`, `raise_manual_runs_disabled`, `reused_manual_run_payload`, `run_ingestion_for_manual`, `recover_integrity_error`, `execute_manual_run`, safety/failure raise helpers | +| `app/api/routers/runs.py` | 440 | Route handlers only + imports | +| `app/api/routers/scholar_helpers.py` | 234 | `serialize_scholar`, `hydrate_scholar_metadata_if_needed`, `enqueue_initial_scrape_job_for_scholar`, `search_kwargs`, `search_response_data`, `require_user_profile`, `read_uploaded_image` | +| `app/api/routers/scholars.py` | 406 | Route handlers only + imports | +| `app/services/ingestion/queue_runner.py` | 379 | `QueueJobRunner` class — drain continuation queue, job lifecycle (drop/retry/reschedule), ingestion dispatch, finalization | +| `app/services/ingestion/scheduler.py` | 312 | `SchedulerService` — auto-run scheduling, `_run_loop`, `_tick_once`, candidate loading, PDF queue drain | + +Also updated one test file (`test_scholars_create_hydration.py`) to monkeypatch `scholar_helpers` instead of the `scholars` router for the 4 helper-related tests. **Files to create:** `app/api/routers/run_serializers.py`, `app/api/routers/run_manual.py`, `app/api/routers/scholar_helpers.py`, `app/services/ingestion/queue_runner.py` diff --git a/app/api/routers/run_manual.py b/app/api/routers/run_manual.py new file mode 100644 index 0000000..6360e34 --- /dev/null +++ b/app/api/routers/run_manual.py @@ -0,0 +1,237 @@ +from __future__ import annotations + +import logging +from typing import Any + +from fastapi import Request +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.errors import ApiException +from app.api.responses import success_payload +from app.api.routers.run_serializers import manual_run_payload_from_run +from app.db.models import RunStatus, RunTriggerType +from app.logging_utils import structured_log +from app.services.ingestion import application as ingestion_service +from app.services.ingestion import safety as run_safety_service +from app.services.runs import application as run_service +from app.services.settings import application as user_settings_service +from app.settings import settings + +logger = logging.getLogger(__name__) + + +async def load_safety_state( + db_session: AsyncSession, + *, + user_id: int, +) -> dict[str, Any]: + user_settings = await user_settings_service.get_or_create_settings( + db_session, + user_id=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) + structured_log( + logger, + "info", + "api.runs.safety_cooldown_cleared", + user_id=user_id, + reason=previous_safety_state.get("cooldown_reason"), + cooldown_until=previous_safety_state.get("cooldown_until"), + ) + return run_safety_service.get_safety_state_payload(user_settings) + + +def raise_manual_runs_disabled(*, user_id: int, safety_state: dict[str, Any]) -> None: + structured_log( + logger, + "warning", + "api.runs.manual_blocked_policy", + user_id=user_id, + policy={"manual_run_allowed": False}, + safety_state=safety_state, + ) + 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 in (RunStatus.RUNNING, RunStatus.RESOLVING): + 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={"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={"user_id": user_id}, + ) + raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc + if existing_run.status in (RunStatus.RUNNING, RunStatus.RESOLVING): + 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: + structured_log( + logger, + "info", + "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"), + ) + 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={"user_id": user_id}, + ) + raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from exc diff --git a/app/api/routers/run_serializers.py b/app/api/routers/run_serializers.py new file mode 100644 index 0000000..53db3b5 --- /dev/null +++ b/app/api/routers/run_serializers.py @@ -0,0 +1,238 @@ +from __future__ import annotations + +import re +from typing import Any + +from app.api.errors import ApiException +from app.services.runs import application as run_service + +IDEMPOTENCY_HEADER = "Idempotency-Key" +IDEMPOTENCY_MAX_LENGTH = 128 +IDEMPOTENCY_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{8,128}$") + + +def _int_value(value: Any, default: int = 0) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default + + +def _str_value(value: Any) -> str | None: + if value is None: + return None + text = str(value).strip() + return text if text else None + + +def _bool_value(value: Any, default: bool = False) -> bool: + if isinstance(value, bool): + return value + if isinstance(value, str): + lowered = value.strip().lower() + if lowered in {"1", "true", "yes", "on"}: + return True + if lowered in {"0", "false", "no", "off"}: + return False + return default + + +def normalize_idempotency_key(raw_value: str | None) -> str | None: + if raw_value is None: + return None + candidate = raw_value.strip() + if not candidate: + return None + if len(candidate) > IDEMPOTENCY_MAX_LENGTH or not IDEMPOTENCY_PATTERN.match(candidate): + raise ApiException( + status_code=400, + code="invalid_idempotency_key", + message=("Invalid Idempotency-Key. Use 8-128 characters from: A-Z a-z 0-9 . _ : -"), + ) + return candidate + + +def serialize_run(run) -> dict[str, Any]: + summary = run_service.extract_run_summary(run.error_log) + return { + "id": int(run.id), + "trigger_type": run.trigger_type.value, + "status": run.status.value, + "start_dt": run.start_dt, + "end_dt": run.end_dt, + "scholar_count": int(run.scholar_count or 0), + "new_publication_count": int(run.new_pub_count or 0), + "failed_count": int(summary["failed_count"]), + "partial_count": int(summary["partial_count"]), + } + + +def serialize_queue_item(item) -> dict[str, Any]: + return { + "id": int(item.id), + "scholar_profile_id": int(item.scholar_profile_id), + "scholar_label": item.scholar_label, + "status": item.status, + "reason": item.reason, + "dropped_reason": item.dropped_reason, + "attempt_count": int(item.attempt_count), + "resume_cstart": int(item.resume_cstart), + "next_attempt_dt": item.next_attempt_dt, + "updated_at": item.updated_at, + "last_error": item.last_error, + "last_run_id": item.last_run_id, + } + + +def _normalize_attempt_log(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + normalized: list[dict[str, Any]] = [] + for item in value: + if not isinstance(item, dict): + continue + normalized.append( + { + "attempt": _int_value(item.get("attempt"), 0), + "cstart": _int_value(item.get("cstart"), 0), + "state": _str_value(item.get("state")), + "state_reason": _str_value(item.get("state_reason")), + "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None), + "fetch_error": _str_value(item.get("fetch_error")), + } + ) + return normalized + + +def _normalize_page_logs(value: Any) -> list[dict[str, Any]]: + if not isinstance(value, list): + return [] + normalized: list[dict[str, Any]] = [] + for item in value: + if not isinstance(item, dict): + continue + warning_codes = item.get("warning_codes") + normalized.append( + { + "page": _int_value(item.get("page"), 0), + "cstart": _int_value(item.get("cstart"), 0), + "state": _str_value(item.get("state")) or "unknown", + "state_reason": _str_value(item.get("state_reason")), + "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None), + "publication_count": _int_value(item.get("publication_count"), 0), + "attempt_count": _int_value(item.get("attempt_count"), 0), + "has_show_more_button": _bool_value(item.get("has_show_more_button"), False), + "articles_range": _str_value(item.get("articles_range")), + "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])], + } + ) + return normalized + + +def _normalize_debug(value: Any) -> dict[str, Any] | None: + if not isinstance(value, dict): + return None + marker_counts = value.get("marker_counts_nonzero") + warning_codes = value.get("warning_codes") + return { + "status_code": (_int_value(value.get("status_code")) if value.get("status_code") is not None else None), + "final_url": _str_value(value.get("final_url")), + "fetch_error": _str_value(value.get("fetch_error")), + "body_sha256": _str_value(value.get("body_sha256")), + "body_length": (_int_value(value.get("body_length")) if value.get("body_length") is not None else None), + "has_show_more_button": ( + _bool_value(value.get("has_show_more_button"), False) + if value.get("has_show_more_button") is not None + else None + ), + "articles_range": _str_value(value.get("articles_range")), + "state_reason": _str_value(value.get("state_reason")), + "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])], + "marker_counts_nonzero": { + str(key): _int_value(count, 0) + for key, count in (marker_counts.items() if isinstance(marker_counts, dict) else []) + }, + "page_logs": _normalize_page_logs(value.get("page_logs")), + "attempt_log": _normalize_attempt_log(value.get("attempt_log")), + } + + +def normalize_scholar_result(value: Any) -> dict[str, Any]: + if not isinstance(value, dict): + return { + "scholar_profile_id": 0, + "scholar_id": "unknown", + "state": "unknown", + "state_reason": None, + "outcome": "failed", + "attempt_count": 0, + "publication_count": 0, + "start_cstart": 0, + "continuation_cstart": None, + "continuation_enqueued": False, + "continuation_cleared": False, + "warnings": [], + "error": None, + "debug": None, + } + warnings = value.get("warnings") + return { + "scholar_profile_id": _int_value(value.get("scholar_profile_id"), 0), + "scholar_id": _str_value(value.get("scholar_id")) or "unknown", + "state": _str_value(value.get("state")) or "unknown", + "state_reason": _str_value(value.get("state_reason")), + "outcome": _str_value(value.get("outcome")) or "failed", + "attempt_count": _int_value(value.get("attempt_count"), 0), + "publication_count": _int_value(value.get("publication_count"), 0), + "start_cstart": _int_value(value.get("start_cstart"), 0), + "continuation_cstart": ( + _int_value(value.get("continuation_cstart")) if value.get("continuation_cstart") is not None else None + ), + "continuation_enqueued": _bool_value(value.get("continuation_enqueued"), False), + "continuation_cleared": _bool_value(value.get("continuation_cleared"), False), + "warnings": [str(item) for item in (warnings if isinstance(warnings, list) else [])], + "error": _str_value(value.get("error")), + "debug": _normalize_debug(value.get("debug")), + } + + +def manual_run_payload_from_run( + run, + *, + idempotency_key: str | None, + reused_existing_run: bool, + safety_state: dict[str, Any], +) -> dict[str, Any]: + summary = run_service.extract_run_summary(run.error_log) + return { + "run_id": int(run.id), + "status": run.status.value, + "scholar_count": int(run.scholar_count or 0), + "succeeded_count": int(summary["succeeded_count"]), + "failed_count": int(summary["failed_count"]), + "partial_count": int(summary["partial_count"]), + "new_publication_count": int(run.new_pub_count or 0), + "reused_existing_run": reused_existing_run, + "idempotency_key": idempotency_key, + "safety_state": safety_state, + } + + +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, + } diff --git a/app/api/routers/runs.py b/app/api/routers/runs.py index 8821ebf..687f6ce 100644 --- a/app/api/routers/runs.py +++ b/app/api/routers/runs.py @@ -2,7 +2,6 @@ from __future__ import annotations import asyncio import logging -import re from typing import Any from fastapi import APIRouter, Depends, Query, Request @@ -13,6 +12,21 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_api_current_user from app.api.errors import ApiException from app.api.responses import success_payload +from app.api.routers.run_manual import ( + load_safety_state, + raise_manual_blocked_safety, + raise_manual_failed, + raise_manual_runs_disabled, + recover_integrity_error, + reused_manual_run_payload, +) +from app.api.routers.run_serializers import ( + IDEMPOTENCY_HEADER, + normalize_idempotency_key, + normalize_scholar_result, + serialize_queue_item, + serialize_run, +) from app.api.runtime_deps import get_ingestion_service from app.api.schemas import ( ManualRunEnvelope, @@ -24,9 +38,7 @@ from app.api.schemas import ( ) from app.db.models import RunStatus, RunTriggerType, User from app.db.session import get_db_session -from app.logging_utils import structured_log from app.services.ingestion import application as ingestion_service -from app.services.ingestion import safety as run_safety_service from app.services.runs import application as run_service from app.services.runs.events import event_generator from app.services.settings import application as user_settings_service @@ -38,453 +50,6 @@ _background_tasks: set[asyncio.Task[Any]] = set() router = APIRouter(prefix="/runs", tags=["api-runs"]) ACTIVE_RUN_STATUSES = {RunStatus.RUNNING, RunStatus.RESOLVING} -IDEMPOTENCY_HEADER = "Idempotency-Key" -IDEMPOTENCY_MAX_LENGTH = 128 -IDEMPOTENCY_PATTERN = re.compile(r"^[A-Za-z0-9._:-]{8,128}$") - - -def _int_value(value: Any, default: int = 0) -> int: - try: - return int(value) - except (TypeError, ValueError): - return default - - -def _str_value(value: Any) -> str | None: - if value is None: - return None - text = str(value).strip() - return text if text else None - - -def _bool_value(value: Any, default: bool = False) -> bool: - if isinstance(value, bool): - return value - if isinstance(value, str): - lowered = value.strip().lower() - if lowered in {"1", "true", "yes", "on"}: - return True - if lowered in {"0", "false", "no", "off"}: - return False - return default - - -def _normalize_idempotency_key(raw_value: str | None) -> str | None: - if raw_value is None: - return None - candidate = raw_value.strip() - if not candidate: - return None - if len(candidate) > IDEMPOTENCY_MAX_LENGTH or not IDEMPOTENCY_PATTERN.match(candidate): - raise ApiException( - status_code=400, - code="invalid_idempotency_key", - message=("Invalid Idempotency-Key. Use 8-128 characters from: A-Z a-z 0-9 . _ : -"), - ) - return candidate - - -def _serialize_run(run) -> dict[str, Any]: - summary = run_service.extract_run_summary(run.error_log) - return { - "id": int(run.id), - "trigger_type": run.trigger_type.value, - "status": run.status.value, - "start_dt": run.start_dt, - "end_dt": run.end_dt, - "scholar_count": int(run.scholar_count or 0), - "new_publication_count": int(run.new_pub_count or 0), - "failed_count": int(summary["failed_count"]), - "partial_count": int(summary["partial_count"]), - } - - -def _serialize_queue_item(item) -> dict[str, Any]: - return { - "id": int(item.id), - "scholar_profile_id": int(item.scholar_profile_id), - "scholar_label": item.scholar_label, - "status": item.status, - "reason": item.reason, - "dropped_reason": item.dropped_reason, - "attempt_count": int(item.attempt_count), - "resume_cstart": int(item.resume_cstart), - "next_attempt_dt": item.next_attempt_dt, - "updated_at": item.updated_at, - "last_error": item.last_error, - "last_run_id": item.last_run_id, - } - - -def _normalize_attempt_log(value: Any) -> list[dict[str, Any]]: - if not isinstance(value, list): - return [] - normalized: list[dict[str, Any]] = [] - for item in value: - if not isinstance(item, dict): - continue - normalized.append( - { - "attempt": _int_value(item.get("attempt"), 0), - "cstart": _int_value(item.get("cstart"), 0), - "state": _str_value(item.get("state")), - "state_reason": _str_value(item.get("state_reason")), - "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None), - "fetch_error": _str_value(item.get("fetch_error")), - } - ) - return normalized - - -def _normalize_page_logs(value: Any) -> list[dict[str, Any]]: - if not isinstance(value, list): - return [] - normalized: list[dict[str, Any]] = [] - for item in value: - if not isinstance(item, dict): - continue - warning_codes = item.get("warning_codes") - normalized.append( - { - "page": _int_value(item.get("page"), 0), - "cstart": _int_value(item.get("cstart"), 0), - "state": _str_value(item.get("state")) or "unknown", - "state_reason": _str_value(item.get("state_reason")), - "status_code": (_int_value(item.get("status_code")) if item.get("status_code") is not None else None), - "publication_count": _int_value(item.get("publication_count"), 0), - "attempt_count": _int_value(item.get("attempt_count"), 0), - "has_show_more_button": _bool_value(item.get("has_show_more_button"), False), - "articles_range": _str_value(item.get("articles_range")), - "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])], - } - ) - return normalized - - -def _normalize_debug(value: Any) -> dict[str, Any] | None: - if not isinstance(value, dict): - return None - marker_counts = value.get("marker_counts_nonzero") - warning_codes = value.get("warning_codes") - return { - "status_code": (_int_value(value.get("status_code")) if value.get("status_code") is not None else None), - "final_url": _str_value(value.get("final_url")), - "fetch_error": _str_value(value.get("fetch_error")), - "body_sha256": _str_value(value.get("body_sha256")), - "body_length": (_int_value(value.get("body_length")) if value.get("body_length") is not None else None), - "has_show_more_button": ( - _bool_value(value.get("has_show_more_button"), False) - if value.get("has_show_more_button") is not None - else None - ), - "articles_range": _str_value(value.get("articles_range")), - "state_reason": _str_value(value.get("state_reason")), - "warning_codes": [str(code) for code in (warning_codes if isinstance(warning_codes, list) else [])], - "marker_counts_nonzero": { - str(key): _int_value(count, 0) - for key, count in (marker_counts.items() if isinstance(marker_counts, dict) else []) - }, - "page_logs": _normalize_page_logs(value.get("page_logs")), - "attempt_log": _normalize_attempt_log(value.get("attempt_log")), - } - - -def _normalize_scholar_result(value: Any) -> dict[str, Any]: - if not isinstance(value, dict): - return { - "scholar_profile_id": 0, - "scholar_id": "unknown", - "state": "unknown", - "state_reason": None, - "outcome": "failed", - "attempt_count": 0, - "publication_count": 0, - "start_cstart": 0, - "continuation_cstart": None, - "continuation_enqueued": False, - "continuation_cleared": False, - "warnings": [], - "error": None, - "debug": None, - } - warnings = value.get("warnings") - return { - "scholar_profile_id": _int_value(value.get("scholar_profile_id"), 0), - "scholar_id": _str_value(value.get("scholar_id")) or "unknown", - "state": _str_value(value.get("state")) or "unknown", - "state_reason": _str_value(value.get("state_reason")), - "outcome": _str_value(value.get("outcome")) or "failed", - "attempt_count": _int_value(value.get("attempt_count"), 0), - "publication_count": _int_value(value.get("publication_count"), 0), - "start_cstart": _int_value(value.get("start_cstart"), 0), - "continuation_cstart": ( - _int_value(value.get("continuation_cstart")) if value.get("continuation_cstart") is not None else None - ), - "continuation_enqueued": _bool_value(value.get("continuation_enqueued"), False), - "continuation_cleared": _bool_value(value.get("continuation_cleared"), False), - "warnings": [str(item) for item in (warnings if isinstance(warnings, list) else [])], - "error": _str_value(value.get("error")), - "debug": _normalize_debug(value.get("debug")), - } - - -def _manual_run_payload_from_run( - run, - *, - idempotency_key: str | None, - reused_existing_run: bool, - safety_state: dict[str, Any], -) -> dict[str, Any]: - summary = run_service.extract_run_summary(run.error_log) - return { - "run_id": int(run.id), - "status": run.status.value, - "scholar_count": int(run.scholar_count or 0), - "succeeded_count": int(summary["succeeded_count"]), - "failed_count": int(summary["failed_count"]), - "partial_count": int(summary["partial_count"]), - "new_publication_count": int(run.new_pub_count or 0), - "reused_existing_run": reused_existing_run, - "idempotency_key": idempotency_key, - "safety_state": safety_state, - } - - -async def _load_safety_state( - db_session: AsyncSession, - *, - user_id: int, -) -> dict[str, Any]: - user_settings = await user_settings_service.get_or_create_settings( - db_session, - user_id=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) - structured_log( - logger, - "info", - "api.runs.safety_cooldown_cleared", - user_id=user_id, - reason=previous_safety_state.get("cooldown_reason"), - cooldown_until=previous_safety_state.get("cooldown_until"), - ) - return run_safety_service.get_safety_state_payload(user_settings) - - -def _raise_manual_runs_disabled(*, user_id: int, safety_state: dict[str, Any]) -> None: - structured_log( - logger, - "warning", - "api.runs.manual_blocked_policy", - user_id=user_id, - policy={"manual_run_allowed": False}, - safety_state=safety_state, - ) - 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 in (RunStatus.RUNNING, RunStatus.RESOLVING): - 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={"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={"user_id": user_id}, - ) - raise ApiException(status_code=500, code="manual_run_failed", message="Manual run failed.") from original_exc - if existing_run.status in (RunStatus.RUNNING, RunStatus.RESOLVING): - 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: - structured_log( - logger, - "info", - "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"), - ) - 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={"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( "", @@ -503,14 +68,14 @@ async def list_runs( limit=limit, failed_only=failed_only, ) - safety_state = await _load_safety_state( + safety_state = await load_safety_state( db_session, user_id=current_user.id, ) return success_payload( request, data={ - "runs": [_serialize_run(run) for run in runs], + "runs": [serialize_run(run) for run in runs], "safety_state": safety_state, }, ) @@ -541,16 +106,16 @@ async def get_run( scholar_results = error_log.get("scholar_results") if not isinstance(scholar_results, list): scholar_results = [] - safety_state = await _load_safety_state( + safety_state = await load_safety_state( db_session, user_id=current_user.id, ) return success_payload( request, data={ - "run": _serialize_run(run), + "run": serialize_run(run), "summary": run_service.extract_run_summary(error_log), - "scholar_results": [_normalize_scholar_result(item) for item in scholar_results], + "scholar_results": [normalize_scholar_result(item) for item in scholar_results], "safety_state": safety_state, }, ) @@ -595,7 +160,7 @@ async def cancel_run( if not isinstance(scholar_results, list): scholar_results = [] - safety_state = await _load_safety_state( + safety_state = await load_safety_state( db_session, user_id=current_user.id, ) @@ -603,9 +168,9 @@ async def cancel_run( return success_payload( request, data={ - "run": _serialize_run(run), + "run": serialize_run(run), "summary": run_service.extract_run_summary(error_log), - "scholar_results": [_normalize_scholar_result(item) for item in scholar_results], + "scholar_results": [normalize_scholar_result(item) for item in scholar_results], "safety_state": safety_state, }, ) @@ -621,12 +186,12 @@ async def run_manual( current_user: User = Depends(get_api_current_user), ingest_service: ingestion_service.ScholarIngestionService = Depends(get_ingestion_service), ): - safety_state = await _load_safety_state(db_session, user_id=current_user.id) + safety_state = await load_safety_state(db_session, user_id=current_user.id) if not settings.ingestion_manual_run_allowed: - _raise_manual_runs_disabled(user_id=current_user.id, 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)) - reused_payload = await _reused_manual_run_payload( + idempotency_key = normalize_idempotency_key(request.headers.get(IDEMPOTENCY_HEADER)) + reused_payload = await reused_manual_run_payload( db_session, request=request, user_id=current_user.id, @@ -695,12 +260,12 @@ async def run_manual( "new_publication_count": 0, "reused_existing_run": False, "idempotency_key": idempotency_key, - "safety_state": await _load_safety_state(db_session, user_id=current_user.id), + "safety_state": await load_safety_state(db_session, user_id=current_user.id), }, ) except ingestion_service.RunBlockedBySafetyPolicyError as exc: await db_session.rollback() - _raise_manual_blocked_safety(exc=exc, user_id=current_user.id) + raise_manual_blocked_safety(exc=exc, user_id=current_user.id) except ingestion_service.RunAlreadyInProgressError as exc: await db_session.rollback() raise ApiException( @@ -710,7 +275,7 @@ async def run_manual( ) from exc except IntegrityError as exc: await db_session.rollback() - return await _recover_integrity_error( + return await recover_integrity_error( db_session, request=request, user_id=current_user.id, @@ -719,7 +284,7 @@ async def run_manual( ) except Exception as exc: await db_session.rollback() - _raise_manual_failed(exc=exc, user_id=current_user.id) + raise_manual_failed(exc=exc, user_id=current_user.id) @router.get( @@ -740,7 +305,7 @@ async def list_queue_items( return success_payload( request, data={ - "queue_items": [_serialize_queue_item(item) for item in items], + "queue_items": [serialize_queue_item(item) for item in items], }, ) @@ -776,7 +341,7 @@ async def retry_queue_item( ) return success_payload( request, - data=_serialize_queue_item(queue_item), + data=serialize_queue_item(queue_item), ) @@ -811,7 +376,7 @@ async def drop_queue_item( ) return success_payload( request, - data=_serialize_queue_item(dropped), + data=serialize_queue_item(dropped), ) diff --git a/app/api/routers/scholar_helpers.py b/app/api/routers/scholar_helpers.py new file mode 100644 index 0000000..4023a3a --- /dev/null +++ b/app/api/routers/scholar_helpers.py @@ -0,0 +1,234 @@ +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +from fastapi import UploadFile +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.errors import ApiException +from app.logging_utils import structured_log +from app.services.ingestion import queue as ingestion_queue_service +from app.services.scholar import rate_limit as scholar_rate_limit +from app.services.scholar.source import ScholarSource +from app.services.scholars import application as scholar_service +from app.services.scholars import search_hints as scholar_search_hints +from app.settings import settings + +logger = logging.getLogger(__name__) + +CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS = 5.0 +INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS = 0 +INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON = "scholar_added_initial_scrape" + + +def needs_metadata_hydration(profile) -> bool: + if not profile.profile_image_url: + return True + return not (profile.display_name or "").strip() + + +def is_create_hydration_rate_limited() -> tuple[bool, float]: + remaining_seconds = scholar_rate_limit.remaining_scholar_slot_seconds( + min_interval_seconds=float(settings.ingestion_min_request_delay_seconds), + ) + return remaining_seconds > 0, remaining_seconds + + +def auto_enqueue_new_scholar_enabled() -> bool: + if not settings.scheduler_enabled: + return False + if not settings.ingestion_automation_allowed: + return False + return bool(settings.ingestion_continuation_queue_enabled) + + +async def enqueue_initial_scrape_job_for_scholar( + db_session: AsyncSession, + *, + profile, + user_id: int, +) -> bool: + if not auto_enqueue_new_scholar_enabled(): + return False + try: + await ingestion_queue_service.upsert_job( + db_session, + user_id=user_id, + scholar_profile_id=int(profile.id), + resume_cstart=0, + reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, + run_id=None, + delay_seconds=INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS, + ) + await db_session.commit() + except Exception: + await db_session.rollback() + structured_log( + logger, + "warning", + "api.scholars.initial_scrape_enqueue_failed", + user_id=user_id, + scholar_profile_id=profile.id, + ) + return False + + structured_log( + logger, + "info", + "api.scholars.initial_scrape_enqueued", + user_id=user_id, + scholar_profile_id=profile.id, + reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, + ) + return True + + +def uploaded_image_media_path(scholar_profile_id: int) -> str: + return f"/scholar-images/{scholar_profile_id}/upload" + + +def serialize_scholar(profile) -> dict[str, object]: + uploaded_image_url = None + if profile.profile_image_upload_path: + uploaded_image_url = uploaded_image_media_path(int(profile.id)) + + profile_image_url, profile_image_source = scholar_search_hints.resolve_profile_image( + profile, + uploaded_image_url=uploaded_image_url, + ) + + return { + "id": int(profile.id), + "scholar_id": profile.scholar_id, + "display_name": profile.display_name, + "profile_image_url": profile_image_url, + "profile_image_source": profile_image_source, + "is_enabled": bool(profile.is_enabled), + "baseline_completed": bool(profile.baseline_completed), + "last_run_dt": profile.last_run_dt, + "last_run_status": (profile.last_run_status.value if profile.last_run_status is not None else None), + } + + +async def hydrate_scholar_metadata_if_needed( + db_session: AsyncSession, + *, + profile, + source: ScholarSource, + user_id: int, +): + if not needs_metadata_hydration(profile): + return profile + + should_skip, remaining_seconds = is_create_hydration_rate_limited() + if should_skip: + structured_log( + logger, + "info", + "api.scholars.create_metadata_hydration_skipped", + reason="scholar_request_throttle_active", + user_id=user_id, + scholar_profile_id=profile.id, + retry_after_seconds=round(remaining_seconds, 3), + ) + return profile + + try: + return await asyncio.wait_for( + scholar_service.hydrate_profile_metadata( + db_session, + profile=profile, + source=source, + ), + timeout=CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS, + ) + except TimeoutError: + structured_log( + logger, + "info", + "api.scholars.create_metadata_hydration_skipped", + reason="create_timeout", + user_id=user_id, + scholar_profile_id=profile.id, + ) + except Exception: + structured_log( + logger, + "warning", + "api.scholars.create_metadata_hydration_failed", + user_id=user_id, + scholar_profile_id=profile.id, + ) + return profile + + +def search_kwargs() -> dict[str, Any]: + 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_search_hints.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 diff --git a/app/api/routers/scholars.py b/app/api/routers/scholars.py index 87fd04a..7095471 100644 --- a/app/api/routers/scholars.py +++ b/app/api/routers/scholars.py @@ -1,8 +1,6 @@ from __future__ import annotations -import asyncio import logging -from typing import Any from fastapi import APIRouter, Depends, File, Query, Request, UploadFile from sqlalchemy.ext.asyncio import AsyncSession @@ -10,6 +8,15 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.api.deps import get_api_current_user from app.api.errors import ApiException from app.api.responses import success_payload +from app.api.routers.scholar_helpers import ( + enqueue_initial_scrape_job_for_scholar, + hydrate_scholar_metadata_if_needed, + read_uploaded_image, + require_user_profile, + search_kwargs, + search_response_data, + serialize_scholar, +) from app.api.runtime_deps import get_scholar_source from app.api.schemas import ( DataExportEnvelope, @@ -25,231 +32,14 @@ from app.api.schemas import ( from app.db.models import User from app.db.session import get_db_session from app.logging_utils import structured_log -from app.services.ingestion import queue as ingestion_queue_service from app.services.portability import application as import_export_service -from app.services.scholar import rate_limit as scholar_rate_limit from app.services.scholar.source import ScholarSource from app.services.scholars import application as scholar_service -from app.services.scholars import search_hints as scholar_search_hints from app.settings import settings logger = logging.getLogger(__name__) router = APIRouter(prefix="/scholars", tags=["api-scholars"]) -CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS = 5.0 -INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS = 0 -INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON = "scholar_added_initial_scrape" - - -def _needs_metadata_hydration(profile) -> bool: - if not profile.profile_image_url: - return True - return not (profile.display_name or "").strip() - - -def _is_create_hydration_rate_limited() -> tuple[bool, float]: - remaining_seconds = scholar_rate_limit.remaining_scholar_slot_seconds( - min_interval_seconds=float(settings.ingestion_min_request_delay_seconds), - ) - return remaining_seconds > 0, remaining_seconds - - -def _auto_enqueue_new_scholar_enabled() -> bool: - if not settings.scheduler_enabled: - return False - if not settings.ingestion_automation_allowed: - return False - return bool(settings.ingestion_continuation_queue_enabled) - - -async def _enqueue_initial_scrape_job_for_scholar( - db_session: AsyncSession, - *, - profile, - user_id: int, -) -> bool: - if not _auto_enqueue_new_scholar_enabled(): - return False - try: - await ingestion_queue_service.upsert_job( - db_session, - user_id=user_id, - scholar_profile_id=int(profile.id), - resume_cstart=0, - reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, - run_id=None, - delay_seconds=INITIAL_SCHOLAR_SCRAPE_QUEUE_DELAY_SECONDS, - ) - await db_session.commit() - except Exception: - await db_session.rollback() - structured_log( - logger, - "warning", - "api.scholars.initial_scrape_enqueue_failed", - user_id=user_id, - scholar_profile_id=profile.id, - ) - return False - - structured_log( - logger, - "info", - "api.scholars.initial_scrape_enqueued", - user_id=user_id, - scholar_profile_id=profile.id, - reason=INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON, - ) - return True - - -def _uploaded_image_media_path(scholar_profile_id: int) -> str: - return f"/scholar-images/{scholar_profile_id}/upload" - - -def _serialize_scholar(profile) -> dict[str, object]: - uploaded_image_url = None - if profile.profile_image_upload_path: - uploaded_image_url = _uploaded_image_media_path(int(profile.id)) - - profile_image_url, profile_image_source = scholar_search_hints.resolve_profile_image( - profile, - uploaded_image_url=uploaded_image_url, - ) - - return { - "id": int(profile.id), - "scholar_id": profile.scholar_id, - "display_name": profile.display_name, - "profile_image_url": profile_image_url, - "profile_image_source": profile_image_source, - "is_enabled": bool(profile.is_enabled), - "baseline_completed": bool(profile.baseline_completed), - "last_run_dt": profile.last_run_dt, - "last_run_status": (profile.last_run_status.value if profile.last_run_status is not None else None), - } - - -async def _hydrate_scholar_metadata_if_needed( - db_session: AsyncSession, - *, - profile, - source: ScholarSource, - user_id: int, -): - if not _needs_metadata_hydration(profile): - return profile - - should_skip, remaining_seconds = _is_create_hydration_rate_limited() - if should_skip: - structured_log( - logger, - "info", - "api.scholars.create_metadata_hydration_skipped", - reason="scholar_request_throttle_active", - user_id=user_id, - scholar_profile_id=profile.id, - retry_after_seconds=round(remaining_seconds, 3), - ) - return profile - - try: - return await asyncio.wait_for( - scholar_service.hydrate_profile_metadata( - db_session, - profile=profile, - source=source, - ), - timeout=CREATE_METADATA_HYDRATION_TIMEOUT_SECONDS, - ) - except TimeoutError: - structured_log( - logger, - "info", - "api.scholars.create_metadata_hydration_skipped", - reason="create_timeout", - user_id=user_id, - scholar_profile_id=profile.id, - ) - except Exception: - structured_log( - logger, - "warning", - "api.scholars.create_metadata_hydration_failed", - user_id=user_id, - scholar_profile_id=profile.id, - ) - return profile - - -def _search_kwargs() -> dict[str, Any]: - 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_search_hints.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( @@ -268,7 +58,7 @@ async def list_scholars( return success_payload( request, data={ - "scholars": [_serialize_scholar(profile) for profile in scholars], + "scholars": [serialize_scholar(profile) for profile in scholars], }, ) @@ -354,13 +144,13 @@ async def create_scholar( message=str(exc), ) from exc structured_log(logger, "info", "api.scholars.created", user_id=current_user.id, scholar_profile_id=created.id) - did_queue_initial_scrape = await _enqueue_initial_scrape_job_for_scholar( + did_queue_initial_scrape = await enqueue_initial_scrape_job_for_scholar( db_session, profile=created, user_id=current_user.id, ) if not did_queue_initial_scrape: - created = await _hydrate_scholar_metadata_if_needed( + created = await hydrate_scholar_metadata_if_needed( db_session, profile=created, source=source, @@ -369,7 +159,7 @@ async def create_scholar( return success_payload( request, - data=_serialize_scholar(created), + data=serialize_scholar(created), ) @@ -391,7 +181,7 @@ async def search_scholars( db_session=db_session, query=query, limit=limit, - **_search_kwargs(), + **search_kwargs(), ) except scholar_service.ScholarServiceError as exc: raise ApiException( @@ -411,7 +201,7 @@ async def search_scholars( ) return success_payload( request, - data=_search_response_data(query, parsed), + data=search_response_data(query, parsed), ) @@ -447,7 +237,7 @@ async def toggle_scholar( ) return success_payload( request, - data=_serialize_scholar(updated), + data=serialize_scholar(updated), ) @@ -528,7 +318,7 @@ async def update_scholar_image_url( ) return success_payload( request, - data=_serialize_scholar(updated), + data=serialize_scholar(updated), ) @@ -543,13 +333,13 @@ async def upload_scholar_image( db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_api_current_user), ): - profile = await _require_user_profile( + profile = await require_user_profile( db_session, user_id=current_user.id, scholar_profile_id=scholar_profile_id, ) - image_bytes = await _read_uploaded_image(image) + image_bytes = await read_uploaded_image(image) try: updated = await scholar_service.set_profile_image_upload( db_session, @@ -578,7 +368,7 @@ async def upload_scholar_image( ) return success_payload( request, - data=_serialize_scholar(updated), + data=serialize_scholar(updated), ) @@ -612,5 +402,5 @@ async def clear_scholar_image_customization( structured_log(logger, "info", "api.scholars.image_cleared", user_id=current_user.id, scholar_profile_id=updated.id) return success_payload( request, - data=_serialize_scholar(updated), + data=serialize_scholar(updated), ) diff --git a/app/services/ingestion/queue_runner.py b/app/services/ingestion/queue_runner.py new file mode 100644 index 0000000..82baab3 --- /dev/null +++ b/app/services/ingestion/queue_runner.py @@ -0,0 +1,379 @@ +from __future__ import annotations + +import logging +from datetime import UTC, datetime + +from sqlalchemy import select + +from app.db.models import QueueItemStatus, RunTriggerType, ScholarProfile, UserSetting +from app.db.session import get_session_factory +from app.logging_utils import structured_log +from app.services.ingestion import queue as queue_service +from app.services.ingestion.application import ( + RunAlreadyInProgressError, + RunBlockedBySafetyPolicyError, + ScholarIngestionService, +) +from app.services.scholar.source import LiveScholarSource +from app.settings import settings + +logger = logging.getLogger(__name__) + + +def _effective_request_delay_seconds(value: int | None, *, floor: int) -> int: + try: + parsed = int(value) if value is not None else floor + except (TypeError, ValueError): + parsed = floor + return max(floor, parsed) + + +class QueueJobRunner: + def __init__( + self, + *, + tick_seconds: int, + network_error_retries: int, + retry_backoff_seconds: float, + max_pages_per_scholar: int, + page_size: int, + continuation_queue_enabled: bool, + continuation_base_delay_seconds: int, + continuation_max_delay_seconds: int, + continuation_max_attempts: int, + queue_batch_size: int, + ) -> None: + self._tick_seconds = tick_seconds + self._network_error_retries = network_error_retries + self._retry_backoff_seconds = retry_backoff_seconds + self._max_pages_per_scholar = max_pages_per_scholar + self._page_size = page_size + self._continuation_queue_enabled = continuation_queue_enabled + self._continuation_base_delay_seconds = continuation_base_delay_seconds + self._continuation_max_delay_seconds = continuation_max_delay_seconds + self._continuation_max_attempts = continuation_max_attempts + self._queue_batch_size = queue_batch_size + self._source = LiveScholarSource() + + async def drain_continuation_queue(self) -> None: + now = datetime.now(UTC) + session_factory = get_session_factory() + async with session_factory() as session: + jobs = await queue_service.list_due_jobs( + session, + now=now, + limit=self._queue_batch_size, + ) + for job in jobs: + await self._run_queue_job(job) + + async def _drop_queue_job_if_max_attempts( + self, + job: queue_service.ContinuationQueueJob, + ) -> bool: + if job.attempt_count < self._continuation_max_attempts: + return False + session_factory = get_session_factory() + async with session_factory() as session: + dropped = await queue_service.mark_dropped( + session, + job_id=job.id, + reason="max_attempts_reached", + ) + await session.commit() + if dropped is not None: + 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 + + async def _mark_queue_job_retrying( + self, + job: queue_service.ContinuationQueueJob, + ) -> bool: + session_factory = get_session_factory() + async with session_factory() as session: + queue_item = await queue_service.mark_retrying(session, job_id=job.id) + await session.commit() + if queue_item is None: + return False + return queue_item.status != QueueItemStatus.DROPPED.value + + async def _queue_job_has_available_scholar( + self, + job: queue_service.ContinuationQueueJob, + ) -> bool: + session_factory = get_session_factory() + async with session_factory() as session: + scholar_result = await session.execute( + select(ScholarProfile.id).where( + ScholarProfile.user_id == job.user_id, + ScholarProfile.id == job.scholar_profile_id, + ScholarProfile.is_enabled.is_(True), + ) + ) + scholar_id = scholar_result.scalar_one_or_none() + if scholar_id is not None: + return True + dropped = await queue_service.mark_dropped( + session, + job_id=job.id, + reason="scholar_unavailable", + ) + await session.commit() + if dropped is not None: + 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 + + async def _reschedule_queue_job_lock_active(self, job: queue_service.ContinuationQueueJob) -> None: + session_factory = get_session_factory() + async with session_factory() as recovery_session: + await queue_service.reschedule_job( + recovery_session, + job_id=job.id, + delay_seconds=max(self._tick_seconds, 15), + reason="user_run_lock_active", + error="run_already_in_progress", + ) + await recovery_session.commit() + 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( + self, + job: queue_service.ContinuationQueueJob, + exc: RunBlockedBySafetyPolicyError, + ) -> None: + cooldown_remaining_seconds = max( + self._tick_seconds, + int(exc.safety_state.get("cooldown_remaining_seconds") or 0), + ) + session_factory = get_session_factory() + async with session_factory() as recovery_session: + await queue_service.reschedule_job( + recovery_session, + job_id=job.id, + delay_seconds=max(self._tick_seconds, cooldown_remaining_seconds), + reason="scrape_safety_cooldown", + error=str(exc.message), + ) + await recovery_session.commit() + 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( + self, + job: queue_service.ContinuationQueueJob, + *, + exc: Exception, + ) -> None: + session_factory = get_session_factory() + async with session_factory() as recovery_session: + queue_item = await queue_service.increment_attempt_count(recovery_session, job_id=job.id) + if queue_item is None: + await recovery_session.commit() + return + if int(queue_item.attempt_count) >= self._continuation_max_attempts: + await queue_service.mark_dropped( + recovery_session, + job_id=job.id, + reason="scheduler_exception_max_attempts", + error=str(exc), + ) + await recovery_session.commit() + 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( + 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( + recovery_session, + job_id=job.id, + delay_seconds=delay_seconds, + reason="scheduler_exception", + error=str(exc), + ) + await recovery_session.commit() + logger.exception("scheduler.queue_item_run_failed", extra={"queue_item_id": job.id, "user_id": job.user_id}) + + async def _load_request_delay_for_user(self, user_id: int, *, floor: int) -> int: + session_factory = get_session_factory() + async with session_factory() as session: + result = await session.execute( + select(UserSetting.request_delay_seconds).where(UserSetting.user_id == user_id) + ) + delay = result.scalar_one_or_none() + return _effective_request_delay_seconds(delay, floor=floor) + + async def _run_ingestion_for_queue_job( + self, + job: queue_service.ContinuationQueueJob, + *, + request_delay_floor: int, + ): + request_delay_seconds = await self._load_request_delay_for_user(job.user_id, floor=request_delay_floor) + session_factory = get_session_factory() + async with session_factory() as session: + ingestion = ScholarIngestionService(source=self._source) + try: + return await ingestion.run_for_user( + session, + user_id=job.user_id, + trigger_type=RunTriggerType.SCHEDULED, + request_delay_seconds=request_delay_seconds, + network_error_retries=self._network_error_retries, + retry_backoff_seconds=self._retry_backoff_seconds, + rate_limit_retries=settings.ingestion_rate_limit_retries, + rate_limit_backoff_seconds=settings.ingestion_rate_limit_backoff_seconds, + max_pages_per_scholar=self._max_pages_per_scholar, + page_size=self._page_size, + scholar_profile_ids={job.scholar_profile_id}, + start_cstart_by_scholar_id={job.scholar_profile_id: job.resume_cstart}, + auto_queue_continuations=self._continuation_queue_enabled, + queue_delay_seconds=self._continuation_base_delay_seconds, + 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 RunAlreadyInProgressError: + await session.rollback() + await self._reschedule_queue_job_lock_active(job) + except RunBlockedBySafetyPolicyError as exc: + await session.rollback() + await self._reschedule_queue_job_safety_cooldown(job, exc) + except Exception as exc: + await session.rollback() + await self._reschedule_queue_job_after_exception(job, exc=exc) + return None + + 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: + if int(run_summary.failed_count) <= 0: + queue_item = await queue_service.reset_attempt_count(session, job_id=job.id) + await session.commit() + if queue_item is None: + 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 + 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() + 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() + 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() + 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, + ) + + async def _run_queue_job(self, job: queue_service.ContinuationQueueJob) -> None: + if await self._drop_queue_job_if_max_attempts(job): + return + if not await self._mark_queue_job_retrying(job): + return + if not await self._queue_job_has_available_scholar(job): + return + from app.services.settings import application as user_settings_service + + request_delay_floor = user_settings_service.resolve_request_delay_minimum( + settings.ingestion_min_request_delay_seconds, + ) + run_summary = await self._run_ingestion_for_queue_job(job, request_delay_floor=request_delay_floor) + if run_summary is None: + return + await self._finalize_queue_job_after_run(job, run_summary) diff --git a/app/services/ingestion/scheduler.py b/app/services/ingestion/scheduler.py index 12ba3ef..975cb53 100644 --- a/app/services/ingestion/scheduler.py +++ b/app/services/ingestion/scheduler.py @@ -6,24 +6,21 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta from sqlalchemy import select -from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import ( CrawlRun, - QueueItemStatus, RunTriggerType, - ScholarProfile, User, UserSetting, ) from app.db.session import get_session_factory from app.logging_utils import structured_log -from app.services.ingestion import queue as queue_service from app.services.ingestion.application import ( RunAlreadyInProgressError, RunBlockedBySafetyPolicyError, ScholarIngestionService, ) +from app.services.ingestion.queue_runner import QueueJobRunner from app.services.scholar.source import LiveScholarSource from app.services.settings import application as user_settings_service from app.settings import settings @@ -85,6 +82,18 @@ class SchedulerService: self._queue_batch_size = max(1, int(queue_batch_size)) self._task: asyncio.Task[None] | None = None self._source = LiveScholarSource() + self._queue_runner = QueueJobRunner( + 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 start(self) -> None: if not self._enabled: @@ -136,7 +145,7 @@ class SchedulerService: async def _tick_once(self) -> None: if self._continuation_queue_enabled: - await self._drain_continuation_queue() + await self._queue_runner.drain_continuation_queue() await self._drain_pdf_queue() @@ -281,313 +290,6 @@ class SchedulerService: new_publication_count=run_summary.new_publication_count, ) - async def _drain_continuation_queue(self) -> None: - now = datetime.now(UTC) - session_factory = get_session_factory() - async with session_factory() as session: - jobs = await queue_service.list_due_jobs( - session, - now=now, - limit=self._queue_batch_size, - ) - for job in jobs: - await self._run_queue_job(job) - - async def _drop_queue_job_if_max_attempts( - self, - job: queue_service.ContinuationQueueJob, - ) -> bool: - if job.attempt_count < self._continuation_max_attempts: - return False - session_factory = get_session_factory() - async with session_factory() as session: - dropped = await queue_service.mark_dropped( - session, - job_id=job.id, - reason="max_attempts_reached", - ) - await session.commit() - if dropped is not None: - 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 - - async def _mark_queue_job_retrying( - self, - job: queue_service.ContinuationQueueJob, - ) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: - queue_item = await queue_service.mark_retrying(session, job_id=job.id) - await session.commit() - if queue_item is None: - return False - return queue_item.status != QueueItemStatus.DROPPED.value - - async def _queue_job_has_available_scholar( - self, - job: queue_service.ContinuationQueueJob, - ) -> bool: - session_factory = get_session_factory() - async with session_factory() as session: - scholar_result = await session.execute( - select(ScholarProfile.id).where( - ScholarProfile.user_id == job.user_id, - ScholarProfile.id == job.scholar_profile_id, - ScholarProfile.is_enabled.is_(True), - ) - ) - scholar_id = scholar_result.scalar_one_or_none() - if scholar_id is not None: - return True - dropped = await queue_service.mark_dropped( - session, - job_id=job.id, - reason="scholar_unavailable", - ) - await session.commit() - if dropped is not None: - 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 - - async def _reschedule_queue_job_lock_active(self, job: queue_service.ContinuationQueueJob) -> None: - session_factory = get_session_factory() - async with session_factory() as recovery_session: - await queue_service.reschedule_job( - recovery_session, - job_id=job.id, - delay_seconds=max(self._tick_seconds, 15), - reason="user_run_lock_active", - error="run_already_in_progress", - ) - await recovery_session.commit() - 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( - self, - job: queue_service.ContinuationQueueJob, - exc: RunBlockedBySafetyPolicyError, - ) -> None: - cooldown_remaining_seconds = max( - self._tick_seconds, - int(exc.safety_state.get("cooldown_remaining_seconds") or 0), - ) - session_factory = get_session_factory() - async with session_factory() as recovery_session: - await queue_service.reschedule_job( - recovery_session, - job_id=job.id, - delay_seconds=max(self._tick_seconds, cooldown_remaining_seconds), - reason="scrape_safety_cooldown", - error=str(exc.message), - ) - await recovery_session.commit() - 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( - self, - job: queue_service.ContinuationQueueJob, - *, - exc: Exception, - ) -> None: - session_factory = get_session_factory() - async with session_factory() as recovery_session: - queue_item = await queue_service.increment_attempt_count(recovery_session, job_id=job.id) - if queue_item is None: - await recovery_session.commit() - return - if int(queue_item.attempt_count) >= self._continuation_max_attempts: - await queue_service.mark_dropped( - recovery_session, - job_id=job.id, - reason="scheduler_exception_max_attempts", - error=str(exc), - ) - await recovery_session.commit() - 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( - 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( - recovery_session, - job_id=job.id, - delay_seconds=delay_seconds, - reason="scheduler_exception", - error=str(exc), - ) - await recovery_session.commit() - 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, - job: queue_service.ContinuationQueueJob, - ): - session_factory = get_session_factory() - async with session_factory() as session: - request_delay_seconds = await self._load_request_delay_for_user(session, user_id=job.user_id) - ingestion = ScholarIngestionService(source=self._source) - try: - return await ingestion.run_for_user( - session, - user_id=job.user_id, - trigger_type=RunTriggerType.SCHEDULED, - request_delay_seconds=request_delay_seconds, - network_error_retries=self._network_error_retries, - retry_backoff_seconds=self._retry_backoff_seconds, - rate_limit_retries=settings.ingestion_rate_limit_retries, - rate_limit_backoff_seconds=settings.ingestion_rate_limit_backoff_seconds, - max_pages_per_scholar=self._max_pages_per_scholar, - page_size=self._page_size, - scholar_profile_ids={job.scholar_profile_id}, - start_cstart_by_scholar_id={job.scholar_profile_id: job.resume_cstart}, - auto_queue_continuations=self._continuation_queue_enabled, - queue_delay_seconds=self._continuation_base_delay_seconds, - 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 RunAlreadyInProgressError: - await session.rollback() - await self._reschedule_queue_job_lock_active(job) - except RunBlockedBySafetyPolicyError as exc: - await session.rollback() - await self._reschedule_queue_job_safety_cooldown(job, exc) - except Exception as exc: - await session.rollback() - await self._reschedule_queue_job_after_exception(job, exc=exc) - return None - - 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: - if int(run_summary.failed_count) <= 0: - queue_item = await queue_service.reset_attempt_count(session, job_id=job.id) - await session.commit() - if queue_item is None: - 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 - 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() - 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() - 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() - 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, - ) - - async def _run_queue_job(self, job: queue_service.ContinuationQueueJob) -> None: - if await self._drop_queue_job_if_max_attempts(job): - return - if not await self._mark_queue_job_retrying(job): - return - if not await self._queue_job_has_available_scholar(job): - return - run_summary = await self._run_ingestion_for_queue_job(job) - if run_summary is None: - return - await self._finalize_queue_job_after_run(job, run_summary) - async def _drain_pdf_queue(self) -> None: from app.services.publications.pdf_queue import drain_ready_jobs @@ -608,15 +310,3 @@ class SchedulerService: ) except Exception: logger.exception("scheduler.pdf_queue_drain_failed", extra={}) - - async def _load_request_delay_for_user( - self, - db_session: AsyncSession, - *, - user_id: int, - ) -> int: - result = await db_session.execute( - select(UserSetting.request_delay_seconds).where(UserSetting.user_id == user_id) - ) - delay = result.scalar_one_or_none() - return _effective_request_delay_seconds(delay) diff --git a/tests/unit/test_scholars_create_hydration.py b/tests/unit/test_scholars_create_hydration.py index b452b44..0694b32 100644 --- a/tests/unit/test_scholars_create_hydration.py +++ b/tests/unit/test_scholars_create_hydration.py @@ -2,7 +2,7 @@ from types import SimpleNamespace import pytest -from app.api.routers import scholars as scholars_router +from app.api.routers import scholar_helpers class _FakeSession: @@ -31,17 +31,17 @@ async def test_create_hydration_skips_when_global_throttle_active( raise AssertionError("hydrate_profile_metadata should not run when throttle is active") monkeypatch.setattr( - scholars_router.scholar_rate_limit, + scholar_helpers.scholar_rate_limit, "remaining_scholar_slot_seconds", lambda **_kwargs: 3.5, ) monkeypatch.setattr( - scholars_router.scholar_service, + scholar_helpers.scholar_service, "hydrate_profile_metadata", _fail_hydration, ) - result = await scholars_router._hydrate_scholar_metadata_if_needed( + result = await scholar_helpers.hydrate_scholar_metadata_if_needed( db_session=None, profile=profile, source=object(), @@ -66,17 +66,17 @@ async def test_create_hydration_runs_when_throttle_is_clear( return profile monkeypatch.setattr( - scholars_router.scholar_rate_limit, + scholar_helpers.scholar_rate_limit, "remaining_scholar_slot_seconds", lambda **_kwargs: 0.0, ) monkeypatch.setattr( - scholars_router.scholar_service, + scholar_helpers.scholar_service, "hydrate_profile_metadata", _hydrate_profile_metadata, ) - result = await scholars_router._hydrate_scholar_metadata_if_needed( + result = await scholar_helpers.hydrate_scholar_metadata_if_needed( db_session=None, profile=profile, source=object(), @@ -98,17 +98,17 @@ async def test_initial_scrape_job_enqueued_on_create( upsert_calls.append(kwargs) monkeypatch.setattr( - scholars_router, - "_auto_enqueue_new_scholar_enabled", + scholar_helpers, + "auto_enqueue_new_scholar_enabled", lambda: True, ) monkeypatch.setattr( - scholars_router.ingestion_queue_service, + scholar_helpers.ingestion_queue_service, "upsert_job", _fake_upsert_job, ) - queued = await scholars_router._enqueue_initial_scrape_job_for_scholar( + queued = await scholar_helpers.enqueue_initial_scrape_job_for_scholar( session, profile=profile, user_id=9, @@ -117,7 +117,7 @@ async def test_initial_scrape_job_enqueued_on_create( assert queued is True assert session.commits == 1 assert session.rollbacks == 0 - assert upsert_calls[0]["reason"] == scholars_router.INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON + assert upsert_calls[0]["reason"] == scholar_helpers.INITIAL_SCHOLAR_SCRAPE_QUEUE_REASON @pytest.mark.asyncio @@ -128,12 +128,12 @@ async def test_initial_scrape_job_not_enqueued_when_disabled( session = _FakeSession() monkeypatch.setattr( - scholars_router, - "_auto_enqueue_new_scholar_enabled", + scholar_helpers, + "auto_enqueue_new_scholar_enabled", lambda: False, ) - queued = await scholars_router._enqueue_initial_scrape_job_for_scholar( + queued = await scholar_helpers.enqueue_initial_scrape_job_for_scholar( session, profile=profile, user_id=11,