diff --git a/shelfmark/core/release_search.py b/shelfmark/core/release_search.py new file mode 100644 index 00000000..44064483 --- /dev/null +++ b/shelfmark/core/release_search.py @@ -0,0 +1,93 @@ +"""Shared release-search helpers. + +Extracted from the ``/api/releases`` route so the same per-source search logic can +be reused outside the HTTP route (for example by background automation) without +going through Flask. Behaviour for the HTTP route is preserved: the route delegates +its inner per-source search to :func:`search_source_releases`. +""" + +from __future__ import annotations + +import sqlite3 +from typing import TYPE_CHECKING + +from shelfmark.core.logger import setup_logger +from shelfmark.core.search_plan import build_release_search_plan + +if TYPE_CHECKING: + from shelfmark.core.models import SearchFilters + from shelfmark.metadata_providers import BookMetadata + from shelfmark.release_sources import Release, ReleaseSource + +logger = setup_logger(__name__) + +# Mirror of main._OPERATIONAL_ERRORS so a misbehaving source can't crash a caller. +_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error) + + +def search_source_releases( + source_name: str, + search_book: BookMetadata, + *, + languages: list[str] | None = None, + manual_query: str | None = None, + indexers: list[str] | None = None, + expand_search: bool = False, + content_type: str = "ebook", + source_filters: SearchFilters | None = None, + user_id: int | None = None, +) -> tuple[ReleaseSource | None, list[Release], str | None]: + """Search a single release source, returning any error instead of raising. + + Returns ``(source, releases, error_message)``. On failure ``source`` is ``None`` + and ``error_message`` describes the problem. ``user_id`` lets the search plan + pick up that user's default languages when no explicit filter is given. + """ + from shelfmark.release_sources import SourceUnavailableError, get_source + + try: + source = get_source(source_name) + + plan = build_release_search_plan( + search_book, + languages=languages, + manual_query=manual_query, + indexers=indexers, + source_filters=source_filters, + user_id=user_id, + ) + + if plan.source_filters is not None: + planned_query = plan.manual_query or plan.primary_query + planned_query_type = "query" + elif plan.manual_query: + planned_query = plan.manual_query + planned_query_type = "manual" + elif not expand_search and plan.isbn_candidates: + planned_query = plan.isbn_candidates[0] + planned_query_type = "isbn" + else: + planned_query = plan.primary_query + planned_query_type = "title_author" + + logger.debug( + "Searching %s: %s='%s' (title='%s', authors=%s, expand=%s, content_type=%s)", + source_name, + planned_query_type, + planned_query, + search_book.title, + search_book.authors, + expand_search, + content_type, + ) + + releases = source.search( + search_book, plan, expand_search=expand_search, content_type=content_type + ) + except ValueError: + return None, [], f"Unknown source: {source_name}" + except (SourceUnavailableError, *_OPERATIONAL_ERRORS) as exc: + logger.warning("Release search failed for source %s: %s", source_name, exc) + return None, [], f"{source_name}: {exc!s}" + else: + return source, releases, None diff --git a/shelfmark/main.py b/shelfmark/main.py index dc8dab52..f5359b02 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -2829,7 +2829,7 @@ def api_releases() -> Response | tuple[Response, int]: try: from dataclasses import asdict - from shelfmark.core.search_plan import build_release_search_plan + from shelfmark.core.release_search import search_source_releases from shelfmark.metadata_providers import ( BookMetadata, get_provider, @@ -2848,54 +2848,17 @@ def api_releases() -> Response | tuple[Response, int]: source_name: str, search_book: BookMetadata ) -> tuple[Any | None, list[Any], str | None]: """Search one source and return any error message instead of raising.""" - try: - source = get_source(source_name) - - plan = build_release_search_plan( - search_book, - languages=browse_filters.lang - if source_query_filters is not None - else languages, - manual_query=query_text if source_query_filters is not None else manual_query, - indexers=indexers, - source_filters=source_query_filters, - user_id=db_user_id, - ) - - if plan.source_filters is not None: - planned_query = plan.manual_query or plan.primary_query - planned_query_type = "query" - elif plan.manual_query: - planned_query = plan.manual_query - planned_query_type = "manual" - elif not expand_search and plan.isbn_candidates: - planned_query = plan.isbn_candidates[0] - planned_query_type = "isbn" - else: - planned_query = plan.primary_query - planned_query_type = "title_author" - - logger.debug( - "Searching %s: %s='%s' (title='%s', authors=%s, expand=%s, content_type=%s)", - source_name, - planned_query_type, - planned_query, - search_book.title, - search_book.authors, - expand_search, - content_type, - ) - - releases = source.search( - search_book, plan, expand_search=expand_search, content_type=content_type - ) - except ValueError: - return None, [], f"Unknown source: {source_name}" - except (SourceUnavailableError, *_OPERATIONAL_ERRORS) as e: - logger.warning("Release search failed for source %s: %s", source_name, e) - return None, [], f"{source_name}: {e!s}" - else: - return source, releases, None + return search_source_releases( + source_name, + search_book, + languages=(browse_filters.lang if source_query_filters is not None else languages), + manual_query=(query_text if source_query_filters is not None else manual_query), + indexers=indexers, + expand_search=expand_search, + content_type=content_type, + source_filters=source_query_filters, + user_id=db_user_id, + ) provider = request.args.get("provider", "").strip() book_id = request.args.get("book_id", "").strip() diff --git a/tests/core/test_release_search.py b/tests/core/test_release_search.py new file mode 100644 index 00000000..3434d28a --- /dev/null +++ b/tests/core/test_release_search.py @@ -0,0 +1,119 @@ +"""Tests for the per-source release search helper extracted from ``/api/releases``.""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from shelfmark.core import release_search +from shelfmark.metadata_providers import BookMetadata +from shelfmark.release_sources import SourceUnavailableError + + +def _book(**overrides: Any) -> BookMetadata: + fields: dict[str, Any] = { + "provider": "manual", + "provider_id": "1", + "title": "Dungeon Crawler Carl", + "authors": ["Matt Dinniman"], + "isbn_13": "9780593820247", + } + fields.update(overrides) + return BookMetadata(**fields) + + +class _Source: + def __init__(self, result: list[str] | None = None, error: Exception | None = None) -> None: + self.result = result if result is not None else [] + self.error = error + self.calls: list[tuple[Any, Any, bool, str]] = [] + + def search( + self, book: BookMetadata, plan: Any, *, expand_search: bool, content_type: str + ) -> list[str]: + self.calls.append((book, plan, expand_search, content_type)) + if self.error is not None: + raise self.error + return list(self.result) + + +def test_unknown_source_returns_error_tuple(monkeypatch: pytest.MonkeyPatch) -> None: + def _raise(name: str) -> Any: + msg = f"Unknown source: {name}" + raise ValueError(msg) + + monkeypatch.setattr("shelfmark.release_sources.get_source", _raise) + + source, releases, error = release_search.search_source_releases("nope", _book()) + + assert source is None + assert releases == [] + assert error == "Unknown source: nope" + + +@pytest.mark.parametrize( + "exc", + [SourceUnavailableError("down"), OSError("boom"), RuntimeError("bad")], + ids=["unavailable", "oserror", "runtime"], +) +def test_source_failures_become_messages(monkeypatch: pytest.MonkeyPatch, exc: Exception) -> None: + src = _Source(error=exc) + monkeypatch.setattr("shelfmark.release_sources.get_source", lambda _name: src) + + source, releases, error = release_search.search_source_releases("s", _book()) + + assert source is None + assert releases == [] + assert error == f"s: {exc!s}" + + +def test_success_returns_source_and_forwards_search_kwargs( + monkeypatch: pytest.MonkeyPatch, +) -> None: + src = _Source(result=["r1", "r2"]) + monkeypatch.setattr("shelfmark.release_sources.get_source", lambda _name: src) + + source, releases, error = release_search.search_source_releases( + "s", _book(), expand_search=True, content_type="audiobook" + ) + + assert source is src + assert releases == ["r1", "r2"] + assert error is None + book, plan, expand_search, content_type = src.calls[0] + assert book.title == "Dungeon Crawler Carl" + assert "dungeon crawler carl" in plan.primary_query.lower() + assert plan.isbn_candidates == ["9780593820247"] + assert expand_search is True + assert content_type == "audiobook" + + +def test_plan_receives_filters_and_user_id(monkeypatch: pytest.MonkeyPatch) -> None: + captured: dict[str, Any] = {} + real_build = release_search.build_release_search_plan + + def _build(book: BookMetadata, **kwargs: Any) -> Any: + captured.update(kwargs) + return real_build(book, **kwargs) + + monkeypatch.setattr(release_search, "build_release_search_plan", _build) + src = _Source() + monkeypatch.setattr("shelfmark.release_sources.get_source", lambda _name: src) + + release_search.search_source_releases( + "s", + _book(), + languages=["en"], + manual_query="dcc", + indexers=["idx"], + user_id=42, + ) + + assert captured == { + "languages": ["en"], + "manual_query": "dcc", + "indexers": ["idx"], + "source_filters": None, + "user_id": 42, + }