starred/shelfmark
1
0
Fork 0
mirror of https://github.com/calibrain/shelfmark.git synced 2026-09-24 18:50:19 +01:00
Code Issues Packages Projects Releases 10 Wiki Activity

Enhance Hardcover lists (#710)

Browse Source
This commit is contained in:
Alex
2026-03-06 10:45:15 +00:00
committed by GitHub
parent 4b0d1aef13
commit d6d10a450e
2 changed files with 291 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
shelfmark/metadata_providers/hardcover.py
+156 -45
View File
@@ -95,6 +95,11 @@ query GetUserLists {
me {
id
username
want_to_read_books: user_books_aggregate(where: {status_id: {_eq: 1}}) {
aggregate {
count(columns: [book_id], distinct: true)
}
}
lists(order_by: {name: asc}) {
id
name
@@ -116,6 +121,56 @@ query GetUserLists {
}
"""
USER_BOOKS_BY_STATUS_QUERY = """
query GetCurrentUserBooksByStatus($statusId: Int!, $limit: Int!, $offset: Int!) {
me {
status_books: user_books(
where: {status_id: {_eq: $statusId}}
distinct_on: [book_id]
order_by: [{book_id: asc}, {created_at: desc}]
limit: $limit
offset: $offset
) {
book {
id
title
subtitle
slug
release_date
headline
description
pages
rating
ratings_count
users_count
cached_image
cached_contributors
contributions(where: {contribution: {_eq: "Author"}}) {
author {
name
}
}
featured_book_series {
position
series {
name
primary_books_count
}
}
}
}
status_books_aggregate: user_books_aggregate(where: {status_id: {_eq: $statusId}}) {
aggregate {
count(columns: [book_id], distinct: true)
}
}
}
}
"""
HARDCOVER_WANT_TO_READ_STATUS_ID = 1
HARDCOVER_STATUS_PREFIX = "status:"
# Mapping from abstract sort order to Hardcover sort parameter
# Note: release_year is more consistently populated than release_date_i
@@ -498,51 +553,7 @@ class HardcoverProvider(MetadataProvider):
if not isinstance(book_data, dict) or not book_data:
continue
try:
author_names: List[str] = []
for contrib in book_data.get("contributions", []) or []:
if not isinstance(contrib, dict):
continue
author_data = contrib.get("author", {})
if isinstance(author_data, dict):
author_name = str(author_data.get("name") or "").strip()
if author_name:
author_names.append(author_name)
if not author_names:
for contrib in book_data.get("cached_contributors", []) or []:
if isinstance(contrib, dict):
nested_author = contrib.get("author", {})
if isinstance(nested_author, dict):
nested_name = str(nested_author.get("name") or "").strip()
if nested_name:
author_names.append(nested_name)
continue
flat_name = str(contrib.get("name") or "").strip()
if flat_name:
author_names.append(flat_name)
elif isinstance(contrib, str):
normalized = contrib.strip()
if normalized:
author_names.append(normalized)
search_like_item = {
"id": book_data.get("id"),
"title": book_data.get("title"),
"subtitle": book_data.get("subtitle"),
"slug": book_data.get("slug"),
"release_date": book_data.get("release_date"),
"headline": book_data.get("headline"),
"description": book_data.get("description"),
"rating": book_data.get("rating"),
"ratings_count": book_data.get("ratings_count"),
"users_count": book_data.get("users_count"),
"image": book_data.get("cached_image"),
"author_names": author_names,
}
parsed_book = self._parse_search_result(search_like_item)
parsed_book = self._parse_book(book_data)
if parsed_book:
books.append(parsed_book)
except Exception as exc:
@@ -642,6 +653,78 @@ class HardcoverProvider(MetadataProvider):
"""Cached wrapper keyed by Hardcover user id to avoid cross-user cache leakage."""
return self._fetch_user_lists()
def _fetch_current_user_books_by_status(self, status_id: int, page: int, limit: int) -> SearchResult:
"""Fetch the current user's Hardcover books for a specific status shelf."""
if not self.api_key:
return SearchResult(books=[], page=page, total_found=0, has_more=False)
connected_user_id = self._resolve_current_user_id()
if not connected_user_id:
return SearchResult(books=[], page=page, total_found=0, has_more=False)
return self._fetch_user_books_by_status_cached(connected_user_id, status_id, page, limit)
@cacheable(ttl_key="METADATA_CACHE_SEARCH_TTL", ttl_default=300, key_prefix="hardcover:user_books:status")
def _fetch_user_books_by_status_cached(
self,
_cache_user_id: str,
status_id: int,
page: int,
limit: int,
) -> SearchResult:
"""Cached wrapper keyed by Hardcover user id and status shelf."""
return self._fetch_user_books_by_status(status_id, page, limit)
def _fetch_user_books_by_status(self, status_id: int, page: int, limit: int) -> SearchResult:
"""Fetch books from the current user's Hardcover status shelf."""
if not self.api_key:
return SearchResult(books=[], page=page, total_found=0, has_more=False)
offset = (page - 1) * limit
result = self._execute_query(
USER_BOOKS_BY_STATUS_QUERY,
{
"statusId": status_id,
"limit": limit,
"offset": offset,
},
)
if not result:
return SearchResult(books=[], page=page, total_found=0, has_more=False)
me_data = result.get("me", {})
if isinstance(me_data, list) and me_data:
me_data = me_data[0]
if not isinstance(me_data, dict):
return SearchResult(books=[], page=page, total_found=0, has_more=False)
status_books = me_data.get("status_books", [])
aggregate_data = me_data.get("status_books_aggregate", {})
aggregate = aggregate_data.get("aggregate", {}) if isinstance(aggregate_data, dict) else {}
count_raw = aggregate.get("count", 0) if isinstance(aggregate, dict) else 0
try:
total_found = int(count_raw)
except (TypeError, ValueError):
total_found = 0
books: List[BookMetadata] = []
for item in status_books:
if not isinstance(item, dict):
continue
book_data = item.get("book", {})
if not isinstance(book_data, dict) or not book_data:
continue
try:
parsed_book = self._parse_book(book_data)
if parsed_book:
books.append(parsed_book)
except Exception as exc:
logger.debug(f"Failed to parse Hardcover status book for status_id={status_id}: {exc}")
has_more = offset + len(status_books) < total_found
return SearchResult(books=books, page=page, total_found=total_found, has_more=has_more)
def _fetch_user_lists(self) -> List[Dict[str, str]]:
"""Fetch raw list options from Hardcover me query."""
result = self._execute_query(USER_LISTS_QUERY, {})
@@ -664,6 +747,27 @@ class HardcoverProvider(MetadataProvider):
except (TypeError, ValueError):
return name
want_to_read_count_data = me_data.get("want_to_read_books", {})
want_to_read_aggregate = (
want_to_read_count_data.get("aggregate", {})
if isinstance(want_to_read_count_data, dict)
else {}
)
want_to_read_count = (
want_to_read_aggregate.get("count")
if isinstance(want_to_read_aggregate, dict)
else None
)
want_to_read_value = f"{HARDCOVER_STATUS_PREFIX}{HARDCOVER_WANT_TO_READ_STATUS_ID}"
seen_values.add(want_to_read_value)
options.append(
{
"value": want_to_read_value,
"label": _format_label("Want to Read", want_to_read_count),
"group": "My Books",
}
)
for list_item in me_data.get("lists", []):
if not isinstance(list_item, dict):
continue
@@ -733,6 +837,13 @@ class HardcoverProvider(MetadataProvider):
# Advanced filter list selector (shared fetch path with URL detection)
list_value_from_field = str(options.fields.get("hardcover_list", "")).strip()
if list_value_from_field:
if list_value_from_field.startswith(HARDCOVER_STATUS_PREFIX):
try:
status_id = int(list_value_from_field.split(":", 1)[1])
return self._fetch_current_user_books_by_status(status_id, options.page, options.limit)
except (IndexError, ValueError):
logger.debug(f"Invalid Hardcover status field value: {list_value_from_field}")
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
if list_value_from_field.startswith("id:"):
try:
list_id = int(list_value_from_field.split(":", 1)[1])
tests/metadata/test_hardcover_lists.py
+135
View File
@@ -0,0 +1,135 @@
from shelfmark.metadata_providers import MetadataSearchOptions, SearchResult
from shelfmark.metadata_providers.hardcover import (
HARDCOVER_WANT_TO_READ_STATUS_ID,
HardcoverProvider,
)
class TestHardcoverLists:
def test_fetch_user_lists_includes_want_to_read_shelf(self, monkeypatch):
provider = HardcoverProvider(api_key="test-token")
monkeypatch.setattr(
provider,
"_execute_query",
lambda query, variables: {
"me": {
"username": "alex",
"want_to_read_books": {
"aggregate": {
"count": 7,
}
},
"lists": [
{
"id": 42,
"name": "Sci-Fi Favourites",
"slug": "sci-fi-favourites",
"books_count": 12,
}
],
"followed_lists": [],
}
},
)
options = provider._fetch_user_lists()
assert options[0] == {
"value": f"status:{HARDCOVER_WANT_TO_READ_STATUS_ID}",
"label": "Want to Read (7)",
"group": "My Books",
}
assert options[1] == {
"value": "id:42",
"label": "Sci-Fi Favourites (12)",
"group": "My Lists",
}
def test_search_paginated_uses_status_field_as_list_source(self, monkeypatch):
provider = HardcoverProvider(api_key="test-token")
expected = SearchResult(books=[], page=2, total_found=14, has_more=True)
captured: dict[str, int] = {}
def fake_fetch(status_id: int, page: int, limit: int) -> SearchResult:
captured["status_id"] = status_id
captured["page"] = page
captured["limit"] = limit
return expected
monkeypatch.setattr(provider, "_fetch_current_user_books_by_status", fake_fetch)
result = provider.search_paginated(
MetadataSearchOptions(
query="",
page=2,
limit=20,
fields={"hardcover_list": f"status:{HARDCOVER_WANT_TO_READ_STATUS_ID}"},
)
)
assert result == expected
assert captured == {
"status_id": HARDCOVER_WANT_TO_READ_STATUS_ID,
"page": 2,
"limit": 20,
}
def test_fetch_current_user_books_by_status_returns_books(self, monkeypatch):
provider = HardcoverProvider(api_key="test-token")
captured: dict[str, object] = {}
monkeypatch.setattr(provider, "_resolve_current_user_id", lambda: "123")
def fake_execute(query: str, variables):
captured["query"] = query
captured["variables"] = variables
return {
"me": {
"status_books": [
{
"book": {
"id": 9000,
"title": "Dune",
"subtitle": None,
"slug": "dune",
"release_date": "1965-08-01",
"headline": None,
"description": "Arrakis.",
"rating": 4.6,
"ratings_count": 100,
"users_count": 200,
"cached_image": {"url": "https://example.com/dune.jpg"},
"cached_contributors": [{"name": "Frank Herbert"}],
"contributions": [],
"featured_book_series": None,
}
}
],
"status_books_aggregate": {
"aggregate": {
"count": 1,
}
},
}
}
monkeypatch.setattr(provider, "_execute_query", fake_execute)
result = provider._fetch_current_user_books_by_status(
HARDCOVER_WANT_TO_READ_STATUS_ID,
page=1,
limit=10,
)
assert captured["variables"] == {
"statusId": HARDCOVER_WANT_TO_READ_STATUS_ID,
"limit": 10,
"offset": 0,
}
assert "distinct_on: [book_id]" in str(captured["query"])
assert result.total_found == 1
assert result.has_more is False
assert len(result.books) == 1
assert result.books[0].title == "Dune"
assert result.books[0].authors == ["Frank Herbert"]