From 99a6aa206bb66a642ca7a0f566c3bc2bdca11d83 Mon Sep 17 00:00:00 2001 From: ProcyonX1 <247258290+ProcyonX1@users.noreply.github.com> Date: Fri, 23 Oct 2026 02:37:15 +0200 Subject: [PATCH] feat(metadata): add Deutsche Nationalbibliothek (DNB) metadata provider (#1547) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## feat(metadata): add Deutsche Nationalbibliothek (DNB) metadata provider This adds the Deutsche Nationalbibliothek (DNB) as a metadata provider. It covers German editions far better than the existing providers. No API key, disabled by default, and the same pattern as the other providers. It uses DNB's public SRU interface and parses MARC21-XML. Thats one reason why this needs more code than the Google Books or Open Library providers. The others are the search (DNB has no relevance ranking) and the language filter, see below. ### Notes - **Search box.** Search-box text goes from precise to loose until a stage has hits: the complete title, then the title as a phrase, then every word in a title or a person's name, last without German articles. - **"Only show editions in my book languages"** (optional, on by default). DNB also catalogs a lot of translations. DNB filters by the searching user's Default Book Languages, else the system default. - **No covers.** DNB's cover service is behind Anubis bot protection, so the provider returns no cover URLs. - **Proxy support** already included like Open Library since #1526 ### Changes outside the provider - DNB is in the onboarding dropdown. - Calibre's `dnb-idn` identifier is mapped, so DNB results show library ownership - In the env-var docs I added only the DNB section. Regenerating the whole file on main currently also reorders unrelated sections and drops `SHELFMARK_API_KEY_READONLY` (and the Databáze knih section sits inside the Moly.hu block). ### Ideas - **Deduplication.** DNB lists every edition. Kafka's "Die Verwandlung" has around 200 German editions of the same text. Grouping them would touch the API, the frontend and release search. - **Language-aware metadata search** for all providers. The other providers ignore Default Book Languages in metadata search. The `user_id` added here could serve them too. --- docs/environment-variables.md | 42 + scripts/generate_env_docs.py | 1 + shelfmark/core/library_providers/calibre.py | 1 + shelfmark/core/onboarding.py | 5 + shelfmark/main.py | 2 +- shelfmark/metadata_providers/README.md | 2 + shelfmark/metadata_providers/__init__.py | 5 + shelfmark/metadata_providers/dnb.py | 1032 +++++++++++++++++++ tests/core/test_library_provider_calibre.py | 3 +- tests/core/test_metadata_search_library.py | 24 + tests/core/test_onboarding.py | 41 + tests/metadata/test_dnb_parse.py | 759 ++++++++++++++ 12 files changed, 1915 insertions(+), 2 deletions(-) create mode 100644 shelfmark/metadata_providers/dnb.py create mode 100644 tests/metadata/test_dnb_parse.py diff --git a/docs/environment-variables.md b/docs/environment-variables.md index 5ae2540e..8f9aa6ba 100644 --- a/docs/environment-variables.md +++ b/docs/environment-variables.md @@ -28,6 +28,7 @@ This document lists all configuration options that can be set via environment va - [Google Books](#metadata-providers-google-books) - [Moly.hu](#metadata-providers-moly.hu) - [Databáze knih](#metadata-providers-databáze-knih) + - [DNB](#metadata-providers-dnb) - [Direct Download](#direct-download) - [Download Sources](#direct-download-download-sources) - [Cloudflare Bypass](#direct-download-cloudflare-bypass) @@ -2742,6 +2743,47 @@ Enable Moly.hu as a metadata provider for book searches +### Metadata Providers: DNB + +| Variable | Description | Type | Default | +|----------|-------------|------|---------| +| `DNB_ENABLED` | Enable DNB as a metadata provider for book searches | boolean | `false` | +| `DNB_DEFAULT_SORT` | Default sort order for DNB search results. | string (choice) | `relevance` | +| `DNB_USE_BOOK_LANGUAGES` | Filter DNB results by Default Book Languages: each user's own setting, else the system default. DNB also catalogs translations into many other languages. | boolean | `true` | + +
+Detailed descriptions + +#### `DNB_ENABLED` + +**Enable DNB** + +Enable DNB as a metadata provider for book searches + +- **Type:** boolean +- **Default:** `false` + +#### `DNB_DEFAULT_SORT` + +**Default Sort Order** + +Default sort order for DNB search results. + +- **Type:** string (choice) +- **Default:** `relevance` +- **Options:** `relevance` (Most relevant), `newest` (Newest), `oldest` (Oldest) + +#### `DNB_USE_BOOK_LANGUAGES` + +**Only show editions in my book languages** + +Filter DNB results by Default Book Languages: each user's own setting, else the system default. DNB also catalogs translations into many other languages. + +- **Type:** boolean +- **Default:** `true` + +
+ ## Direct Download ### Direct Download: Download Sources diff --git a/scripts/generate_env_docs.py b/scripts/generate_env_docs.py index dbca3226..8094a34c 100755 --- a/scripts/generate_env_docs.py +++ b/scripts/generate_env_docs.py @@ -362,6 +362,7 @@ def generate_env_docs() -> str: # Import settings modules to ensure all settings are registered import shelfmark.config.security import shelfmark.config.settings + import shelfmark.metadata_providers.dnb import shelfmark.metadata_providers.googlebooks import shelfmark.metadata_providers.hardcover import shelfmark.metadata_providers.openlibrary diff --git a/shelfmark/core/library_providers/calibre.py b/shelfmark/core/library_providers/calibre.py index 1c0611af..80ecc017 100644 --- a/shelfmark/core/library_providers/calibre.py +++ b/shelfmark/core/library_providers/calibre.py @@ -39,6 +39,7 @@ _ISBN_TYPES = frozenset({"isbn", "isbn13", "isbn-13", "isbn10", "isbn-10"}) _ASIN_TYPES = frozenset({"amazon", "mobi-asin", "asin"}) # Calibre identifier type -> Shelfmark metadata provider name. _EXTERNAL_ID_TYPES = { + "dnb-idn": "dnb", "hardcover-id": "hardcover", "google": "googlebooks", "openlibrary": "openlibrary", diff --git a/shelfmark/core/onboarding.py b/shelfmark/core/onboarding.py index 989f1c7e..9a576735 100644 --- a/shelfmark/core/onboarding.py +++ b/shelfmark/core/onboarding.py @@ -265,6 +265,11 @@ def get_metadata_provider_fields() -> list[SettingsField]: "label": "Google Books", "description": "Google's book database with good coverage. Requires free API key.", }, + { + "value": "dnb", + "label": "DNB (Deutsche Nationalbibliothek)", + "description": "German national library catalog with the best coverage of German-language editions. No API key required.", + }, ] # Clone with onboarding-specific options and default diff --git a/shelfmark/main.py b/shelfmark/main.py index 79b30484..72f2c642 100644 --- a/shelfmark/main.py +++ b/shelfmark/main.py @@ -2984,7 +2984,7 @@ def api_metadata_search() -> Response | tuple[Response, int]: return jsonify({"error": "Either 'query' or search field values are required"}), 400 options = MetadataSearchOptions( - query=query, limit=limit, page=page, sort=sort_order, fields=fields + query=query, limit=limit, page=page, sort=sort_order, fields=fields, user_id=db_user_id ) search_result = provider.search_paginated(options) diff --git a/shelfmark/metadata_providers/README.md b/shelfmark/metadata_providers/README.md index 5a789441..61607177 100644 --- a/shelfmark/metadata_providers/README.md +++ b/shelfmark/metadata_providers/README.md @@ -13,6 +13,7 @@ Metadata providers allow searching for books and retrieving detailed metadata (t | **Hardcover** | Yes (API key) | Modern book tracking platform with GraphQL API. Get your key at [hardcover.app/account/api](https://hardcover.app/account/api) | | **Open Library** | No | Free, open-source library catalog from the Internet Archive. Rate limited to ~100 requests/minute | | **Google Books** | Yes (API key) | Google's book database with broad coverage and a free API key option | +| **DNB** | No | Deutsche Nationalbibliothek, the German national library catalog. Best coverage for German-language editions | ## Core Components @@ -68,6 +69,7 @@ class MetadataSearchOptions: sort: SortOrder = SortOrder.RELEVANCE limit: int = 40 page: int = 1 + user_id: int | None = None # Searching user, for per-user settings ``` ### SortOrder diff --git a/shelfmark/metadata_providers/__init__.py b/shelfmark/metadata_providers/__init__.py index 77aca6f9..fa1188ab 100644 --- a/shelfmark/metadata_providers/__init__.py +++ b/shelfmark/metadata_providers/__init__.py @@ -172,6 +172,8 @@ class MetadataSearchOptions: limit: int = 40 page: int = 1 fields: dict[str, Any] = field(default_factory=dict) # Custom search field values + # Searching user, for per-user settings; kept out of repr and thus out of cache keys + user_id: int | None = field(default=None, repr=False) @dataclass @@ -719,3 +721,6 @@ with suppress(ImportError): with suppress(ImportError): from shelfmark.metadata_providers import databazeknih as databazeknih + +with suppress(ImportError): + from shelfmark.metadata_providers import dnb as dnb diff --git a/shelfmark/metadata_providers/dnb.py b/shelfmark/metadata_providers/dnb.py new file mode 100644 index 00000000..99a64fd8 --- /dev/null +++ b/shelfmark/metadata_providers/dnb.py @@ -0,0 +1,1032 @@ +"""Deutsche Nationalbibliothek (DNB) metadata provider via its SRU API. No API key required. + +Record heuristics adapted from the Calibre DNB_DE plugin and its Calibre-Web / +LazyLibrarian port (GPL v3). API: https://www.dnb.de/sru +""" + +import re +import unicodedata +from dataclasses import replace +from typing import TYPE_CHECKING, Any, ClassVar + +import requests +from bs4 import BeautifulSoup +from defusedxml import ElementTree as DefusedElementTree +from defusedxml.common import DefusedXmlException + +from shelfmark.core.cache import cacheable +from shelfmark.core.config import config as app_config +from shelfmark.core.languages import normalize_language +from shelfmark.core.logger import setup_logger +from shelfmark.core.settings_registry import ( + ActionButton, + CheckboxField, + HeadingField, + SelectField, + SettingsField, + register_settings, +) +from shelfmark.download.network import get_proxies, get_ssl_verify +from shelfmark.metadata_providers import ( + BookMetadata, + DisplayField, + MetadataProvider, + MetadataSearchOptions, + SearchField, + SearchResult, + SearchType, + SortOrder, + TextSearchField, + register_provider, +) +from shelfmark.metadata_providers.openlibrary import RateLimiter + +if TYPE_CHECKING: + from xml.etree.ElementTree import Element + +logger = setup_logger(__name__) + +DNB_SRU_URL = "https://services.dnb.de/sru/dnb" +DNB_PERMALINK_URL = "https://d-nb.info/" +# No covers: DNB's cover service is behind bot protection (Anubis) + +# Allowed hosts for blurb links in field 856 (deposit.dnb.de is legacy) +_BLURB_URL_PREFIXES = ( + "https://services.dnb.de/plus/", + "http://deposit.dnb.de/", + "https://deposit.dnb.de/", +) + +USER_AGENT = "Shelfmark (DNB metadata provider; +https://github.com/calibrain/shelfmark)" + +# DNB publishes no limit; stay polite +RATE_LIMIT_REQUESTS = 60 +RATE_LIMIT_WINDOW_SECONDS = 60 + +# SRU caps maximumRecords at 100 +MAX_RECORDS_PER_REQUEST = 100 +ISBN_13_LENGTH = 13 +# The number index can match more than the book itself +ISBN_LOOKUP_RECORDS = 5 + +# Plausibility bounds for parsed years and page counts +MIN_YEAR = 1400 +MAX_YEAR = 2100 +MAX_PAGES = 100_000 + +# Relator codes ($4) that never mean author (translator, narrator, ...) +_NON_AUTHOR_RELATORS = frozenset( + { + "act", + "aui", + "cmp", + "cnd", + "cov", + "ill", + "itr", + "nrt", + "pht", + "prf", + "sng", + "spk", + "trl", + "wpr", + } +) + +# Page counts such as 336 Seiten, 1.024 S. or 224 pages +_PAGES_RE = re.compile(r"(\d{1,3}(?:\.\d{3})+|\d+)\s*(?:Seiten|S\.|pages?\b)", re.IGNORECASE) + +# German transliterations, so "Schaetzing" and "Schätzing" compare equal +_UMLAUTS = str.maketrans({"ä": "ae", "ö": "oe", "ü": "ue"}) + +# MARC 21 language codes, the only form DNB's spr index accepts (German: ger, not deu) +_MARC_LANGUAGE_CODES = { + "af": "afr", + "ar": "ara", + "az": "aze", + "ba": "bak", + "be": "bel", + "bg": "bul", + "bn": "ben", + "bo": "tib", + "bs": "bos", + "ca": "cat", + "cs": "cze", + "da": "dan", + "de": "ger", + "el": "gre", + "en": "eng", + "eo": "epo", + "es": "spa", + "et": "est", + "fa": "per", + "fi": "fin", + "fil": "fil", + "fr": "fre", + "ga": "gle", + "gd": "gla", + "gl": "glg", + "gu": "guj", + "gv": "glv", + "he": "heb", + "hi": "hin", + "hr": "hrv", + "hu": "hun", + "hy": "arm", + "id": "ind", + "is": "ice", + "it": "ita", + "ja": "jpn", + "jv": "jav", + "ka": "geo", + "kk": "kaz", + "kn": "kan", + "ko": "kor", + "ky": "kir", + "la": "lat", + "lt": "lit", + "lv": "lav", + "ml": "mal", + "mn": "mon", + "mr": "mar", + "ms": "may", + "my": "bur", + "nl": "dut", + "no": "nor", + "pa": "pan", + "pl": "pol", + "pt": "por", + "qu": "que", + "ro": "rum", + "ru": "rus", + "rw": "kin", + "sa": "san", + "shn": "shn", + "sk": "slo", + "sl": "slv", + "sq": "alb", + "sr": "srp", + "sv": "swe", + "sw": "swa", + "ta": "tam", + "te": "tel", + "th": "tha", + "tr": "tur", + "ug": "uig", + "uk": "ukr", + "ur": "urd", + "vi": "vie", + "zh": "chi", + "zh-Hant": "chi", # MARC has one code for both scripts +} + +_NS = { + "srw": "http://www.loc.gov/zing/srw/", + "diag": "http://www.loc.gov/zing/srw/diagnostic/", + "marc": "http://www.loc.gov/MARC21/slim", +} + +# Films, music and microfiches share the catalog with books +_EXCLUDED_MATERIALS = "NOT (mat=film OR mat=music OR mat=microfiches)" + +# DNB has no relevance ranking; "Most relevant" is its default order (newest first) +SORT_MAPPING: dict[SortOrder, str | None] = { + SortOrder.RELEVANCE: None, + SortOrder.NEWEST: "jhr/sort.descending", + SortOrder.OLDEST: "jhr/sort.ascending", +} + +# Pseudo-index for search-box words: title or person name ("woe" also matches subjects) +_TITLE_OR_NAME = "tit|per" + +# Tried in front of search-box text +_LEADING_ARTICLES = ("Der", "Die", "Das", "Ein", "Eine") + +# Dropped in the word-by-word fallback +_GERMAN_JOINERS = frozenset({"ein", "eine", "einer", "der", "die", "das", "und", "oder"}) + +# Publisher imprints DNB records as series (list from the Calibre DNB_DE plugin) +_UNWANTED_SERIES_RE = re.compile( + # Names that start an imprint series + r"^(?:Allgemeine|Ariadne-Krimi|Bastei[ -]L|Beck Paperback|Beck-.*berater|Beck'sche Reihe" + r"|bt-Kinder|C.-Bertelsmann|C.H. Beck|dtv|Edition C|Ein Goldmann|Knaur|Premium|Ravensburger" + r"|Rororo|Ullstein|Unionsverlag|Volk-und-Welt)" + # Complete imprint names + r"|^(?:Roman|Science-fiction|\[Ariadne\]|Ariadne|atb|BvT|bb|Bibliothek Suhrkamp|BLT" + r"|DLV-Taschenbuch|Edition Suhrkamp|Edition Lingen Stiftung|Edition Metzgenstein|ETB" + r"|Oettinger-Taschenbuch|Haymon-Taschenbuch|Mira Taschenbuch|Suhrkamp-Taschenbuch|Hey|btb" + r"|Sammlung Luchterhand|blanvalet|KiWi|Piper|Goldmann|Moewig|Fischer Klassik|hey! shorties" + r"|Phantastische Bibliothek|Horror-Bibliothek)$", + flags=re.IGNORECASE, +) + +_TRANSLATOR_SUFFIX_RE = re.compile(r"^(.+) [/:] [Aa]us dem .+? von(\s\w+)+$") +# "Angaben aus der Verlagsmeldung" header and the title heading after it +_BLANK_HTML = r"(?:\s||

)*" +_PUBLISHER_NOTE_RE = re.compile( + rf"{_BLANK_HTML}Angaben aus der Verlagsmeldung{_BLANK_HTML}(?:

.*?

)*{_BLANK_HTML}", + flags=re.IGNORECASE, +) + +_rate_limiter = RateLimiter(RATE_LIMIT_REQUESTS, RATE_LIMIT_WINDOW_SECONDS) + + +def _clean_text(value: str | None) -> str: + """Drop non-sorting markers, collapse whitespace, compose Unicode. + + DNB sends decomposed umlauts ('a' + U+0308), which indexers wouldn't match. + """ + if not value: + return "" + value = unicodedata.normalize("NFC", value.replace("\x98", "").replace("\x9c", "")) + return " ".join(value.split()) + + +def _strip_isbd(value: str) -> str: + """Strip trailing ISBD punctuation (' /', ' :', ' ;') left on older records.""" + return value.rstrip(" /:;,=") + + +def _sanitize_term(value: str) -> str: + """Remove characters that would break out of a quoted CQL term.""" + return " ".join(value.replace('"', " ").replace("\\", " ").split()) + + +def _parse_year(value: str | None) -> int | None: + """First plausible four-digit year in a date string ('[2022]', '©2022', '202611').""" + match = re.search(r"\d{4}", value or "") + if not match: + return None + year = int(match.group()) + return year if MIN_YEAR <= year <= MAX_YEAR else None + + +def _name_tokens(value: str) -> set[str]: + """Comparable name parts: casefolded, transliterated, accents and initials dropped.""" + folded = unicodedata.normalize("NFC", value).casefold().translate(_UMLAUTS) + decomposed = unicodedata.normalize("NFKD", folded) + plain = "".join(char for char in decomposed if not unicodedata.combining(char)) + return set(re.findall(r"\w{2,}", plain)) + + +def _filter_by_author(books: list[BookMetadata], author: str) -> list[BookMetadata]: + """Drop books by other people (DNB's person index also matches subjects). + + One shared name part keeps a book; a page without any match stays unfiltered (pseudonyms). + """ + wanted = _name_tokens(author) + matching = [book for book in books if any(wanted & _name_tokens(name) for name in book.authors)] + return matching or books + + +def _book_languages(user_id: int | None) -> list[str]: + """The user's Default Book Languages (else the system default), if DNB should use them.""" + if app_config.get("DNB_USE_BOOK_LANGUAGES", True) is not True: + return [] + value = app_config.get("BOOK_LANGUAGE", [], user_id=user_id) + if isinstance(value, str): + value = value.split(",") + if not isinstance(value, list): + return [] + return [text for item in value if (text := str(item).strip())] + + +def _language_clause(languages: list[str]) -> str | None: + """CQL language filter, or None for all; languages without a MARC code are skipped.""" + codes = dict.fromkeys( + marc + for language in languages + if (marc := _MARC_LANGUAGE_CODES.get(normalize_language(language) or "")) + ) + if not codes: + return None + return "(" + " OR ".join(f"spr={code}" for code in codes) + ")" + + +def _parse_number(value: str | None) -> float | None: + if not value: + return None + match = re.search(r"\d+(?:[.,]\d+)?", value) + if not match: + return None + return float(match.group().replace(",", ".")) + + +def _valid_isbn(candidate: str) -> str | None: + """Return a normalized ISBN-10/13 (digits, with optional X check digit), else None.""" + digits = candidate.replace("-", "").strip() + if len(digits) == ISBN_13_LENGTH and digits.isdigit(): + return digits + if len(digits) == 10 and re.fullmatch(r"\d{9}[\dXx]", digits): + return digits.upper() + return None + + +def _format_author(name: str) -> str: + """Turn DNB's 'Rowling, J. K. [Hrsg.]' (or 'Rowling, J. K.,') into 'J. K. Rowling'.""" + name = re.sub(r" \[.*\]$", "", name).rstrip(" ,") + return re.sub(r"^([^,]+), (.+)$", r"\2 \1", name) + + +def _clean_title(title: str) -> str: + """Remove a trailing translator note ('... / aus dem Englischen von ...').""" + match = _TRANSLATOR_SUFFIX_RE.search(title) + return match.group(1) if match else title + + +def _drop_interpolations(value: str) -> str: + """Remove cataloguer insertions: '20000 [Zwanzigtausend] Meilen' -> '20000 Meilen'.""" + cleaned = re.sub(r"\s*\[[^\]]*\]", "", value).strip(" ;:,/") + return cleaned or value.strip("[] ") + + +def _clean_publisher(name: str | None) -> str | None: + """Publisher without bracketed notes ('[u.a.]', '[Vertrieb]'); '[s. n.]' means none.""" + if not name: + return None + name = name.replace("@", "").strip() # "@" is a sorting marker in older records + if re.fullmatch(r"\[?s\.\s*n\.\]?", name, flags=re.IGNORECASE): # sine nomine + return None + return _strip_isbd(_drop_interpolations(name)).strip("[] ") or None + + +def _clean_series(series: str | None, publisher: str | None) -> str | None: + """Return the series name, or None if it is really a publisher imprint.""" + series = _strip_isbd(_clean_text(series)) + if not series: + return None + + if publisher: + if publisher.lower() == series.lower(): + return None + match = re.search(r"^(\w\w\w\w+)", publisher) + if match and re.search(r"^\W*" + re.escape(match.group(1)), series, flags=re.IGNORECASE): + return None + + return None if _UNWANTED_SERIES_RE.search(series) else series + + +def _strip_german_joiners(words: list[str]) -> list[str]: + return [word for word in words if word.lower() not in _GERMAN_JOINERS] + + +def _cql_term(index: str, value: str) -> str: + if index == _TITLE_OR_NAME: + return f'(tit="{value}" OR per="{value}")' + return f'{index}="{value}"' + + +def _cql_words(index: str, value: str) -> str: + words = value.split() + if index != "per": + words = _strip_german_joiners(words) or words + return " AND ".join(_cql_term(index, word) for word in words) + + +def _build_cql_variants(terms: list[tuple[str, str]]) -> list[tuple[str, str]]: + """(hit-count probe, query) pairs from precise to loose; the first probe with hits wins.""" + if len(terms) == 1 and terms[0][0] == _TITLE_OR_NAME: + return _search_box_variants(terms[0][1]) + exact = " AND ".join(_cql_term(index, value) for index, value in terms) + loose = " AND ".join(_cql_words(index, value) for index, value in terms) + return [(exact, exact)] if loose == exact else [(exact, exact), (loose, loose)] + + +def _search_box_variants(text: str) -> list[tuple[str, str]]: + """Variants for search-box text (a title, a name, or both). + + DNB has no relevance ranking, so precision comes from the query. Title + stages also include records where a word is a person's name: + 1. complete title, also with a leading article ("Das Boot", "Herr der Ringe") + 2. title phrase ("Remarque Im Westen nichts Neues") + 3. every word in a title or name ("Kafka Verwandlung") + 4. the same without German articles + """ + words = text.split() + every_word = " AND ".join(_cql_term(_TITLE_OR_NAME, word) for word in words) + some_name = " OR ".join(_cql_term("per", word) for word in words) + + def with_names(title_match: str) -> tuple[str, str]: + return title_match, f"({title_match} OR ({every_word} AND ({some_name})))" + + forms = [text] + if words[0].lower() not in {article.lower() for article in _LEADING_ARTICLES}: + forms += [f"{article} {text}" for article in _LEADING_ARTICLES] + variants = [with_names("(" + " OR ".join(_cql_term("tst", form) for form in forms) + ")")] + if len(words) > 1: + variants.append(with_names(_cql_term("tit", text))) + variants.append((every_word, every_word)) + loose = _cql_words(_TITLE_OR_NAME, text) + if loose != every_word: + variants.append((loose, loose)) + return variants + + +def _blurb_to_text(html: str) -> str | None: + """Convert a DNB blurb page to plain text with paragraph breaks.""" + if "Zugriff derzeit nicht möglich" in html: + return None + # Only
and block ends separate lines + html = _PUBLISHER_NOTE_RE.sub("", " ".join(html.split())) + soup = BeautifulSoup(html, "html.parser") + for br in soup.find_all("br"): + br.replace_with("\n") + for block in soup.find_all(["p", "div", "h3"]): + block.append("\n") + lines = [" ".join(line.split()) for line in soup.get_text().splitlines()] + text = re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip() + return text or None + + +def _datafields(record: Element, tag: str) -> list[Element]: + return record.findall(f"marc:datafield[@tag='{tag}']", _NS) + + +def _subfields(field: Element, code: str) -> list[str]: + subfields = field.findall(f"marc:subfield[@code='{code}']", _NS) + return [value for sub in subfields if (value := _clean_text(sub.text))] + + +def _first_subfield(field: Element, code: str) -> str | None: + values = _subfields(field, code) + return values[0] if values else None + + +def _record_idn(record: Element) -> str: + """The record's IDN from 001, or from 016 on records without one.""" + idn = _clean_text(record.findtext("marc:controlfield[@tag='001']", namespaces=_NS)) + if idn: + return idn + return next( + (value for field in _datafields(record, "016") for value in _subfields(field, "a")), "" + ) + + +def _record_isbns(record: Element) -> list[str]: + """Every valid ISBN in 020 $a, in record order.""" + return [ + isbn + for field in _datafields(record, "020") + for value in _subfields(field, "a") + for candidate in re.findall(r"[\dXx-]{10,17}", value) + if (isbn := _valid_isbn(candidate)) + ] + + +def _as_isbn13(isbn: str) -> str: + """ISBN-13 form of a normalized ISBN, so an ISBN-10 and its ISBN-13 compare equal.""" + if len(isbn) == ISBN_13_LENGTH: + return isbn + core = "978" + isbn[:9] + checksum = sum(int(digit) * (3 if position % 2 else 1) for position, digit in enumerate(core)) + return core + str(-checksum % 10) + + +@register_provider("dnb") +class DNBProvider(MetadataProvider): + """Deutsche Nationalbibliothek metadata provider (SRU API, MARC21-XML).""" + + name = "dnb" + display_name = "DNB" + requires_auth = False + supported_sorts: ClassVar[tuple[SortOrder, ...]] = ( + SortOrder.RELEVANCE, + SortOrder.NEWEST, + SortOrder.OLDEST, + ) + search_fields: ClassVar[tuple[SearchField, ...]] = ( + TextSearchField( + key="author", + label="Author", + description="Search by author name", + ), + TextSearchField( + key="title", + label="Title", + description="Search by book title", + ), + ) + + def __init__(self) -> None: + """Initialize provider.""" + self.session = requests.Session() + self.session.headers["User-Agent"] = USER_AGENT + + def is_available(self) -> bool: + """DNB needs no authentication.""" + return True + + def _fetch(self, url: str, params: dict[str, Any] | None = None) -> requests.Response | None: + _rate_limiter.wait_if_needed() + try: + response = self.session.get( + url, params=params, timeout=15, proxies=get_proxies(url), verify=get_ssl_verify(url) + ) + response.raise_for_status() + except requests.Timeout: + logger.warning("DNB request timed out: %s", url) + return None + except requests.HTTPError as e: + status = e.response.status_code if e.response is not None else "error" + logger.warning("DNB request failed with HTTP %s: %s", status, url) + return None + except requests.RequestException as e: + logger.warning("DNB request failed (%s): %s", e, url) + return None + return response + + def _query( + self, + cql: str, + *, + limit: int, + start: int = 1, + sort: SortOrder = SortOrder.RELEVANCE, + ) -> tuple[list[Element], int] | None: + """Run an SRU query: (records, total hits), or None on any failure (never cached).""" + query = f"{cql} {_EXCLUDED_MATERIALS}" + sort_key = SORT_MAPPING.get(sort) + if sort_key: + query += f" sortBy {sort_key}" + + response = self._fetch( + DNB_SRU_URL, + { + "version": "1.1", + "operation": "searchRetrieve", + "recordSchema": "MARC21-xml", + # Safety net; search_paginated clamps already + "maximumRecords": max(0, min(limit, MAX_RECORDS_PER_REQUEST)), + "startRecord": max(1, start), + "query": query, + }, + ) + if response is None: + return None + + try: + root = DefusedElementTree.fromstring(response.content) + except DefusedElementTree.ParseError, DefusedXmlException: + logger.warning("DNB returned an unparseable response for query: %s", cql) + return None + + if root.tag != f"{{{_NS['srw']}}}searchRetrieveResponse": + logger.warning("DNB returned a non-SRU response for query: %s", cql) + return None + + total = root.findtext("srw:numberOfRecords", default="", namespaces=_NS).strip() + if not total.isdigit(): + # SRU reports errors, including paging past the last hit, as diagnostics + message = root.findtext(".//diag:message", default="", namespaces=_NS) + logger.warning("DNB diagnostic for query '%s': %s", cql, message) + return None + + records = root.findall("srw:records/srw:record/srw:recordData/marc:record", _NS) + return records, int(total) + + def search(self, options: MetadataSearchOptions) -> list[BookMetadata]: + """Search the DNB catalog.""" + return self.search_paginated(options).books + + def search_paginated(self, options: MetadataSearchOptions) -> SearchResult: + """Search the DNB catalog with exact pagination from the SRU hit count.""" + # Clamp API input; a limit of 0 would page forever through empty pages + options = replace( + options, + page=max(1, options.page), + limit=max(1, min(options.limit, MAX_RECORDS_PER_REQUEST)), + ) + # A pasted ISBN goes straight to the ISBN lookup + looks_like_isbn = ( + options.search_type == SearchType.GENERAL + and not options.fields + and _valid_isbn(options.query.replace(" ", "")) is not None + ) + if options.search_type == SearchType.ISBN or looks_like_isbn: + books = [book] if (book := self.search_by_isbn(options.query)) else [] + return SearchResult(books=books, page=options.page, total_found=len(books)) + + terms = self._search_terms(options) + if not terms: + return SearchResult(books=[], page=options.page) + + fields_key = ":".join(f"{k}={v}" for k, v in sorted(options.fields.items())) + cache_key = ( + f"{options.query}:{options.search_type.value}:{options.sort.value}:" + f"{options.limit}:{options.page}:{fields_key}" + ) + queries = _build_cql_variants(terms) + # A language passed with the search wins over the book languages + language_clause = _language_clause( + [options.language] if options.language else _book_languages(options.user_id) + ) + if language_clause: + queries = [ + (f"{probe} AND {language_clause}", f"{cql} AND {language_clause}") + for probe, cql in queries + ] + author = dict(terms).get("per") + return self._search_cached(cache_key, queries, author, options) or SearchResult( + books=[], page=options.page + ) + + def _search_terms(self, options: MetadataSearchOptions) -> list[tuple[str, str]]: + """(SRU index, value) pairs for the search: person, title, or search-box text.""" + author = _sanitize_term(options.fields.get("author") or "") + title = _sanitize_term(options.fields.get("title") or "") + query = _sanitize_term(options.query) + + if not author and not title: + if options.search_type == SearchType.AUTHOR: + author = query + elif options.search_type == SearchType.TITLE: + title = query + + terms: list[tuple[str, str]] = [] + if author: + terms.append(("per", author)) + if title: + terms.append(("tit", title)) + if not terms and query: + terms.append((_TITLE_OR_NAME, query)) + return terms + + @cacheable(ttl_key="METADATA_CACHE_SEARCH_TTL", ttl_default=300, key_prefix="dnb:search") + def _search_cached( + self, + cache_key: str, + queries: list[tuple[str, str]], + author: str | None, + options: MetadataSearchOptions, + ) -> SearchResult | None: + # None on failure, so it isn't cached + start = (options.page - 1) * options.limit + 1 + result = self._run_search(queries, limit=options.limit, start=start, sort=options.sort) + if result is None: + return None + + cql, records, total = result + books = [book for record in records if (book := self._parse_record(record))] + if author: + books = _filter_by_author(books, author) + logger.info("DNB search '%s' returned %s results", cql, len(books)) + # Paging follows the raw records, so filtered pages don't end it early + return SearchResult( + books=books, + page=options.page, + total_found=total, + has_more=start - 1 + len(records) < total, + ) + + def _run_search( + self, queries: list[tuple[str, str]], *, limit: int, start: int, sort: SortOrder + ) -> tuple[str, list[Element], int] | None: + """Run the first variant whose probe has hits (the last one unconditionally). + + Chosen by hit count, so every page uses the same variant; a variant that + is its own probe uses its page 1 as the count. + """ + for probe, cql in queries[:-1]: + page_is_probe = start == 1 and probe == cql + result = self._query(probe, limit=limit if page_is_probe else 0, sort=sort) + if result is None: + return None + records, total = result + if total and page_is_probe: + return cql, records, total + if total: + break + else: + cql = queries[-1][1] + + result = self._query(cql, limit=limit, start=start, sort=sort) + return None if result is None else (cql, *result) + + @cacheable(ttl_key="METADATA_CACHE_BOOK_TTL", ttl_default=600, key_prefix="dnb:book") + def get_book(self, book_id: str) -> BookMetadata | None: + """Get book details by DNB IDN (e.g. '1420086006').""" + idn = book_id.strip().upper() + if not re.fullmatch(r"\d+[\dX]", idn): + return None + result = self._query(f"idn={idn}", limit=1) + records = result[0] if result else [] + record = next((record for record in records if _record_idn(record) == idn), None) + return self._parse_full_record(record) if record is not None else None + + @cacheable(ttl_key="METADATA_CACHE_BOOK_TTL", ttl_default=600, key_prefix="dnb:isbn") + def search_by_isbn(self, isbn: str) -> BookMetadata | None: + """Search for a book by ISBN-10 or ISBN-13.""" + clean_isbn = _valid_isbn(isbn.replace(" ", "")) + if not clean_isbn: + return None + # The number index also matches ISSNs etc.; require the ISBN itself (10 or 13) + wanted = _as_isbn13(clean_isbn) + result = self._query(f"num={clean_isbn}", limit=ISBN_LOOKUP_RECORDS) + records = result[0] if result else [] + record = next( + ( + record + for record in records + if wanted in {_as_isbn13(found) for found in _record_isbns(record)} + ), + None, + ) + return self._parse_full_record(record) if record is not None else None + + def _parse_full_record(self, record: Element) -> BookMetadata | None: + """Parse a record for the details view, including the linked blurb.""" + return self._parse_record(record, description=self._fetch_description(record)) + + def _fetch_description(self, record: Element) -> str | None: + """Fetch the publisher blurb linked from field 856, if any.""" + for field in _datafields(record, "856"): + for url in _subfields(field, "u"): + if not url.startswith(_BLURB_URL_PREFIXES): + continue + _rate_limiter.wait_if_needed() + try: + response = self.session.get( + url, timeout=15, proxies=get_proxies(url), verify=get_ssl_verify(url) + ) + response.raise_for_status() + except requests.RequestException as e: + logger.debug("DNB blurb fetch failed for %s: %s", url, e) + continue + description = _blurb_to_text(response.text) + if description: + return description + return None + + def _parse_record(self, record: Element, description: str | None = None) -> BookMetadata | None: + """Parse a MARC21 record into BookMetadata.""" + # Videos can slip past the material filter + if any( + (_first_subfield(field, "a") or "").lower() == "video" + for field in _datafields(record, "337") + ): + return None + + idn = _record_idn(record) + publisher, publish_year = self._parse_publication(record) + title, subtitle, series_name, series_position = self._parse_title(record, publisher) + authors = self._parse_authors(record) + if not idn or not title or not authors: + return None + + # Series: multi-part title, then 830/490, then a "(Series 3)" suffix (always stripped) + if not series_name: + series_name, series_position = self._parse_series_statement(record, publisher) + guessed = self._guess_series_from_title(title, publisher) + if guessed: + title, guessed_name, guessed_position = guessed + if not series_name: + series_name, series_position = guessed_name, guessed_position + + isbns = _record_isbns(record) + isbn_13 = next((isbn for isbn in isbns if len(isbn) == ISBN_13_LENGTH), None) + isbn_10 = next((isbn for isbn in isbns if len(isbn) != ISBN_13_LENGTH), None) + + display_fields: list[DisplayField] = [] + pages = self._parse_pages(record) + if pages: + display_fields.append(DisplayField(label="Pages", value=pages, icon="book")) + + return BookMetadata( + provider=self.name, + provider_id=idn, + provider_display_name=self.display_name, + title=title, + subtitle=subtitle, + authors=authors, + isbn_13=isbn_13, + isbn_10=isbn_10, + # Blurb first, 520 summary as fallback + description=description or self._parse_summary(record), + publisher=publisher, + publish_year=publish_year, + language=self._parse_language(record), + genres=self._parse_subjects(record)[:5], + source_url=DNB_PERMALINK_URL + idn, + search_title=title, + search_author=authors[0], + series_name=series_name, + series_position=series_position, + display_fields=display_fields, + ) + + def _parse_title( + self, record: Element, publisher: str | None + ) -> tuple[str | None, str | None, str | None, float | None]: + """Parse field 245 into (title, subtitle, series, series position). + + Multi-part works are catalogued as '$a Series $n 3 $p Part title'. + """ + fields = _datafields(record, "245") + if not fields: + return None, None, None, None + field = fields[0] + + code_a = [_strip_isbd(_drop_interpolations(value)) for value in _subfields(field, "a")] + code_n = [ + match.group() + for value in _subfields(field, "n") + if (match := re.search(r"\d+(?:[,.]\d+)?", value)) + ] + code_p = [_strip_isbd(_drop_interpolations(value)) for value in _subfields(field, "p")] + + title_parts = code_a + series = None + position = None + if code_a and code_n: + if code_p: + title_parts = [code_p[-1]] + series_parts = [code_a[0], *code_p[: min(len(code_p), len(code_n)) - 1]] + for i in range(min(len(series_parts), len(code_n) - 1)): + series_parts[i] += f" {code_n[i]}" + series = _clean_series(" - ".join(series_parts), publisher) + position = _parse_number(code_n[-1]) + + title = _clean_title(" : ".join(part for part in title_parts if part)) or None + subtitle = _first_subfield(field, "b") + subtitle = _clean_title(_strip_isbd(_drop_interpolations(subtitle))) if subtitle else None + return title, subtitle or None, series, position if series else None + + def _parse_series_statement( + self, record: Element, publisher: str | None + ) -> tuple[str | None, float | None]: + """Series from 830 (authorized form) or 490 (as printed), with a volume number.""" + for tag in ("830", "490"): + for field in _datafields(record, tag): + position = _parse_number(_first_subfield(field, "v")) + if position is None: + continue + series = _clean_series(_first_subfield(field, "a"), publisher) + if series: + return series, position + return None, None + + def _guess_series_from_title( + self, title: str, publisher: str | None + ) -> tuple[str, str, float] | None: + """Split 'Title (Series 3)' into (title, series, position).""" + parts = title.split(" (") + if len(parts) != 2: + return None + text_part, index_part = parts + try: + series, index = index_part.split(")", 1)[0].rsplit(" ", 1) + except ValueError: + return None + series = _clean_series(series, publisher) + position = _parse_number(index) if index[-1:].isdigit() else None + if not series or position is None: + return None + text_part = text_part.strip(" -–—:") + return (text_part or title), series, position + + def _parse_authors(self, record: Element) -> list[str]: + """Authors (100/700 with $4 aut), else everyone except translators, narrators etc.""" + persons = [*_datafields(record, "100"), *_datafields(record, "700")] + authors = [field for field in persons if "aut" in _subfields(field, "4")] + if not authors: + authors = [ + field + for field in persons + if not _NON_AUTHOR_RELATORS.intersection(_subfields(field, "4")) + ] + names = [name for field in authors for name in _subfields(field, "a")] + return list(dict.fromkeys(_format_author(name) for name in names)) + + def _parse_publication(self, record: Element) -> tuple[str | None, int | None]: + """Publisher and year. + + - Publisher only from publication statements (264 ind2=1, 260), never printers. + - Year from those first, then other 264s, 263 (announced) and 008. + """ + statements = sorted( + [*_datafields(record, "264"), *_datafields(record, "260")], + key=lambda field: (field.get("tag"), field.get("ind2")) != ("264", "1"), + ) + publisher = None + year = None + for field in statements: + is_publication = field.get("tag") == "260" or field.get("ind2") == "1" + if is_publication and not publisher: + name = _first_subfield(field, "b") + publisher = _clean_publisher(name) + year = year or _parse_year(_first_subfield(field, "c")) + for field in _datafields(record, "263"): + year = year or _parse_year(_first_subfield(field, "a")) + if not year: + # Positions 7-10 hold the (first) publication date + fixed = record.findtext("marc:controlfield[@tag='008']", default="", namespaces=_NS) + year = _parse_year(fixed[7:11]) + return publisher or None, year + + def _parse_language(self, record: Element) -> str | None: + for field in _datafields(record, "041"): + for code in _subfields(field, "a"): + language = normalize_language(code) + if language: + return language + # 008 positions 35-37 + fixed = record.findtext("marc:controlfield[@tag='008']", default="", namespaces=_NS) + return normalize_language(fixed[35:38]) if len(fixed) >= 38 else None + + def _parse_subjects(self, record: Element) -> list[str]: + """GND subject headings first, then other 6XX subject terms.""" + gnd: list[str] = [] + other: list[str] = [] + for field in _datafields(record, "689"): + gnd.extend(_subfields(field, "a")) + for field in record.findall("marc:datafield", _NS): + tag = field.get("tag", "") + # 653: publisher keywords, not subjects + if not (tag.isdigit() and 600 <= int(tag) <= 655) or tag == "653": + continue + # Skip qualified terms like (Produktform)Paperback + terms = [term for term in _subfields(field, "a") if not term.startswith("(")] + if "gnd" in _subfields(field, "2"): + gnd.extend(terms) + for term in terms: + other.extend(part.strip() for part in re.split(r"[,;]", term)) + return list(dict.fromkeys(term for term in [*gnd, *other] if len(term) >= 2)) + + def _parse_pages(self, record: Element) -> str | None: + for field in _datafields(record, "300"): + for extent in _subfields(field, "a"): + match = _PAGES_RE.search(extent) + if match and 0 < (pages := int(match.group(1).replace(".", ""))) <= MAX_PAGES: + return str(pages) + return None + + def _parse_summary(self, record: Element) -> str | None: + summaries = [ + text for field in _datafields(record, "520") for text in _subfields(field, "a") + ] + summary = "\n\n".join(summaries) + # Some summaries are HTML + return (_blurb_to_text(summary) if "<" in summary else summary) or None + + +def _test_dnb_connection() -> dict[str, Any]: + """Test connectivity to the DNB SRU interface.""" + if DNBProvider()._query('tit="test"', limit=0) is None: + return {"success": False, "message": "Could not reach the DNB catalog (details in the log)"} + return {"success": True, "message": "Successfully connected to the DNB catalog"} + + +# DNB sort options for settings UI +_DNB_SORT_OPTIONS = [ + {"value": "relevance", "label": "Most relevant"}, + {"value": "newest", "label": "Newest"}, + {"value": "oldest", "label": "Oldest"}, +] + + +@register_settings("dnb", "DNB", icon="library", order=55, group="metadata_providers") +def dnb_settings() -> list[SettingsField]: + """DNB metadata provider settings.""" + return [ + HeadingField( + key="dnb_heading", + title="Deutsche Nationalbibliothek", + description=( + "The German National Library catalogs every book published in Germany, " + "plus German-language and translated works from abroad. " + "No API key required." + ), + link_url="https://www.dnb.de", + link_text="dnb.de", + ), + CheckboxField( + key="DNB_ENABLED", + label="Enable DNB", + description="Enable DNB as a metadata provider for book searches", + default=False, + ), + ActionButton( + key="test_connection", + label="Test Connection", + description="Verify the DNB catalog is accessible", + style="primary", + callback=_test_dnb_connection, + ), + SelectField( + key="DNB_DEFAULT_SORT", + label="Default Sort Order", + description="Default sort order for DNB search results.", + options=_DNB_SORT_OPTIONS, + default="relevance", + ), + CheckboxField( + key="DNB_USE_BOOK_LANGUAGES", + label="Only show editions in my book languages", + description=( + "Filter DNB results by Default Book Languages: each user's own setting, else " + "the system default. DNB also catalogs translations into many other languages." + ), + default=True, + ), + ] diff --git a/tests/core/test_library_provider_calibre.py b/tests/core/test_library_provider_calibre.py index 6894c84f..a6e8d822 100644 --- a/tests/core/test_library_provider_calibre.py +++ b/tests/core/test_library_provider_calibre.py @@ -42,6 +42,7 @@ _IDENTIFIERS = [ (2, "goodreads", "12067"), (4, "isbn-13", "978-0-8041-3902-1"), (4, "openlibrary", "OL17091839W"), + (4, "dnb-idn", "1061425843"), ] @@ -119,7 +120,7 @@ def test_indexes_titles_authors_series_and_identifiers( martian = _by_token(entries, "martian") assert martian.isbns == {"9780804139021"} - assert martian.external_ids == {("openlibrary", "OL17091839W")} + assert martian.external_ids == {("openlibrary", "OL17091839W"), ("dnb", "1061425843")} draft = _by_token(entries, "draft") assert draft.tokens == {"untitled", "draft"} diff --git a/tests/core/test_metadata_search_library.py b/tests/core/test_metadata_search_library.py index 17bba7f4..b188e09e 100644 --- a/tests/core/test_metadata_search_library.py +++ b/tests/core/test_metadata_search_library.py @@ -69,3 +69,27 @@ def test_search_results_omit_library_when_no_check_is_enabled(main_module, clien result = _search(main_module, client, lambda book: None) assert "library" not in result + + +def test_search_passes_the_logged_in_user_to_the_provider(main_module, client): + # Providers may resolve per-user settings, e.g. DNB's book languages + seen = [] + + class _RecordingProvider(_StubProvider): + def search_paginated(self, options): + seen.append(options.user_id) + return super().search_paginated(options) + + with client.session_transaction() as session: + session["db_user_id"] = 7 + with ( + patch.object(main_module, "get_auth_mode", return_value="none"), + patch("shelfmark.metadata_providers.is_provider_registered", return_value=True), + patch("shelfmark.metadata_providers.is_provider_enabled", return_value=True), + patch("shelfmark.metadata_providers.get_provider_kwargs", return_value={}), + patch("shelfmark.metadata_providers.get_provider", return_value=_RecordingProvider([])), + ): + response = client.get("/api/metadata/search?query=carl&provider=stub") + + assert response.status_code == 200, response.get_json() + assert seen == [7] diff --git a/tests/core/test_onboarding.py b/tests/core/test_onboarding.py index 2b14d515..99587d1e 100644 --- a/tests/core/test_onboarding.py +++ b/tests/core/test_onboarding.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections import defaultdict import shelfmark.config.settings as settings_config +import shelfmark.metadata_providers.dnb as dnb_provider import shelfmark.metadata_providers.googlebooks as googlebooks_provider import shelfmark.metadata_providers.hardcover as hardcover_provider import shelfmark.metadata_providers.openlibrary as openlibrary_provider @@ -12,6 +13,7 @@ import shelfmark.release_sources.prowlarr.settings as prowlarr_settings _REGISTERED_SETTINGS_MODULES = ( settings_config, + dnb_provider, googlebooks_provider, hardcover_provider, openlibrary_provider, @@ -178,3 +180,42 @@ def test_save_onboarding_settings_skips_hidden_fields(monkeypatch): assert all("EXT_BYPASSER_URL" not in payload for payload in bypass_payloads) assert all("EXT_BYPASSER_PATH" not in payload for payload in bypass_payloads) assert all("EXT_BYPASSER_TIMEOUT" not in payload for payload in bypass_payloads) + + +def test_onboarding_offers_dnb(): + from shelfmark.core.onboarding import get_metadata_provider_fields + + provider_field = next( + field + for field in get_metadata_provider_fields() + if getattr(field, "key", None) == "METADATA_PROVIDER" + ) + + assert "dnb" in {option["value"] for option in provider_field.options} + + +def test_save_onboarding_settings_enables_dnb(monkeypatch): + import shelfmark.core.config as app_config + import shelfmark.core.onboarding as onboarding + + save_calls: list[tuple[str, dict[str, object]]] = [] + + def _fake_save(tab_name: str, values: dict[str, object]) -> bool: + save_calls.append((tab_name, dict(values))) + return True + + monkeypatch.setattr(onboarding, "save_config_file", _fake_save) + monkeypatch.setattr(onboarding, "mark_onboarding_complete", lambda: True) + monkeypatch.setattr(app_config.config, "refresh", lambda: None, raising=False) + + result = onboarding.save_onboarding_settings( + {"SEARCH_MODE": "universal", "METADATA_PROVIDER": "dnb"} + ) + + grouped_calls = _group_save_calls(save_calls) + + assert result == {"success": True, "message": "Onboarding complete!"} + assert {"DNB_ENABLED": True} in grouped_calls["dnb"] + assert any( + payload.get("METADATA_PROVIDER") == "dnb" for payload in grouped_calls["search_mode"] + ) diff --git a/tests/metadata/test_dnb_parse.py b/tests/metadata/test_dnb_parse.py new file mode 100644 index 00000000..a4a71ede --- /dev/null +++ b/tests/metadata/test_dnb_parse.py @@ -0,0 +1,759 @@ +import unicodedata + +import pytest +import requests + +import shelfmark.metadata_providers.dnb as dnb +from shelfmark.core.cache import get_metadata_cache +from shelfmark.core.languages import known_language_codes +from shelfmark.metadata_providers import ( + MetadataSearchOptions, + SearchType, + SortOrder, + get_provider_sort_options, +) +from shelfmark.metadata_providers.dnb import ( + DNBProvider, + _clean_series, + _format_author, + dnb_settings, +) + +EXCLUDED = " NOT (mat=film OR mat=music OR mat=microfiches)" +UMLAUT = "\N{COMBINING DIAERESIS}" # DNB writes "ä" as "a" + this mark + + +@pytest.fixture(autouse=True) +def _isolated_provider(monkeypatch): + get_metadata_cache().clear() + # The limiter is shared across tests; past 60 fake requests it would sleep a minute + monkeypatch.setattr(dnb._rate_limiter, "wait_if_needed", lambda: None) + # Independent of a local /config; the book_languages fixture turns the filter on + original_get = dnb.app_config.get + monkeypatch.setattr( + dnb.app_config, + "get", + lambda key, *args, **kwargs: ( + False if key == "DNB_USE_BOOK_LANGUAGES" else original_get(key, *args, **kwargs) + ), + ) + + +@pytest.fixture +def book_languages(monkeypatch): + """Set the DNB checkbox and BOOK_LANGUAGE (system-wide and per user).""" + original_get = dnb.app_config.get + + def configure(system, *, enabled=True, per_user=None): + def get(key, default=None, user_id=None): + if key == "DNB_USE_BOOK_LANGUAGES": + return enabled + if key == "BOOK_LANGUAGE": + return (per_user or {}).get(user_id, system) + return original_get(key, default, user_id=user_id) + + monkeypatch.setattr(dnb.app_config, "get", get) + + return configure + + +def _field(tag, *subfields, ind2=" "): + subs = "".join(f'{value}' for code, value in subfields) + return f'{subs}' + + +def _record(*fields, idn="1", title="Krabat", author="Preußler, Otfried", fixed=None): + """A minimal MARC21 record; `fixed` is the 008 fixed-length data field.""" + parts = [f'{idn}'] + if fixed: + parts.append(f'{fixed}') + if author: + parts.append(_field("100", ("a", author), ("4", "aut"))) + parts.append(_field("245", ("a", title))) + return ( + f'{"".join(parts)}{"".join(fields)}' + ) + + +# Modern RDA record: series in 830 and repeated as "(Harry Potter 1)" in the title +HARRY_POTTER = """ + + 1420086006 + 250101s2026 gw ||||| |||| 00||||ger + + 9783551323507 + : EUR 6.99 (DE) + + 355132350X + ger + + Rowling, J. K. + aut + + + Harry Potter und der Stein der Weisen (Harry Potter 1) + J.K. Rowling + + + Hamburg + Carlsen + 2026 + + 336 Seiten + + ohne Hilfsmittel zu benutzen + + + (Produktform)Paperback / softback + + Harry Potter Geschenk + + Fantasy + gnd + + + Kinderbücher, Jugendbücher + gatbeg + + + Fritz, Klaus + trl + + + Harry Potter + 1 + + + https://services.dnb.de/plus/idn/1420086006/blurb/ + Inhaltstext + + http://www.carlsen.de + +""" + +# Older record: multi-part title in 245 $n/$p, non-sorting markers, ISBD punctuation, +# translator note in the subtitle, 260 instead of 264, no relator code on the author +# (but one on the translator), and a trailing comma after the author's name +RAD_DER_ZEIT = """ + + 95752128X + 3-442-24603-0 kart. + ger + Jordan, Robert, + + König, Karin + trl + + + ˜Dasœ Rad der Zeit + 3 + ˜Dieœ Suche : + Roman / aus dem Amerikanischen von Karin König + + + München : + Goldmann, + 1999 + + + Goldmann ; + 24603 + + +""" + +VIDEO = _record(_field("337", ("a", "video")), idn="111", title="Harry Potter") +NO_PERSONS = _record(idn="222", title="Anthologie", author=None) + +DIAGNOSTIC_XML = """ +1.1 + +info:srw/diagnostic/1/1General System Error +'startIndex' must be less-than-or-equal to 'endIndex' + +""" + +# defusedxml must refuse to resolve external entities +XXE_XML = ( + ']>' + "&leak;" +) + +BLURB_HTML = """ +

Angaben aus der Verlagsmeldung

+

Harry Potter und der Stein der Weisen

Bis zu seinem elften Geburtstag glaubt Harry, +er sei ein ganz normaler Junge.

Ein Klassiker
Millionen +von Lesern sind mit Harry aufgewachsen.

+""" + + +def _sru(*records, total=None): + body = "".join( + "MARC21-xmlxml" + f"{record}{position}" + for position, record in enumerate(records, 1) + ) + count = len(records) if total is None else total + return ( + '' + '1.1' + f"{count}{body}" + "" + ) + + +class _DnbResponse: + def __init__(self, text, status_code=200): + self.text = text + self.content = text.encode("utf-8") + self.status_code = status_code + + def raise_for_status(self): + if self.status_code >= 400: + raise requests.HTTPError(response=self) + + +class _DnbSession: + """Serves canned SRU responses; queries for which `hits(query)` is false find nothing. + + Records each SRU request as (maximumRecords, startRecord, query). + """ + + def __init__(self, search_xml, hits=lambda query: True, blurb=BLURB_HTML): + self.search_xml = search_xml + self.hits = hits + self.blurb = blurb + self.calls = 0 + self.queries = [] + + def get(self, url, params=None, **kwargs): + self.calls += 1 + if params is None: + return _DnbResponse(self.blurb) + self.queries.append((params["maximumRecords"], params["startRecord"], params["query"])) + if not self.hits(params["query"]): + return _DnbResponse(_sru(total=0)) + if params["maximumRecords"] == 0: + return _DnbResponse(_sru(total=1)) + return _DnbResponse(self.search_xml) + + +class _SequenceSession: + """Answers each request with the next canned response, or raises it.""" + + def __init__(self, *replies): + self.replies = list(replies) + + def get(self, url, params=None, **kwargs): + reply = self.replies.pop(0) + if isinstance(reply, Exception): + raise reply + return reply + + +def _provider(session): + provider = DNBProvider() + provider.session = session + return provider + + +def _search(*records, **fields): + """Run a one-request search over the given records.""" + session = _DnbSession(_sru(*records)) + return _provider(session).search(MetadataSearchOptions(query="Krabat", fields=fields)) + + +def _queries(**options): + """The SRU queries one search over the Harry Potter record sends.""" + session = _DnbSession(_sru(HARRY_POTTER)) + _provider(session).search(MetadataSearchOptions(**options)) + return [query for _, _, query in session.queries] + + +class TestDnbSearch: + def test_search_parses_results(self): + session = _DnbSession(_sru(HARRY_POTTER, RAD_DER_ZEIT, VIDEO, NO_PERSONS, total=10)) + provider = _provider(session) + + result = provider.search_paginated( + MetadataSearchOptions(query="", fields={"title": "Harry Potter"}) + ) + + # The exact phrase has hits, so its first page is the only request + assert session.queries == [(40, 1, 'tit="Harry Potter"' + EXCLUDED)] + # Video and person-less records are skipped + assert [book.provider_id for book in result.books] == ["1420086006", "95752128X"] + assert result.total_found == 10 + assert result.has_more is True + + book = result.books[0] + assert book.provider == "dnb" + assert book.title == "Harry Potter und der Stein der Weisen" + assert book.authors == ["J. K. Rowling"] # translator excluded + assert book.isbn_13 == "9783551323507" + assert book.isbn_10 == "355132350X" + assert book.publisher == "Carlsen" + assert book.publish_year == 2026 + assert book.language == "de" + assert book.series_name == "Harry Potter" + assert book.series_position == 1.0 + # Publisher keywords (653) and qualifier-prefixed terms are not genres + assert book.genres == ["Fantasy", "Kinderbücher", "Jugendbücher"] + assert book.cover_url is None # DNB's cover service is behind bot protection + assert book.source_url == "https://d-nb.info/1420086006" + assert book.description is None # blurbs are only fetched for book details + assert {f.label: f.value for f in book.display_fields} == {"Pages": "336"} + + def test_search_parses_multipart_title(self): + book = _search(RAD_DER_ZEIT)[0] + + assert book.title == "Die Suche" + assert book.subtitle == "Roman" + assert book.authors == ["Robert Jordan"] # trailing comma dropped, translator excluded + assert book.series_name == "Das Rad der Zeit" + assert book.series_position == 3.0 + assert book.isbn_10 == "3442246030" + assert book.isbn_13 is None + assert book.publisher == "Goldmann" + assert book.publish_year == 1999 + + @pytest.mark.parametrize( + ("text", "stage_with_hits", "searched"), + [ + # A complete title: exactly that title, plus records where a word is a name + ("Der Schwarm", "tst", "complete title"), + # A phrase within a title ("Die Verwandlung Kafka" naming a study guide) + ("Die Verwandlung Kafka", "tit", "title phrase"), + # Name and title words mixed: every word in the title or a name + ("Die Verwandlung Kafka", None, "every word"), + ], + ids=["complete-title", "title-phrase", "name-and-title-words"], + ) + def test_search_box_text(self, text, stage_with_hits, searched): + words = text.split() + every_word = " AND ".join(f'(tit="{w}" OR per="{w}")' for w in words) + some_name = " OR ".join(f'per="{w}"' for w in words) + complete, phrase = f'(tst="{text}")', f'tit="{text}"' + expected = { + "complete title": f"({complete} OR ({every_word} AND ({some_name})))", + "title phrase": f"({phrase} OR ({every_word} AND ({some_name})))", + "every word": every_word, + }[searched] + probes = [complete] if stage_with_hits == "tst" else [complete, phrase] + + def hits(query): + stage = ( + "tst" if query.startswith("(tst=") else "tit" if query.startswith("tit=") else None + ) + return stage is None or stage == stage_with_hits + + session = _DnbSession(_sru(HARRY_POTTER), hits=hits) + + _provider(session).search(MetadataSearchOptions(query=text, sort=SortOrder.OLDEST)) + + sort = " sortBy jhr/sort.ascending" + assert session.queries == [(0, 1, probe + EXCLUDED + sort) for probe in probes] + [ + (40, 1, expected + EXCLUDED + sort) + ] + + def test_search_box_also_tries_a_leading_article(self): + # "Herr der Ringe" is how people type "Der Herr der Ringe" + session = _DnbSession(_sru(HARRY_POTTER)) + + _provider(session).search(MetadataSearchOptions(query="Herr der Ringe")) + + assert session.queries[0][2].startswith( + '(tst="Herr der Ringe" OR tst="Der Herr der Ringe" OR tst="Die Herr der Ringe"' + ) + + def test_search_falls_back_to_word_query(self): + exact = 'per="Rowling" AND tit="Die Stein der Weisen"' + session = _DnbSession(_sru(HARRY_POTTER), hits=lambda query: exact not in query) + + books = _provider(session).search( + MetadataSearchOptions( + query="", fields={"author": "Rowling", "title": "Die Stein der Weisen"} + ) + ) + + assert [book.provider_id for book in books] == ["1420086006"] + # German articles are dropped from the word-by-word title query + assert session.queries == [ + (40, 1, exact + EXCLUDED), + (40, 1, 'per="Rowling" AND tit="Stein" AND tit="Weisen"' + EXCLUDED), + ] + + def test_later_pages_check_the_hit_count_first(self): + # So page 2 runs against the same query variant as page 1 + session = _DnbSession(_sru(HARRY_POTTER, total=100)) + + _provider(session).search( + MetadataSearchOptions(query="", fields={"title": "Harry Potter"}, page=2) + ) + + assert session.queries == [ + (0, 1, 'tit="Harry Potter"' + EXCLUDED), + (40, 41, 'tit="Harry Potter"' + EXCLUDED), + ] + + def test_page_and_limit_are_clamped(self): + # A limit of 0 would otherwise page forever through empty pages + session = _DnbSession(_sru(HARRY_POTTER, total=1)) + + result = _provider(session).search_paginated( + MetadataSearchOptions(query="Rowling", limit=0, page=-3) + ) + + assert session.queries[-1][:2] == (1, 1) # the page request + assert (result.page, result.has_more) == (1, False) + + @pytest.mark.parametrize( + "reply", + [ + _DnbResponse("Wartungsarbeiten"), + _DnbResponse(DIAGNOSTIC_XML), + _DnbResponse(XXE_XML), + _DnbResponse("", status_code=503), + requests.Timeout(), + ], + ids=["maintenance-page", "diagnostic", "external-entity", "http-503", "timeout"], + ) + def test_unusable_responses_are_not_cached(self, reply): + provider = _provider(_SequenceSession(reply, _DnbResponse(_sru(HARRY_POTTER)))) + options = MetadataSearchOptions(query="", fields={"author": "Rowling"}) # one request + + assert provider.search(options) == [] + assert [book.provider_id for book in provider.search(options)] == ["1420086006"] + + @pytest.mark.parametrize( + "options", + [ + MetadataSearchOptions(query=' " \\ '), + MetadataSearchOptions(query='9783641216177" or tit=*', search_type=SearchType.ISBN), + ], + ids=["blank", "isbn-injection"], + ) + def test_unusable_input_makes_no_request(self, options): + session = _DnbSession(_sru(HARRY_POTTER)) + + assert _provider(session).search(options) == [] + assert session.calls == 0 + + def test_author_search_keeps_only_that_persons_books(self): + # DNB's person index also matches people a book is about + schaetzing = _record(idn="1", author="Scha" + UMLAUT + "tzing, Frank") + about_him = _record(idn="2", author="Seibt, Gustav") + + # "Schaetzing" and "Schätzing" compare equal + books = _search(schaetzing, about_him, author="Schaetzing") + + assert [book.provider_id for book in books] == ["1"] + + def test_author_filter_keeps_page_when_nothing_matches(self): + # A pseudonym shares no name part with the searched person + books = _search(_record(author="Bachman, Richard"), author="King") + + assert [book.authors for book in books] == [["Richard Bachman"]] + + def test_people_with_other_roles_are_not_authors(self): + narrated = _record(_field("700", ("a", "Berben, Iris"), ("4", "nrt")), idn="1", author=None) + anthology = _record( + _field("700", ("a", "Usher, Shaun"), ("4", "edt")), + _field("700", ("a", "Eidinger, Lars"), ("4", "nrt")), + idn="2", + author=None, + ) + + books = _search(narrated, anthology) + + # Narrator-only records have no author and are skipped + assert [(book.provider_id, book.authors) for book in books] == [("2", ["Shaun Usher"])] + + @pytest.mark.parametrize( + ("fields", "fixed", "year", "publisher"), + [ + ( + [ + _field("264", ("c", "©2022"), ind2="4"), + _field("264", ("b", "Booket"), ("c", "[2023]"), ind2="1"), + ], + None, + 2023, + "Booket", + ), + ([], "240724s2024 gw ||||| |||| 00||||ger ", 2024, None), + ([_field("264", ("c", "[9999]"), ind2="1")], "870101s1987", 1987, None), + ( + [ + _field("264", ("c", "2021"), ind2="1"), + _field("264", ("b", "Druckerei Muster"), ind2="3"), + ], + None, + 2021, + None, + ), + ], + ids=[ + "publication-over-copyright", + "fixed-field", + "implausible-year", + "printer-is-not-publisher", + ], + ) + def test_publication_year_and_publisher(self, fields, fixed, year, publisher): + book = _search(_record(*fields, fixed=fixed))[0] + + assert (book.publish_year, book.publisher) == (year, publisher) + + @pytest.mark.parametrize( + ("extent", "pages"), + [ + ("1.024 S.", "1024"), + ("1 Online-Ressource (224 pages)", "224"), + ("999999 Seiten", None), + ("3 CDs", None), + ], + ) + def test_page_count(self, extent, pages): + book = _search(_record(_field("300", ("a", extent))))[0] + + assert [f.value for f in book.display_fields if f.label == "Pages"] == ( + [pages] if pages else [] + ) + + @pytest.mark.parametrize( + ("title", "publisher", "expected"), + [ + # Spelled-out numbers become the title indexers know; "[s. n.]" means no publisher + ( + "20000 [Zwanzigtausend] Meilen unter den Meeren", + "[s. n.] @", + ("20000 Meilen unter den Meeren", None), + ), + # A wholly bracketed title keeps its text; publisher notes are dropped + ( + "[Briefe]", + "RM-Buch-und-Medien-Vertrieb [u.a.]", + ("Briefe", "RM-Buch-und-Medien-Vertrieb"), + ), + ], + ids=["insertion-and-no-publisher", "bracketed-title-and-note"], + ) + def test_cataloguer_brackets_are_removed(self, title, publisher, expected): + book = _search(_record(_field("264", ("b", publisher), ind2="1"), title=title))[0] + + assert (book.title, book.publisher) == expected + + def test_all_text_is_composed_unicode(self): + record = _record( + _field("264", ("b", "Bu" + UMLAUT + "cher"), ind2="1"), + _field("520", ("a", "U" + UMLAUT + "ber Gro" + UMLAUT + "ße")), + title="Verha" + UMLAUT + "ltnis", + author="Mu" + UMLAUT + "ller, Jo" + UMLAUT + "rg", + ) + + book = _search(record)[0] + + values = [book.title, *book.authors, book.publisher, book.description] + assert all(unicodedata.is_normalized("NFC", value) for value in values) + assert book.authors == [ + unicodedata.normalize("NFC", "Jo" + UMLAUT + "rg Mu" + UMLAUT + "ller") + ] + + +class TestDnbLookups: + def test_get_book_fetches_blurb(self): + session = _DnbSession(_sru(HARRY_POTTER)) + + book = _provider(session).get_book("1420086006") + + assert book is not None + assert session.queries == [(1, 1, "idn=1420086006" + EXCLUDED)] + assert book.description == ( + "Bis zu seinem elften Geburtstag glaubt Harry, er sei ein ganz normaler Junge.\n\n" + "Ein Klassiker\n" + "Millionen von Lesern sind mit Harry aufgewachsen." + ) + + def test_requests_use_the_configured_proxy(self, monkeypatch): + # The proxy from Settings > Network, which requests' own env handling misses for HTTPS + proxies = {"https": "http://proxy.local:8118"} + monkeypatch.setattr(dnb, "get_proxies", lambda url: proxies if "dnb.de" in url else {}) + session = _DnbSession(_sru(HARRY_POTTER)) + serve, used = session.get, [] + + def get(url, params=None, **kwargs): + used.append(kwargs.get("proxies")) + return serve(url, params, **kwargs) + + monkeypatch.setattr(session, "get", get) + _provider(session).get_book("1420086006") + + assert used == [proxies, proxies] # catalog record, then blurb + + def test_get_book_checks_the_id(self): + session = _DnbSession(_sru(HARRY_POTTER)) + provider = _provider(session) + + assert provider.get_book('1" OR tit="x') is None + assert session.calls == 0 + + assert provider.get_book("1420086007") is None + assert session.calls == 1 # no blurb fetched for the wrong record + + def test_blurbs_are_only_fetched_from_dnb(self): + # Catalogue links can point anywhere, including internal addresses + record = _record(_field("856", ("u", "http://169.254.169.254/latest/meta-data/")), idn="77") + session = _DnbSession(_sru(record)) + + book = _provider(session).get_book("77") + + assert book.description is None + assert session.calls == 1 # the record lookup only + + def test_description_comes_from_blurb_or_summary(self): + # Some publishers deliver the summary as HTML, escaped inside the MARC XML + summary = "Kurz.<br><br>Zusammengefasst." + record = HARRY_POTTER.replace("", _field("520", ("a", summary)) + "") + provider = _provider(_DnbSession(_sru(record))) + + assert provider.search(MetadataSearchOptions(query="Rowling"))[0].description == ( + "Kurz.\n\nZusammengefasst." + ) + assert provider.get_book("1420086006").description.startswith("Bis zu seinem") + + def test_isbn_in_search_box_uses_the_isbn_lookup(self): + session = _DnbSession(_sru(HARRY_POTTER)) + + result = _provider(session).search_paginated( + MetadataSearchOptions(query="978-3-551-32350-7") + ) + + assert [book.provider_id for book in result.books] == ["1420086006"] + assert result.total_found == 1 + assert session.queries == [(5, 1, "num=9783551323507" + EXCLUDED)] + + def test_isbn_lookup_checks_the_record(self): + # The number index also matches ISSNs and other identifiers + other = _record(_field("020", ("a", "9783492612418")), idn="2") + isbn13_only = _record(_field("020", ("a", "9783551323507")), idn="3") + + def lookup(isbn, *records): + get_metadata_cache().clear() + book = _provider(_DnbSession(_sru(*records))).search_by_isbn(isbn) + return book.provider_id if book else None + + assert lookup("9783551323507", other, HARRY_POTTER) == "1420086006" + assert lookup("9783551323507", other) is None + assert lookup("3-551-32350-X", isbn13_only) == "3" + + +class TestDnbLanguageFilter: + @pytest.mark.parametrize( + ("enabled", "system", "clause"), + [ + (False, ["de"], None), + (True, [], None), + (True, ["de", "en"], "(spr=ger OR spr=eng)"), + (True, "de, xx", "(spr=ger)"), + ], + ids=["checkbox-off", "no-languages", "german-english", "legacy-string-unknown-skipped"], + ) + def test_book_languages_restrict_the_search(self, book_languages, enabled, system, clause): + book_languages(system, enabled=enabled) + + expected = 'per="Rowling"' + (f" AND {clause}" if clause else "") + assert _queries(query="", fields={"author": "Rowling"}) == [expected + EXCLUDED] + + @pytest.mark.parametrize( + ("user_id", "clause"), + [(7, "(spr=ger)"), (8, "(spr=eng)"), (None, "(spr=eng)")], + ids=["own-setting", "no-own-setting", "not-logged-in"], + ) + def test_the_searching_users_own_languages_win(self, book_languages, user_id, clause): + # System default English; user 7 chose German for their own account + book_languages(["en"], per_user={7: ["de"]}) + + assert _queries(query="", fields={"author": "Rowling"}, user_id=user_id) == [ + f'per="Rowling" AND {clause}' + EXCLUDED + ] + + def test_cached_results_follow_the_languages_not_the_user(self, book_languages): + # User 7 chose German; users 8 and 9 use the system default English + book_languages(["en"], per_user={7: ["de"]}) + session = _DnbSession(_sru(HARRY_POTTER)) + + for user_id in (7, 8, 9): + options = MetadataSearchOptions(query="", fields={"author": "Rowling"}, user_id=user_id) + _provider(session).search(options) + + assert [query for _, _, query in session.queries] == [ + 'per="Rowling" AND (spr=ger)' + EXCLUDED, + 'per="Rowling" AND (spr=eng)' + EXCLUDED, # user 9 gets user 8's cached page + ] + + def test_every_selectable_language_can_be_filtered(self): + # A language without a MARC code would silently drop out of the filter + unmapped = [code for code in known_language_codes() if dnb._language_clause([code]) is None] + assert unmapped == [] + + def test_language_passed_with_the_search_wins(self, book_languages): + book_languages(["de", "en"]) + + assert _queries(query="", fields={"author": "Rowling"}, language="fr") == [ + 'per="Rowling" AND (spr=fre)' + EXCLUDED + ] + + def test_search_box_hit_counts_are_filtered_too(self, book_languages): + book_languages(["de"]) + + queries = _queries(query="Rowling") + + assert len(queries) == 2 # complete-title count, then the page + assert all(query.endswith(" AND (spr=ger)" + EXCLUDED) for query in queries) + + def test_direct_lookups_ignore_the_setting(self, book_languages): + # A specific IDN or ISBN must be found whatever its language + book_languages(["fr"]) + session = _DnbSession(_sru(HARRY_POTTER)) + provider = _provider(session) + + provider.get_book("1420086006") + provider.search_by_isbn("9783551323507") + + assert [query for _, _, query in session.queries] == [ + "idn=1420086006" + EXCLUDED, + "num=9783551323507" + EXCLUDED, + ] + + +class TestDnbSettings: + def test_defaults(self): + # Same sorts as the other providers, and the book languages apply out of the box + fields = {field.key: field for field in dnb_settings()} + + assert [option["value"] for option in get_provider_sort_options("dnb")] == [ + "relevance", + "newest", + "oldest", + ] + assert fields["DNB_DEFAULT_SORT"].default == "relevance" + assert fields["DNB_USE_BOOK_LANGUAGES"].default is True + + @pytest.mark.parametrize(("reply", "success"), [(([], 0), True), (None, False)]) + def test_connection_test_reports_reachability(self, monkeypatch, reply, success): + monkeypatch.setattr(DNBProvider, "_query", lambda self, cql, **kwargs: reply) + + assert dnb._test_dnb_connection()["success"] is success + + +class TestDnbHelpers: + def test_clean_series_drops_publisher_imprints(self): + assert _clean_series("Goldmann ;", "Goldmann") is None + assert _clean_series("dtv premium", None) is None + assert _clean_series("Harry Potter", "Carlsen") == "Harry Potter" + + def test_format_author(self): + assert _format_author("Rowling, J. K. [Verfasser]") == "J. K. Rowling" + assert _format_author("Wharram, C. C.,") == "C. C. Wharram" + assert _format_author("Homer") == "Homer"