-**Multi-user book tracking webapp** — maintain four reading lists, import books from Open Library & Google Books, scrape cover art, and get rich reading analytics — all on your own hardware.
+**Multi-user book tracking webapp** — maintain four reading lists, import books from six metadata sources, scrape cover art, and get rich reading analytics — all on your own hardware.
> `docker compose up -d` → full data ownership, no vendor lock-in, no API keys required.
@@ -62,7 +62,7 @@ Open **http://localhost:8001** and create your account.
## Why LibrisLog?
- **Your data, your rules.** Fully self-hosted under MIT license — no ads, no user tracking, no vendor lock-in. A single SQLite file you can back up anytime.
-- **No API keys required.** Works with Open Library out of the box. Add Google Books or Hardcover.app tokens optionally for richer search results.
+- **No API keys required.** Four sources work out of the box: Open Library, BnF, Inventaire.io, and the Library of Congress. BnF and Inventaire also find books that Open Library misses, which matters for non-English editions. Add Google Books or Hardcover.app tokens optionally (both are off in `.env.example`; set their key and their `*_IMPORT_SEARCH_ENABLED=true` to use them), and switch any source on or off individually. The Library of Congress endpoint is only reachable over plain HTTP, so it sends queries unencrypted unless you disable it — see [Configuration](https://docs.librislog.app/guide/configuration).
- **Rich insights from day one.** Calendar heatmap, language/status/page distribution charts, books finished per month/year, top authors — all on your hardware.
- **Multi-user from the start.** User roles (admin/user), optional OIDC SSO, per-user libraries. One instance works for your whole household or small group.
- **Import any format you have.** Goodreads or Bookstats exports with automatic field mapping, generic CSV or Excel (XLSX) with per-field Python transforms, JSON, ZIP with covers.
@@ -80,11 +80,13 @@ Open **http://localhost:8001** and create your account.
- **Library** — Grid/list view, search and sort, four reading statuses (Want to Read, Currently Reading, Read, Did Not Finish)
- **Reading progress** — Page-level slider, full progress timeline per book with edit/history
- **Statistics dashboard** — Calendar heatmap, distribution charts, books finished per period, top authors
-- **Book import** — Search Open Library, Google Books, Hardcover.app. Scan ISBN barcodes on mobile. Manual entry for anything not found
+- **Reading streaks & goals** — Current and all-time longest reading streak, plus optional daily, monthly, and yearly goals
+- **Book import** — Search Open Library, BnF, Inventaire.io, the Library of Congress, Hardcover.app, and Google Books in parallel, each source individually switchable. Results describing the same book are grouped so you can compare editions side by side before importing. Scan ISBN barcodes on mobile, collect candidates in a basket, or enter manually
- **Data portability** — Export as JSON, CSV, or ZIP with covers. Import from the Goodreads or Bookstats presets, or any CSV/Excel file with custom field mapping
- **Cover management** — Automatic multi-source cover search with manual override, URL paste, or file upload
- **Data hygiene** — Find and fix missing metadata (covers, page counts, authors) in bulk
- **Multi-user** — Admin/user roles, per-user libraries, optional OIDC login
+- **Public profiles** — Shareable, read-only profile links with configurable sections, audience, language, and optional expiry
- **Themes** — Light, dark, and custom DaisyUI themes with persistent preferences
- **Administration** — Full backup/restore of the SQLite database, user management, API key management
diff --git a/backend/app/config.py b/backend/app/config.py
index 9876a1d6..76bae68c 100644
--- a/backend/app/config.py
+++ b/backend/app/config.py
@@ -41,6 +41,16 @@ 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 and Google Books need credentials to answer at all. The built-in default
+ # stays true so upgrading never silently drops a source someone had configured
+ # before these switches existed; without a token/key the source is skipped anyway.
+ # .env.example and the docs recommend setting these to false when unused.
+ 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..b619cc84 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -76,9 +76,46 @@ async def _telemetry_heartbeat() -> None:
await asyncio.sleep(_TELEMETRY_INTERVAL_SECONDS)
+def _validate_import_sources() -> None:
+ """Ensure at least one import search provider can actually answer a query.
+
+ Counting enabled flags alone is not enough: Hardcover and Google Books require a
+ token/API key, so being switched on without credentials still means no source can
+ answer. This checks for a provider that is both enabled and usable.
+ """
+ from app.services import book_import
+
+ active: list[str] = []
+ for spec in book_import._build_sources(""):
+ if not spec.enabled():
+ continue
+ if spec.stage == "hardcover" and not settings.hardcover_app_api_token.strip():
+ continue
+ active.append(spec.stage)
+
+ if book_import._google_fallback_enabled() and settings.google_books_api_key.strip():
+ active.append("google_books")
+
+ if not active:
+ keyless = ", ".join(
+ sorted(
+ f"{spec.stage.upper()}_IMPORT_SEARCH_ENABLED"
+ for spec in book_import._build_sources("")
+ if spec.stage not in {"hardcover", "google_books"}
+ )
+ )
+ raise RuntimeError(
+ "No usable book import source is configured. "
+ f"Set at least one of {keyless} to true, or provide "
+ "HARDCOVER_APP_API_TOKEN with HARDCOVER_IMPORT_SEARCH_ENABLED=true, "
+ "or GOOGLE_BOOKS_API_KEY with GOOGLE_BOOKS_IMPORT_SEARCH_ENABLED=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:{formatAuthors(group.authors, group.authors[0])}
{/if}