diff --git a/shelfmark/core/metadata_routes.py b/shelfmark/core/metadata_routes.py new file mode 100644 index 00000000..067134ce --- /dev/null +++ b/shelfmark/core/metadata_routes.py @@ -0,0 +1,541 @@ +"""Metadata providers: the ``/api/metadata/...`` search, book and target routes.""" + +from __future__ import annotations + +import sqlite3 +from contextlib import suppress +from functools import wraps +from typing import TYPE_CHECKING, Any + +from flask import jsonify, request, session + +from shelfmark.core.config import config as app_config +from shelfmark.core.logger import setup_logger +from shelfmark.core.request_helpers import get_session_db_user_id + +if TYPE_CHECKING: + from collections.abc import Callable + + from flask import Flask + from werkzeug.wrappers import Response + + from shelfmark.metadata_providers import BookMetadata, MetadataProvider + +logger = setup_logger(__name__) + +_OPERATIONAL_ERRORS = (OSError, RuntimeError, TypeError, ValueError, sqlite3.Error) +_IMPORT_OPERATIONAL_ERRORS = (ImportError, *_OPERATIONAL_ERRORS) + + +def _resolve_metadata_provider(provider_name: str) -> MetadataProvider: + """Validate, instantiate and return a ready metadata provider. + + Raises appropriate HTTP-friendly exceptions on failure. + """ + from shelfmark.metadata_providers import ( + get_provider, + get_provider_kwargs, + is_provider_registered, + ) + + if not is_provider_registered(provider_name): + msg = f"Unknown metadata provider: {provider_name}" + raise ValueError(msg) + + kwargs = get_provider_kwargs(provider_name) + prov = get_provider(provider_name, **kwargs) + + if not prov.is_available(): + msg = f"Provider '{provider_name}' is not available" + raise RuntimeError(msg) + + return prov + + +def _handle_target_errors( + fallback_message: str, +) -> Callable[ + [Callable[..., Response | tuple[Response, int]]], Callable[..., Response | tuple[Response, int]] +]: + """Wrap a metadata-target route with standard error handling.""" + + def decorator( + fn: Callable[..., Response | tuple[Response, int]], + ) -> Callable[..., Response | tuple[Response, int]]: + @wraps(fn) + def wrapper(*args: object, **kwargs: object) -> Response | tuple[Response, int]: + try: + return fn(*args, **kwargs) + except (NotImplementedError, ValueError) as e: + return jsonify({"error": str(e)}), 400 + except RuntimeError as e: + return jsonify({"error": str(e)}), 502 + except (OSError, TypeError, sqlite3.Error) as e: + logger.error_trace(f"{fallback_message}: {e}") + return jsonify({"error": fallback_message}), 500 + + return wrapper + + return decorator + + +def register_metadata_routes( + app: Flask, + *, + login_required: Callable[..., Callable[..., Response | tuple[Response, int]]], + resolve_auth_mode: Callable[[], str], + attach_library_ownership: Callable[[BookMetadata, dict[str, Any]], None], +) -> None: + """Register the ``/api/metadata/...`` routes. + + ``login_required`` is the app's decorator. ``resolve_auth_mode`` is called per request so a + runtime patch of the auth mode still takes effect. ``attach_library_ownership`` is shared + with the releases routes, which is why it stays in ``main.py``. + """ + + def _session_may_use_provider_account() -> bool: + """Whether this session may read or change the metadata provider's own account. + + A provider such as Hardcover is signed in as one account, the instance owner's, and its lists + and shelves belong to that person. With no authentication configured the one user is the owner. + Otherwise only an admin may use them. + """ + return resolve_auth_mode() == "none" or bool(session.get("is_admin", False)) + + @app.route("/api/metadata/providers", methods=["GET"]) + @login_required + def api_metadata_providers() -> Response | tuple[Response, int]: + """Get list of available metadata providers. + + Returns: + flask.Response: JSON with list of providers and their status. + + """ + try: + from shelfmark.metadata_providers import ( + get_configured_provider_name, + get_provider, + get_provider_kwargs, + list_providers, + ) + + app_config.refresh() + db_user_id = get_session_db_user_id(session) + + configured_metadata_provider = get_configured_provider_name( + content_type="ebook", + user_id=db_user_id, + fallback_to_main=True, + ) + configured_audiobook_metadata_provider = get_configured_provider_name( + content_type="audiobook", + user_id=db_user_id, + fallback_to_main=False, + ) + configured_combined_metadata_provider = get_configured_provider_name( + content_type="combined", + user_id=db_user_id, + fallback_to_main=False, + ) + providers = [] + for info in list_providers(): + enabled_key = f"{info['name'].upper()}_ENABLED" + provider_info = { + "name": info["name"], + "display_name": info["display_name"], + "requires_auth": info["requires_auth"], + "enabled": app_config.get(enabled_key, False) is True, + "available": False, + } + + try: + kwargs = get_provider_kwargs(info["name"]) + provider = get_provider(info["name"], **kwargs) + provider_info["available"] = provider.is_available() + except _OPERATIONAL_ERRORS as exc: + logger.debug( + "Metadata provider %s availability check failed: %s", + info["name"], + exc, + ) + + providers.append(provider_info) + + return jsonify( + { + "providers": providers, + "configured_provider": configured_metadata_provider or None, + "configured_provider_audiobook": configured_audiobook_metadata_provider or None, + "configured_provider_combined": configured_combined_metadata_provider or None, + } + ) + except _IMPORT_OPERATIONAL_ERRORS as e: + logger.error_trace(f"Metadata providers error: {e}") + return jsonify({"error": str(e)}), 500 + + @app.route("/api/metadata/config", methods=["GET"]) + @login_required + def api_metadata_config() -> Response | tuple[Response, int]: + """Return provider-specific metadata search config for the active session.""" + try: + from shelfmark.metadata_providers import ( + get_configured_provider_name, + get_provider, + get_provider_capabilities, + get_provider_default_sort, + get_provider_kwargs, + get_provider_search_fields, + get_provider_sort_options, + is_provider_registered, + ) + + app_config.refresh() + content_type = request.args.get("content_type", "ebook").strip() + provider_name = request.args.get("provider", "").strip() + + db_user_id = get_session_db_user_id(session) + + if not provider_name: + provider_name = get_configured_provider_name( + content_type=content_type, + user_id=db_user_id, + fallback_to_main=True, + ) + + if not provider_name: + return jsonify( + { + "provider": None, + "display_name": None, + "enabled": False, + "available": False, + "search_fields": [], + "capabilities": [], + "sort_options": [{"value": "relevance", "label": "Most relevant"}], + "default_sort": "relevance", + } + ) + + if not is_provider_registered(provider_name): + return jsonify({"error": f"Unknown metadata provider: {provider_name}"}), 400 + + kwargs = get_provider_kwargs(provider_name) + provider = get_provider(provider_name, **kwargs) + enabled_key = f"{provider_name.upper()}_ENABLED" + provider_enabled = app_config.get(enabled_key, False) is True + provider_available = provider.is_available() + + return jsonify( + { + "provider": provider_name, + "display_name": provider.display_name, + "enabled": provider_enabled, + "available": provider_available, + "search_fields": get_provider_search_fields(provider_name), + "capabilities": get_provider_capabilities(provider_name), + "sort_options": get_provider_sort_options(provider_name), + "default_sort": get_provider_default_sort(provider_name, user_id=db_user_id), + } + ) + except _IMPORT_OPERATIONAL_ERRORS as e: + logger.error_trace(f"Metadata config error: {e}") + return jsonify({"error": str(e)}), 500 + + @app.route("/api/metadata/search", methods=["GET"]) + @login_required + def api_metadata_search() -> Response | tuple[Response, int]: + """Search for books using the configured metadata provider. + + Query Parameters: + query (str): Search query (required) + limit (int): Maximum number of results (default: 40, max: 100) + sort (str): Sort order - relevance, popularity, rating, newest, oldest (default: relevance) + [dynamic fields]: Provider-specific search fields passed as query params + + Returns: + flask.Response: JSON with list of books from metadata provider. + + """ + try: + from dataclasses import asdict + + from shelfmark.metadata_providers import ( + CheckboxSearchField, + MetadataSearchOptions, + NumberSearchField, + SortOrder, + get_configured_provider, + get_provider, + get_provider_kwargs, + is_provider_enabled, + is_provider_registered, + ) + + query = request.args.get("query", "").strip() + content_type = request.args.get("content_type", "ebook").strip() + provider_name = request.args.get("provider", "").strip() + + try: + limit = min(int(request.args.get("limit", 40)), 100) + except ValueError: + limit = 40 + + try: + page = max(1, int(request.args.get("page", 1))) + except ValueError: + page = 1 + + # Parse sort parameter + sort_value = request.args.get("sort", "relevance").lower() + try: + sort_order = SortOrder(sort_value) + except ValueError: + sort_order = SortOrder.RELEVANCE + + db_user_id = get_session_db_user_id(session) + + if provider_name: + if not is_provider_registered(provider_name): + return jsonify( + { + "error": f"Unknown metadata provider: {provider_name}", + "message": f"Unknown metadata provider: {provider_name}", + } + ), 400 + if not is_provider_enabled(provider_name): + return jsonify( + { + "error": f"Metadata provider '{provider_name}' is not enabled", + "message": f"{provider_name} is not enabled. Enable it in Settings first.", + } + ), 503 + + kwargs = get_provider_kwargs(provider_name) + provider = get_provider(provider_name, **kwargs) + else: + provider = get_configured_provider(content_type=content_type, user_id=db_user_id) + + if not provider: + return jsonify( + { + "error": "No metadata provider configured", + "message": "No metadata provider configured. Enable one in Settings.", + } + ), 503 + + if not provider.is_available(): + return jsonify( + { + "error": f"Metadata provider '{provider.name}' is not available", + "message": f"{provider.display_name} is not available. Check configuration in Settings.", + } + ), 503 + + # Extract custom search field values from query params + fields: dict[str, Any] = {} + may_use_account = _session_may_use_provider_account() + for search_field in provider.search_fields: + if search_field.key in provider.account_search_fields and not may_use_account: + continue + value = request.args.get(search_field.key) + if value is not None: + # Strip string values to handle whitespace-only input + value = value.strip() + if value != "": + # Parse value based on field type + if isinstance(search_field, CheckboxSearchField): + fields[search_field.key] = value.lower() in ("true", "1", "yes", "on") + elif isinstance(search_field, NumberSearchField): + with suppress(ValueError): + fields[search_field.key] = int(value) + else: + fields[search_field.key] = value + + # Require either a query or at least one field value + if not query and not fields: + return jsonify({"error": "Either 'query' or search field values are required"}), 400 + + options = MetadataSearchOptions( + query=query, + limit=limit, + page=page, + sort=sort_order, + fields=fields, + user_id=db_user_id, + ) + search_result = provider.search_paginated(options) + + # Convert BookMetadata objects to dicts + books_data = [asdict(book) for book in search_result.books] + + # Transform cover_url to local proxy URLs when caching is enabled + from shelfmark.core.utils import transform_cover_url + + for book_dict in books_data: + if book_dict.get("cover_url"): + cache_id = f"{book_dict['provider']}_{book_dict['provider_id']}" + book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id) + for book, book_dict in zip(search_result.books, books_data, strict=True): + attach_library_ownership(book, book_dict) + + response_data = { + "books": books_data, + "provider": provider.name, + "query": query, + "page": search_result.page, + "total_found": search_result.total_found, + "has_more": search_result.has_more, + } + if search_result.source_url: + response_data["source_url"] = search_result.source_url + if search_result.source_title: + response_data["source_title"] = search_result.source_title + return jsonify(response_data) + except _IMPORT_OPERATIONAL_ERRORS as e: + logger.error_trace(f"Metadata search error: {e}") + return jsonify({"error": str(e)}), 500 + + @app.route("/api/metadata/field-options", methods=["GET"]) + @login_required + def api_metadata_field_options() -> Response: + """Return dynamic search-field options for a metadata provider.""" + try: + from shelfmark.metadata_providers import ( + get_configured_provider, + get_provider, + get_provider_kwargs, + is_provider_registered, + ) + + field_key = request.args.get("field", "").strip() + provider_name = request.args.get("provider", "").strip() + content_type = request.args.get("content_type", "ebook").strip() + query_text = request.args.get("query", "").strip() + + if not field_key: + return jsonify({"options": []}) + + db_user_id = get_session_db_user_id(session) + + provider = None + if provider_name: + if not is_provider_registered(provider_name): + return jsonify({"options": []}) + kwargs = get_provider_kwargs(provider_name) + provider = get_provider(provider_name, **kwargs) + else: + provider = get_configured_provider(content_type=content_type, user_id=db_user_id) + + if not provider or not provider.is_available(): + return jsonify({"options": []}) + + if ( + field_key in provider.account_search_fields + and not _session_may_use_provider_account() + ): + return jsonify({"options": []}) + + options = provider.get_search_field_options(field_key, query=query_text or None) + return jsonify({"options": options}) + except _OPERATIONAL_ERRORS as e: + logger.warning("Metadata field options endpoint error: %s", e) + return jsonify({"options": []}) + + @app.route("/api/metadata/book//", methods=["GET"]) + @login_required + def api_metadata_book(provider: str, book_id: str) -> Response | tuple[Response, int]: + """Get detailed book information from a metadata provider. + + Path Parameters: + provider (str): Provider name (e.g., "hardcover", "openlibrary") + book_id (str): Book ID in the provider's system + + Returns: + flask.Response: JSON with book details. + + """ + try: + from dataclasses import asdict + + prov = _resolve_metadata_provider(provider) + + book = prov.get_book(book_id) + if not book: + return jsonify({"error": "Book not found"}), 404 + + book_dict = asdict(book) + attach_library_ownership(book, book_dict) + + # Transform cover_url to local proxy URL when caching is enabled + from shelfmark.core.utils import transform_cover_url + + if book_dict.get("cover_url"): + cache_id = f"{provider}_{book_id}" + book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id) + + return jsonify(book_dict) + except ValueError as e: + return jsonify({"error": str(e)}), 400 + except RuntimeError as e: + return jsonify({"error": str(e)}), 503 + except (OSError, TypeError, sqlite3.Error) as e: + logger.error_trace(f"Metadata book error: {e}") + return jsonify({"error": str(e)}), 500 + + @app.route("/api/metadata/book///targets", methods=["GET"]) + @login_required + @_handle_target_errors("Failed to load book targets") + def api_metadata_book_targets(provider: str, book_id: str) -> Response | tuple[Response, int]: + """Get provider-managed list/status targets for a specific book.""" + prov = _resolve_metadata_provider(provider) + if not _session_may_use_provider_account(): + return jsonify({"options": []}) + return jsonify({"options": prov.get_book_targets(book_id)}) + + @app.route("/api/metadata/book//targets/batch", methods=["POST"]) + @login_required + @_handle_target_errors("Failed to load book targets") + def api_metadata_book_targets_batch(provider: str) -> Response | tuple[Response, int]: + """Get provider-managed list/status targets for multiple books.""" + prov = _resolve_metadata_provider(provider) + + payload = request.get_json(silent=True) or {} + raw_ids = payload.get("book_ids", []) if isinstance(payload, dict) else [] + if not isinstance(raw_ids, list) or not raw_ids: + return jsonify({"error": "book_ids must be a non-empty array"}), 400 + + book_ids = [str(bid) for bid in raw_ids[:50]] + + if not _session_may_use_provider_account(): + return jsonify({"results": {book_id: [] for book_id in book_ids}}) + return jsonify({"results": prov.get_book_targets_batch(book_ids)}) + + @app.route("/api/metadata/book///targets", methods=["PUT"]) + @login_required + @_handle_target_errors("Failed to update book targets") + def api_metadata_book_targets_update( + provider: str, book_id: str + ) -> Response | tuple[Response, int]: + """Set whether a book belongs to a provider-managed list or shelf.""" + if not _session_may_use_provider_account(): + return jsonify({"error": "Admin access required"}), 403 + prov = _resolve_metadata_provider(provider) + + payload = request.get_json(silent=True) or {} + target = str(payload.get("target", "")).strip() if isinstance(payload, dict) else "" + selected = payload.get("selected") if isinstance(payload, dict) else None + if not target: + return jsonify({"error": "target is required"}), 400 + if not isinstance(selected, bool): + return jsonify({"error": "selected must be a boolean"}), 400 + + result = prov.set_book_target_state(book_id, target, selected=selected) + response: dict = { + "success": True, + "changed": bool(result.get("changed", True)), + "selected": selected, + } + deselected = result.get("deselected_target") + if isinstance(deselected, str) and deselected: + response["deselected_target"] = deselected + return jsonify(response) diff --git a/shelfmark/main.py b/shelfmark/main.py index 1de6f203..eef7fa3d 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -7,7 +7,6 @@ import re import sqlite3 import time import unicodedata -from contextlib import suppress from datetime import UTC, datetime, timedelta from functools import wraps from importlib import import_module @@ -58,6 +57,7 @@ from shelfmark.core.cwa_user_sync import upsert_cwa_user from shelfmark.core.download_history_service import DownloadHistoryService from shelfmark.core.external_user_linking import upsert_external_user from shelfmark.core.logger import setup_logger +from shelfmark.core.metadata_routes import register_metadata_routes from shelfmark.core.models import ( TERMINAL_QUEUE_STATUSES, QueueStatus, @@ -118,7 +118,7 @@ from shelfmark.release_sources import ( if TYPE_CHECKING: from shelfmark.core.models import DownloadTask - from shelfmark.metadata_providers import BookMetadata, MetadataProvider + from shelfmark.metadata_providers import BookMetadata logger = setup_logger(__name__) FLASK_SECRET_KEY_MIN_BYTES = 32 @@ -2726,341 +2726,6 @@ def api_auth_check() -> Response | tuple[Response, int]: ) -@app.route("/api/metadata/providers", methods=["GET"]) -@login_required -def api_metadata_providers() -> Response | tuple[Response, int]: - """Get list of available metadata providers. - - Returns: - flask.Response: JSON with list of providers and their status. - - """ - try: - from shelfmark.metadata_providers import ( - get_configured_provider_name, - get_provider, - get_provider_kwargs, - list_providers, - ) - - app_config.refresh() - db_user_id = get_session_db_user_id(session) - - configured_metadata_provider = get_configured_provider_name( - content_type="ebook", - user_id=db_user_id, - fallback_to_main=True, - ) - configured_audiobook_metadata_provider = get_configured_provider_name( - content_type="audiobook", - user_id=db_user_id, - fallback_to_main=False, - ) - configured_combined_metadata_provider = get_configured_provider_name( - content_type="combined", - user_id=db_user_id, - fallback_to_main=False, - ) - providers = [] - for info in list_providers(): - enabled_key = f"{info['name'].upper()}_ENABLED" - provider_info = { - "name": info["name"], - "display_name": info["display_name"], - "requires_auth": info["requires_auth"], - "enabled": app_config.get(enabled_key, False) is True, - "available": False, - } - - try: - kwargs = get_provider_kwargs(info["name"]) - provider = get_provider(info["name"], **kwargs) - provider_info["available"] = provider.is_available() - except _OPERATIONAL_ERRORS as exc: - logger.debug( - "Metadata provider %s availability check failed: %s", - info["name"], - exc, - ) - - providers.append(provider_info) - - return jsonify( - { - "providers": providers, - "configured_provider": configured_metadata_provider or None, - "configured_provider_audiobook": configured_audiobook_metadata_provider or None, - "configured_provider_combined": configured_combined_metadata_provider or None, - } - ) - except _IMPORT_OPERATIONAL_ERRORS as e: - logger.error_trace(f"Metadata providers error: {e}") - return jsonify({"error": str(e)}), 500 - - -@app.route("/api/metadata/config", methods=["GET"]) -@login_required -def api_metadata_config() -> Response | tuple[Response, int]: - """Return provider-specific metadata search config for the active session.""" - try: - from shelfmark.metadata_providers import ( - get_configured_provider_name, - get_provider, - get_provider_capabilities, - get_provider_default_sort, - get_provider_kwargs, - get_provider_search_fields, - get_provider_sort_options, - is_provider_registered, - ) - - app_config.refresh() - content_type = request.args.get("content_type", "ebook").strip() - provider_name = request.args.get("provider", "").strip() - - db_user_id = get_session_db_user_id(session) - - if not provider_name: - provider_name = get_configured_provider_name( - content_type=content_type, - user_id=db_user_id, - fallback_to_main=True, - ) - - if not provider_name: - return jsonify( - { - "provider": None, - "display_name": None, - "enabled": False, - "available": False, - "search_fields": [], - "capabilities": [], - "sort_options": [{"value": "relevance", "label": "Most relevant"}], - "default_sort": "relevance", - } - ) - - if not is_provider_registered(provider_name): - return jsonify({"error": f"Unknown metadata provider: {provider_name}"}), 400 - - kwargs = get_provider_kwargs(provider_name) - provider = get_provider(provider_name, **kwargs) - enabled_key = f"{provider_name.upper()}_ENABLED" - provider_enabled = app_config.get(enabled_key, False) is True - provider_available = provider.is_available() - - return jsonify( - { - "provider": provider_name, - "display_name": provider.display_name, - "enabled": provider_enabled, - "available": provider_available, - "search_fields": get_provider_search_fields(provider_name), - "capabilities": get_provider_capabilities(provider_name), - "sort_options": get_provider_sort_options(provider_name), - "default_sort": get_provider_default_sort(provider_name, user_id=db_user_id), - } - ) - except _IMPORT_OPERATIONAL_ERRORS as e: - logger.error_trace(f"Metadata config error: {e}") - return jsonify({"error": str(e)}), 500 - - -@app.route("/api/metadata/search", methods=["GET"]) -@login_required -def api_metadata_search() -> Response | tuple[Response, int]: - """Search for books using the configured metadata provider. - - Query Parameters: - query (str): Search query (required) - limit (int): Maximum number of results (default: 40, max: 100) - sort (str): Sort order - relevance, popularity, rating, newest, oldest (default: relevance) - [dynamic fields]: Provider-specific search fields passed as query params - - Returns: - flask.Response: JSON with list of books from metadata provider. - - """ - try: - from dataclasses import asdict - - from shelfmark.metadata_providers import ( - CheckboxSearchField, - MetadataSearchOptions, - NumberSearchField, - SortOrder, - get_configured_provider, - get_provider, - get_provider_kwargs, - is_provider_enabled, - is_provider_registered, - ) - - query = request.args.get("query", "").strip() - content_type = request.args.get("content_type", "ebook").strip() - provider_name = request.args.get("provider", "").strip() - - try: - limit = min(int(request.args.get("limit", 40)), 100) - except ValueError: - limit = 40 - - try: - page = max(1, int(request.args.get("page", 1))) - except ValueError: - page = 1 - - # Parse sort parameter - sort_value = request.args.get("sort", "relevance").lower() - try: - sort_order = SortOrder(sort_value) - except ValueError: - sort_order = SortOrder.RELEVANCE - - db_user_id = get_session_db_user_id(session) - - if provider_name: - if not is_provider_registered(provider_name): - return jsonify( - { - "error": f"Unknown metadata provider: {provider_name}", - "message": f"Unknown metadata provider: {provider_name}", - } - ), 400 - if not is_provider_enabled(provider_name): - return jsonify( - { - "error": f"Metadata provider '{provider_name}' is not enabled", - "message": f"{provider_name} is not enabled. Enable it in Settings first.", - } - ), 503 - - kwargs = get_provider_kwargs(provider_name) - provider = get_provider(provider_name, **kwargs) - else: - provider = get_configured_provider(content_type=content_type, user_id=db_user_id) - - if not provider: - return jsonify( - { - "error": "No metadata provider configured", - "message": "No metadata provider configured. Enable one in Settings.", - } - ), 503 - - if not provider.is_available(): - return jsonify( - { - "error": f"Metadata provider '{provider.name}' is not available", - "message": f"{provider.display_name} is not available. Check configuration in Settings.", - } - ), 503 - - # Extract custom search field values from query params - fields: dict[str, Any] = {} - may_use_account = _session_may_use_provider_account() - for search_field in provider.search_fields: - if search_field.key in provider.account_search_fields and not may_use_account: - continue - value = request.args.get(search_field.key) - if value is not None: - # Strip string values to handle whitespace-only input - value = value.strip() - if value != "": - # Parse value based on field type - if isinstance(search_field, CheckboxSearchField): - fields[search_field.key] = value.lower() in ("true", "1", "yes", "on") - elif isinstance(search_field, NumberSearchField): - with suppress(ValueError): - fields[search_field.key] = int(value) - else: - fields[search_field.key] = value - - # Require either a query or at least one field value - if not query and not fields: - return jsonify({"error": "Either 'query' or search field values are required"}), 400 - - options = MetadataSearchOptions( - query=query, limit=limit, page=page, sort=sort_order, fields=fields, user_id=db_user_id - ) - search_result = provider.search_paginated(options) - - # Convert BookMetadata objects to dicts - books_data = [asdict(book) for book in search_result.books] - - # Transform cover_url to local proxy URLs when caching is enabled - from shelfmark.core.utils import transform_cover_url - - for book_dict in books_data: - if book_dict.get("cover_url"): - cache_id = f"{book_dict['provider']}_{book_dict['provider_id']}" - book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id) - for book, book_dict in zip(search_result.books, books_data, strict=True): - _attach_library_ownership(book, book_dict) - - response_data = { - "books": books_data, - "provider": provider.name, - "query": query, - "page": search_result.page, - "total_found": search_result.total_found, - "has_more": search_result.has_more, - } - if search_result.source_url: - response_data["source_url"] = search_result.source_url - if search_result.source_title: - response_data["source_title"] = search_result.source_title - return jsonify(response_data) - except _IMPORT_OPERATIONAL_ERRORS as e: - logger.error_trace(f"Metadata search error: {e}") - return jsonify({"error": str(e)}), 500 - - -@app.route("/api/metadata/field-options", methods=["GET"]) -@login_required -def api_metadata_field_options() -> Response: - """Return dynamic search-field options for a metadata provider.""" - try: - from shelfmark.metadata_providers import ( - get_configured_provider, - get_provider, - get_provider_kwargs, - is_provider_registered, - ) - - field_key = request.args.get("field", "").strip() - provider_name = request.args.get("provider", "").strip() - content_type = request.args.get("content_type", "ebook").strip() - query_text = request.args.get("query", "").strip() - - if not field_key: - return jsonify({"options": []}) - - db_user_id = get_session_db_user_id(session) - - provider = None - if provider_name: - if not is_provider_registered(provider_name): - return jsonify({"options": []}) - kwargs = get_provider_kwargs(provider_name) - provider = get_provider(provider_name, **kwargs) - else: - provider = get_configured_provider(content_type=content_type, user_id=db_user_id) - - if not provider or not provider.is_available(): - return jsonify({"options": []}) - - if field_key in provider.account_search_fields and not _session_may_use_provider_account(): - return jsonify({"options": []}) - - options = provider.get_search_field_options(field_key, query=query_text or None) - return jsonify({"options": options}) - except _OPERATIONAL_ERRORS as e: - logger.warning("Metadata field options endpoint error: %s", e) - return jsonify({"options": []}) - - def _attach_library_ownership(book: Any, book_dict: dict[str, Any]) -> None: """Add per-format ownership under ``library`` when a library check is enabled.""" from shelfmark.core import library_index @@ -3070,169 +2735,12 @@ def _attach_library_ownership(book: Any, book_dict: dict[str, Any]) -> None: book_dict["library"] = owned -def _session_may_use_provider_account() -> bool: - """Whether this session may read or change the metadata provider's own account. - - A provider such as Hardcover is signed in as one account, the instance owner's, and its lists - and shelves belong to that person. With no authentication configured the one user is the owner. - Otherwise only an admin may use them. - """ - return get_auth_mode() == "none" or bool(session.get("is_admin", False)) - - -def _resolve_metadata_provider(provider_name: str) -> MetadataProvider: - """Validate, instantiate and return a ready metadata provider. - - Raises appropriate HTTP-friendly exceptions on failure. - """ - from shelfmark.metadata_providers import ( - get_provider, - get_provider_kwargs, - is_provider_registered, - ) - - if not is_provider_registered(provider_name): - msg = f"Unknown metadata provider: {provider_name}" - raise ValueError(msg) - - kwargs = get_provider_kwargs(provider_name) - prov = get_provider(provider_name, **kwargs) - - if not prov.is_available(): - msg = f"Provider '{provider_name}' is not available" - raise RuntimeError(msg) - - return prov - - -@app.route("/api/metadata/book//", methods=["GET"]) -@login_required -def api_metadata_book(provider: str, book_id: str) -> Response | tuple[Response, int]: - """Get detailed book information from a metadata provider. - - Path Parameters: - provider (str): Provider name (e.g., "hardcover", "openlibrary") - book_id (str): Book ID in the provider's system - - Returns: - flask.Response: JSON with book details. - - """ - try: - from dataclasses import asdict - - prov = _resolve_metadata_provider(provider) - - book = prov.get_book(book_id) - if not book: - return jsonify({"error": "Book not found"}), 404 - - book_dict = asdict(book) - _attach_library_ownership(book, book_dict) - - # Transform cover_url to local proxy URL when caching is enabled - from shelfmark.core.utils import transform_cover_url - - if book_dict.get("cover_url"): - cache_id = f"{provider}_{book_id}" - book_dict["cover_url"] = transform_cover_url(book_dict["cover_url"], cache_id) - - return jsonify(book_dict) - except ValueError as e: - return jsonify({"error": str(e)}), 400 - except RuntimeError as e: - return jsonify({"error": str(e)}), 503 - except (OSError, TypeError, sqlite3.Error) as e: - logger.error_trace(f"Metadata book error: {e}") - return jsonify({"error": str(e)}), 500 - - -def _handle_target_errors( - fallback_message: str, -) -> Callable[ - [Callable[..., Response | tuple[Response, int]]], Callable[..., Response | tuple[Response, int]] -]: - """Wrap a metadata-target route with standard error handling.""" - - def decorator( - fn: Callable[..., Response | tuple[Response, int]], - ) -> Callable[..., Response | tuple[Response, int]]: - @wraps(fn) - def wrapper(*args: object, **kwargs: object) -> Response | tuple[Response, int]: - try: - return fn(*args, **kwargs) - except (NotImplementedError, ValueError) as e: - return jsonify({"error": str(e)}), 400 - except RuntimeError as e: - return jsonify({"error": str(e)}), 502 - except (OSError, TypeError, sqlite3.Error) as e: - logger.error_trace(f"{fallback_message}: {e}") - return jsonify({"error": fallback_message}), 500 - - return wrapper - - return decorator - - -@app.route("/api/metadata/book///targets", methods=["GET"]) -@login_required -@_handle_target_errors("Failed to load book targets") -def api_metadata_book_targets(provider: str, book_id: str) -> Response | tuple[Response, int]: - """Get provider-managed list/status targets for a specific book.""" - prov = _resolve_metadata_provider(provider) - if not _session_may_use_provider_account(): - return jsonify({"options": []}) - return jsonify({"options": prov.get_book_targets(book_id)}) - - -@app.route("/api/metadata/book//targets/batch", methods=["POST"]) -@login_required -@_handle_target_errors("Failed to load book targets") -def api_metadata_book_targets_batch(provider: str) -> Response | tuple[Response, int]: - """Get provider-managed list/status targets for multiple books.""" - prov = _resolve_metadata_provider(provider) - - payload = request.get_json(silent=True) or {} - raw_ids = payload.get("book_ids", []) if isinstance(payload, dict) else [] - if not isinstance(raw_ids, list) or not raw_ids: - return jsonify({"error": "book_ids must be a non-empty array"}), 400 - - book_ids = [str(bid) for bid in raw_ids[:50]] - - if not _session_may_use_provider_account(): - return jsonify({"results": {book_id: [] for book_id in book_ids}}) - return jsonify({"results": prov.get_book_targets_batch(book_ids)}) - - -@app.route("/api/metadata/book///targets", methods=["PUT"]) -@login_required -@_handle_target_errors("Failed to update book targets") -def api_metadata_book_targets_update( - provider: str, book_id: str -) -> Response | tuple[Response, int]: - """Set whether a book belongs to a provider-managed list or shelf.""" - if not _session_may_use_provider_account(): - return jsonify({"error": "Admin access required"}), 403 - prov = _resolve_metadata_provider(provider) - - payload = request.get_json(silent=True) or {} - target = str(payload.get("target", "")).strip() if isinstance(payload, dict) else "" - selected = payload.get("selected") if isinstance(payload, dict) else None - if not target: - return jsonify({"error": "target is required"}), 400 - if not isinstance(selected, bool): - return jsonify({"error": "selected must be a boolean"}), 400 - - result = prov.set_book_target_state(book_id, target, selected=selected) - response: dict = { - "success": True, - "changed": bool(result.get("changed", True)), - "selected": selected, - } - deselected = result.get("deselected_target") - if isinstance(deselected, str) and deselected: - response["deselected_target"] = deselected - return jsonify(response) +register_metadata_routes( + app, + login_required=login_required, + resolve_auth_mode=_resolve_auth_mode_for_routes, + attach_library_ownership=_attach_library_ownership, +) @app.route("/api/releases", methods=["GET"]) diff --git a/tests/core/test_metadata_routes.py b/tests/core/test_metadata_routes.py new file mode 100644 index 00000000..dc771f30 --- /dev/null +++ b/tests/core/test_metadata_routes.py @@ -0,0 +1,348 @@ +"""Behaviour of the `/api/metadata/...` routes that no other test pins down.""" + +from __future__ import annotations + +import importlib +from unittest.mock import patch + +import pytest + +from shelfmark.metadata_providers import BookMetadata + + +@pytest.fixture(scope="module") +def main_module(): + """Import `shelfmark.main` with background startup disabled.""" + with patch("shelfmark.download.orchestrator.start"): + import shelfmark.main as main + + importlib.reload(main) + return main + + +class _Provider: + name = "stub" + display_name = "Stub" + search_fields: list = [] + account_search_fields: frozenset = frozenset() + + def __init__(self) -> None: + self.available = True + self.book: BookMetadata | None = BookMetadata( + provider="stub", provider_id="7", title="Dune", authors=["Frank Herbert"] + ) + self.raise_on_targets: Exception | None = None + + def is_available(self) -> bool: + return self.available + + def get_book(self, book_id: str) -> BookMetadata | None: + return self.book + + def get_book_targets(self, book_id: str) -> list: + if self.raise_on_targets: + raise self.raise_on_targets + return [{"value": "list:1"}] + + def get_book_targets_batch(self, ids: list) -> dict: + return {i: [] for i in ids} + + def set_book_target_state(self, book_id: str, target: str, *, selected: bool) -> dict: + return {"changed": False, "deselected_target": "list:2"} + + +@pytest.fixture +def stub() -> _Provider: + return _Provider() + + +@pytest.fixture +def client(main_module, stub): + registered = {"stub": True} + with ( + patch("shelfmark.metadata_providers.get_provider", return_value=stub), + patch( + "shelfmark.metadata_providers.is_provider_registered", + side_effect=lambda name: registered.get(name, False), + ), + patch("shelfmark.metadata_providers.get_provider_kwargs", return_value={}), + patch.object(main_module, "get_auth_mode", return_value="none"), + ): + yield main_module.app.test_client() + + +class TestAuth: + @pytest.mark.parametrize( + ("method", "path"), + [ + ("get", "/api/metadata/providers"), + ("get", "/api/metadata/config"), + ("get", "/api/metadata/search?query=x"), + ("get", "/api/metadata/field-options?field=a"), + ("get", "/api/metadata/book/stub/7"), + ("get", "/api/metadata/book/stub/7/targets"), + ("post", "/api/metadata/book/stub/targets/batch"), + ("put", "/api/metadata/book/stub/7/targets"), + ], + ) + def test_every_route_needs_a_session(self, main_module, method, path) -> None: + client = main_module.app.test_client() + with patch.object(main_module, "get_auth_mode", return_value="builtin"): + resp = getattr(client, method)(path) + + assert resp.status_code == 401 + + +class TestProviders: + def test_lists_providers_with_availability_and_configured_names(self, client, stub) -> None: + stub.available = False + info = {"name": "stub", "display_name": "Stub", "requires_auth": True} + with ( + patch("shelfmark.metadata_providers.list_providers", return_value=[info]), + patch( + "shelfmark.metadata_providers.get_configured_provider_name", + side_effect=lambda content_type, **_: {"ebook": "stub", "audiobook": ""}.get( + content_type, "" + ), + ), + ): + resp = client.get("/api/metadata/providers") + + assert resp.status_code == 200 + assert resp.json == { + "providers": [ + { + "name": "stub", + "display_name": "Stub", + "requires_auth": True, + "enabled": False, + "available": False, + } + ], + "configured_provider": "stub", + "configured_provider_audiobook": None, + "configured_provider_combined": None, + } + + def test_a_provider_that_fails_its_check_is_listed_unavailable(self, client) -> None: + info = {"name": "stub", "display_name": "Stub", "requires_auth": False} + with ( + patch("shelfmark.metadata_providers.list_providers", return_value=[info]), + patch("shelfmark.metadata_providers.get_provider", side_effect=RuntimeError("boom")), + patch("shelfmark.metadata_providers.get_configured_provider_name", return_value=""), + ): + resp = client.get("/api/metadata/providers") + + assert resp.status_code == 200 + assert resp.json["providers"][0]["available"] is False + + +class TestConfig: + def test_no_provider_configured_returns_the_empty_shape(self, client) -> None: + with patch("shelfmark.metadata_providers.get_configured_provider_name", return_value=""): + resp = client.get("/api/metadata/config") + + assert resp.status_code == 200 + assert resp.json["provider"] is None + assert resp.json["search_fields"] == [] + assert resp.json["default_sort"] == "relevance" + + def test_unknown_provider_is_a_400(self, client) -> None: + resp = client.get("/api/metadata/config?provider=nope") + + assert resp.status_code == 400 + assert resp.json == {"error": "Unknown metadata provider: nope"} + + def test_known_provider_returns_its_search_config(self, client) -> None: + with ( + patch("shelfmark.metadata_providers.get_provider_search_fields", return_value=[1]), + patch("shelfmark.metadata_providers.get_provider_capabilities", return_value=["c"]), + patch("shelfmark.metadata_providers.get_provider_sort_options", return_value=[2]), + patch("shelfmark.metadata_providers.get_provider_default_sort", return_value="newest"), + ): + resp = client.get("/api/metadata/config?provider=stub") + + assert resp.json == { + "provider": "stub", + "display_name": "Stub", + "enabled": False, + "available": True, + "search_fields": [1], + "capabilities": ["c"], + "sort_options": [2], + "default_sort": "newest", + } + + +class TestSearchErrors: + def test_unknown_provider_is_a_400(self, client) -> None: + resp = client.get("/api/metadata/search?provider=nope&query=x") + + assert resp.status_code == 400 + assert resp.json["error"] == "Unknown metadata provider: nope" + + def test_disabled_provider_is_a_503(self, client) -> None: + with patch("shelfmark.metadata_providers.is_provider_enabled", return_value=False): + resp = client.get("/api/metadata/search?provider=stub&query=x") + + assert resp.status_code == 503 + assert resp.json["error"] == "Metadata provider 'stub' is not enabled" + + def test_no_configured_provider_is_a_503(self, client) -> None: + with patch("shelfmark.metadata_providers.get_configured_provider", return_value=None): + resp = client.get("/api/metadata/search?query=x") + + assert resp.status_code == 503 + assert resp.json["error"] == "No metadata provider configured" + + def test_unavailable_provider_is_a_503(self, client, stub) -> None: + stub.available = False + with patch("shelfmark.metadata_providers.is_provider_enabled", return_value=True): + resp = client.get("/api/metadata/search?provider=stub&query=x") + + assert resp.status_code == 503 + assert resp.json["error"] == "Metadata provider 'stub' is not available" + + def test_no_query_and_no_fields_is_a_400(self, client) -> None: + with patch("shelfmark.metadata_providers.is_provider_enabled", return_value=True): + resp = client.get("/api/metadata/search?provider=stub") + + assert resp.status_code == 400 + assert resp.json == {"error": "Either 'query' or search field values are required"} + + +class TestFieldOptions: + def test_no_field_gives_no_options(self, client) -> None: + assert client.get("/api/metadata/field-options").json == {"options": []} + + def test_unknown_provider_gives_no_options(self, client) -> None: + resp = client.get("/api/metadata/field-options?provider=nope&field=a") + + assert resp.json == {"options": []} + + def test_unavailable_provider_gives_no_options(self, client, stub) -> None: + stub.available = False + + assert client.get("/api/metadata/field-options?provider=stub&field=a").json == { + "options": [] + } + + +class TestBook: + def test_returns_the_book(self, client) -> None: + resp = client.get("/api/metadata/book/stub/7") + + assert resp.status_code == 200 + assert resp.json["title"] == "Dune" + assert resp.json["provider_id"] == "7" + + def test_missing_book_is_a_404(self, client, stub) -> None: + stub.book = None + + resp = client.get("/api/metadata/book/stub/7") + + assert resp.status_code == 404 + assert resp.json == {"error": "Book not found"} + + def test_unknown_provider_is_a_400(self, client) -> None: + resp = client.get("/api/metadata/book/nope/7") + + assert resp.status_code == 400 + assert resp.json == {"error": "Unknown metadata provider: nope"} + + def test_unavailable_provider_is_a_503(self, client, stub) -> None: + stub.available = False + + resp = client.get("/api/metadata/book/stub/7") + + assert resp.status_code == 503 + assert resp.json == {"error": "Provider 'stub' is not available"} + + def test_a_library_ownership_check_adds_the_library_key(self, client) -> None: + with patch("shelfmark.core.library_index.ownership", return_value={"epub": True}): + resp = client.get("/api/metadata/book/stub/7") + + assert resp.json["library"] == {"epub": True} + + +class TestTargetErrors: + def test_unknown_provider_is_a_400(self, client) -> None: + resp = client.get("/api/metadata/book/nope/7/targets") + + assert resp.status_code == 400 + + def test_unavailable_provider_is_a_502(self, client, stub) -> None: + stub.available = False + + resp = client.get("/api/metadata/book/stub/7/targets") + + assert resp.status_code == 502 + assert resp.json == {"error": "Provider 'stub' is not available"} + + def test_not_implemented_is_a_400(self, client, stub) -> None: + stub.raise_on_targets = NotImplementedError("no targets here") + + resp = client.get("/api/metadata/book/stub/7/targets") + + assert resp.status_code == 400 + assert resp.json == {"error": "no targets here"} + + def test_an_operational_error_gets_the_fallback_message(self, client, stub) -> None: + stub.raise_on_targets = OSError("secret detail") + + resp = client.get("/api/metadata/book/stub/7/targets") + + assert resp.status_code == 500 + assert resp.json == {"error": "Failed to load book targets"} + + def test_the_update_route_uses_its_own_fallback_message(self, client, stub) -> None: + with patch.object(stub, "set_book_target_state", side_effect=TypeError("x")): + resp = client.put( + "/api/metadata/book/stub/7/targets", json={"target": "list:1", "selected": True} + ) + + assert resp.status_code == 500 + assert resp.json == {"error": "Failed to update book targets"} + + +class TestTargetPayloads: + @pytest.mark.parametrize("body", [{}, {"book_ids": []}, {"book_ids": "x"}, []]) + def test_batch_needs_a_non_empty_list(self, client, body) -> None: + resp = client.post("/api/metadata/book/stub/targets/batch", json=body) + + assert resp.status_code == 400 + assert resp.json == {"error": "book_ids must be a non-empty array"} + + def test_batch_caps_the_ids_at_fifty_and_stringifies_them(self, client) -> None: + resp = client.post( + "/api/metadata/book/stub/targets/batch", json={"book_ids": list(range(60))} + ) + + assert resp.status_code == 200 + assert set(resp.json["results"]) == {str(i) for i in range(50)} + + def test_update_needs_a_target(self, client) -> None: + resp = client.put("/api/metadata/book/stub/7/targets", json={"selected": True}) + + assert resp.status_code == 400 + assert resp.json == {"error": "target is required"} + + def test_update_needs_a_boolean_selected(self, client) -> None: + resp = client.put( + "/api/metadata/book/stub/7/targets", json={"target": "list:1", "selected": "yes"} + ) + + assert resp.status_code == 400 + assert resp.json == {"error": "selected must be a boolean"} + + def test_update_reports_the_change_and_the_deselected_target(self, client) -> None: + resp = client.put( + "/api/metadata/book/stub/7/targets", json={"target": "list:1", "selected": True} + ) + + assert resp.json == { + "success": True, + "changed": False, + "selected": True, + "deselected_target": "list:2", + }