From a1950e977fa358f07f7fdbef394b8588d90faaa1 Mon Sep 17 00:00:00 2001 From: codebude Date: Wed, 14 Oct 2026 20:41:29 +0200 Subject: [PATCH 1/5] Add BnF, Inventaire and Library of Congress import sources to backend --- .env.example | 23 + backend/app/config.py | 6 + backend/app/main.py | 24 + backend/app/routers/import_.py | 5 + backend/app/schemas.py | 2 +- backend/app/services/book_import.py | 1168 ++++++++++++++++++++++++--- backend/tests/test_book_import.py | 619 +++++++++++++- backend/tests/test_import.py | 40 + backend/tests/test_main.py | 50 ++ 9 files changed, 1841 insertions(+), 96 deletions(-) diff --git a/.env.example b/.env.example index bc6123dc..7b4b3fad 100644 --- a/.env.example +++ b/.env.example @@ -28,6 +28,29 @@ DASHBOARD_QUOTE_CACHE_TTL=86400 # in seconds; default 24h # Frontend settings (build-time) PUBLIC_DEFAULT_LOCALE=en # UI default locale: en | de +# Book import search providers +# At least one source must stay enabled. If every *_IMPORT_SEARCH_ENABLED is false the +# backend refuses to start and names the variables in the error. +# Setting GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED=false also disables the on-demand +# "Search Google Books too" button. +OPEN_LIBRARY_IMPORT_SEARCH_ENABLED=true +HARDCOVER_IMPORT_SEARCH_ENABLED=true +GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED=true + +# BnF (Bibliotheque nationale de France) - no API key needed, HTTPS +BNF_IMPORT_SEARCH_ENABLED=true + +# Inventaire.io - no API key needed, HTTPS, ISBN searches only +INVENTAIRE_IMPORT_SEARCH_ENABLED=true + +# Library of Congress - no API key needed +# WARNING: This source is reached over PLAIN HTTP (unencrypted) on port 210; the LC +# offers no HTTPS for its SRU endpoint. Book queries and responses travel unencrypted +# and can be observed or tampered with on the network path. Only public bibliographic +# data is transferred. Set this to false if unencrypted egress to port 210 is not +# acceptable in your environment. All other sources use HTTPS. +LIBRARY_OF_CONGRESS_IMPORT_SEARCH_ENABLED=false + # Cover scraper settings THALIA_COVER_SEARCH_ENABLED=false # For research only. Do not use in production without understanding the legal implications of scraping. diff --git a/backend/app/config.py b/backend/app/config.py index 9876a1d6..0cb3d34e 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -41,6 +41,12 @@ class Settings(BaseSettings): cover_import_timeout_seconds: int = 15 hardcover_app_api_token: str = "" thalia_cover_search_enabled: bool = False + open_library_import_search_enabled: bool = True + hardcover_import_search_enabled: bool = True + google_books_import_search_enabled: bool = True + bnf_import_search_enabled: bool = True + inventaire_import_search_enabled: bool = True + library_of_congress_import_search_enabled: bool = True embed_enabled: bool = True mail_server: str = "" mail_port: int = 587 diff --git a/backend/app/main.py b/backend/app/main.py index 432219eb..62449b61 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -76,9 +76,33 @@ async def _telemetry_heartbeat() -> None: await asyncio.sleep(_TELEMETRY_INTERVAL_SECONDS) +def _validate_import_sources() -> None: + """Ensure at least one import search provider is enabled. + + The list of providers is taken from the book import registry, so a newly added + source cannot silently escape this check. + """ + from app.services import book_import + + active = [spec.stage for spec in book_import._build_sources("") if spec.enabled()] + if book_import._google_fallback_enabled(): + active.append("google_books") + + if not active: + variables = ", ".join( + sorted(f"{spec.stage.upper()}_IMPORT_SEARCH_ENABLED" for spec in book_import._build_sources("")) + ) + raise RuntimeError( + "At least one import search provider must be enabled. " + f"Set at least one of {variables}, or " + "GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED, to true." + ) + + @asynccontextmanager async def lifespan(app: FastAPI): """Application lifespan: create required directories and start background tasks.""" + _validate_import_sources() Path(settings.covers_dir).mkdir(parents=True, exist_ok=True) Path(settings.import_temp_dir).mkdir(parents=True, exist_ok=True) cleanup_temp_files() diff --git a/backend/app/routers/import_.py b/backend/app/routers/import_.py index 8df096cd..15856d1f 100644 --- a/backend/app/routers/import_.py +++ b/backend/app/routers/import_.py @@ -17,6 +17,7 @@ from app.schemas import BookImportCandidate, BookImportRequest, BookRead from app.services import book_import from app.services.authors import resolve_authors_payload, sync_book_authors +from app.services.cover_import import is_safe_cover_import_url from app.services.cover_storage import download_cover from app.services.tags import build_book_read, sync_book_tags @@ -134,6 +135,10 @@ async def import_book( # Attempt to download and cache the cover locally; fall back to external URL. cover_url = c.cover_url + if cover_url and not is_safe_cover_import_url(cover_url): + # The candidate is user-submitted, so its cover URL is untrusted input. + logger.warning("Rejecting cover URL for unsafe address: %s", cover_url) + cover_url = None if cover_url: async with httpx.AsyncClient(timeout=15.0, follow_redirects=True) as client: filename = await download_cover(cover_url, settings.covers_dir, client, current_user.id) diff --git a/backend/app/schemas.py b/backend/app/schemas.py index 6ebe2e57..51431797 100644 --- a/backend/app/schemas.py +++ b/backend/app/schemas.py @@ -116,7 +116,7 @@ class BookImportCandidate(SQLModel): language: Optional[str] = None tags: Optional[str] = None blurb: Optional[str] = None - source: str # "open_library" | "google_books" | "hardcover" + source: str # "open_library" | "google_books" | "hardcover" | "bnf" | "inventaire" | "library_of_congress" class CoverCandidate(SQLModel): diff --git a/backend/app/services/book_import.py b/backend/app/services/book_import.py index fbbe5082..062ba965 100644 --- a/backend/app/services/book_import.py +++ b/backend/app/services/book_import.py @@ -3,20 +3,30 @@ Search priority: 1. Open Library (no API key required) - 2. Google Books (optional API key, used as fallback) + 2. BnF (no API key required) + 3. Inventaire (no API key required, ISBN-only) + 4. Library of Congress (no API key required) + 5. Hardcover (API token required, skipped if not configured) + 6. Google Books (optional API key, used as fallback when all above return empty) -Both return a list of BookImportCandidate objects normalised to the same schema. +All return a list of BookImportCandidate objects normalised to the same schema. """ import asyncio import logging import random +import re +import xml.etree.ElementTree as ET from collections.abc import AsyncGenerator -from typing import Literal, Optional +from dataclasses import dataclass +from typing import Awaitable, Callable, Literal, Optional import httpx import pycountry +from defusedxml import ElementTree as SafeET +from defusedxml.common import DefusedXmlException +from app.config import settings from app.schemas import BookImportCandidate from app.services.cover_import import is_safe_cover_import_url from app.services.isbn_utils import normalize_isbn @@ -26,6 +36,17 @@ OPEN_LIBRARY_SEARCH_URL: str = "https://openlibrary.org/search.json" OPEN_LIBRARY_COVER_URL: str = "https://covers.openlibrary.org/b/id/{cover_id}-L.jpg" +BNF_SRU_URL: str = "https://catalogue.bnf.fr/api/SRU" +INVENTAIRE_ENTITIES_URL: str = "https://inventaire.io/api/entities/by-uris" +LOC_SRU_URL: str = "http://lx2.loc.gov:210/LCDB" + +INVENTAIRE_COVER_BASE_URL: str = "https://inventaire.io/img/entities" + +# The LoC SRU server listens on plain HTTP port 210, which is frequently blocked or +# silently dropped by egress firewalls. Give it a shorter budget than the shared +# client so a blocked LoC cannot hold up the whole search. +_LOC_TIMEOUT_SECONDS: float = 5.0 + _USER_AGENT: str = "LibrisLog/1.0 (book import; +https://github.com/codebude/librislog)" GOOGLE_BOOKS_SEARCH_URL: str = "https://www.googleapis.com/books/v1/volumes" @@ -79,7 +100,13 @@ class SourceBackendError(Exception): """Raised when an external book source returns an error response.""" - def __init__(self, source: Literal["open_library", "google_books"], status_code: int | None = None) -> None: + def __init__( + self, + source: Literal[ + "open_library", "google_books", "hardcover", "bnf", "inventaire", "library_of_congress" + ], + status_code: int | None = None, + ) -> None: self.source = source self.status_code = status_code super().__init__(f"{source} backend error") @@ -94,6 +121,138 @@ def _truncate_api_key(api_key: str) -> str: return f"{api_key[:4]}...{api_key[-4:]}" +# ── Source registry ─────────────────────────────────────────────────────────── + + +def _ol_enabled() -> bool: + return settings.open_library_import_search_enabled + + +def _bnf_enabled() -> bool: + return settings.bnf_import_search_enabled + + +def _inventaire_enabled() -> bool: + return settings.inventaire_import_search_enabled + + +def _loc_enabled() -> bool: + return settings.library_of_congress_import_search_enabled + + +def _hc_enabled() -> bool: + return settings.hardcover_import_search_enabled + + +def _google_fallback_enabled() -> bool: + """Return True when the Google Books fallback may run.""" + return settings.google_books_import_search_enabled + + +def _skip_reason(stage: str, search_type: str, hardcover_api_token: str) -> str | None: + """Return why a source cannot run for this query, or None when it can. + + A skipped source is reported once and never queried, so the UI does not claim a + search ran when nothing was sent. + """ + if stage == "hardcover" and not hardcover_api_token.strip(): + return "no_api_token" + if stage == "inventaire" and search_type != "isbn": + # Inventaire exposes no usable title-search route. + return "title_search_unsupported" + return None + + +@dataclass(frozen=True) +class SourceSpec: + """Configuration for a single book import search provider.""" + + stage: str + display_name: str + enabled: Callable[[], bool] + search: Callable[[str, str, httpx.AsyncClient], Awaitable[list[BookImportCandidate]]] + + +def _build_sources(hardcover_api_token: str) -> tuple[SourceSpec, ...]: + """Return the parallel search sources in result-order. + + Hardcover is always listed so the UI can report it as skipped when no token is + configured; its dispatcher is bound to the token passed by the caller rather than + to settings, so an explicit token argument keeps working exactly as before. + """ + return ( + SourceSpec( + stage="open_library", + display_name="Open Library", + enabled=_ol_enabled, + search=_dispatch_open_library, + ), + SourceSpec( + stage="bnf", + display_name="BnF", + enabled=_bnf_enabled, + search=_dispatch_bnf, + ), + SourceSpec( + stage="inventaire", + display_name="Inventaire", + enabled=_inventaire_enabled, + search=_dispatch_inventaire, + ), + SourceSpec( + stage="library_of_congress", + display_name="Library of Congress", + enabled=_loc_enabled, + search=_dispatch_loc, + ), + SourceSpec( + stage="hardcover", + display_name="Hardcover", + enabled=_hc_enabled, + search=_make_hardcover_dispatcher(hardcover_api_token), + ), + ) + + +def _make_hardcover_dispatcher( + api_token: str, +) -> Callable[[str, str, httpx.AsyncClient], Awaitable[list[BookImportCandidate]]]: + """Build a Hardcover search callable bound to ``api_token``.""" + + async def _dispatch(query: str, search_type: str, client: httpx.AsyncClient) -> list[BookImportCandidate]: + if not api_token.strip(): + return [] + return await _search_hardcover(query, search_type, api_token, client) + + return _dispatch + + +# Indirection through module-level wrappers so tests can monkeypatch the +# underlying search functions (see tests/test_book_import.py). +async def _dispatch_open_library( + query: str, search_type: str, client: httpx.AsyncClient +) -> list[BookImportCandidate]: + return await _search_open_library(query, search_type, client) + + +async def _dispatch_bnf( + query: str, search_type: str, client: httpx.AsyncClient +) -> list[BookImportCandidate]: + return await _search_bnf(query, search_type, client) + + +async def _dispatch_inventaire( + query: str, search_type: str, client: httpx.AsyncClient +) -> list[BookImportCandidate]: + return await _search_inventaire(query, search_type, client) + + +async def _dispatch_loc( + query: str, search_type: str, client: httpx.AsyncClient +) -> list[BookImportCandidate]: + return await _search_loc(query, search_type, client) + + # ── Public API ──────────────────────────────────────────────────────────────── @@ -109,26 +268,16 @@ async def search_with_progress( """Search external book sources and yield progress events as dicts. Event shapes (stage / extra fields): - open_library / status=searching - open_library / status=done, count=int - open_library / status=error, reason=str - hardcover / status=searching - hardcover / status=done, count=int - hardcover / status=skipped, reason=str - hardcover / status=error, reason=str - google_books / status=searching - google_books / status=done, count=int - google_books / status=skipped, reason=str - google_books / status=error, reason=str - complete / results=list[dict] - error / message=str + All sources / status=searching, done (count), error (reason), skipped (reason) + complete / results=list[dict] + error / message=str Args: query: Search query string. search_type: ``"title"`` or ``"isbn"``. api_key: Google Books API key (optional). - hardcover_api_token: Hardcover.app API token (optional). - mode: ``"auto"`` (OL + HC, fallback to GB) or ``"google_only"``. + hardcover_api_token: Hardcover.app API token (optional, legacy). + mode: ``"auto"`` (all enabled sources, fallback to GB) or ``"google_only"``. http_client: Reusable httpx client (created internally if omitted). Yields: @@ -149,7 +298,14 @@ async def search_with_progress( results: list[BookImportCandidate] = [] if mode == "google_only": - if not api_key: + if not _google_fallback_enabled(): + logger.warning( + "Google-only search requested for %r but the Google Books source " + "is disabled (GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED=false)", + query, + ) + yield {"stage": "google_books", "status": "skipped", "reason": "disabled"} + elif not api_key: logger.warning("Google-only search requested for %r but API key is not set", query) yield {"stage": "google_books", "status": "skipped", "reason": "no_api_key"} else: @@ -167,52 +323,78 @@ async def search_with_progress( logger.warning("Google Books backend error for %r: status=%s", query, exc.status_code) yield {"stage": "google_books", "status": "error", "reason": "backend_error"} else: - ol_events: list[dict] = [] - hc_events: list[dict] = [] - ol_results: list[BookImportCandidate] = [] - hc_results: list[BookImportCandidate] = [] - - yield {"stage": "open_library", "status": "searching"} - if hardcover_api_token.strip(): - yield {"stage": "hardcover", "status": "searching"} - - async def _run_ol() -> None: - nonlocal ol_results, ol_events - try: - r = await _search_open_library(query, search_type, client) - ol_results = r - ol_events.append({"stage": "open_library", "status": "done", "count": len(r)}) - except SourceBackendError as exc: - ol_events.append({"stage": "open_library", "status": "error", "reason": "backend_error"}) - - async def _run_hc() -> None: - nonlocal hc_results, hc_events - if not hardcover_api_token.strip(): - hc_events.append({"stage": "hardcover", "status": "skipped", "reason": "no_api_token"}) - return - try: - r = await _search_hardcover(query, search_type, hardcover_api_token, client) - hc_results = r - hc_events.append({"stage": "hardcover", "status": "done", "count": len(r)}) - except Exception: - logger.exception("hardcover search error") - hc_events.append({"stage": "hardcover", "status": "error", "reason": "backend_error"}) - - await asyncio.gather(_run_ol(), _run_hc()) - - for e in ol_events: - yield e - for e in hc_events: - yield e - - results = _merge_results(ol_results, hc_results) + enabled_sources = [s for s in _build_sources(hardcover_api_token) if s.enabled()] + + if enabled_sources: + all_events: dict[str, list[dict]] = {s.stage: [] for s in enabled_sources} + all_results: dict[str, list[BookImportCandidate]] = {s.stage: [] for s in enabled_sources} + + # A source that cannot run for this query gets exactly one terminal + # `skipped` event and is excluded from the gather, so it never also + # reports a result count. + runnable: list[SourceSpec] = [] + for s in enabled_sources: + reason = _skip_reason(s.stage, search_type, hardcover_api_token) + if reason is not None: + all_events[s.stage].append( + {"stage": s.stage, "status": "skipped", "reason": reason} + ) + else: + all_events[s.stage].append({"stage": s.stage, "status": "searching"}) + runnable.append(s) + + async def _run_source(spec: SourceSpec) -> None: + try: + r = await spec.search(query, search_type, client) + all_results[spec.stage] = r + all_events[spec.stage].append( + {"stage": spec.stage, "status": "done", "count": len(r)} + ) + except SourceBackendError as exc: + logger.warning( + "%s backend error for %r: status=%s", spec.stage, query, exc.status_code + ) + all_events[spec.stage].append( + {"stage": spec.stage, "status": "error", "reason": "backend_error"} + ) + except Exception: + logger.exception("%s search error", spec.stage) + all_events[spec.stage].append( + {"stage": spec.stage, "status": "error", "reason": "backend_error"} + ) + + await asyncio.gather(*[_run_source(s) for s in runnable]) + + for s in enabled_sources: + for e in all_events[s.stage]: + yield e + + merged: list[BookImportCandidate] = [] + for s in runnable: + merged = _merge_results(merged, all_results[s.stage]) + results = merged + else: + # Every parallel source is switched off. Google Books can still answer + # as the fallback, which is the only remaining provider. + logger.warning( + "No parallel import search provider is enabled for %r — " + "only the Google Books fallback is available", + query, + ) if not results: - if not api_key: + if not _google_fallback_enabled(): + logger.warning( + "No results for %r and the Google Books fallback is disabled " + "(GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED=false)", + query, + ) + yield {"stage": "google_books", "status": "skipped", "reason": "disabled"} + elif not api_key: logger.warning("No results for %r and GOOGLE_BOOKS_API_KEY is not set", query) yield {"stage": "google_books", "status": "skipped", "reason": "no_api_key"} else: - logger.info("No OL/HC results — falling back to Google Books") + logger.info("No results from any source — falling back to Google Books") logger.debug( "Google Books invocation — mode=%s query=%r search_type=%r api_key=%s", mode, query, search_type, _truncate_api_key(api_key), @@ -244,17 +426,17 @@ async def search( hardcover_api_token: str = "", http_client: Optional[httpx.AsyncClient] = None, ) -> list[BookImportCandidate]: - """Search Open Library in parallel with Hardcover; fall back to Google Books if both empty. + """Search all enabled external book sources; fall back to Google Books if all return empty. Args: query: Search query string. search_type: ``"title"`` or ``"isbn"``. api_key: Google Books API key (optional). - hardcover_api_token: Hardcover.app API token (optional). + hardcover_api_token: Hardcover.app API token (optional, legacy). http_client: Reusable httpx client (created internally if omitted). Returns: - A deduplicated list of BookImportCandidate objects. + A list of BookImportCandidate objects. """ logger.debug("search() called — query=%r search_type=%r has_api_key=%s has_hc_token=%s", query, search_type, bool(api_key), bool(hardcover_api_token)) @@ -266,32 +448,48 @@ async def search( ) try: - tasks = [_search_open_library(query, search_type, client)] - if hardcover_api_token.strip(): - tasks.append(_search_hardcover(query, search_type, hardcover_api_token, client)) - - results_list = await asyncio.gather(*tasks, return_exceptions=True) - - ol_results = results_list[0] if not isinstance(results_list[0], BaseException) else [] - if isinstance(results_list[0], SourceBackendError): - logger.warning("Open Library backend error for %r: status=%s", query, results_list[0].status_code) - elif isinstance(results_list[0], Exception): - logger.warning("Open Library error for %r: %s", query, results_list[0]) - - hc_results: list[BookImportCandidate] = [] - if len(results_list) > 1: - if not isinstance(results_list[1], BaseException): - hc_results = results_list[1] - else: - logger.warning("Hardcover error for %r: %s", query, results_list[1]) + enabled_sources = [s for s in _build_sources(hardcover_api_token) if s.enabled()] + results: list[BookImportCandidate] = [] + + if enabled_sources: + all_results: list[list[BookImportCandidate]] = [] + + async def _run_source(spec: SourceSpec) -> list[BookImportCandidate]: + try: + r = await spec.search(query, search_type, client) + logger.info("%s returned %d result(s) for %r", spec.stage, len(r), query) + return r + except SourceBackendError as exc: + logger.warning("%s backend error for %r: status=%s", spec.stage, query, exc.status_code) + return [] + except Exception as exc: + logger.warning("%s error for %r: %s", spec.stage, query, exc) + return [] + + results_list = await asyncio.gather(*[_run_source(s) for s in enabled_sources]) - logger.info("Open Library returned %d result(s) for %r", len(ol_results), query) - logger.info("Hardcover returned %d result(s) for %r", len(hc_results), query) + for r in results_list: + all_results.append(r) - results = _merge_results(ol_results, hc_results) + for r in all_results: + results = _merge_results(results, r) + else: + # Every parallel source is switched off. Google Books can still answer as + # the fallback, which is the only remaining provider. + logger.warning( + "No parallel import search provider is enabled for %r — " + "only the Google Books fallback is available", + query, + ) if not results: - if not api_key: + if not _google_fallback_enabled(): + logger.warning( + "No results for %r and the Google Books fallback is disabled " + "(GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED=false)", + query, + ) + elif not api_key: logger.warning( "No results for %r and GOOGLE_BOOKS_API_KEY is not set — " "Google Books requires an API key for all requests. " @@ -299,7 +497,7 @@ async def search( query, ) else: - logger.info("No OL/HC results — falling back to Google Books") + logger.info("No results from any source — falling back to Google Books") try: results = await _search_google_books(query, search_type, api_key, client) except SourceBackendError as exc: @@ -826,26 +1024,810 @@ def map_hardcover(edition: dict) -> BookImportCandidate | None: ) -# ── Merge / Deduplicate ─────────────────────────────────────────────────────── +# ── BnF ──────────────────────────────────────────────────────────────────────── -def _merge_results( - primary: list[BookImportCandidate], - secondary: list[BookImportCandidate], +async def _search_bnf( + query: str, + search_type: str, + client: httpx.AsyncClient, ) -> list[BookImportCandidate]: - """Merge two candidate lists, preserving every candidate in input order. + """Search BnF catalogue general via SRU. + + Raises: + SourceBackendError: On HTTP errors from BnF. + """ + if search_type == "isbn": + try: + isbn13 = normalize_isbn(query) + except ValueError: + logger.warning("BnF invalid ISBN: %s", query) + return [] + query_str = f'bib.isbn adj "{isbn13}"' + else: + term = _sanitize_cql_term(query) + if not term: + logger.debug("BnF: title query is empty after sanitizing") + return [] + query_str = f'bib.title all "{term}" and bib.doctype adj "a"' + + params: dict = { + "version": "1.2", + "operation": "searchRetrieve", + "query": query_str, + "recordSchema": "dublincore", + "maximumRecords": 10, + "sortBy": "dat/sort.descending", + } + + logger.debug("BnF request — url=%s params=%s", BNF_SRU_URL, params) + + try: + resp = await client.get(BNF_SRU_URL, params=params, headers={"User-Agent": _USER_AGENT}) + logger.debug("BnF response — status=%d body_size=%d bytes", + resp.status_code, len(resp.content)) + resp.raise_for_status() + except httpx.HTTPStatusError as exc: + logger.warning("BnF HTTP error: %s %s", exc.response.status_code, exc.response.text[:200]) + raise SourceBackendError("bnf", exc.response.status_code) from exc + except httpx.HTTPError as exc: + logger.warning("BnF request failed: %s", exc) + raise SourceBackendError("bnf") from exc + + candidates = map_bnf_response(resp.text) + for c in candidates: + logger.debug(" BnF candidate: title=%r isbn=%r", c.title, c.isbn) + return candidates + + +def map_bnf_response(xml_text: str) -> list[BookImportCandidate]: + """Parse BnF SRU Dublin Core XML response into BookImportCandidate list.""" + namespaces = { + "srw": "http://www.loc.gov/zing/srw/", + "oai_dc": "http://www.openarchives.org/OAI/2.0/oai_dc/", + "dc": "http://purl.org/dc/elements/1.1/", + } + + try: + # The payload comes from a remote server, so parse it with entity expansion + # and external entity resolution disabled. + root = SafeET.fromstring(xml_text) + except (ET.ParseError, DefusedXmlException) as exc: + logger.warning("BnF XML parse error: %s", exc) + return [] + + records = root.findall(".//oai_dc:dc", namespaces) + + # BnF answers one ISBN with every record that carries it, which includes a + # commercial barcode stub ("Code a barres commercial : EAN ..."). The marker also + # appears on genuine records, so it only ranks a record lower instead of dropping + # it: within an ISBN group the richest, least barcode-ish record wins. + best_by_key: dict[str, tuple[BookImportCandidate, bool]] = {} + order: list[str] = [] + seen_arks: set[str] = set() + + for record in records: + candidate = map_bnf_record(record, namespaces) + if candidate is None: + continue + ark = _extract_bnf_ark(record, namespaces) + if ark and ark in seen_arks: + continue + if ark: + seen_arks.add(ark) + + key = candidate.isbn or ark or f"__record_{len(order)}" + is_shell = _has_bnf_barcode_marker(record, namespaces) + incumbent = best_by_key.get(key) + if incumbent is None: + best_by_key[key] = (candidate, is_shell) + order.append(key) + continue + if not is_shell and (incumbent[1] or _bnf_record_richness(candidate) > _bnf_record_richness(incumbent[0])): + best_by_key[key] = (candidate, is_shell) + + return [best_by_key[key][0] for key in order] + + +def _bnf_record_richness(candidate: BookImportCandidate) -> int: + """Count populated fields, used to pick the better of two BnF records.""" + return sum( + 1 + for value in ( + candidate.subtitle, + candidate.author, + candidate.publisher, + candidate.published_year, + candidate.page_count, + candidate.language, + candidate.tags, + ) + if value + ) + + +def _has_bnf_barcode_marker(record: ET.Element, namespaces: dict[str, str]) -> bool: + """Return True when a BnF record is (or includes) the commercial barcode stub.""" + for description in record.findall("dc:description", namespaces): + if description.text and "Code à barres commercial" in description.text: + return True + return False + + +def _extract_bnf_isbn(record: ET.Element, namespaces: dict[str, str]) -> str | None: + """Extract the ISBN from a BnF record's dc:identifier values ('ISBN 9791020923554').""" + for identifier in record.findall("dc:identifier", namespaces): + text = (identifier.text or "").strip() + match = re.search(r"ISBN\s+([0-9Xx][0-9Xx\s-]{8,}[0-9Xx])", text, re.IGNORECASE) + if not match: + continue + try: + return normalize_isbn(match.group(1)) + except ValueError: + continue + return None + + +def map_bnf_record(record: ET.Element, namespaces: dict[str, str]) -> BookImportCandidate | None: + """Map a single BnF Dublin Core record to BookImportCandidate.""" + title_elem = record.find("dc:title", namespaces) + if title_elem is None or not title_elem.text: + return None + + raw_title = title_elem.text + title, subtitle = _split_bnf_title(raw_title) + + authors: list[str] = [] + for creator in record.findall("dc:creator", namespaces): + if creator.text: + author = _clean_bnf_author(creator.text) + if author: + authors.append(author) + for contributor in record.findall("dc:contributor", namespaces): + if contributor.text: + author = _clean_bnf_author(contributor.text) + if author and author not in authors: + authors.append(author) + author_str = ", ".join(authors) if authors else None + + publishers = [p.text.strip() for p in record.findall("dc:publisher", namespaces) if p.text] + publisher = _clean_bnf_publisher(publishers[0]) if publishers else None + + date_elem = record.find("dc:date", namespaces) + published_year = None + if date_elem is not None and date_elem.text: + try: + published_year = int(date_elem.text[:4]) + except (ValueError, IndexError): + pass + + format_elem = record.find("dc:format", namespaces) + page_count = None + if format_elem is not None and format_elem.text: + page_count = _parse_page_count(format_elem.text) + + languages = [l.text for l in record.findall("dc:language", namespaces) if l.text] + language = _normalize_language_code(languages[0]) if languages else None + + subjects = [_clean_bnf_tag(s.text) for s in record.findall("dc:subject", namespaces) if s.text] + subjects = [s for s in subjects if s] + tags = ", ".join(subjects[:3]) if subjects else None + + return BookImportCandidate( + title=title, + subtitle=subtitle, + author=author_str, + authors=authors, + isbn=_extract_bnf_isbn(record, namespaces), + cover_url=None, + publisher=publisher, + published_year=published_year, + page_count=page_count, + language=language, + tags=tags, + source="bnf", + ) + + +def _clean_bnf_publisher(raw: str) -> str: + """Drop the BnF publication-place suffix, e.g. 'Editions Gallimard (Paris)'.""" + return re.sub(r"\s*\([^()]*\)\s*$", "", raw).strip() or raw.strip() + + +def _split_bnf_title(raw_title: str) -> tuple[str, Optional[str]]: + """Split a BnF dc:title into (title, subtitle). + + BnF conventions seen in the catalogue: + "L'heure du Sud : ou l'invention ... (Éd. 2024)" -> edition statement, dropped + "Dune / Frank Herbert" -> statement of responsibility, dropped + "Titre : sous-titre" -> real subtitle, kept + """ + title = raw_title.strip() + + # Drop a trailing edition statement. BnF returns both accented and unaccented + # forms, e.g. "(Ed. 2024)" and "(Éd. 2024)". + title = re.sub( + r"\s*\((?:[ÉEeé]d\.?|[ÉEeé]dition)\s*[^()]*\)\s*$", + "", + title, + ).strip() + + # Drop a trailing statement of responsibility after " / ". + if " / " in title: + title = title.split(" / ", 1)[0].strip() + + if " : " in title: + head, tail = title.split(" : ", 1) + subtitle = tail.split(" / ", 1)[0].strip() if " / " in tail else tail.strip() + # The edition statement may sit after the subtitle rather than the whole title. + subtitle = re.sub( + r"\s*\((?:[ÉEeé]d\.?|[ÉEeé]dition)\s*[^()]*\)\s*$", + "", + subtitle, + ).strip() + return head.strip(), subtitle or None + + return title, None + + +# BnF subject headings use " -- " for subdivisions ("Relations internationales -- +# 2001-...."). Keep the top-level heading only. +_BNF_SUBDIVISION_RE = re.compile(r"\s+--\s+.*$") + + +def _clean_bnf_tag(tag: str) -> str: + """Strip the subdivision suffix from a single BnF subject heading.""" + return _BNF_SUBDIVISION_RE.sub("", tag).strip() or tag.strip() + + +def _sanitize_cql_term(value: str) -> str: + """Make a user query safe to interpolate inside a quoted CQL string. + + A stray double quote would otherwise terminate the CQL string early; BnF answers + such a query with a diagnostic document and zero records instead of an error, so + the breakage would be silent. + """ + cleaned = value.replace('"', " ").replace("\\", " ") + return re.sub(r"\s+", " ", cleaned).strip() + + +# Role terms BnF appends after a personal name (UNIMARC 7xx role codes rendered in +# French). Matching a known vocabulary keeps real names such as "Ursula K. Le Guin" +# intact, which a generic "strip everything after a period" rule would mangle. +_BNF_ROLE_RE = re.compile( + r"\.\s*(?:auteur|editeur|éditeur|traducteur|postfacier|préfacier|prefacier|" + r"illustrateur|directeur de publication|photographe|compilateur|annotateur|" + r"correcteur|dessinateur|graveur|interviewer|prologue|epilogue|épilogue|" + r"preface|préface|enregistrement|editing|revue de|de)\b.*$", + re.IGNORECASE, +) + + +def _clean_bnf_author(raw: str) -> str: + """Strip BnF role suffixes from a creator/contributor string. + + Handles "Badie, Bertrand (1950-....). Éditeur scientifique" and + "Labica, Thierry. Postfacier" down to the bare name. + """ + author = raw.strip() + + # Life dates first: "Name (1900-1980)" or "Name (1950-.... ; journaliste)". + author = re.sub(r"\s*\((?:\d{4}|-{3})[^)]*\)", "", author).strip() + + # Then a trailing role phrase: "Name. Postfacier". + author = _BNF_ROLE_RE.sub("", author).strip() + + return author.rstrip(".").strip() or raw.strip() + + +def _extract_bnf_ark(record: ET.Element, namespaces: dict[str, str]) -> str | None: + """Extract BnF ARK identifier from a record.""" + for identifier in record.findall("dc:identifier", namespaces): + if identifier.text and "ark:/" in identifier.text: + return identifier.text.strip() + return None + + +# ── Inventaire.io ─────────────────────────────────────────────────────────────── + +# Claims pulled along with the edition in the single by-uris request: +# P629 -> the work, P50 -> authors on that work, P123 -> publisher. +# Each relative is returned as a full entity with resolved labels. +INVENTAIRE_RELATIVES: tuple[str, ...] = ("wdt:P629", "wdt:P50", "wdt:P123") + +_LANG_PREFERENCE: tuple[str, ...] = ("en", "fr", "de", "es", "it") + + +async def _search_inventaire( + query: str, + search_type: str, + client: httpx.AsyncClient, +) -> list[BookImportCandidate]: + """Search inventaire.io by ISBN via the entities API. + + Inventaire is ISBN-only: it has no usable title-search route. + + Raises: + SourceBackendError: On HTTP errors from inventaire.io. + """ + if search_type != "isbn": + logger.debug("Inventaire: skipping title search (ISBN-only provider)") + return [] + + try: + isbn13 = normalize_isbn(query) + except ValueError: + logger.warning("Inventaire invalid ISBN: %s", query) + return [] + + # relatives must be repeated query params; a comma-joined value returns HTTP 400. + params: list[tuple[str, str]] = [("uris", f"isbn:{isbn13}")] + params += [("relatives", rel) for rel in INVENTAIRE_RELATIVES] + params.append(("lang", "en")) + + logger.debug("Inventaire request — url=%s isbn=%s", INVENTAIRE_ENTITIES_URL, isbn13) + + try: + resp = await client.get(INVENTAIRE_ENTITIES_URL, params=params, headers={"User-Agent": _USER_AGENT}) + logger.debug("Inventaire response — status=%d body_size=%d bytes", + resp.status_code, len(resp.content)) + resp.raise_for_status() + except httpx.HTTPStatusError as exc: + logger.warning("Inventaire HTTP error: %s %s", exc.response.status_code, exc.response.text[:200]) + raise SourceBackendError("inventaire", exc.response.status_code) from exc + except httpx.HTTPError as exc: + logger.warning("Inventaire request failed: %s", exc) + raise SourceBackendError("inventaire") from exc + + candidates = map_inventaire_response(resp.json()) + for c in candidates: + logger.debug(" Inventaire candidate: title=%r author=%r isbn=%r cover=%r", + c.title, c.author, c.isbn, c.cover_url) + return candidates + + +def map_inventaire_response(data: dict) -> list[BookImportCandidate]: + """Parse an inventaire.io by-uris response into a BookImportCandidate list. + + The response is ``{"entities": {uri: entity}, "redirects": {"isbn:...": uri}}``. + Every claim value is a plain string; the ISBN redirect points at the edition. + """ + entities: dict[str, dict] = data.get("entities") or {} + if not entities: + return [] + + edition_uri = _find_inventaire_edition_uri(data.get("redirects") or {}, entities) + if edition_uri is None: + return [] + + edition = entities[edition_uri] + candidate = map_inventaire_entity(edition, entities, entities[edition_uri].get("labels") or {}) + if candidate is None: + return [] + if not candidate.isbn: + candidate = candidate.model_copy(update={"isbn": _isbn_from_uri(edition_uri)}) + return [candidate] + + +def _find_inventaire_edition_uri(redirects: dict[str, str], entities: dict[str, dict]) -> str | None: + """Return the URI of the edition entity in an inventaire.io response.""" + for uri, target in redirects.items(): + if uri.startswith("isbn:") and target in entities: + return target + # Fall back to any entity that actually is an edition. + for uri, entity in entities.items(): + if entity.get("type") == "edition": + return uri + return None + + +def _isbn_from_uri(uri: str) -> str | None: + """Return the ISBN carried by an ``isbn:`` URI, normalized. + + Returns None when the value is not a valid ISBN, so a malformed identifier is + never persisted. + """ + if not uri.startswith("isbn:"): + return None + raw = uri[len("isbn:"):] + try: + return normalize_isbn(raw) + except ValueError: + logger.debug("Inventaire returned a non-ISBN identifier: %r", raw) + return None + + +def map_inventaire_entity( + edition: dict, + entities: dict[str, dict], + labels: dict[str, str] | None = None, +) -> BookImportCandidate | None: + """Map an inventaire.io edition entity to BookImportCandidate.""" + claims: dict[str, list[str]] = edition.get("claims") or {} + + title = _inventaire_claim(claims, "wdt:P1476") + if not title: + title = _inventaire_label(labels or {}) + if not title: + return None + subtitle = _inventaire_claim(claims, "wdt:P1680") + + isbn: str | None = None + for raw in _inventaire_claims(claims, "wdt:P212") + _inventaire_claims(claims, "wdt:P957"): + try: + isbn = normalize_isbn(raw) + break + except ValueError: + continue + + published_year = _inventaire_year(_inventaire_claim(claims, "wdt:P577")) + page_count = _inventaire_int(_inventaire_claim(claims, "wdt:P1104")) + + language = _normalize_language_code(edition.get("originalLang") or None) + + cover_url = _inventaire_cover(edition, claims) + + authors = _inventaire_authors(claims, entities) + author = ", ".join(authors) if authors else None + + publisher = _inventaire_publisher(claims, entities) + + return BookImportCandidate( + title=title, + subtitle=subtitle, + author=author, + authors=authors, + isbn=isbn, + cover_url=cover_url, + publisher=publisher, + published_year=published_year, + page_count=page_count, + language=language, + tags=None, + source="inventaire", + ) + + +def _inventaire_claims(claims: dict[str, list[str]], claim_id: str) -> list[str]: + """Return the string values of a claim, tolerating legacy dict-shaped values.""" + values: list[str] = [] + for raw in claims.get(claim_id) or []: + if isinstance(raw, str) and raw: + values.append(raw) + elif isinstance(raw, dict): + value = raw.get("value") + if isinstance(value, str) and value: + values.append(value) + return values + + +def _inventaire_claim(claims: dict[str, list[str]], claim_id: str) -> str | None: + """Return the first string value of a claim.""" + values = _inventaire_claims(claims, claim_id) + return values[0] if values else None + + +def _inventaire_label(labels: dict[str, str]) -> str | None: + """Pick a label in the reader's likely language, ignoring generated ones.""" + for key in _LANG_PREFERENCE: + value = labels.get(key) + if value: + return value + return None + + +def _inventaire_year(raw: str | None) -> int | None: + """Extract the year from a Wikidata date value such as '2024-03-01'.""" + if not raw: + return None + match = re.match(r"^(-?\d{1,4})", raw.strip()) + if not match: + return None + try: + return int(match.group(1)) + except ValueError: + return None + + +def _inventaire_int(raw: str | None) -> int | None: + """Parse a numeric claim value such as a page count.""" + if not raw: + return None + match = re.search(r"\d+", raw) + if not match: + return None + return int(match.group(0)) + + +def _inventaire_cover(edition: dict, claims: dict[str, list[str]]) -> str | None: + """Build the absolute cover URL from the image hash, if one exists.""" + cover_hash = _inventaire_claim(claims, "invp:P2") + if not cover_hash: + image_url = (edition.get("image") or {}).get("url") or "" + cover_hash = image_url.rsplit("/", 1)[-1] if image_url else "" + if not cover_hash: + return None + cover_url = f"{INVENTAIRE_COVER_BASE_URL}/{cover_hash}" + return cover_url if is_safe_cover_import_url(cover_url) else None + + +def _inventaire_authors(claims: dict[str, list[str]], entities: dict[str, dict]) -> list[str]: + """Resolve author labels from the work entity referenced by the edition. + + Native inventaire works often carry no P50, so the edition's own P50 is used as a + fallback when the work yields nothing. + """ + work_uri = _inventaire_claim(claims, "wdt:P629") + work = entities.get(work_uri) if work_uri else None + + authors = _inventaire_author_labels((work or {}).get("claims") or {}, entities) + if not authors: + authors = _inventaire_author_labels(claims, entities) + return authors + + +def _inventaire_author_labels(claims: dict[str, list[str]], entities: dict[str, dict]) -> list[str]: + """Resolve P50 author URIs into display labels.""" + authors: list[str] = [] + for author_uri in _inventaire_claims(claims, "wdt:P50"): + entity = entities.get(author_uri) + label = _inventaire_label((entity or {}).get("labels") or {}) + if label and label not in authors: + authors.append(label) + return authors + + +def _inventaire_publisher(claims: dict[str, list[str]], entities: dict[str, dict]) -> str | None: + """Resolve the publisher label from the edition's P123 relative.""" + publisher_uri = _inventaire_claim(claims, "wdt:P123") + if not publisher_uri: + return None + entity = entities.get(publisher_uri) + if not entity: + return None + return _inventaire_label(entity.get("labels") or {}) + + +# ── Library of Congress ──────────────────────────────────────────────────────── + + +async def _search_loc( + query: str, + search_type: str, + client: httpx.AsyncClient, +) -> list[BookImportCandidate]: + """Search Library of Congress via SRU. + + Raises: + SourceBackendError: On HTTP errors from LoC. + """ + if search_type == "isbn": + try: + isbn13 = normalize_isbn(query) + except ValueError: + logger.warning("LoC invalid ISBN: %s", query) + return [] + query_str = f"bath.isbn={isbn13}" + else: + term = _sanitize_cql_term(query) + if not term: + logger.debug("LoC: title query is empty after sanitizing") + return [] + query_str = f"bath.title={term}" + + params: dict = { + "version": "1.1", + "operation": "searchRetrieve", + "query": query_str, + "recordSchema": "mods", + "maximumRecords": 10, + } + + logger.debug("LoC request — url=%s params=%s", LOC_SRU_URL, params) + + try: + # LoC is reached over plain HTTP on port 210, which many firewalls drop rather + # than refuse. A shorter timeout keeps a blocked LoC from stalling the search. + resp = await client.get( + LOC_SRU_URL, + params=params, + headers={"User-Agent": _USER_AGENT}, + timeout=_LOC_TIMEOUT_SECONDS, + ) + logger.debug("LoC response — status=%d body_size=%d bytes", + resp.status_code, len(resp.content)) + resp.raise_for_status() + except httpx.HTTPStatusError as exc: + logger.warning("LoC HTTP error: %s %s", exc.response.status_code, exc.response.text[:200]) + raise SourceBackendError("library_of_congress", exc.response.status_code) from exc + except httpx.HTTPError as exc: + logger.warning("LoC request failed: %s", exc) + raise SourceBackendError("library_of_congress") from exc + + candidates = map_loc_response(resp.text) + for c in candidates: + logger.debug(" LoC candidate: title=%r isbn=%r", c.title, c.isbn) + return candidates + + +def map_loc_response(xml_text: str) -> list[BookImportCandidate]: + """Parse LoC SRU MODS XML response into BookImportCandidate list.""" + namespaces = { + "srw": "http://www.loc.gov/zing/srw/", + "mods": "http://www.loc.gov/mods/v3", + } + + try: + # The payload comes from a remote server, so parse it with entity expansion + # and external entity resolution disabled. + root = SafeET.fromstring(xml_text) + except (ET.ParseError, DefusedXmlException) as exc: + logger.warning("LoC XML parse error: %s", exc) + return [] + + records = root.findall(".//mods:mods", namespaces) + candidates: list[BookImportCandidate] = [] + + for record in records: + candidate = map_loc_record(record, namespaces) + if candidate is not None: + candidates.append(candidate) + + return candidates + + +def map_loc_record(record: ET.Element, namespaces: dict[str, str]) -> BookImportCandidate | None: + """Map a single LoC MODS record to BookImportCandidate.""" + title_info = record.find("mods:titleInfo", namespaces) + if title_info is None: + return None + + title_elem = title_info.find("mods:title", namespaces) + if title_elem is None or not title_elem.text: + return None + + title = title_elem.text.strip() + + subtitle_elem = title_info.find("mods:subTitle", namespaces) + subtitle = subtitle_elem.text.strip() if subtitle_elem is not None and subtitle_elem.text else None + + authors: list[str] = [] + for name in record.findall("mods:name", namespaces): + # Only personal names are book authors; corporate names would be the publisher. + name_type = name.get("type", "") + if name_type and name_type != "personal": + continue + name_part = name.find("mods:namePart", namespaces) + if name_part is not None and name_part.text: + author = _reverse_loc_author(name_part.text.strip()) + if author and author not in authors: + authors.append(author) + author_str = ", ".join(authors) if authors else None + + isbn = _extract_loc_isbn(record, namespaces) + + origin_info = record.find("mods:originInfo", namespaces) + publisher = None + published_year = None + if origin_info is not None: + publisher_elem = origin_info.find("mods:publisher", namespaces) + if publisher_elem is not None and publisher_elem.text: + publisher = publisher_elem.text.strip() + + date_elem = origin_info.find("mods:dateIssued", namespaces) + if date_elem is not None and date_elem.text: + try: + published_year = int(date_elem.text[:4]) + except (ValueError, IndexError): + pass + + language = None + for lang in record.findall("mods:language", namespaces): + term = lang.find("mods:languageTerm", namespaces) + if term is not None and term.text: + language = _normalize_language_code(term.text.strip()) + break + + page_count = None + physical_desc = record.find("mods:physicalDescription", namespaces) + if physical_desc is not None: + extent = physical_desc.find("mods:extent", namespaces) + if extent is not None and extent.text: + page_count = _parse_page_count(extent.text) + + tags = None + # MODS nests topics inside , so search descendants rather than direct children. + topics = [t.text for t in record.findall(".//mods:topic", namespaces) if t.text] + if topics: + tags = ", ".join(topics[:3]) + + return BookImportCandidate( + title=title, + subtitle=subtitle, + author=author_str, + authors=authors, + isbn=isbn, + cover_url=None, + publisher=publisher, + published_year=published_year, + page_count=page_count, + language=language, + tags=tags, + source="library_of_congress", + ) + + +def _extract_loc_isbn(record: ET.Element, namespaces: dict[str, str]) -> str | None: + """Extract an ISBN from MODS identifier elements, preferring type='isbn'.""" + fallbacks: list[str] = [] + for identifier in record.findall("mods:identifier", namespaces): + text = (identifier.text or "").strip() + if not text or not re.fullmatch(r"[0-9Xx][0-9Xx\s-]{8,}[0-9Xx]", text): + continue + try: + normalized = normalize_isbn(text) + except ValueError: + continue + if identifier.get("type") == "isbn": + return normalized + fallbacks.append(normalized) + return fallbacks[0] if fallbacks else None + + +def _reverse_loc_author(raw: str) -> str: + """Reverse LoC's 'LastName, FirstName' form into 'FirstName LastName'. + + Real MODS namePart values carry trailing separators in both forms: + "Herbert, Frank." (abbreviation period) and "Horydczak, Theodor," (list comma). + Both must be stripped before the name is reversed. + """ + name = raw.strip().rstrip(".,").strip() + if ", " in name: + last, first = name.split(", ", 1) + reversed_name = f"{first.strip().rstrip('.,')} {last.strip().rstrip('.,')}".strip() + return reversed_name or name + return name or raw.strip() + + +# ── Merge / Deduplicate ─────────────────────────────────────────────────────── + + +def _merge_results(*lists: list[BookImportCandidate]) -> list[BookImportCandidate]: + """Merge multiple candidate lists, preserving every candidate in input order. The frontend is responsible for grouping variants that represent the same book (e.g. by ISBN) and letting the user pick the best record. A user can only own one book per exact ISBN string, so keeping all provider-specific records lets the user compare data quality before importing. """ - return primary + secondary + merged: list[BookImportCandidate] = [] + for lst in lists: + merged.extend(lst) + return merged # ── Helpers ─────────────────────────────────────────────────────────────────── +def _parse_page_count(text: str) -> Optional[int]: + """Extract page count from a format string like '1 vol. (265 p.) ; 20 cm' or '412 p. ; 22 cm.'.""" + # Try parenthesized format first: (265 p.) + match = re.search(r"\(([\d,\s]+)\s*p\.\)", text) + if match: + try: + return int(match.group(1).replace(",", "").replace(" ", "")) + except ValueError: + pass + # Try plain format: 412 p. (or 1,234 p.) + match = re.search(r"([\d,]+)\s*p\.", text) + if match: + try: + return int(match.group(1).replace(",", "").replace(" ", "")) + except ValueError: + pass + return None + + def _pick_isbn(isbns: list[str]) -> Optional[str]: """Return the first ISBN-13 found, or the first ISBN-10, or None.""" for isbn in isbns: diff --git a/backend/tests/test_book_import.py b/backend/tests/test_book_import.py index 2cfb7850..65ec8e87 100644 --- a/backend/tests/test_book_import.py +++ b/backend/tests/test_book_import.py @@ -81,6 +81,9 @@ async def test_search_with_progress_auto_ol_and_hc(monkeypatch: pytest.MonkeyPat fake_ol = BookImportCandidate(title="OL Book", source="open_library") monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[fake_ol])) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) events = [] async for e in bi.search_with_progress("query", "title", hardcover_api_token="token"): @@ -91,17 +94,27 @@ async def test_search_with_progress_auto_ol_and_hc(monkeypatch: pytest.MonkeyPat @pytest.mark.anyio -async def test_search_with_progress_auto_hc_skipped_no_token() -> None: +async def test_search_with_progress_auto_hc_skipped_no_token(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) events = [] async for e in bi.search_with_progress("query", "title"): events.append(e) assert any(e.get("stage") == "hardcover" and e.get("status") == "skipped" for e in events) + # A skipped source must be reported exactly once: never also as a result count. + hardcover = [e for e in events if e.get("stage") == "hardcover"] + assert len(hardcover) == 1 @pytest.mark.anyio async def test_search_with_progress_auto_fallback_google(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) fake_gb = BookImportCandidate(title="GB Book", source="google_books") monkeypatch.setattr(bi, "_search_google_books", AsyncMock(return_value=[fake_gb])) @@ -117,6 +130,9 @@ async def test_search_with_progress_auto_fallback_google(monkeypatch: pytest.Mon async def test_search_with_progress_auto_fallback_google_error(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) monkeypatch.setattr( bi, "_search_google_books", AsyncMock(side_effect=bi.SourceBackendError("google_books", 500)), @@ -131,11 +147,14 @@ async def test_search_with_progress_auto_fallback_google_error(monkeypatch: pyte @pytest.mark.anyio async def test_search_with_progress_general_exception(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(bi, "_search_open_library", AsyncMock(side_effect=RuntimeError("boom"))) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) events = [] async for e in bi.search_with_progress("query", "title"): events.append(e) - assert any(e.get("stage") == "error" for e in events) + assert any(e.get("stage") == "open_library" and e.get("status") == "error" for e in events) # ── search ───────────────────────────────────────────────────────────────────── @@ -147,6 +166,9 @@ async def test_search_ol_exception(monkeypatch: pytest.MonkeyPatch) -> None: AsyncMock(side_effect=bi.SourceBackendError("open_library", 503)), ) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) results = await bi.search("query", "title") assert results == [] @@ -159,6 +181,9 @@ async def test_search_ol_generic_exception(monkeypatch: pytest.MonkeyPatch) -> N AsyncMock(side_effect=RuntimeError("boom")), ) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) results = await bi.search("query", "title") assert results == [] @@ -173,6 +198,9 @@ async def test_search_ol_generic_exception_with_hc(monkeypatch: pytest.MonkeyPat ) fake_hc = BookImportCandidate(title="HC Book", source="hardcover") monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[fake_hc])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) results = await bi.search("query", "title", hardcover_api_token="token") assert len(results) == 1 @@ -185,6 +213,9 @@ async def test_search_hc_exception(monkeypatch: pytest.MonkeyPatch) -> None: bi, "_search_hardcover", AsyncMock(side_effect=RuntimeError("boom")), ) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) results = await bi.search("query", "title", hardcover_api_token="token") assert results == [] @@ -194,6 +225,9 @@ async def test_search_hc_exception(monkeypatch: pytest.MonkeyPatch) -> None: async def test_search_fallback_google_no_key(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) results = await bi.search("query", "title") assert results == [] @@ -203,6 +237,9 @@ async def test_search_fallback_google_no_key(monkeypatch: pytest.MonkeyPatch) -> async def test_search_fallback_google_success(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) fake = BookImportCandidate(title="Book", source="google_books") monkeypatch.setattr(bi, "_search_google_books", AsyncMock(return_value=[fake])) @@ -214,6 +251,9 @@ async def test_search_fallback_google_success(monkeypatch: pytest.MonkeyPatch) - async def test_search_fallback_google_error(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) monkeypatch.setattr( bi, "_search_google_books", AsyncMock(side_effect=bi.SourceBackendError("google_books", 500)), @@ -1074,6 +1114,9 @@ async def test_search_with_progress_ol_error_event(monkeypatch: pytest.MonkeyPat AsyncMock(side_effect=bi.SourceBackendError("open_library", 500)), ) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) events = [] async for e in bi.search_with_progress("query", "title"): @@ -1214,6 +1257,9 @@ async def test_search_with_progress_hc_done_event(monkeypatch: pytest.MonkeyPatc fake_hc = BookImportCandidate(title="HC Book", source="hardcover") monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) monkeypatch.setattr(bi, "_search_hardcover", AsyncMock(return_value=[fake_hc])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) events = [] async for e in bi.search_with_progress("query", "title", hardcover_api_token="token"): @@ -1230,8 +1276,577 @@ async def test_search_with_progress_hc_error_event(monkeypatch: pytest.MonkeyPat bi, "_search_hardcover", AsyncMock(side_effect=RuntimeError("boom")), ) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) events = [] async for e in bi.search_with_progress("query", "title", hardcover_api_token="token"): events.append(e) assert any(e.get("stage") == "hardcover" and e.get("status") == "error" for e in events) + + +# ── BnF ────────────────────────────────────────────────────────────────────────── + +# Two records for one ISBN: a commercial barcode stub and the real bibliographic +# record, as catalogue.bnf.fr actually answers a bib.isbn query. +BNF_SAMPLE_XML = """ + + + + + + L'heure du Sud : ou l'invention d'un nouvel ordre mondial (Ed. 2024) + Badie, Bertrand + Vidal, Dominique + Editions les Liens qui liberent (Paris) + 2024 + 1 volume 295 p ; 22 cm + fre + http://catalogue.bnf.fr/ark:/12148/cb48773987b + ISBN 9791020923554 + Code à barres commercial : EAN 9791020923554 + + + + + + + L'heure du Sud : ou l'invention d'un nouvel ordre mondial / sous la direction de Bertrand Badie + Badie, Bertrand (1950-....).Editeur scientifique + les Liens qui liberent (Paris) + 2024 + 1 vol. (295 p.) : ill. ; 22 cm + fre + http://catalogue.bnf.fr/ark:/12148/cb47695466k + ISBN 9791020923554 + Relations internationales -- 2001-.... + + + + + +""" + + +def test_map_bnf_response_basic() -> None: + candidates = bi.map_bnf_response(BNF_SAMPLE_XML) + # Both records carry the same ISBN, so only the richer one survives. + assert len(candidates) == 1 + c = candidates[0] + assert c.title == "L'heure du Sud" + assert c.subtitle == "ou l'invention d'un nouvel ordre mondial" + assert c.authors == ["Badie, Bertrand"] + assert c.isbn == "9791020923554" + assert c.publisher == "les Liens qui liberent" + assert c.published_year == 2024 + assert c.page_count == 295 + assert c.language == "FR" + assert c.tags == "Relations internationales" + assert c.source == "bnf" + assert c.cover_url is None + + +def test_map_bnf_response_empty() -> None: + xml = """ + + +""" + assert bi.map_bnf_response(xml) == [] + + +def test_map_bnf_response_malformed_xml() -> None: + assert bi.map_bnf_response(" None: + xml = """ + +""" + assert bi.map_bnf_response(xml) == [] + + +# ── Inventaire ───────────────────────────────────────────────────────────────── + +INVENTAIRE_SAMPLE = { + # Shape verified against https://inventaire.io/api/entities/by-uris: + # claim values are plain strings, and P629/P50/P123 relatives come back as entities. + "entities": { + "inv:edition123": { + "type": "edition", + "originalLang": "fr", + "labels": {"fromclaims": "L'heure du Sud"}, + "claims": { + "wdt:P1476": ["L'heure du Sud"], + "wdt:P1680": ["ou l'invention d'un nouvel ordre mondial"], + "wdt:P212": ["979-10-209-2355-4"], + "wdt:P577": ["2024"], + "wdt:P1104": ["295"], + "wdt:P123": ["wd:Q3236382"], + "wdt:P629": ["inv:work123"], + "invp:P2": ["abcdef123456"], + }, + }, + "inv:work123": { + "type": "work", + "labels": {"fr": "L'heure du Sud"}, + "claims": {"wdt:P50": ["wd:Q7934"]}, + }, + "wd:Q7934": { + "type": "human", + "labels": {"en": "Bertrand Badie"}, + }, + "wd:Q3236382": { + "type": "publisher", + "labels": {"en": "Editions les Liens qui liberent"}, + }, + }, + "redirects": {"isbn:9791020923554": "inv:edition123"}, +} + + +def test_map_inventaire_response_basic() -> None: + candidates = bi.map_inventaire_response(INVENTAIRE_SAMPLE) + assert len(candidates) == 1 + c = candidates[0] + assert c.title == "L'heure du Sud" + assert c.subtitle == "ou l'invention d'un nouvel ordre mondial" + assert c.isbn == "9791020923554" + assert c.published_year == 2024 + assert c.page_count == 295 + assert c.language == "FR" + assert c.source == "inventaire" + assert c.cover_url == "https://inventaire.io/img/entities/abcdef123456" + assert c.authors == ["Bertrand Badie"] + assert c.publisher == "Editions les Liens qui liberent" + + +def test_map_inventaire_response_authors_fall_back_to_edition_claims() -> None: + data = { + "entities": { + "inv:edition123": { + "type": "edition", + "labels": {"fromclaims": "Plain Edition"}, + "claims": { + "wdt:P1476": ["Plain Edition"], + "wdt:P212": ["9782072716645"], + "wdt:P50": ["wd:Q43499"], + }, + }, + "inv:work123": {"type": "work", "labels": {}, "claims": {}}, + "wd:Q43499": {"type": "human", "labels": {"en": "Erasmus"}}, + }, + "redirects": {"isbn:9782072716645": "inv:edition123"}, + } + candidates = bi.map_inventaire_response(data) + assert candidates[0].authors == ["Erasmus"] + + +def test_map_inventaire_response_without_title_is_skipped() -> None: + data = { + "entities": {"inv:edition123": {"type": "edition", "labels": {}, "claims": {}}}, + "redirects": {"isbn:9791020923554": "inv:edition123"}, + } + assert bi.map_inventaire_response(data) == [] + + +def test_map_inventaire_response_empty() -> None: + assert bi.map_inventaire_response({"entities": {}, "redirects": {}}) == [] + + +def test_map_inventaire_response_no_entities() -> None: + assert bi.map_inventaire_response({}) == [] + + +# ── Library of Congress ─────────────────────────────────────────────────────── + +LOC_SAMPLE_XML = """ + + + + + + + Dune + Book One of the Dune Chronicles + + + Herbert, Frank. + + + Chilton Books + + + Chilton Books + 1965 + + + eng + + + 412 p. ; 22 cm. + + + Science fiction + + 9780441013593 + 65021248 + + + + + +""" + + +def test_map_loc_response_basic() -> None: + candidates = bi.map_loc_response(LOC_SAMPLE_XML) + assert len(candidates) == 1 + c = candidates[0] + assert c.title == "Dune" + assert c.subtitle == "Book One of the Dune Chronicles" + assert c.author == "Frank Herbert" + assert c.authors == ["Frank Herbert"] + assert c.publisher == "Chilton Books" + assert c.published_year == 1965 + assert c.page_count == 412 + assert c.language == "EN" + assert c.isbn == "9780441013593" + assert c.tags == "Science fiction" + assert c.source == "library_of_congress" + assert c.cover_url is None + + +def test_map_loc_response_reads_topics_nested_in_subject() -> None: + """MODS nests topics under , so only a descendant search finds them.""" + xml = """ + + + + Nested Topics + + Outer + Inner + + + +""" + candidates = bi.map_loc_response(xml) + assert candidates[0].tags == "Outer, Inner" + + +def test_map_loc_response_accepts_spaced_isbn() -> None: + xml = """ + + + + Spaced + 0-441-01359-8 + + +""" + candidates = bi.map_loc_response(xml) + assert candidates[0].isbn == "9780441013593" + + +def test_map_loc_response_empty() -> None: + xml = """ + + +""" + assert bi.map_loc_response(xml) == [] + + +def test_map_loc_response_malformed_xml() -> None: + assert bi.map_loc_response(" None: + """Each stage reports exactly one terminal event; no stage ends up twice.""" + monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[BookImportCandidate(title="B", source="bnf")])) + monkeypatch.setattr(bi, "_search_inventaire", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) + monkeypatch.setattr(bi.settings, "library_of_congress_import_search_enabled", True) + + events = [] + async for e in bi.search_with_progress("query", "isbn"): + events.append(e) + + terminal = {"done", "skipped", "error"} + by_stage: dict[str, list[str]] = {} + for e in events: + if e.get("stage") in ("complete", "error"): + continue + by_stage.setdefault(e["stage"], []).append(e.get("status")) + + for stage, statuses in by_stage.items(): + assert statuses.count("searching") <= 1, (stage, statuses) + assert len([s for s in statuses if s in terminal]) == 1, (stage, statuses) + + +@pytest.mark.anyio +async def test_search_with_progress_skips_inventaire_for_title_search( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Inventaire has no title search, so it must be skipped rather than report 0.""" + inventaire = AsyncMock(return_value=[]) + monkeypatch.setattr(bi, "_search_open_library", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_bnf", AsyncMock(return_value=[])) + monkeypatch.setattr(bi, "_search_inventaire", inventaire) + monkeypatch.setattr(bi, "_search_loc", AsyncMock(return_value=[])) + + events = [] + async for e in bi.search_with_progress("query", "title"): + events.append(e) + + inventaire_events = [e for e in events if e.get("stage") == "inventaire"] + assert len(inventaire_events) == 1 + assert inventaire_events[0]["status"] == "skipped" + assert inventaire_events[0]["reason"] == "title_search_unsupported" + inventaire.assert_not_awaited() + + +@pytest.mark.anyio +async def test_search_with_progress_google_only_mode_respects_the_flag( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(bi.settings, "google_books_import_search_enabled", False) + google = AsyncMock(return_value=[]) + + events = [] + async for e in bi.search_with_progress("query", "title", api_key="key", mode="google_only"): + events.append(e) + + assert any( + e.get("stage") == "google_books" and e.get("reason") == "disabled" for e in events + ) + google.assert_not_awaited() + + +# ── Parser edge cases found against live payloads ────────────────────────────── + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("Herbert, Frank.", "Frank Herbert"), + ("Herbert, Frank,", "Frank Herbert"), + ("Horydczak, Theodor,", "Theodor Horydczak"), + ("Mary Shelley", "Mary Shelley"), + ], +) +def test_reverse_loc_author_handles_trailing_separators(raw: str, expected: str) -> None: + assert bi._reverse_loc_author(raw) == expected + + +def test_split_bnf_title_strips_accented_edition_statements() -> None: + assert bi._split_bnf_title("L'heure du Sud : ou l'invention (Éd. 2024)") == ( + "L'heure du Sud", + "ou l'invention", + ) + assert bi._split_bnf_title("Série noire : intégrale (Édition 2020)") == ( + "Série noire", + "intégrale", + ) + assert bi._split_bnf_title("Dune / Frank Herbert") == ("Dune", None) + + +def test_clean_bnf_tag_drops_subdivisions() -> None: + assert bi._clean_bnf_tag("Relations internationales -- 2001-....") == "Relations internationales" + assert bi._clean_bnf_tag("Pays en développement") == "Pays en développement" + + +@pytest.mark.parametrize( + "raw", + ['Le Seigneur des "Anneaux"', 'He said "hello"', 'back\\slash', '"', " "], +) +def test_sanitize_cql_term_removes_quotes_and_backslashes(raw: str) -> None: + cleaned = bi._sanitize_cql_term(raw) + assert '"' not in cleaned + assert "\\" not in cleaned + + +@pytest.mark.anyio +async def test_search_bnf_title_with_quotes_still_queries(monkeypatch: pytest.MonkeyPatch) -> None: + """A quote in a title must not break the CQL query into a silent empty result.""" + seen: dict = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["query"] = request.url.params.get("query", "") + return httpx.Response(200, text="") + + transport = httpx.MockTransport(handler) + async with httpx.AsyncClient(transport=transport) as client: + results = await bi._search_bnf('Le Seigneur des "Anneaux"', "title", client) + + assert results == [] + assert '""' not in seen["query"] + assert seen["query"].count('"') % 2 == 0 + + +# ── Hardened XML parsing ─────────────────────────────────────────────────────── + +_BILLION_LAUGHS = """ + + + +]> + + + &lol3; + +""" + +_XXE_FILE_READ = """ +]> + + + &xxe; + +""" + + +def test_map_loc_response_rejects_entity_expansion() -> None: + """A hostile SRU payload must not be expanded, and must yield no candidates.""" + assert bi.map_loc_response(_BILLION_LAUGHS) == [] + + +def test_map_loc_response_rejects_external_entity() -> None: + """External entity references must not be resolved to local files.""" + assert bi.map_loc_response(_XXE_FILE_READ) == [] + + +def test_map_bnf_response_rejects_entity_expansion() -> None: + payload = _BILLION_LAUGHS.replace("mods:mods", "oai_dc:dc").replace( + 'xmlns:mods="http://www.loc.gov/mods/v3"', + 'xmlns:oai_dc="http://www.openarchives.org/OAI/2.0/oai_dc/"', + ).replace("mods:titleInfo", "").replace("mods:title", "dc:title").replace( + "", "" + ).replace("", "") + assert bi.map_bnf_response(payload) == [] + + +# ── Google fallback gating ───────────────────────────────────────────────────── + +@pytest.mark.anyio +async def test_search_uses_google_when_all_parallel_sources_disabled(monkeypatch: pytest.MonkeyPatch) -> None: + """With every parallel source off, Google Books must still answer as the fallback.""" + for flag, value in ( + ("open_library_import_search_enabled", False), + ("bnf_import_search_enabled", False), + ("inventaire_import_search_enabled", False), + ("library_of_congress_import_search_enabled", False), + ("hardcover_import_search_enabled", False), + ("google_books_import_search_enabled", True), + ): + monkeypatch.setattr(bi.settings, flag, value) + + fake_gb = BookImportCandidate(title="GB Book", source="google_books") + monkeypatch.setattr(bi, "_search_google_books", AsyncMock(return_value=[fake_gb])) + + results = await bi.search("query", "title", api_key="key") + assert [r.source for r in results] == ["google_books"] + + +@pytest.mark.anyio +async def test_search_skips_google_when_fallback_disabled(monkeypatch: pytest.MonkeyPatch) -> None: + """Disabling the fallback must stop the Google request entirely.""" + for flag, value in ( + ("open_library_import_search_enabled", False), + ("bnf_import_search_enabled", False), + ("inventaire_import_search_enabled", False), + ("library_of_congress_import_search_enabled", False), + ("hardcover_import_search_enabled", False), + ("google_books_import_search_enabled", False), + ): + monkeypatch.setattr(bi.settings, flag, value) + + google = AsyncMock(return_value=[BookImportCandidate(title="GB", source="google_books")]) + monkeypatch.setattr(bi, "_search_google_books", google) + + results = await bi.search("query", "title", api_key="key") + assert results == [] + google.assert_not_awaited() + + +@pytest.mark.anyio +async def test_search_with_progress_reports_disabled_google_fallback(monkeypatch: pytest.MonkeyPatch) -> None: + for flag, value in ( + ("open_library_import_search_enabled", False), + ("bnf_import_search_enabled", False), + ("inventaire_import_search_enabled", False), + ("library_of_congress_import_search_enabled", False), + ("hardcover_import_search_enabled", False), + ("google_books_import_search_enabled", False), + ): + monkeypatch.setattr(bi.settings, flag, value) + + google = AsyncMock(return_value=[BookImportCandidate(title="GB", source="google_books")]) + monkeypatch.setattr(bi, "_search_google_books", google) + + events = [] + async for e in bi.search_with_progress("query", "title", api_key="key"): + events.append(e) + + assert any( + e.get("stage") == "google_books" and e.get("reason") == "disabled" for e in events + ) + google.assert_not_awaited() + + +@pytest.mark.anyio +async def test_search_with_progress_still_completes_without_parallel_sources( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A Google-only deployment is valid and must still emit a complete event.""" + for flag, value in ( + ("open_library_import_search_enabled", False), + ("bnf_import_search_enabled", False), + ("inventaire_import_search_enabled", False), + ("library_of_congress_import_search_enabled", False), + ("hardcover_import_search_enabled", False), + ("google_books_import_search_enabled", True), + ): + monkeypatch.setattr(bi.settings, flag, value) + + fake_gb = BookImportCandidate(title="GB Book", source="google_books") + monkeypatch.setattr(bi, "_search_google_books", AsyncMock(return_value=[fake_gb])) + + events = [] + async for e in bi.search_with_progress("query", "title", api_key="key"): + events.append(e) + + complete = [e for e in events if e.get("stage") == "complete"] + assert len(complete) == 1 + assert len(complete[0]["results"]) == 1 + + +# ── _parse_page_count ────────────────────────────────────────────────────────── + +def test_parse_page_count_valid() -> None: + assert bi._parse_page_count("1 vol. (265 p.) ; 20 cm") == 265 + assert bi._parse_page_count("412 p. ; 22 cm.") == 412 + assert bi._parse_page_count("1,234 p.") == 1234 + assert bi._parse_page_count("528 p. : 23 cm.") == 528 + + +def test_parse_page_count_invalid() -> None: + assert bi._parse_page_count("no pages") is None + assert bi._parse_page_count("") is None + + +# ── Source registry ───────────────────────────────────────────────────────────── + +def test_book_sources_registry() -> None: + stages = [s.stage for s in bi._build_sources("")] + assert "open_library" in stages + assert "bnf" in stages + assert "inventaire" in stages + assert "library_of_congress" in stages + assert "hardcover" in stages diff --git a/backend/tests/test_import.py b/backend/tests/test_import.py index 60646a33..0c5ce455 100644 --- a/backend/tests/test_import.py +++ b/backend/tests/test_import.py @@ -23,6 +23,20 @@ from app.services import book_import +@pytest.fixture(autouse=True) +def _only_legacy_sources(monkeypatch: MonkeyPatch) -> None: + """Run this module's tests with only Open Library, Hardcover and Google Books active. + + The tests here predate the BnF / Inventaire / Library of Congress providers and + assert on exact result counts from the original two parallel sources. Disabling + the newer keyless providers keeps those expectations valid and keeps the suite + free of real network calls. Their own behaviour is covered in test_book_import.py. + """ + monkeypatch.setattr(settings, "bnf_import_search_enabled", False) + monkeypatch.setattr(settings, "inventaire_import_search_enabled", False) + monkeypatch.setattr(settings, "library_of_congress_import_search_enabled", False) + + # ── Fake response data ───────────────────────────────────────────────────────── OPEN_LIBRARY_DUNE_DOC = { @@ -1006,6 +1020,32 @@ async def fake_download(url: str, covers_dir: str, http_client: httpx.AsyncClien assert resp.json()["cover_url"] is None +def test_import_book_rejects_private_cover_url(client: TestClient, monkeypatch: MonkeyPatch, tmp_path: Path) -> None: + """A candidate cover URL pointing at a private address must never be fetched (SSRF).""" + monkeypatch.setattr(settings, "covers_dir", str(tmp_path)) + fetched: list[str] = [] + + async def fake_download(url: str, covers_dir: str, http_client: httpx.AsyncClient, user_id: int | None) -> str | None: + fetched.append(url) # pragma: no cover - must never run + return "cover.jpg" + + monkeypatch.setattr(import_router, "download_cover", fake_download) + + payload = { + "candidate": { + "title": "Internal Service", + "isbn": "9780000000001", + "cover_url": "http://169.254.169.254/latest/meta-data/", + "source": "bnf", + }, + "reading_status": "want_to_read", + } + resp = client.post("/api/import", json=payload) + assert resp.status_code == 201 + assert resp.json()["cover_url"] is None + assert fetched == [] + + def test_import_book_no_cover_url_skips_download(client: TestClient, monkeypatch: MonkeyPatch, tmp_path: Path) -> None: """When the candidate has no cover_url, download_cover is never called.""" monkeypatch.setattr(settings, "covers_dir", str(tmp_path)) diff --git a/backend/tests/test_main.py b/backend/tests/test_main.py index d96e9ff9..caf51e92 100644 --- a/backend/tests/test_main.py +++ b/backend/tests/test_main.py @@ -193,6 +193,56 @@ async def call_next(request: Request) -> Response: assert received["scheme"] == "http" +# ── Import source startup validation ─────────────────────────────────────────── + +def _import_source_flags() -> list[str]: + """Every per-source enable flag, derived from the registry itself.""" + from app.services import book_import + + return [f"{spec.stage}_import_search_enabled" for spec in book_import._build_sources("")] + + +@pytest.mark.anyio +async def test_lifespan_refuses_to_start_without_any_import_source(monkeypatch) -> None: + """Disabling every book import source must stop startup with an actionable error.""" + from app.config import settings + from app.services import book_import + import app.main as main_module + + for flag in _import_source_flags(): + monkeypatch.setattr(settings, flag, False) + monkeypatch.setattr(settings, "google_books_import_search_enabled", False) + + with patch("app.main._periodic_maintenance", new=AsyncMock()): + with pytest.raises(RuntimeError) as exc_info: + async with main_module.lifespan(main_module.app): + pass + + message = str(exc_info.value) + assert "At least one import search provider" in message + # The error must name the variables the operator can set. + for stage in (s.stage for s in book_import._build_sources("")): + assert f"{stage.upper()}_IMPORT_SEARCH_ENABLED" in message + assert "GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED" in message + + +@pytest.mark.anyio +async def test_lifespan_starts_with_a_single_import_source(monkeypatch) -> None: + """One enabled source is enough to start.""" + from app.config import settings + import app.main as main_module + + for flag in _import_source_flags(): + monkeypatch.setattr(settings, flag, False) + monkeypatch.setattr(settings, "google_books_import_search_enabled", False) + monkeypatch.setattr(settings, "bnf_import_search_enabled", True) + + with patch("app.main._periodic_maintenance", new=AsyncMock()): + with patch("app.main._telemetry_heartbeat", new=AsyncMock()): + async with main_module.lifespan(main_module.app): + pass + + @pytest.mark.anyio async def test_lifespan_logs_warning_when_mail_not_configured(monkeypatch) -> None: """Lifespan should warn when MAIL_SERVER or MAIL_FROM are not configured.""" From 2e0377ec7889537a9ad2e2f5b2ad2cb96e0563ab Mon Sep 17 00:00:00 2001 From: codebude Date: Wed, 14 Oct 2026 20:43:02 +0200 Subject: [PATCH 2/5] Add new search sources, localize import source names and report each search stage --- .../src/lib/components/AddBookModal.svelte | 10 +- .../src/lib/components/ImportSearch.svelte | 85 +++++++------ .../src/lib/components/ImportSearch.test.ts | 112 ++++++++++++++++-- frontend/src/lib/i18n/locales/de.json | 13 +- frontend/src/lib/i18n/locales/en.json | 13 +- frontend/src/lib/i18n/locales/es.json | 13 +- frontend/src/lib/i18n/locales/fr.json | 13 +- frontend/src/lib/i18n/locales/zh.json | 13 +- frontend/src/lib/types.ts | 27 +++-- 9 files changed, 213 insertions(+), 86 deletions(-) diff --git a/frontend/src/lib/components/AddBookModal.svelte b/frontend/src/lib/components/AddBookModal.svelte index a176ea5f..76127178 100644 --- a/frontend/src/lib/components/AddBookModal.svelte +++ b/frontend/src/lib/components/AddBookModal.svelte @@ -228,12 +228,17 @@ {#if open}
-
-
+
+ +

