329 lines
11 KiB
Python
329 lines
11 KiB
Python
"""Hardcover, the ebook provider with usable German metadata.
|
|
|
|
Needs an API token (Settings → API in your Hardcover account, put it into
|
|
WORDARR_HARDCOVER_TOKEN); without one the provider stays silent instead of
|
|
breaking the combined ebook search.
|
|
|
|
The GraphQL search is Typesense backed and returns its hits as untyped JSON, so
|
|
every field is read defensively. A book row describes the work, not one edition,
|
|
so when a language is requested the matching editions are fetched in a second
|
|
query and their title/ISBN/cover replace the ones of the original edition.
|
|
Series are one query - unlike Audible, Hardcover knows them.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from collections import Counter
|
|
|
|
import httpx
|
|
|
|
from .base import MetadataResult
|
|
|
|
ENDPOINT = "https://api.hardcover.app/v1/graphql"
|
|
TOKEN = os.environ.get("WORDARR_HARDCOVER_TOKEN", "").strip()
|
|
LANGUAGES = {"german": "de", "english": "en"}
|
|
RESULT_LIMIT = 10
|
|
SERIES_LIMIT = 500 # books per series
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
_SEARCH = """
|
|
query Search($q: String!, $per: Int!) {
|
|
search(query: $q, query_type: "Book", per_page: $per, page: 1) {
|
|
results
|
|
}
|
|
}
|
|
"""
|
|
|
|
_EDITIONS = """
|
|
query Editions($ids: [Int!]!, $lang: String!) {
|
|
editions(
|
|
where: {
|
|
book_id: {_in: $ids}
|
|
_or: [{language: {code2: {_eq: $lang}}}, {language_id: {_is_null: true}}]
|
|
}
|
|
order_by: {users_count: desc_nulls_last}
|
|
) {
|
|
book_id
|
|
title
|
|
isbn_13
|
|
isbn_10
|
|
release_date
|
|
cached_image
|
|
language { code2 }
|
|
}
|
|
other: editions(
|
|
where: {book_id: {_in: $ids}, language: {code2: {_neq: $lang}}}
|
|
distinct_on: book_id
|
|
) {
|
|
book_id
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
_SERIES = """
|
|
query Series($id: Int!, $limit: Int!) {
|
|
series(where: {id: {_eq: $id}}) {
|
|
name
|
|
book_series(order_by: {position: asc_nulls_last}, limit: $limit) {
|
|
position
|
|
book {
|
|
id
|
|
title
|
|
release_year
|
|
users_count
|
|
cached_image
|
|
contributions(limit: 3) { author { name } }
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
async def _post(client: httpx.AsyncClient, query: str, variables: dict) -> dict:
|
|
resp = await client.post(
|
|
ENDPOINT,
|
|
json={"query": query, "variables": variables},
|
|
headers={"Authorization": f"Bearer {TOKEN}"},
|
|
)
|
|
resp.raise_for_status()
|
|
payload = resp.json()
|
|
if payload.get("errors"):
|
|
raise RuntimeError(payload["errors"][0].get("message", "hardcover error"))
|
|
return payload.get("data") or {}
|
|
|
|
|
|
def _image_url(value) -> str:
|
|
# covers arrive as {"url": …} on books and as a json string or dict on editions
|
|
if isinstance(value, dict):
|
|
return value.get("url") or ""
|
|
return value if isinstance(value, str) and value.startswith("http") else ""
|
|
|
|
|
|
def _authors(doc: dict) -> str:
|
|
names = doc.get("author_names")
|
|
if not names:
|
|
names = [
|
|
((c or {}).get("author") or {}).get("name", "")
|
|
for c in doc.get("contributions") or []
|
|
]
|
|
return ", ".join(n for n in names if n)
|
|
|
|
|
|
def _isbn(*candidates) -> str:
|
|
for value in candidates:
|
|
if isinstance(value, list):
|
|
value = next((v for v in value if v), "")
|
|
if value:
|
|
return str(value)
|
|
return ""
|
|
|
|
|
|
def _series_of(doc: dict) -> tuple[str, str]:
|
|
"""Name and id of the featured series; series_names[0] can be another one."""
|
|
featured = ((doc.get("featured_series") or {}).get("series")) or {}
|
|
if featured.get("name"):
|
|
return featured["name"], str(featured.get("id") or "")
|
|
names = doc.get("series_names") or []
|
|
ids = doc.get("series_ids") or []
|
|
return (names[0] if names else ""), (str(ids[0]) if ids else "")
|
|
|
|
|
|
def _volume(position) -> int | None:
|
|
# position 0 means "somewhere in this series", not "episode zero"
|
|
try:
|
|
number = int(float(position))
|
|
except (TypeError, ValueError):
|
|
return None
|
|
return number or None
|
|
|
|
|
|
def _to_result(doc: dict, language: str) -> MetadataResult:
|
|
series, series_id = _series_of(doc)
|
|
year = doc.get("release_year")
|
|
return MetadataResult(
|
|
media_type="ebook",
|
|
source="hardcover",
|
|
title=doc.get("title", ""),
|
|
authors=_authors(doc),
|
|
external_id=_isbn(doc.get("isbns")),
|
|
year=int(str(year)) if str(year).isdigit() else None,
|
|
language=language,
|
|
series=series,
|
|
series_id=series_id,
|
|
volume=_volume(doc.get("featured_series_position")),
|
|
cover_url=_image_url(doc.get("image")),
|
|
)
|
|
|
|
|
|
async def _localized(client: httpx.AsyncClient, results: dict[int, MetadataResult],
|
|
language: str) -> dict[int, MetadataResult]:
|
|
"""The hits that have an edition in the wanted language, swapped in.
|
|
|
|
An edition without a language counts too, unless the book has one in another
|
|
language: translations are often a book row of their own whose editions nobody
|
|
tagged ("Tote Augen"), while an English row always has tagged editions.
|
|
"""
|
|
code = LANGUAGES[language]
|
|
data = await _post(client, _EDITIONS, {"ids": list(results), "lang": code})
|
|
foreign = {ed.get("book_id") for ed in data.get("other") or []}
|
|
known: dict[int, MetadataResult] = {}
|
|
unknown: dict[int, MetadataResult] = {}
|
|
for ed in data.get("editions") or []:
|
|
book_id = ed.get("book_id")
|
|
base = results.get(book_id)
|
|
if base is None:
|
|
continue
|
|
if (ed.get("language") or {}).get("code2") == code:
|
|
target, tag = known, language
|
|
elif book_id not in foreign:
|
|
target, tag = unknown, ""
|
|
else:
|
|
continue
|
|
if book_id in target:
|
|
continue # the first edition is the most read one
|
|
release = str(ed.get("release_date") or "")
|
|
target[book_id] = base.model_copy(update={
|
|
"title": ed.get("title") or base.title,
|
|
"external_id": (_isbn(ed.get("isbn_13"), ed.get("isbn_10"))
|
|
or base.external_id),
|
|
"year": int(release[:4]) if release[:4].isdigit() else base.year,
|
|
"language": tag,
|
|
"cover_url": _image_url(ed.get("cached_image")) or base.cover_url,
|
|
})
|
|
return {**unknown, **known}
|
|
|
|
|
|
async def _localize(client: httpx.AsyncClient, results: dict[int, MetadataResult],
|
|
language: str) -> list[MetadataResult]:
|
|
"""Replace each hit with its edition in the wanted language, drop the rest.
|
|
Hits whose language is only unknown go last."""
|
|
localized = await _localized(client, results, language)
|
|
hits = [localized[book_id] for book_id in results if book_id in localized]
|
|
return sorted(hits, key=lambda r: not r.language)
|
|
|
|
|
|
async def search(query: str, language: str = "",
|
|
fallback: bool = True) -> list[MetadataResult]:
|
|
if not TOKEN:
|
|
return []
|
|
async with httpx.AsyncClient(timeout=15) as client:
|
|
data = await _post(client, _SEARCH, {"q": query.strip(), "per": RESULT_LIMIT})
|
|
hits = ((data.get("search") or {}).get("results") or {}).get("hits") or []
|
|
|
|
results: dict[int, MetadataResult] = {}
|
|
for hit in hits:
|
|
doc = hit.get("document") or {}
|
|
book_id = doc.get("id")
|
|
try:
|
|
book_id = int(str(book_id))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if doc.get("title"):
|
|
results.setdefault(book_id, _to_result(doc, ""))
|
|
|
|
if language in LANGUAGES and results:
|
|
try:
|
|
return await _localize(client, results, language)
|
|
except (httpx.HTTPError, RuntimeError) as exc:
|
|
# without the edition lookup the hits are still worth showing
|
|
log.warning("hardcover edition lookup failed for %r: %s", query, exc)
|
|
return list(results.values())
|
|
|
|
|
|
def _count(value) -> int:
|
|
try:
|
|
return int(str(value))
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def _author_set(book: dict) -> set[str]:
|
|
return {name.strip().lower()
|
|
for name in _authors(book).split(", ") if name.strip()}
|
|
|
|
|
|
def _main_author(authors: dict[int, set[str]]) -> str:
|
|
"""The name behind the series, or "" when the rows do not agree on one.
|
|
|
|
Translators and publishers ride along in the contributions, so the author of
|
|
the series is simply the name most rows share - and a row without it is the
|
|
foreign book that someone filed under this position by mistake.
|
|
"""
|
|
tally = Counter(name for names in authors.values() for name in names)
|
|
if not tally:
|
|
return ""
|
|
name, shared = tally.most_common(1)[0]
|
|
return name if shared * 2 > len(authors) else ""
|
|
|
|
|
|
async def series_books(series_id: str, language: str = "") -> list[MetadataResult]:
|
|
"""Every volume of a series, by position. A position often holds several book
|
|
rows - other editions of the story, and now and then a foreign book someone
|
|
filed here by mistake. The slot goes to the row that is by the author of the
|
|
series, has an edition in the wanted language and is the most read of those.
|
|
Unnumbered books go last."""
|
|
if not TOKEN:
|
|
return []
|
|
try:
|
|
numeric_id = int(str(series_id))
|
|
except (TypeError, ValueError):
|
|
return []
|
|
|
|
async with httpx.AsyncClient(timeout=30) as client:
|
|
data = await _post(client, _SERIES, {"id": numeric_id, "limit": SERIES_LIMIT})
|
|
row = (data.get("series") or [{}])[0]
|
|
name = row.get("name", "")
|
|
|
|
slots: dict[int, list[int]] = {} # volume -> book rows filed under it
|
|
loose: list[int] = []
|
|
books: dict[int, MetadataResult] = {}
|
|
authors: dict[int, set[str]] = {}
|
|
readers: dict[int, int] = {}
|
|
for entry in row.get("book_series") or []:
|
|
book = entry.get("book") or {}
|
|
try:
|
|
book_id = int(str(book.get("id")))
|
|
except (TypeError, ValueError):
|
|
continue
|
|
if not book.get("title") or book_id in books:
|
|
continue
|
|
volume = _volume(entry.get("position"))
|
|
books[book_id] = _to_result(
|
|
{**book, "image": book.get("cached_image"),
|
|
"series_names": [name], "series_ids": [numeric_id],
|
|
"featured_series_position": volume},
|
|
"",
|
|
)
|
|
authors[book_id] = _author_set(book)
|
|
readers[book_id] = _count(book.get("users_count"))
|
|
if volume is None:
|
|
loose.append(book_id)
|
|
else:
|
|
slots.setdefault(volume, []).append(book_id)
|
|
|
|
localized: dict[int, MetadataResult] = {}
|
|
if language in LANGUAGES and books:
|
|
try:
|
|
localized = await _localized(client, books, language)
|
|
except (httpx.HTTPError, RuntimeError) as exc:
|
|
log.warning("hardcover series editions failed for %s: %s",
|
|
series_id, exc)
|
|
|
|
author = _main_author(authors)
|
|
|
|
def rank(book_id: int) -> tuple:
|
|
# an untagged edition is no evidence: bundles and Spanish rows have them too
|
|
edition = localized.get(book_id)
|
|
return (author in authors[book_id] if author else True,
|
|
bool(edition and edition.language), readers[book_id])
|
|
|
|
def pick(candidates: list[int]) -> MetadataResult:
|
|
best = max(candidates, key=rank)
|
|
return localized.get(best) or books[best]
|
|
|
|
return ([pick(slots[volume]) for volume in sorted(slots)]
|
|
+ [localized.get(i) or books[i] for i in loose])
|