diff --git a/DECOMPOSITION_SLICES.md b/DECOMPOSITION_SLICES.md new file mode 100644 index 0000000..8e22261 --- /dev/null +++ b/DECOMPOSITION_SLICES.md @@ -0,0 +1,212 @@ +# Scholarr Decomposition Slices + +## How to use this file + +This file contains self-contained decomposition prompts ("slices") to bring the codebase into compliance with `agents.md` file size rules. Each slice is an independent task for an LLM executor. + +**Rules from `agents.md`:** +- File length: 400 lines target, 600 lines hard ceiling +- Function length: 50 lines max +- All business logic in `app/services//` +- Read `agents.md` fully before starting any slice + +**Execution protocol:** +1. Pick the next incomplete slice in order (slices depend on prior ones) +2. Read the prompt section — it contains everything you need +3. Read all source files mentioned before making changes +4. Make the changes described — no business logic changes, only file reorganization +5. Verify: all imports resolve, all tests pass +6. Commit with the specified message + +--- + +## Progress + +| # | Slice | Status | +|---|-------|--------| +| 1 | Schema split | **DONE** | +| 2 | Ingestion: pagination + publication upsert | **DONE** | +| 3 | Ingestion: enrichment + scholar processing + run completion | pending | +| 4 | Scholars service + PDF queue | pending | +| 5 | Routers + scheduler | pending | + +--- + +## Slice 1: Split `app/api/schemas.py` into domain packages — DONE + +Completed. The 945-line `app/api/schemas.py` was split into `app/api/schemas/` package: + +| File | Lines | Contents | +|---|---|---| +| `common.py` | 39 | `ApiMeta`, `ApiErrorData`, `ApiErrorEnvelope`, `MessageData`, `MessageEnvelope` | +| `auth.py` | 73 | `SessionUserData`, `AuthMe*`, `CsrfBootstrap*`, `Login*`, `ChangePasswordRequest` | +| `scholars.py` | 153 | `ScholarItem*`, `ScholarSearch*`, `ScholarExport*`, `DataExport*`, `DataImport*` | +| `publications.py` | 151 | `DisplayIdentifierData`, `PublicationItem*`, `MarkAll*`, `MarkSelected*`, `RetryPublication*`, `TogglePublication*` | +| `runs.py` | 226 | `RunListItem*`, `RunSummary*`, `RunDebug*`, `RunScholarResult*`, `RunDetail*`, `ManualRun*`, `Queue*`, `ScrapeSafety*` | +| `admin.py` | 289 | All `Admin*` models including PDF queue, repair, near-duplicate schemas | +| `settings.py` | 54 | `SettingsPolicy*`, `Settings*`, `SettingsUpdateRequest` | +| `__init__.py` | 7 | Wildcard re-exports — all `from app.api.schemas import X` imports unchanged | + +--- + +## Slice 2: Decompose `app/services/ingestion/application.py` — pagination + publication upsert — DONE + +Completed. Extracted three modules from `application.py` (2,955 → 2,084 lines): + +| File | Lines | Contents | +|---|---|---| +| `app/services/ingestion/page_fetch.py` | 219 | `PageFetcher` class — single-page fetch, parse-or-layout-error, retry with backoff | +| `app/services/ingestion/pagination.py` | 486 | `PaginationEngine` class — multi-page orchestration, loop state, short-circuit, fingerprint checks | +| `app/services/ingestion/publication_upsert.py` | 239 | Module-level functions — `resolve_publication`, `upsert_profile_publications`, find/create/update helpers | + +Also fixed two external import paths (`portability/normalize.py`, `portability/publication_import.py`) that imported `normalize_title`/`build_publication_url` from `application.py` — now correctly sourced from `fingerprints.py`. Updated one integration test to monkeypatch the new module-level function instead of the old instance method. + +--- + +## Slice 3: Decompose `app/services/ingestion/application.py` — enrichment, scholar processing, run completion + +**Why:** Three remaining logical chunks. Extracting all three brings `application.py` down to ~500 lines (the orchestration spine). + +**Files to create:** `app/services/ingestion/enrichment.py`, `app/services/ingestion/scholar_processing.py`, `app/services/ingestion/run_completion.py` + +**Files to modify:** `app/services/ingestion/application.py` + +**Prompt:** + +You are working on the scholarr repository. Read `agents.md` for project conventions. Continue decomposing `app/services/ingestion/application.py` (slice 2 is complete — `page_fetch.py`, `pagination.py`, and `publication_upsert.py` already extracted). Extract three remaining chunks. + +**A. `app/services/ingestion/enrichment.py` (~300 lines)** + +Move these methods (post-run OpenAlex enrichment): + +- `_run_is_canceled`, `_enrich_pending_publications`, `_discover_identifiers_for_enrichment` +- `_publish_identifier_update_event`, `_enrich_publications_with_openalex` + +Create an `EnrichmentRunner` class that receives service dependencies (OpenAlex client, identifier service, dedup service, DB session factory) in its constructor. + +**B. `app/services/ingestion/scholar_processing.py` (~400 lines)** + +Move these methods (per-scholar outcome resolution): + +- `_assert_valid_paged_parse_result`, `_apply_first_page_profile_metadata`, `_build_result_entry` +- `_skipped_no_change_outcome`, `_upsert_publications_outcome`, `_upsert_success_or_exception` +- `_upsert_success`, `_upsert_exception_outcome`, `_parse_failure_outcome` +- `_sync_continuation_queue`, `_process_scholar`, `_process_scholar_inner` +- `_fetch_and_prepare_scholar_result`, `_resolve_scholar_outcome`, `_unexpected_scholar_exception_outcome` + +These become methods on a `ScholarProcessor` class or module-level functions. + +**C. `app/services/ingestion/run_completion.py` (~300 lines)** + +Move these methods (run finalization + alerting): + +- `_classify_failure_bucket` (standalone function) +- `_summarize_failures`, `_build_alert_summary`, `_apply_safety_outcome`, `_finalize_run_record` +- `_resolve_run_status`, `_resolve_continuation_queue_target`, `_build_failure_debug_context` +- `_complete_run_for_user` + +These become module-level functions accepting explicit parameters. + +**After all extractions, `application.py` should contain only:** + +- `ScholarIngestionService.__init__` and config helpers +- Run lifecycle: `initialize_run`, `execute_run`, `run_for_user` +- Scholar iteration orchestration: `_run_scholar_iteration` +- Progress tracking: `_result_counters`, `_adjust_progress_counts` +- Safety gate integration, background task management + +Target: `application.py` ≤ 500 lines. No new file exceeds 400 lines. All tests must pass unchanged. + +Commit message: `refactor: extract enrichment, scholar processing, and run completion from ingestion service` + +--- + +## Slice 4: Decompose `app/services/scholars/application.py` (996 lines) and `app/services/publications/pdf_queue.py` (969 lines) + +**Why:** Two service files nearly 2x the hard ceiling. Each has clear internal seams. + +**Files to create:** `app/services/scholars/author_search.py`, `app/services/scholars/author_search_cache.py`, `app/services/publications/pdf_queue_queries.py`, `app/services/publications/pdf_queue_resolution.py` + +**Files to modify:** `app/services/scholars/application.py`, `app/services/publications/pdf_queue.py` + +**Prompt:** + +You are working on the scholarr repository. Read `agents.md` for project conventions. Two service files exceed the 600-line ceiling. Decompose both. + +**A. `app/services/scholars/application.py` (996 lines)** + +This file mixes scholar CRUD (~350 lines) with the entire author search pipeline (~650 lines). Split them: + +Create `app/services/scholars/author_search_cache.py` (~150 lines): Move all `_serialize_*` and `_deserialize_*` functions. Move `_cache_get_author_search_result`, `_cache_set_author_search_result`, `_prune_author_search_cache`. These are pure functions operating on the DB cache table. + +Create `app/services/scholars/author_search.py` (~500 lines): Move remaining author search functions — cooldown management, throttle logic, retry wrappers, circuit breaker, lock acquisition, and `search_author_candidates`. Import cache functions from `author_search_cache.py`. Accept DB session and settings as explicit parameters. + +Update `application.py`: Keep only scholar CRUD methods. Import and delegate to `search_author_candidates` from `author_search.py`. Target: ≤ 350 lines. + +**B. `app/services/publications/pdf_queue.py` (969 lines)** + +Three concerns in one file. Split by concern: + +Create `app/services/publications/pdf_queue_queries.py` (~330 lines): Move SQL query builders (`_tracked_queue_select_base`, `_tracked_queue_select`, etc.), result hydration (`_queue_item_from_row`, `_hydrated_queue_items`), listing/counting/pagination (`list_pdf_queue_items`, `count_pdf_queue_items`, `list_pdf_queue_page`), retry item builders and missing-PDF candidate queries. + +Create `app/services/publications/pdf_queue_resolution.py` (~300 lines): Move task execution (`_mark_attempt_started`, `_failed_outcome`, `_fetch_outcome_for_row`), outcome persistence (`_apply_publication_update`, `_apply_job_outcome`, `_persist_outcome`), scheduling (`_resolve_publication_row`, `_run_resolution_task`, `_register_task`, `_drop_finished_task`, `_schedule_rows`). + +Keep in `pdf_queue.py` (~340 lines): dataclasses, constants, enqueueing logic, public API. + +No file should exceed 500 lines. All tests must pass unchanged. + +Commit message: `refactor: decompose scholars service and pdf_queue into focused modules` + +--- + +## Slice 5: Decompose router files and scheduler + +**Why:** Three files slightly over the 600-line ceiling. Each needs helper extraction. + +**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` + +**Files to modify:** `app/api/routers/runs.py` (875 → ~405), `app/api/routers/scholars.py` (616 → ~396), `app/services/ingestion/scheduler.py` (622 → ~280) + +**Prompt:** + +You are working on the scholarr repository. Read `agents.md` for project conventions. Three files are slightly over the 600-line ceiling. Extract helpers from each. + +**A. `app/api/routers/runs.py` (875 lines) → 3 files** + +Create `app/api/routers/run_serializers.py` (~250 lines): Move all `_serialize_*`, `_normalize_*`, `_int_value`, `_str_value`, `_bool_value` helper functions. Move `_manual_run_payload_from_run` and `_manual_run_success_payload`. + +Create `app/api/routers/run_manual.py` (~220 lines): Move `_load_safety_state`, `_raise_manual_runs_disabled`, `_reused_manual_run_payload`, `_run_ingestion_for_manual`, `_recover_integrity_error`, `_execute_manual_run`, `_raise_manual_blocked_safety`, `_raise_manual_failed`. + +Keep in `runs.py` (~405 lines): only route handler functions + imports. + +**B. `app/api/routers/scholars.py` (616 lines) → 2 files** + +Create `app/api/routers/scholar_helpers.py` (~220 lines): Move `_needs_metadata_hydration`, `_is_create_hydration_rate_limited`, `_auto_enqueue_new_scholar_enabled`, `_enqueue_initial_scrape_job_for_scholar`, `_uploaded_image_media_path`, `_serialize_scholar`, `_hydrate_scholar_metadata_if_needed`, `_search_kwargs`, `_search_response_data`, `_read_uploaded_image`, `_require_user_profile`. + +Keep in `scholars.py` (~396 lines): only route handlers. + +**C. `app/services/ingestion/scheduler.py` (622 lines) → 2 files** + +Create `app/services/ingestion/queue_runner.py` (~350 lines): Move all continuation-queue job processing — `_drain_continuation_queue`, `_drop_queue_job_if_max_attempts`, `_mark_queue_job_retrying`, `_queue_job_has_available_scholar`, all `_reschedule_*` methods, `_run_ingestion_for_queue_job`, `_finalize_queue_job_after_run`, `_run_queue_job`. Create a `QueueJobRunner` class receiving config params in constructor. + +Keep in `scheduler.py` (~280 lines): auto-run scheduling, `_run_loop`, `_tick_once`, candidate loading. + +All tests must pass unchanged after each extraction. + +Commit message: `refactor: extract helpers from oversized router and scheduler files` + +--- + +## Verification (after all slices) + +```bash +# No Python file exceeds 600 lines +find app/ -name "*.py" -exec wc -l {} + | awk '$1 > 600 && !/total/ {print "FAIL:", $0}' + +# All tests pass +docker compose -f docker-compose.yml -f docker-compose.dev.yml run --rm app python -m pytest + +# Linting +docker compose -f docker-compose.yml -f docker-compose.dev.yml run --rm app ruff check . +docker compose -f docker-compose.yml -f docker-compose.dev.yml run --rm app ruff format --check . +``` diff --git a/app/services/ingestion/application.py b/app/services/ingestion/application.py index 9aa49b9..abe23a3 100644 --- a/app/services/ingestion/application.py +++ b/app/services/ingestion/application.py @@ -21,7 +21,6 @@ from app.db.models import ( ) from app.logging_utils import structured_log from app.services.arxiv.errors import ArxivRateLimitError -from app.services.doi.normalize import first_doi_from_texts from app.services.ingestion import queue as queue_service from app.services.ingestion import safety as run_safety_service from app.services.ingestion.constants import ( @@ -37,16 +36,10 @@ from app.services.ingestion.constants import ( ) from app.services.ingestion.fingerprints import ( _build_body_excerpt, - _dedupe_publication_candidates, - _next_cstart_value, - build_initial_page_fingerprint, - build_publication_fingerprint, - build_publication_url, - canonical_title_for_dedup, - normalize_title, ) +from app.services.ingestion.pagination import PaginationEngine +from app.services.ingestion.publication_upsert import upsert_profile_publications from app.services.ingestion.types import ( - PagedLoopState, PagedParseResult, RunAlertSummary, RunAlreadyInProgressError, @@ -62,8 +55,6 @@ from app.services.scholar.parser import ( ParsedProfilePage, ParseState, PublicationCandidate, - ScholarParserError, - parse_profile_page, ) from app.services.scholar.source import FetchResult, ScholarSource from app.services.settings import application as user_settings_service @@ -115,6 +106,7 @@ _background_tasks: set[asyncio.Task[Any]] = set() class ScholarIngestionService: def __init__(self, *, source: ScholarSource) -> None: self._source = source + self._pagination = PaginationEngine(source=source) @staticmethod def _effective_request_delay_seconds(value: int) -> int: @@ -290,7 +282,9 @@ class ScholarIngestionService: paged_parse_result: PagedParseResult, ) -> None: parsed_page = paged_parse_result.parsed_page - if parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS} and any(code.startswith("layout_") for code in parsed_page.warnings): + if parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS} and any( + code.startswith("layout_") for code in parsed_page.warnings + ): raise RuntimeError(f"Layout warning marked as terminal for scholar_id={scholar_id}.") for publication in paged_parse_result.publications: if not publication.title.strip(): @@ -445,7 +439,7 @@ class ScholarIngestionService: has_partial_publication_set: bool, ) -> ScholarProcessingOutcome: # We no longer aggregate and upsert the publications here. - # Eager DB insertions have already saved and committed them inside `_fetch_and_parse_all_pages_with_retry` and `_paginate_loop`. + # Eager DB insertions have already saved and committed them inside `fetch_and_parse_all_pages` and `_paginate_loop`. discovered_count = paged_parse_result.discovered_publication_count is_partial = ( paged_parse_result.has_more_remaining @@ -683,7 +677,7 @@ class ScholarIngestionService: page_size: int, ) -> tuple[datetime, PagedParseResult, dict[str, Any]]: run_dt = datetime.now(UTC) - paged_parse_result = await self._fetch_and_parse_all_pages_with_retry( + paged_parse_result = await self._pagination.fetch_and_parse_all_pages( scholar=scholar, run=run, db_session=db_session, @@ -696,6 +690,7 @@ class ScholarIngestionService: max_pages=max_pages_per_scholar, page_size=page_size, previous_initial_page_fingerprint_sha256=scholar.last_initial_page_fingerprint_sha256, + upsert_publications_fn=upsert_profile_publications, ) 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) @@ -1701,650 +1696,6 @@ class ScholarIngestionService: ) return self._run_execution_summary(run=run, scholars=scholars, progress=progress) - async def _fetch_profile_page( - self, - *, - scholar_id: str, - cstart: int, - page_size: int, - ) -> FetchResult: - try: - page_fetcher = getattr(self._source, "fetch_profile_page_html", None) - if callable(page_fetcher): - return await page_fetcher( - scholar_id, - cstart=cstart, - pagesize=page_size, - ) - if cstart <= 0: - return await self._source.fetch_profile_html(scholar_id) - return FetchResult( - requested_url=( - f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}" - ), - status_code=None, - final_url=None, - body="", - error="source_does_not_support_pagination", - ) - except Exception as exc: - logger.exception( - "ingestion.fetch_unexpected_error", - extra={ - "scholar_id": scholar_id, - "cstart": cstart, - "page_size": page_size, - }, - ) - return FetchResult( - requested_url=( - f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}" - ), - status_code=None, - final_url=None, - body="", - error=str(exc), - ) - - @staticmethod - def _should_retry_page_fetch( - *, - parsed_page: ParsedProfilePage, - network_attempt_count: int, - rate_limit_attempt_count: int, - network_error_retries: int, - rate_limit_retries: int, - ) -> bool: - if parsed_page.state == ParseState.NETWORK_ERROR: - return network_attempt_count <= network_error_retries - if ( - parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA - and parsed_page.state_reason == "blocked_http_429_rate_limited" - ): - return rate_limit_attempt_count <= rate_limit_retries - return False - - @staticmethod - async def _sleep_retry_backoff( - *, - scholar_id: str, - cstart: int, - network_attempt_count: int, - rate_limit_attempt_count: int, - retry_backoff_seconds: float, - rate_limit_backoff_seconds: float, - state_reason: str, - ) -> None: - if state_reason == "blocked_http_429_rate_limited": - sleep_seconds = rate_limit_backoff_seconds * rate_limit_attempt_count - attempt_label = rate_limit_attempt_count - else: - sleep_seconds = retry_backoff_seconds * (2 ** (network_attempt_count - 1)) - attempt_label = network_attempt_count - - structured_log( - logger, - "warning", - "ingestion.scholar_retry_scheduled", - scholar_id=scholar_id, - cstart=cstart, - attempt_count=attempt_label, - sleep_seconds=sleep_seconds, - state_reason=state_reason, - ) - if sleep_seconds > 0: - await asyncio.sleep(sleep_seconds) - - async def _fetch_and_parse_page_with_retry( - self, - *, - scholar_id: str, - cstart: int, - page_size: int, - network_error_retries: int, - retry_backoff_seconds: float, - rate_limit_retries: int, - rate_limit_backoff_seconds: float, - ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]: - network_attempts = 0 - rate_limit_attempts = 0 - attempt_log: list[dict[str, Any]] = [] - fetch_result: FetchResult | None = None - parsed_page: ParsedProfilePage | None = None - - while True: - fetch_result = await self._fetch_profile_page( - scholar_id=scholar_id, - cstart=cstart, - page_size=page_size, - ) - parsed_page = self._parse_profile_page_or_layout_error(fetch_result=fetch_result) - - if parsed_page.state == ParseState.NETWORK_ERROR: - network_attempts += 1 - total_attempts = network_attempts - elif ( - parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA - and parsed_page.state_reason == "blocked_http_429_rate_limited" - ): - rate_limit_attempts += 1 - total_attempts = rate_limit_attempts - else: - total_attempts = network_attempts + rate_limit_attempts + 1 - - attempt_log.append( - { - "attempt": total_attempts, - "cstart": cstart, - "state": parsed_page.state.value, - "state_reason": parsed_page.state_reason, - "status_code": fetch_result.status_code, - "fetch_error": fetch_result.error, - } - ) - - if not self._should_retry_page_fetch( - parsed_page=parsed_page, - network_attempt_count=network_attempts, - rate_limit_attempt_count=rate_limit_attempts, - network_error_retries=network_error_retries, - rate_limit_retries=rate_limit_retries, - ): - break - - await self._sleep_retry_backoff( - scholar_id=scholar_id, - cstart=cstart, - network_attempt_count=network_attempts, - rate_limit_attempt_count=rate_limit_attempts, - retry_backoff_seconds=max(float(retry_backoff_seconds), 0.0), - rate_limit_backoff_seconds=max(float(rate_limit_backoff_seconds), 0.0), - state_reason=parsed_page.state_reason, - ) - - if fetch_result is None or parsed_page is None: - raise RuntimeError("Fetch-and-parse retry loop produced no result.") - return fetch_result, parsed_page, attempt_log - - def _parse_profile_page_or_layout_error( - self, - *, - fetch_result: FetchResult, - ) -> ParsedProfilePage: - try: - return parse_profile_page(fetch_result) - except ScholarParserError as exc: - return self._parsed_page_from_parser_error( - fetch_result=fetch_result, - code=exc.code, - ) - - @staticmethod - def _parsed_page_from_parser_error( - *, - fetch_result: FetchResult, - code: str, - ) -> ParsedProfilePage: - return ParsedProfilePage( - state=ParseState.LAYOUT_CHANGED, - state_reason=code, - profile_name=None, - profile_image_url=None, - publications=[], - marker_counts={}, - warnings=[code], - has_show_more_button=False, - has_operation_error_banner=False, - articles_range=None, - ) - - @staticmethod - def _should_skip_no_change( - *, - start_cstart: int, - first_page_fingerprint_sha256: str | None, - previous_initial_page_fingerprint_sha256: str | None, - parsed_page: ParsedProfilePage, - ) -> bool: - return ( - start_cstart <= 0 - and first_page_fingerprint_sha256 is not None - and previous_initial_page_fingerprint_sha256 is not None - and first_page_fingerprint_sha256 == previous_initial_page_fingerprint_sha256 - and parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS} - ) - - @staticmethod - def _skip_no_change_result( - *, - fetch_result: FetchResult, - parsed_page: ParsedProfilePage, - first_page_fingerprint_sha256: str | None, - attempt_log: list[dict[str, Any]], - page_logs: list[dict[str, Any]], - ) -> PagedParseResult: - return PagedParseResult( - fetch_result=fetch_result, - parsed_page=parsed_page, - first_page_fetch_result=fetch_result, - first_page_parsed_page=parsed_page, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - publications=[], - attempt_log=attempt_log, - page_logs=page_logs, - pages_fetched=1, - pages_attempted=1, - has_more_remaining=False, - pagination_truncated_reason=None, - continuation_cstart=None, - skipped_no_change=True, - discovered_publication_count=0, - ) - - @staticmethod - def _initial_failure_result( - *, - fetch_result: FetchResult, - parsed_page: ParsedProfilePage, - first_page_fingerprint_sha256: str | None, - start_cstart: int, - attempt_log: list[dict[str, Any]], - page_logs: list[dict[str, Any]], - ) -> PagedParseResult: - continuation_cstart = start_cstart if parsed_page.state == ParseState.NETWORK_ERROR else None - return PagedParseResult( - fetch_result=fetch_result, - parsed_page=parsed_page, - first_page_fetch_result=fetch_result, - first_page_parsed_page=parsed_page, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - publications=[], - attempt_log=attempt_log, - page_logs=page_logs, - pages_fetched=0, - pages_attempted=1, - has_more_remaining=False, - pagination_truncated_reason=None, - continuation_cstart=continuation_cstart, - skipped_no_change=False, - discovered_publication_count=0, - ) - - @staticmethod - def _build_loop_state( - *, - start_cstart: int, - fetch_result: FetchResult, - parsed_page: ParsedProfilePage, - attempt_log: list[dict[str, Any]], - page_logs: list[dict[str, Any]], - ) -> PagedLoopState: - next_cstart = _next_cstart_value( - articles_range=parsed_page.articles_range, - fallback=start_cstart + len(parsed_page.publications), - ) - return PagedLoopState( - fetch_result=fetch_result, - parsed_page=parsed_page, - attempt_log=attempt_log, - page_logs=page_logs, - publications=list(parsed_page.publications), - pages_fetched=1, - pages_attempted=1, - current_cstart=start_cstart, - next_cstart=next_cstart, - ) - - @staticmethod - def _set_truncated_state( - *, - state: PagedLoopState, - reason: str, - continuation_cstart: int, - ) -> None: - state.has_more_remaining = True - state.pagination_truncated_reason = reason - state.continuation_cstart = continuation_cstart - - def _should_stop_pagination(self, *, state: PagedLoopState, bounded_max_pages: int) -> bool: - if state.pages_fetched >= bounded_max_pages: - self._set_truncated_state( - state=state, - reason="max_pages_reached", - continuation_cstart=( - state.next_cstart if state.next_cstart > state.current_cstart else state.current_cstart - ), - ) - return True - if state.next_cstart <= state.current_cstart: - self._set_truncated_state( - state=state, - reason="pagination_cursor_stalled", - continuation_cstart=state.current_cstart, - ) - return True - return False - - async def _fetch_next_page( - self, - *, - scholar_id: str, - state: PagedLoopState, - request_delay_seconds: int, - bounded_page_size: int, - network_error_retries: int, - retry_backoff_seconds: float, - rate_limit_retries: int, - rate_limit_backoff_seconds: float, - ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]: - if request_delay_seconds > 0: - await asyncio.sleep(float(request_delay_seconds)) - state.current_cstart = state.next_cstart - return await self._fetch_and_parse_page_with_retry( - scholar_id=scholar_id, - cstart=state.current_cstart, - page_size=bounded_page_size, - network_error_retries=network_error_retries, - retry_backoff_seconds=retry_backoff_seconds, - rate_limit_retries=rate_limit_retries, - rate_limit_backoff_seconds=rate_limit_backoff_seconds, - ) - - @staticmethod - def _record_next_page( - *, - state: PagedLoopState, - fetch_result: FetchResult, - parsed_page: ParsedProfilePage, - page_attempt_log: list[dict[str, Any]], - ) -> None: - state.pages_attempted += 1 - state.attempt_log.extend(page_attempt_log) - state.page_logs.append( - { - "page": state.pages_attempted, - "cstart": state.current_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": len(page_attempt_log), - } - ) - state.fetch_result = fetch_result - state.parsed_page = parsed_page - - @staticmethod - def _handle_page_state_transition(*, state: PagedLoopState) -> bool: - if state.parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}: - ScholarIngestionService._set_truncated_state( - state=state, - reason=f"page_state_{state.parsed_page.state.value}", - continuation_cstart=state.current_cstart, - ) - return True - if state.parsed_page.state == ParseState.NO_RESULTS and len(state.parsed_page.publications) == 0: - state.pages_fetched += 1 - return True - state.pages_fetched += 1 - state.publications.extend(state.parsed_page.publications) - state.next_cstart = _next_cstart_value( - articles_range=state.parsed_page.articles_range, - fallback=state.current_cstart + len(state.parsed_page.publications), - ) - return False - - async def _fetch_initial_page_context( - self, - *, - scholar_id: str, - start_cstart: int, - bounded_page_size: int, - network_error_retries: int, - retry_backoff_seconds: float, - rate_limit_retries: int, - rate_limit_backoff_seconds: float, - ) -> tuple[FetchResult, ParsedProfilePage, str | None, list[dict[str, Any]], list[dict[str, Any]]]: - fetch_result, parsed_page, first_attempt_log = await self._fetch_and_parse_page_with_retry( - scholar_id=scholar_id, - cstart=start_cstart, - page_size=bounded_page_size, - network_error_retries=network_error_retries, - retry_backoff_seconds=retry_backoff_seconds, - rate_limit_retries=rate_limit_retries, - rate_limit_backoff_seconds=rate_limit_backoff_seconds, - ) - first_page_fingerprint_sha256 = build_initial_page_fingerprint(parsed_page) - attempt_log = list(first_attempt_log) - page_logs = [ - { - "page": 1, - "cstart": start_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": len(first_attempt_log), - } - ] - return fetch_result, parsed_page, first_page_fingerprint_sha256, attempt_log, page_logs - - async def _paginate_loop( - self, - *, - scholar: ScholarProfile, - run: CrawlRun, - db_session: AsyncSession, - state: PagedLoopState, - bounded_max_pages: int, - request_delay_seconds: int, - bounded_page_size: int, - network_error_retries: int, - retry_backoff_seconds: float, - rate_limit_retries: int, - rate_limit_backoff_seconds: float, - ) -> None: - # Cross-page canonical dedup state; grows across all pages of this scholar. - seen_canonical: set[str] = set() - - if state.parsed_page.publications: - deduped_first = _dedupe_publication_candidates( - list(state.parsed_page.publications), seen_canonical=seen_canonical - ) - if deduped_first: - discovered_count = await self._upsert_profile_publications( - db_session, run=run, scholar=scholar, publications=deduped_first - ) - state.discovered_publication_count += discovered_count - - while state.parsed_page.has_show_more_button: - await db_session.refresh(run) - if run.status == RunStatus.CANCELED: - structured_log( - logger, - "info", - "ingestion.pagination_canceled", - run_id=run.id, - ) - self._set_truncated_state( - state=state, - reason="run_canceled", - continuation_cstart=state.current_cstart, - ) - return - - if self._should_stop_pagination(state=state, bounded_max_pages=bounded_max_pages): - return - next_fetch_result, next_parsed_page, next_attempt_log = await self._fetch_next_page( - scholar_id=scholar.scholar_id, - state=state, - request_delay_seconds=request_delay_seconds, - bounded_page_size=bounded_page_size, - network_error_retries=network_error_retries, - retry_backoff_seconds=retry_backoff_seconds, - rate_limit_retries=rate_limit_retries, - rate_limit_backoff_seconds=rate_limit_backoff_seconds, - ) - self._record_next_page( - state=state, - fetch_result=next_fetch_result, - parsed_page=next_parsed_page, - page_attempt_log=next_attempt_log, - ) - - # Deduplicate against all publications already committed this run, - # then immediately commit the surviving candidates to the DB. - if next_parsed_page.publications: - deduped_next = _dedupe_publication_candidates( - list(next_parsed_page.publications), seen_canonical=seen_canonical - ) - if deduped_next: - discovered_count = await self._upsert_profile_publications( - db_session, run=run, scholar=scholar, publications=deduped_next - ) - state.discovered_publication_count += discovered_count - - if self._handle_page_state_transition(state=state): - return - - @staticmethod - def _result_from_pagination_state( - *, - state: PagedLoopState, - first_page_fetch_result: FetchResult, - first_page_parsed_page: ParsedProfilePage, - first_page_fingerprint_sha256: str | None, - ) -> PagedParseResult: - return PagedParseResult( - fetch_result=state.fetch_result, - parsed_page=state.parsed_page, - first_page_fetch_result=first_page_fetch_result, - first_page_parsed_page=first_page_parsed_page, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - publications=_dedupe_publication_candidates(state.publications), - attempt_log=state.attempt_log, - page_logs=state.page_logs, - pages_fetched=state.pages_fetched, - pages_attempted=state.pages_attempted, - has_more_remaining=state.has_more_remaining, - pagination_truncated_reason=state.pagination_truncated_reason, - continuation_cstart=state.continuation_cstart, - skipped_no_change=False, - discovered_publication_count=state.discovered_publication_count, - ) - - def _short_circuit_initial_page( - self, - *, - start_cstart: int, - previous_initial_page_fingerprint_sha256: str | None, - fetch_result: FetchResult, - parsed_page: ParsedProfilePage, - first_page_fingerprint_sha256: str | None, - attempt_log: list[dict[str, Any]], - page_logs: list[dict[str, Any]], - ) -> PagedParseResult | None: - if self._should_skip_no_change( - start_cstart=start_cstart, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256, - parsed_page=parsed_page, - ): - return self._skip_no_change_result( - fetch_result=fetch_result, - parsed_page=parsed_page, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - attempt_log=attempt_log, - page_logs=page_logs, - ) - if parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}: - return self._initial_failure_result( - fetch_result=fetch_result, - parsed_page=parsed_page, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - start_cstart=start_cstart, - attempt_log=attempt_log, - page_logs=page_logs, - ) - return None - - async def _fetch_and_parse_all_pages_with_retry( - self, - *, - scholar: ScholarProfile, - run: CrawlRun, - db_session: AsyncSession, - start_cstart: int, - request_delay_seconds: int, - network_error_retries: int, - retry_backoff_seconds: float, - rate_limit_retries: int, - rate_limit_backoff_seconds: float, - max_pages: int, - page_size: int, - previous_initial_page_fingerprint_sha256: str | None = None, - ) -> PagedParseResult: - bounded_max_pages = max(1, int(max_pages)) - bounded_page_size = max(1, int(page_size)) - ( - fetch_result, - parsed_page, - first_page_fingerprint_sha256, - attempt_log, - page_logs, - ) = await self._fetch_initial_page_context( - scholar_id=scholar.scholar_id, - start_cstart=start_cstart, - bounded_page_size=bounded_page_size, - network_error_retries=network_error_retries, - retry_backoff_seconds=retry_backoff_seconds, - rate_limit_retries=rate_limit_retries, - rate_limit_backoff_seconds=rate_limit_backoff_seconds, - ) - shortcut_result = self._short_circuit_initial_page( - start_cstart=start_cstart, - previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256, - fetch_result=fetch_result, - parsed_page=parsed_page, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - attempt_log=attempt_log, - page_logs=page_logs, - ) - if shortcut_result is not None: - return shortcut_result - state = self._build_loop_state( - start_cstart=start_cstart, - fetch_result=fetch_result, - parsed_page=parsed_page, - attempt_log=attempt_log, - page_logs=page_logs, - ) - await self._paginate_loop( - scholar=scholar, - run=run, - db_session=db_session, - state=state, - bounded_max_pages=bounded_max_pages, - request_delay_seconds=request_delay_seconds, - bounded_page_size=bounded_page_size, - network_error_retries=network_error_retries, - retry_backoff_seconds=retry_backoff_seconds, - rate_limit_retries=rate_limit_retries, - rate_limit_backoff_seconds=rate_limit_backoff_seconds, - ) - return self._result_from_pagination_state( - state=state, - first_page_fetch_result=fetch_result, - first_page_parsed_page=parsed_page, - first_page_fingerprint_sha256=first_page_fingerprint_sha256, - ) - async def _run_is_canceled( self, db_session: AsyncSession, @@ -2641,228 +1992,6 @@ class ScholarIngestionService: enriched.append(p) return enriched - async def _upsert_profile_publications( - self, - db_session: AsyncSession, - *, - run: CrawlRun, - scholar: ScholarProfile, - publications: list[PublicationCandidate], - ) -> int: - # We no longer enrich inline here. Enrichment is now deferred to the end of the run. - seen_publication_ids: set[int] = set() - discovered_count = 0 - - for candidate in publications: - publication = await self._resolve_publication(db_session, candidate) - if publication.id in seen_publication_ids: - continue - seen_publication_ids.add(publication.id) - - link_result = await db_session.execute( - select(ScholarPublication).where( - ScholarPublication.scholar_profile_id == scholar.id, - ScholarPublication.publication_id == publication.id, - ) - ) - link = link_result.scalar_one_or_none() - if link is not None: - continue - - link = ScholarPublication( - scholar_profile_id=scholar.id, - publication_id=publication.id, - is_read=False, - first_seen_run_id=run.id, - ) - db_session.add(link) - discovered_count += 1 - - await self._commit_discovered_publication( - db_session, - run=run, - scholar=scholar, - publication=publication, - ) - - if not scholar.baseline_completed: - scholar.baseline_completed = True - - return discovered_count - - async def _commit_discovered_publication( - self, - db_session: AsyncSession, - *, - run: CrawlRun, - scholar: ScholarProfile, - publication: Publication, - ) -> None: - run.new_pub_count = int(run.new_pub_count or 0) + 1 - await db_session.commit() - await run_events.publish( - run_id=run.id, - event_type="publication_discovered", - data={ - "publication_id": publication.id, - "title": publication.title_raw, - "pub_url": publication.pub_url, - "scholar_profile_id": scholar.id, - "scholar_label": scholar.display_name or scholar.scholar_id, - "first_seen_at": datetime.now(UTC).isoformat(), - "new_publication_count": int(run.new_pub_count or 0), - }, - ) - - @staticmethod - def _validate_publication_candidate(candidate: PublicationCandidate) -> None: - if not candidate.title.strip(): - raise RuntimeError("Publication candidate is missing title.") - if candidate.citation_count is not None and int(candidate.citation_count) < 0: - raise RuntimeError("Publication candidate has negative citation_count.") - - async def _find_publication_by_cluster( - self, - db_session: AsyncSession, - *, - cluster_id: str | None, - ) -> Publication | None: - if not cluster_id: - return None - result = await db_session.execute(select(Publication).where(Publication.cluster_id == cluster_id)) - return result.scalar_one_or_none() - - async def _find_publication_by_fingerprint( - self, - db_session: AsyncSession, - *, - fingerprint: str, - ) -> Publication | None: - result = await db_session.execute(select(Publication).where(Publication.fingerprint_sha256 == fingerprint)) - return result.scalar_one_or_none() - - @staticmethod - def _select_existing_publication( - *, - cluster_publication: Publication | None, - fingerprint_publication: Publication | None, - ) -> Publication | None: - if cluster_publication is not None: - return cluster_publication - return fingerprint_publication - - @staticmethod - def _compute_canonical_title_hash(title: str) -> str: - canonical = canonical_title_for_dedup(title) - return hashlib.sha256(canonical.encode("utf-8")).hexdigest() - - async def _find_publication_by_canonical_title_hash( - self, - db_session: AsyncSession, - *, - canonical_title_hash: str, - ) -> Publication | None: - result = await db_session.execute( - select(Publication).where(Publication.canonical_title_hash == canonical_title_hash) - ) - return result.scalar_one_or_none() - - async def _create_publication( - self, - db_session: AsyncSession, - *, - candidate: PublicationCandidate, - fingerprint: str, - ) -> Publication: - publication = Publication( - cluster_id=candidate.cluster_id, - fingerprint_sha256=fingerprint, - title_raw=candidate.title, - title_normalized=normalize_title(candidate.title), - canonical_title_hash=self._compute_canonical_title_hash(candidate.title), - year=candidate.year, - citation_count=int(candidate.citation_count or 0), - author_text=candidate.authors_text, - venue_text=candidate.venue_text, - pub_url=build_publication_url(candidate.title_url), - pdf_url=None, - ) - db_session.add(publication) - await db_session.flush() - return publication - - @staticmethod - def _update_existing_publication( - *, - publication: Publication, - candidate: PublicationCandidate, - ) -> None: - if candidate.cluster_id and publication.cluster_id is None: - publication.cluster_id = candidate.cluster_id - publication.title_raw = candidate.title - publication.title_normalized = normalize_title(candidate.title) - if candidate.year is not None: - publication.year = candidate.year - if candidate.citation_count is not None: - publication.citation_count = int(candidate.citation_count) - if candidate.authors_text: - publication.author_text = candidate.authors_text - if candidate.venue_text: - publication.venue_text = candidate.venue_text - if candidate.title_url: - publication.pub_url = build_publication_url(candidate.title_url) - first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title) - - async def _resolve_publication( - self, - db_session: AsyncSession, - candidate: PublicationCandidate, - ) -> Publication: - self._validate_publication_candidate(candidate) - fingerprint = build_publication_fingerprint(candidate) - cluster_publication = await self._find_publication_by_cluster( - db_session, - cluster_id=candidate.cluster_id, - ) - fingerprint_publication = await self._find_publication_by_fingerprint( - db_session, - fingerprint=fingerprint, - ) - publication = self._select_existing_publication( - cluster_publication=cluster_publication, - fingerprint_publication=fingerprint_publication, - ) - if publication is None: - # Fallback: canonical title hash — catches cross-scholar noise variants - # (e.g. "Adam preprint (2014)" vs "Adam arXiv 2017" → same normalized form) - canonical_hash = self._compute_canonical_title_hash(candidate.title) - publication = await self._find_publication_by_canonical_title_hash( - db_session, - canonical_title_hash=canonical_hash, - ) - if publication is None: - created = await self._create_publication( - db_session, - candidate=candidate, - fingerprint=fingerprint, - ) - # Sync identifiers from local fields only for fast UI response - await identifier_service.sync_identifiers_for_publication_fields( - db_session, - publication=created, - ) - return created - self._update_existing_publication( - publication=publication, - candidate=candidate, - ) - # Sync identifiers from local fields only for fast UI response - await identifier_service.sync_identifiers_for_publication_fields( - db_session, - publication=publication, - ) - return publication - def _resolve_run_status( self, *, diff --git a/app/services/ingestion/page_fetch.py b/app/services/ingestion/page_fetch.py new file mode 100644 index 0000000..008cb6b --- /dev/null +++ b/app/services/ingestion/page_fetch.py @@ -0,0 +1,217 @@ +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +from app.logging_utils import structured_log +from app.services.scholar.parser import ( + ParsedProfilePage, + ParseState, + ScholarParserError, + parse_profile_page, +) +from app.services.scholar.source import FetchResult, ScholarSource + +logger = logging.getLogger(__name__) + + +class PageFetcher: + """Fetches and parses a single Google Scholar profile page with retry logic.""" + + def __init__(self, *, source: ScholarSource) -> None: + self._source = source + + async def fetch_profile_page( + self, + *, + scholar_id: str, + cstart: int, + page_size: int, + ) -> FetchResult: + try: + page_fetcher = getattr(self._source, "fetch_profile_page_html", None) + if callable(page_fetcher): + return await page_fetcher( + scholar_id, + cstart=cstart, + pagesize=page_size, + ) + if cstart <= 0: + return await self._source.fetch_profile_html(scholar_id) + return FetchResult( + requested_url=( + f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}" + ), + status_code=None, + final_url=None, + body="", + error="source_does_not_support_pagination", + ) + except Exception as exc: + logger.exception( + "ingestion.fetch_unexpected_error", + extra={ + "scholar_id": scholar_id, + "cstart": cstart, + "page_size": page_size, + }, + ) + return FetchResult( + requested_url=( + f"https://scholar.google.com/citations?hl=en&user={scholar_id}&cstart={cstart}&pagesize={page_size}" + ), + status_code=None, + final_url=None, + body="", + error=str(exc), + ) + + # ── Parse helpers ──────────────────────────────────────────────── + + def parse_page_or_layout_error( + self, + *, + fetch_result: FetchResult, + ) -> ParsedProfilePage: + try: + return parse_profile_page(fetch_result) + except ScholarParserError as exc: + return self._parsed_page_from_parser_error(code=exc.code) + + @staticmethod + def _parsed_page_from_parser_error(*, code: str) -> ParsedProfilePage: + return ParsedProfilePage( + state=ParseState.LAYOUT_CHANGED, + state_reason=code, + profile_name=None, + profile_image_url=None, + publications=[], + marker_counts={}, + warnings=[code], + has_show_more_button=False, + has_operation_error_banner=False, + articles_range=None, + ) + + # ── Retry logic ────────────────────────────────────────────────── + + @staticmethod + def _should_retry( + *, + parsed_page: ParsedProfilePage, + network_attempt_count: int, + rate_limit_attempt_count: int, + network_error_retries: int, + rate_limit_retries: int, + ) -> bool: + if parsed_page.state == ParseState.NETWORK_ERROR: + return network_attempt_count <= network_error_retries + if ( + parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA + and parsed_page.state_reason == "blocked_http_429_rate_limited" + ): + return rate_limit_attempt_count <= rate_limit_retries + return False + + @staticmethod + async def _sleep_backoff( + *, + scholar_id: str, + cstart: int, + network_attempt_count: int, + rate_limit_attempt_count: int, + retry_backoff_seconds: float, + rate_limit_backoff_seconds: float, + state_reason: str, + ) -> None: + if state_reason == "blocked_http_429_rate_limited": + sleep_seconds = rate_limit_backoff_seconds * rate_limit_attempt_count + attempt_label = rate_limit_attempt_count + else: + sleep_seconds = retry_backoff_seconds * (2 ** (network_attempt_count - 1)) + attempt_label = network_attempt_count + + structured_log( + logger, + "warning", + "ingestion.scholar_retry_scheduled", + scholar_id=scholar_id, + cstart=cstart, + attempt_count=attempt_label, + sleep_seconds=sleep_seconds, + state_reason=state_reason, + ) + if sleep_seconds > 0: + await asyncio.sleep(sleep_seconds) + + async def fetch_and_parse_with_retry( + self, + *, + scholar_id: str, + cstart: int, + page_size: int, + network_error_retries: int, + retry_backoff_seconds: float, + rate_limit_retries: int, + rate_limit_backoff_seconds: float, + ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]: + network_attempts = 0 + rate_limit_attempts = 0 + attempt_log: list[dict[str, Any]] = [] + fetch_result: FetchResult | None = None + parsed_page: ParsedProfilePage | None = None + + while True: + fetch_result = await self.fetch_profile_page( + scholar_id=scholar_id, + cstart=cstart, + page_size=page_size, + ) + parsed_page = self.parse_page_or_layout_error(fetch_result=fetch_result) + + if parsed_page.state == ParseState.NETWORK_ERROR: + network_attempts += 1 + total_attempts = network_attempts + elif ( + parsed_page.state == ParseState.BLOCKED_OR_CAPTCHA + and parsed_page.state_reason == "blocked_http_429_rate_limited" + ): + rate_limit_attempts += 1 + total_attempts = rate_limit_attempts + else: + total_attempts = network_attempts + rate_limit_attempts + 1 + + attempt_log.append( + { + "attempt": total_attempts, + "cstart": cstart, + "state": parsed_page.state.value, + "state_reason": parsed_page.state_reason, + "status_code": fetch_result.status_code, + "fetch_error": fetch_result.error, + } + ) + + if not self._should_retry( + parsed_page=parsed_page, + network_attempt_count=network_attempts, + rate_limit_attempt_count=rate_limit_attempts, + network_error_retries=network_error_retries, + rate_limit_retries=rate_limit_retries, + ): + break + + await self._sleep_backoff( + scholar_id=scholar_id, + cstart=cstart, + network_attempt_count=network_attempts, + rate_limit_attempt_count=rate_limit_attempts, + retry_backoff_seconds=max(float(retry_backoff_seconds), 0.0), + rate_limit_backoff_seconds=max(float(rate_limit_backoff_seconds), 0.0), + state_reason=parsed_page.state_reason, + ) + + if fetch_result is None or parsed_page is None: + raise RuntimeError("Fetch-and-parse retry loop produced no result.") + return fetch_result, parsed_page, attempt_log diff --git a/app/services/ingestion/pagination.py b/app/services/ingestion/pagination.py new file mode 100644 index 0000000..9e761b8 --- /dev/null +++ b/app/services/ingestion/pagination.py @@ -0,0 +1,486 @@ +from __future__ import annotations + +import asyncio +import logging +from typing import Any + +from app.db.models import CrawlRun, RunStatus, ScholarProfile +from app.logging_utils import structured_log +from app.services.ingestion.fingerprints import ( + _dedupe_publication_candidates, + _next_cstart_value, + build_initial_page_fingerprint, +) +from app.services.ingestion.page_fetch import PageFetcher +from app.services.ingestion.types import PagedLoopState, PagedParseResult +from app.services.scholar.parser import ParsedProfilePage, ParseState +from app.services.scholar.source import FetchResult, ScholarSource + +logger = logging.getLogger(__name__) + + +class PaginationEngine: + """Fetches and paginates Google Scholar profile pages. + + Pure HTTP + parsing — no DB writes except via the provided upsert callback. + """ + + def __init__(self, *, source: ScholarSource) -> None: + self._source = source + self._fetcher = PageFetcher(source=source) + + # ── No-change / initial failure short-circuits ─────────────────── + + @staticmethod + def _should_skip_no_change( + *, + start_cstart: int, + first_page_fingerprint_sha256: str | None, + previous_initial_page_fingerprint_sha256: str | None, + parsed_page: ParsedProfilePage, + ) -> bool: + return ( + start_cstart <= 0 + and first_page_fingerprint_sha256 is not None + and previous_initial_page_fingerprint_sha256 is not None + and first_page_fingerprint_sha256 == previous_initial_page_fingerprint_sha256 + and parsed_page.state in {ParseState.OK, ParseState.NO_RESULTS} + ) + + @staticmethod + def _skip_no_change_result( + *, + fetch_result: FetchResult, + parsed_page: ParsedProfilePage, + first_page_fingerprint_sha256: str | None, + attempt_log: list[dict[str, Any]], + page_logs: list[dict[str, Any]], + ) -> PagedParseResult: + return PagedParseResult( + fetch_result=fetch_result, + parsed_page=parsed_page, + first_page_fetch_result=fetch_result, + first_page_parsed_page=parsed_page, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + publications=[], + attempt_log=attempt_log, + page_logs=page_logs, + pages_fetched=1, + pages_attempted=1, + has_more_remaining=False, + pagination_truncated_reason=None, + continuation_cstart=None, + skipped_no_change=True, + discovered_publication_count=0, + ) + + @staticmethod + def _initial_failure_result( + *, + fetch_result: FetchResult, + parsed_page: ParsedProfilePage, + first_page_fingerprint_sha256: str | None, + start_cstart: int, + attempt_log: list[dict[str, Any]], + page_logs: list[dict[str, Any]], + ) -> PagedParseResult: + continuation_cstart = start_cstart if parsed_page.state == ParseState.NETWORK_ERROR else None + return PagedParseResult( + fetch_result=fetch_result, + parsed_page=parsed_page, + first_page_fetch_result=fetch_result, + first_page_parsed_page=parsed_page, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + publications=[], + attempt_log=attempt_log, + page_logs=page_logs, + pages_fetched=0, + pages_attempted=1, + has_more_remaining=False, + pagination_truncated_reason=None, + continuation_cstart=continuation_cstart, + skipped_no_change=False, + discovered_publication_count=0, + ) + + # ── Loop state management ──────────────────────────────────────── + + @staticmethod + def _build_loop_state( + *, + start_cstart: int, + fetch_result: FetchResult, + parsed_page: ParsedProfilePage, + attempt_log: list[dict[str, Any]], + page_logs: list[dict[str, Any]], + ) -> PagedLoopState: + next_cstart = _next_cstart_value( + articles_range=parsed_page.articles_range, + fallback=start_cstart + len(parsed_page.publications), + ) + return PagedLoopState( + fetch_result=fetch_result, + parsed_page=parsed_page, + attempt_log=attempt_log, + page_logs=page_logs, + publications=list(parsed_page.publications), + pages_fetched=1, + pages_attempted=1, + current_cstart=start_cstart, + next_cstart=next_cstart, + ) + + @staticmethod + def _set_truncated_state( + *, + state: PagedLoopState, + reason: str, + continuation_cstart: int, + ) -> None: + state.has_more_remaining = True + state.pagination_truncated_reason = reason + state.continuation_cstart = continuation_cstart + + def _should_stop_pagination(self, *, state: PagedLoopState, bounded_max_pages: int) -> bool: + if state.pages_fetched >= bounded_max_pages: + self._set_truncated_state( + state=state, + reason="max_pages_reached", + continuation_cstart=( + state.next_cstart if state.next_cstart > state.current_cstart else state.current_cstart + ), + ) + return True + if state.next_cstart <= state.current_cstart: + self._set_truncated_state( + state=state, + reason="pagination_cursor_stalled", + continuation_cstart=state.current_cstart, + ) + return True + return False + + # ── Multi-page fetch helpers ───────────────────────────────────── + + async def _fetch_next_page( + self, + *, + scholar_id: str, + state: PagedLoopState, + request_delay_seconds: int, + bounded_page_size: int, + network_error_retries: int, + retry_backoff_seconds: float, + rate_limit_retries: int, + rate_limit_backoff_seconds: float, + ) -> tuple[FetchResult, ParsedProfilePage, list[dict[str, Any]]]: + if request_delay_seconds > 0: + await asyncio.sleep(float(request_delay_seconds)) + state.current_cstart = state.next_cstart + return await self._fetcher.fetch_and_parse_with_retry( + scholar_id=scholar_id, + cstart=state.current_cstart, + page_size=bounded_page_size, + network_error_retries=network_error_retries, + retry_backoff_seconds=retry_backoff_seconds, + rate_limit_retries=rate_limit_retries, + rate_limit_backoff_seconds=rate_limit_backoff_seconds, + ) + + @staticmethod + def _record_next_page( + *, + state: PagedLoopState, + fetch_result: FetchResult, + parsed_page: ParsedProfilePage, + page_attempt_log: list[dict[str, Any]], + ) -> None: + state.pages_attempted += 1 + state.attempt_log.extend(page_attempt_log) + state.page_logs.append( + { + "page": state.pages_attempted, + "cstart": state.current_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": len(page_attempt_log), + } + ) + state.fetch_result = fetch_result + state.parsed_page = parsed_page + + def _handle_page_state_transition(self, *, state: PagedLoopState) -> bool: + if state.parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}: + self._set_truncated_state( + state=state, + reason=f"page_state_{state.parsed_page.state.value}", + continuation_cstart=state.current_cstart, + ) + return True + if state.parsed_page.state == ParseState.NO_RESULTS and len(state.parsed_page.publications) == 0: + state.pages_fetched += 1 + return True + state.pages_fetched += 1 + state.publications.extend(state.parsed_page.publications) + state.next_cstart = _next_cstart_value( + articles_range=state.parsed_page.articles_range, + fallback=state.current_cstart + len(state.parsed_page.publications), + ) + return False + + async def _fetch_initial_page_context( + self, + *, + scholar_id: str, + start_cstart: int, + bounded_page_size: int, + network_error_retries: int, + retry_backoff_seconds: float, + rate_limit_retries: int, + rate_limit_backoff_seconds: float, + ) -> tuple[FetchResult, ParsedProfilePage, str | None, list[dict[str, Any]], list[dict[str, Any]]]: + fetch_result, parsed_page, first_attempt_log = await self._fetcher.fetch_and_parse_with_retry( + scholar_id=scholar_id, + cstart=start_cstart, + page_size=bounded_page_size, + network_error_retries=network_error_retries, + retry_backoff_seconds=retry_backoff_seconds, + rate_limit_retries=rate_limit_retries, + rate_limit_backoff_seconds=rate_limit_backoff_seconds, + ) + first_page_fingerprint_sha256 = build_initial_page_fingerprint(parsed_page) + attempt_log = list(first_attempt_log) + page_logs = [ + { + "page": 1, + "cstart": start_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": len(first_attempt_log), + } + ] + return fetch_result, parsed_page, first_page_fingerprint_sha256, attempt_log, page_logs + + # ── Pagination loop ────────────────────────────────────────────── + + async def _paginate_loop( + self, + *, + scholar: ScholarProfile, + run: CrawlRun, + db_session: Any, + state: PagedLoopState, + bounded_max_pages: int, + request_delay_seconds: int, + bounded_page_size: int, + network_error_retries: int, + retry_backoff_seconds: float, + rate_limit_retries: int, + rate_limit_backoff_seconds: float, + upsert_publications_fn: Any, + ) -> None: + seen_canonical: set[str] = set() + + if state.parsed_page.publications: + deduped_first = _dedupe_publication_candidates( + list(state.parsed_page.publications), seen_canonical=seen_canonical + ) + if deduped_first: + discovered_count = await upsert_publications_fn( + db_session, run=run, scholar=scholar, publications=deduped_first + ) + state.discovered_publication_count += discovered_count + + while state.parsed_page.has_show_more_button: + await db_session.refresh(run) + if run.status == RunStatus.CANCELED: + structured_log( + logger, + "info", + "ingestion.pagination_canceled", + run_id=run.id, + ) + self._set_truncated_state( + state=state, + reason="run_canceled", + continuation_cstart=state.current_cstart, + ) + return + + if self._should_stop_pagination(state=state, bounded_max_pages=bounded_max_pages): + return + next_fetch_result, next_parsed_page, next_attempt_log = await self._fetch_next_page( + scholar_id=scholar.scholar_id, + state=state, + request_delay_seconds=request_delay_seconds, + bounded_page_size=bounded_page_size, + network_error_retries=network_error_retries, + retry_backoff_seconds=retry_backoff_seconds, + rate_limit_retries=rate_limit_retries, + rate_limit_backoff_seconds=rate_limit_backoff_seconds, + ) + self._record_next_page( + state=state, + fetch_result=next_fetch_result, + parsed_page=next_parsed_page, + page_attempt_log=next_attempt_log, + ) + + if next_parsed_page.publications: + deduped_next = _dedupe_publication_candidates( + list(next_parsed_page.publications), seen_canonical=seen_canonical + ) + if deduped_next: + discovered_count = await upsert_publications_fn( + db_session, run=run, scholar=scholar, publications=deduped_next + ) + state.discovered_publication_count += discovered_count + + if self._handle_page_state_transition(state=state): + return + + @staticmethod + def _result_from_pagination_state( + *, + state: PagedLoopState, + first_page_fetch_result: FetchResult, + first_page_parsed_page: ParsedProfilePage, + first_page_fingerprint_sha256: str | None, + ) -> PagedParseResult: + return PagedParseResult( + fetch_result=state.fetch_result, + parsed_page=state.parsed_page, + first_page_fetch_result=first_page_fetch_result, + first_page_parsed_page=first_page_parsed_page, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + publications=_dedupe_publication_candidates(state.publications), + attempt_log=state.attempt_log, + page_logs=state.page_logs, + pages_fetched=state.pages_fetched, + pages_attempted=state.pages_attempted, + has_more_remaining=state.has_more_remaining, + pagination_truncated_reason=state.pagination_truncated_reason, + continuation_cstart=state.continuation_cstart, + skipped_no_change=False, + discovered_publication_count=state.discovered_publication_count, + ) + + def _short_circuit_initial_page( + self, + *, + start_cstart: int, + previous_initial_page_fingerprint_sha256: str | None, + fetch_result: FetchResult, + parsed_page: ParsedProfilePage, + first_page_fingerprint_sha256: str | None, + attempt_log: list[dict[str, Any]], + page_logs: list[dict[str, Any]], + ) -> PagedParseResult | None: + if self._should_skip_no_change( + start_cstart=start_cstart, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256, + parsed_page=parsed_page, + ): + return self._skip_no_change_result( + fetch_result=fetch_result, + parsed_page=parsed_page, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + attempt_log=attempt_log, + page_logs=page_logs, + ) + if parsed_page.state not in {ParseState.OK, ParseState.NO_RESULTS}: + return self._initial_failure_result( + fetch_result=fetch_result, + parsed_page=parsed_page, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + start_cstart=start_cstart, + attempt_log=attempt_log, + page_logs=page_logs, + ) + return None + + # ── Public entry point ─────────────────────────────────────────── + + async def fetch_and_parse_all_pages( + self, + *, + scholar: ScholarProfile, + run: CrawlRun, + db_session: Any, + start_cstart: int, + request_delay_seconds: int, + network_error_retries: int, + retry_backoff_seconds: float, + rate_limit_retries: int, + rate_limit_backoff_seconds: float, + max_pages: int, + page_size: int, + previous_initial_page_fingerprint_sha256: str | None = None, + upsert_publications_fn: Any = None, + ) -> PagedParseResult: + bounded_max_pages = max(1, int(max_pages)) + bounded_page_size = max(1, int(page_size)) + ( + fetch_result, + parsed_page, + first_page_fingerprint_sha256, + attempt_log, + page_logs, + ) = await self._fetch_initial_page_context( + scholar_id=scholar.scholar_id, + start_cstart=start_cstart, + bounded_page_size=bounded_page_size, + network_error_retries=network_error_retries, + retry_backoff_seconds=retry_backoff_seconds, + rate_limit_retries=rate_limit_retries, + rate_limit_backoff_seconds=rate_limit_backoff_seconds, + ) + shortcut_result = self._short_circuit_initial_page( + start_cstart=start_cstart, + previous_initial_page_fingerprint_sha256=previous_initial_page_fingerprint_sha256, + fetch_result=fetch_result, + parsed_page=parsed_page, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + attempt_log=attempt_log, + page_logs=page_logs, + ) + if shortcut_result is not None: + return shortcut_result + state = self._build_loop_state( + start_cstart=start_cstart, + fetch_result=fetch_result, + parsed_page=parsed_page, + attempt_log=attempt_log, + page_logs=page_logs, + ) + await self._paginate_loop( + scholar=scholar, + run=run, + db_session=db_session, + state=state, + bounded_max_pages=bounded_max_pages, + request_delay_seconds=request_delay_seconds, + bounded_page_size=bounded_page_size, + network_error_retries=network_error_retries, + retry_backoff_seconds=retry_backoff_seconds, + rate_limit_retries=rate_limit_retries, + rate_limit_backoff_seconds=rate_limit_backoff_seconds, + upsert_publications_fn=upsert_publications_fn, + ) + return self._result_from_pagination_state( + state=state, + first_page_fetch_result=fetch_result, + first_page_parsed_page=parsed_page, + first_page_fingerprint_sha256=first_page_fingerprint_sha256, + ) diff --git a/app/services/ingestion/publication_upsert.py b/app/services/ingestion/publication_upsert.py new file mode 100644 index 0000000..9d1a6b7 --- /dev/null +++ b/app/services/ingestion/publication_upsert.py @@ -0,0 +1,239 @@ +from __future__ import annotations + +import hashlib +import logging +from datetime import UTC, datetime + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.db.models import CrawlRun, Publication, ScholarProfile, ScholarPublication +from app.services.doi.normalize import first_doi_from_texts +from app.services.ingestion.fingerprints import ( + build_publication_fingerprint, + build_publication_url, + canonical_title_for_dedup, + normalize_title, +) +from app.services.publication_identifiers import application as identifier_service +from app.services.runs.events import run_events +from app.services.scholar.parser import PublicationCandidate + +logger = logging.getLogger(__name__) + + +def validate_publication_candidate(candidate: PublicationCandidate) -> None: + if not candidate.title.strip(): + raise RuntimeError("Publication candidate is missing title.") + if candidate.citation_count is not None and int(candidate.citation_count) < 0: + raise RuntimeError("Publication candidate has negative citation_count.") + + +async def find_publication_by_cluster( + db_session: AsyncSession, + *, + cluster_id: str | None, +) -> Publication | None: + if not cluster_id: + return None + result = await db_session.execute(select(Publication).where(Publication.cluster_id == cluster_id)) + return result.scalar_one_or_none() + + +async def find_publication_by_fingerprint( + db_session: AsyncSession, + *, + fingerprint: str, +) -> Publication | None: + result = await db_session.execute(select(Publication).where(Publication.fingerprint_sha256 == fingerprint)) + return result.scalar_one_or_none() + + +def select_existing_publication( + *, + cluster_publication: Publication | None, + fingerprint_publication: Publication | None, +) -> Publication | None: + if cluster_publication is not None: + return cluster_publication + return fingerprint_publication + + +def compute_canonical_title_hash(title: str) -> str: + canonical = canonical_title_for_dedup(title) + return hashlib.sha256(canonical.encode("utf-8")).hexdigest() + + +async def find_publication_by_canonical_title_hash( + db_session: AsyncSession, + *, + canonical_title_hash: str, +) -> Publication | None: + result = await db_session.execute( + select(Publication).where(Publication.canonical_title_hash == canonical_title_hash) + ) + return result.scalar_one_or_none() + + +async def create_publication( + db_session: AsyncSession, + *, + candidate: PublicationCandidate, + fingerprint: str, +) -> Publication: + publication = Publication( + cluster_id=candidate.cluster_id, + fingerprint_sha256=fingerprint, + title_raw=candidate.title, + title_normalized=normalize_title(candidate.title), + canonical_title_hash=compute_canonical_title_hash(candidate.title), + year=candidate.year, + citation_count=int(candidate.citation_count or 0), + author_text=candidate.authors_text, + venue_text=candidate.venue_text, + pub_url=build_publication_url(candidate.title_url), + pdf_url=None, + ) + db_session.add(publication) + await db_session.flush() + return publication + + +def update_existing_publication( + *, + publication: Publication, + candidate: PublicationCandidate, +) -> None: + if candidate.cluster_id and publication.cluster_id is None: + publication.cluster_id = candidate.cluster_id + publication.title_raw = candidate.title + publication.title_normalized = normalize_title(candidate.title) + if candidate.year is not None: + publication.year = candidate.year + if candidate.citation_count is not None: + publication.citation_count = int(candidate.citation_count) + if candidate.authors_text: + publication.author_text = candidate.authors_text + if candidate.venue_text: + publication.venue_text = candidate.venue_text + if candidate.title_url: + publication.pub_url = build_publication_url(candidate.title_url) + first_doi_from_texts(candidate.title_url, candidate.venue_text, candidate.title) + + +async def resolve_publication( + db_session: AsyncSession, + candidate: PublicationCandidate, +) -> Publication: + validate_publication_candidate(candidate) + fingerprint = build_publication_fingerprint(candidate) + cluster_publication = await find_publication_by_cluster( + db_session, + cluster_id=candidate.cluster_id, + ) + fingerprint_publication = await find_publication_by_fingerprint( + db_session, + fingerprint=fingerprint, + ) + publication = select_existing_publication( + cluster_publication=cluster_publication, + fingerprint_publication=fingerprint_publication, + ) + if publication is None: + canonical_hash = compute_canonical_title_hash(candidate.title) + publication = await find_publication_by_canonical_title_hash( + db_session, + canonical_title_hash=canonical_hash, + ) + if publication is None: + created = await create_publication( + db_session, + candidate=candidate, + fingerprint=fingerprint, + ) + await identifier_service.sync_identifiers_for_publication_fields( + db_session, + publication=created, + ) + return created + update_existing_publication( + publication=publication, + candidate=candidate, + ) + await identifier_service.sync_identifiers_for_publication_fields( + db_session, + publication=publication, + ) + return publication + + +async def upsert_profile_publications( + db_session: AsyncSession, + *, + run: CrawlRun, + scholar: ScholarProfile, + publications: list[PublicationCandidate], +) -> int: + seen_publication_ids: set[int] = set() + discovered_count = 0 + + for candidate in publications: + publication = await resolve_publication(db_session, candidate) + if publication.id in seen_publication_ids: + continue + seen_publication_ids.add(publication.id) + + link_result = await db_session.execute( + select(ScholarPublication).where( + ScholarPublication.scholar_profile_id == scholar.id, + ScholarPublication.publication_id == publication.id, + ) + ) + link = link_result.scalar_one_or_none() + if link is not None: + continue + + link = ScholarPublication( + scholar_profile_id=scholar.id, + publication_id=publication.id, + is_read=False, + first_seen_run_id=run.id, + ) + db_session.add(link) + discovered_count += 1 + + await commit_discovered_publication( + db_session, + run=run, + scholar=scholar, + publication=publication, + ) + + if not scholar.baseline_completed: + scholar.baseline_completed = True + + return discovered_count + + +async def commit_discovered_publication( + db_session: AsyncSession, + *, + run: CrawlRun, + scholar: ScholarProfile, + publication: Publication, +) -> None: + run.new_pub_count = int(run.new_pub_count or 0) + 1 + await db_session.commit() + await run_events.publish( + run_id=run.id, + event_type="publication_discovered", + data={ + "publication_id": publication.id, + "title": publication.title_raw, + "pub_url": publication.pub_url, + "scholar_profile_id": scholar.id, + "scholar_label": scholar.display_name or scholar.scholar_id, + "first_seen_at": datetime.now(UTC).isoformat(), + "new_publication_count": int(run.new_pub_count or 0), + }, + ) diff --git a/app/services/portability/normalize.py b/app/services/portability/normalize.py index 41e017f..9353d2a 100644 --- a/app/services/portability/normalize.py +++ b/app/services/portability/normalize.py @@ -3,7 +3,7 @@ from __future__ import annotations import hashlib from typing import Any -from app.services.ingestion.application import normalize_title +from app.services.ingestion.fingerprints import normalize_title from app.services.portability.constants import ( MAX_IMPORT_PUBLICATIONS, MAX_IMPORT_SCHOLARS, diff --git a/app/services/portability/publication_import.py b/app/services/portability/publication_import.py index a347f4a..e2eb96a 100644 --- a/app/services/portability/publication_import.py +++ b/app/services/portability/publication_import.py @@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from app.db.models import Publication, ScholarProfile, ScholarPublication from app.services.doi.normalize import normalize_doi -from app.services.ingestion.application import build_publication_url, normalize_title +from app.services.ingestion.fingerprints import build_publication_url, normalize_title from app.services.portability.normalize import ( _normalize_citation_count, _normalize_optional_text, diff --git a/tests/integration/test_run_lifecycle_consistency.py b/tests/integration/test_run_lifecycle_consistency.py index 7592abf..f8239e4 100644 --- a/tests/integration/test_run_lifecycle_consistency.py +++ b/tests/integration/test_run_lifecycle_consistency.py @@ -326,7 +326,6 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( db_session.add_all([run, publication]) await db_session.commit() - service = ScholarIngestionService(source=object()) call_count = 0 async def _resolve_publication_stub(*_args: Any, **_kwargs: Any) -> Publication: @@ -336,7 +335,9 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( return publication raise RuntimeError("mid_page_failure") - monkeypatch.setattr(service, "_resolve_publication", _resolve_publication_stub) + from app.services.ingestion import publication_upsert + + monkeypatch.setattr(publication_upsert, "resolve_publication", _resolve_publication_stub) from app.db.models import ScholarProfile from app.services.scholar.parser_types import PublicationCandidate @@ -368,7 +369,7 @@ async def test_partial_discovery_exception_keeps_new_pub_count_consistent( ] with pytest.raises(RuntimeError, match="mid_page_failure"): - await service._upsert_profile_publications( + await publication_upsert.upsert_profile_publications( db_session, run=run, scholar=scholar,