from __future__ import annotations import logging import re import unicodedata from dataclasses import dataclass from difflib import SequenceMatcher from typing import Any import httpx from config.settings import settings logger = logging.getLogger(__name__) SEARCH_URL = "https://openlibrary.org/search.json" COVERS_URL = "https://covers.openlibrary.org/b/id/{cover_id}-{size}.jpg" SEARCH_FIELDS = ( "key,title,author_name,cover_i,first_publish_year,subject,language," "edition_key,number_of_pages_median,publisher" ) HIGH_CONFIDENCE = 0.8 LOW_CONFIDENCE = 0.6 # EPUB release noise often copied from filenames, e.g. "Title [6494] (r2.3)" _TITLE_NOISE_PATTERNS = ( re.compile(r"\s*\[\d+\]"), # [6494] re.compile(r"\s*\([rv][\d.]+\)", re.IGNORECASE), # (r2.3), (v1.0) re.compile(r"\s*\(rev[\d.]*\)", re.IGNORECASE), # (rev2) ) def _sanitize_search_title(title: str) -> str: cleaned = title.strip() for pattern in _TITLE_NOISE_PATTERNS: cleaned = pattern.sub("", cleaned) return re.sub(r"\s+", " ", cleaned).strip() @dataclass(frozen=True) class OpenLibraryHit: work_key: str title: str authors: list[str] cover_id: int | None first_publish_year: int | None subjects: list[str] languages: list[str] publishers: list[str] edition_key: str | None number_of_pages_median: int | None match_language: str score: float @dataclass(frozen=True) class OpenLibraryMetadata: match_language: str match_score: float match_status: str openlibrary: dict[str, Any] def _normalize(text: str) -> str: cleaned = re.sub(r"[^\w\s]", " ", _strip_accents(text).lower()) return re.sub(r"\s+", " ", cleaned).strip() def _strip_accents(text: str) -> str: normalized = unicodedata.normalize("NFKD", text) return "".join(ch for ch in normalized if not unicodedata.combining(ch)) def _build_search_params( title: str, author: str, *, lang: str | None, search_mode: str, ) -> dict[str, str | int]: title = title.strip() author = author.strip() params: dict[str, str | int] = {"limit": 10, "fields": SEARCH_FIELDS} if search_mode == "spanish_q": q_parts = ["language:spa", title] if author: q_parts.append(author) params["q"] = " ".join(q_parts) elif search_mode == "q": params["q"] = " ".join(part for part in (title, author) if part) elif search_mode == "q_unaccent": params["q"] = " ".join( part for part in (_strip_accents(title), _strip_accents(author) if author else "") if part ) else: params["title"] = title if author: params["author"] = author if lang: params["lang"] = lang return params def _title_similarity(a: str, b: str) -> float: na, nb = _normalize(a), _normalize(b) if not na or not nb: return 0.0 if na in nb or nb in na: return 1.0 return SequenceMatcher(None, na, nb).ratio() def _author_overlap(user_author: str, ol_authors: list[str]) -> float: if not user_author.strip(): return 0.5 if ol_authors else 0.0 user_tokens = set(_normalize(user_author).split()) if not user_tokens: return 0.0 best = 0.0 for name in ol_authors: name_tokens = set(_normalize(name).split()) if not name_tokens: continue overlap = len(user_tokens & name_tokens) / len(user_tokens) best = max(best, overlap) if user_tokens <= name_tokens or name_tokens <= user_tokens: best = max(best, 0.95) return best def _score_hit(title: str, author: str, doc: dict[str, Any], *, lang: str) -> float: ol_title = doc.get("title") or "" ol_authors = doc.get("author_name") or [] title_score = _title_similarity(title, ol_title) author_score = _author_overlap(author, ol_authors) combined = (title_score * 0.6) + (author_score * 0.4) if doc.get("cover_i"): combined += 0.05 return min(combined, 1.0) def _parse_hit(doc: dict[str, Any], *, lang: str, score: float) -> OpenLibraryHit: edition_keys = doc.get("edition_key") or [] edition_key = edition_keys[0] if edition_keys else None cover_id = doc.get("cover_i") return OpenLibraryHit( work_key=doc.get("key") or "", title=doc.get("title") or "", authors=list(doc.get("author_name") or []), cover_id=int(cover_id) if cover_id else None, first_publish_year=doc.get("first_publish_year"), subjects=list(doc.get("subject") or [])[:10], languages=list(doc.get("language") or []), publishers=list(doc.get("publisher") or [])[:5], edition_key=edition_key, number_of_pages_median=doc.get("number_of_pages_median"), match_language=lang, score=score, ) def build_cover_url(cover_id: int, size: str = "L") -> str: return COVERS_URL.format(cover_id=cover_id, size=size) def _client() -> httpx.Client: read_timeout = settings.OPENLIBRARY_TIMEOUT_SECONDS connect_timeout = settings.OPENLIBRARY_CONNECT_TIMEOUT_SECONDS return httpx.Client( timeout=httpx.Timeout( connect=connect_timeout, read=read_timeout, write=read_timeout, pool=connect_timeout, ), headers={"User-Agent": settings.OPENLIBRARY_USER_AGENT}, follow_redirects=True, ) def search_works( title: str, author: str, *, lang: str | None = None, search_mode: str = "title", client: httpx.Client | None = None, ) -> list[OpenLibraryHit]: if not title.strip(): return [] params = _build_search_params(title, author, lang=lang, search_mode=search_mode) try: if client is not None: response = client.get(SEARCH_URL, params=params) response.raise_for_status() docs = response.json().get("docs") or [] else: with _client() as owned_client: response = owned_client.get(SEARCH_URL, params=params) response.raise_for_status() docs = response.json().get("docs") or [] except (httpx.HTTPError, ValueError) as exc: logger.warning("Open Library search failed: %s", exc) return [] hits: list[OpenLibraryHit] = [] for doc in docs: score = _score_hit(title, author, doc, lang=lang or "") if score < LOW_CONFIDENCE: continue hits.append(_parse_hit(doc, lang=lang or "", score=score)) hits.sort(key=lambda h: (h.score, h.cover_id is not None), reverse=True) return hits def pick_best_match(title: str, author: str, hits: list[OpenLibraryHit]) -> OpenLibraryHit | None: return hits[0] if hits else None def download_cover(cover_id: int) -> bytes | None: url = build_cover_url(cover_id, size="L") try: with _client() as client: response = client.get(url) if response.status_code == 404: return None response.raise_for_status() content_type = response.headers.get("content-type", "") if not content_type.startswith("image/"): return None return response.content except httpx.HTTPError as exc: logger.warning("Open Library cover download failed for %s: %s", cover_id, exc) return None def _hit_to_openlibrary_dict(hit: OpenLibraryHit) -> dict[str, Any]: return { "work_key": hit.work_key, "edition_key": hit.edition_key, "title": hit.title, "authors": hit.authors, "cover_id": hit.cover_id, "cover_url": build_cover_url(hit.cover_id) if hit.cover_id else None, "first_publish_year": hit.first_publish_year, "subjects": hit.subjects, "languages": hit.languages, "publishers": hit.publishers, "number_of_pages_median": hit.number_of_pages_median, } def fetch_metadata(title: str, author: str) -> OpenLibraryMetadata | None: if not settings.OPENLIBRARY_ENABLED: return None preferred = settings.OPENLIBRARY_PREFERRED_LANG fallback = settings.OPENLIBRARY_FALLBACK_LANG search_title = _sanitize_search_title(title) if not search_title: return OpenLibraryMetadata( match_language=preferred, match_score=0.0, match_status="not_found", openlibrary={}, ) search_plan: list[tuple[str, str | None]] = [ ("spanish_q", preferred), ("title", preferred), ("q", None), ("q_unaccent", None), ] if fallback != preferred: search_plan.extend([("title", fallback), ("q", fallback)]) hits: list[OpenLibraryHit] = [] with _client() as client: for search_mode, lang in search_plan: hits = search_works(search_title, author, lang=lang, search_mode=search_mode, client=client) if hits: break hit = pick_best_match(title, author, hits) if not hit: return OpenLibraryMetadata( match_language=preferred, match_score=0.0, match_status="not_found", openlibrary={}, ) status = "matched" if hit.score >= HIGH_CONFIDENCE else "partial" return OpenLibraryMetadata( match_language=hit.match_language, match_score=hit.score, match_status=status, openlibrary=_hit_to_openlibrary_dict(hit), )