mirror of
https://github.com/calibrain/shelfmark.git
synced 2026-09-24 22:05:20 +01:00
Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
4881adc19f | ||
|
|
ee54033d23 | ||
|
|
3554d01c81 | ||
|
|
9dd445f2af | ||
|
|
4e41b1a8ec |
@@ -14,6 +14,7 @@ This document lists all configuration options that can be set via environment va
|
||||
- [Network](#network)
|
||||
- [Advanced](#advanced)
|
||||
- [Prowlarr](#prowlarr)
|
||||
- [Newznab](#newznab)
|
||||
- [AudiobookBay](#audiobookbay)
|
||||
- [IRC](#irc)
|
||||
- [Download Clients](#download-clients)
|
||||
@@ -124,6 +125,7 @@ Show the onboarding wizard on first run. Set to false to skip (useful for epheme
|
||||
|
||||
| Variable | Description | Type | Default |
|
||||
|----------|-------------|------|---------|
|
||||
| `SEARCH_PAGE_TITLE` | Title shown above the main search box on the homepage. | string | `Shelfmark` |
|
||||
| `CALIBRE_WEB_URL` | Adds a navigation button to your book library (Calibre-Web Automated, Grimmory, etc). | string | _none_ |
|
||||
| `AUDIOBOOK_LIBRARY_URL` | Adds a separate navigation button for your audiobook library (Audiobookshelf, Plex, etc). When both URLs are set, icons are shown instead of text. | string | _none_ |
|
||||
| `SUPPORTED_FORMATS` | Book formats to include in search results. ZIP/RAR archives are extracted automatically and book files are used if found. | string (comma-separated) | `epub,mobi,azw3,fb2,djvu,cbz,cbr` |
|
||||
@@ -133,6 +135,15 @@ Show the onboarding wizard on first run. Set to false to skip (useful for epheme
|
||||
<details>
|
||||
<summary>Detailed descriptions</summary>
|
||||
|
||||
#### `SEARCH_PAGE_TITLE`
|
||||
|
||||
**Search Page Title**
|
||||
|
||||
Title shown above the main search box on the homepage.
|
||||
|
||||
- **Type:** string
|
||||
- **Default:** `Shelfmark`
|
||||
|
||||
#### `CALIBRE_WEB_URL`
|
||||
|
||||
**Library URL**
|
||||
@@ -294,8 +305,8 @@ The release source tab to open by default in the release modal for audiobooks. U
|
||||
| `BOOKS_OUTPUT_MODE` | Choose where completed book files are sent. | string (choice) | `folder` |
|
||||
| `INGEST_DIR` | Directory where downloaded files are saved. Use {User} for per-user folders (e.g. /books/{User}). | string | `/books` |
|
||||
| `FILE_ORGANIZATION` | Choose how downloaded book files are named and organized. | string (choice) | `rename` |
|
||||
| `TEMPLATE_RENAME` | Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads. | string | `{Author} - {Title} ({Year})` |
|
||||
| `TEMPLATE_ORGANIZE` | Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. | string | `{Author}/{Title} ({Year})` |
|
||||
| `TEMPLATE_RENAME` | Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads. | string | `{Author} - {Title} ({Year})` |
|
||||
| `TEMPLATE_ORGANIZE` | Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. | string | `{Author}/{Title} ({Year})` |
|
||||
| `HARDLINK_TORRENTS` | Create hardlinks instead of copying. Preserves seeding but archives won't be extracted. Don't use if destination is a library ingest folder. | boolean | `false` |
|
||||
| `BOOKLORE_HOST` | Base URL of your Grimmory instance | string | _none_ |
|
||||
| `BOOKLORE_USERNAME` | Grimmory account username | string | _none_ |
|
||||
@@ -311,13 +322,13 @@ The release source tab to open by default in the release modal for audiobooks. U
|
||||
| `EMAIL_SMTP_USERNAME` | SMTP username (leave empty for no authentication). | string | _none_ |
|
||||
| `EMAIL_SMTP_PASSWORD` | SMTP password (required if Username is set). | string (secret) | _none_ |
|
||||
| `EMAIL_FROM` | From address used for the email. You can include a display name (e.g., Shelfmark <mail@example.com>). Leave blank to default to the SMTP username (when it is an email address). | string | _none_ |
|
||||
| `EMAIL_SUBJECT_TEMPLATE` | Email subject. Variables: {Author}, {Title}, {Year}, {Series}, {SeriesPosition}, {Subtitle}, {Format}. | string | `{Title}` |
|
||||
| `EMAIL_SUBJECT_TEMPLATE` | Email subject. Variables: {Author}, {Title}, {PrimaryTitle}, {Year}, {Series}, {SeriesPosition}, {Subtitle}, {Format}. | string | `{Title}` |
|
||||
| `EMAIL_SMTP_TIMEOUT_SECONDS` | How long to wait for SMTP operations before failing. | number | `60` |
|
||||
| `EMAIL_ALLOW_UNVERIFIED_TLS` | Disable TLS certificate verification (not recommended). | boolean | `false` |
|
||||
| `DESTINATION_AUDIOBOOK` | Directory where downloaded audiobook files are saved. Leave empty to use the Books destination. | string | _none_ |
|
||||
| `FILE_ORGANIZATION_AUDIOBOOK` | Choose how downloaded audiobook files are named and organized. | string (choice) | `rename` |
|
||||
| `TEMPLATE_AUDIOBOOK_RENAME` | Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads. | string | `{Author} - {Title}` |
|
||||
| `TEMPLATE_AUDIOBOOK_ORGANIZE` | Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. | string | `{Author}/{Title}` |
|
||||
| `TEMPLATE_AUDIOBOOK_RENAME` | Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads. | string | `{Author} - {Title}` |
|
||||
| `TEMPLATE_AUDIOBOOK_ORGANIZE` | Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. | string | `{Author}/{Title}/{Title}` |
|
||||
| `HARDLINK_TORRENTS_AUDIOBOOK` | Create hardlinks instead of copying. Preserves seeding but archives won't be extracted. Don't use if destination is a library ingest folder. | boolean | `true` |
|
||||
| `AUTO_OPEN_DOWNLOADS_SIDEBAR` | Automatically open the downloads sidebar when a new download is queued. | boolean | `false` |
|
||||
| `DOWNLOAD_TO_BROWSER_CONTENT_TYPES` | Automatically download completed files to your browser for the selected content types. | string (comma-separated) | _empty list_ |
|
||||
@@ -361,7 +372,7 @@ Choose how downloaded book files are named and organized.
|
||||
|
||||
**Naming Template**
|
||||
|
||||
Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads.
|
||||
Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads.
|
||||
|
||||
- **Type:** string
|
||||
- **Default:** `{Author} - {Title} ({Year})`
|
||||
@@ -370,7 +381,7 @@ Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename wi
|
||||
|
||||
**Path Template**
|
||||
|
||||
Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty.
|
||||
Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension). Universal adds: {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty.
|
||||
|
||||
- **Type:** string
|
||||
- **Default:** `{Author}/{Title} ({Year})`
|
||||
@@ -524,7 +535,7 @@ From address used for the email. You can include a display name (e.g., Shelfmark
|
||||
|
||||
**Subject Template**
|
||||
|
||||
Email subject. Variables: {Author}, {Title}, {Year}, {Series}, {SeriesPosition}, {Subtitle}, {Format}.
|
||||
Email subject. Variables: {Author}, {Title}, {PrimaryTitle}, {Year}, {Series}, {SeriesPosition}, {Subtitle}, {Format}.
|
||||
|
||||
- **Type:** string
|
||||
- **Default:** `{Title}`
|
||||
@@ -571,7 +582,7 @@ Choose how downloaded audiobook files are named and organized.
|
||||
|
||||
**Naming Template**
|
||||
|
||||
Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads.
|
||||
Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. Rename templates are filename-only (no '/' or '\'); use Organize for folders. Applies to single-file downloads.
|
||||
|
||||
- **Type:** string
|
||||
- **Default:** `{Author} - {Title}`
|
||||
@@ -580,10 +591,10 @@ Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename wi
|
||||
|
||||
**Path Template**
|
||||
|
||||
Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty.
|
||||
Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} (source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, {PrimaryTitle}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty.
|
||||
|
||||
- **Type:** string
|
||||
- **Default:** `{Author}/{Title}`
|
||||
- **Default:** `{Author}/{Title}/{Title}`
|
||||
|
||||
#### `HARDLINK_TORRENTS_AUDIOBOOK`
|
||||
|
||||
@@ -639,7 +650,7 @@ How long to keep completed/failed downloads in the queue display.
|
||||
|
||||
| Variable | Description | Type | Default |
|
||||
|----------|-------------|------|---------|
|
||||
| `AUTH_METHOD` | Select the authentication method for accessing Shelfmark. | string (choice) | `none` |
|
||||
| `AUTH_METHOD` | Select the authentication method for accessing Shelfmark. Restart container after changing Calibre-Web passwords. | string (choice) | `none` |
|
||||
| `PROXY_AUTH_USER_HEADER` | The HTTP header your proxy uses to pass the authenticated username. | string | `X-Auth-User` |
|
||||
| `PROXY_AUTH_LOGOUT_URL` | The URL to redirect users to for logging out. Leave empty to disable logout functionality. | string | _empty string_ |
|
||||
| `PROXY_AUTH_ADMIN_GROUP_HEADER` | Optional: header your proxy uses to pass user groups/roles. | string | `X-Auth-Groups` |
|
||||
@@ -661,7 +672,7 @@ How long to keep completed/failed downloads in the queue display.
|
||||
|
||||
**Authentication Method**
|
||||
|
||||
Select the authentication method for accessing Shelfmark.
|
||||
Select the authentication method for accessing Shelfmark. Restart container after changing Calibre-Web passwords.
|
||||
|
||||
- **Type:** string (choice)
|
||||
- **Default:** `none`
|
||||
@@ -1115,6 +1126,57 @@ Automatically retry search without category filtering if no results are found
|
||||
|
||||
</details>
|
||||
|
||||
## Newznab
|
||||
|
||||
| Variable | Description | Type | Default |
|
||||
|----------|-------------|------|---------|
|
||||
| `NEWZNAB_ENABLED` | Enable searching for books via a Newznab-compatible indexer | boolean | `false` |
|
||||
| `NEWZNAB_URL` | Base URL of your Newznab indexer or aggregator | string | _none_ |
|
||||
| `NEWZNAB_API_KEY` | Your Newznab API key (leave blank if not required) | string (secret) | _none_ |
|
||||
| `NEWZNAB_AUTO_EXPAND` | Automatically retry search without category filtering if no results are found | boolean | `false` |
|
||||
|
||||
<details>
|
||||
<summary>Detailed descriptions</summary>
|
||||
|
||||
#### `NEWZNAB_ENABLED`
|
||||
|
||||
**Enable Newznab source**
|
||||
|
||||
Enable searching for books via a Newznab-compatible indexer
|
||||
|
||||
- **Type:** boolean
|
||||
- **Default:** `false`
|
||||
|
||||
#### `NEWZNAB_URL`
|
||||
|
||||
**Newznab URL**
|
||||
|
||||
Base URL of your Newznab indexer or aggregator
|
||||
|
||||
- **Type:** string
|
||||
- **Default:** _none_
|
||||
- **Required:** Yes
|
||||
|
||||
#### `NEWZNAB_API_KEY`
|
||||
|
||||
**API Key**
|
||||
|
||||
Your Newznab API key (leave blank if not required)
|
||||
|
||||
- **Type:** string (secret)
|
||||
- **Default:** _none_
|
||||
|
||||
#### `NEWZNAB_AUTO_EXPAND`
|
||||
|
||||
**Auto-expand search on no results**
|
||||
|
||||
Automatically retry search without category filtering if no results are found
|
||||
|
||||
- **Type:** boolean
|
||||
- **Default:** `false`
|
||||
|
||||
</details>
|
||||
|
||||
## AudiobookBay
|
||||
|
||||
| Variable | Description | Type | Default |
|
||||
|
||||
+19
-2
@@ -106,6 +106,18 @@ if [ ! -x "$PYTHON_BIN" ]; then
|
||||
PYTHON_BIN="python3"
|
||||
fi
|
||||
|
||||
# Defensive: some orchestrators (e.g. Unraid Dockhand templates) inject a default
|
||||
# PATH that drops the venv bin directory baked in by the Dockerfile. Prepend it
|
||||
# so subprocesses launched without an absolute path still resolve correctly.
|
||||
case ":${PATH}:" in
|
||||
*":/app/.venv/bin:"*) ;;
|
||||
*) export PATH="/app/.venv/bin:${PATH}" ;;
|
||||
esac
|
||||
GUNICORN_BIN="/app/.venv/bin/gunicorn"
|
||||
if [ ! -x "$GUNICORN_BIN" ]; then
|
||||
GUNICORN_BIN="gunicorn"
|
||||
fi
|
||||
|
||||
# Print build version
|
||||
echo "Build version: $BUILD_VERSION"
|
||||
echo "Release version: $RELEASE_VERSION"
|
||||
@@ -443,7 +455,7 @@ fi
|
||||
# upgrades work reliably on customer machines.
|
||||
# Map app LOG_LEVEL (often DEBUG/INFO/...) to gunicorn's --log-level (lowercase).
|
||||
gunicorn_loglevel=$([ "$DEBUG" = "true" ] && echo debug || echo "${LOG_LEVEL:-info}" | tr '[:upper:]' '[:lower:]')
|
||||
command="gunicorn --log-level ${gunicorn_loglevel} --access-logfile - --error-logfile - --worker-class geventwebsocket.gunicorn.workers.GeventWebSocketWorker --workers 1 -t 300 -b ${FLASK_HOST:-0.0.0.0}:${FLASK_PORT:-8084} shelfmark.main:app"
|
||||
command="${GUNICORN_BIN} --log-level ${gunicorn_loglevel} --access-logfile - --error-logfile - --worker-class geventwebsocket.gunicorn.workers.GeventWebSocketWorker --workers 1 -t 300 -b ${FLASK_HOST:-0.0.0.0}:${FLASK_PORT:-8084} shelfmark.main:app"
|
||||
|
||||
# If DEBUG and not using an external bypass
|
||||
if [ "$DEBUG" = "true" ] && [ "$USING_EXTERNAL_BYPASSER" != "true" ]; then
|
||||
@@ -512,7 +524,12 @@ else
|
||||
fi
|
||||
|
||||
RUNTIME_HOME=$(resolve_runtime_home)
|
||||
require_writable_dir "$RUNTIME_HOME" "Home"
|
||||
if [ "$RUN_AS_NON_ROOT" = "true" ]; then
|
||||
require_writable_dir "$RUNTIME_HOME" "Home"
|
||||
else
|
||||
mkdir -p "$RUNTIME_HOME"
|
||||
make_writable "$RUNTIME_HOME" tree
|
||||
fi
|
||||
|
||||
if [ "$RUN_AS_NON_ROOT" = "true" ]; then
|
||||
echo "Startup mode: non-root"
|
||||
|
||||
@@ -23,7 +23,6 @@ dependencies = [
|
||||
"transmission-rpc",
|
||||
"authlib>=1.7.0,<1.8",
|
||||
"apprise>=1.9.0",
|
||||
"Pillow>=11.0.0",
|
||||
]
|
||||
|
||||
[project.optional-dependencies]
|
||||
|
||||
@@ -310,7 +310,7 @@ def generate_env_docs() -> str:
|
||||
|
||||
def _generate_tab_docs(tab: Any, group_prefix: str | None = None) -> list[str]:
|
||||
"""Generate documentation for a single settings tab."""
|
||||
from shelfmark.core.settings_registry import ActionButton, CustomComponentField, HeadingField
|
||||
from shelfmark.core.settings_registry import iter_value_fields
|
||||
|
||||
lines = []
|
||||
|
||||
@@ -323,17 +323,9 @@ def _generate_tab_docs(tab: Any, group_prefix: str | None = None) -> list[str]:
|
||||
lines.append("")
|
||||
|
||||
# Collect env-supported fields
|
||||
env_fields = []
|
||||
for field in tab.fields:
|
||||
# Skip non-value fields
|
||||
if isinstance(field, (ActionButton, CustomComponentField, HeadingField)):
|
||||
continue
|
||||
|
||||
# Skip fields that don't support ENV vars
|
||||
if not getattr(field, "env_supported", True):
|
||||
continue
|
||||
|
||||
env_fields.append(field)
|
||||
env_fields = [
|
||||
field for field in iter_value_fields(tab) if getattr(field, "env_supported", True)
|
||||
]
|
||||
|
||||
if not env_fields:
|
||||
lines.append("_No environment variables for this section._")
|
||||
|
||||
@@ -938,7 +938,7 @@ def download_settings() -> list[SettingsField]:
|
||||
SelectField(
|
||||
key="FILE_ORGANIZATION",
|
||||
label="File Organization",
|
||||
description="Choose how downloaded book files are named and organized. ",
|
||||
description="Choose how downloaded book files are named and organized.",
|
||||
options=[
|
||||
{
|
||||
"value": "none",
|
||||
@@ -966,7 +966,14 @@ def download_settings() -> list[SettingsField]:
|
||||
_naming_template_field(
|
||||
key="TEMPLATE_RENAME",
|
||||
label="Naming Template",
|
||||
description="Filename template for single-file book downloads.",
|
||||
description=(
|
||||
"Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} "
|
||||
"(source filename without extension). Universal adds: {Series}, "
|
||||
"{SeriesPosition}, {Subtitle}, {PrimaryTitle}. Use arbitrary prefix/suffix: "
|
||||
"{Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. "
|
||||
"Rename templates are filename-only (no '/' or '\\'); use Organize for folders. "
|
||||
"Applies to single-file downloads."
|
||||
),
|
||||
default="{Author} - {Title} ({Year})",
|
||||
placeholder="{Author} - {Title} ({Year})",
|
||||
show_when=[
|
||||
@@ -978,7 +985,12 @@ def download_settings() -> list[SettingsField]:
|
||||
_naming_template_field(
|
||||
key="TEMPLATE_ORGANIZE",
|
||||
label="Path Template",
|
||||
description="Folder and filename template for book downloads.",
|
||||
description=(
|
||||
"Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, "
|
||||
"{OriginalName} (source filename without extension). Universal adds: {Series}, "
|
||||
"{SeriesPosition}, {Subtitle}, {PrimaryTitle}. Use arbitrary prefix/suffix: "
|
||||
"{Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty."
|
||||
),
|
||||
default="{Author}/{Title} ({Year})",
|
||||
placeholder="{Author}/{Series/}{Title} ({Year})",
|
||||
show_when=[
|
||||
@@ -1236,7 +1248,14 @@ def download_settings() -> list[SettingsField]:
|
||||
_naming_template_field(
|
||||
key="TEMPLATE_AUDIOBOOK_RENAME",
|
||||
label="Naming Template",
|
||||
description="Filename template for single-file audiobook downloads.",
|
||||
description=(
|
||||
"Variables: {Author}, {Title}, {Year}, {User}, {OriginalName} "
|
||||
"(source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, "
|
||||
"{PrimaryTitle}, {PartNumber}. Use arbitrary prefix/suffix: "
|
||||
"{Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty. "
|
||||
"Rename templates are filename-only (no '/' or '\\'); use Organize for folders. "
|
||||
"Applies to single-file downloads."
|
||||
),
|
||||
default="{Author} - {Title}",
|
||||
placeholder="{Author} - {Title}{ - Part }{PartNumber}",
|
||||
show_when={"field": "FILE_ORGANIZATION_AUDIOBOOK", "value": "rename"},
|
||||
@@ -1246,8 +1265,13 @@ def download_settings() -> list[SettingsField]:
|
||||
_naming_template_field(
|
||||
key="TEMPLATE_AUDIOBOOK_ORGANIZE",
|
||||
label="Path Template",
|
||||
description="Folder and filename template for audiobook downloads.",
|
||||
default="{Author}/{Title}",
|
||||
description=(
|
||||
"Use / to create folders. Variables: {Author}, {Title}, {Year}, {User}, "
|
||||
"{OriginalName} (source filename without extension), {Series}, {SeriesPosition}, "
|
||||
"{Subtitle}, {PrimaryTitle}, {PartNumber}. Use arbitrary prefix/suffix: "
|
||||
"{Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty."
|
||||
),
|
||||
default="{Author}/{Title}/{Title}",
|
||||
placeholder="{Author}/{Series/}{Title}{ - Part }{PartNumber}",
|
||||
show_when={"field": "FILE_ORGANIZATION_AUDIOBOOK", "value": "organize"},
|
||||
universal_only=True,
|
||||
|
||||
@@ -49,9 +49,6 @@ TRANSIENT_CACHE_TTL = 60
|
||||
|
||||
_MIN_WEBP_HEADER_LENGTH = 12
|
||||
HTTP_NOT_FOUND = HTTPStatus.NOT_FOUND
|
||||
MAX_VARIANT_DIMENSION = 1024
|
||||
WEBP_DEFAULT_QUALITY = 80
|
||||
JPEG_DEFAULT_QUALITY = 85
|
||||
|
||||
|
||||
def _detect_image_type(data: bytes) -> tuple[str, str] | None:
|
||||
@@ -75,164 +72,6 @@ def _detect_image_type(data: bytes) -> tuple[str, str] | None:
|
||||
return None
|
||||
|
||||
|
||||
def normalize_variant_dimension(value: object) -> int | None:
|
||||
"""Normalize a requested variant dimension, clamping to a safe upper bound."""
|
||||
dimension = coerce_int(value, 0)
|
||||
if dimension <= 0:
|
||||
return None
|
||||
return min(dimension, MAX_VARIANT_DIMENSION)
|
||||
|
||||
|
||||
def normalize_variant_format(value: object) -> str | None:
|
||||
"""Normalize a requested output image format."""
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
|
||||
normalized = value.strip().lower()
|
||||
if normalized in {"jpg", "jpeg"}:
|
||||
return "jpeg"
|
||||
if normalized in {"png", "webp"}:
|
||||
return normalized
|
||||
return None
|
||||
|
||||
|
||||
def build_variant_cache_id(
|
||||
cache_id: str,
|
||||
*,
|
||||
width: int | None,
|
||||
height: int | None,
|
||||
image_format: str | None,
|
||||
) -> str:
|
||||
"""Build a cache key for a derived cover variant."""
|
||||
width_token = str(width) if width is not None else "auto"
|
||||
height_token = str(height) if height is not None else "auto"
|
||||
format_token = image_format or "original"
|
||||
return f"{cache_id}__w{width_token}_h{height_token}_f{format_token}"
|
||||
|
||||
|
||||
def _calculate_variant_size(
|
||||
*,
|
||||
source_width: int,
|
||||
source_height: int,
|
||||
width: int | None,
|
||||
height: int | None,
|
||||
) -> tuple[int, int]:
|
||||
"""Calculate the output size while preserving aspect ratio and avoiding upscaling."""
|
||||
if width is None and height is None:
|
||||
return source_width, source_height
|
||||
|
||||
width_ratio = (width / source_width) if width is not None else None
|
||||
height_ratio = (height / source_height) if height is not None else None
|
||||
|
||||
if width_ratio is not None and height_ratio is not None:
|
||||
scale = min(width_ratio, height_ratio, 1.0)
|
||||
elif width_ratio is not None:
|
||||
scale = min(width_ratio, 1.0)
|
||||
elif height_ratio is not None:
|
||||
scale = min(height_ratio, 1.0)
|
||||
else:
|
||||
scale = 1.0
|
||||
|
||||
return (
|
||||
max(1, round(source_width * scale)),
|
||||
max(1, round(source_height * scale)),
|
||||
)
|
||||
|
||||
|
||||
def _normalize_source_format(image_data: bytes) -> str | None:
|
||||
"""Return the normalized detected source image format."""
|
||||
detected = _detect_image_type(image_data)
|
||||
if not detected:
|
||||
return None
|
||||
|
||||
content_type, _ext = detected
|
||||
if content_type == "image/jpeg":
|
||||
return "jpeg"
|
||||
if content_type == "image/png":
|
||||
return "png"
|
||||
if content_type == "image/webp":
|
||||
return "webp"
|
||||
return None
|
||||
|
||||
|
||||
def create_image_variant(
|
||||
image_data: bytes,
|
||||
*,
|
||||
width: int | None = None,
|
||||
height: int | None = None,
|
||||
image_format: str | None = None,
|
||||
) -> tuple[bytes, str] | None:
|
||||
"""Create a resized and/or transcoded image variant.
|
||||
|
||||
Returns None when no variant is needed or the image cannot be safely transformed.
|
||||
"""
|
||||
requested_format = normalize_variant_format(image_format)
|
||||
if width is None and height is None and requested_format is None:
|
||||
return None
|
||||
|
||||
source_format = _normalize_source_format(image_data)
|
||||
|
||||
try:
|
||||
from PIL import Image, ImageOps, UnidentifiedImageError
|
||||
except ImportError:
|
||||
logger.warning("Pillow is not installed; serving original cover image")
|
||||
return None
|
||||
|
||||
try:
|
||||
with Image.open(BytesIO(image_data)) as source_image:
|
||||
if getattr(source_image, "is_animated", False):
|
||||
return None
|
||||
|
||||
image = ImageOps.exif_transpose(source_image)
|
||||
source_width, source_height = image.size
|
||||
output_width, output_height = _calculate_variant_size(
|
||||
source_width=source_width,
|
||||
source_height=source_height,
|
||||
width=width,
|
||||
height=height,
|
||||
)
|
||||
|
||||
needs_resize = (output_width, output_height) != (source_width, source_height)
|
||||
output_format = requested_format or source_format
|
||||
|
||||
if not needs_resize and output_format == source_format:
|
||||
return None
|
||||
|
||||
if needs_resize:
|
||||
image = image.resize((output_width, output_height), Image.Resampling.LANCZOS)
|
||||
|
||||
if output_format == "jpeg":
|
||||
if image.mode not in {"RGB", "L"}:
|
||||
image = image.convert("RGB")
|
||||
content_type = "image/jpeg"
|
||||
save_kwargs: dict[str, Any] = {
|
||||
"format": "JPEG",
|
||||
"quality": JPEG_DEFAULT_QUALITY,
|
||||
"optimize": True,
|
||||
}
|
||||
elif output_format == "png":
|
||||
if image.mode not in {"1", "L", "LA", "P", "PA", "RGB", "RGBA"}:
|
||||
image = image.convert("RGBA")
|
||||
content_type = "image/png"
|
||||
save_kwargs = {"format": "PNG", "optimize": True}
|
||||
else:
|
||||
if image.mode not in {"RGB", "RGBA"}:
|
||||
image = image.convert("RGBA" if "A" in image.getbands() else "RGB")
|
||||
content_type = "image/webp"
|
||||
save_kwargs = {
|
||||
"format": "WEBP",
|
||||
"quality": WEBP_DEFAULT_QUALITY,
|
||||
"method": 6,
|
||||
}
|
||||
|
||||
output = BytesIO()
|
||||
image.save(output, **save_kwargs)
|
||||
return output.getvalue(), content_type
|
||||
except (OSError, UnidentifiedImageError, ValueError) as exc:
|
||||
logger.warning("Failed to derive image variant: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
class ImageCacheService:
|
||||
"""Persistent image cache with LRU eviction and TTL support."""
|
||||
|
||||
|
||||
@@ -333,7 +333,7 @@ def get_all_settings_tabs() -> list[SettingsTab]:
|
||||
return sorted(_SETTINGS_REGISTRY.values(), key=lambda t: (t.order, t.name))
|
||||
|
||||
|
||||
def _iter_value_fields(tab: SettingsTab) -> Iterator[FieldBase]:
|
||||
def iter_value_fields(tab: SettingsTab) -> Iterator[FieldBase]:
|
||||
"""Yield value-bearing fields for a tab."""
|
||||
for settings_field in tab.fields:
|
||||
if isinstance(settings_field, CustomComponentField):
|
||||
@@ -360,7 +360,7 @@ def get_settings_field_map(
|
||||
|
||||
field_map: dict[str, tuple[FieldBase, str]] = {}
|
||||
for tab in tabs:
|
||||
for settings_field in _iter_value_fields(tab):
|
||||
for settings_field in iter_value_fields(tab):
|
||||
field_map[settings_field.key] = (settings_field, tab.name)
|
||||
return field_map
|
||||
|
||||
@@ -494,7 +494,7 @@ def initialize_default_configs() -> bool:
|
||||
|
||||
# Collect default values for all fields
|
||||
defaults = {}
|
||||
for field in _iter_value_fields(tab):
|
||||
for field in iter_value_fields(tab):
|
||||
# Only include fields that have a non-None default
|
||||
if field.default is not None:
|
||||
defaults[field.key] = field.default
|
||||
@@ -536,7 +536,7 @@ def sync_env_to_config() -> None:
|
||||
for tab in get_all_settings_tabs():
|
||||
values_to_sync = {}
|
||||
|
||||
for settings_field in _iter_value_fields(tab):
|
||||
for settings_field in iter_value_fields(tab):
|
||||
# Skip fields that don't support ENV vars
|
||||
if not getattr(settings_field, "env_supported", True):
|
||||
continue
|
||||
|
||||
+21
-65
@@ -1597,9 +1597,6 @@ def api_cover(cover_id: str) -> Response | tuple[Response, int]:
|
||||
|
||||
Query Parameters:
|
||||
url (str): Base64-encoded original image URL (required on first request)
|
||||
w (int): Optional max width for a derived image variant
|
||||
h (int): Optional max height for a derived image variant
|
||||
format (str): Optional output format for a derived image variant (webp/png/jpeg)
|
||||
|
||||
Returns:
|
||||
flask.Response: Binary image data with appropriate Content-Type, or 404.
|
||||
@@ -1609,84 +1606,43 @@ def api_cover(cover_id: str) -> Response | tuple[Response, int]:
|
||||
import base64
|
||||
|
||||
from shelfmark.config.env import is_covers_cache_enabled
|
||||
from shelfmark.core.image_cache import (
|
||||
build_variant_cache_id,
|
||||
create_image_variant,
|
||||
get_image_cache,
|
||||
normalize_variant_dimension,
|
||||
normalize_variant_format,
|
||||
)
|
||||
from shelfmark.core.image_cache import get_image_cache
|
||||
|
||||
# Check if caching is enabled
|
||||
if not is_covers_cache_enabled():
|
||||
return jsonify({"error": "Cover caching is disabled"}), 404
|
||||
|
||||
cache = get_image_cache()
|
||||
width = normalize_variant_dimension(request.args.get("w"))
|
||||
height = normalize_variant_dimension(request.args.get("h"))
|
||||
image_format = normalize_variant_format(request.args.get("format"))
|
||||
variant_cache_id = (
|
||||
build_variant_cache_id(
|
||||
cover_id,
|
||||
width=width,
|
||||
height=height,
|
||||
image_format=image_format,
|
||||
)
|
||||
if width is not None or height is not None or image_format is not None
|
||||
else None
|
||||
)
|
||||
|
||||
def make_cover_response(
|
||||
image_data: bytes,
|
||||
content_type: str,
|
||||
*,
|
||||
cache_status: str,
|
||||
) -> Response:
|
||||
response = app.response_class(response=image_data, status=200, mimetype=content_type)
|
||||
response.headers["Cache-Control"] = "public, max-age=86400"
|
||||
response.headers["X-Cache"] = cache_status
|
||||
return response
|
||||
|
||||
# Try to get from cache first
|
||||
cache_lookup_id = variant_cache_id or cover_id
|
||||
cached = cache.get(cache_lookup_id)
|
||||
cached = cache.get(cover_id)
|
||||
if cached:
|
||||
image_data, content_type = cached
|
||||
return make_cover_response(image_data, content_type, cache_status="HIT")
|
||||
response = app.response_class(response=image_data, status=200, mimetype=content_type)
|
||||
response.headers["Cache-Control"] = "public, max-age=86400"
|
||||
response.headers["X-Cache"] = "HIT"
|
||||
return response
|
||||
|
||||
# Cache miss - get URL from query parameter
|
||||
encoded_url = request.args.get("url")
|
||||
original: tuple[bytes, str] | None = cache.get(cover_id) if variant_cache_id else None
|
||||
if not encoded_url:
|
||||
return jsonify({"error": "Cover URL not provided"}), 404
|
||||
|
||||
if original is None:
|
||||
if not encoded_url:
|
||||
return jsonify({"error": "Cover URL not provided"}), 404
|
||||
try:
|
||||
original_url = base64.urlsafe_b64decode(encoded_url).decode()
|
||||
except (binascii.Error, UnicodeDecodeError) as e:
|
||||
logger.warning("Failed to decode cover URL: %s", e)
|
||||
return jsonify({"error": "Invalid cover URL encoding"}), 400
|
||||
|
||||
try:
|
||||
original_url = base64.urlsafe_b64decode(encoded_url).decode()
|
||||
except (binascii.Error, UnicodeDecodeError) as e:
|
||||
logger.warning("Failed to decode cover URL: %s", e)
|
||||
return jsonify({"error": "Invalid cover URL encoding"}), 400
|
||||
# Fetch and cache the image
|
||||
result = cache.fetch_and_cache(cover_id, original_url)
|
||||
if not result:
|
||||
return jsonify({"error": "Failed to fetch cover image"}), 404
|
||||
|
||||
# Fetch and cache the original image
|
||||
original = cache.fetch_and_cache(cover_id, original_url)
|
||||
if not original:
|
||||
return jsonify({"error": "Failed to fetch cover image"}), 404
|
||||
|
||||
image_data, content_type = original
|
||||
|
||||
if variant_cache_id:
|
||||
variant = create_image_variant(
|
||||
image_data,
|
||||
width=width,
|
||||
height=height,
|
||||
image_format=image_format,
|
||||
)
|
||||
if variant:
|
||||
image_data, content_type = variant
|
||||
cache.put(variant_cache_id, image_data, content_type)
|
||||
|
||||
response = make_cover_response(image_data, content_type, cache_status="MISS")
|
||||
image_data, content_type = result
|
||||
response = app.response_class(response=image_data, status=200, mimetype=content_type)
|
||||
response.headers["Cache-Control"] = "public, max-age=86400"
|
||||
response.headers["X-Cache"] = "MISS"
|
||||
except _IMPORT_OPERATIONAL_ERRORS as e:
|
||||
logger.error_trace(f"Cover fetch error: {e}")
|
||||
return jsonify({"error": str(e)}), 500
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -1,35 +0,0 @@
|
||||
"""Hardcover metadata provider package."""
|
||||
|
||||
from shelfmark.core.cache import get_metadata_cache
|
||||
from shelfmark.core.config import config as app_config
|
||||
|
||||
from .auth import _get_connected_user_id, _get_connected_username, _save_connected_user
|
||||
from .constants import (
|
||||
HARDCOVER_LIST_ID_PREFIX,
|
||||
HARDCOVER_STATUS_GROUP,
|
||||
HARDCOVER_STATUS_PREFIX,
|
||||
HARDCOVER_WRITABLE_TARGET_GROUPS,
|
||||
)
|
||||
from .models import HardcoverBookTargetState, HardcoverGraphQLError, HardcoverTargetPayloadError
|
||||
from .parsing import _compute_search_title, _simplify_author_for_search
|
||||
from .provider import HardcoverProvider
|
||||
from .settings import hardcover_settings
|
||||
|
||||
__all__ = [
|
||||
"HARDCOVER_LIST_ID_PREFIX",
|
||||
"HARDCOVER_STATUS_GROUP",
|
||||
"HARDCOVER_STATUS_PREFIX",
|
||||
"HARDCOVER_WRITABLE_TARGET_GROUPS",
|
||||
"HardcoverBookTargetState",
|
||||
"HardcoverGraphQLError",
|
||||
"HardcoverProvider",
|
||||
"HardcoverTargetPayloadError",
|
||||
"_compute_search_title",
|
||||
"_get_connected_user_id",
|
||||
"_get_connected_username",
|
||||
"_save_connected_user",
|
||||
"_simplify_author_for_search",
|
||||
"app_config",
|
||||
"get_metadata_cache",
|
||||
"hardcover_settings",
|
||||
]
|
||||
@@ -1,36 +0,0 @@
|
||||
"""Persistence helpers for the connected Hardcover account."""
|
||||
|
||||
|
||||
def _save_connected_user(user_id: str | None, username: str | None) -> None:
|
||||
"""Save or clear connected user metadata in config."""
|
||||
from shelfmark.core.settings_registry import load_config_file, save_config_file
|
||||
|
||||
config = load_config_file("hardcover")
|
||||
if user_id:
|
||||
config["_connected_user_id"] = user_id
|
||||
else:
|
||||
config.pop("_connected_user_id", None)
|
||||
|
||||
if username:
|
||||
config["_connected_username"] = username
|
||||
else:
|
||||
config.pop("_connected_username", None)
|
||||
|
||||
save_config_file("hardcover", config)
|
||||
|
||||
|
||||
def _get_connected_username() -> str | None:
|
||||
"""Get the stored connected username."""
|
||||
from shelfmark.core.settings_registry import load_config_file
|
||||
|
||||
config = load_config_file("hardcover")
|
||||
return config.get("_connected_username")
|
||||
|
||||
|
||||
def _get_connected_user_id() -> str | None:
|
||||
"""Get the stored connected Hardcover user id."""
|
||||
from shelfmark.core.settings_registry import load_config_file
|
||||
|
||||
config = load_config_file("hardcover")
|
||||
value = config.get("_connected_user_id")
|
||||
return str(value) if value is not None else None
|
||||
@@ -1,105 +0,0 @@
|
||||
"""GraphQL transport helpers for Hardcover."""
|
||||
|
||||
from http import HTTPStatus
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
|
||||
from shelfmark.core.logger import setup_logger
|
||||
from shelfmark.download.network import get_ssl_verify
|
||||
|
||||
from .constants import HARDCOVER_API_URL
|
||||
from .models import HardcoverGraphQLError
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
def _extract_graphql_error_message(payload: Any) -> str:
|
||||
"""Extract a readable message from a GraphQL error payload."""
|
||||
if not isinstance(payload, dict):
|
||||
return ""
|
||||
|
||||
errors = payload.get("errors", [])
|
||||
if not isinstance(errors, list):
|
||||
return ""
|
||||
|
||||
messages: list[str] = []
|
||||
for error in errors:
|
||||
if not isinstance(error, dict):
|
||||
continue
|
||||
message = str(error.get("message") or "").strip()
|
||||
if message:
|
||||
messages.append(message)
|
||||
|
||||
return "; ".join(messages)
|
||||
|
||||
|
||||
class HardcoverClientMixin:
|
||||
session: requests.Session
|
||||
|
||||
def _execute_query(
|
||||
self,
|
||||
query: str,
|
||||
variables: dict[str, Any],
|
||||
*,
|
||||
raise_on_error: bool = False,
|
||||
) -> dict | None:
|
||||
"""Execute a GraphQL query and return data or None on error."""
|
||||
|
||||
def _raise_graphql_error(message: str) -> None:
|
||||
raise HardcoverGraphQLError(message)
|
||||
|
||||
try:
|
||||
response = self.session.post(
|
||||
HARDCOVER_API_URL,
|
||||
json={"query": query, "variables": variables},
|
||||
timeout=15,
|
||||
verify=get_ssl_verify(HARDCOVER_API_URL),
|
||||
)
|
||||
response.raise_for_status()
|
||||
|
||||
data = response.json()
|
||||
|
||||
if "errors" in data:
|
||||
logger.error("GraphQL errors: %s", data["errors"])
|
||||
if raise_on_error:
|
||||
message = (
|
||||
_extract_graphql_error_message(data) or "Hardcover rejected this request"
|
||||
)
|
||||
_raise_graphql_error(message)
|
||||
return None
|
||||
|
||||
return data.get("data")
|
||||
|
||||
except requests.Timeout as e:
|
||||
logger.warning("Hardcover API request timed out")
|
||||
if raise_on_error:
|
||||
msg = "Hardcover API request timed out"
|
||||
raise RuntimeError(msg) from e
|
||||
return None
|
||||
except requests.HTTPError as e:
|
||||
if e.response.status_code == HTTPStatus.UNAUTHORIZED:
|
||||
logger.exception("Hardcover API key is invalid")
|
||||
if raise_on_error:
|
||||
msg = "Hardcover API key is invalid"
|
||||
raise RuntimeError(msg) from e
|
||||
else:
|
||||
logger.exception("Hardcover API HTTP error")
|
||||
if raise_on_error:
|
||||
msg = f"Hardcover API HTTP error: {e}"
|
||||
raise RuntimeError(msg) from e
|
||||
return None
|
||||
except HardcoverGraphQLError:
|
||||
raise
|
||||
except ValueError as e:
|
||||
logger.exception("Hardcover API returned invalid JSON")
|
||||
if raise_on_error:
|
||||
msg = "Hardcover API returned an invalid response"
|
||||
raise RuntimeError(msg) from e
|
||||
return None
|
||||
except (TypeError, requests.RequestException) as e:
|
||||
logger.exception("Hardcover API request failed")
|
||||
if raise_on_error:
|
||||
msg = "Hardcover API request failed"
|
||||
raise RuntimeError(msg) from e
|
||||
return None
|
||||
@@ -1,61 +0,0 @@
|
||||
"""Constants for the Hardcover metadata provider."""
|
||||
|
||||
import re
|
||||
|
||||
from shelfmark.metadata_providers import SearchType, SortOrder
|
||||
|
||||
HARDCOVER_API_URL = "https://api.hardcover.app/v1/graphql"
|
||||
HARDCOVER_PAGE_SIZE = 25 # Hardcover API returns max 25 results per page
|
||||
HARDCOVER_MIN_AUTHOR_PARTS = 2
|
||||
HARDCOVER_MIN_TYPEAHEAD_QUERY_LENGTH = 2
|
||||
HARDCOVER_MAX_SERIES_OPTIONS = 7
|
||||
HARDCOVER_API_KEY_MIN_LENGTH = 100
|
||||
HARDCOVER_LIST_URL_PATTERN = re.compile(
|
||||
r"^/(?:@([\w.-]+)/)?lists?/([\w-]+)/?$",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
HARDCOVER_STATUS_PREFIX = "status:"
|
||||
HARDCOVER_STATUSES: list[dict] = [
|
||||
{"id": 1, "label": "Want to Read", "slug": "want-to-read", "query_key": "want_to_read_count"},
|
||||
{
|
||||
"id": 2,
|
||||
"label": "Currently Reading",
|
||||
"slug": "currently-reading",
|
||||
"query_key": "currently_reading_count",
|
||||
},
|
||||
{"id": 3, "label": "Read", "slug": "read", "query_key": "read_count"},
|
||||
{
|
||||
"id": 5,
|
||||
"label": "Did Not Finish",
|
||||
"slug": "did-not-finish",
|
||||
"query_key": "did_not_finish_count",
|
||||
},
|
||||
]
|
||||
HARDCOVER_STATUS_URL_SLUGS: dict[int, str] = {s["id"]: s["slug"] for s in HARDCOVER_STATUSES}
|
||||
HARDCOVER_STATUS_GROUP = "Reading Status"
|
||||
HARDCOVER_LIST_ID_PREFIX = "id:"
|
||||
HARDCOVER_WRITABLE_TARGET_GROUPS = {HARDCOVER_STATUS_GROUP, "My Lists"}
|
||||
|
||||
SORT_MAPPING: dict[SortOrder, str] = {
|
||||
SortOrder.RELEVANCE: "_text_match:desc,users_count:desc",
|
||||
SortOrder.POPULARITY: "users_count:desc",
|
||||
SortOrder.RATING: "rating:desc",
|
||||
SortOrder.NEWEST: "release_year:desc",
|
||||
SortOrder.OLDEST: "release_year:asc",
|
||||
}
|
||||
SEARCH_TYPE_FIELDS: dict[SearchType, str] = {
|
||||
SearchType.GENERAL: "title,isbns,series_names,author_names,alternative_titles",
|
||||
SearchType.TITLE: "title,alternative_titles",
|
||||
SearchType.AUTHOR: "author_names",
|
||||
# ISBN is handled separately via search_by_isbn()
|
||||
}
|
||||
SERIES_SEARCH_FIELDS = "name,books,author_name"
|
||||
SERIES_SEARCH_WEIGHTS = "2,1,1"
|
||||
SERIES_SEARCH_SORT = "_text_match:desc,readers_count:desc"
|
||||
AUTHOR_SUGGESTION_FIELDS = "name,name_personal,alternate_names"
|
||||
AUTHOR_SUGGESTION_WEIGHTS = "4,3,2"
|
||||
AUTHOR_SUGGESTION_SORT = "_text_match:desc,books_count:desc"
|
||||
TITLE_SUGGESTION_FIELDS = "title,alternative_titles"
|
||||
TITLE_SUGGESTION_WEIGHTS = "5,2"
|
||||
TITLE_SUGGESTION_SORT = "_text_match:desc,users_count:desc"
|
||||
@@ -1,400 +0,0 @@
|
||||
"""Hardcover list and status-shelf workflows."""
|
||||
|
||||
from typing import TYPE_CHECKING, Any
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from shelfmark.core.cache import cacheable
|
||||
from shelfmark.core.logger import setup_logger
|
||||
from shelfmark.core.request_helpers import coerce_int
|
||||
from shelfmark.metadata_providers import BookMetadata, SearchResult
|
||||
|
||||
from .auth import _get_connected_user_id, _get_connected_username, _save_connected_user
|
||||
from .constants import (
|
||||
HARDCOVER_LIST_URL_PATTERN,
|
||||
HARDCOVER_STATUS_GROUP,
|
||||
HARDCOVER_STATUS_PREFIX,
|
||||
HARDCOVER_STATUS_URL_SLUGS,
|
||||
HARDCOVER_STATUSES,
|
||||
)
|
||||
from .queries import (
|
||||
LIST_BOOKS_BY_ID_QUERY,
|
||||
LIST_LOOKUP_QUERY,
|
||||
USER_BOOKS_BY_STATUS_QUERY,
|
||||
USER_LISTS_QUERY,
|
||||
)
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
class HardcoverListsMixin:
|
||||
if TYPE_CHECKING:
|
||||
api_key: str
|
||||
|
||||
def _execute_query(
|
||||
self,
|
||||
query: str,
|
||||
variables: dict[str, Any],
|
||||
*,
|
||||
raise_on_error: bool = False,
|
||||
) -> dict[str, Any] | None: ...
|
||||
|
||||
def _parse_book(self, book: dict[str, Any]) -> BookMetadata: ...
|
||||
|
||||
def _detect_list_url(self, query: str) -> tuple[str | None, str] | None:
|
||||
"""Detect and extract optional owner username + list slug from a URL string."""
|
||||
candidate = query.strip()
|
||||
if not candidate:
|
||||
return None
|
||||
|
||||
parsed = urlparse(candidate)
|
||||
if parsed.scheme not in {"http", "https"}:
|
||||
return None
|
||||
|
||||
hostname = (parsed.hostname or "").lower()
|
||||
if hostname not in {"hardcover.app", "www.hardcover.app"}:
|
||||
return None
|
||||
|
||||
match = HARDCOVER_LIST_URL_PATTERN.match(parsed.path or "")
|
||||
if not match:
|
||||
return None
|
||||
|
||||
owner_username = match.group(1).strip() if match.group(1) else None
|
||||
slug = match.group(2).strip()
|
||||
if not slug:
|
||||
return None
|
||||
|
||||
return owner_username, slug
|
||||
|
||||
@cacheable(ttl_key="METADATA_CACHE_SEARCH_TTL", ttl_default=300, key_prefix="hardcover:list:id")
|
||||
def _fetch_list_books_by_id(self, list_id: int, page: int, limit: int) -> SearchResult:
|
||||
"""Fetch list books by unique Hardcover list ID."""
|
||||
if not self.api_key:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
offset = (page - 1) * limit
|
||||
|
||||
result = self._execute_query(
|
||||
LIST_BOOKS_BY_ID_QUERY,
|
||||
{
|
||||
"id": list_id,
|
||||
"limit": limit,
|
||||
"offset": offset,
|
||||
},
|
||||
)
|
||||
if not result:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
lists = result.get("lists", [])
|
||||
if not lists:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
list_data = lists[0] if isinstance(lists[0], dict) else {}
|
||||
list_books = list_data.get("list_books", []) if isinstance(list_data, dict) else []
|
||||
books_count_raw = list_data.get("books_count", 0) if isinstance(list_data, dict) else 0
|
||||
|
||||
# Build source URL and title from list metadata
|
||||
source_url = None
|
||||
source_title = str(list_data.get("name") or "").strip() or None
|
||||
list_slug = str(list_data.get("slug") or "").strip()
|
||||
user_data = list_data.get("user", {})
|
||||
owner_username = (
|
||||
str(user_data.get("username") or "").strip() if isinstance(user_data, dict) else ""
|
||||
)
|
||||
if list_slug and owner_username:
|
||||
source_url = f"https://hardcover.app/@{owner_username}/lists/{list_slug}"
|
||||
|
||||
try:
|
||||
books_count = int(books_count_raw)
|
||||
except TypeError, ValueError:
|
||||
books_count = 0
|
||||
|
||||
books: list[BookMetadata] = []
|
||||
for item in list_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 (AttributeError, IndexError, KeyError, TypeError, ValueError) as exc:
|
||||
logger.debug("Failed to parse Hardcover list book for list_id=%s: %s", list_id, exc)
|
||||
|
||||
has_more = offset + len(list_books) < books_count
|
||||
return SearchResult(
|
||||
books=books,
|
||||
page=page,
|
||||
total_found=books_count,
|
||||
has_more=has_more,
|
||||
source_url=source_url,
|
||||
source_title=source_title,
|
||||
)
|
||||
|
||||
@cacheable(
|
||||
ttl_key="METADATA_CACHE_SEARCH_TTL", ttl_default=300, key_prefix="hardcover:list:slug"
|
||||
)
|
||||
def _fetch_list_books(
|
||||
self, slug: str, owner_username: str | None, page: int, limit: int
|
||||
) -> SearchResult:
|
||||
"""Fetch list books by slug, optionally disambiguating by owner username."""
|
||||
if not self.api_key:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
lookup = self._execute_query(LIST_LOOKUP_QUERY, {"slug": slug})
|
||||
if not lookup:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
lists = lookup.get("lists", [])
|
||||
if not isinstance(lists, list) or not lists:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
selected: dict[str, Any] | None = None
|
||||
normalized_owner = owner_username.lower() if owner_username else None
|
||||
if normalized_owner:
|
||||
for item in lists:
|
||||
if not isinstance(item, dict):
|
||||
continue
|
||||
owner_data = item.get("user", {})
|
||||
if not isinstance(owner_data, dict):
|
||||
continue
|
||||
candidate_owner = str(owner_data.get("username") or "").strip().lower()
|
||||
if candidate_owner == normalized_owner:
|
||||
selected = item
|
||||
break
|
||||
|
||||
if selected is None:
|
||||
first_item = lists[0]
|
||||
selected = first_item if isinstance(first_item, dict) else None
|
||||
|
||||
if not selected:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
list_id = coerce_int(selected.get("id"), 0)
|
||||
if list_id < 1:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
return self._fetch_list_books_by_id(list_id, page, limit)
|
||||
|
||||
def _resolve_current_user_id(self) -> str | None:
|
||||
"""Resolve current Hardcover user id from saved settings or API me query."""
|
||||
connected_user_id = _get_connected_user_id()
|
||||
if connected_user_id:
|
||||
return connected_user_id
|
||||
|
||||
result = self._execute_query("query { me { id, username } }", {})
|
||||
if not result:
|
||||
return None
|
||||
|
||||
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 None
|
||||
|
||||
user_id_raw = me_data.get("id")
|
||||
if user_id_raw is None:
|
||||
return None
|
||||
|
||||
user_id = str(user_id_raw)
|
||||
username_raw = me_data.get("username")
|
||||
username = str(username_raw).strip() if username_raw else _get_connected_username()
|
||||
_save_connected_user(user_id, username)
|
||||
return user_id
|
||||
|
||||
def get_user_lists(self) -> list[dict[str, str]]:
|
||||
"""Get authenticated user's own and followed Hardcover lists."""
|
||||
if not self.api_key:
|
||||
return []
|
||||
|
||||
connected_user_id = self._resolve_current_user_id()
|
||||
if not connected_user_id:
|
||||
return self._fetch_user_lists()
|
||||
|
||||
return self._get_user_lists_cached(connected_user_id)
|
||||
|
||||
@cacheable(ttl=120, key_prefix="hardcover:user_lists")
|
||||
def _get_user_lists_cached(self, _cache_user_id: str) -> list[dict[str, str]]:
|
||||
"""Return cached user lists keyed by Hardcover user id."""
|
||||
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:
|
||||
"""Return cached status-shelf books keyed by user id and 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 (AttributeError, KeyError, TypeError, ValueError) as exc:
|
||||
logger.debug(
|
||||
"Failed to parse Hardcover status book for status_id=%s: %s", status_id, exc
|
||||
)
|
||||
|
||||
has_more = offset + len(status_books) < total_found
|
||||
|
||||
# Build source URL for the status shelf
|
||||
source_url = None
|
||||
url_slug = HARDCOVER_STATUS_URL_SLUGS.get(status_id)
|
||||
username = _get_connected_username()
|
||||
if url_slug and username:
|
||||
source_url = f"https://hardcover.app/@{username}/books/{url_slug}"
|
||||
|
||||
return SearchResult(
|
||||
books=books,
|
||||
page=page,
|
||||
total_found=total_found,
|
||||
has_more=has_more,
|
||||
source_url=source_url,
|
||||
)
|
||||
|
||||
def _fetch_user_lists(self) -> list[dict[str, str]]:
|
||||
"""Fetch raw list options from Hardcover me query."""
|
||||
result = self._execute_query(USER_LISTS_QUERY, {})
|
||||
if not result:
|
||||
return []
|
||||
|
||||
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 []
|
||||
|
||||
options: list[dict[str, str]] = []
|
||||
seen_values: set[str] = set()
|
||||
current_username = str(me_data.get("username") or "").strip()
|
||||
|
||||
def _format_label(name: str, books_count: Any) -> str:
|
||||
try:
|
||||
return f"{name} ({int(books_count)})"
|
||||
except TypeError, ValueError:
|
||||
return name
|
||||
|
||||
for status in HARDCOVER_STATUSES:
|
||||
count_data = me_data.get(status["query_key"], {})
|
||||
aggregate = count_data.get("aggregate", {}) if isinstance(count_data, dict) else {}
|
||||
count = aggregate.get("count") if isinstance(aggregate, dict) else None
|
||||
value = f"{HARDCOVER_STATUS_PREFIX}{status['id']}"
|
||||
seen_values.add(value)
|
||||
options.append(
|
||||
{
|
||||
"value": value,
|
||||
"label": _format_label(status["label"], count),
|
||||
"group": HARDCOVER_STATUS_GROUP,
|
||||
}
|
||||
)
|
||||
|
||||
for list_item in me_data.get("lists", []):
|
||||
if not isinstance(list_item, dict):
|
||||
continue
|
||||
list_id = list_item.get("id")
|
||||
slug = str(list_item.get("slug") or "").strip()
|
||||
name = str(list_item.get("name") or "").strip()
|
||||
value = f"id:{list_id}" if list_id is not None else slug
|
||||
if not value or not name or value in seen_values:
|
||||
continue
|
||||
seen_values.add(value)
|
||||
options.append(
|
||||
{
|
||||
"value": value,
|
||||
"label": _format_label(name, list_item.get("books_count")),
|
||||
"group": "My Lists",
|
||||
}
|
||||
)
|
||||
|
||||
for followed_item in me_data.get("followed_lists", []):
|
||||
if not isinstance(followed_item, dict):
|
||||
continue
|
||||
|
||||
list_item = followed_item.get("list", {})
|
||||
if not isinstance(list_item, dict):
|
||||
continue
|
||||
|
||||
list_id = list_item.get("id")
|
||||
slug = str(list_item.get("slug") or "").strip()
|
||||
name = str(list_item.get("name") or "").strip()
|
||||
value = f"id:{list_id}" if list_id is not None else slug
|
||||
if not value or not name or value in seen_values:
|
||||
continue
|
||||
seen_values.add(value)
|
||||
|
||||
option: dict[str, str] = {
|
||||
"value": value,
|
||||
"label": _format_label(name, list_item.get("books_count")),
|
||||
"group": "Followed Lists",
|
||||
}
|
||||
owner_data = list_item.get("user", {})
|
||||
if isinstance(owner_data, dict):
|
||||
owner_username = str(owner_data.get("username") or "").strip()
|
||||
if owner_username:
|
||||
option["description"] = f"by @{owner_username}"
|
||||
elif current_username:
|
||||
option["description"] = f"by @{current_username}"
|
||||
options.append(option)
|
||||
|
||||
return options
|
||||
@@ -1,20 +0,0 @@
|
||||
"""Small Hardcover-specific models and errors."""
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class HardcoverBookTargetState:
|
||||
"""Current Hardcover target state for a specific book."""
|
||||
|
||||
user_book_id: int | None
|
||||
status_id: int | None
|
||||
list_book_ids: dict[int, int]
|
||||
|
||||
|
||||
class HardcoverGraphQLError(ValueError):
|
||||
"""GraphQL request was rejected by Hardcover."""
|
||||
|
||||
|
||||
class HardcoverTargetPayloadError(RuntimeError):
|
||||
"""Hardcover returned an invalid payload while loading book targets."""
|
||||
@@ -1,611 +0,0 @@
|
||||
"""Parsing and search-normalization helpers for Hardcover payloads."""
|
||||
|
||||
import re
|
||||
from contextlib import suppress
|
||||
from datetime import datetime
|
||||
from typing import Any
|
||||
|
||||
from shelfmark.core.logger import setup_logger
|
||||
from shelfmark.core.request_helpers import normalize_optional_text
|
||||
from shelfmark.metadata_providers import BookMetadata, DisplayField
|
||||
|
||||
from .constants import HARDCOVER_MIN_AUTHOR_PARTS
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
def _combine_headline_description(headline: str | None, description: str | None) -> str | None:
|
||||
"""Combine headline (tagline) and description into a single description."""
|
||||
if headline and description:
|
||||
return f"{headline}\n\n{description}"
|
||||
return headline or description
|
||||
|
||||
|
||||
def _extract_cover_url(data: dict, *keys: str) -> str | None:
|
||||
"""Extract cover URL from data dict, trying multiple keys.
|
||||
|
||||
Handles both string URLs and dict with 'url' key.
|
||||
"""
|
||||
for key in keys:
|
||||
value = data.get(key)
|
||||
if value:
|
||||
if isinstance(value, str):
|
||||
return value
|
||||
if isinstance(value, dict):
|
||||
return value.get("url")
|
||||
return None
|
||||
|
||||
|
||||
def _extract_publish_year(data: dict) -> int | None:
|
||||
"""Extract publish year from release_year or release_date fields."""
|
||||
if data.get("release_year"):
|
||||
try:
|
||||
return int(data["release_year"])
|
||||
except ValueError, TypeError:
|
||||
pass
|
||||
if data.get("release_date"):
|
||||
try:
|
||||
return int(str(data["release_date"])[:4])
|
||||
except ValueError, TypeError:
|
||||
pass
|
||||
return None
|
||||
|
||||
|
||||
def _parse_release_date(value: Any) -> datetime | None:
|
||||
"""Parse Hardcover release dates stored as YYYY-MM-DD strings."""
|
||||
if not value:
|
||||
return None
|
||||
|
||||
normalized_value = str(value).strip()
|
||||
if not normalized_value:
|
||||
return None
|
||||
|
||||
try:
|
||||
return datetime.fromisoformat(normalized_value[:10])
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_series_position(value: Any) -> float | None:
|
||||
"""Normalize a series position to a float for sorting and grouping."""
|
||||
if value is None:
|
||||
return None
|
||||
|
||||
try:
|
||||
return float(value)
|
||||
except TypeError, ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _normalize_hardcover_api_key(value: object) -> str:
|
||||
"""Normalize Hardcover API keys, stripping copied auth-header prefixes."""
|
||||
normalized_value = normalize_optional_text(value) or ""
|
||||
return normalized_value.removeprefix("Bearer ").strip()
|
||||
|
||||
|
||||
def _normalize_search_text(value: str) -> str:
|
||||
"""Normalize free-text search input for matching and caching."""
|
||||
return " ".join(value.split()).strip()
|
||||
|
||||
|
||||
def _unwrap_hit_document(hit: Any) -> dict[str, Any] | None:
|
||||
"""Extract the document dict from a Typesense hit, or return None."""
|
||||
if not isinstance(hit, dict):
|
||||
return None
|
||||
item = hit.get("document", hit)
|
||||
return item if isinstance(item, dict) else None
|
||||
|
||||
|
||||
def _search_tokens(value: str) -> list[str]:
|
||||
"""Tokenize search text for lightweight prefix matching."""
|
||||
return re.findall(r"[a-z0-9']+", value.casefold())
|
||||
|
||||
|
||||
def _query_matches_author_name(query: str, author_name: str) -> bool:
|
||||
"""Return True when the query looks like an author-name search."""
|
||||
normalized_query = _normalize_search_text(query)
|
||||
normalized_author_name = _normalize_search_text(author_name)
|
||||
if not normalized_query or not normalized_author_name:
|
||||
return False
|
||||
|
||||
query_folded = normalized_query.casefold()
|
||||
author_folded = normalized_author_name.casefold()
|
||||
if query_folded in author_folded:
|
||||
return True
|
||||
|
||||
query_tokens = _search_tokens(normalized_query)
|
||||
author_tokens = _search_tokens(normalized_author_name)
|
||||
if not query_tokens or not author_tokens:
|
||||
return False
|
||||
|
||||
return all(
|
||||
any(author_token.startswith(query_token) for author_token in author_tokens)
|
||||
for query_token in query_tokens
|
||||
)
|
||||
|
||||
|
||||
def _split_part_base_title(title: str) -> str | None:
|
||||
"""Extract the base title from segmented part releases like ', Part 2'."""
|
||||
normalized_title = _normalize_search_text(title)
|
||||
if not normalized_title:
|
||||
return None
|
||||
|
||||
match = re.match(r"^(?P<base>.+?),\s*Part\s+\d+$", normalized_title, re.IGNORECASE)
|
||||
if not match:
|
||||
return None
|
||||
|
||||
base_title = str(match.group("base") or "").strip()
|
||||
return base_title or None
|
||||
|
||||
|
||||
def _series_allows_split_parts(series_name: str) -> bool:
|
||||
"""Return True for series that intentionally organize split-part releases."""
|
||||
normalized_name = _normalize_search_text(series_name).casefold()
|
||||
if not normalized_name:
|
||||
return False
|
||||
|
||||
markers = (
|
||||
"dramatized adaptation",
|
||||
"graphicaudio",
|
||||
"graphic audio",
|
||||
"(3 parts)",
|
||||
"(2 parts)",
|
||||
"(4 parts)",
|
||||
)
|
||||
return any(marker in normalized_name for marker in markers)
|
||||
|
||||
|
||||
def _extract_typesense_hits(result: dict[str, Any]) -> tuple[list[dict[str, Any]], int]:
|
||||
"""Extract hit documents + total count from Hardcover search output."""
|
||||
root = result.get("search", result) if isinstance(result, dict) else {}
|
||||
results_obj = root.get("results", {}) if isinstance(root, dict) else {}
|
||||
if isinstance(results_obj, dict):
|
||||
hits = results_obj.get("hits", [])
|
||||
found_count = results_obj.get("found", 0)
|
||||
else:
|
||||
hits = results_obj if isinstance(results_obj, list) else []
|
||||
found_count = 0
|
||||
return hits, found_count
|
||||
|
||||
|
||||
def _build_source_url(slug: str) -> str | None:
|
||||
"""Build Hardcover source URL from book slug."""
|
||||
return f"https://hardcover.app/books/{slug}" if slug else None
|
||||
|
||||
|
||||
def _is_probably_series_position(subtitle: str) -> bool:
|
||||
normalized = subtitle.strip().lower()
|
||||
|
||||
# Common patterns: "Book One", "Book 1", "Part 2", "Volume III", etc.
|
||||
if re.match(
|
||||
r"^(book|part|volume|vol\.?|episode)\s+([0-9]+|[ivxlcdm]+|one|two|three|four|five|six|seven|eight|nine|ten)\b",
|
||||
normalized,
|
||||
):
|
||||
return True
|
||||
|
||||
# e.g. "A Novel", "An Epic Fantasy", etc. These add noise to indexer queries.
|
||||
if normalized in {"a novel", "a novella", "a story", "a memoir"}:
|
||||
return True
|
||||
|
||||
# Descriptive subtitles like "A [Name] Novel", "An [Name] Mystery", etc.
|
||||
genre_words = (
|
||||
"novel",
|
||||
"novella",
|
||||
"story",
|
||||
"memoir",
|
||||
"tale",
|
||||
"thriller",
|
||||
"mystery",
|
||||
"romance",
|
||||
"adventure",
|
||||
"epic",
|
||||
"saga",
|
||||
"chronicle",
|
||||
"fantasy",
|
||||
"novel-in-stories",
|
||||
)
|
||||
genre_pattern = "|".join(re.escape(w) for w in genre_words)
|
||||
return bool(re.match(rf"^an?\s+.+\s+({genre_pattern})$", normalized))
|
||||
|
||||
|
||||
def _strip_parenthetical_suffix(title: str) -> str:
|
||||
# Drop trailing qualifiers like "(Unabridged)", "(Illustrated Edition)", etc.
|
||||
return re.sub(r"\s*\([^)]*\)\s*$", "", title).strip()
|
||||
|
||||
|
||||
def _simplify_author_for_search(author: str) -> str | None:
|
||||
"""Return a looser author string for indexer searches.
|
||||
|
||||
Primary goal: reduce mismatch between metadata providers and indexers.
|
||||
Indexers store author names inconsistently ("R.A.", "R. A.", "Salvatore, R.A.")
|
||||
so initials add noise and hurt recall.
|
||||
|
||||
Heuristics:
|
||||
- Strip all initials (single or compound), keeping only full names
|
||||
e.g. "R. A. Salvatore" -> "Salvatore", "George R.R. Martin" -> "George Martin"
|
||||
- Preserve suffixes like "Jr."/"Sr."/"III" as they sometimes matter
|
||||
"""
|
||||
if not author:
|
||||
return None
|
||||
|
||||
normalized = " ".join(author.split()).strip()
|
||||
if not normalized:
|
||||
return None
|
||||
|
||||
# Handle "Last, First ..." -> "First ... Last"
|
||||
if "," in normalized:
|
||||
parts = [p.strip() for p in normalized.split(",") if p.strip()]
|
||||
if len(parts) >= HARDCOVER_MIN_AUTHOR_PARTS:
|
||||
normalized = " ".join([*parts[1:], parts[0]]).strip()
|
||||
|
||||
tokens = normalized.split(" ")
|
||||
if len(tokens) < HARDCOVER_MIN_AUTHOR_PARTS:
|
||||
return None
|
||||
|
||||
keep_suffixes = {"jr", "jr.", "sr", "sr.", "ii", "iii", "iv", "v"}
|
||||
|
||||
simplified: list[str] = []
|
||||
for idx, token in enumerate(tokens):
|
||||
t = token.strip()
|
||||
if not t:
|
||||
continue
|
||||
|
||||
t_lower = t.lower()
|
||||
is_suffix = (idx == len(tokens) - 1) and (t_lower in keep_suffixes)
|
||||
if is_suffix:
|
||||
simplified.append(t)
|
||||
continue
|
||||
|
||||
# Drop all initials: "R.", "R", "R.R.", "J.K.", etc.
|
||||
if re.match(r"^[A-Za-z]$|^([A-Za-z]\.)+[A-Za-z]?$", t):
|
||||
continue
|
||||
|
||||
simplified.append(t)
|
||||
|
||||
if not simplified:
|
||||
return None
|
||||
|
||||
candidate = " ".join(simplified).strip()
|
||||
if candidate.lower() == normalized.lower():
|
||||
return None
|
||||
|
||||
return candidate
|
||||
|
||||
|
||||
def _compute_search_title(
|
||||
title: str,
|
||||
subtitle: str | None,
|
||||
*,
|
||||
series_name: str | None = None,
|
||||
) -> str | None:
|
||||
"""Compute a provider-specific, *looser* title for indexer searching.
|
||||
|
||||
Goal: produce a string that maximizes recall in downstream sources (Prowlarr,
|
||||
IRC bots, etc.). Being too detailed is counterproductive.
|
||||
|
||||
Hardcover often stores titles in a "Series: Book Title" format and places the
|
||||
standalone book title in `subtitle`. When this appears to be the case, prefer
|
||||
the subtitle (unless it looks like a series position or other noise).
|
||||
|
||||
Additional heuristics:
|
||||
- If Hardcover prefixes the series in the title, remove it.
|
||||
- Drop trailing parenthetical qualifiers.
|
||||
"""
|
||||
if not title:
|
||||
return None
|
||||
|
||||
original_title = " ".join(title.split()).strip()
|
||||
|
||||
normalized_title = _strip_parenthetical_suffix(original_title)
|
||||
|
||||
normalized_subtitle = " ".join(subtitle.split()).strip() if subtitle else ""
|
||||
normalized_subtitle = (
|
||||
_strip_parenthetical_suffix(normalized_subtitle) if normalized_subtitle else ""
|
||||
)
|
||||
|
||||
if normalized_subtitle and normalized_subtitle.lower() == normalized_title.lower():
|
||||
normalized_subtitle = ""
|
||||
|
||||
# If subtitle is noise, strip it from the title and use just the prefix.
|
||||
if normalized_subtitle and _is_probably_series_position(normalized_subtitle):
|
||||
match = re.match(r"^(.+?)\s*:\s*(.+)$", normalized_title)
|
||||
if match:
|
||||
suffix = _strip_parenthetical_suffix(match.group(2).strip())
|
||||
if (
|
||||
normalized_subtitle.lower() == suffix.lower()
|
||||
or normalized_subtitle.lower() in suffix.lower()
|
||||
):
|
||||
return None
|
||||
|
||||
# Prefer subtitle when it looks like the real title.
|
||||
if normalized_subtitle and not _is_probably_series_position(normalized_subtitle):
|
||||
match = re.match(r"^(.+?)\s*:\s*(.+)$", normalized_title)
|
||||
if match:
|
||||
prefix = match.group(1).strip()
|
||||
suffix = _strip_parenthetical_suffix(match.group(2).strip())
|
||||
|
||||
prefix_words = len(prefix.split()) if prefix else 0
|
||||
subtitle_words = len(normalized_subtitle.split())
|
||||
|
||||
series_normalized = " ".join(series_name.split()).strip() if series_name else ""
|
||||
if series_normalized and prefix.lower() == series_normalized.lower():
|
||||
return normalized_subtitle
|
||||
|
||||
# If the subtitle is much longer than the prefix, treat it as a descriptive subtitle.
|
||||
if prefix and subtitle_words >= (prefix_words + 4):
|
||||
return prefix
|
||||
|
||||
# Otherwise assume "Series: Book Title" and prefer the subtitle.
|
||||
if (
|
||||
normalized_subtitle.lower() == suffix.lower()
|
||||
or normalized_subtitle.lower() in suffix.lower()
|
||||
):
|
||||
return normalized_subtitle
|
||||
|
||||
# Fallback: if title contains the subtitle, this is likely "Series: Subtitle".
|
||||
if normalized_subtitle.lower() in normalized_title.lower():
|
||||
return normalized_subtitle
|
||||
|
||||
# If we know the series name (from full book fetch), strip it.
|
||||
if series_name:
|
||||
series_normalized = " ".join(series_name.split()).strip()
|
||||
if series_normalized:
|
||||
# Common Hardcover format: "Series: Book Title".
|
||||
prefix = f"{series_normalized}:"
|
||||
if normalized_title.lower().startswith(prefix.lower()):
|
||||
candidate = normalized_title[len(prefix) :].strip()
|
||||
candidate = _strip_parenthetical_suffix(candidate)
|
||||
if candidate and candidate.lower() != normalized_title.lower():
|
||||
return candidate
|
||||
|
||||
# Last resort: return a cleaned version of the title if we removed noise.
|
||||
if normalized_title and normalized_title.lower() != original_title.lower():
|
||||
return normalized_title
|
||||
|
||||
return None
|
||||
|
||||
|
||||
class HardcoverParsingMixin:
|
||||
def _parse_search_result(self, item: dict) -> BookMetadata | None:
|
||||
"""Parse a search result item into BookMetadata."""
|
||||
try:
|
||||
book_id = item.get("id") or item.get("document", {}).get("id")
|
||||
title = item.get("title") or item.get("document", {}).get("title")
|
||||
|
||||
if not book_id or not title:
|
||||
return None
|
||||
|
||||
# Extract authors - use contribution_types to filter author_names if available
|
||||
authors = []
|
||||
|
||||
author_names = item.get("author_names", [])
|
||||
if isinstance(author_names, str):
|
||||
author_names = [author_names]
|
||||
|
||||
contribution_types = item.get("contribution_types", [])
|
||||
|
||||
# If we have parallel arrays, filter to only "Author" contributions
|
||||
if contribution_types and len(contribution_types) == len(author_names):
|
||||
for name, contrib_type in zip(author_names, contribution_types, strict=True):
|
||||
if contrib_type == "Author":
|
||||
authors.append(name)
|
||||
elif author_names:
|
||||
# No contribution_types or length mismatch - use all names as fallback
|
||||
authors = author_names
|
||||
|
||||
# Normalize whitespace in author names (some API data has multiple spaces)
|
||||
authors = [" ".join(name.split()) for name in authors]
|
||||
|
||||
search_author = _simplify_author_for_search(authors[0]) if authors else None
|
||||
|
||||
cover_url = _extract_cover_url(item, "image")
|
||||
publish_year = _extract_publish_year(item)
|
||||
source_url = _build_source_url(item.get("slug", ""))
|
||||
|
||||
# Build display fields from Hardcover-specific data
|
||||
display_fields = []
|
||||
|
||||
# Rating (e.g., "4.5 (3,764)")
|
||||
rating = item.get("rating")
|
||||
ratings_count = item.get("ratings_count")
|
||||
if rating is not None:
|
||||
rating_str = f"{rating:.1f}"
|
||||
if ratings_count:
|
||||
rating_str += f" ({ratings_count:,})"
|
||||
display_fields.append(DisplayField(label="Rating", value=rating_str, icon="star"))
|
||||
|
||||
# Readers (users who have this book)
|
||||
users_count = item.get("users_count")
|
||||
if users_count:
|
||||
display_fields.append(
|
||||
DisplayField(label="Readers", value=f"{users_count:,}", icon="users")
|
||||
)
|
||||
|
||||
# Combine headline and description if both present
|
||||
headline = item.get("headline")
|
||||
description = item.get("description")
|
||||
full_description = _combine_headline_description(headline, description)
|
||||
|
||||
# Extract subtitle if available in search results
|
||||
subtitle = item.get("subtitle")
|
||||
|
||||
return BookMetadata(
|
||||
provider="hardcover",
|
||||
provider_id=str(book_id),
|
||||
title=title,
|
||||
subtitle=subtitle,
|
||||
search_title=_compute_search_title(title, subtitle),
|
||||
search_author=search_author,
|
||||
provider_display_name="Hardcover",
|
||||
authors=authors,
|
||||
cover_url=cover_url,
|
||||
description=full_description,
|
||||
publish_year=publish_year,
|
||||
source_url=source_url,
|
||||
display_fields=display_fields,
|
||||
)
|
||||
|
||||
except (AttributeError, KeyError, TypeError, ValueError) as e:
|
||||
logger.debug("Failed to parse Hardcover search result: %s", e)
|
||||
return None
|
||||
|
||||
def _parse_book(self, book: dict) -> BookMetadata:
|
||||
"""Parse a book object into BookMetadata."""
|
||||
title = str(book.get("title") or "")
|
||||
subtitle = book.get("subtitle")
|
||||
|
||||
# Extract authors - try contributions first (filtered), fall back to cached_contributors
|
||||
authors = []
|
||||
contributions = book.get("contributions") or []
|
||||
cached_contributors = book.get("cached_contributors") or []
|
||||
|
||||
# Try contributions first (filtered to "Author" role only - cleaner data)
|
||||
for contrib in contributions:
|
||||
author = contrib.get("author", {})
|
||||
if author and author.get("name"):
|
||||
authors.append(author["name"])
|
||||
|
||||
# Fallback to cached_contributors if no authors found
|
||||
if not authors:
|
||||
for contrib in cached_contributors:
|
||||
if isinstance(contrib, dict):
|
||||
# Handle nested structure: {"author": {"name": "..."}, "contribution": ...}
|
||||
if contrib.get("author", {}).get("name"):
|
||||
authors.append(contrib["author"]["name"])
|
||||
# Handle flat structure: {"name": "..."}
|
||||
elif contrib.get("name"):
|
||||
authors.append(contrib["name"])
|
||||
elif isinstance(contrib, str):
|
||||
authors.append(contrib)
|
||||
|
||||
# Normalize whitespace in author names (some API data has multiple spaces)
|
||||
authors = [" ".join(name.split()) for name in authors]
|
||||
|
||||
search_author = _simplify_author_for_search(authors[0]) if authors else None
|
||||
|
||||
cover_url = _extract_cover_url(book, "cached_image", "image")
|
||||
publish_year = _extract_publish_year(book)
|
||||
|
||||
# Extract genres from cached_tags
|
||||
genres = []
|
||||
for tag in book.get("cached_tags", []):
|
||||
if isinstance(tag, dict) and tag.get("tag"):
|
||||
genres.append(tag["tag"])
|
||||
elif isinstance(tag, str):
|
||||
genres.append(tag)
|
||||
|
||||
# Get ISBN from direct fields, default_physical_edition, or editions
|
||||
isbn_10 = book.get("isbn_10")
|
||||
isbn_13 = book.get("isbn_13")
|
||||
|
||||
if not isbn_10 and not isbn_13:
|
||||
# Try default_physical_edition first
|
||||
edition = book.get("default_physical_edition")
|
||||
if edition:
|
||||
isbn_10 = edition.get("isbn_10")
|
||||
isbn_13 = edition.get("isbn_13")
|
||||
|
||||
# Fallback to editions array
|
||||
if not isbn_10 and not isbn_13 and book.get("editions"):
|
||||
for ed in book["editions"]:
|
||||
if not isbn_10 and ed.get("isbn_10"):
|
||||
isbn_10 = ed["isbn_10"]
|
||||
if not isbn_13 and ed.get("isbn_13"):
|
||||
isbn_13 = ed["isbn_13"]
|
||||
if isbn_10 and isbn_13:
|
||||
break
|
||||
|
||||
source_url = _build_source_url(book.get("slug", ""))
|
||||
|
||||
# Combine headline and description if both present
|
||||
headline = book.get("headline")
|
||||
description = book.get("description")
|
||||
full_description = _combine_headline_description(headline, description)
|
||||
|
||||
# Extract series info from featured_book_series
|
||||
series_id = None
|
||||
series_name = None
|
||||
series_position = None
|
||||
series_count = None
|
||||
featured_series = book.get("featured_book_series")
|
||||
if featured_series:
|
||||
series_position = featured_series.get("position")
|
||||
series_data = featured_series.get("series")
|
||||
if series_data:
|
||||
if series_data.get("id") is not None:
|
||||
series_id = str(series_data.get("id"))
|
||||
series_name = series_data.get("name")
|
||||
series_count = series_data.get("primary_books_count")
|
||||
|
||||
# Extract titles by language from editions
|
||||
# This allows searching with localized titles when language filter is active
|
||||
titles_by_language: dict[str, str] = {}
|
||||
editions = book.get("editions", [])
|
||||
for edition in editions:
|
||||
edition_title = edition.get("title")
|
||||
lang_data = edition.get("language")
|
||||
if edition_title and lang_data:
|
||||
# Store by various language identifiers for flexible matching
|
||||
# Language name (e.g., "German", "English")
|
||||
lang_name = lang_data.get("language")
|
||||
# 2-letter code (e.g., "de", "en")
|
||||
code2 = lang_data.get("code2")
|
||||
# 3-letter code (e.g., "deu", "eng")
|
||||
code3 = lang_data.get("code3")
|
||||
|
||||
# Store with all available keys (first title wins for each language)
|
||||
if lang_name and lang_name not in titles_by_language:
|
||||
titles_by_language[lang_name] = edition_title
|
||||
if code2 and code2 not in titles_by_language:
|
||||
titles_by_language[code2] = edition_title
|
||||
if code3 and code3 not in titles_by_language:
|
||||
titles_by_language[code3] = edition_title
|
||||
|
||||
# Build display fields from Hardcover-specific metrics
|
||||
display_fields: list[DisplayField] = []
|
||||
|
||||
rating = book.get("rating")
|
||||
ratings_count = book.get("ratings_count")
|
||||
if rating is not None:
|
||||
try:
|
||||
rating_str = f"{float(rating):.1f}"
|
||||
except TypeError, ValueError:
|
||||
rating_str = str(rating)
|
||||
|
||||
if ratings_count:
|
||||
with suppress(TypeError, ValueError):
|
||||
rating_str += f" ({int(ratings_count):,})"
|
||||
|
||||
display_fields.append(DisplayField(label="Rating", value=rating_str, icon="star"))
|
||||
|
||||
users_count = book.get("users_count")
|
||||
if users_count:
|
||||
try:
|
||||
readers_value = f"{int(users_count):,}"
|
||||
except TypeError, ValueError:
|
||||
readers_value = str(users_count)
|
||||
display_fields.append(DisplayField(label="Readers", value=readers_value, icon="users"))
|
||||
|
||||
return BookMetadata(
|
||||
provider="hardcover",
|
||||
provider_id=str(book["id"]),
|
||||
title=title,
|
||||
subtitle=subtitle,
|
||||
search_title=_compute_search_title(title, subtitle, series_name=series_name),
|
||||
search_author=search_author,
|
||||
provider_display_name="Hardcover",
|
||||
authors=authors,
|
||||
isbn_10=isbn_10,
|
||||
isbn_13=isbn_13,
|
||||
cover_url=cover_url,
|
||||
description=full_description,
|
||||
publish_year=publish_year,
|
||||
genres=genres,
|
||||
source_url=source_url,
|
||||
series_id=series_id,
|
||||
series_name=series_name,
|
||||
series_position=series_position,
|
||||
series_count=series_count,
|
||||
titles_by_language=titles_by_language,
|
||||
display_fields=display_fields,
|
||||
)
|
||||
@@ -1,106 +0,0 @@
|
||||
"""Hardcover.app metadata provider. Requires API key."""
|
||||
|
||||
from typing import Any, ClassVar
|
||||
|
||||
import requests
|
||||
|
||||
from shelfmark.core.config import config as app_config
|
||||
from shelfmark.metadata_providers import (
|
||||
DynamicSelectSearchField,
|
||||
MetadataCapability,
|
||||
MetadataProvider,
|
||||
SearchField,
|
||||
SortOrder,
|
||||
TextSearchField,
|
||||
register_provider,
|
||||
register_provider_kwargs,
|
||||
)
|
||||
|
||||
from .client import HardcoverClientMixin
|
||||
from .lists import HardcoverListsMixin
|
||||
from .parsing import HardcoverParsingMixin, _normalize_hardcover_api_key
|
||||
from .search import HardcoverSearchMixin
|
||||
from .targets import HardcoverTargetsMixin
|
||||
|
||||
|
||||
@register_provider_kwargs("hardcover")
|
||||
def _hardcover_kwargs() -> dict[str, Any]:
|
||||
"""Provide Hardcover-specific constructor kwargs."""
|
||||
return {"api_key": app_config.get("HARDCOVER_API_KEY", "")}
|
||||
|
||||
|
||||
@register_provider("hardcover")
|
||||
class HardcoverProvider(
|
||||
HardcoverSearchMixin,
|
||||
HardcoverListsMixin,
|
||||
HardcoverTargetsMixin,
|
||||
HardcoverClientMixin,
|
||||
HardcoverParsingMixin,
|
||||
MetadataProvider,
|
||||
):
|
||||
"""Hardcover.app metadata provider using GraphQL API."""
|
||||
|
||||
name = "hardcover"
|
||||
display_name = "Hardcover"
|
||||
requires_auth = True
|
||||
supported_sorts: ClassVar[tuple[SortOrder, ...]] = (
|
||||
SortOrder.RELEVANCE,
|
||||
SortOrder.POPULARITY,
|
||||
SortOrder.RATING,
|
||||
SortOrder.NEWEST,
|
||||
SortOrder.OLDEST,
|
||||
SortOrder.SERIES_ORDER,
|
||||
)
|
||||
capabilities: ClassVar[tuple[MetadataCapability, ...]] = (
|
||||
MetadataCapability(
|
||||
key="view_series",
|
||||
field_key="series",
|
||||
sort=SortOrder.SERIES_ORDER,
|
||||
),
|
||||
)
|
||||
search_fields: ClassVar[tuple[SearchField, ...]] = (
|
||||
TextSearchField(
|
||||
key="author",
|
||||
label="Author",
|
||||
placeholder="Search author...",
|
||||
description="Search by author name",
|
||||
suggestions_endpoint="/api/metadata/field-options?provider=hardcover&field=author",
|
||||
),
|
||||
TextSearchField(
|
||||
key="title",
|
||||
label="Title",
|
||||
placeholder="Search title...",
|
||||
description="Search by book title",
|
||||
),
|
||||
TextSearchField(
|
||||
key="series",
|
||||
label="Series",
|
||||
placeholder="Search series...",
|
||||
description="Search by series name",
|
||||
suggestions_endpoint="/api/metadata/field-options?provider=hardcover&field=series",
|
||||
),
|
||||
DynamicSelectSearchField(
|
||||
key="hardcover_list",
|
||||
label="List",
|
||||
options_endpoint="/api/metadata/field-options?provider=hardcover&field=hardcover_list",
|
||||
placeholder="Browse a list...",
|
||||
description="Browse books from a Hardcover list",
|
||||
),
|
||||
)
|
||||
|
||||
def __init__(self, api_key: str | None = None) -> None:
|
||||
"""Initialize provider with optional API key (falls back to config)."""
|
||||
raw_key = api_key or app_config.get("HARDCOVER_API_KEY", "")
|
||||
self.api_key = _normalize_hardcover_api_key(raw_key)
|
||||
self.session = requests.Session()
|
||||
if self.api_key:
|
||||
self.session.headers.update(
|
||||
{
|
||||
"Authorization": f"Bearer {self.api_key}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
)
|
||||
|
||||
def is_available(self) -> bool:
|
||||
"""Check if provider is configured with an API key."""
|
||||
return bool(self.api_key)
|
||||
@@ -1,525 +0,0 @@
|
||||
"""GraphQL operations used by the Hardcover metadata provider."""
|
||||
|
||||
LIST_LOOKUP_QUERY = """
|
||||
query LookupListsBySlug($slug: String!) {
|
||||
lists(where: {slug: {_eq: $slug}}, limit: 20) {
|
||||
id
|
||||
slug
|
||||
user {
|
||||
username
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
LIST_BOOKS_BY_ID_QUERY = """
|
||||
query GetListBooksById($id: Int!, $limit: Int!, $offset: Int!) {
|
||||
lists(where: {id: {_eq: $id}}, limit: 1) {
|
||||
name
|
||||
slug
|
||||
user {
|
||||
username
|
||||
}
|
||||
books_count
|
||||
list_books(order_by: {position: asc}, 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 {
|
||||
id
|
||||
name
|
||||
primary_books_count
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
USER_LISTS_QUERY = """
|
||||
query GetUserLists {
|
||||
me {
|
||||
id
|
||||
username
|
||||
want_to_read_count: user_books_aggregate(where: {status_id: {_eq: 1}}) {
|
||||
aggregate {
|
||||
count(columns: [book_id], distinct: true)
|
||||
}
|
||||
}
|
||||
currently_reading_count: user_books_aggregate(where: {status_id: {_eq: 2}}) {
|
||||
aggregate {
|
||||
count(columns: [book_id], distinct: true)
|
||||
}
|
||||
}
|
||||
read_count: user_books_aggregate(where: {status_id: {_eq: 3}}) {
|
||||
aggregate {
|
||||
count(columns: [book_id], distinct: true)
|
||||
}
|
||||
}
|
||||
did_not_finish_count: user_books_aggregate(where: {status_id: {_eq: 5}}) {
|
||||
aggregate {
|
||||
count(columns: [book_id], distinct: true)
|
||||
}
|
||||
}
|
||||
lists(order_by: {name: asc}) {
|
||||
id
|
||||
name
|
||||
slug
|
||||
books_count
|
||||
}
|
||||
followed_lists(order_by: {created_at: desc}) {
|
||||
list {
|
||||
id
|
||||
name
|
||||
slug
|
||||
books_count
|
||||
user {
|
||||
username
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
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 {
|
||||
id
|
||||
name
|
||||
primary_books_count
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
status_books_aggregate: user_books_aggregate(where: {status_id: {_eq: $statusId}}) {
|
||||
aggregate {
|
||||
count(columns: [book_id], distinct: true)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
BOOK_TARGET_MEMBERSHIP_QUERY = """
|
||||
query GetBookTargetMembership($bookId: Int!) {
|
||||
me {
|
||||
user_books(where: {book_id: {_eq: $bookId}}, limit: 1, order_by: [{created_at: desc}]) {
|
||||
id
|
||||
status_id
|
||||
}
|
||||
lists {
|
||||
id
|
||||
list_books(where: {book_id: {_eq: $bookId}}, limit: 1) {
|
||||
id
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
BOOK_TARGET_MEMBERSHIP_BATCH_QUERY = """
|
||||
query GetBookTargetMembershipBatch($bookIds: [Int!]!) {
|
||||
me {
|
||||
user_books(where: {book_id: {_in: $bookIds}}, order_by: [{created_at: desc}]) {
|
||||
id
|
||||
book_id
|
||||
status_id
|
||||
}
|
||||
lists {
|
||||
id
|
||||
list_books(where: {book_id: {_in: $bookIds}}) {
|
||||
id
|
||||
book_id
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
INSERT_USER_BOOK_MUTATION = """
|
||||
mutation AddBookToStatus($bookId: Int!, $statusId: Int!) {
|
||||
insert_user_book(object: {book_id: $bookId, status_id: $statusId}) {
|
||||
id
|
||||
error
|
||||
user_book {
|
||||
id
|
||||
book_id
|
||||
status_id
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
UPDATE_USER_BOOK_MUTATION = """
|
||||
mutation UpdateBookStatus($userBookId: Int!, $statusId: Int!) {
|
||||
update_user_book(id: $userBookId, object: {status_id: $statusId}) {
|
||||
id
|
||||
error
|
||||
user_book {
|
||||
id
|
||||
book_id
|
||||
status_id
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
DELETE_USER_BOOK_MUTATION = """
|
||||
mutation RemoveBookStatus($userBookId: Int!) {
|
||||
delete_user_book(id: $userBookId) {
|
||||
id
|
||||
book_id
|
||||
user_id
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
INSERT_LIST_BOOK_MUTATION = """
|
||||
mutation AddBookToList($bookId: Int!, $listId: Int!) {
|
||||
insert_list_book(object: {book_id: $bookId, list_id: $listId}) {
|
||||
id
|
||||
list_book {
|
||||
id
|
||||
book_id
|
||||
list_id
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
DELETE_LIST_BOOK_MUTATION = """
|
||||
mutation RemoveBookFromList($listBookId: Int!) {
|
||||
delete_list_book(id: $listBookId) {
|
||||
id
|
||||
list_id
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
SEARCH_FIELD_OPTIONS_QUERY = """
|
||||
query SearchFieldOptions(
|
||||
$query: String!,
|
||||
$queryType: String!,
|
||||
$limit: Int!,
|
||||
$page: Int!,
|
||||
$sort: String,
|
||||
$fields: String,
|
||||
$weights: String
|
||||
) {
|
||||
search(
|
||||
query: $query,
|
||||
query_type: $queryType,
|
||||
per_page: $limit,
|
||||
page: $page,
|
||||
sort: $sort,
|
||||
fields: $fields,
|
||||
weights: $weights
|
||||
) {
|
||||
results
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
SERIES_BY_AUTHOR_IDS_QUERY = """
|
||||
query SeriesByAuthorIds($authorIds: [Int!], $limit: Int!) {
|
||||
series(
|
||||
where: {
|
||||
author_id: {_in: $authorIds},
|
||||
canonical_id: {_is_null: true},
|
||||
state: {_eq: "active"}
|
||||
},
|
||||
limit: $limit,
|
||||
order_by: [{primary_books_count: desc_nulls_last}, {books_count: desc}, {name: asc}]
|
||||
) {
|
||||
id
|
||||
name
|
||||
primary_books_count
|
||||
books_count
|
||||
author {
|
||||
name
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
SERIES_BOOKS_BY_ID_QUERY = """
|
||||
query GetSeriesBooks($seriesId: Int!) {
|
||||
series(where: {id: {_eq: $seriesId}}, limit: 1) {
|
||||
id
|
||||
name
|
||||
primary_books_count
|
||||
book_series(
|
||||
where: {
|
||||
book: {
|
||||
canonical_id: {_is_null: true},
|
||||
state: {_in: ["normalized", "normalizing"]}
|
||||
}
|
||||
}
|
||||
order_by: [{position: asc_nulls_last}, {book_id: asc}]
|
||||
) {
|
||||
position
|
||||
book {
|
||||
id
|
||||
title
|
||||
subtitle
|
||||
slug
|
||||
release_date
|
||||
headline
|
||||
description
|
||||
pages
|
||||
rating
|
||||
ratings_count
|
||||
users_count
|
||||
compilation
|
||||
editions_count
|
||||
cached_image
|
||||
cached_contributors
|
||||
contributions(where: {contribution: {_eq: "Author"}}) {
|
||||
author {
|
||||
name
|
||||
}
|
||||
}
|
||||
featured_book_series {
|
||||
position
|
||||
series {
|
||||
id
|
||||
name
|
||||
primary_books_count
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
AUTHOR_BOOKS_BY_ID_QUERY = """
|
||||
query GetAuthorBooks($authorId: Int!, $limit: Int!, $offset: Int!) {
|
||||
authors(where: {id: {_eq: $authorId}}, limit: 1) {
|
||||
name
|
||||
contributions(
|
||||
where: {
|
||||
contributable_type: {_eq: "Book"},
|
||||
book: {
|
||||
canonical_id: {_is_null: true},
|
||||
state: {_in: ["normalized", "normalizing"]}
|
||||
}
|
||||
},
|
||||
order_by: [
|
||||
{book: {users_count: desc_nulls_last}},
|
||||
{book: {ratings_count: desc_nulls_last}},
|
||||
{book: {release_date: asc_nulls_last}},
|
||||
{book: {id: asc}}
|
||||
],
|
||||
limit: $limit,
|
||||
offset: $offset
|
||||
) {
|
||||
contribution
|
||||
book {
|
||||
id
|
||||
title
|
||||
subtitle
|
||||
slug
|
||||
release_date
|
||||
headline
|
||||
description
|
||||
pages
|
||||
rating
|
||||
ratings_count
|
||||
users_count
|
||||
compilation
|
||||
editions_count
|
||||
cached_image
|
||||
cached_contributors
|
||||
contributions(where: {contribution: {_eq: "Author"}}) {
|
||||
author {
|
||||
name
|
||||
}
|
||||
}
|
||||
featured_book_series {
|
||||
position
|
||||
series {
|
||||
id
|
||||
name
|
||||
primary_books_count
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
contributions_aggregate(
|
||||
where: {
|
||||
contributable_type: {_eq: "Book"},
|
||||
book: {
|
||||
canonical_id: {_is_null: true},
|
||||
state: {_in: ["normalized", "normalizing"]}
|
||||
}
|
||||
}
|
||||
) {
|
||||
aggregate {
|
||||
count
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
SEARCH_BOOKS_WITH_FIELDS_QUERY = """
|
||||
query SearchBooks(
|
||||
$query: String!,
|
||||
$limit: Int!,
|
||||
$page: Int!,
|
||||
$sort: String,
|
||||
$fields: String,
|
||||
$weights: String
|
||||
) {
|
||||
search(
|
||||
query: $query,
|
||||
query_type: "Book",
|
||||
per_page: $limit,
|
||||
page: $page,
|
||||
sort: $sort,
|
||||
fields: $fields,
|
||||
weights: $weights
|
||||
) {
|
||||
results
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
SEARCH_BOOKS_QUERY = """
|
||||
query SearchBooks($query: String!, $limit: Int!, $page: Int!, $sort: String) {
|
||||
search(query: $query, query_type: "Book", per_page: $limit, page: $page, sort: $sort) {
|
||||
results
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
GET_BOOK_QUERY = """
|
||||
query GetBook($id: Int!) {
|
||||
books(where: {id: {_eq: $id}}, limit: 1) {
|
||||
id
|
||||
title
|
||||
subtitle
|
||||
slug
|
||||
release_date
|
||||
headline
|
||||
description
|
||||
pages
|
||||
cached_image
|
||||
cached_tags
|
||||
cached_contributors
|
||||
contributions(where: {contribution: {_eq: "Author"}}) {
|
||||
author {
|
||||
name
|
||||
}
|
||||
}
|
||||
default_physical_edition {
|
||||
isbn_10
|
||||
isbn_13
|
||||
}
|
||||
featured_book_series {
|
||||
position
|
||||
series {
|
||||
id
|
||||
name
|
||||
primary_books_count
|
||||
}
|
||||
}
|
||||
editions(
|
||||
distinct_on: language_id
|
||||
order_by: [{language_id: asc}, {users_count: desc}]
|
||||
limit: 200
|
||||
) {
|
||||
title
|
||||
language {
|
||||
language
|
||||
code2
|
||||
code3
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
|
||||
SEARCH_BY_ISBN_QUERY = """
|
||||
query SearchByISBN($isbn: String!) {
|
||||
editions(
|
||||
where: {
|
||||
_or: [
|
||||
{isbn_10: {_eq: $isbn}},
|
||||
{isbn_13: {_eq: $isbn}}
|
||||
]
|
||||
},
|
||||
limit: 1
|
||||
) {
|
||||
isbn_10
|
||||
isbn_13
|
||||
book {
|
||||
id
|
||||
title
|
||||
subtitle
|
||||
slug
|
||||
release_date
|
||||
headline
|
||||
description
|
||||
pages
|
||||
cached_image
|
||||
cached_tags
|
||||
contributions(where: {contribution: {_eq: "Author"}}) {
|
||||
author {
|
||||
name
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
"""
|
||||
@@ -1,844 +0,0 @@
|
||||
"""Search, typeahead, series, and book lookup workflows for Hardcover."""
|
||||
|
||||
from datetime import UTC, datetime
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from shelfmark.core.cache import cacheable
|
||||
from shelfmark.core.config import config as app_config
|
||||
from shelfmark.core.logger import setup_logger
|
||||
from shelfmark.core.request_helpers import coerce_bool, coerce_int
|
||||
from shelfmark.metadata_providers import (
|
||||
BookMetadata,
|
||||
MetadataSearchOptions,
|
||||
SearchResult,
|
||||
SearchType,
|
||||
SortOrder,
|
||||
)
|
||||
|
||||
from .constants import (
|
||||
AUTHOR_SUGGESTION_FIELDS,
|
||||
AUTHOR_SUGGESTION_SORT,
|
||||
AUTHOR_SUGGESTION_WEIGHTS,
|
||||
HARDCOVER_LIST_ID_PREFIX,
|
||||
HARDCOVER_MAX_SERIES_OPTIONS,
|
||||
HARDCOVER_MIN_TYPEAHEAD_QUERY_LENGTH,
|
||||
HARDCOVER_PAGE_SIZE,
|
||||
HARDCOVER_STATUS_PREFIX,
|
||||
SERIES_SEARCH_FIELDS,
|
||||
SERIES_SEARCH_SORT,
|
||||
SERIES_SEARCH_WEIGHTS,
|
||||
SORT_MAPPING,
|
||||
TITLE_SUGGESTION_FIELDS,
|
||||
TITLE_SUGGESTION_SORT,
|
||||
TITLE_SUGGESTION_WEIGHTS,
|
||||
)
|
||||
from .parsing import (
|
||||
_extract_typesense_hits,
|
||||
_normalize_search_text,
|
||||
_normalize_series_position,
|
||||
_parse_release_date,
|
||||
_query_matches_author_name,
|
||||
_series_allows_split_parts,
|
||||
_split_part_base_title,
|
||||
_unwrap_hit_document,
|
||||
)
|
||||
from .queries import (
|
||||
AUTHOR_BOOKS_BY_ID_QUERY,
|
||||
GET_BOOK_QUERY,
|
||||
SEARCH_BOOKS_QUERY,
|
||||
SEARCH_BOOKS_WITH_FIELDS_QUERY,
|
||||
SEARCH_BY_ISBN_QUERY,
|
||||
SEARCH_FIELD_OPTIONS_QUERY,
|
||||
SERIES_BOOKS_BY_ID_QUERY,
|
||||
SERIES_BY_AUTHOR_IDS_QUERY,
|
||||
)
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
class HardcoverSearchMixin:
|
||||
if TYPE_CHECKING:
|
||||
api_key: str
|
||||
|
||||
def _detect_list_url(self, query: str) -> tuple[str | None, str] | None: ...
|
||||
|
||||
def _execute_query(
|
||||
self,
|
||||
query: str,
|
||||
variables: dict[str, Any],
|
||||
*,
|
||||
raise_on_error: bool = False,
|
||||
) -> dict[str, Any] | None: ...
|
||||
|
||||
def _fetch_current_user_books_by_status(
|
||||
self, status_id: int, page: int, limit: int
|
||||
) -> SearchResult: ...
|
||||
|
||||
def _fetch_list_books(
|
||||
self, slug: str, owner_username: str | None, page: int, limit: int
|
||||
) -> SearchResult: ...
|
||||
|
||||
def _fetch_list_books_by_id(self, list_id: int, page: int, limit: int) -> SearchResult: ...
|
||||
|
||||
def _parse_book(self, book: dict[str, Any]) -> BookMetadata: ...
|
||||
|
||||
@staticmethod
|
||||
def _parse_prefixed_int(value: str, label: str = "target") -> int: ...
|
||||
|
||||
def _parse_search_result(self, item: dict[str, Any]) -> BookMetadata | None: ...
|
||||
|
||||
def get_user_lists(self) -> list[dict[str, str]]: ...
|
||||
|
||||
def _build_search_params(
|
||||
self, default_query: str, author: str, title: str, series: str
|
||||
) -> tuple[str, str | None, str | None]:
|
||||
"""Build search query, fields, and weights based on provided values.
|
||||
|
||||
Returns (query, fields, weights) tuple. Fields/weights are None for general search.
|
||||
"""
|
||||
if author and not title and not series:
|
||||
return author, None, None
|
||||
if title and not author and not series:
|
||||
return title, "title,alternative_titles", "5,1"
|
||||
if author and title and not series:
|
||||
return f"{title} {author}", "title,alternative_titles,author_names", "5,1,3"
|
||||
return default_query, None, None
|
||||
|
||||
def get_search_field_options(
|
||||
self,
|
||||
field_key: str,
|
||||
query: str | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
"""Provide dynamic options for Hardcover-specific advanced fields."""
|
||||
if field_key == "author":
|
||||
return self._search_author_options(query or "")
|
||||
if field_key == "title":
|
||||
return self._search_title_options(query or "")
|
||||
if field_key == "series":
|
||||
return self._search_series_options(query or "")
|
||||
if field_key == "hardcover_list":
|
||||
return self.get_user_lists()
|
||||
return []
|
||||
|
||||
def _search_field_hits(
|
||||
self,
|
||||
*,
|
||||
query: str,
|
||||
query_type: str,
|
||||
limit: int,
|
||||
sort: str | None,
|
||||
fields: str | None,
|
||||
weights: str | None,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Run a Hardcover search request for field-level typeahead options."""
|
||||
normalized_query = _normalize_search_text(query)
|
||||
if not self.api_key or len(normalized_query) < HARDCOVER_MIN_TYPEAHEAD_QUERY_LENGTH:
|
||||
return []
|
||||
|
||||
result = self._execute_query(
|
||||
SEARCH_FIELD_OPTIONS_QUERY,
|
||||
{
|
||||
"query": normalized_query,
|
||||
"queryType": query_type,
|
||||
"limit": limit,
|
||||
"page": 1,
|
||||
"sort": sort,
|
||||
"fields": fields,
|
||||
"weights": weights,
|
||||
},
|
||||
)
|
||||
if not result:
|
||||
return []
|
||||
|
||||
hits, _found_count = _extract_typesense_hits(result)
|
||||
return hits
|
||||
|
||||
def _search_series_by_matching_author(self, query: str) -> list[dict[str, Any]]:
|
||||
"""Return direct series rows when the query clearly matches an author."""
|
||||
author_hits = self._search_field_hits(
|
||||
query=query,
|
||||
query_type="Author",
|
||||
limit=2,
|
||||
sort=AUTHOR_SUGGESTION_SORT,
|
||||
fields=AUTHOR_SUGGESTION_FIELDS,
|
||||
weights=AUTHOR_SUGGESTION_WEIGHTS,
|
||||
)
|
||||
|
||||
author_ids: list[int] = []
|
||||
for hit in author_hits:
|
||||
item = _unwrap_hit_document(hit)
|
||||
if item is None:
|
||||
continue
|
||||
|
||||
author_name = str(item.get("name") or "").strip()
|
||||
if not _query_matches_author_name(query, author_name):
|
||||
continue
|
||||
|
||||
author_id = coerce_int(item.get("id"), 0)
|
||||
if author_id < 1:
|
||||
continue
|
||||
|
||||
if author_id not in author_ids:
|
||||
author_ids.append(author_id)
|
||||
|
||||
if not author_ids:
|
||||
return []
|
||||
|
||||
result = self._execute_query(
|
||||
SERIES_BY_AUTHOR_IDS_QUERY,
|
||||
{
|
||||
"authorIds": author_ids,
|
||||
"limit": 7,
|
||||
},
|
||||
)
|
||||
if not result:
|
||||
return []
|
||||
|
||||
series_rows = result.get("series", [])
|
||||
return [row for row in series_rows if isinstance(row, dict)]
|
||||
|
||||
@cacheable(ttl=120, key_prefix="hardcover:author:options")
|
||||
def _search_author_options(self, query: str) -> list[dict[str, str]]:
|
||||
"""Return typeahead options for Hardcover author search."""
|
||||
hits = self._search_field_hits(
|
||||
query=query,
|
||||
query_type="Author",
|
||||
limit=7,
|
||||
sort=AUTHOR_SUGGESTION_SORT,
|
||||
fields=AUTHOR_SUGGESTION_FIELDS,
|
||||
weights=AUTHOR_SUGGESTION_WEIGHTS,
|
||||
)
|
||||
options: list[dict[str, str]] = []
|
||||
seen_labels: set[str] = set()
|
||||
|
||||
for hit in hits:
|
||||
item = _unwrap_hit_document(hit)
|
||||
if item is None:
|
||||
continue
|
||||
|
||||
author_id = coerce_int(item.get("id"), 0)
|
||||
label = str(item.get("name") or "").strip()
|
||||
normalized_label = label.casefold()
|
||||
if author_id < 1 or not label or normalized_label in seen_labels:
|
||||
continue
|
||||
|
||||
seen_labels.add(normalized_label)
|
||||
options.append({"value": f"id:{author_id}", "label": label})
|
||||
|
||||
return options
|
||||
|
||||
@cacheable(ttl=120, key_prefix="hardcover:title:options")
|
||||
def _search_title_options(self, query: str) -> list[dict[str, str]]:
|
||||
"""Return typeahead options for Hardcover title search."""
|
||||
hits = self._search_field_hits(
|
||||
query=query,
|
||||
query_type="Book",
|
||||
limit=7,
|
||||
sort=TITLE_SUGGESTION_SORT,
|
||||
fields=TITLE_SUGGESTION_FIELDS,
|
||||
weights=TITLE_SUGGESTION_WEIGHTS,
|
||||
)
|
||||
|
||||
exclude_compilations = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_COMPILATIONS", False),
|
||||
default=False,
|
||||
)
|
||||
exclude_unreleased = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_UNRELEASED", False),
|
||||
default=False,
|
||||
)
|
||||
current_year = datetime.now(UTC).year
|
||||
|
||||
options: list[dict[str, str]] = []
|
||||
seen_labels: set[str] = set()
|
||||
|
||||
for hit in hits:
|
||||
item = _unwrap_hit_document(hit)
|
||||
if item is None:
|
||||
continue
|
||||
|
||||
if exclude_compilations and item.get("compilation"):
|
||||
continue
|
||||
|
||||
if exclude_unreleased:
|
||||
release_year = item.get("release_year")
|
||||
try:
|
||||
if release_year is not None and int(release_year) > current_year:
|
||||
continue
|
||||
except TypeError, ValueError:
|
||||
pass
|
||||
|
||||
label = str(item.get("title") or "").strip()
|
||||
normalized_label = label.casefold()
|
||||
if not label or normalized_label in seen_labels:
|
||||
continue
|
||||
|
||||
seen_labels.add(normalized_label)
|
||||
options.append({"value": label, "label": label})
|
||||
|
||||
return options
|
||||
|
||||
def _format_series_option_description(self, item: dict[str, Any]) -> str | None:
|
||||
"""Build a short description for a series suggestion option."""
|
||||
author_name = item.get("author_name")
|
||||
if not author_name:
|
||||
author_data = item.get("author")
|
||||
if isinstance(author_data, dict):
|
||||
author_name = author_data.get("name")
|
||||
|
||||
parts: list[str] = []
|
||||
if author_name:
|
||||
parts.append(f"by {author_name}")
|
||||
|
||||
books_count = item.get("primary_books_count")
|
||||
if books_count is None:
|
||||
books_count = item.get("books_count")
|
||||
|
||||
try:
|
||||
if books_count is not None:
|
||||
books_count_int = int(books_count)
|
||||
parts.append(f"{books_count_int} book{'s' if books_count_int != 1 else ''}")
|
||||
except TypeError, ValueError:
|
||||
pass
|
||||
|
||||
return " • ".join(parts) if parts else None
|
||||
|
||||
@cacheable(ttl=120, key_prefix="hardcover:series:options")
|
||||
def _search_series_options(self, query: str) -> list[dict[str, str]]:
|
||||
"""Return typeahead options for Hardcover series search."""
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||||
author_future = executor.submit(self._search_series_by_matching_author, query)
|
||||
series_future = executor.submit(
|
||||
self._search_field_hits,
|
||||
query=query,
|
||||
query_type="Series",
|
||||
limit=7,
|
||||
sort=SERIES_SEARCH_SORT,
|
||||
fields=SERIES_SEARCH_FIELDS,
|
||||
weights=SERIES_SEARCH_WEIGHTS,
|
||||
)
|
||||
|
||||
author_series = author_future.result()
|
||||
hits = series_future.result()
|
||||
options: list[dict[str, str]] = []
|
||||
seen_values: set[str] = set()
|
||||
|
||||
series_items: list[dict[str, Any]] = []
|
||||
series_items.extend(author_series)
|
||||
series_items.extend(doc for hit in hits if (doc := _unwrap_hit_document(hit)) is not None)
|
||||
|
||||
for item in series_items:
|
||||
series_id = item.get("id")
|
||||
name = str(item.get("name") or "").strip()
|
||||
if series_id is None or not name:
|
||||
continue
|
||||
|
||||
value = f"id:{series_id}"
|
||||
if value in seen_values:
|
||||
continue
|
||||
seen_values.add(value)
|
||||
|
||||
option: dict[str, str] = {
|
||||
"value": value,
|
||||
"label": name,
|
||||
}
|
||||
description = self._format_series_option_description(item)
|
||||
if description:
|
||||
option["description"] = description
|
||||
options.append(option)
|
||||
if len(options) >= HARDCOVER_MAX_SERIES_OPTIONS:
|
||||
break
|
||||
|
||||
return options
|
||||
|
||||
def _resolve_series_search_value(self, series_value: str) -> dict[str, Any] | None:
|
||||
"""Resolve a series field value to a canonical Hardcover series."""
|
||||
normalized_value = _normalize_search_text(series_value)
|
||||
if not normalized_value:
|
||||
return None
|
||||
|
||||
if normalized_value.startswith(HARDCOVER_LIST_ID_PREFIX):
|
||||
try:
|
||||
return {"id": self._parse_prefixed_int(normalized_value, "series id")}
|
||||
except ValueError:
|
||||
logger.debug("Invalid Hardcover series id field value: %s", normalized_value)
|
||||
return None
|
||||
|
||||
result = self._execute_query(
|
||||
SEARCH_FIELD_OPTIONS_QUERY,
|
||||
{
|
||||
"query": normalized_value,
|
||||
"queryType": "Series",
|
||||
"limit": 10,
|
||||
"page": 1,
|
||||
"sort": SERIES_SEARCH_SORT,
|
||||
"fields": SERIES_SEARCH_FIELDS,
|
||||
"weights": SERIES_SEARCH_WEIGHTS,
|
||||
},
|
||||
)
|
||||
if not result:
|
||||
return None
|
||||
|
||||
hits, _found_count = _extract_typesense_hits(result)
|
||||
if not hits:
|
||||
return None
|
||||
|
||||
normalized_lookup = normalized_value.lower()
|
||||
candidates: list[dict[str, Any]] = []
|
||||
for hit in hits:
|
||||
item = _unwrap_hit_document(hit)
|
||||
if item is None:
|
||||
continue
|
||||
series_id = coerce_int(item.get("id"), 0)
|
||||
if series_id < 1:
|
||||
continue
|
||||
name = str(item.get("name") or "").strip()
|
||||
if not name:
|
||||
continue
|
||||
candidates.append({"id": series_id, "name": name})
|
||||
|
||||
if not candidates:
|
||||
return None
|
||||
|
||||
exact_match = next(
|
||||
(
|
||||
candidate
|
||||
for candidate in candidates
|
||||
if candidate["name"].lower() == normalized_lookup
|
||||
),
|
||||
None,
|
||||
)
|
||||
return exact_match or candidates[0]
|
||||
|
||||
@cacheable(
|
||||
ttl_key="METADATA_CACHE_SEARCH_TTL", ttl_default=300, key_prefix="hardcover:series:rows:v4"
|
||||
)
|
||||
def _fetch_series_ordered_rows(
|
||||
self,
|
||||
series_id: int,
|
||||
*,
|
||||
exclude_compilations: bool,
|
||||
exclude_unreleased: bool,
|
||||
) -> dict[str, Any]:
|
||||
"""Fetch and process all books for a series (cached independently of page)."""
|
||||
empty: dict[str, Any] = {"rows": [], "series_name": "", "total": 0}
|
||||
if not self.api_key:
|
||||
return empty
|
||||
|
||||
result = self._execute_query(
|
||||
SERIES_BOOKS_BY_ID_QUERY,
|
||||
{"seriesId": series_id},
|
||||
)
|
||||
if not result:
|
||||
return empty
|
||||
|
||||
series_items = result.get("series", [])
|
||||
if not isinstance(series_items, list) or not series_items:
|
||||
return empty
|
||||
|
||||
series_data = series_items[0] if isinstance(series_items[0], dict) else {}
|
||||
series_name = (
|
||||
str(series_data.get("name") or "").strip() if isinstance(series_data, dict) else ""
|
||||
)
|
||||
allow_split_parts = _series_allows_split_parts(series_name)
|
||||
today = datetime.now(UTC).date()
|
||||
|
||||
book_series_rows = (
|
||||
series_data.get("book_series", []) if isinstance(series_data, dict) else []
|
||||
)
|
||||
rows_by_position: dict[float, dict[str, Any]] = {}
|
||||
for row in book_series_rows:
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
book_data = row.get("book", {})
|
||||
if not isinstance(book_data, dict) or not book_data:
|
||||
continue
|
||||
if exclude_compilations and book_data.get("compilation"):
|
||||
continue
|
||||
if not allow_split_parts and _split_part_base_title(str(book_data.get("title") or "")):
|
||||
continue
|
||||
|
||||
position = _normalize_series_position(row.get("position"))
|
||||
if position is None:
|
||||
continue
|
||||
|
||||
release_date = _parse_release_date(book_data.get("release_date"))
|
||||
if exclude_unreleased and (release_date is None or release_date.date() > today):
|
||||
continue
|
||||
|
||||
sort_key = (
|
||||
1 if release_date and release_date.date() <= today else 0,
|
||||
0 if book_data.get("compilation") else 1,
|
||||
coerce_int(book_data.get("users_count"), 0),
|
||||
coerce_int(book_data.get("ratings_count"), 0),
|
||||
coerce_int(book_data.get("editions_count"), 0),
|
||||
-coerce_int(book_data.get("id"), 0),
|
||||
)
|
||||
existing_row = rows_by_position.get(position)
|
||||
if existing_row is None:
|
||||
rows_by_position[position] = {"row": row, "sort_key": sort_key}
|
||||
continue
|
||||
if sort_key > existing_row["sort_key"]:
|
||||
rows_by_position[position] = {"row": row, "sort_key": sort_key}
|
||||
|
||||
ordered_rows = [
|
||||
entry["row"]
|
||||
for _position, entry in sorted(rows_by_position.items(), key=lambda item: item[0])
|
||||
]
|
||||
return {"rows": ordered_rows, "series_name": series_name, "total": len(ordered_rows)}
|
||||
|
||||
def _fetch_series_books_by_id(
|
||||
self,
|
||||
series_id: int,
|
||||
page: int,
|
||||
limit: int,
|
||||
*,
|
||||
exclude_compilations: bool,
|
||||
exclude_unreleased: bool,
|
||||
) -> SearchResult:
|
||||
"""Fetch books for a Hardcover series in canonical series order."""
|
||||
cached = self._fetch_series_ordered_rows(
|
||||
series_id,
|
||||
exclude_compilations=exclude_compilations,
|
||||
exclude_unreleased=exclude_unreleased,
|
||||
)
|
||||
ordered_rows = cached["rows"]
|
||||
series_name = cached["series_name"]
|
||||
total_found = cached["total"]
|
||||
|
||||
offset = (page - 1) * limit
|
||||
page_rows = ordered_rows[offset : offset + limit]
|
||||
|
||||
books: list[BookMetadata] = []
|
||||
for row in page_rows:
|
||||
book_data = row.get("book", {})
|
||||
if not isinstance(book_data, dict) or not book_data:
|
||||
continue
|
||||
try:
|
||||
parsed_book = self._parse_book(book_data)
|
||||
if not parsed_book:
|
||||
continue
|
||||
parsed_book.series_id = str(series_id)
|
||||
if series_name:
|
||||
parsed_book.series_name = series_name
|
||||
parsed_book.series_position = row.get("position")
|
||||
parsed_book.series_count = total_found
|
||||
books.append(parsed_book)
|
||||
except (AttributeError, IndexError, KeyError, TypeError, ValueError) as exc:
|
||||
logger.debug(
|
||||
"Failed to parse Hardcover series book for series_id=%s: %s", series_id, exc
|
||||
)
|
||||
|
||||
has_more = offset + len(page_rows) < total_found
|
||||
return SearchResult(books=books, page=page, total_found=total_found, has_more=has_more)
|
||||
|
||||
def _fetch_author_books_by_id(
|
||||
self,
|
||||
author_id: int,
|
||||
page: int,
|
||||
limit: int,
|
||||
*,
|
||||
exclude_compilations: bool,
|
||||
exclude_unreleased: bool,
|
||||
) -> SearchResult:
|
||||
"""Fetch books for a selected Hardcover author."""
|
||||
if not self.api_key:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
offset = (page - 1) * limit
|
||||
result = self._execute_query(
|
||||
AUTHOR_BOOKS_BY_ID_QUERY,
|
||||
{"authorId": author_id, "limit": limit, "offset": offset},
|
||||
)
|
||||
if not result:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
author_items = result.get("authors", [])
|
||||
if not isinstance(author_items, list) or not author_items:
|
||||
return SearchResult(books=[], page=page, total_found=0, has_more=False)
|
||||
|
||||
author_data = author_items[0] if isinstance(author_items[0], dict) else {}
|
||||
contributions = (
|
||||
author_data.get("contributions", []) if isinstance(author_data, dict) else []
|
||||
)
|
||||
aggregate = (
|
||||
author_data.get("contributions_aggregate", {}) if isinstance(author_data, dict) else {}
|
||||
)
|
||||
total_found = coerce_int(
|
||||
aggregate.get("aggregate", {}).get("count") if isinstance(aggregate, dict) else 0,
|
||||
0,
|
||||
)
|
||||
today = datetime.now(UTC).date()
|
||||
|
||||
books: list[BookMetadata] = []
|
||||
for row in contributions:
|
||||
if not isinstance(row, dict):
|
||||
continue
|
||||
contribution = str(row.get("contribution") or "").strip()
|
||||
if contribution and "author" not in contribution.casefold():
|
||||
continue
|
||||
book_data = row.get("book", {})
|
||||
if not isinstance(book_data, dict) or not book_data:
|
||||
continue
|
||||
if exclude_compilations and book_data.get("compilation"):
|
||||
continue
|
||||
release_date = _parse_release_date(book_data.get("release_date"))
|
||||
if exclude_unreleased and (release_date is None or release_date.date() > today):
|
||||
continue
|
||||
try:
|
||||
parsed_book = self._parse_book(book_data)
|
||||
books.append(parsed_book)
|
||||
except (AttributeError, IndexError, KeyError, TypeError, ValueError) as exc:
|
||||
logger.debug(
|
||||
"Failed to parse Hardcover author book for author_id=%s: %s",
|
||||
author_id,
|
||||
exc,
|
||||
)
|
||||
|
||||
has_more = offset + len(contributions) < total_found
|
||||
return SearchResult(books=books, page=page, total_found=total_found, has_more=has_more)
|
||||
|
||||
def search(self, options: MetadataSearchOptions) -> list[BookMetadata]:
|
||||
"""Search for books using Hardcover's search API."""
|
||||
return self.search_paginated(options).books
|
||||
|
||||
def search_paginated(self, options: MetadataSearchOptions) -> SearchResult:
|
||||
"""Search for books with pagination info."""
|
||||
if not self.api_key:
|
||||
logger.warning("Hardcover API key not configured")
|
||||
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
|
||||
|
||||
# Allow pasting a Hardcover list URL directly in the search input
|
||||
list_url_parts = self._detect_list_url(options.query)
|
||||
if list_url_parts:
|
||||
owner_username, list_slug = list_url_parts
|
||||
return self._fetch_list_books(list_slug, owner_username, options.page, options.limit)
|
||||
|
||||
# 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 = self._parse_prefixed_int(list_value_from_field, "status")
|
||||
return self._fetch_current_user_books_by_status(
|
||||
status_id, options.page, options.limit
|
||||
)
|
||||
except ValueError:
|
||||
logger.debug("Invalid Hardcover status field value: %s", list_value_from_field)
|
||||
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
|
||||
if list_value_from_field.startswith(HARDCOVER_LIST_ID_PREFIX):
|
||||
try:
|
||||
list_id = self._parse_prefixed_int(list_value_from_field, "list")
|
||||
return self._fetch_list_books_by_id(list_id, options.page, options.limit)
|
||||
except ValueError:
|
||||
logger.debug("Invalid hardcover_list field value: %s", list_value_from_field)
|
||||
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
|
||||
return self._fetch_list_books(list_value_from_field, None, options.page, options.limit)
|
||||
|
||||
series_value_from_field = str(options.fields.get("series", "")).strip()
|
||||
if series_value_from_field:
|
||||
resolved_series = self._resolve_series_search_value(series_value_from_field)
|
||||
if not resolved_series:
|
||||
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
|
||||
exclude_compilations = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_COMPILATIONS", False),
|
||||
default=False,
|
||||
)
|
||||
exclude_unreleased = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_UNRELEASED", False),
|
||||
default=False,
|
||||
)
|
||||
return self._fetch_series_books_by_id(
|
||||
int(resolved_series["id"]),
|
||||
options.page,
|
||||
options.limit,
|
||||
exclude_compilations=exclude_compilations,
|
||||
exclude_unreleased=exclude_unreleased,
|
||||
)
|
||||
|
||||
author_value_from_field = str(options.fields.get("author", "")).strip()
|
||||
if author_value_from_field.startswith(HARDCOVER_LIST_ID_PREFIX):
|
||||
try:
|
||||
author_id = self._parse_prefixed_int(author_value_from_field, "author id")
|
||||
except ValueError:
|
||||
logger.debug("Invalid Hardcover author id field value: %s", author_value_from_field)
|
||||
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
|
||||
exclude_compilations = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_COMPILATIONS", False),
|
||||
default=False,
|
||||
)
|
||||
exclude_unreleased = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_UNRELEASED", False),
|
||||
default=False,
|
||||
)
|
||||
return self._fetch_author_books_by_id(
|
||||
author_id,
|
||||
options.page,
|
||||
options.limit,
|
||||
exclude_compilations=exclude_compilations,
|
||||
exclude_unreleased=exclude_unreleased,
|
||||
)
|
||||
|
||||
# Handle ISBN search separately
|
||||
if options.search_type == SearchType.ISBN:
|
||||
result = self.search_by_isbn(options.query)
|
||||
books = [result] if result else []
|
||||
return SearchResult(books=books, page=1, total_found=len(books), has_more=False)
|
||||
|
||||
# Build cache key from options (include fields and settings for cache differentiation)
|
||||
fields_key = ":".join(f"{k}={v}" for k, v in sorted(options.fields.items()))
|
||||
exclude_compilations = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_COMPILATIONS", False),
|
||||
default=False,
|
||||
)
|
||||
exclude_unreleased = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_UNRELEASED", False),
|
||||
default=False,
|
||||
)
|
||||
cache_key = f"{options.query}:{options.search_type.value}:{options.sort.value}:{options.limit}:{options.page}:{fields_key}:excl_comp={exclude_compilations}:excl_unrel={exclude_unreleased}"
|
||||
return self._search_cached(cache_key, options)
|
||||
|
||||
@cacheable(ttl_key="METADATA_CACHE_SEARCH_TTL", ttl_default=300, key_prefix="hardcover:search")
|
||||
def _search_cached(self, cache_key: str, options: MetadataSearchOptions) -> SearchResult:
|
||||
"""Return cached Hardcover search results."""
|
||||
# Determine query and fields based on custom search fields
|
||||
# Note: Hardcover API requires 'weights' when using 'fields' parameter
|
||||
author_value = options.fields.get("author", "").strip()
|
||||
title_value = options.fields.get("title", "").strip()
|
||||
|
||||
# Build query and field configuration based on which fields are provided
|
||||
query, search_fields, search_weights = self._build_search_params(
|
||||
options.query, author_value, title_value, ""
|
||||
)
|
||||
|
||||
graphql_query = SEARCH_BOOKS_WITH_FIELDS_QUERY if search_fields else SEARCH_BOOKS_QUERY
|
||||
|
||||
# Map abstract sort order to Hardcover's sort parameter
|
||||
sort_param = SORT_MAPPING.get(options.sort, SORT_MAPPING[SortOrder.RELEVANCE])
|
||||
|
||||
variables = {
|
||||
"query": query,
|
||||
"limit": options.limit,
|
||||
"page": options.page,
|
||||
"sort": sort_param,
|
||||
}
|
||||
|
||||
if search_fields:
|
||||
variables["fields"] = search_fields
|
||||
variables["weights"] = search_weights
|
||||
|
||||
try:
|
||||
result = self._execute_query(graphql_query, variables)
|
||||
if not result:
|
||||
logger.debug("Hardcover search: No result from API")
|
||||
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
|
||||
|
||||
# Extract hits from Typesense response
|
||||
hits, found_count = _extract_typesense_hits(result)
|
||||
|
||||
# Parse hits, filtering compilations and unreleased books if enabled
|
||||
exclude_compilations = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_COMPILATIONS", False),
|
||||
default=False,
|
||||
)
|
||||
exclude_unreleased = coerce_bool(
|
||||
app_config.get("HARDCOVER_EXCLUDE_UNRELEASED", False),
|
||||
default=False,
|
||||
)
|
||||
current_year = datetime.now(UTC).year
|
||||
books = []
|
||||
for hit in hits:
|
||||
item = _unwrap_hit_document(hit)
|
||||
if item is None:
|
||||
continue
|
||||
if exclude_compilations and item.get("compilation"):
|
||||
continue
|
||||
if exclude_unreleased:
|
||||
release_year = item.get("release_year")
|
||||
if release_year is not None and release_year > current_year:
|
||||
continue
|
||||
book = self._parse_search_result(item)
|
||||
if book:
|
||||
books.append(book)
|
||||
|
||||
logger.info(
|
||||
"Hardcover search '%s' (fields=%s) returned %s results",
|
||||
query,
|
||||
search_fields,
|
||||
len(books),
|
||||
)
|
||||
|
||||
# Calculate if there are more results
|
||||
results_so_far = (options.page - 1) * HARDCOVER_PAGE_SIZE + len(hits)
|
||||
has_more = results_so_far < found_count
|
||||
|
||||
return SearchResult(
|
||||
books=books, page=options.page, total_found=found_count, has_more=has_more
|
||||
)
|
||||
|
||||
except AttributeError, KeyError, TypeError, ValueError:
|
||||
logger.exception("Hardcover search error")
|
||||
return SearchResult(books=[], page=options.page, total_found=0, has_more=False)
|
||||
|
||||
@cacheable(ttl_key="METADATA_CACHE_BOOK_TTL", ttl_default=600, key_prefix="hardcover:book")
|
||||
def get_book(self, book_id: str) -> BookMetadata | None:
|
||||
"""Get book details by Hardcover ID."""
|
||||
if not self.api_key:
|
||||
logger.warning("Hardcover API key not configured")
|
||||
return None
|
||||
|
||||
try:
|
||||
book_id_int = int(book_id)
|
||||
result = self._execute_query(GET_BOOK_QUERY, {"id": book_id_int})
|
||||
if not result:
|
||||
return None
|
||||
|
||||
books = result.get("books", [])
|
||||
if not books:
|
||||
return None
|
||||
|
||||
return self._parse_book(books[0])
|
||||
|
||||
except ValueError:
|
||||
logger.exception("Invalid book ID: %s", book_id)
|
||||
return None
|
||||
except AttributeError, KeyError, TypeError:
|
||||
logger.exception("Hardcover get_book error")
|
||||
return None
|
||||
|
||||
@cacheable(ttl_key="METADATA_CACHE_BOOK_TTL", ttl_default=600, key_prefix="hardcover:isbn")
|
||||
def search_by_isbn(self, isbn: str) -> BookMetadata | None:
|
||||
"""Search for a book by ISBN-10 or ISBN-13."""
|
||||
if not self.api_key:
|
||||
logger.warning("Hardcover API key not configured")
|
||||
return None
|
||||
|
||||
# Clean ISBN (remove hyphens)
|
||||
clean_isbn = isbn.replace("-", "").strip()
|
||||
|
||||
try:
|
||||
result = self._execute_query(SEARCH_BY_ISBN_QUERY, {"isbn": clean_isbn})
|
||||
if not result:
|
||||
return None
|
||||
|
||||
editions = result.get("editions", [])
|
||||
if not editions:
|
||||
logger.debug("No Hardcover book found for ISBN: %s", isbn)
|
||||
return None
|
||||
|
||||
edition = editions[0]
|
||||
book_data = edition.get("book", {})
|
||||
if not book_data:
|
||||
return None
|
||||
|
||||
# Add ISBN data from edition to book data
|
||||
book_data["isbn_10"] = edition.get("isbn_10")
|
||||
book_data["isbn_13"] = edition.get("isbn_13")
|
||||
|
||||
return self._parse_book(book_data)
|
||||
|
||||
except AttributeError, IndexError, KeyError, TypeError, ValueError:
|
||||
logger.exception("Hardcover ISBN search error")
|
||||
return None
|
||||
@@ -1,154 +0,0 @@
|
||||
"""Settings registration for the Hardcover metadata provider."""
|
||||
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
|
||||
from shelfmark.core.config import config as app_config
|
||||
from shelfmark.core.logger import setup_logger
|
||||
from shelfmark.core.settings_registry import (
|
||||
ActionButton,
|
||||
CheckboxField,
|
||||
HeadingField,
|
||||
PasswordField,
|
||||
SelectField,
|
||||
SettingsField,
|
||||
register_settings,
|
||||
)
|
||||
|
||||
from .auth import _get_connected_username, _save_connected_user
|
||||
from .constants import HARDCOVER_API_KEY_MIN_LENGTH
|
||||
from .parsing import _normalize_hardcover_api_key
|
||||
from .provider import HardcoverProvider
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
def _test_hardcover_connection(current_values: dict[str, Any] | None = None) -> dict[str, Any]:
|
||||
"""Test the Hardcover API connection using current form values."""
|
||||
current_values = current_values or {}
|
||||
|
||||
# Use current form values first, fall back to saved config
|
||||
raw_key = current_values.get("HARDCOVER_API_KEY") or app_config.get("HARDCOVER_API_KEY", "")
|
||||
api_key = _normalize_hardcover_api_key(raw_key)
|
||||
|
||||
key_len = len(api_key) if api_key else 0
|
||||
logger.debug("Hardcover test: key length=%s", key_len)
|
||||
|
||||
if not api_key:
|
||||
# Clear any stored connection metadata since there's no key
|
||||
_save_connected_user(None, None)
|
||||
return {"success": False, "message": "API key is required"}
|
||||
|
||||
if key_len < HARDCOVER_API_KEY_MIN_LENGTH:
|
||||
return {
|
||||
"success": False,
|
||||
"message": (
|
||||
f"API key seems too short ({key_len} chars). "
|
||||
f"Expected {HARDCOVER_API_KEY_MIN_LENGTH}+ chars."
|
||||
),
|
||||
}
|
||||
|
||||
connection_result = {"success": False, "message": "API request failed - check your API key"}
|
||||
try:
|
||||
provider = HardcoverProvider(api_key=api_key)
|
||||
# Use the 'me' query to test connection (recommended by API docs)
|
||||
result = provider._execute_query("query { me { id, username } }", {})
|
||||
if result is not None:
|
||||
# Handle both single object and array response formats
|
||||
me_data = result.get("me", {})
|
||||
if isinstance(me_data, list) and me_data:
|
||||
me_data = me_data[0]
|
||||
user_id = (
|
||||
str(me_data.get("id"))
|
||||
if isinstance(me_data, dict) and me_data.get("id") is not None
|
||||
else None
|
||||
)
|
||||
username = (
|
||||
me_data.get("username", "Unknown") if isinstance(me_data, dict) else "Unknown"
|
||||
)
|
||||
|
||||
# Save connected user metadata for persistent display + per-user list caching
|
||||
_save_connected_user(user_id, username)
|
||||
connection_result = {"success": True, "message": f"Connected as: {username}"}
|
||||
else:
|
||||
_save_connected_user(None, None)
|
||||
except (AttributeError, KeyError, requests.RequestException, TypeError, ValueError) as e:
|
||||
logger.exception("Hardcover connection test failed")
|
||||
_save_connected_user(None, None)
|
||||
return {"success": False, "message": f"Connection failed: {e!s}"}
|
||||
|
||||
return connection_result
|
||||
|
||||
|
||||
_HARDCOVER_SORT_OPTIONS = [
|
||||
{"value": "relevance", "label": "Most relevant"},
|
||||
{"value": "popularity", "label": "Most popular"},
|
||||
{"value": "rating", "label": "Highest rated"},
|
||||
{"value": "newest", "label": "Newest"},
|
||||
{"value": "oldest", "label": "Oldest"},
|
||||
]
|
||||
|
||||
|
||||
@register_settings("hardcover", "Hardcover", icon="book", order=51, group="metadata_providers")
|
||||
def hardcover_settings() -> list[SettingsField]:
|
||||
"""Hardcover metadata provider settings."""
|
||||
# Check for connected username to show status
|
||||
connected_user = _get_connected_username()
|
||||
test_button_description = (
|
||||
f"Connected as: {connected_user}" if connected_user else "Verify your API key works"
|
||||
)
|
||||
|
||||
return [
|
||||
HeadingField(
|
||||
key="hardcover_heading",
|
||||
title="Hardcover",
|
||||
description="A modern book tracking and discovery platform with a comprehensive API.",
|
||||
link_url="https://hardcover.app",
|
||||
link_text="hardcover.app",
|
||||
),
|
||||
CheckboxField(
|
||||
key="HARDCOVER_ENABLED",
|
||||
label="Enable Hardcover",
|
||||
description="Enable Hardcover as a metadata provider for book searches",
|
||||
default=False,
|
||||
),
|
||||
PasswordField(
|
||||
key="HARDCOVER_API_KEY",
|
||||
label="API Key",
|
||||
description="Get your API key from hardcover.app/account/api",
|
||||
required=True,
|
||||
),
|
||||
ActionButton(
|
||||
key="test_connection",
|
||||
label="Test Connection",
|
||||
description=test_button_description,
|
||||
style="primary",
|
||||
callback=_test_hardcover_connection,
|
||||
),
|
||||
SelectField(
|
||||
key="HARDCOVER_DEFAULT_SORT",
|
||||
label="Default Sort Order",
|
||||
description="Default sort order for Hardcover search results.",
|
||||
options=_HARDCOVER_SORT_OPTIONS,
|
||||
default="relevance",
|
||||
),
|
||||
CheckboxField(
|
||||
key="HARDCOVER_EXCLUDE_COMPILATIONS",
|
||||
label="Exclude Compilations",
|
||||
description="Filter out compilations, anthologies, and omnibus editions from search results",
|
||||
default=False,
|
||||
),
|
||||
CheckboxField(
|
||||
key="HARDCOVER_EXCLUDE_UNRELEASED",
|
||||
label="Exclude Unreleased Books",
|
||||
description="Filter out books with a release year in the future",
|
||||
default=False,
|
||||
),
|
||||
CheckboxField(
|
||||
key="HARDCOVER_AUTO_REMOVE_ON_DOWNLOAD",
|
||||
label="Auto-Remove from List on Download",
|
||||
description="Automatically remove a book from the active Hardcover list when you download it",
|
||||
default=True,
|
||||
),
|
||||
]
|
||||
@@ -1,438 +0,0 @@
|
||||
"""Hardcover list/status target read and mutation workflows."""
|
||||
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
from shelfmark.core.cache import cache_key
|
||||
from shelfmark.core.logger import setup_logger
|
||||
from shelfmark.core.request_helpers import coerce_int
|
||||
|
||||
from .constants import (
|
||||
HARDCOVER_LIST_ID_PREFIX,
|
||||
HARDCOVER_STATUS_PREFIX,
|
||||
HARDCOVER_WRITABLE_TARGET_GROUPS,
|
||||
)
|
||||
from .models import HardcoverBookTargetState, HardcoverTargetPayloadError
|
||||
from .queries import (
|
||||
BOOK_TARGET_MEMBERSHIP_BATCH_QUERY,
|
||||
BOOK_TARGET_MEMBERSHIP_QUERY,
|
||||
DELETE_LIST_BOOK_MUTATION,
|
||||
DELETE_USER_BOOK_MUTATION,
|
||||
INSERT_LIST_BOOK_MUTATION,
|
||||
INSERT_USER_BOOK_MUTATION,
|
||||
UPDATE_USER_BOOK_MUTATION,
|
||||
)
|
||||
|
||||
logger = setup_logger(__name__)
|
||||
|
||||
|
||||
def _metadata_cache() -> Any:
|
||||
from shelfmark.metadata_providers import hardcover
|
||||
|
||||
return hardcover.get_metadata_cache()
|
||||
|
||||
|
||||
class HardcoverTargetsMixin:
|
||||
if TYPE_CHECKING:
|
||||
api_key: str
|
||||
|
||||
def _execute_query(
|
||||
self,
|
||||
query: str,
|
||||
variables: dict[str, Any],
|
||||
*,
|
||||
raise_on_error: bool = False,
|
||||
) -> dict[str, Any] | None: ...
|
||||
|
||||
def _resolve_current_user_id(self) -> str | None: ...
|
||||
|
||||
def get_user_lists(self) -> list[dict[str, str]]: ...
|
||||
|
||||
def get_book_targets(self, book_id: str) -> list[dict[str, Any]]:
|
||||
"""Get writable Hardcover list/status targets for a specific book."""
|
||||
if not self.api_key:
|
||||
return []
|
||||
|
||||
book_id_int = coerce_int(book_id, 0)
|
||||
if book_id_int < 1:
|
||||
msg = "book_id must be a valid Hardcover book id"
|
||||
raise ValueError(msg)
|
||||
|
||||
state = self._fetch_book_target_state(book_id_int)
|
||||
options: list[dict[str, Any]] = [
|
||||
dict(option)
|
||||
for option in self.get_user_lists()
|
||||
if option.get("group") in HARDCOVER_WRITABLE_TARGET_GROUPS
|
||||
]
|
||||
|
||||
for option in options:
|
||||
value = str(option.get("value") or "").strip()
|
||||
option["checked"] = self._is_target_checked(value, state)
|
||||
option["writable"] = True
|
||||
|
||||
return options
|
||||
|
||||
def set_book_target_state(
|
||||
self,
|
||||
book_id: str,
|
||||
target: str,
|
||||
*,
|
||||
selected: bool,
|
||||
) -> dict[str, Any]:
|
||||
"""Set whether a Hardcover book belongs to a status shelf or user list."""
|
||||
if not self.api_key:
|
||||
msg = "Hardcover is not configured"
|
||||
raise ValueError(msg)
|
||||
|
||||
book_id_int = coerce_int(book_id, 0)
|
||||
if book_id_int < 1:
|
||||
msg = "book_id must be a valid Hardcover book id"
|
||||
raise ValueError(msg)
|
||||
|
||||
selected_target = str(target or "").strip()
|
||||
if not selected_target:
|
||||
msg = "target is required"
|
||||
raise ValueError(msg)
|
||||
|
||||
if selected_target not in self._get_writable_targets():
|
||||
msg = "Unsupported Hardcover target"
|
||||
raise ValueError(msg)
|
||||
|
||||
state = self._fetch_book_target_state(book_id_int)
|
||||
status_ids_to_invalidate: set[int] = set()
|
||||
list_ids_to_invalidate: set[int] = set()
|
||||
deselected_target: str | None = None
|
||||
|
||||
if selected_target.startswith(HARDCOVER_STATUS_PREFIX):
|
||||
status_id = self._parse_prefixed_int(selected_target, "status target")
|
||||
previous_status_id = state.status_id
|
||||
changed = self._set_status_target_state(
|
||||
book_id_int,
|
||||
status_id,
|
||||
selected=selected,
|
||||
state=state,
|
||||
)
|
||||
if changed:
|
||||
if previous_status_id is not None:
|
||||
status_ids_to_invalidate.add(previous_status_id)
|
||||
if selected and previous_status_id != status_id:
|
||||
deselected_target = f"{HARDCOVER_STATUS_PREFIX}{previous_status_id}"
|
||||
status_ids_to_invalidate.add(status_id)
|
||||
elif selected_target.startswith(HARDCOVER_LIST_ID_PREFIX):
|
||||
list_id = self._parse_prefixed_int(selected_target, "list target")
|
||||
changed = self._set_list_target_state(
|
||||
book_id_int,
|
||||
list_id,
|
||||
selected=selected,
|
||||
state=state,
|
||||
)
|
||||
if changed:
|
||||
list_ids_to_invalidate.add(list_id)
|
||||
else:
|
||||
msg = "Unsupported Hardcover target"
|
||||
raise ValueError(msg)
|
||||
|
||||
if changed:
|
||||
self._invalidate_book_target_caches(
|
||||
connected_user_id=self._resolve_current_user_id(),
|
||||
status_ids=status_ids_to_invalidate,
|
||||
list_ids=list_ids_to_invalidate,
|
||||
)
|
||||
|
||||
result_data: dict[str, Any] = {"changed": changed}
|
||||
if deselected_target:
|
||||
result_data["deselected_target"] = deselected_target
|
||||
return result_data
|
||||
|
||||
@staticmethod
|
||||
def _unwrap_me_data(result: dict | None) -> dict:
|
||||
"""Extract and validate the ``me`` payload from a GraphQL result."""
|
||||
if not isinstance(result, dict):
|
||||
msg = "Hardcover could not load book targets"
|
||||
raise HardcoverTargetPayloadError(msg)
|
||||
|
||||
me_data = result.get("me", {})
|
||||
if isinstance(me_data, list) and me_data:
|
||||
me_data = me_data[0]
|
||||
if not isinstance(me_data, dict):
|
||||
msg = "Hardcover returned an invalid target payload"
|
||||
raise HardcoverTargetPayloadError(msg)
|
||||
return me_data
|
||||
|
||||
def _fetch_book_target_state(self, book_id: int) -> HardcoverBookTargetState:
|
||||
"""Load current Hardcover membership state for a specific book."""
|
||||
result = self._execute_query(
|
||||
BOOK_TARGET_MEMBERSHIP_QUERY,
|
||||
{"bookId": book_id},
|
||||
raise_on_error=True,
|
||||
)
|
||||
me_data = self._unwrap_me_data(result)
|
||||
|
||||
user_book_id: int | None = None
|
||||
status_id: int | None = None
|
||||
user_books = me_data.get("user_books", [])
|
||||
if isinstance(user_books, list) and user_books:
|
||||
latest_user_book = user_books[0] if isinstance(user_books[0], dict) else {}
|
||||
user_book_id = coerce_int(latest_user_book.get("id"), 0) or None
|
||||
status_id = coerce_int(latest_user_book.get("status_id"), 0) or None
|
||||
|
||||
list_book_ids: dict[int, int] = {}
|
||||
for user_list in me_data.get("lists", []):
|
||||
if not isinstance(user_list, dict):
|
||||
continue
|
||||
list_id = coerce_int(user_list.get("id"), 0)
|
||||
if list_id < 1:
|
||||
continue
|
||||
|
||||
list_books = user_list.get("list_books", [])
|
||||
if not isinstance(list_books, list) or not list_books:
|
||||
continue
|
||||
|
||||
list_book = list_books[0] if isinstance(list_books[0], dict) else {}
|
||||
list_book_id = coerce_int(list_book.get("id"), 0)
|
||||
if list_book_id > 0:
|
||||
list_book_ids[list_id] = list_book_id
|
||||
|
||||
return HardcoverBookTargetState(
|
||||
user_book_id=user_book_id,
|
||||
status_id=status_id,
|
||||
list_book_ids=list_book_ids,
|
||||
)
|
||||
|
||||
def _fetch_book_target_states_batch(
|
||||
self,
|
||||
book_ids: list[int],
|
||||
) -> dict[int, HardcoverBookTargetState]:
|
||||
"""Load Hardcover membership state for multiple books in one query."""
|
||||
result = self._execute_query(
|
||||
BOOK_TARGET_MEMBERSHIP_BATCH_QUERY,
|
||||
{"bookIds": book_ids},
|
||||
raise_on_error=True,
|
||||
)
|
||||
me_data = self._unwrap_me_data(result)
|
||||
|
||||
# Group user_books by book_id (keep only the latest per book)
|
||||
user_book_by_book: dict[int, dict] = {}
|
||||
for ub in me_data.get("user_books", []):
|
||||
if not isinstance(ub, dict):
|
||||
continue
|
||||
bid = coerce_int(ub.get("book_id"), 0)
|
||||
if bid > 0 and bid not in user_book_by_book:
|
||||
user_book_by_book[bid] = ub
|
||||
|
||||
# Group list_book memberships by book_id
|
||||
list_book_ids_by_book: dict[int, dict[int, int]] = {}
|
||||
for user_list in me_data.get("lists", []):
|
||||
if not isinstance(user_list, dict):
|
||||
continue
|
||||
list_id = coerce_int(user_list.get("id"), 0)
|
||||
if list_id < 1:
|
||||
continue
|
||||
for lb in user_list.get("list_books", []):
|
||||
if not isinstance(lb, dict):
|
||||
continue
|
||||
bid = coerce_int(lb.get("book_id"), 0)
|
||||
lb_id = coerce_int(lb.get("id"), 0)
|
||||
if bid > 0 and lb_id > 0:
|
||||
list_book_ids_by_book.setdefault(bid, {})[list_id] = lb_id
|
||||
|
||||
states: dict[int, HardcoverBookTargetState] = {}
|
||||
for bid in book_ids:
|
||||
ub = user_book_by_book.get(bid)
|
||||
states[bid] = HardcoverBookTargetState(
|
||||
user_book_id=coerce_int(ub.get("id"), 0) or None if ub else None,
|
||||
status_id=coerce_int(ub.get("status_id"), 0) or None if ub else None,
|
||||
list_book_ids=list_book_ids_by_book.get(bid, {}),
|
||||
)
|
||||
return states
|
||||
|
||||
def get_book_targets_batch(self, book_ids: list[str]) -> dict[str, list[dict[str, Any]]]:
|
||||
"""Get writable Hardcover list/status targets for multiple books."""
|
||||
if not self.api_key or not book_ids:
|
||||
return {bid: [] for bid in book_ids}
|
||||
|
||||
int_ids = []
|
||||
id_map: dict[int, str] = {}
|
||||
for bid in book_ids:
|
||||
int_id = coerce_int(bid, 0)
|
||||
if int_id > 0:
|
||||
int_ids.append(int_id)
|
||||
id_map[int_id] = bid
|
||||
|
||||
if not int_ids:
|
||||
return {bid: [] for bid in book_ids}
|
||||
|
||||
states = self._fetch_book_target_states_batch(int_ids)
|
||||
writable_options: list[dict[str, Any]] = [
|
||||
dict(option)
|
||||
for option in self.get_user_lists()
|
||||
if option.get("group") in HARDCOVER_WRITABLE_TARGET_GROUPS
|
||||
]
|
||||
|
||||
results: dict[str, list[dict[str, Any]]] = {}
|
||||
for int_id, str_id in id_map.items():
|
||||
state = states.get(
|
||||
int_id,
|
||||
HardcoverBookTargetState(
|
||||
user_book_id=None,
|
||||
status_id=None,
|
||||
list_book_ids={},
|
||||
),
|
||||
)
|
||||
options = [dict(opt) for opt in writable_options]
|
||||
for option in options:
|
||||
value = str(option.get("value") or "").strip()
|
||||
option["checked"] = self._is_target_checked(value, state)
|
||||
option["writable"] = True
|
||||
results[str_id] = options
|
||||
|
||||
# Fill in any book_ids that didn't parse as valid ints
|
||||
for bid in book_ids:
|
||||
if bid not in results:
|
||||
results[bid] = []
|
||||
|
||||
return results
|
||||
|
||||
def _get_writable_targets(self) -> set[str]:
|
||||
"""Return the set of writable Hardcover targets for the current user."""
|
||||
writable_targets: set[str] = set()
|
||||
for option in self.get_user_lists():
|
||||
value = str(option.get("value") or "").strip()
|
||||
if (
|
||||
option.get("group") in HARDCOVER_WRITABLE_TARGET_GROUPS
|
||||
and value
|
||||
and value.startswith((HARDCOVER_STATUS_PREFIX, HARDCOVER_LIST_ID_PREFIX))
|
||||
):
|
||||
writable_targets.add(value)
|
||||
return writable_targets
|
||||
|
||||
def _is_target_checked(self, target: str, state: HardcoverBookTargetState) -> bool:
|
||||
"""Return whether a target is currently selected for the book."""
|
||||
if target.startswith(HARDCOVER_STATUS_PREFIX):
|
||||
return state.status_id == self._parse_prefixed_int(target)
|
||||
if target.startswith(HARDCOVER_LIST_ID_PREFIX):
|
||||
return self._parse_prefixed_int(target) in state.list_book_ids
|
||||
return False
|
||||
|
||||
def _set_status_target_state(
|
||||
self,
|
||||
book_id: int,
|
||||
status_id: int,
|
||||
*,
|
||||
selected: bool,
|
||||
state: HardcoverBookTargetState,
|
||||
) -> bool:
|
||||
"""Set whether the book belongs to a Hardcover status shelf."""
|
||||
if selected:
|
||||
if state.user_book_id is None:
|
||||
result = self._execute_query(
|
||||
INSERT_USER_BOOK_MUTATION,
|
||||
{"bookId": book_id, "statusId": status_id},
|
||||
raise_on_error=True,
|
||||
)
|
||||
self._check_mutation_result(result, "insert_user_book")
|
||||
return True
|
||||
|
||||
if state.status_id == status_id:
|
||||
return False
|
||||
|
||||
result = self._execute_query(
|
||||
UPDATE_USER_BOOK_MUTATION,
|
||||
{"userBookId": state.user_book_id, "statusId": status_id},
|
||||
raise_on_error=True,
|
||||
)
|
||||
self._check_mutation_result(result, "update_user_book")
|
||||
return True
|
||||
|
||||
if state.user_book_id is None or state.status_id != status_id:
|
||||
return False
|
||||
|
||||
result = self._execute_query(
|
||||
DELETE_USER_BOOK_MUTATION,
|
||||
{"userBookId": state.user_book_id},
|
||||
raise_on_error=True,
|
||||
)
|
||||
self._check_mutation_result(result, "delete_user_book", check_error=False)
|
||||
return True
|
||||
|
||||
def _set_list_target_state(
|
||||
self,
|
||||
book_id: int,
|
||||
list_id: int,
|
||||
*,
|
||||
selected: bool,
|
||||
state: HardcoverBookTargetState,
|
||||
) -> bool:
|
||||
"""Set whether the book belongs to a Hardcover list."""
|
||||
list_book_id = state.list_book_ids.get(list_id)
|
||||
|
||||
if selected:
|
||||
if list_book_id is not None:
|
||||
return False
|
||||
|
||||
result = self._execute_query(
|
||||
INSERT_LIST_BOOK_MUTATION,
|
||||
{"bookId": book_id, "listId": list_id},
|
||||
raise_on_error=True,
|
||||
)
|
||||
self._check_mutation_result(result, "insert_list_book")
|
||||
return True
|
||||
|
||||
if list_book_id is None:
|
||||
return False
|
||||
|
||||
result = self._execute_query(
|
||||
DELETE_LIST_BOOK_MUTATION,
|
||||
{"listBookId": list_book_id},
|
||||
raise_on_error=True,
|
||||
)
|
||||
self._check_mutation_result(result, "delete_list_book", check_error=False)
|
||||
return True
|
||||
|
||||
def _invalidate_book_target_caches(
|
||||
self,
|
||||
*,
|
||||
connected_user_id: str | None,
|
||||
status_ids: set[int],
|
||||
list_ids: set[int],
|
||||
) -> None:
|
||||
"""Invalidate caches affected by a target membership change."""
|
||||
metadata_cache = _metadata_cache()
|
||||
|
||||
if connected_user_id:
|
||||
metadata_cache.invalidate(cache_key("hardcover:user_lists", connected_user_id))
|
||||
for status_id in status_ids:
|
||||
metadata_cache.invalidate_prefix(
|
||||
cache_key("hardcover:user_books:status", connected_user_id, status_id)
|
||||
)
|
||||
|
||||
for list_id in list_ids:
|
||||
metadata_cache.invalidate_prefix(cache_key("hardcover:list:id", list_id))
|
||||
|
||||
@staticmethod
|
||||
def _parse_prefixed_int(value: str, label: str = "target") -> int:
|
||||
"""Parse an integer from a colon-prefixed value like 'status:1' or 'id:42'."""
|
||||
try:
|
||||
return int(value.split(":", 1)[1])
|
||||
except (IndexError, ValueError) as exc:
|
||||
msg = f"Invalid Hardcover {label}"
|
||||
raise ValueError(msg) from exc
|
||||
|
||||
@staticmethod
|
||||
def _check_mutation_result(result: Any, key: str, *, check_error: bool = True) -> None:
|
||||
"""Raise if a Hardcover mutation failed.
|
||||
|
||||
When *check_error* is True (the default) the ``error`` field inside
|
||||
the payload is inspected and surfaced as a ``ValueError``. Pass
|
||||
``check_error=False`` for delete mutations that don't return an
|
||||
error field.
|
||||
"""
|
||||
payload = result.get(key, {}) if isinstance(result, dict) else {}
|
||||
if isinstance(payload, dict):
|
||||
if check_error:
|
||||
error_text = str(payload.get("error") or "").strip()
|
||||
if error_text:
|
||||
raise ValueError(error_text)
|
||||
if payload.get("id") is not None:
|
||||
return
|
||||
msg = "Hardcover could not complete this action"
|
||||
raise RuntimeError(msg)
|
||||
@@ -7,7 +7,6 @@ import { useMountEffect } from '../hooks/useMountEffect';
|
||||
import type { Book, ButtonStateInfo } from '../types';
|
||||
import { isMetadataBook } from '../types';
|
||||
import { bookSupportsTargets } from '../utils/bookTargetLoader';
|
||||
import { getSizedCoverUrl } from '../utils/covers';
|
||||
import { isUserCancelledError } from '../utils/errors';
|
||||
import { BookTargetDropdown } from './BookTargetDropdown';
|
||||
|
||||
@@ -137,10 +136,6 @@ export const DetailsModal = ({
|
||||
const artworkMaxWidth = isSquareCover
|
||||
? 'min(45vw, 400px, calc(90vh - 220px))'
|
||||
: 'min(45vw, 520px, calc((90vh - 220px) / 1.6))';
|
||||
const optimizedPreview = getSizedCoverUrl(book.preview, {
|
||||
width: isSquareCover ? 640 : 480,
|
||||
height: isSquareCover ? 640 : 720,
|
||||
});
|
||||
const additionalInfo =
|
||||
book.info && Object.keys(book.info).length > 0
|
||||
? Object.entries(book.info).filter(([key]) => {
|
||||
@@ -206,17 +201,14 @@ export const DetailsModal = ({
|
||||
<div className="min-h-0 flex-1 overflow-y-auto px-5 py-6">
|
||||
<div className="flex flex-col gap-6 lg:min-h-0 lg:flex-row lg:items-stretch lg:gap-8">
|
||||
<div className="flex w-full justify-center lg:w-auto lg:flex-none lg:justify-start lg:self-stretch lg:pr-4">
|
||||
{optimizedPreview ? (
|
||||
{book.preview ? (
|
||||
<div
|
||||
className="flex w-full items-center justify-center lg:h-full lg:max-w-none"
|
||||
style={{ maxHeight: artworkMaxHeight, maxWidth: artworkMaxWidth }}
|
||||
>
|
||||
<img
|
||||
src={optimizedPreview}
|
||||
src={book.preview}
|
||||
alt="Book cover"
|
||||
width={isSquareCover ? 640 : 480}
|
||||
height={isSquareCover ? 640 : 720}
|
||||
decoding="async"
|
||||
className="h-auto max-h-full w-auto max-w-full rounded-xl object-contain shadow-lg"
|
||||
style={{ maxHeight: '100%', maxWidth: '100%' }}
|
||||
/>
|
||||
|
||||
@@ -22,7 +22,6 @@ import type {
|
||||
import { isMetadataBook } from '../types';
|
||||
import { bookSupportsTargets } from '../utils/bookTargetLoader';
|
||||
import { getColorStyleFromHint } from '../utils/colorMaps';
|
||||
import { getSizedCoverUrl } from '../utils/covers';
|
||||
import {
|
||||
LANGUAGE_OPTION_DEFAULT,
|
||||
getLanguageFilterValues,
|
||||
@@ -211,9 +210,8 @@ function StarRating({ rating, maxRating = 5 }: { rating: number; maxRating?: num
|
||||
const ReleaseThumbnail = ({ preview, title }: { preview?: string; title?: string }) => {
|
||||
const [imageLoaded, setImageLoaded] = useState(false);
|
||||
const [imageError, setImageError] = useState(false);
|
||||
const optimizedPreview = getSizedCoverUrl(preview, { width: 32, height: 48 });
|
||||
|
||||
if (!optimizedPreview || imageError) {
|
||||
if (!preview || imageError) {
|
||||
return (
|
||||
<div
|
||||
className="flex h-10 w-7 shrink-0 items-center justify-center rounded-sm bg-zinc-200 text-[7px] font-medium text-zinc-500 sm:h-12 sm:w-8 sm:text-[8px] dark:bg-zinc-700 dark:text-zinc-400"
|
||||
@@ -230,13 +228,10 @@ const ReleaseThumbnail = ({ preview, title }: { preview?: string; title?: string
|
||||
<div className="absolute inset-0 animate-pulse bg-linear-to-r from-gray-200 via-gray-100 to-gray-200 dark:from-gray-700 dark:via-gray-600 dark:to-gray-700" />
|
||||
)}
|
||||
<img
|
||||
src={optimizedPreview}
|
||||
src={preview}
|
||||
alt={title || 'Book cover'}
|
||||
className="h-full w-full object-cover object-top"
|
||||
loading="lazy"
|
||||
decoding="async"
|
||||
width={32}
|
||||
height={48}
|
||||
onLoad={() => setImageLoaded(true)}
|
||||
onError={() => setImageError(true)}
|
||||
style={{ opacity: imageLoaded ? 1 : 0, transition: 'opacity 0.2s ease-in-out' }}
|
||||
@@ -1242,10 +1237,6 @@ const ReleaseModalSession = ({
|
||||
} else if (book.series_name) {
|
||||
coverSizeClassName = 'h-[144px] w-24';
|
||||
}
|
||||
const modalPreview = getSizedCoverUrl(book.preview, {
|
||||
width: book.cover_aspect === 'square' ? 144 : 96,
|
||||
height: 144,
|
||||
});
|
||||
|
||||
let combinedFooterEbookMode = combinedEbookMode;
|
||||
if (combinedPhase === 'ebook') {
|
||||
@@ -1334,14 +1325,13 @@ const ReleaseModalSession = ({
|
||||
{/* Mobile: static thumbnail always visible */}
|
||||
{!isRequestMode && (
|
||||
<div className="shrink-0 sm:hidden">
|
||||
{modalPreview ? (
|
||||
{book.preview ? (
|
||||
<img
|
||||
src={modalPreview}
|
||||
src={book.preview}
|
||||
alt=""
|
||||
width={book.cover_aspect === 'square' ? 68 : 46}
|
||||
height={68}
|
||||
className={`rounded-sm object-cover shadow-md ${book.cover_aspect === 'square' ? 'object-center' : 'object-top'}`}
|
||||
decoding="async"
|
||||
style={{
|
||||
width: book.cover_aspect === 'square' ? 68 : 46,
|
||||
height: 68,
|
||||
@@ -1375,14 +1365,13 @@ const ReleaseModalSession = ({
|
||||
className="transition-opacity duration-300 ease-out"
|
||||
style={{ opacity: showHeaderThumb ? 1 : 0 }}
|
||||
>
|
||||
{modalPreview ? (
|
||||
{book.preview ? (
|
||||
<img
|
||||
src={modalPreview}
|
||||
src={book.preview}
|
||||
alt=""
|
||||
width={book.cover_aspect === 'square' ? 68 : 46}
|
||||
height={68}
|
||||
className={`rounded-sm object-cover shadow-md ${book.cover_aspect === 'square' ? 'object-center' : 'object-top'}`}
|
||||
decoding="async"
|
||||
style={{
|
||||
width: book.cover_aspect === 'square' ? 68 : 46,
|
||||
height: 68,
|
||||
@@ -1445,13 +1434,10 @@ const ReleaseModalSession = ({
|
||||
ref={bookSummaryRef}
|
||||
className="flex gap-4 border-b border-(--border-muted) px-5 py-4"
|
||||
>
|
||||
{modalPreview ? (
|
||||
{book.preview ? (
|
||||
<img
|
||||
src={modalPreview}
|
||||
src={book.preview}
|
||||
alt="Book cover"
|
||||
width={book.cover_aspect === 'square' ? (book.series_name ? 144 : 120) : 96}
|
||||
height={book.series_name ? 144 : 120}
|
||||
decoding="async"
|
||||
className={`hidden shrink-0 rounded-lg object-cover shadow-md sm:block ${coverAspectClassName} ${coverSizeClassName}`}
|
||||
/>
|
||||
) : (
|
||||
|
||||
@@ -5,7 +5,6 @@ import { useEscapeKey } from '../hooks/useEscapeKey';
|
||||
import { useMountEffect } from '../hooks/useMountEffect';
|
||||
import { getMetadataBookInfo } from '../services/api';
|
||||
import type { CreateRequestPayload } from '../types';
|
||||
import { getSizedCoverUrl } from '../utils/covers';
|
||||
import type { RequestConfirmationPreview } from '../utils/requestConfirmation';
|
||||
import {
|
||||
applyRequestNoteToPayload,
|
||||
@@ -176,7 +175,6 @@ function RequestConfirmationModalSession({
|
||||
|
||||
const titleId = 'request-confirmation-modal-title';
|
||||
const confirmDisabled = isSubmitting || (allowNotes && note.length > MAX_REQUEST_NOTE_LENGTH);
|
||||
const previewImage = getSizedCoverUrl(preview.preview, { width: 64, height: 96 });
|
||||
|
||||
const submit = async () => {
|
||||
if (confirmDisabled) {
|
||||
@@ -242,15 +240,11 @@ function RequestConfirmationModalSession({
|
||||
<div className="rounded-xl border border-(--border-muted) bg-(--bg-soft) px-4 py-4">
|
||||
<div className="flex gap-4">
|
||||
<div className="h-24 w-16 shrink-0 overflow-hidden rounded-lg border border-(--border-muted) bg-(--bg)">
|
||||
{previewImage ? (
|
||||
{preview.preview ? (
|
||||
<img
|
||||
src={previewImage}
|
||||
src={preview.preview}
|
||||
alt={`${preview.title} cover`}
|
||||
className="h-full w-full object-cover object-top"
|
||||
loading="lazy"
|
||||
decoding="async"
|
||||
width={64}
|
||||
height={96}
|
||||
/>
|
||||
) : (
|
||||
<div className="flex h-full w-full items-center justify-center text-[10px] opacity-60">
|
||||
|
||||
@@ -3,7 +3,6 @@ import { useLayoutEffect, useMemo, useRef, useState } from 'react';
|
||||
|
||||
import type { RequestRecord } from '../../types';
|
||||
import { withBasePath } from '../../utils/basePath';
|
||||
import { getSizedCoverUrl } from '../../utils/covers';
|
||||
import { Tooltip } from '../shared/Tooltip';
|
||||
import type { ActivityCardAction } from './activityCardModel';
|
||||
import { buildActivityCardModel } from './activityCardModel';
|
||||
@@ -500,7 +499,6 @@ export const ActivityCard = ({
|
||||
const titleLineRef = useRef<HTMLParagraphElement | null>(null);
|
||||
const [badgeOverflow, setBadgeOverflow] = useState<Record<string, boolean>>({});
|
||||
const [titleOverflow, setTitleOverflow] = useState(false);
|
||||
const previewImage = getSizedCoverUrl(item.preview, { width: 48, height: 72 });
|
||||
|
||||
useLayoutEffect(() => {
|
||||
const measureBadgeOverflow = () => {
|
||||
@@ -711,15 +709,11 @@ export const ActivityCard = ({
|
||||
<div className="flex items-start gap-3">
|
||||
{/* Artwork */}
|
||||
<div className="h-18 w-12 shrink-0 overflow-hidden rounded-sm bg-gray-200 dark:bg-gray-700">
|
||||
{previewImage ? (
|
||||
{item.preview ? (
|
||||
<img
|
||||
src={previewImage}
|
||||
src={item.preview}
|
||||
alt={`${item.title} cover`}
|
||||
className="h-full w-full object-cover object-top"
|
||||
loading="lazy"
|
||||
decoding="async"
|
||||
width={48}
|
||||
height={72}
|
||||
/>
|
||||
) : (
|
||||
<BookFallback />
|
||||
|
||||
@@ -3,7 +3,6 @@ import { useState } from 'react';
|
||||
import { useSearchMode } from '../../contexts/SearchModeContext';
|
||||
import type { Book, ButtonStateInfo } from '../../types';
|
||||
import { bookSupportsTargets } from '../../utils/bookTargetLoader';
|
||||
import { getSizedCoverUrl } from '../../utils/covers';
|
||||
import { BookActionButton } from '../BookActionButton';
|
||||
import { BookTargetDropdown } from '../BookTargetDropdown';
|
||||
import { DisplayFieldBadges } from '../shared';
|
||||
@@ -42,11 +41,6 @@ export const CardView = ({
|
||||
const [dropdownOpen, setDropdownOpen] = useState(false);
|
||||
const targetProvider = book.provider;
|
||||
const targetBookId = book.provider_id;
|
||||
const isSquareCover = book.cover_aspect === 'square';
|
||||
const optimizedPreview = getSizedCoverUrl(book.preview, {
|
||||
width: 292,
|
||||
height: isSquareCover ? 292 : 438,
|
||||
});
|
||||
let zIndex: number | undefined;
|
||||
if (dropdownOpen) {
|
||||
zIndex = 20;
|
||||
@@ -103,7 +97,7 @@ export const CardView = ({
|
||||
#{book.series_position}
|
||||
</div>
|
||||
)}
|
||||
{optimizedPreview && !imageError ? (
|
||||
{book.preview && !imageError ? (
|
||||
<>
|
||||
{!imageLoaded && (
|
||||
<div className="absolute inset-0">
|
||||
@@ -111,13 +105,10 @@ export const CardView = ({
|
||||
</div>
|
||||
)}
|
||||
<img
|
||||
src={optimizedPreview}
|
||||
src={book.preview}
|
||||
alt={book.title || 'Book cover'}
|
||||
className="h-full w-full"
|
||||
loading="lazy"
|
||||
decoding="async"
|
||||
width={292}
|
||||
height={isSquareCover ? 292 : 438}
|
||||
style={{
|
||||
opacity: imageLoaded ? 1 : 0,
|
||||
transition: 'opacity 0.3s ease-in-out',
|
||||
|
||||
@@ -3,7 +3,6 @@ import { useState } from 'react';
|
||||
import { useSearchMode } from '../../contexts/SearchModeContext';
|
||||
import type { Book, ButtonStateInfo } from '../../types';
|
||||
import { bookSupportsTargets } from '../../utils/bookTargetLoader';
|
||||
import { getSizedCoverUrl } from '../../utils/covers';
|
||||
import { BookActionButton } from '../BookActionButton';
|
||||
import { BookTargetDropdown } from '../BookTargetDropdown';
|
||||
import { DisplayFieldBadges, DisplayFieldIcon } from '../shared';
|
||||
@@ -45,11 +44,6 @@ export const CompactView = ({
|
||||
const targetProvider = book.provider;
|
||||
const targetBookId = book.provider_id;
|
||||
const microphoneField = book.display_fields?.find((field) => field.icon === 'microphone');
|
||||
const isSquareCover = book.cover_aspect === 'square';
|
||||
const optimizedPreview = getSizedCoverUrl(book.preview, {
|
||||
width: 120,
|
||||
height: isSquareCover ? 120 : 180,
|
||||
});
|
||||
let zIndex: number | undefined;
|
||||
if (dropdownOpen) {
|
||||
zIndex = 20;
|
||||
@@ -103,7 +97,7 @@ export const CompactView = ({
|
||||
#{book.series_position}
|
||||
</div>
|
||||
)}
|
||||
{optimizedPreview && !imageError ? (
|
||||
{book.preview && !imageError ? (
|
||||
<>
|
||||
{!imageLoaded && (
|
||||
<div className="absolute inset-0">
|
||||
@@ -111,13 +105,10 @@ export const CompactView = ({
|
||||
</div>
|
||||
)}
|
||||
<img
|
||||
src={optimizedPreview}
|
||||
src={book.preview}
|
||||
alt={book.title || 'Book cover'}
|
||||
className="h-full w-full"
|
||||
loading="lazy"
|
||||
decoding="async"
|
||||
width={120}
|
||||
height={isSquareCover ? 120 : 180}
|
||||
style={{
|
||||
opacity: imageLoaded ? 1 : 0,
|
||||
transition: 'opacity 0.3s ease-in-out',
|
||||
|
||||
@@ -4,7 +4,6 @@ import { useSearchMode } from '../../contexts/SearchModeContext';
|
||||
import type { Book, ButtonStateInfo, DisplayField } from '../../types';
|
||||
import { bookSupportsTargets } from '../../utils/bookTargetLoader';
|
||||
import { getFormatColor, getLanguageColor } from '../../utils/colorMaps';
|
||||
import { getSizedCoverUrl } from '../../utils/covers';
|
||||
import { BookActionButton } from '../BookActionButton';
|
||||
import { BookTargetDropdown } from '../BookTargetDropdown';
|
||||
import { DisplayFieldIcon, DisplayFieldBadge } from '../shared';
|
||||
@@ -48,12 +47,8 @@ const ListViewThumbnail = ({
|
||||
const [imageError, setImageError] = useState(false);
|
||||
const isSquare = coverAspect === 'square';
|
||||
const sizeClass = isSquare ? 'w-10 h-10 sm:w-14 sm:h-14' : 'w-7 h-10 sm:w-10 sm:h-14';
|
||||
const optimizedPreview = getSizedCoverUrl(preview, {
|
||||
width: isSquare ? 56 : 40,
|
||||
height: isSquare ? 56 : 56,
|
||||
});
|
||||
|
||||
if (!optimizedPreview || imageError) {
|
||||
if (!preview || imageError) {
|
||||
return (
|
||||
<div
|
||||
className={`${sizeClass} flex items-center justify-center rounded-sm bg-gray-200 text-[8px] font-medium text-gray-500 sm:text-[9px] dark:bg-gray-700 dark:text-gray-300`}
|
||||
@@ -72,13 +67,10 @@ const ListViewThumbnail = ({
|
||||
<div className="absolute inset-0 animate-pulse bg-linear-to-r from-gray-200 via-gray-100 to-gray-200 dark:from-gray-700 dark:via-gray-600 dark:to-gray-700" />
|
||||
)}
|
||||
<img
|
||||
src={optimizedPreview}
|
||||
src={preview}
|
||||
alt={title || 'Book cover'}
|
||||
className={`h-full w-full object-cover ${isSquare ? 'object-center' : 'object-top'}`}
|
||||
loading="lazy"
|
||||
decoding="async"
|
||||
width={isSquare ? 56 : 40}
|
||||
height={isSquare ? 56 : 56}
|
||||
onLoad={() => setImageLoaded(true)}
|
||||
onError={() => setImageError(true)}
|
||||
style={{ opacity: imageLoaded ? 1 : 0, transition: 'opacity 0.2s ease-in-out' }}
|
||||
|
||||
@@ -106,7 +106,6 @@ export const NamingTemplateField = ({
|
||||
disabled={fieldDisabled}
|
||||
className="w-full rounded-lg border border-(--border-muted) bg-(--bg-soft) px-3 py-2 text-sm transition-colors focus:border-sky-500 focus:ring-2 focus:ring-sky-500/50 focus:outline-hidden disabled:cursor-not-allowed disabled:opacity-60"
|
||||
/>
|
||||
{boundField.description && <p className="text-xs opacity-60">{boundField.description}</p>}
|
||||
</div>
|
||||
|
||||
{(hasPathSeparatorInFilename || preview.unknownTokens.length > 0) && (
|
||||
|
||||
@@ -1,33 +0,0 @@
|
||||
import { describe, expect, it } from 'vitest';
|
||||
|
||||
import { getSizedCoverUrl } from '../utils/covers';
|
||||
|
||||
describe('getSizedCoverUrl', () => {
|
||||
it('adds size and format params to local cover proxy URLs', () => {
|
||||
expect(
|
||||
getSizedCoverUrl('/api/covers/book-1?url=abc', {
|
||||
width: 120,
|
||||
height: 180,
|
||||
}),
|
||||
).toBe('/api/covers/book-1?url=abc&w=120&h=180&format=webp');
|
||||
});
|
||||
|
||||
it('leaves external preview URLs alone', () => {
|
||||
expect(
|
||||
getSizedCoverUrl('https://covers.example.com/book.jpg', {
|
||||
width: 120,
|
||||
height: 180,
|
||||
}),
|
||||
).toBe('https://covers.example.com/book.jpg');
|
||||
});
|
||||
|
||||
it('preserves absolute proxy URLs', () => {
|
||||
expect(
|
||||
getSizedCoverUrl('https://bookrequest.example.com/api/covers/book-1?url=abc', {
|
||||
width: 56,
|
||||
height: 56,
|
||||
format: 'png',
|
||||
}),
|
||||
).toBe('https://bookrequest.example.com/api/covers/book-1?url=abc&w=56&h=56&format=png');
|
||||
});
|
||||
});
|
||||
@@ -2,11 +2,18 @@ import { describe, expect, it } from 'vitest';
|
||||
|
||||
import {
|
||||
buildNamingTemplatePreview,
|
||||
NAMING_TEMPLATE_TOKENS,
|
||||
renderNamingTemplate,
|
||||
SAMPLE_NAMING_METADATA,
|
||||
} from '../utils/namingTemplatePreview';
|
||||
|
||||
describe('namingTemplatePreview', () => {
|
||||
it('groups primary title with universal variables', () => {
|
||||
expect(NAMING_TEMPLATE_TOKENS.find((token) => token.token === 'PrimaryTitle')?.group).toBe(
|
||||
'Universal',
|
||||
);
|
||||
});
|
||||
|
||||
it('renders primary title in path previews', () => {
|
||||
const preview = buildNamingTemplatePreview(
|
||||
'{Author}/{Series/}{SeriesPosition - }{PrimaryTitle} ({Year})',
|
||||
|
||||
@@ -1,66 +0,0 @@
|
||||
const COVER_PROXY_PATH = '/api/covers/';
|
||||
const LOCAL_URL_BASE = 'http://shelfmark.local';
|
||||
const DEFAULT_COVER_FORMAT = 'webp';
|
||||
const MAX_COVER_DIMENSION = 1024;
|
||||
|
||||
type CoverFormat = 'jpeg' | 'png' | 'webp';
|
||||
|
||||
interface SizedCoverUrlOptions {
|
||||
width?: number;
|
||||
height?: number;
|
||||
format?: CoverFormat;
|
||||
}
|
||||
|
||||
const normalizeDimension = (value?: number) => {
|
||||
if (typeof value !== 'number' || !Number.isFinite(value)) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
const rounded = Math.round(value);
|
||||
if (rounded <= 0) {
|
||||
return undefined;
|
||||
}
|
||||
|
||||
return Math.min(rounded, MAX_COVER_DIMENSION);
|
||||
};
|
||||
|
||||
export const getSizedCoverUrl = (
|
||||
preview?: string,
|
||||
{ width, height, format = DEFAULT_COVER_FORMAT }: SizedCoverUrlOptions = {},
|
||||
) => {
|
||||
if (!preview) {
|
||||
return preview;
|
||||
}
|
||||
|
||||
const isRelativeUrl = preview.startsWith('/');
|
||||
|
||||
let url: URL;
|
||||
try {
|
||||
url = new URL(preview, LOCAL_URL_BASE);
|
||||
} catch {
|
||||
return preview;
|
||||
}
|
||||
|
||||
if (!url.pathname.includes(COVER_PROXY_PATH)) {
|
||||
return preview;
|
||||
}
|
||||
|
||||
const normalizedWidth = normalizeDimension(width);
|
||||
const normalizedHeight = normalizeDimension(height);
|
||||
|
||||
if (normalizedWidth !== undefined) {
|
||||
url.searchParams.set('w', String(normalizedWidth));
|
||||
}
|
||||
|
||||
if (normalizedHeight !== undefined) {
|
||||
url.searchParams.set('h', String(normalizedHeight));
|
||||
}
|
||||
|
||||
if (format) {
|
||||
url.searchParams.set('format', format);
|
||||
}
|
||||
|
||||
const search = url.searchParams.toString();
|
||||
const relativeUrl = `${url.pathname}${search ? `?${search}` : ''}${url.hash}`;
|
||||
return isRelativeUrl ? relativeUrl : url.toString();
|
||||
};
|
||||
@@ -39,7 +39,7 @@ export const NAMING_TEMPLATE_TOKENS: NamingTemplateToken[] = [
|
||||
label: 'Primary title',
|
||||
description: 'Title without the subtitle suffix',
|
||||
value: 'The Hound of the Baskervilles',
|
||||
group: 'Core',
|
||||
group: 'Universal',
|
||||
},
|
||||
{
|
||||
token: 'Year',
|
||||
|
||||
@@ -156,6 +156,21 @@ def test_download_settings_naming_templates_use_wrapped_custom_component():
|
||||
assert value_key not in fields_by_key
|
||||
|
||||
|
||||
def test_download_settings_naming_template_value_fields_are_registered():
|
||||
import shelfmark.config.settings # noqa: F401
|
||||
from shelfmark.core import settings_registry
|
||||
|
||||
field_map = settings_registry.get_settings_field_map(tab_name="downloads")
|
||||
|
||||
for value_key in (
|
||||
"TEMPLATE_RENAME",
|
||||
"TEMPLATE_ORGANIZE",
|
||||
"TEMPLATE_AUDIOBOOK_RENAME",
|
||||
"TEMPLATE_AUDIOBOOK_ORGANIZE",
|
||||
):
|
||||
assert value_key in field_map
|
||||
|
||||
|
||||
def test_download_settings_naming_template_serialization_keeps_value_fields_hidden():
|
||||
from shelfmark.config.settings import download_settings
|
||||
from shelfmark.core import settings_registry
|
||||
|
||||
@@ -38,3 +38,22 @@ def test_generated_env_docs_describe_mirror_lists_as_comma_separated_strings() -
|
||||
"| `LIBGEN_MIRROR_URLS` | Mirrors are tried in the order you add them until one works. | "
|
||||
"string (comma-separated) | _empty list_ |"
|
||||
) in docs
|
||||
|
||||
|
||||
def test_generated_env_docs_include_custom_component_value_fields() -> None:
|
||||
docs = generate_env_docs()
|
||||
|
||||
for env_var in (
|
||||
"TEMPLATE_RENAME",
|
||||
"TEMPLATE_ORGANIZE",
|
||||
"TEMPLATE_AUDIOBOOK_RENAME",
|
||||
"TEMPLATE_AUDIOBOOK_ORGANIZE",
|
||||
):
|
||||
assert f"`{env_var}`" in docs
|
||||
|
||||
assert (
|
||||
"| `TEMPLATE_AUDIOBOOK_ORGANIZE` | Use / to create folders. Variables: "
|
||||
"{Author}, {Title}, {Year}, {User}, {OriginalName} "
|
||||
"(source filename without extension), {Series}, {SeriesPosition}, {Subtitle}, "
|
||||
"{PrimaryTitle}, {PartNumber}. Use arbitrary prefix/suffix:"
|
||||
) in docs
|
||||
|
||||
@@ -1,25 +1,8 @@
|
||||
"""Tests for targeted image cache safety and fetch fallbacks."""
|
||||
|
||||
from io import BytesIO
|
||||
|
||||
import requests
|
||||
from PIL import Image
|
||||
|
||||
from shelfmark.core.image_cache import (
|
||||
ImageCacheService,
|
||||
build_variant_cache_id,
|
||||
create_image_variant,
|
||||
normalize_variant_dimension,
|
||||
normalize_variant_format,
|
||||
)
|
||||
|
||||
|
||||
def _make_image_bytes(
|
||||
*, width: int = 400, height: int = 600, image_format: str = "JPEG", color: str = "navy"
|
||||
) -> bytes:
|
||||
buffer = BytesIO()
|
||||
Image.new("RGB", (width, height), color=color).save(buffer, format=image_format)
|
||||
return buffer.getvalue()
|
||||
from shelfmark.core.image_cache import ImageCacheService
|
||||
|
||||
|
||||
def test_is_safe_url_rejects_invalid_ipv6_url() -> None:
|
||||
@@ -37,58 +20,3 @@ def test_fetch_and_cache_returns_none_on_request_exception(tmp_path, monkeypatch
|
||||
|
||||
assert cache.fetch_and_cache("cover-1", "https://example.com/cover.jpg") is None
|
||||
assert "cover-1" not in cache._index
|
||||
|
||||
|
||||
def test_create_image_variant_resizes_and_transcodes_to_webp() -> None:
|
||||
variant = create_image_variant(
|
||||
_make_image_bytes(),
|
||||
width=120,
|
||||
height=180,
|
||||
image_format="webp",
|
||||
)
|
||||
|
||||
assert variant is not None
|
||||
variant_bytes, content_type = variant
|
||||
assert content_type == "image/webp"
|
||||
|
||||
with Image.open(BytesIO(variant_bytes)) as image:
|
||||
assert image.size == (120, 180)
|
||||
|
||||
|
||||
def test_create_image_variant_preserves_aspect_ratio_for_single_dimension() -> None:
|
||||
variant = create_image_variant(
|
||||
_make_image_bytes(),
|
||||
width=120,
|
||||
image_format="jpeg",
|
||||
)
|
||||
|
||||
assert variant is not None
|
||||
variant_bytes, content_type = variant
|
||||
assert content_type == "image/jpeg"
|
||||
|
||||
with Image.open(BytesIO(variant_bytes)) as image:
|
||||
assert image.size == (120, 180)
|
||||
|
||||
|
||||
def test_create_image_variant_returns_none_when_no_change_needed() -> None:
|
||||
image_bytes = _make_image_bytes(image_format="WEBP")
|
||||
|
||||
assert create_image_variant(image_bytes, image_format="webp") is None
|
||||
|
||||
|
||||
def test_variant_helpers_normalize_requested_variant_values() -> None:
|
||||
assert normalize_variant_dimension("240") == 240
|
||||
assert normalize_variant_dimension("0") is None
|
||||
assert normalize_variant_dimension("99999") == 1024
|
||||
assert normalize_variant_format("jpg") == "jpeg"
|
||||
assert normalize_variant_format("weBp") == "webp"
|
||||
assert normalize_variant_format("gif") is None
|
||||
assert (
|
||||
build_variant_cache_id(
|
||||
"cover-123",
|
||||
width=120,
|
||||
height=180,
|
||||
image_format="webp",
|
||||
)
|
||||
== "cover-123__w120_h180_fwebp"
|
||||
)
|
||||
|
||||
@@ -202,7 +202,7 @@ while true; do
|
||||
exit 1
|
||||
fi
|
||||
|
||||
CURRENT_LOG=$(tail -n 1 /var/log/tor/notices.log 2>/dev/null)
|
||||
CURRENT_LOG=$(tail -n 1 /var/log/tor/notices.log 2>/dev/null || true)
|
||||
printf "\r\033[K[%ds] %s" "$ELAPSED" "$CURRENT_LOG"
|
||||
sleep 1
|
||||
done
|
||||
|
||||
@@ -718,39 +718,6 @@ wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/51/fe/53ac0cd932db5dcaf55961bc7cb7afdca8d80d8cc7406ed661f0c7dc111a/pdbp-1.8.2-py3-none-any.whl", hash = "sha256:d4fd05e177636b5ccd0b2e03e378cec57afc06149e5fd975de6f8ddb3d0109a8", size = 21969, upload-time = "2026-01-14T03:10:27.062Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pillow"
|
||||
version = "12.2.0"
|
||||
source = { registry = "https://pypi.org/simple" }
|
||||
sdist = { url = "https://files.pythonhosted.org/packages/8c/21/c2bcdd5906101a30244eaffc1b6e6ce71a31bd0742a01eb89e660ebfac2d/pillow-12.2.0.tar.gz", hash = "sha256:a830b1a40919539d07806aa58e1b114df53ddd43213d9c8b75847eee6c0182b5", size = 46987819, upload-time = "2026-04-01T14:46:17.687Z" }
|
||||
wheels = [
|
||||
{ url = "https://files.pythonhosted.org/packages/bf/98/4595daa2365416a86cb0d495248a393dfc84e96d62ad080c8546256cb9c0/pillow-12.2.0-cp314-cp314-ios_13_0_arm64_iphoneos.whl", hash = "sha256:3adc9215e8be0448ed6e814966ecf3d9952f0ea40eb14e89a102b87f450660d8", size = 4100848, upload-time = "2026-04-01T14:44:48.48Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/0b/79/40184d464cf89f6663e18dfcf7ca21aae2491fff1a16127681bf1fa9b8cf/pillow-12.2.0-cp314-cp314-ios_13_0_arm64_iphonesimulator.whl", hash = "sha256:6a9adfc6d24b10f89588096364cc726174118c62130c817c2837c60cf08a392b", size = 4176515, upload-time = "2026-04-01T14:44:51.353Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/b0/63/703f86fd4c422a9cf722833670f4f71418fb116b2853ff7da722ea43f184/pillow-12.2.0-cp314-cp314-ios_13_0_x86_64_iphonesimulator.whl", hash = "sha256:6a6e67ea2e6feda684ed370f9a1c52e7a243631c025ba42149a2cc5934dec295", size = 3640159, upload-time = "2026-04-01T14:44:53.588Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/71/e0/fb22f797187d0be2270f83500aab851536101b254bfa1eae10795709d283/pillow-12.2.0-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:2bb4a8d594eacdfc59d9e5ad972aa8afdd48d584ffd5f13a937a664c3e7db0ed", size = 5312185, upload-time = "2026-04-01T14:44:56.039Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/ba/8c/1a9e46228571de18f8e28f16fabdfc20212a5d019f3e3303452b3f0a580d/pillow-12.2.0-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:80b2da48193b2f33ed0c32c38140f9d3186583ce7d516526d462645fd98660ae", size = 4695386, upload-time = "2026-04-01T14:44:58.663Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/70/62/98f6b7f0c88b9addd0e87c217ded307b36be024d4ff8869a812b241d1345/pillow-12.2.0-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:22db17c68434de69d8ecfc2fe821569195c0c373b25cccb9cbdacf2c6e53c601", size = 6280384, upload-time = "2026-04-01T14:45:01.5Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/5e/03/688747d2e91cfbe0e64f316cd2e8005698f76ada3130d0194664174fa5de/pillow-12.2.0-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:7b14cc0106cd9aecda615dd6903840a058b4700fcb817687d0ee4fc8b6e389be", size = 8091599, upload-time = "2026-04-01T14:45:04.5Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/f6/35/577e22b936fcdd66537329b33af0b4ccfefaeabd8aec04b266528cddb33c/pillow-12.2.0-cp314-cp314-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:8cbeb542b2ebc6fcdacabf8aca8c1a97c9b3ad3927d46b8723f9d4f033288a0f", size = 6396021, upload-time = "2026-04-01T14:45:07.117Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/11/8d/d2532ad2a603ca2b93ad9f5135732124e57811d0168155852f37fbce2458/pillow-12.2.0-cp314-cp314-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:4bfd07bc812fbd20395212969e41931001fd59eb55a60658b0e5710872e95286", size = 7083360, upload-time = "2026-04-01T14:45:09.763Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/5e/26/d325f9f56c7e039034897e7380e9cc202b1e368bfd04d4cbe6a441f02885/pillow-12.2.0-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:9aba9a17b623ef750a4d11b742cbafffeb48a869821252b30ee21b5e91392c50", size = 6507628, upload-time = "2026-04-01T14:45:12.378Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/5f/f7/769d5632ffb0988f1c5e7660b3e731e30f7f8ec4318e94d0a5d674eb65a4/pillow-12.2.0-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:deede7c263feb25dba4e82ea23058a235dcc2fe1f6021025dc71f2b618e26104", size = 7209321, upload-time = "2026-04-01T14:45:15.122Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/6a/7a/c253e3c645cd47f1aceea6a8bacdba9991bf45bb7dfe927f7c893e89c93c/pillow-12.2.0-cp314-cp314-win32.whl", hash = "sha256:632ff19b2778e43162304d50da0181ce24ac5bb8180122cbe1bf4673428328c7", size = 6479723, upload-time = "2026-04-01T14:45:17.797Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/cd/8b/601e6566b957ca50e28725cb6c355c59c2c8609751efbecd980db44e0349/pillow-12.2.0-cp314-cp314-win_amd64.whl", hash = "sha256:4e6c62e9d237e9b65fac06857d511e90d8461a32adcc1b9065ea0c0fa3a28150", size = 7217400, upload-time = "2026-04-01T14:45:20.529Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/d6/94/220e46c73065c3e2951bb91c11a1fb636c8c9ad427ac3ce7d7f3359b9b2f/pillow-12.2.0-cp314-cp314-win_arm64.whl", hash = "sha256:b1c1fbd8a5a1af3412a0810d060a78b5136ec0836c8a4ef9aa11807f2a22f4e1", size = 2554835, upload-time = "2026-04-01T14:45:23.162Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/b6/ab/1b426a3974cb0e7da5c29ccff4807871d48110933a57207b5a676cccc155/pillow-12.2.0-cp314-cp314t-macosx_10_15_x86_64.whl", hash = "sha256:57850958fe9c751670e49b2cecf6294acc99e562531f4bd317fa5ddee2068463", size = 5314225, upload-time = "2026-04-01T14:45:25.637Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/19/1e/dce46f371be2438eecfee2a1960ee2a243bbe5e961890146d2dee1ff0f12/pillow-12.2.0-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:d5d38f1411c0ed9f97bcb49b7bd59b6b7c314e0e27420e34d99d844b9ce3b6f3", size = 4698541, upload-time = "2026-04-01T14:45:28.355Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/55/c3/7fbecf70adb3a0c33b77a300dc52e424dc22ad8cdc06557a2e49523b703d/pillow-12.2.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:5c0a9f29ca8e79f09de89293f82fc9b0270bb4af1d58bc98f540cc4aedf03166", size = 6322251, upload-time = "2026-04-01T14:45:30.924Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/1c/3c/7fbc17cfb7e4fe0ef1642e0abc17fc6c94c9f7a16be41498e12e2ba60408/pillow-12.2.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:1610dd6c61621ae1cf811bef44d77e149ce3f7b95afe66a4512f8c59f25d9ebe", size = 8127807, upload-time = "2026-04-01T14:45:33.908Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/ff/c3/a8ae14d6defd2e448493ff512fae903b1e9bd40b72efb6ec55ce0048c8ce/pillow-12.2.0-cp314-cp314t-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0a34329707af4f73cf1782a36cd2289c0368880654a2c11f027bcee9052d35dd", size = 6433935, upload-time = "2026-04-01T14:45:36.623Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/6e/32/2880fb3a074847ac159d8f902cb43278a61e85f681661e7419e6596803ed/pillow-12.2.0-cp314-cp314t-manylinux_2_27_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8e9c4f5b3c546fa3458a29ab22646c1c6c787ea8f5ef51300e5a60300736905e", size = 7116720, upload-time = "2026-04-01T14:45:39.258Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/46/87/495cc9c30e0129501643f24d320076f4cc54f718341df18cc70ec94c44e1/pillow-12.2.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:fb043ee2f06b41473269765c2feae53fc2e2fbf96e5e22ca94fb5ad677856f06", size = 6540498, upload-time = "2026-04-01T14:45:41.879Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/18/53/773f5edca692009d883a72211b60fdaf8871cbef075eaa9d577f0a2f989e/pillow-12.2.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:f278f034eb75b4e8a13a54a876cc4a5ab39173d2cdd93a638e1b467fc545ac43", size = 7239413, upload-time = "2026-04-01T14:45:44.705Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/c9/e4/4b64a97d71b2a83158134abbb2f5bd3f8a2ea691361282f010998f339ec7/pillow-12.2.0-cp314-cp314t-win32.whl", hash = "sha256:6bb77b2dcb06b20f9f4b4a8454caa581cd4dd0643a08bacf821216a16d9c8354", size = 6482084, upload-time = "2026-04-01T14:45:47.568Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/ba/13/306d275efd3a3453f72114b7431c877d10b1154014c1ebbedd067770d629/pillow-12.2.0-cp314-cp314t-win_amd64.whl", hash = "sha256:6562ace0d3fb5f20ed7290f1f929cae41b25ae29528f2af1722966a0a02e2aa1", size = 7225152, upload-time = "2026-04-01T14:45:50.032Z" },
|
||||
{ url = "https://files.pythonhosted.org/packages/ff/6e/cf826fae916b8658848d7b9f38d88da6396895c676e8086fc0988073aaf8/pillow-12.2.0-cp314-cp314t-win_arm64.whl", hash = "sha256:aa88ccfe4e32d362816319ed727a004423aab09c5cea43c01a4b435643fa34eb", size = 2556579, upload-time = "2026-04-01T14:45:52.529Z" },
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "pip"
|
||||
version = "26.0.1"
|
||||
@@ -1373,7 +1340,6 @@ dependencies = [
|
||||
{ name = "gevent" },
|
||||
{ name = "gevent-websocket" },
|
||||
{ name = "gunicorn" },
|
||||
{ name = "pillow" },
|
||||
{ name = "psutil" },
|
||||
{ name = "python-socketio" },
|
||||
{ name = "qbittorrent-api" },
|
||||
@@ -1416,7 +1382,6 @@ requires-dist = [
|
||||
{ name = "gevent" },
|
||||
{ name = "gevent-websocket" },
|
||||
{ name = "gunicorn" },
|
||||
{ name = "pillow", specifier = ">=11.0.0" },
|
||||
{ name = "psutil" },
|
||||
{ name = "pyautogui", marker = "extra == 'browser'" },
|
||||
{ name = "python-socketio" },
|
||||
|
||||
Reference in New Issue
Block a user