Archived
crisleo94
6b4c0c43f8
- add uv configuration for the backend - update frontend to make auth work - add new auth endpoints - add bookmars feat - add reader feat
305 lines
9.3 KiB
Python
305 lines
9.3 KiB
Python
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),
|
|
)
|