{$_('app.addBook')}

+
+
= { + open_library: 'import.sourceNameOpenLibrary', + hardcover: 'import.sourceNameHardcover', + google_books: 'import.sourceNameGoogleBooks', + bnf: 'import.sourceNameBnf', + inventaire: 'import.sourceNameInventaire', + library_of_congress: 'import.sourceNameLibraryOfCongress' + }; + + function sourceLabel(stage: string): string { + const key = SOURCE_LABEL_KEY[stage]; + return key ? $_(key) : stage; + } + function stageLabel(s: SearchStage): string { - if (s.stage === 'open_library') { - if (s.status === 'searching') return $_('import.sourceOpenLibrarySearching'); - if ('reason' in s) return $_('import.sourceBackendError', { values: { source: 'Open Library' } }); - return $_('import.resultCount', { - values: { source: 'Open Library', count: s.count, suffix: s.count === 1 ? '' : 's' } - }); - } - if (s.stage === 'hardcover') { - if (s.status === 'searching') return $_('import.sourceHardcoverSearching'); - if (s.status === 'skipped') return $_('import.sourceHardcoverSkipped'); - if ('reason' in s) return $_('import.sourceBackendError', { values: { source: 'Hardcover' } }); - return $_('import.resultCount', { - values: { source: 'Hardcover', count: s.count, suffix: s.count === 1 ? '' : 's' } - }); - } - if (s.stage === 'google_books') { - if (s.status === 'searching') return $_('import.sourceGoogleSearching'); - if (s.status === 'skipped') return $_('import.sourceSkipped'); - if ('reason' in s) return $_('import.sourceBackendError', { values: { source: 'Google Books' } }); - return $_('import.resultCount', { - values: { source: 'Google Books', count: s.count, suffix: s.count === 1 ? '' : 's' } - }); - } if (s.stage === 'error') return $_('import.sourceError', { values: { message: s.message } }); - return ''; + if (s.stage === 'complete') return ''; + const source = sourceLabel(s.stage); + if (s.status === 'searching') return $_('import.sourceSearching', { values: { source } }); + if (s.status === 'skipped') { + // The backend reports why a source was skipped; surface the concrete reason. + if (s.reason === 'title_search_unsupported') return $_('import.sourceTitleSearchUnsupported', { values: { source } }); + if (s.stage === 'hardcover') return $_('import.sourceHardcoverSkipped'); + return $_('import.sourceSkipped', { values: { source } }); + } + if ('reason' in s) return $_('import.sourceBackendError', { values: { source } }); + return $_('import.resultCount', { + values: { source, count: s.count, suffix: s.count === 1 ? '' : 's' } + }); } function stageIcon(s: SearchStage): string { - if (s.stage === 'open_library' || s.stage === 'hardcover' || s.stage === 'google_books') { - if (s.status === 'searching') return '◌'; - if (s.status === 'skipped') return '—'; - if ('reason' in s) return '!'; - return '✓'; - } if (s.stage === 'error') return '✗'; - return ''; + if (s.stage === 'complete') return ''; + if (s.status === 'searching') return '◌'; + if (s.status === 'skipped') return '—'; + if ('reason' in s) return '!'; + return '✓'; } function stageClass(s: SearchStage): string { if (s.stage === 'error') return 'text-error'; - if (s.stage === 'hardcover' && s.status === 'skipped') return 'text-base-content/40'; - if (s.stage === 'google_books' && s.status === 'skipped') return 'text-base-content/40'; - if ( - (s.stage === 'open_library' || s.stage === 'hardcover' || s.stage === 'google_books') && - s.status === 'searching' - ) - return 'text-base-content/60 animate-pulse'; - if ((s.stage === 'open_library' || s.stage === 'hardcover' || s.stage === 'google_books') && 'reason' in s) { - return 'text-warning'; + if (s.stage === 'complete') return 'text-base-content/70'; + if (s.status === 'skipped') return 'text-base-content/40'; + if (s.status === 'searching') return 'text-base-content/60 animate-pulse'; + if ('reason' in s) return 'text-warning'; + return 'text-base-content/70'; } - return 'text-base-content/70'; -} function updateImportedLookups(books: Book[]) { const isbnSet = new Set(); @@ -452,7 +447,7 @@

{formatAuthors(group.authors, group.authors[0])}

{/if}
- {selected.source} + {sourceLabel(selected.source)} {#if selected.published_year} · {selected.published_year} @@ -517,7 +512,7 @@