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"