Compare commits

..
Author SHA1 Message Date
Alex 3ab503d82d Split hardcover 2026-04-29 19:37:27 +01:00
107 changed files with 4646 additions and 7027 deletions
+3 -17
View File
@@ -1,17 +1,10 @@
version: 2
updates:
# Python dependencies
# Dependabot supports uv version updates, but GitHub currently lists uv
# security updates as "Not applicable"; daily checks keep uv.lock moving
# while repo-level Dependabot alerts/security updates cover supported ecosystems.
- package-ecosystem: "uv"
directory: "/"
schedule:
interval: "daily"
time: "05:00"
timezone: "Europe/London"
cooldown:
default-days: 3
interval: "weekly"
open-pull-requests-limit: 10
groups:
python-deps:
@@ -23,25 +16,20 @@ updates:
directory: "/src/frontend"
schedule:
interval: "weekly"
cooldown:
default-days: 3
open-pull-requests-limit: 10
groups:
npm-deps:
patterns: ["*"]
update-types: ["minor", "patch"]
# Dockerfile base image digests. When a tag stays the same, Dependabot titles
# can only show digest prefixes, so keep the group name explicit.
# Dockerfile base images
- package-ecosystem: "docker"
directory: "/"
schedule:
interval: "weekly"
cooldown:
default-days: 3
open-pull-requests-limit: 5
groups:
docker-base-image-digests:
docker-images:
patterns: ["*"]
ignore:
# Node.js: block major-version bumps so dependabot never proposes
@@ -55,8 +43,6 @@ updates:
directory: "/"
schedule:
interval: "weekly"
cooldown:
default-days: 3
open-pull-requests-limit: 5
groups:
gh-actions:
+3 -3
View File
@@ -25,14 +25,14 @@ jobs:
uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
- name: Initialize CodeQL
uses: github/codeql-action/init@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v3
uses: github/codeql-action/init@95e58e9a2cdfd71adc6e0353d5c52f41a045d225 # v3
with:
languages: ${{ matrix.language }}
- name: Autobuild
uses: github/codeql-action/autobuild@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v3
uses: github/codeql-action/autobuild@95e58e9a2cdfd71adc6e0353d5c52f41a045d225 # v3
- name: Perform CodeQL Analysis
uses: github/codeql-action/analyze@68bde559dea0fdcac2102bfdf6230c5f70eb485e # v3
uses: github/codeql-action/analyze@95e58e9a2cdfd71adc6e0353d5c52f41a045d225 # v3
with:
category: "/language:${{ matrix.language }}"
+3 -18
View File
@@ -4,7 +4,7 @@ ARG BUILDPLATFORM
ARG BUILDARCH
# Frontend build stage.
FROM --platform=$BUILDPLATFORM node:24-alpine@sha256:d1b3b4da11eefd5941e7f0b9cf17783fc99d9c6fc34884a665f40a06dbdfc94f AS frontend-builder
FROM --platform=$BUILDPLATFORM node:24-alpine AS frontend-builder
# Helpful debug output to see what platforms BuildKit thinks it's using
RUN echo "BUILDPLATFORM=$BUILDPLATFORM BUILDARCH=$BUILDARCH TARGETPLATFORM=$TARGETPLATFORM TARGETARCH=$TARGETARCH"
@@ -25,9 +25,9 @@ COPY src/frontend/ ./
RUN npm run build
# Use python-slim as the base image
FROM python:3.14-slim@sha256:1697e8e8d39bf168e177ac6b5fdab6df86d81cfc24dae17dfb96cfc3ef76b4dd AS base
FROM python:3.14-slim AS base
COPY --from=ghcr.io/astral-sh/uv:0.11.3@sha256:90bbb3c16635e9627f49eec6539f956d70746c409209041800a0280b93152823 /uv /uvx /bin/
COPY --from=ghcr.io/astral-sh/uv:0.11.3 /uv /uvx /bin/
# Add build argument for version
ARG BUILD_VERSION
@@ -103,15 +103,6 @@ COPY pyproject.toml uv.lock ./
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --locked --no-default-groups
# Runtime dependencies are installed into /app/.venv during the build. Remove the
# base image's system pip so stale installer CVEs do not ship in the final image.
RUN rm -rf \
/usr/local/bin/pip \
/usr/local/bin/pip3 \
/usr/local/bin/pip3.* \
/usr/local/lib/python*/site-packages/pip \
/usr/local/lib/python*/site-packages/pip-*.dist-info
# Copy application code *after* dependencies are installed
COPY . .
@@ -173,9 +164,6 @@ RUN apt-get update && \
RUN --mount=type=cache,target=/root/.cache/uv \
uv sync --locked --no-default-groups --extra browser
# uv is only needed while building the image.
RUN rm -f /usr/bin/uv /usr/bin/uvx
# Keep SeleniumBase's bundled driver cache writable for the fixed non-root user.
RUN SELENIUMBASE_DRIVERS_DIR=$(/app/.venv/bin/python -c "import pathlib, seleniumbase; print(pathlib.Path(seleniumbase.__file__).resolve().parent / 'drivers')") && \
chown -R 1000:1000 "${SELENIUMBASE_DRIVERS_DIR}" && \
@@ -192,7 +180,4 @@ FROM base AS shelfmark-lite
ENV USING_EXTERNAL_BYPASSER=true
# uv is only needed while building the image.
RUN rm -f /usr/bin/uv /usr/bin/uvx
CMD ["/app/entrypoint.sh"]
+14 -110
View File
@@ -14,7 +14,6 @@ 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)
@@ -31,7 +30,7 @@ This document lists all configuration options that can be set via environment va
## Bootstrap Configuration
These environment variables are used at startup before the settings system loads. They typically configure paths, server settings, and authentication startup behavior.
These environment variables are used at startup before the settings system loads. They typically configure paths and server settings.
| Variable | Description | Type | Default |
|----------|-------------|------|---------|
@@ -43,9 +42,6 @@ These environment variables are used at startup before the settings system loads
| `FLASK_PORT` | Port number for the Flask web server. | number | `8084` |
| `SESSION_COOKIE_SECURE` | Enable secure cookies (requires HTTPS). | boolean | `false` |
| `CWA_DB_PATH` | Path to the Calibre-Web database for authentication integration. | string (path) | `/auth/app.db` |
| `HIDE_LOCAL_AUTH` | Hide the username/password login form when OIDC is active. | boolean | `false` |
| `DISABLE_LOCAL_AUTH` | Disable username/password login and remove the local-admin prerequisite for OIDC. Implies HIDE_LOCAL_AUTH; with AUTH_METHOD=builtin, everyone is locked out until auth env vars are changed. | boolean | `false` |
| `OIDC_AUTO_REDIRECT` | Automatically redirect to the OIDC provider instead of showing the login page. | boolean | `false` |
| `DOCKERMODE` | Indicates the application is running inside a Docker container. | boolean | `false` |
| `ONBOARDING` | Show the onboarding wizard on first run. Set to false to skip (useful for ephemeral storage). | boolean | `true` |
@@ -108,27 +104,6 @@ Path to the Calibre-Web database for authentication integration.
- **Type:** string (path)
- **Default:** `/auth/app.db`
#### `HIDE_LOCAL_AUTH`
Hide the username/password login form when OIDC is active.
- **Type:** boolean
- **Default:** `false`
#### `DISABLE_LOCAL_AUTH`
Disable username/password login and remove the local-admin prerequisite for OIDC. Implies HIDE_LOCAL_AUTH; with AUTH_METHOD=builtin, everyone is locked out until auth env vars are changed.
- **Type:** boolean
- **Default:** `false`
#### `OIDC_AUTO_REDIRECT`
Automatically redirect to the OIDC provider instead of showing the login page.
- **Type:** boolean
- **Default:** `false`
#### `DOCKERMODE`
Indicates the application is running inside a Docker container.
@@ -149,7 +124,6 @@ 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` |
@@ -159,15 +133,6 @@ 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**
@@ -329,8 +294,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}, {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})` |
| `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})` |
| `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_ |
@@ -346,13 +311,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}, {PrimaryTitle}, {Year}, {Series}, {SeriesPosition}, {Subtitle}, {Format}. | string | `{Title}` |
| `EMAIL_SUBJECT_TEMPLATE` | Email subject. Variables: {Author}, {Title}, {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}, {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}` |
| `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}` |
| `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_ |
@@ -396,7 +361,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}, {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.
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.
- **Type:** string
- **Default:** `{Author} - {Title} ({Year})`
@@ -405,7 +370,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}, {PrimaryTitle}. 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}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty.
- **Type:** string
- **Default:** `{Author}/{Title} ({Year})`
@@ -559,7 +524,7 @@ From address used for the email. You can include a display name (e.g., Shelfmark
**Subject Template**
Email subject. Variables: {Author}, {Title}, {PrimaryTitle}, {Year}, {Series}, {SeriesPosition}, {Subtitle}, {Format}.
Email subject. Variables: {Author}, {Title}, {Year}, {Series}, {SeriesPosition}, {Subtitle}, {Format}.
- **Type:** string
- **Default:** `{Title}`
@@ -606,7 +571,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}, {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.
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.
- **Type:** string
- **Default:** `{Author} - {Title}`
@@ -615,10 +580,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}, {PrimaryTitle}, {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}, {PartNumber}. Use arbitrary prefix/suffix: {Vol. SeriesPosition - } outputs 'Vol. 2 - ' when set, nothing when empty.
- **Type:** string
- **Default:** `{Author}/{Title}/{Title}`
- **Default:** `{Author}/{Title}`
#### `HARDLINK_TORRENTS_AUDIOBOOK`
@@ -674,7 +639,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. Restart container after changing Calibre-Web passwords. | string (choice) | `none` |
| `AUTH_METHOD` | Select the authentication method for accessing Shelfmark. | 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` |
@@ -696,7 +661,7 @@ How long to keep completed/failed downloads in the queue display.
**Authentication Method**
Select the authentication method for accessing Shelfmark. Restart container after changing Calibre-Web passwords.
Select the authentication method for accessing Shelfmark.
- **Type:** string (choice)
- **Default:** `none`
@@ -1097,7 +1062,6 @@ How long to cache individual book details. Default: 600 (10 minutes). Max: 60480
| `PROWLARR_API_KEY` | Found in Prowlarr: Settings > General > API Key | string (secret) | _none_ |
| `PROWLARR_INDEXERS` | Select which indexers to search. 📚 = has book categories. Leave empty to search all. | string (comma-separated) | _empty list_ |
| `PROWLARR_AUTO_EXPAND` | Automatically retry search without category filtering if no results are found | boolean | `false` |
| `PROWLARR_USE_SEED_PREFERENCES` | Apply per-indexer seed time and ratio preferences from Prowlarr when sending torrents to the download client | boolean | `false` |
<details>
<summary>Detailed descriptions</summary>
@@ -1149,66 +1113,6 @@ Automatically retry search without category filtering if no results are found
- **Type:** boolean
- **Default:** `false`
#### `PROWLARR_USE_SEED_PREFERENCES`
**Use Prowlarr seed preferences**
Apply per-indexer seed time and ratio preferences from Prowlarr when sending torrents to the download client
- **Type:** boolean
- **Default:** `false`
</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
+1 -2
View File
@@ -39,10 +39,9 @@ These optional environment variables control login page behavior when OIDC is en
| Variable | Description | Default |
|----------|-------------|---------|
| `HIDE_LOCAL_AUTH` | Hide the username/password login option, so only the OIDC button is shown | `false` |
| `DISABLE_LOCAL_AUTH` | Disable username/password login and remove the local-admin prerequisite for OIDC. Implies `HIDE_LOCAL_AUTH`; with `AUTH_METHOD=builtin`, everyone is locked out until auth env vars are changed. | `false` |
| `OIDC_AUTO_REDIRECT` | Automatically redirect to the OIDC provider instead of showing the login page | `false` |
If `DISABLE_LOCAL_AUTH` and `OIDC_AUTO_REDIRECT` are both enabled, users are redirected straight to the OIDC provider. On failure they return to the login page with an error message but no password fallback.
If both are enabled, users are redirected straight to the OIDC provider. On failure they return to the login page with an error message but no password fallback.
## Troubleshooting
+27 -43
View File
@@ -106,18 +106,6 @@ 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"
@@ -322,27 +310,6 @@ require_writable_dir() {
fi
}
fail_unwritable_config_dir() {
local folder="$1"
local owner
owner=$(stat -c '%u:%g' "$folder" 2>/dev/null || echo "unknown")
echo ""
echo "========================================================"
echo "ERROR: Config directory is not writable!"
echo ""
echo "Config directory: $folder"
echo "Current owner: $owner"
echo "Configured runtime identity: ${RUN_UID}:${RUN_GID}"
echo ""
echo "To fix this permanently, run on your HOST machine:"
echo " chown -R $RUN_UID:$RUN_GID /path/to/config"
echo "========================================================"
echo ""
exit 1
}
resolve_runtime_home() {
local runtime_home
@@ -438,15 +405,37 @@ else
# Config is Shelfmark-owned state, so it keeps the thorough repair path.
make_writable "${CONFIG_DIR:-/config}" tree
# Refuse to continue if the config directory is still not writable after repair.
# Fallback to root if config dir is still not writable (common on NAS/Unraid after upgrade from v0.4.0)
CONFIG_PATH=${CONFIG_DIR:-/config}
set +e
test_write "$CONFIG_PATH" >/dev/null 2>&1
config_ok=$?
set -e
if [ $config_ok -ne 0 ]; then
fail_unwritable_config_dir "$CONFIG_PATH"
if [ $config_ok -ne 0 ] && [ "$RUN_UID" != "0" ]; then
config_owner=$(stat -c '%u' "$CONFIG_PATH" 2>/dev/null || echo "unknown")
if [ "$config_owner" = "0" ]; then
echo ""
echo "========================================================"
echo "WARNING: Permission issue detected!"
echo ""
echo "Config directory is owned by root but PUID=$RUN_UID."
echo "This typically happens after upgrading from v0.4.0 where"
echo "PUID/PGID settings were not respected."
echo ""
echo "Falling back to running as root to prevent data loss."
echo ""
echo "To fix this permanently, run on your HOST machine:"
echo " chown -R $RUN_UID:$RUN_GID /path/to/config"
echo ""
echo "Then restart the container."
echo "========================================================"
echo ""
RUN_UID=0
RUN_GID=0
USERNAME=root
TARGET_USER_SPEC="0:0"
fi
fi
fi
@@ -454,7 +443,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_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"
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"
# If DEBUG and not using an external bypass
if [ "$DEBUG" = "true" ] && [ "$USING_EXTERNAL_BYPASSER" != "true" ]; then
@@ -523,12 +512,7 @@ else
fi
RUNTIME_HOME=$(resolve_runtime_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
require_writable_dir "$RUNTIME_HOME" "Home"
if [ "$RUN_AS_NON_ROOT" = "true" ]; then
echo "Startup mode: non-root"
+5 -4
View File
@@ -21,15 +21,16 @@ dependencies = [
"rarfile",
"qbittorrent-api",
"transmission-rpc",
"authlib>=1.7.2,<1.8",
"apprise>=1.10.0",
"authlib>=1.7.0,<1.8",
"apprise>=1.9.0",
"Pillow>=11.0.0",
]
[project.optional-dependencies]
browser = [
"pyvirtualdisplay",
"pyautogui",
"seleniumbase==4.48.4",
"seleniumbase==4.48.2",
"python-xlib",
]
@@ -40,7 +41,7 @@ dev = [
"pytest",
"pytest-cov",
"pytest-xdist>=3.8.0",
"ruff==0.15.12",
"ruff==0.15.11",
"vulture>=2.14",
]
+13 -23
View File
@@ -172,24 +172,6 @@ def _generate_bootstrap_env_docs() -> list[str]:
"type": "string (path)",
"default": "/auth/app.db",
},
{
"name": "HIDE_LOCAL_AUTH",
"description": "Hide the username/password login form when OIDC is active.",
"type": "boolean",
"default": "false",
},
{
"name": "DISABLE_LOCAL_AUTH",
"description": "Disable username/password login and remove the local-admin prerequisite for OIDC. Implies HIDE_LOCAL_AUTH; with AUTH_METHOD=builtin, everyone is locked out until auth env vars are changed.",
"type": "boolean",
"default": "false",
},
{
"name": "OIDC_AUTO_REDIRECT",
"description": "Automatically redirect to the OIDC provider instead of showing the login page.",
"type": "boolean",
"default": "false",
},
{
"name": "DOCKERMODE",
"description": "Indicates the application is running inside a Docker container.",
@@ -207,7 +189,7 @@ def _generate_bootstrap_env_docs() -> list[str]:
lines = [
"## Bootstrap Configuration",
"",
"These environment variables are used at startup before the settings system loads. They typically configure paths, server settings, and authentication startup behavior.",
"These environment variables are used at startup before the settings system loads. They typically configure paths and server settings.",
"",
"| Variable | Description | Type | Default |",
"|----------|-------------|------|---------|",
@@ -328,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 iter_value_fields
from shelfmark.core.settings_registry import ActionButton, CustomComponentField, HeadingField
lines = []
@@ -341,9 +323,17 @@ def _generate_tab_docs(tab: Any, group_prefix: str | None = None) -> list[str]:
lines.append("")
# Collect env-supported fields
env_fields = [
field for field in iter_value_fields(tab) if getattr(field, "env_supported", True)
]
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)
if not env_fields:
lines.append("_No environment variables for this section._")
-1
View File
@@ -121,7 +121,6 @@ SESSION_COOKIE_SECURE_ENV = os.getenv("SESSION_COOKIE_SECURE", "false")
SESSION_COOKIE_NAME = "shelfmark_session"
CWA_DB_PATH = _resolve_cwa_db_path()
HIDE_LOCAL_AUTH = string_to_bool(os.getenv("HIDE_LOCAL_AUTH", "false"))
DISABLE_LOCAL_AUTH = string_to_bool(os.getenv("DISABLE_LOCAL_AUTH", "false"))
OIDC_AUTO_REDIRECT = string_to_bool(os.getenv("OIDC_AUTO_REDIRECT", "false"))
+6 -12
View File
@@ -76,7 +76,7 @@ def _test_oidc_connection(current_values: dict[str, Any] | None = None) -> dict[
@register_settings("security", "Security", icon="shield", order=5)
def security_settings() -> list[SettingsField]:
"""Security and authentication settings."""
from shelfmark.config.env import CWA_DB_PATH, DISABLE_LOCAL_AUTH
from shelfmark.config.env import CWA_DB_PATH
cwa_db_available = CWA_DB_PATH is not None and CWA_DB_PATH.exists()
@@ -108,17 +108,11 @@ def security_settings() -> list[SettingsField]:
),
show_when=_auth_condition("builtin"),
),
*(
[]
if DISABLE_LOCAL_AUTH
else [
CustomComponentField(
key="oidc_admin_requirement",
component="oidc_admin_hint",
label="A local admin account is required before OIDC can be enabled.",
show_when=_auth_condition("oidc"),
),
]
CustomComponentField(
key="oidc_admin_requirement",
component="oidc_admin_hint",
label="A local admin account is required before OIDC can be enabled.",
show_when=_auth_condition("oidc"),
),
*(
[]
+1 -2
View File
@@ -4,7 +4,6 @@ import os
from pathlib import Path
from typing import TYPE_CHECKING, Any
from shelfmark.config.env import DISABLE_LOCAL_AUTH
from shelfmark.core.user_db import UserDB
from shelfmark.core.utils import normalize_http_url
from shelfmark.download.network import get_ssl_verify
@@ -79,7 +78,7 @@ def on_save_security(
auth_method = str(effective_values.get("AUTH_METHOD", "") or "").strip().lower()
if auth_method == "oidc":
if not DISABLE_LOCAL_AUTH and not _has_local_password_admin():
if not _has_local_password_admin():
return {"error": True, "message": _OIDC_LOCKOUT_MESSAGE, "values": normalized_values}
missing_fields = _get_missing_oidc_required_fields(effective_values)
+6 -30
View File
@@ -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,14 +966,7 @@ def download_settings() -> list[SettingsField]:
_naming_template_field(
key="TEMPLATE_RENAME",
label="Naming Template",
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."
),
description="Filename template for single-file book downloads.",
default="{Author} - {Title} ({Year})",
placeholder="{Author} - {Title} ({Year})",
show_when=[
@@ -985,12 +978,7 @@ def download_settings() -> list[SettingsField]:
_naming_template_field(
key="TEMPLATE_ORGANIZE",
label="Path Template",
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."
),
description="Folder and filename template for book downloads.",
default="{Author}/{Title} ({Year})",
placeholder="{Author}/{Series/}{Title} ({Year})",
show_when=[
@@ -1248,14 +1236,7 @@ def download_settings() -> list[SettingsField]:
_naming_template_field(
key="TEMPLATE_AUDIOBOOK_RENAME",
label="Naming Template",
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."
),
description="Filename template for single-file audiobook downloads.",
default="{Author} - {Title}",
placeholder="{Author} - {Title}{ - Part }{PartNumber}",
show_when={"field": "FILE_ORGANIZATION_AUDIOBOOK", "value": "rename"},
@@ -1265,13 +1246,8 @@ def download_settings() -> list[SettingsField]:
_naming_template_field(
key="TEMPLATE_AUDIOBOOK_ORGANIZE",
label="Path Template",
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}",
description="Folder and filename template for audiobook downloads.",
default="{Author}/{Title}",
placeholder="{Author}/{Series/}{Title}{ - Part }{PartNumber}",
show_when={"field": "FILE_ORGANIZATION_AUDIOBOOK", "value": "organize"},
universal_only=True,
+2 -6
View File
@@ -71,16 +71,14 @@ def determine_auth_mode(
cwa_db_path: object | None,
*,
has_local_admin: bool = True,
disable_local_auth: bool = False,
) -> str:
"""Determine active auth mode from security config and runtime prerequisites."""
auth_mode = security_config.get("AUTH_METHOD", "none")
local_admin_available = has_local_admin or disable_local_auth
if auth_mode == AUTH_SOURCE_CWA and cwa_db_path:
return AUTH_SOURCE_CWA
if auth_mode == AUTH_SOURCE_BUILTIN and local_admin_available:
if auth_mode == AUTH_SOURCE_BUILTIN and has_local_admin:
return AUTH_SOURCE_BUILTIN
if auth_mode == AUTH_SOURCE_PROXY and security_config.get("PROXY_AUTH_USER_HEADER"):
@@ -88,7 +86,7 @@ def determine_auth_mode(
if (
auth_mode == AUTH_SOURCE_OIDC
and local_admin_available
and has_local_admin
and security_config.get("OIDC_DISCOVERY_URL")
and security_config.get("OIDC_CLIENT_ID")
):
@@ -104,7 +102,6 @@ def load_active_auth_mode(
) -> str:
"""Resolve active auth mode using current security config and runtime prerequisites."""
try:
from shelfmark.config.env import DISABLE_LOCAL_AUTH
from shelfmark.core.config import config as app_config
security_config = {
@@ -117,7 +114,6 @@ def load_active_auth_mode(
security_config,
cwa_db_path,
has_local_admin=has_local_password_admin(user_db),
disable_local_auth=DISABLE_LOCAL_AUTH,
)
except ImportError, OSError, RuntimeError, TypeError, ValueError, sqlite3.Error:
return "none"
+182 -70
View File
@@ -8,7 +8,7 @@ import time
from http import HTTPStatus
from io import BytesIO
from typing import TYPE_CHECKING, Any
from urllib.parse import urljoin, urlparse
from urllib.parse import urlparse
import requests
@@ -39,7 +39,6 @@ FETCH_HEADERS = {
# Maximum image size to fetch (5 MB)
MAX_IMAGE_SIZE = 5 * 1024 * 1024
MAX_REDIRECTS = 5
# Negative cache TTL (for failed fetches) - 1 hour
NEGATIVE_CACHE_TTL = 3600
@@ -50,6 +49,9 @@ 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:
@@ -73,6 +75,164 @@ 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."""
@@ -483,85 +643,29 @@ class ImageCacheService:
}
@staticmethod
def _prepare_safe_url(url: str) -> str | None:
"""Prepare and validate a URL before fetching it."""
if "\\" in url or any(ord(char) < 32 for char in url):
return None
def _is_safe_url(url: str) -> bool:
"""Check that a URL is safe to fetch (no SSRF to internal resources)."""
try:
prepared = requests.Request("GET", url).prepare()
prepared_url = prepared.url
if not isinstance(prepared_url, str):
return None
parsed = urlparse(prepared_url)
parsed = urlparse(url)
hostname = parsed.hostname
except requests.exceptions.RequestException, ValueError:
return None
if not prepared_url:
return None
if "\\" in prepared_url or any(ord(char) < 32 for char in prepared_url):
return None
netloc_lower = parsed.netloc.lower()
if "%2f" in netloc_lower or "%5c" in netloc_lower:
return None
except ValueError:
return False
if parsed.scheme not in ("http", "https"):
return None
return False
if not hostname:
return None
return False
try:
resolved = socket.getaddrinfo(hostname, None)
for _, _, _, _, sockaddr in resolved:
ip = ipaddress.ip_address(sockaddr[0])
if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_reserved:
return None
return False
except socket.gaierror, ValueError:
return None
return False
return prepared_url
@staticmethod
def _is_safe_url(url: str) -> bool:
"""Check that a URL is safe to fetch (no SSRF to internal resources)."""
return ImageCacheService._prepare_safe_url(url) is not None
def _fetch_safe_response(self, url: str) -> requests.Response | None:
"""Fetch a URL after validating the initial URL and each redirect."""
current_url = self._prepare_safe_url(url)
if not current_url:
logger.warning("Blocked request to disallowed URL: %s", url)
return None
for _ in range(MAX_REDIRECTS + 1):
response = requests.get(
current_url,
timeout=(5, 10),
headers=FETCH_HEADERS,
stream=True,
verify=get_ssl_verify(current_url),
allow_redirects=False,
)
if not response.is_redirect:
return response
location = response.headers.get("location")
response.close()
if not location:
return None
redirect_url = urljoin(current_url, location)
next_url = self._prepare_safe_url(redirect_url)
if not next_url:
logger.warning("Blocked redirect to disallowed URL: %s", redirect_url)
return None
current_url = next_url
return None
return True
def fetch_and_cache(self, cache_id: str, url: str) -> tuple[bytes, str] | None:
"""Fetch an image from URL and cache it.
@@ -576,9 +680,17 @@ class ImageCacheService:
"""
cached_data: tuple[bytes, str] | None = None
try:
response = self._fetch_safe_response(url)
if response is None:
if not self._is_safe_url(url):
logger.warning("Blocked request to disallowed URL: %s", url)
return None
response = requests.get(
url,
timeout=(5, 10),
headers=FETCH_HEADERS,
stream=True,
verify=get_ssl_verify(url),
)
response.raise_for_status()
# Validate content type
+1 -11
View File
@@ -73,16 +73,6 @@ def _has_username_or_email(claims: dict[str, Any]) -> bool:
return False
def _is_email_verified(claims: dict[str, Any]) -> bool:
"""Return True when claims explicitly mark the email address as verified."""
email_verified = claims.get("email_verified")
if isinstance(email_verified, bool):
return email_verified
if isinstance(email_verified, str):
return email_verified.strip().lower() == "true"
return False
def _login_error_url(message: str) -> str:
"""Build a login URL (with script_root) that includes an OIDC error message."""
script_root = request.script_root.rstrip("/")
@@ -305,7 +295,7 @@ def register_oidc_routes(app: Flask, user_db: UserDB) -> None:
if admin_group and use_admin_group:
is_admin = admin_group in groups
allow_email_link = bool(user_info.get("email")) and _is_email_verified(claims)
allow_email_link = bool(user_info.get("email"))
user = provision_oidc_user(
user_db,
user_info,
+4 -49
View File
@@ -10,7 +10,7 @@ A mapping rewrites a remote path prefix into a local path prefix.
from __future__ import annotations
from dataclasses import dataclass
from pathlib import Path, PureWindowsPath
from pathlib import Path
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@@ -50,42 +50,6 @@ def _normalize_host(host: str) -> str:
return str(host or "").strip().lower()
def _is_relative_to(path: Path, prefix: Path) -> bool:
try:
path.relative_to(prefix)
except ValueError:
return False
return True
def _join_contained_path(local_prefix: str, remainder: str) -> Path | None:
local_path = Path(local_prefix)
if remainder:
remainder_path = Path(remainder)
windows_remainder_path = PureWindowsPath(remainder)
if (
remainder_path.is_absolute()
or windows_remainder_path.is_absolute()
or ".." in remainder_path.parts
or ".." in windows_remainder_path.parts
):
return None
remapped = local_path / remainder_path
else:
remapped = local_path
resolved_local_path = local_path.resolve(strict=False)
resolved_remapped = remapped.resolve(strict=False)
if not _is_relative_to(resolved_remapped, resolved_local_path):
return None
return remapped
def parse_remote_path_mappings(value: object) -> list[RemotePathMapping]:
"""Parse configured remote-path mapping rows into normalized mappings."""
if not value or not isinstance(value, list):
@@ -117,12 +81,8 @@ def remap_remote_to_local_with_match(
mappings: Iterable[RemotePathMapping],
host: str,
remote_path: str | Path,
) -> tuple[Path | None, bool]:
"""Remap a remote path and report whether a configured mapping matched.
Returns ``(None, True)`` when a mapping prefix matched but the remainder was
unsafe to join under the local prefix.
"""
) -> tuple[Path, bool]:
"""Remap a remote path and report whether a configured mapping matched."""
host_normalized = _normalize_host(host)
remote_normalized = _normalize_prefix(str(remote_path))
@@ -159,10 +119,7 @@ def remap_remote_to_local_with_match(
remainder = remainder.removeprefix("/")
remapped = _join_contained_path(local_prefix, remainder)
if remapped is None:
return None, True
remapped = Path(local_prefix) / remainder if remainder else Path(local_prefix)
return remapped, True
return Path(remote_normalized), False
@@ -177,8 +134,6 @@ def remap_remote_to_local(
host=host,
remote_path=remote_path,
)
if remapped is None:
return Path(str(remote_path))
return remapped
-28
View File
@@ -220,26 +220,6 @@ def _normalize_release_result_request_payload(
return "release", normalized_release_data
def _validate_release_source_matches_policy_context(
*,
source: str,
release_data: object,
) -> None:
if not isinstance(release_data, dict):
return
release_source = normalize_source(release_data.get("source"))
if release_source in {"", "*"} or release_source == source:
return
msg = "Policy context source must match release_data.source"
raise RequestServiceError(
msg,
status_code=400,
code="policy_source_mismatch",
)
def _resolve_request_title(request_row: dict[str, Any]) -> str:
return _resolve_title_from_book_data(request_row.get("book_data"))
@@ -337,10 +317,6 @@ def _prepare_request_create_arguments(
content_type = normalize_content_type(
context.get("content_type") or data.get("content_type") or book_data.get("content_type")
)
_validate_release_source_matches_policy_context(
source=source,
release_data=release_data,
)
request_level, release_data = _normalize_release_result_request_payload(
source=source,
request_level=request_level,
@@ -348,10 +324,6 @@ def _prepare_request_create_arguments(
release_data=release_data,
content_type=content_type,
)
_validate_release_source_matches_policy_context(
source=source,
release_data=release_data,
)
global_settings, user_settings, effective, requests_enabled = _resolve_effective_policy(
user_db,
+4 -4
View File
@@ -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
+5 -24
View File
@@ -1,21 +1,18 @@
"""Archive extraction utilities for downloaded book archives."""
import shutil
import tempfile
import zipfile
from pathlib import Path
from typing import TYPE_CHECKING, cast
from typing import TYPE_CHECKING
from shelfmark.core.logger import setup_logger
from shelfmark.core.utils import is_audiobook as check_audiobook
from shelfmark.download.fs import atomic_move
from shelfmark.download.fs import atomic_write
from shelfmark.download.postprocess.policy import (
get_supported_audiobook_formats,
get_supported_formats,
)
logger = setup_logger(__name__)
_ARCHIVE_COPY_CHUNK_SIZE = 1024 * 1024
if TYPE_CHECKING:
import rarfile
@@ -211,25 +208,9 @@ def _extract_files_from_archive(archive: ArchiveType, output_dir: Path) -> list[
logger.warning("Path traversal attempt blocked: %r", info.filename)
continue
temp_path: Path | None = None
try:
with (
archive.open(info) as src,
tempfile.NamedTemporaryFile(
dir=output_dir,
prefix=".shelfmark-extract-",
suffix=".tmp",
delete=False,
) as temp_file,
):
temp_path = Path(temp_file.name)
shutil.copyfileobj(src, temp_file, length=_ARCHIVE_COPY_CHUNK_SIZE)
final_path = atomic_move(cast("Path", temp_path), target_path)
except Exception:
if temp_path is not None:
temp_path.unlink(missing_ok=True)
raise
with archive.open(info) as src:
data = src.read()
final_path = atomic_write(target_path, data)
extracted_files.append(final_path)
logger.debug("Extracted: %s", filename)
+1 -25
View File
@@ -276,18 +276,7 @@ class ExternalClientHandler(DownloadHandler, ABC):
remote_path=source_path_obj,
)
if matched_mapping:
if remapped is None:
logger.warning(
"Refusing to delete download data for %s %s because remote path mapping rejected unsafe path: %s",
client.name,
download_id,
source_path_obj,
)
return
delete_path = remapped
else:
delete_path = source_path_obj
delete_path = remapped if matched_mapping else source_path_obj
if str(delete_path) in ("", "/"):
logger.warning(
@@ -446,19 +435,6 @@ class ExternalClientHandler(DownloadHandler, ABC):
)
if matched_mapping:
if remapped is None:
message = (
f"Remote path mapping rejected unsafe path '{source_path_obj}'. "
f"Check Settings > Advanced > Remote Path Mappings."
)
failure_log = "Remote path mapping rejected unsafe path for %s (%s): %s"
failure_args = (client.name, download_id, source_path_obj)
if log_details:
logger.error(failure_log, *failure_args)
else:
logger.debug(failure_log, *failure_args)
return None, message
remapped_exists, remapped_error = _probe_completed_path(remapped)
if log_details:
+6 -37
View File
@@ -2,10 +2,9 @@
from __future__ import annotations
import os
import time
from http import HTTPStatus
from pathlib import Path, PurePosixPath, PureWindowsPath
from pathlib import Path
from types import SimpleNamespace
from typing import NoReturn, TypedDict
@@ -47,13 +46,6 @@ _HTTP_STATUS_NOT_FOUND = HTTPStatus.NOT_FOUND
_ONE_WEEK_IN_SECONDS = 604800
class _UnsafeQBittorrentPath:
pass
_UNSAFE_QBITTORRENT_PATH = _UnsafeQBittorrentPath()
class _QBittorrentAddKwargs(TypedDict, total=False):
rename: str
category: str
@@ -144,24 +136,6 @@ def _is_explicit_add_failure(raw_result: object) -> bool:
return normalized in {"fail", "fails", "error", "errors"}
def _build_qbittorrent_child_path(base_path: object, child_path: object) -> str | None:
"""Build a qBittorrent-reported child path without allowing escape from base."""
if not isinstance(base_path, str) or not base_path:
return None
if not isinstance(child_path, str) or not child_path:
return None
child = child_path.replace("\\", "/")
posix_child = PurePosixPath(child)
windows_child = PureWindowsPath(child_path)
if posix_child.is_absolute() or windows_child.is_absolute() or windows_child.drive:
return None
if any(part == ".." for part in posix_child.parts):
return None
return os.path.normpath(str(Path(base_path) / child))
@register_client("torrent")
class QBittorrentClient(DownloadClient):
"""qBittorrent download client."""
@@ -655,18 +629,16 @@ class QBittorrentClient(DownloadClient):
download_id = getattr(torrent, "hash", "")
if isinstance(download_id, str) and download_id:
derived = self._derive_download_path_from_files(download_id)
if derived and not isinstance(derived, _UnsafeQBittorrentPath):
if derived:
return derived
# Legacy fallback: save_path + name (for older clients/emulators)
return _build_qbittorrent_child_path(
return self._build_path(
getattr(torrent, "save_path", ""),
getattr(torrent, "name", ""),
)
def _derive_download_path_from_files(
self, download_id: str
) -> str | _UnsafeQBittorrentPath | None:
def _derive_download_path_from_files(self, download_id: str) -> str | None:
"""Derive completed download path using `/torrents/properties` + `/torrents/files`.
This mirrors how common automation apps derive the path when
@@ -713,12 +685,9 @@ class QBittorrentClient(DownloadClient):
first_name_norm = first_name.replace("\\", "/")
top_level = first_name_norm.split("/", 1)[0]
if not top_level:
return _UNSAFE_QBITTORRENT_PATH
return None
derived = _build_qbittorrent_child_path(save_path, top_level)
if derived is None:
return _UNSAFE_QBITTORRENT_PATH
return os.path.normpath(derived)
return os.path.normpath(str(Path(save_path) / top_level))
except _QBITTORRENT_CLIENT_ERRORS as e:
logger.debug(
"qBittorrent could not derive path from files: %s: %s",
+6 -41
View File
@@ -33,24 +33,6 @@ _SABNZBD_CLIENT_ERRORS = (
_SabnzbdRequestParam = str | int | float | bool
def _url_origin(value: str) -> tuple[str, str, int] | None:
try:
parsed = urlparse(value)
port = parsed.port
except ValueError:
return None
scheme = parsed.scheme.lower()
hostname = (parsed.hostname or "").lower()
if scheme not in {"http", "https"} or not hostname:
return None
if port is None:
port = 443 if scheme == "https" else 80
return scheme, hostname, port
def _parse_eta(eta_str: str) -> int | None:
"""Parse SABnzbd ETA string (format: 'H:MM:SS') to seconds."""
if not eta_str or eta_str == "0:00:00":
@@ -238,18 +220,6 @@ class SABnzbdClient(DownloadClient):
response.raise_for_status()
return response.content
def _can_prefetch_nzb_url(self, url: str) -> bool:
target_origin = _url_origin(url)
if target_origin is None:
return False
for key in ("PROWLARR_URL", "NEWZNAB_URL"):
trusted_url = normalize_http_config_url(config.get(key, ""))
if trusted_url and _url_origin(trusted_url) == target_origin:
return True
return False
def _get_prowlarr_headers(self, url: str) -> dict:
# TODO(shelfmark): Move this source-specific Prowlarr auth handling into a source hook.
api_key = str(config.get("PROWLARR_API_KEY", "") or "").strip()
@@ -356,20 +326,15 @@ class SABnzbdClient(DownloadClient):
try:
logger.debug("Adding NZB to SABnzbd: %s", name)
if self._can_prefetch_nzb_url(url):
nzb_filename = self._build_nzb_filename(name, url)
nzb_content = self._fetch_nzb_content(url)
result = self._api_post_file(nzb_content, nzb_filename, name, resolved_category)
nzo_id = self._extract_nzo_id(result)
logger.info("Added NZB to SABnzbd: %s", nzo_id)
else:
logger.info("Skipping SABnzbd addfile prefetch for untrusted NZB URL")
nzo_id = ""
nzb_filename = self._build_nzb_filename(name, url)
nzb_content = self._fetch_nzb_content(url)
result = self._api_post_file(nzb_content, nzb_filename, name, resolved_category)
nzo_id = self._extract_nzo_id(result)
logger.info("Added NZB to SABnzbd: %s", nzo_id)
except _SABNZBD_CLIENT_ERRORS as e:
logger.warning("SABnzbd addfile failed, falling back to addurl: %s", e)
else:
if nzo_id:
return nzo_id
return nzo_id
try:
result = self._api_call(
+1 -42
View File
@@ -7,13 +7,12 @@ import hashlib
import re
from binascii import Error as BinasciiError
from dataclasses import dataclass
from urllib.parse import ParseResult, parse_qs, urljoin, urlparse
from urllib.parse import parse_qs, urljoin, urlparse
import requests
from shelfmark.core.config import config
from shelfmark.core.logger import setup_logger
from shelfmark.core.utils import normalize_http_url
from shelfmark.download.network import get_ssl_verify
logger = setup_logger(__name__)
@@ -33,7 +32,6 @@ _TORRENT_FETCH_ERRORS = (
ValueError,
)
_TORRENT_PARSE_ERRORS = (IndexError, KeyError, TypeError, ValueError)
_TRUSTED_TORRENT_FETCH_URL_CONFIG_KEYS = ("PROWLARR_URL", "NEWZNAB_URL")
type BencodeValue = dict[str | bytes, BencodeValue] | list[BencodeValue] | int | bytes | str
@@ -95,9 +93,6 @@ def extract_torrent_info(
# Not a magnet - try to fetch and parse the .torrent file
if not fetch_torrent:
return TorrentInfo(info_hash=expected_hash, torrent_data=None, is_magnet=False)
if not _is_trusted_torrent_fetch_url(url):
logger.debug("Skipping torrent prefetch for untrusted URL: %s...", url[:80])
return TorrentInfo(info_hash=expected_hash, torrent_data=None, is_magnet=False)
headers: dict[str, str] = {"Accept": "application/x-bittorrent"}
# TODO(shelfmark): Move this source-specific Prowlarr auth handling into a source hook.
@@ -138,12 +133,6 @@ def extract_torrent_info(
is_magnet=True,
magnet_url=redirect_url,
)
if not _is_trusted_torrent_fetch_url(redirect_url):
logger.debug(
"Skipping torrent prefetch redirect to untrusted URL: %s...",
redirect_url[:80],
)
return TorrentInfo(info_hash=expected_hash, torrent_data=None, is_magnet=False)
# Not a magnet redirect, follow it manually
logger.debug("Following redirect to: %s...", redirect_url[:80])
resp = requests.get(
@@ -183,36 +172,6 @@ def extract_torrent_info(
return TorrentInfo(info_hash=expected_hash, torrent_data=None, is_magnet=False)
def _is_trusted_torrent_fetch_url(url: str) -> bool:
parsed = urlparse(url)
origin = _url_origin(parsed)
if origin is None:
return False
for key in _TRUSTED_TORRENT_FETCH_URL_CONFIG_KEYS:
configured_url = str(config.get(key, "") or "").strip()
if not configured_url:
continue
configured_origin = _url_origin(urlparse(normalize_http_url(configured_url)))
if configured_origin == origin:
return True
return False
def _url_origin(parsed_url: ParseResult) -> tuple[str, str, int] | None:
scheme = parsed_url.scheme.lower()
if scheme not in {"http", "https"}:
return None
hostname = parsed_url.hostname
if not hostname:
return None
default_port = 443 if scheme == "https" else 80
return (scheme, hostname.lower(), parsed_url.port or default_port)
def parse_transmission_url(url: str) -> tuple[str, str, int, str]:
"""Parse Transmission URL into (protocol, host, port, path)."""
parsed = urlparse(url)
+12 -16
View File
@@ -229,10 +229,6 @@ def _is_permission_error(e: Exception) -> bool:
return isinstance(e, PermissionError) or (isinstance(e, OSError) and e.errno == errno.EPERM)
def _should_fallback_to_content_copy(error: Exception) -> bool:
return _is_permission_error(error) or (isinstance(error, OSError) and error.errno == errno.EIO)
def _system_op(op: str, source: Path, dest: Path) -> None:
"""Execute system command (mv or cp) as final fallback."""
logger.warning("Attempting system %s as final fallback: %s -> %s", op, source, dest)
@@ -467,9 +463,9 @@ def atomic_move(source_path: Path, dest_path: Path, max_attempts: int = 100) ->
try:
run_blocking_io(shutil.copy2, str(source_path), str(temp_path))
except (PermissionError, OSError) as copy_error:
if _should_fallback_to_content_copy(copy_error):
if _is_permission_error(copy_error):
logger.debug(
"copy2 failed during move-copy, falling back to copyfile (%s -> %s): %s",
"Permission error during move-copy, falling back to copyfile (%s -> %s): %s",
source_path,
temp_path,
copy_error,
@@ -587,7 +583,7 @@ def atomic_hardlink(source_path: Path, dest_path: Path, max_attempts: int = 100)
error=e,
)
if permission_error or _hardlink_not_supported(e):
logger.warning(
logger.debug(
"Hardlink failed (%s), falling back to copy: %s -> %s",
e,
source_path,
@@ -635,16 +631,16 @@ def atomic_copy(source_path: Path, dest_path: Path, max_attempts: int = 100) ->
try:
run_blocking_io(shutil.copy2, str(source_path), str(temp_path))
except (PermissionError, OSError) as e:
if _should_fallback_to_content_copy(e):
if _is_permission_error(e):
log_transfer_permission_context(
"atomic_copy",
source=source_path,
dest=temp_path,
error=e,
)
# Handle NFS permission errors immediately here
if _is_permission_error(e):
log_transfer_permission_context(
"atomic_copy",
source=source_path,
dest=temp_path,
error=e,
)
logger.debug(
"copy2 failed during copy, falling back to copyfile (%s -> %s): %s",
"Permission error during copy, falling back to copyfile (%s -> %s): %s",
source_path,
temp_path,
e,
+12 -43
View File
@@ -29,7 +29,6 @@ from shelfmark.download.postprocess.pipeline import is_torrent_source, safe_clea
from shelfmark.download.postprocess.router import post_process_download
from shelfmark.release_sources import (
get_handler,
get_source,
get_source_display_name,
)
@@ -109,13 +108,6 @@ def _parse_release_search_mode(value: object) -> SearchMode:
raise ValueError(msg)
def _source_unavailable_message(source_name: str) -> str | None:
source = get_source(source_name)
if source.is_available():
return None
return f"{source.display_name} is unavailable. Enable and configure the source in Settings."
def _optional_number(value: object) -> float | None:
if isinstance(value, bool):
return float(value)
@@ -146,6 +138,13 @@ def _optional_positive_int(value: object) -> int | None:
return parsed if parsed > 0 else None
def _seed_time_seconds_to_minutes(value: object) -> int | None:
seed_time_seconds = _optional_positive_int(value)
if seed_time_seconds is None:
return None
return (seed_time_seconds + 59) // 60
def _config_float(value: object, default: float) -> float:
if isinstance(value, bool) or value is None:
return default
@@ -167,34 +166,19 @@ def _build_retry_resolution_fields(
if not isinstance(extra, dict):
extra = {}
retry_download_url = normalize_optional_text(release_data.get("download_url"))
protocol = normalize_optional_text(release_data.get("protocol"))
source = normalize_optional_text(release_data.get("source"))
if source is not None:
handler = get_handler(source)
source_retry_fields = handler.build_retry_resolution_fields(release_data)
retry_download_url = (
normalize_optional_text(source_retry_fields.get("retry_download_url"))
or retry_download_url
)
protocol = (
normalize_optional_text(source_retry_fields.get("retry_download_protocol")) or protocol
)
ratio_limit = _optional_number(release_data.get("ratio_limit"))
if ratio_limit is None and config.get("PROWLARR_USE_SEED_PREFERENCES", False):
ratio_limit = _optional_number(extra.get("configured_ratio_limit"))
if ratio_limit is None:
ratio_limit = _optional_number(extra.get("minimum_ratio"))
seeding_time_limit_minutes = _optional_positive_int(
release_data.get("seeding_time_limit_minutes")
)
if seeding_time_limit_minutes is None and config.get("PROWLARR_USE_SEED_PREFERENCES", False):
seeding_time_limit_minutes = _optional_positive_int(
extra.get("configured_seed_time_minutes")
)
if seeding_time_limit_minutes is None:
seeding_time_limit_minutes = _seed_time_seconds_to_minutes(extra.get("minimum_seed_time"))
return {
"retry_download_url": retry_download_url,
"retry_download_url": normalize_optional_text(release_data.get("download_url")),
"retry_download_protocol": protocol.lower() if protocol is not None else None,
"retry_release_name": normalize_optional_text(release_data.get("title")),
"retry_expected_hash": normalize_optional_text(
@@ -215,11 +199,6 @@ def queue_release(
"""Add a release to the download queue. Returns (success, error_message)."""
try:
source = release_data["source"]
unavailable_message = _source_unavailable_message(source)
if unavailable_message:
logger.warning("Rejected queue request for unavailable source %s", source)
return False, unavailable_message
extra = release_data.get("extra", {})
raw_request_id = release_data.get("_request_id")
request_id: int | None = None
@@ -611,16 +590,6 @@ def _download_task(task_id: str, cancel_flag: Event) -> str | None:
logger.error("Task not found in queue: %s", task_id)
return None
unavailable_message = _source_unavailable_message(task.source)
if unavailable_message:
logger.warning("Task %s: source unavailable: %s", task_id, unavailable_message)
_capture_task_error(
task,
message=unavailable_message,
exc_type="SourceUnavailable",
)
return None
title_label = task.title or "Unknown title"
logger.info(
"Task %s: starting download (%s) - %s",
+15 -11
View File
@@ -13,6 +13,7 @@ from shelfmark.core.naming import (
build_library_path,
derive_primary_title,
parse_naming_template,
same_filesystem,
sanitize_filename,
)
from shelfmark.core.utils import is_audiobook as check_audiobook
@@ -38,7 +39,10 @@ _TRANSFER_PROCESS_ERRORS = (AttributeError, KeyError, OSError, RuntimeError, Typ
def should_hardlink(task: DownloadTask) -> bool:
"""Check if hardlinking is enabled for this torrent-backed task."""
"""Check if hardlinking is enabled for this task (Prowlarr torrents only)."""
if task.source != "prowlarr":
return False
if not task.original_download_path:
return False
@@ -92,21 +96,21 @@ def resolve_hardlink_source(
if hardlink_enabled and task.original_download_path:
hardlink_source = Path(task.original_download_path)
hardlink_source_exists = run_blocking_io(hardlink_source.exists)
if hardlink_source_exists:
if (
destination
and hardlink_source_exists
and run_blocking_io(same_filesystem, hardlink_source, destination)
):
use_hardlink = True
source_path = hardlink_source
logger.info(
"Hardlink enabled for task %s; attempting link from %s to %s",
task.task_id,
elif hardlink_source_exists:
logger.warning(
"Cannot hardlink: %s and %s are on different filesystems. Falling back to copy. To fix: ensure torrent client downloads to same filesystem as destination.",
hardlink_source,
destination,
)
else:
logger.warning(
"Hardlink enabled for task %s, but source path does not exist: %s",
task.task_id,
hardlink_source,
)
if status_callback:
status_callback("resolving", "Cannot hardlink (different filesystems), using copy")
return TransferPlan(
source_path=source_path,
+65 -127
View File
@@ -26,7 +26,6 @@ from shelfmark.config.env import (
BUILD_VERSION,
CONFIG_DIR,
CWA_DB_PATH,
DISABLE_LOCAL_AUTH,
FLASK_HOST,
FLASK_PORT,
HIDE_LOCAL_AUTH,
@@ -1478,43 +1477,6 @@ def _download_row_owned_by_actor(
return False
def _resolve_queue_actor() -> tuple[bool, int | None, str | None, Response | None]:
is_admin, db_user_id, can_access_status = _resolve_status_scope()
actor_username = session.get("user_id")
normalized_actor_username = actor_username if isinstance(actor_username, str) else None
if not is_admin and (not can_access_status or db_user_id is None):
return (
is_admin,
db_user_id,
normalized_actor_username,
jsonify({"error": "User identity unavailable", "code": "user_identity_unavailable"}),
)
return is_admin, db_user_id, normalized_actor_username, None
def _queue_task_visible_to_actor(
task_id: str,
*,
is_admin: bool,
actor_user_id: int | None,
actor_username: str | None,
) -> bool:
if is_admin:
return True
task = backend.book_queue.get_task(task_id)
if task is None:
return False
return _task_owned_by_actor(
task,
actor_user_id=actor_user_id,
actor_username=actor_username,
)
backend.book_queue.set_queue_hook(_record_download_queued)
backend.book_queue.set_terminal_status_hook(_record_download_terminal_snapshot)
@@ -1624,7 +1586,6 @@ def api_local_download() -> Response | tuple[Response, int]:
@app.route("/api/covers/<cover_id>", methods=["GET"])
@login_required
def api_cover(cover_id: str) -> Response | tuple[Response, int]:
"""Serve a cached book cover image.
@@ -1636,6 +1597,9 @@ 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.
@@ -1645,43 +1609,84 @@ 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 get_image_cache
from shelfmark.core.image_cache import (
build_variant_cache_id,
create_image_variant,
get_image_cache,
normalize_variant_dimension,
normalize_variant_format,
)
# 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
)
# Try to get from cache first
cached = cache.get(cover_id)
if cached:
image_data, content_type = cached
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"] = "HIT"
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)
if cached:
image_data, content_type = cached
return make_cover_response(image_data, content_type, cache_status="HIT")
# Cache miss - get URL from query parameter
encoded_url = request.args.get("url")
if not encoded_url:
return jsonify({"error": "Cover URL not provided"}), 404
original: tuple[bytes, str] | None = cache.get(cover_id) if variant_cache_id else None
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
if original is None:
if not encoded_url:
return jsonify({"error": "Cover URL not provided"}), 404
# 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
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
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"
# 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")
except _IMPORT_OPERATIONAL_ERRORS as e:
logger.error_trace(f"Cover fetch error: {e}")
return jsonify({"error": str(e)}), 500
@@ -1836,22 +1841,6 @@ def api_set_priority(book_id: str) -> Response | tuple[Response, int]:
return jsonify({"error": "Priority not provided"}), 400
priority = int(data["priority"])
is_admin, db_user_id, actor_username, identity_error = _resolve_queue_actor()
if identity_error is not None:
return identity_error, 403
task = backend.book_queue.get_task(book_id)
if task is None:
return jsonify({"error": "Failed to update priority or book not found"}), 404
if not is_admin and not _task_owned_by_actor(
task,
actor_user_id=db_user_id,
actor_username=actor_username,
):
return jsonify({"error": "Forbidden", "code": "download_not_owned"}), 403
success = backend.set_book_priority(book_id, priority)
if success:
@@ -1890,23 +1879,6 @@ def api_reorder_queue() -> Response | tuple[Response, int]:
if not isinstance(priority, int):
return jsonify({"error": f"Invalid priority for book {book_id}"}), 400
is_admin, db_user_id, actor_username, identity_error = _resolve_queue_actor()
if identity_error is not None:
return identity_error, 403
if not is_admin:
owned_book_priorities = {}
for book_id in book_priorities:
task = backend.book_queue.get_task(str(book_id))
if task is None:
continue
if not _task_owned_by_actor(
task, actor_user_id=db_user_id, actor_username=actor_username
):
return jsonify({"error": "Forbidden", "code": "download_not_owned"}), 403
owned_book_priorities[book_id] = book_priorities[book_id]
book_priorities = owned_book_priorities
success = backend.reorder_queue(book_priorities)
if success:
@@ -1928,20 +1900,6 @@ def api_queue_order() -> Response | tuple[Response, int]:
"""
try:
queue_order = backend.get_queue_order()
is_admin, db_user_id, actor_username, identity_error = _resolve_queue_actor()
if identity_error is not None:
return identity_error, 403
if not is_admin:
queue_order = [
item
for item in queue_order
if _queue_task_visible_to_actor(
str(item.get("id", "")),
is_admin=False,
actor_user_id=db_user_id,
actor_username=actor_username,
)
]
return jsonify({"queue": queue_order})
except _OPERATIONAL_ERRORS as e:
logger.error_trace(f"Queue order error: {e}")
@@ -1959,20 +1917,6 @@ def api_active_downloads() -> Response | tuple[Response, int]:
"""
try:
active_downloads = backend.get_active_downloads()
is_admin, db_user_id, actor_username, identity_error = _resolve_queue_actor()
if identity_error is not None:
return identity_error, 403
if not is_admin:
active_downloads = [
task_id
for task_id in active_downloads
if _queue_task_visible_to_actor(
task_id,
is_admin=False,
actor_user_id=db_user_id,
actor_username=actor_username,
)
]
return jsonify({"active_downloads": active_downloads})
except _OPERATIONAL_ERRORS as e:
logger.error_trace(f"Active downloads error: {e}")
@@ -2055,9 +1999,6 @@ def api_login() -> Response | tuple[Response, int]:
if auth_mode == "proxy":
return jsonify({"error": "Proxy authentication is enabled"}), 401
if auth_mode in ("builtin", "oidc") and DISABLE_LOCAL_AUTH:
return jsonify({"error": "Local authentication is disabled"}), 403
if auth_mode == "oidc" and HIDE_LOCAL_AUTH:
return jsonify({"error": "Local authentication is disabled"}), 403
@@ -2287,9 +2228,6 @@ def api_auth_check() -> Response | tuple[Response, int]:
if logout_url:
response_data["logout_url"] = logout_url
if auth_mode in ("builtin", "oidc") and DISABLE_LOCAL_AUTH:
response_data["hide_local_auth"] = True
# Add custom OIDC button label and SSO enforcement flags if configured
if auth_mode == "oidc":
oidc_button_label = app_config.get("OIDC_BUTTON_LABEL", "")
+10 -19
View File
@@ -47,11 +47,6 @@ _HTTP_STATUS_NOT_FOUND = HTTPStatus.NOT_FOUND
GOOGLE_BOOKS_BASE_URL = "https://www.googleapis.com/books/v1"
class _GoogleBooksRequestError(Exception):
"""Raised when Google Books does not return a usable API response."""
# Sort mapping - Google only supports "relevance" and "newest"
SORT_MAPPING: dict[SortOrder, str | None] = {
SortOrder.RELEVANCE: None, # Default, no param needed
@@ -122,10 +117,7 @@ class GoogleBooksProvider(MetadataProvider):
f"{options.query}:{options.search_type.value}:{options.sort.value}:"
f"{options.language}:{options.limit}:{options.page}:{fields_key}"
)
try:
return self._search_cached(cache_key, options)
except _GoogleBooksRequestError:
return []
return self._search_cached(cache_key, options)
@cacheable(
ttl_key="METADATA_CACHE_SEARCH_TTL",
@@ -174,19 +166,18 @@ class GoogleBooksProvider(MetadataProvider):
if options.language:
params["langRestrict"] = options.language
result = self._make_request("/volumes", params)
if result is None:
raise _GoogleBooksRequestError
books: list[BookMetadata] = []
try:
items = result.get("items", [])
for item in items:
book = self._parse_volume(item)
if book:
books.append(book)
result = self._make_request("/volumes", params)
if result:
items = result.get("items", [])
logger.info("Google Books search '%s' returned %s results", query, len(books))
for item in items:
book = self._parse_volume(item)
if book:
books.append(book)
logger.info("Google Books search '%s' returned %s results", query, len(books))
except Exception:
logger.exception("Google Books search error")
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,35 @@
"""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",
]
@@ -0,0 +1,36 @@
"""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
@@ -0,0 +1,105 @@
"""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
@@ -0,0 +1,61 @@
"""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"
@@ -0,0 +1,400 @@
"""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
@@ -0,0 +1,20 @@
"""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."""
@@ -0,0 +1,611 @@
"""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,
)
@@ -0,0 +1,106 @@
"""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)
@@ -0,0 +1,525 @@
"""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
}
}
}
}
}
"""
@@ -0,0 +1,844 @@
"""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
@@ -0,0 +1,154 @@
"""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,
),
]
@@ -0,0 +1,438 @@
"""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)
-4
View File
@@ -390,10 +390,6 @@ class DownloadHandler(ABC):
"""
return
def build_retry_resolution_fields(self, release_data: dict[str, Any]) -> dict[str, Any]:
"""Return private queue-time fields needed for restart-safe retry."""
return {}
@abstractmethod
def cancel(self, task_id: str) -> bool:
"""Cancel an in-progress download."""
@@ -24,8 +24,6 @@ if TYPE_CHECKING:
from shelfmark.core.models import DownloadTask
logger = setup_logger(__name__)
DEFAULT_ABB_HOSTNAME = "audiobookbay.lu"
ALLOWED_DETAIL_URL_SCHEMES = {"https"}
def _resolve_configured_hostname() -> str:
@@ -34,23 +32,6 @@ def _resolve_configured_hostname() -> str:
return normalize_hostname(configured_hostname if isinstance(configured_hostname, str) else "")
def _resolve_allowed_detail_hostname() -> str:
"""Return the ABB hostname allowed for queued detail URLs."""
return _resolve_configured_hostname() or DEFAULT_ABB_HOSTNAME
def _detail_url_matches_host(detail_url: str, hostname: str) -> bool:
"""Return True when a detail URL uses the allowed ABB scheme and host."""
parsed = urlparse(detail_url)
detail_hostname = normalize_hostname(parsed.hostname)
allowed_hostname = normalize_hostname(hostname).lower().rstrip(".")
return (
parsed.scheme.lower() in ALLOWED_DETAIL_URL_SCHEMES
and bool(detail_hostname)
and detail_hostname.lower().rstrip(".") == allowed_hostname
)
@register_handler("audiobookbay")
class AudiobookBayHandler(ExternalClientHandler):
"""Handler for AudiobookBay downloads via configured torrent client."""
@@ -88,14 +69,9 @@ class AudiobookBayHandler(ExternalClientHandler):
logger.warning("Missing details URL for AudiobookBay task: %s", task.task_id)
return None
hostname = _resolve_allowed_detail_hostname()
if not _detail_url_matches_host(detail_url, hostname):
status_callback("error", "Invalid AudiobookBay details URL")
logger.warning(
"Rejected AudiobookBay details URL with invalid scheme or host: %s",
detail_url,
)
return None
hostname = _resolve_configured_hostname()
if not hostname:
hostname = normalize_hostname(urlparse(detail_url).hostname)
status_callback("resolving", "Extracting magnet link")
magnet_link = scraper.extract_magnet_link(detail_url, hostname)
+4 -35
View File
@@ -14,7 +14,7 @@ from typing import TYPE_CHECKING, Self
from shelfmark.core.logger import setup_logger
from .dcc import DCCError, DCCOffer, parse_dcc_send, validate_dcc_endpoint
from .dcc import DCCOffer, parse_dcc_send
if TYPE_CHECKING:
from collections.abc import Iterator
@@ -400,33 +400,6 @@ class IRCClient:
self.send_notice(sender, f"\x01VERSION {self.version}\x01")
logger.debug("Sent VERSION to %s", sender)
@staticmethod
def _sender_nick(msg: IRCMessage) -> str | None:
"""Extract the nick from a message prefix."""
if not msg.prefix:
return None
return msg.prefix.split("!", maxsplit=1)[0]
def _is_allowed_dcc_sender(
self,
msg: IRCMessage,
expected_senders: set[str] | None,
) -> bool:
allowed_senders = expected_senders or self.online_servers
if not allowed_senders:
return True
sender = self._sender_nick(msg)
if sender is None:
logger.warning("Ignoring DCC offer without sender prefix")
return False
normalized_allowed = {nick.casefold() for nick in allowed_senders}
if sender.casefold() not in normalized_allowed:
logger.warning("Ignoring DCC offer from unexpected sender: %s", sender)
return False
return True
def read_messages(self, *, auto_handle: bool = True) -> Iterator[IRCMessage]:
"""Read and yield IRC messages, optionally auto-handling PING/VERSION."""
for line in self._recv_lines():
@@ -449,7 +422,6 @@ class IRCClient:
timeout: float = 60.0,
*,
result_type: bool = False,
expected_senders: set[str] | None = None,
) -> DCCOffer | None:
"""Wait for a DCC SEND offer. Returns None on timeout or no results."""
target_event = IRCEvent.SEARCH_RESULT if result_type else IRCEvent.BOOK_RESULT
@@ -461,15 +433,12 @@ class IRCClient:
return None
if msg.event == target_event:
if not self._is_allowed_dcc_sender(msg, expected_senders):
continue
try:
offer = parse_dcc_send(msg.raw)
validate_dcc_endpoint(offer)
logger.info("Received DCC offer: %s", offer.filename)
except DCCError:
logger.exception("Rejected DCC offer")
continue
except Exception:
logger.exception("Failed to parse DCC")
return None
else:
return offer
+2 -47
View File
@@ -7,8 +7,6 @@ import re
import socket
import struct
from dataclasses import dataclass
from ipaddress import ip_address
from pathlib import PureWindowsPath
from typing import TYPE_CHECKING
from shelfmark.core.logger import setup_logger
@@ -61,10 +59,6 @@ class DCCConnectionError(DCCError):
"""Failed to connect to DCC sender."""
class DCCSecurityError(DCCError):
"""Rejected unsafe DCC offer metadata."""
def int_to_ip(ip_int: int) -> str:
"""Convert 32-bit integer (DCC format) to dotted IP notation."""
packed = struct.pack(">I", ip_int)
@@ -82,53 +76,15 @@ def parse_dcc_send(text: str) -> DCCOffer:
ip_int = int(match.group(2))
port = int(match.group(3))
size = int(match.group(4))
try:
ip = int_to_ip(ip_int)
except struct.error as e:
msg = f"Invalid DCC IP integer: {ip_int}"
raise DCCParseError(msg) from e
return DCCOffer(
filename=safe_dcc_filename(filename),
ip=ip,
filename=filename,
ip=int_to_ip(ip_int),
port=port,
size=size,
)
def safe_dcc_filename(filename: str) -> str:
"""Return a DCC filename that cannot escape its destination directory."""
safe_name = filename.strip()
windows_path = PureWindowsPath(safe_name)
if (
not safe_name
or safe_name in {".", ".."}
or "/" in safe_name
or "\\" in safe_name
or windows_path.drive
):
msg = f"Rejected unsafe DCC filename: {filename!r}"
raise DCCSecurityError(msg)
return safe_name
def validate_dcc_endpoint(offer: DCCOffer) -> None:
"""Reject DCC endpoints that can target local/internal network services."""
if not 1 <= offer.port <= 65535:
msg = f"Rejected invalid DCC port: {offer.port}"
raise DCCSecurityError(msg)
try:
address = ip_address(offer.ip)
except ValueError as e:
msg = f"Rejected invalid DCC IP address: {offer.ip}"
raise DCCSecurityError(msg) from e
if not address.is_global:
msg = f"Rejected non-public DCC endpoint: {offer.ip}"
raise DCCSecurityError(msg)
def download_dcc(
offer: DCCOffer,
dest_path: Path,
@@ -137,7 +93,6 @@ def download_dcc(
timeout: float = 30.0,
) -> None:
"""Download file via DCC protocol to dest_path. Raises DCCError on failure."""
validate_dcc_endpoint(offer)
logger.info("DCC connecting to %s:%s for %s", offer.ip, offer.port, offer.filename)
try:
+3 -16
View File
@@ -12,7 +12,7 @@ from shelfmark.core.logger import setup_logger
from shelfmark.release_sources import DownloadHandler, register_handler
from .connection_manager import connection_manager
from .dcc import DCCError, download_dcc, safe_dcc_filename
from .dcc import DCCError, download_dcc
if TYPE_CHECKING:
from collections.abc import Callable
@@ -23,15 +23,6 @@ if TYPE_CHECKING:
logger = setup_logger(__name__)
def _server_from_download_request(download_request: str) -> str | None:
"""Extract the expected IRC bot nick from a release request line."""
stripped = download_request.strip()
if not stripped.startswith("!"):
return None
server = stripped[1:].split(maxsplit=1)[0]
return server or None
def _config_text(key: str) -> str:
"""Read a string config value with whitespace trimmed."""
value = config.get(key, "")
@@ -81,7 +72,6 @@ class IRCDownloadHandler(DownloadHandler):
"""Download a release via IRC DCC. task.task_id contains the IRC request string."""
download_request = task.task_id
logger.info("IRC download: %s...", download_request[:60])
expected_server = _server_from_download_request(download_request)
# Get IRC settings
server = _config_text("IRC_SERVER")
@@ -133,8 +123,7 @@ class IRCDownloadHandler(DownloadHandler):
# Phase 3: Wait for DCC offer
status_callback("resolving", "Waiting for bot response")
wait_kwargs = {"expected_senders": {expected_server}} if expected_server else {}
offer = client.wait_for_dcc(timeout=120.0, result_type=False, **wait_kwargs)
offer = client.wait_for_dcc(timeout=120.0, result_type=False)
if not offer:
status_callback("error", "No response from bot")
@@ -148,9 +137,7 @@ class IRCDownloadHandler(DownloadHandler):
status_callback("downloading", "")
# Get file extension from offer filename
ext = (
Path(safe_dcc_filename(offer.filename)).suffix.lstrip(".") or task.format or "epub"
)
ext = Path(offer.filename).suffix.lstrip(".") or task.format or "epub"
# Stage to temp directory (lazy import to avoid circular import)
from shelfmark.download.staging import get_staging_path
+3 -4
View File
@@ -30,7 +30,7 @@ from shelfmark.release_sources import (
)
from .connection_manager import connection_manager
from .dcc import DCCError, download_dcc, safe_dcc_filename
from .dcc import DCCError, download_dcc
from .parser import SearchResult, extract_results_from_zip, parse_results_file
logger = setup_logger(__name__)
@@ -227,8 +227,7 @@ class IRCReleaseSource(ReleaseSource):
# Wait for results DCC - this is the long wait
_emit_status(f"Connected to #{channel} - Waiting for results...", phase="searching")
wait_kwargs = {"expected_senders": {search_bot}} if search_bot else {}
offer = client.wait_for_dcc(timeout=60.0, result_type=True, **wait_kwargs)
offer = client.wait_for_dcc(timeout=60.0, result_type=True)
if not offer:
logger.info("No search results received")
_emit_status("No results found", phase="complete")
@@ -248,7 +247,7 @@ class IRCReleaseSource(ReleaseSource):
# Download results file
_emit_status(f"Connected to #{channel} - Downloading results...", phase="downloading")
with tempfile.TemporaryDirectory() as tmpdir:
result_path = Path(tmpdir) / safe_dcc_filename(offer.filename)
result_path = Path(tmpdir) / offer.filename
download_dcc(offer, result_path, timeout=30.0)
# Parse results
@@ -8,7 +8,6 @@ if TYPE_CHECKING:
from shelfmark.core.models import DownloadTask
from shelfmark.core.logger import setup_logger
from shelfmark.core.request_helpers import normalize_optional_text
from shelfmark.download.clients import DownloadClient, get_client, list_configured_clients
from shelfmark.download.clients.base_handler import (
COMPLETED_PATH_MAX_ATTEMPTS as _DEFAULT_COMPLETED_PATH_MAX_ATTEMPTS,
@@ -84,44 +83,6 @@ class NewznabHandler(ExternalClientHandler):
def _completed_path_max_attempts(self) -> int:
return COMPLETED_PATH_MAX_ATTEMPTS
def build_retry_resolution_fields(self, release_data: dict) -> dict:
source_id = normalize_optional_text(release_data.get("source_id"))
if source_id is None:
return {}
result = get_release(source_id)
if result is None:
return {}
return {
"retry_download_url": normalize_optional_text(_get_download_url(result)),
"retry_download_protocol": normalize_optional_text(_get_protocol(result)),
}
@classmethod
def _restore_download_request_from_task(cls, task: DownloadTask) -> DownloadRequest | None:
retry_download_url = normalize_optional_text(getattr(task, "retry_download_url", None))
retry_download_protocol = normalize_optional_text(
getattr(task, "retry_download_protocol", None)
)
if retry_download_url is None or retry_download_protocol is None:
return None
protocol = retry_download_protocol.lower()
if protocol not in {"torrent", "usenet"}:
return None
return DownloadRequest(
url=retry_download_url,
protocol=protocol,
release_name=(
normalize_optional_text(getattr(task, "retry_release_name", None))
or task.title
or "Unknown"
),
expected_hash=normalize_optional_text(getattr(task, "retry_expected_hash", None)),
)
def _resolve_download(
self,
task: DownloadTask,
@@ -129,10 +90,6 @@ class NewznabHandler(ExternalClientHandler):
) -> DownloadRequest | None:
result = get_release(task.task_id)
if not result:
restored_request = self._restore_download_request_from_task(task)
if restored_request is not None:
logger.info("Restored Newznab download request for retry: %s", task.task_id)
return restored_request
logger.warning("Newznab release cache miss: %s", task.task_id)
status_callback("error", "Release not found in cache (may have expired)")
return None
+2 -1
View File
@@ -109,6 +109,7 @@ def _newznab_result_to_release(result: dict, content_type: str = "ebook") -> Rel
if is_freeleech:
add_flag("FreeLeech")
download_url = str(result.get("downloadUrl") or "").strip()
info_url = result.get("infoUrl") or result.get("guid")
return Release(
@@ -119,7 +120,7 @@ def _newznab_result_to_release(result: dict, content_type: str = "ebook") -> Rel
language=None,
size=_parse_size(size_bytes),
size_bytes=size_bytes,
download_url=None,
download_url=download_url or None,
info_url=info_url,
protocol=protocol,
indexer=indexer,
+2 -60
View File
@@ -3,7 +3,7 @@
from collections.abc import Mapping
from contextlib import suppress
from http import HTTPStatus
from typing import Any, TypedDict
from typing import Any
import requests
@@ -11,7 +11,7 @@ from shelfmark.core.logger import setup_logger
from shelfmark.core.utils import normalize_http_url
from shelfmark.download.network import get_ssl_verify
from shelfmark.release_sources.prowlarr.torznab import parse_torznab_xml
from shelfmark.release_sources.prowlarr.utils import coerce_float_like, coerce_int_like
from shelfmark.release_sources.prowlarr.utils import coerce_int_like
logger = setup_logger(__name__)
@@ -27,15 +27,6 @@ _PROWLARR_CLIENT_ERRORS = (
)
class IndexerSeedSettings(TypedDict, total=False):
ratio_limit: float
seeding_time_limit_minutes: int
_INDEXER_FIELD_SEED_RATIO = "torrentBaseSettings.seedRatio"
_INDEXER_FIELD_SEED_TIME_MINUTES = "torrentBaseSettings.seedTime"
def _normalize_json_object(payload: object, *, context: str) -> dict[str, Any]:
"""Return a JSON object payload with string keys or raise on unexpected shapes."""
if not isinstance(payload, Mapping):
@@ -61,19 +52,6 @@ def _normalize_json_object_list(payload: object, *, context: str) -> list[dict[s
return [_normalize_json_object(item, context=context) for item in payload]
def _get_field_value(fields: object, name: str) -> object | None:
if not isinstance(fields, list):
return None
for field in fields:
if not isinstance(field, Mapping):
continue
if field.get("name") == name:
return field.get("value")
return None
class ProwlarrClient:
"""Client for interacting with the Prowlarr API."""
@@ -205,42 +183,6 @@ class ProwlarrClient:
return enriched_ids
def get_indexer_seed_settings(
self, *, restrict_to: list[int] | None = None
) -> dict[int, IndexerSeedSettings]:
"""Return configured per-indexer torrent share limits.
Prowlarr exposes seedTime in minutes, which is also the unit expected by
torrent clients.
"""
settings_by_indexer: dict[int, IndexerSeedSettings] = {}
for idx in self.get_enabled_indexers_detailed():
idx_id_int = coerce_int_like(idx.get("id"))
if idx_id_int is None:
continue
if restrict_to is not None and idx_id_int not in restrict_to:
continue
if str(idx.get("protocol") or "").lower() != "torrent":
continue
fields = idx.get("fields")
ratio_limit = coerce_float_like(_get_field_value(fields, _INDEXER_FIELD_SEED_RATIO))
seeding_time_limit = coerce_int_like(
_get_field_value(fields, _INDEXER_FIELD_SEED_TIME_MINUTES)
)
settings: IndexerSeedSettings = {}
if ratio_limit is not None and ratio_limit > 0:
settings["ratio_limit"] = ratio_limit
if seeding_time_limit is not None and seeding_time_limit > 0:
settings["seeding_time_limit_minutes"] = seeding_time_limit
if settings:
settings_by_indexer[idx_id_int] = settings
return settings_by_indexer
def get_enabled_indexers(self) -> list[dict[str, Any]]:
"""Get enabled indexers with book capability info."""
indexers = self.get_indexers()
+21 -28
View File
@@ -1,6 +1,6 @@
"""Prowlarr download handler - resolves releases and delegates lifecycle to shared clients."""
from typing import TYPE_CHECKING, Any
from typing import TYPE_CHECKING
from shelfmark.core.config import config
from shelfmark.core.logger import setup_logger
@@ -51,11 +51,22 @@ COMPLETED_PATH_RETRY_INTERVAL = _DEFAULT_COMPLETED_PATH_RETRY_INTERVAL
COMPLETED_PATH_MAX_ATTEMPTS = _DEFAULT_COMPLETED_PATH_MAX_ATTEMPTS
def _coerce_positive_minutes(raw_minutes: object) -> int | None:
minutes = coerce_int_like(raw_minutes)
if minutes is None:
def _coerce_seed_time_minutes(raw_seed_time: object) -> int | None:
"""Convert Prowlarr's minimum seed time from seconds to whole minutes."""
if raw_seed_time is None:
return None
return minutes if minutes > 0 else None
seed_time_seconds = coerce_int_like(raw_seed_time)
if seed_time_seconds is None:
logger.warning("Invalid Prowlarr minimumSeedTime value: %r", raw_seed_time)
return None
if seed_time_seconds < 0:
logger.warning("Ignoring negative Prowlarr minimumSeedTime value: %s", seed_time_seconds)
return None
# Round up so we never under-seed when a tracker uses a non-minute boundary.
return (seed_time_seconds + 59) // 60
@register_handler("prowlarr")
@@ -79,22 +90,6 @@ class ProwlarrHandler(ExternalClientHandler):
def _completed_path_max_attempts(self) -> int:
return COMPLETED_PATH_MAX_ATTEMPTS
def build_retry_resolution_fields(self, release_data: dict[str, Any]) -> dict[str, Any]:
source_id = normalize_optional_text(release_data.get("source_id"))
if source_id is None:
return {}
prowlarr_result = get_release(source_id)
if prowlarr_result is None:
return {}
return {
"retry_download_url": normalize_optional_text(
get_preferred_download_url(prowlarr_result)
),
"retry_download_protocol": normalize_optional_text(get_protocol(prowlarr_result)),
}
@classmethod
def _restore_download_request_from_task(cls, task: DownloadTask) -> DownloadRequest | None:
"""Rebuild a DownloadRequest when the in-memory Prowlarr cache is gone."""
@@ -162,14 +157,12 @@ class ProwlarrHandler(ExternalClientHandler):
release_name = prowlarr_result.get("title") or task.title or "Unknown"
expected_hash = str(prowlarr_result.get("infoHash") or "").strip() or None
seeding_time_limit = None
ratio_limit = None
if config.get("PROWLARR_USE_SEED_PREFERENCES", False):
raw_configured_seed_time = prowlarr_result.get("configuredSeedTimeMinutes")
raw_configured_ratio = prowlarr_result.get("configuredRatioLimit")
# Seed criteria from the indexer (Torznab attributes)
raw_seed_time = prowlarr_result.get("minimumSeedTime")
raw_ratio = prowlarr_result.get("minimumRatio")
seeding_time_limit = _coerce_positive_minutes(raw_configured_seed_time)
ratio_limit = float(raw_configured_ratio) if raw_configured_ratio is not None else None
seeding_time_limit = _coerce_seed_time_minutes(raw_seed_time)
ratio_limit = float(raw_ratio) if raw_ratio is not None else None
return DownloadRequest(
url=download_url,
@@ -190,11 +190,4 @@ def prowlarr_config_settings() -> list[SettingsField]:
description="Automatically retry search without category filtering if no results are found",
show_when={"field": "PROWLARR_ENABLED", "value": True},
),
CheckboxField(
key="PROWLARR_USE_SEED_PREFERENCES",
label="Use Prowlarr seed preferences",
default=False,
description="Apply per-indexer seed time and ratio preferences from Prowlarr when sending torrents to the download client",
show_when={"field": "PROWLARR_ENABLED", "value": True},
),
]
+8 -36
View File
@@ -27,11 +27,12 @@ from shelfmark.release_sources import (
SortOption,
register_source,
)
from shelfmark.release_sources.prowlarr.api import IndexerSeedSettings, ProwlarrClient
from shelfmark.release_sources.prowlarr.api import ProwlarrClient
from shelfmark.release_sources.prowlarr.cache import cache_release
from shelfmark.release_sources.prowlarr.utils import (
coerce_float_like,
coerce_int_like,
get_preferred_download_url,
get_protocol,
)
@@ -406,7 +407,7 @@ def _prowlarr_result_to_release(
language=language_detected,
size=_parse_size(size_bytes),
size_bytes=size_bytes,
download_url=None,
download_url=get_preferred_download_url(result),
info_url=result.get("infoUrl") or result.get("guid"),
protocol=(
ReleaseProtocol.TORRENT
@@ -432,8 +433,8 @@ def _prowlarr_result_to_release(
"freeleech": is_freeleech,
"download_volume_factor": result.get("downloadVolumeFactor"),
"upload_volume_factor": result.get("uploadVolumeFactor"),
"configured_ratio_limit": result.get("configuredRatioLimit"),
"configured_seed_time_minutes": result.get("configuredSeedTimeMinutes"),
"minimum_ratio": result.get("minimumRatio"),
"minimum_seed_time": result.get("minimumSeedTime"),
"info_hash": result.get("infoHash"),
"formats": formats or None,
"formats_display": formats_display,
@@ -443,27 +444,6 @@ def _prowlarr_result_to_release(
)
def _apply_indexer_seed_settings(
result: dict,
indexer_seed_settings: dict[int, IndexerSeedSettings],
) -> dict:
indexer_id = _coerce_indexer_id(result.get("indexerId"))
if indexer_id is None:
return result
seed_settings = indexer_seed_settings.get(indexer_id)
if not seed_settings:
return result
enriched_result = dict(result)
if "ratio_limit" in seed_settings:
enriched_result["configuredRatioLimit"] = seed_settings["ratio_limit"]
if "seeding_time_limit_minutes" in seed_settings:
enriched_result["configuredSeedTimeMinutes"] = seed_settings["seeding_time_limit_minutes"]
return enriched_result
@register_source("prowlarr")
class ProwlarrSource(ReleaseSource):
"""Prowlarr release source for ebooks and audiobooks."""
@@ -782,11 +762,6 @@ class ProwlarrSource(ReleaseSource):
# Some indexers benefit from title+author queries and extra format detection.
enriched_indexer_ids = client.get_enriched_indexer_ids(restrict_to=indexer_ids)
enriched_indexer_ids_set = set(enriched_indexer_ids)
indexer_seed_settings = (
client.get_indexer_seed_settings(restrict_to=indexer_ids)
if config.get("PROWLARR_USE_SEED_PREFERENCES", False)
else {}
)
def _check_timeout() -> None:
if time.monotonic() > deadline:
@@ -864,18 +839,15 @@ class ProwlarrSource(ReleaseSource):
results: list[Release] = []
enriched_source_ids: set[str] = set()
for raw_result in all_results:
result_with_seed_settings = _apply_indexer_seed_settings(
raw_result, indexer_seed_settings
)
idx_id = result_with_seed_settings.get("indexerId")
for r in all_results:
idx_id = r.get("indexerId")
idx_id_int = _coerce_indexer_id(idx_id)
is_enriched = bool(
idx_id_int is not None and idx_id_int in enriched_indexer_ids_set
)
release = _prowlarr_result_to_release(
result_with_seed_settings,
r,
content_type,
enable_format_detection=is_enriched,
)
@@ -140,6 +140,9 @@ def parse_torznab_xml(xml_text: str) -> list[dict[str, Any]]:
download_volume_factor = _coerce_float(attrs.get("downloadvolumefactor"))
upload_volume_factor = _coerce_float(attrs.get("uploadvolumefactor"))
minimum_ratio = _coerce_float(attrs.get("minimumratio"))
minimum_seed_time = _coerce_int(attrs.get("minimumseedtime"))
cleaned_title = _strip_author_from_title(title, author)
results.append(
@@ -165,6 +168,8 @@ def parse_torznab_xml(xml_text: str) -> list[dict[str, Any]]:
"bookTitle": book_title,
"downloadVolumeFactor": download_volume_factor,
"uploadVolumeFactor": upload_volume_factor,
"minimumRatio": minimum_ratio,
"minimumSeedTime": minimum_seed_time,
# Pass through all torznab attributes for tooltip display
"torznabAttrs": attrs,
}
-1
View File
@@ -58,7 +58,6 @@
"react/jsx-no-useless-fragment": "error",
"react/self-closing-comp": "error",
"typescript/switch-exhaustiveness-check": "error",
"jsx-a11y/prefer-tag-over-role": "off",
"no-restricted-imports": [
"error",
{
+416 -395
View File
File diff suppressed because it is too large Load Diff
+9 -9
View File
@@ -18,23 +18,23 @@
},
"dependencies": {
"@tailwindcss/vite": "^4.2.4",
"react": "^19.2.6",
"react-dom": "^19.2.6",
"react-router-dom": "^7.15.0",
"react": "^19.2.5",
"react-dom": "^19.2.5",
"react-router-dom": "^7.14.2",
"socket.io-client": "^4.7.5"
},
"devDependencies": {
"@types/node": "^25.6.2",
"@types/node": "^25.6.0",
"@types/react": "^19.2.14",
"@types/react-dom": "^19.2.3",
"@vitejs/plugin-react": "^6.0.1",
"knip": "^6.12.1",
"oxfmt": "^0.48.0",
"oxlint": "^1.63.0",
"oxlint-tsgolint": "^0.22.1",
"knip": "^6.6.2",
"oxfmt": "^0.46.0",
"oxlint": "^1.61.0",
"oxlint-tsgolint": "^0.21.1",
"tailwindcss": "^4.2.2",
"typescript": "^6.0.3",
"vite": "^8.0.11",
"vite": "^8.0.10",
"vitest": "^4.1.5"
}
}
+10 -2
View File
@@ -7,6 +7,7 @@ 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';
@@ -136,6 +137,10 @@ 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]) => {
@@ -201,14 +206,17 @@ 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">
{book.preview ? (
{optimizedPreview ? (
<div
className="flex w-full items-center justify-center lg:h-full lg:max-w-none"
style={{ maxHeight: artworkMaxHeight, maxWidth: artworkMaxWidth }}
>
<img
src={book.preview}
src={optimizedPreview}
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 -8
View File
@@ -22,6 +22,7 @@ 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,
@@ -210,8 +211,9 @@ 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 (!preview || imageError) {
if (!optimizedPreview || 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"
@@ -228,10 +230,13 @@ 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={preview}
src={optimizedPreview}
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' }}
@@ -1237,6 +1242,10 @@ 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') {
@@ -1325,13 +1334,14 @@ const ReleaseModalSession = ({
{/* Mobile: static thumbnail always visible */}
{!isRequestMode && (
<div className="shrink-0 sm:hidden">
{book.preview ? (
{modalPreview ? (
<img
src={book.preview}
src={modalPreview}
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,
@@ -1365,13 +1375,14 @@ const ReleaseModalSession = ({
className="transition-opacity duration-300 ease-out"
style={{ opacity: showHeaderThumb ? 1 : 0 }}
>
{book.preview ? (
{modalPreview ? (
<img
src={book.preview}
src={modalPreview}
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,
@@ -1434,10 +1445,13 @@ const ReleaseModalSession = ({
ref={bookSummaryRef}
className="flex gap-4 border-b border-(--border-muted) px-5 py-4"
>
{book.preview ? (
{modalPreview ? (
<img
src={book.preview}
src={modalPreview}
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,6 +5,7 @@ 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,
@@ -175,6 +176,7 @@ 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) {
@@ -240,11 +242,15 @@ 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)">
{preview.preview ? (
{previewImage ? (
<img
src={preview.preview}
src={previewImage}
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,6 +3,7 @@ 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';
@@ -499,6 +500,7 @@ 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 = () => {
@@ -709,11 +711,15 @@ 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">
{item.preview ? (
{previewImage ? (
<img
src={item.preview}
src={previewImage}
alt={`${item.title} cover`}
className="h-full w-full object-cover object-top"
loading="lazy"
decoding="async"
width={48}
height={72}
/>
) : (
<BookFallback />
@@ -3,6 +3,7 @@ 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';
@@ -41,6 +42,11 @@ 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;
@@ -97,7 +103,7 @@ export const CardView = ({
#{book.series_position}
</div>
)}
{book.preview && !imageError ? (
{optimizedPreview && !imageError ? (
<>
{!imageLoaded && (
<div className="absolute inset-0">
@@ -105,10 +111,13 @@ export const CardView = ({
</div>
)}
<img
src={book.preview}
src={optimizedPreview}
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,6 +3,7 @@ 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';
@@ -44,6 +45,11 @@ 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;
@@ -97,7 +103,7 @@ export const CompactView = ({
#{book.series_position}
</div>
)}
{book.preview && !imageError ? (
{optimizedPreview && !imageError ? (
<>
{!imageLoaded && (
<div className="absolute inset-0">
@@ -105,10 +111,13 @@ export const CompactView = ({
</div>
)}
<img
src={book.preview}
src={optimizedPreview}
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,6 +4,7 @@ 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';
@@ -47,8 +48,12 @@ 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 (!preview || imageError) {
if (!optimizedPreview || 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`}
@@ -67,10 +72,13 @@ 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={preview}
src={optimizedPreview}
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,6 +106,7 @@ 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) && (
@@ -18,10 +18,6 @@ export const OidcEnvInfo = (_props: CustomSettingsFieldRendererProps) => {
{' '}
<span className="opacity-40"># Hide the local login form</span>
{'\n'}
{' '}- <span className="text-blue-400">DISABLE_LOCAL_AUTH</span>=
<span className="text-green-400">true</span>{' '}
<span className="opacity-40"># Disable username/password login</span>
{'\n'}
{' '}- <span className="text-blue-400">OIDC_AUTO_REDIRECT</span>=
<span className="text-green-400">true</span>
{' '}
+4 -4
View File
@@ -75,8 +75,8 @@ const hydrateSettingsResponse = (response: SettingsResponse): HydratedSettingsSt
});
const values = extractSettingsValues(tabs);
if (values.general && Object.prototype.hasOwnProperty.call(values.general, THEME_FIELD.key)) {
values.general[THEME_FIELD.key] = getStoredThemePreference();
if (values.general && Object.prototype.hasOwnProperty.call(values.general, '_THEME')) {
values.general._THEME = getStoredThemePreference();
}
return {
@@ -208,7 +208,7 @@ export function useSettings(): UseSettingsReturn {
const updateValue = useCallback(
(tabName: string, key: string, value: unknown) => {
if (key === THEME_FIELD.key && typeof value === 'string') {
if (key === '_THEME' && typeof value === 'string') {
setThemePreference(value);
setOriginalValues((prev) => ({
...prev,
@@ -277,7 +277,7 @@ export function useSettings(): UseSettingsReturn {
if (tab) {
for (const field of getValueBearingFields(tab.fields)) {
if (field.fromEnv) continue; // Skip env-locked fields
if (field.key === THEME_FIELD.key) continue; // Skip client-side only theme field
if (field.key === '_THEME') continue; // Skip client-side only theme field
const value = tabValues[field.key];
const originalValue = originalTabValues[field.key];
+33
View File
@@ -0,0 +1,33 @@
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');
});
});
@@ -4,9 +4,7 @@ import type { Language } from '../types/index';
import {
LANGUAGE_OPTION_ALL,
LANGUAGE_OPTION_DEFAULT,
buildLanguageNormalizer,
getReleaseSearchLanguageParams,
releaseLanguageMatchesFilter,
} from '../utils/languageFilters';
const supportedLanguages: Language[] = [
@@ -37,33 +35,4 @@ describe('languageFilters release search params', () => {
expect(result).toEqual(['de', 'hu']);
});
it('normalizes legacy default language names when combined with explicit filters', () => {
const result = getReleaseSearchLanguageParams(
[LANGUAGE_OPTION_DEFAULT, 'de'],
supportedLanguages,
['english'],
);
expect(result).toEqual(['en', 'de']);
});
});
describe('releaseLanguageMatchesFilter', () => {
it('matches release language names against legacy default language names', () => {
const normalizer = buildLanguageNormalizer(supportedLanguages);
expect(releaseLanguageMatchesFilter('English', ['english'], normalizer)).toBe(true);
});
it('keeps English-only issue 948 fallback results with a legacy English default', () => {
const normalizer = buildLanguageNormalizer(supportedLanguages);
const issue948Languages = [...Array<string>(48).fill('en'), 'de, en', 'en, es'];
const visibleLanguages = issue948Languages.filter((language) =>
releaseLanguageMatchesFilter(language, ['english'], normalizer),
);
expect(visibleLanguages).toHaveLength(48);
});
});
@@ -2,18 +2,11 @@ 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})',
+4 -4
View File
@@ -31,13 +31,13 @@ const resolveBasePath = (): string => {
};
// Lazy initialization to ensure DOM is ready when base path is resolved
let cachedBasePath: string | null = null;
let _basePath: string | null = null;
export const getBasePath = (): string => {
if (cachedBasePath === null) {
cachedBasePath = normalizeBasePath(resolveBasePath());
if (_basePath === null) {
_basePath = normalizeBasePath(resolveBasePath());
}
return cachedBasePath;
return _basePath;
};
export const withBasePath = (path: string): string => {
+66
View File
@@ -0,0 +1,66 @@
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();
};
+5 -14
View File
@@ -48,11 +48,8 @@ export const getLanguageFilterValues = (
return null;
}
const languageNormalizer = buildLanguageNormalizer(supportedLanguages);
const supportedCodes = new Set(supportedLanguages.map((lang) => lang.code.toLowerCase()));
const defaultCodes = defaultLanguageCodes
.map((code) => languageNormalizer.get(code.toLowerCase()) ?? code.toLowerCase())
.filter((code) => supportedCodes.has(code));
const supportedCodes = new Set(supportedLanguages.map((lang) => lang.code));
const defaultCodes = defaultLanguageCodes.filter((code) => supportedCodes.has(code));
const resolved = new Set<string>();
uniqueSelection.forEach((code) => {
@@ -61,9 +58,8 @@ export const getLanguageFilterValues = (
return;
}
const normalizedCode = languageNormalizer.get(code.toLowerCase()) ?? code.toLowerCase();
if (supportedCodes.has(normalizedCode)) {
resolved.add(normalizedCode);
if (supportedCodes.has(code)) {
resolved.add(code);
}
});
@@ -134,11 +130,6 @@ export const releaseLanguageMatchesFilter = (
return part;
});
const selectedSet = new Set(
selectedCodes.map((code) => {
const normalizedCode = code.toLowerCase();
return languageNormalizer?.get(normalizedCode) ?? normalizedCode;
}),
);
const selectedSet = new Set(selectedCodes.map((c) => c.toLowerCase()));
return releaseCodes.every((code) => selectedSet.has(code));
};
@@ -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: 'Universal',
group: 'Core',
},
{
token: 'Year',
-77
View File
@@ -120,83 +120,6 @@ class TestAudiobookBayHandlerDownload:
)
assert "resolving" in recorder.statuses
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
@patch("shelfmark.release_sources.audiobookbay.handler.config.get")
def test_download_rejects_source_url_host_mismatch(
self, mock_config_get, mock_get_client, mock_extract_magnet
):
"""Test hostile detail URLs are rejected before page fetch."""
mock_config_get.side_effect = lambda key, default="": (
"audiobookbay.lu" if key == "ABB_HOSTNAME" else default
)
handler = AudiobookBayHandler()
task = DownloadTask(
task_id="35f56a3e5734bfa69c3169ee8e605a60",
source="audiobookbay",
title="Test Book",
content_type="audiobook",
source_url="https://169.254.169.254/latest/meta-data/",
)
cancel_flag = Event()
recorder = ProgressRecorder()
result = handler.download(
task=task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
assert recorder.last_status == "error"
assert "details url" in recorder.last_message.lower()
mock_extract_magnet.assert_not_called()
mock_get_client.assert_not_called()
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
@patch("shelfmark.release_sources.audiobookbay.handler.config.get")
def test_download_allows_configured_source_url_host(
self, mock_config_get, mock_get_client, mock_extract_magnet
):
"""Test configured ABB host remains allowed for queued release URLs."""
mock_config_get.side_effect = lambda key, default="": (
"https://audiobookbay.lu/" if key == "ABB_HOSTNAME" else default
)
mock_extract_magnet.return_value = "magnet:?xt=urn:btih:abc123"
mock_client = MagicMock()
mock_client.name = "qbittorrent"
mock_client.find_existing.return_value = None
mock_client.add_download.return_value = "download_id_123"
mock_get_client.return_value = mock_client
handler = AudiobookBayHandler()
task = DownloadTask(
task_id="35f56a3e5734bfa69c3169ee8e605a60",
source="audiobookbay",
title="Test Book",
content_type="audiobook",
source_url="https://audiobookbay.lu/abss/test-book/",
)
cancel_flag = Event()
recorder = ProgressRecorder()
with patch.object(AudiobookBayHandler, "_poll_and_complete", return_value=None):
result = handler.download(
task=task,
cancel_flag=cancel_flag,
progress_callback=recorder.progress_callback,
status_callback=recorder.status_callback,
)
assert result is None
mock_extract_magnet.assert_called_once_with(
"https://audiobookbay.lu/abss/test-book/", "audiobookbay.lu"
)
mock_client.add_download.assert_called_once()
@patch("shelfmark.release_sources.audiobookbay.handler.scraper.extract_magnet_link")
@patch("shelfmark.release_sources.audiobookbay.handler.get_client")
def test_download_existing_complete(self, mock_get_client, mock_extract_magnet):
-15
View File
@@ -156,21 +156,6 @@ 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
@@ -10,9 +10,6 @@ from pathlib import Path
ENTRYPOINT_PATH = Path(__file__).resolve().parents[2] / "entrypoint.sh"
ENTRYPOINT_LOCK_PATH = Path("/tmp/shelfmark_entrypoint_test.lock")
BASH_PATH = shutil.which("bash") or "/bin/bash"
ID_PATH = shutil.which("id") or "/usr/bin/id"
MKDIR_PATH = shutil.which("mkdir") or "/bin/mkdir"
STAT_PATH = shutil.which("stat") or "/usr/bin/stat"
@contextlib.contextmanager
@@ -58,61 +55,6 @@ exit 2
printf '%s' "$HOME" > "$ENTRYPOINT_GUNICORN_HOME_FILE"
printf '%s' "$*" > "$ENTRYPOINT_GUNICORN_ARGS_FILE"
exit 0
""",
)
_write_executable(
bin_dir / "id",
"""#!/bin/sh
if [ -n "${ENTRYPOINT_STUB_CURRENT_UID:-}" ]; then
if [ "$1" = "-u" ]; then
printf '%s\\n' "$ENTRYPOINT_STUB_CURRENT_UID"
exit 0
fi
if [ "$1" = "-g" ]; then
printf '%s\\n' "$ENTRYPOINT_STUB_CURRENT_GID"
exit 0
fi
fi
exec "$ENTRYPOINT_REAL_ID" "$@"
""",
)
_write_executable(
bin_dir / "gosu",
"""#!/bin/sh
shift
if [ "${ENTRYPOINT_STUB_GOSU_FAIL_WRITES:-false}" = "true" ] && [ "$1" = "sh" ] && [ "$2" = "-c" ]; then
exit 1
fi
exec "$@"
""",
)
_write_executable(
bin_dir / "mkdir",
"""#!/bin/sh
for arg in "$@"; do
if [ "$arg" = "/var/log/shelfmark" ]; then
exit 0
fi
done
exec "$ENTRYPOINT_REAL_MKDIR" "$@"
""",
)
_write_executable(
bin_dir / "stat",
"""#!/bin/sh
if [ "$1" = "-c" ]; then
if [ "$2" = "%u:%g" ]; then
printf '%s\\n' "${ENTRYPOINT_STUB_STAT_OWNER:-0:0}"
exit 0
fi
fi
exec "$ENTRYPOINT_REAL_STAT" "$@"
""",
)
_write_executable(
bin_dir / "chown",
"""#!/bin/sh
exit 0
""",
)
@@ -123,8 +65,6 @@ def _run_entrypoint(
tmp_path: Path,
*,
extra_env: dict[str, str] | None = None,
simulate_root_startup: bool = False,
fail_gosu_writes: bool = False,
stub_home: Path | str | None = None,
) -> tuple[subprocess.CompletedProcess[str], Path, Path, Path]:
runtime_home = tmp_path / "runtime-home"
@@ -146,9 +86,6 @@ def _run_entrypoint(
"ENABLE_LOGGING": "false",
"ENTRYPOINT_GUNICORN_ARGS_FILE": str(runtime_args_file),
"ENTRYPOINT_GUNICORN_HOME_FILE": str(runtime_home_file),
"ENTRYPOINT_REAL_ID": ID_PATH,
"ENTRYPOINT_REAL_MKDIR": MKDIR_PATH,
"ENTRYPOINT_REAL_STAT": STAT_PATH,
"ENTRYPOINT_STUB_GID": str(os.getgid()),
"ENTRYPOINT_STUB_HOME": str(stub_home),
"ENTRYPOINT_STUB_UID": str(os.getuid()),
@@ -162,18 +99,6 @@ def _run_entrypoint(
"USING_EXTERNAL_BYPASSER": "true",
}
)
if simulate_root_startup:
env.update(
{
"ENTRYPOINT_STUB_CURRENT_GID": "0",
"ENTRYPOINT_STUB_CURRENT_UID": "0",
"ENTRYPOINT_STUB_STAT_OWNER": "0:0",
"PGID": str(os.getgid()),
"PUID": str(os.getuid()),
}
)
if fail_gosu_writes:
env["ENTRYPOINT_STUB_GOSU_FAIL_WRITES"] = "true"
if extra_env:
env.update(extra_env)
@@ -243,17 +168,3 @@ def test_entrypoint_non_root_mode_requires_writable_config_dir(tmp_path):
f"Config directory is not writable in non-root mode: {readonly_config_dir}" in result.stdout
)
assert "Prepare ownership outside the container" in result.stdout
def test_entrypoint_root_bootstrap_fails_closed_when_config_repair_fails(tmp_path):
result, _, _, _ = _run_entrypoint(
tmp_path,
simulate_root_startup=True,
fail_gosu_writes=True,
)
assert result.returncode == 1
assert "ERROR: Config directory is not writable!" in result.stdout
assert f"Configured runtime identity: {os.getuid()}:{os.getgid()}" in result.stdout
assert f"chown -R {os.getuid()}:{os.getgid()} /path/to/config" in result.stdout
assert "Startup mode: root" not in result.stdout
-19
View File
@@ -38,22 +38,3 @@ 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
-32
View File
@@ -381,16 +381,6 @@ class TestSecuritySettings:
assert "inactive" in hint.label.lower()
assert "local admin" in hint.label.lower()
def test_oidc_admin_requirement_hint_absent_when_local_auth_is_disabled(self):
"""OIDC mode should not show the local-admin warning when local auth is disabled."""
from shelfmark.config.security import security_settings
with patch("shelfmark.config.env.DISABLE_LOCAL_AUTH", True):
fields = security_settings()
hint = next((f for f in fields if f.key == "oidc_admin_requirement"), None)
assert hint is None
def test_builtin_option_label_is_local(self):
"""Builtin auth option should be labeled Local."""
from shelfmark.config.security import security_settings
@@ -439,28 +429,6 @@ class TestSecurityOnSave:
assert result["error"] is True
assert "local admin" in result["message"].lower()
def test_on_save_allows_oidc_without_local_admin_when_local_auth_is_disabled(
self, tmp_path, monkeypatch
):
from shelfmark.config.security import _on_save_security
_set_config_dir(monkeypatch, tmp_path)
UserDB(str(tmp_path / "users.db")).initialize()
with patch("shelfmark.config.security_handlers.DISABLE_LOCAL_AUTH", True):
result = _on_save_security(
{
"AUTH_METHOD": "oidc",
"OIDC_DISCOVERY_URL": (
"https://auth.example.com/.well-known/openid-configuration"
),
"OIDC_CLIENT_ID": "shelfmark",
"OIDC_CLIENT_SECRET": "secret123",
}
)
assert result["error"] is False
def test_on_save_blocks_oidc_when_client_id_is_missing(self, tmp_path, monkeypatch):
from shelfmark.config.security import _on_save_security
-65
View File
@@ -1,65 +0,0 @@
from __future__ import annotations
from pathlib import Path
TOR_SCRIPT_PATH = Path(__file__).resolve().parents[2] / "tor.sh"
def _generated_tor_healthcheck_script() -> str:
script = TOR_SCRIPT_PATH.read_text()
start = script.index("cat <<'HC' > /app/tor_healthcheck.sh")
content_start = script.index("\n", start) + 1
content_end = script.index("\nHC", content_start)
return script[content_start:content_end]
def _tor_script_rule_lines() -> list[str]:
return [
line.strip()
for line in TOR_SCRIPT_PATH.read_text().splitlines()
if line.strip().startswith("iptables ")
]
def _line_index(lines: list[str], needle: str) -> int:
return next(index for index, line in enumerate(lines) if needle in line)
def test_tor_nat_rules_bypass_private_networks_before_tcp_redirect():
lines = _tor_script_rule_lines()
tcp_redirect_index = _line_index(lines, "--syn -j REDIRECT --to-ports 9040")
for cidr in ("127.0.0.0/8", "10.0.0.0/8", "172.16.0.0/12", "192.168.0.0/16"):
rule_index = _line_index(lines, f"-d {cidr} -j RETURN")
assert rule_index < tcp_redirect_index
def test_tor_nat_rules_exempt_tor_process_before_dns_and_tcp_redirects():
lines = _tor_script_rule_lines()
owner_index = _line_index(lines, "-m owner --uid-owner")
udp_dns_index = _line_index(lines, "-p udp --dport 53")
tcp_dns_index = _line_index(lines, "-p tcp --dport 53")
tcp_redirect_index = _line_index(lines, "--syn -j REDIRECT --to-ports 9040")
assert owner_index < udp_dns_index
assert owner_index < tcp_dns_index
assert owner_index < tcp_redirect_index
def test_tor_nat_rules_handle_dns_before_tcp_redirect():
lines = _tor_script_rule_lines()
tcp_redirect_index = _line_index(lines, "--syn -j REDIRECT --to-ports 9040")
assert _line_index(lines, "-p udp --dport 53") < tcp_redirect_index
assert _line_index(lines, "-p tcp --dport 53") < tcp_redirect_index
def test_tor_healthcheck_uses_local_tor_state_without_clear_net_probe():
healthcheck_script = _generated_tor_healthcheck_script()
assert "google.com" not in healthcheck_script
assert "curl " not in healthcheck_script
assert "supervisorctl status tor" in healthcheck_script
assert "Bootstrapped 100%" in healthcheck_script
-28
View File
@@ -126,21 +126,6 @@ class TestLoginSemantics:
assert response.status_code == 403
assert response.get_json()["error"] == "Local authentication is disabled"
@pytest.mark.parametrize("auth_mode", ["builtin", "oidc"])
def test_login_rejects_password_auth_when_local_auth_is_disabled(
self, main_module, client, auth_mode
):
with patch.object(main_module, "get_auth_mode", return_value=auth_mode):
with patch.object(main_module, "DISABLE_LOCAL_AUTH", True):
response = client.post(
"/api/auth/login",
json={"username": "alice", "password": "wrong", "remember_me": False},
)
assert response.status_code == 403
assert response.get_json()["error"] == "Local authentication is disabled"
assert main_module.failed_login_attempts == {}
def test_auth_check_none_mode_reports_full_access(self, main_module, client):
with patch.object(main_module, "get_auth_mode", return_value="none"):
response = client.get("/api/auth/check")
@@ -153,19 +138,6 @@ class TestLoginSemantics:
"is_admin": True,
}
@pytest.mark.parametrize("auth_mode", ["builtin", "oidc"])
def test_auth_check_hides_local_auth_when_disabled(self, main_module, client, auth_mode):
with patch.object(main_module, "get_auth_mode", return_value=auth_mode):
with patch.object(main_module, "DISABLE_LOCAL_AUTH", True):
response = client.get("/api/auth/check")
assert response.status_code == 200
body = response.get_json()
assert body["auth_mode"] == auth_mode
assert body["auth_required"] is True
assert body["authenticated"] is False
assert body["hide_local_auth"] is True
def test_auth_check_includes_display_name_for_authenticated_user(
self, main_module, client, temp_user_db, monkeypatch
):
-28
View File
@@ -1,28 +0,0 @@
"""Cover proxy API security tests."""
from __future__ import annotations
import importlib
from unittest.mock import patch
import pytest
@pytest.fixture(scope="module")
def main_module():
"""Import `shelfmark.main` with background startup disabled."""
with patch("shelfmark.download.orchestrator.start"):
import shelfmark.main as main
importlib.reload(main)
return main
def test_cover_proxy_requires_authentication(main_module) -> None:
client = main_module.app.test_client()
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
response = client.get("/api/covers/test-id")
assert response.status_code == 401
assert response.get_json() == {"error": "Unauthorized"}
-201
View File
@@ -856,204 +856,3 @@ class TestStatusEndpointGuardrails:
assert resp.status_code == 200
assert observed["user_id"] is None
class TestQueueManagementEndpointGuardrails:
def test_non_owner_cannot_set_priority(self, main_module, client):
owner = _create_user(main_module, prefix="owner")
actor = _create_user(main_module, prefix="actor")
_set_authenticated_session(
client,
user_id=actor["username"],
db_user_id=actor["id"],
is_admin=False,
)
task = DownloadTask(
task_id="owned-priority-1",
source="direct_download",
title="Owned Task",
user_id=owner["id"],
username=owner["username"],
)
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(main_module.backend.book_queue, "get_task", return_value=task):
with patch.object(main_module.backend, "set_book_priority") as mock_set_priority:
resp = client.put("/api/queue/owned-priority-1/priority", json={"priority": 1})
assert resp.status_code == 403
assert resp.get_json()["code"] == "download_not_owned"
mock_set_priority.assert_not_called()
def test_owner_can_set_priority(self, main_module, client):
user = _create_user(main_module, prefix="reader")
_set_authenticated_session(
client,
user_id=user["username"],
db_user_id=user["id"],
is_admin=False,
)
task = DownloadTask(
task_id="reader-priority-1",
source="direct_download",
title="Reader Task",
user_id=user["id"],
username=user["username"],
)
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(main_module.backend.book_queue, "get_task", return_value=task):
with patch.object(
main_module.backend, "set_book_priority", return_value=True
) as mock_set_priority:
resp = client.put("/api/queue/reader-priority-1/priority", json={"priority": 2})
assert resp.status_code == 200
assert resp.get_json() == {
"status": "updated",
"book_id": "reader-priority-1",
"priority": 2,
}
mock_set_priority.assert_called_once_with("reader-priority-1", 2)
def test_non_owner_cannot_reorder_other_users_task(self, main_module, client):
owner = _create_user(main_module, prefix="owner")
actor = _create_user(main_module, prefix="actor")
_set_authenticated_session(
client,
user_id=actor["username"],
db_user_id=actor["id"],
is_admin=False,
)
owned_task = DownloadTask(
task_id="actor-reorder-1",
source="direct_download",
title="Actor Task",
user_id=actor["id"],
username=actor["username"],
)
other_task = DownloadTask(
task_id="owner-reorder-1",
source="direct_download",
title="Owner Task",
user_id=owner["id"],
username=owner["username"],
)
def fake_get_task(task_id):
return {
"actor-reorder-1": owned_task,
"owner-reorder-1": other_task,
}.get(task_id)
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module.backend.book_queue, "get_task", side_effect=fake_get_task
):
with patch.object(main_module.backend, "reorder_queue") as mock_reorder:
resp = client.post(
"/api/queue/reorder",
json={
"book_priorities": {
"actor-reorder-1": 1,
"owner-reorder-1": 0,
}
},
)
assert resp.status_code == 403
assert resp.get_json()["code"] == "download_not_owned"
mock_reorder.assert_not_called()
def test_non_admin_queue_order_is_scoped_to_owned_tasks(self, main_module, client):
user = _create_user(main_module, prefix="reader")
other = _create_user(main_module, prefix="other")
_set_authenticated_session(
client,
user_id=user["username"],
db_user_id=user["id"],
is_admin=False,
)
user_task = DownloadTask(
task_id="reader-order-1",
source="direct_download",
title="Reader Task",
user_id=user["id"],
username=user["username"],
)
other_task = DownloadTask(
task_id="other-order-1",
source="direct_download",
title="Other Task",
user_id=other["id"],
username=other["username"],
)
def fake_get_task(task_id):
return {
"reader-order-1": user_task,
"other-order-1": other_task,
}.get(task_id)
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module.backend,
"get_queue_order",
return_value=[
{"id": "reader-order-1", "title": "Reader Task", "priority": 0},
{"id": "other-order-1", "title": "Other Task", "priority": 1},
],
):
with patch.object(
main_module.backend.book_queue, "get_task", side_effect=fake_get_task
):
resp = client.get("/api/queue/order")
assert resp.status_code == 200
assert resp.get_json()["queue"] == [
{"id": "reader-order-1", "title": "Reader Task", "priority": 0}
]
def test_non_admin_active_downloads_are_scoped_to_owned_tasks(self, main_module, client):
user = _create_user(main_module, prefix="reader")
other = _create_user(main_module, prefix="other")
_set_authenticated_session(
client,
user_id=user["username"],
db_user_id=user["id"],
is_admin=False,
)
user_task = DownloadTask(
task_id="reader-active-1",
source="direct_download",
title="Reader Task",
user_id=user["id"],
username=user["username"],
)
other_task = DownloadTask(
task_id="other-active-1",
source="direct_download",
title="Other Task",
user_id=other["id"],
username=other["username"],
)
def fake_get_task(task_id):
return {
"reader-active-1": user_task,
"other-active-1": other_task,
}.get(task_id)
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module.backend,
"get_active_downloads",
return_value=["reader-active-1", "other-active-1"],
):
with patch.object(
main_module.backend.book_queue, "get_task", side_effect=fake_get_task
):
resp = client.get("/api/downloads/active")
assert resp.status_code == 200
assert resp.get_json() == {"active_downloads": ["reader-active-1"]}
-20
View File
@@ -240,26 +240,6 @@ class TestAtomicCopy:
assert result.exists()
assert result.read_text() == "content"
def test_copy_falls_back_when_copy2_hits_fuse_eio(self, tmp_path):
"""Fall back to content copy when FUSE rejects xattr metadata reads."""
import errno
from shelfmark.download.fs import atomic_copy as _atomic_copy
source = tmp_path / "source.txt"
source.write_text("content")
dest = tmp_path / "dest.txt"
with patch(
"shelfmark.download.fs.shutil.copy2",
side_effect=OSError(errno.EIO, "Input/output error"),
):
result = _atomic_copy(source, dest)
assert result == dest
assert result.exists()
assert result.read_text() == "content"
def test_copy_tolerates_post_publish_estale(self, tmp_path, monkeypatch):
"""Treat ESTALE on the final destination as a successful NFS publish."""
import errno
+7 -51
View File
@@ -28,6 +28,7 @@ def _run_organize_post_process(
task,
library: Path,
hardlink_enabled: bool = True,
same_fs: bool = True,
):
from shelfmark.download.postprocess.router import (
post_process_download as _post_process_download,
@@ -36,7 +37,10 @@ def _run_organize_post_process(
status_cb = MagicMock()
cancel_flag = Event()
with patch("shelfmark.core.config.config") as mock_config:
with (
patch("shelfmark.core.config.config") as mock_config,
patch("shelfmark.download.postprocess.transfer.same_filesystem", return_value=same_fs),
):
mock_config.CUSTOM_SCRIPT = None
mock_config.get = MagicMock(
side_effect=lambda key, default=None, **_kwargs: {
@@ -403,32 +407,6 @@ class TestAtomicMove:
assert mock_copy.called
assert mock_fallback.called
def test_cross_filesystem_move_falls_back_when_copy2_hits_fuse_eio(self, tmp_path, monkeypatch):
"""Falls back to content copy when FUSE rejects xattr metadata reads."""
import errno
from shelfmark.download.fs import atomic_move as _atomic_move
source = tmp_path / "source.txt"
source.write_text("content")
dest = tmp_path / "dest.txt"
def _raise_exdev(*_args, **_kwargs):
raise OSError(errno.EXDEV, "Cross-device link")
monkeypatch.setattr(os, "rename", _raise_exdev)
with patch(
"shelfmark.download.fs.shutil.copy2",
side_effect=OSError(errno.EIO, "Input/output error"),
):
result = _atomic_move(source, dest)
assert result == dest
assert not source.exists()
assert dest.exists()
assert dest.read_text() == "content"
def test_cross_filesystem_move_recovers_when_metadata_step_hits_enoent(
self, tmp_path, monkeypatch
):
@@ -718,6 +696,7 @@ class TestHardlinkDecisionLogic:
task=sample_task,
library=library,
hardlink_enabled=True,
same_fs=True,
)
assert result is not None
@@ -771,29 +750,6 @@ class TestHardlinkDecisionLogic:
assert result is not None
assert not staged.exists()
def test_non_prowlarr_torrent_with_original_path_can_hardlink(self, tmp_path, sample_task):
"""Torrent-backed sources such as AudiobookBay can hardlink client files."""
library = tmp_path / "library"
library.mkdir()
source = tmp_path / "downloads" / "book.m4b"
source.parent.mkdir()
source.write_bytes(b"content")
sample_task.source = "audiobookbay"
sample_task.content_type = "audiobook"
sample_task.format = "m4b"
sample_task.original_download_path = str(source)
result, _ = _run_organize_post_process(
temp_file=source,
task=sample_task,
library=library,
hardlink_enabled=True,
)
assert result is not None
assert Path(result).stat().st_ino == source.stat().st_ino
class TestHardlinkInodeVerification:
"""Tests that verify hardlinks share the same inode."""
@@ -1365,7 +1321,7 @@ class TestTorrentSourceCleanupProtection:
from shelfmark.download.postprocess.pipeline import transfer_file_to_library
# Simulate by directly calling transfer_file_to_library with use_hardlink=False
# (this is what happens when hardlinking is disabled before transfer)
# (this is what happens after same_filesystem check fails)
downloads = tmp_path / "downloads"
downloads.mkdir()
torrent_file = downloads / "book.epub"
+73 -53
View File
@@ -1,74 +1,94 @@
"""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
from shelfmark.core.image_cache import (
ImageCacheService,
build_variant_cache_id,
create_image_variant,
normalize_variant_dimension,
normalize_variant_format,
)
def test_fetch_and_cache_rejects_backslash_authority_bypass_before_request(
tmp_path, monkeypatch
) -> None:
cache = ImageCacheService(tmp_path)
calls = []
def fake_get(url, **_kwargs):
calls.append(url)
raise AssertionError("unsafe URL should not be requested")
monkeypatch.setattr("shelfmark.core.image_cache.requests.get", fake_get)
assert cache.fetch_and_cache("cover-ssrf", "http://127.0.0.1:6666\\@1.1.1.1") is None
assert calls == []
assert "cover-ssrf" not in cache._index
def test_is_safe_url_rejects_encoded_separator_in_authority() -> None:
assert ImageCacheService._is_safe_url("http://127.0.0.1:6666%5c@1.1.1.1") is False
assert ImageCacheService._is_safe_url("http://127.0.0.1:6666%2f@1.1.1.1") is False
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()
def test_is_safe_url_rejects_invalid_ipv6_url() -> None:
assert ImageCacheService._is_safe_url("http://[") is False
def test_fetch_and_cache_blocks_unsafe_redirect(tmp_path, monkeypatch) -> None:
cache = ImageCacheService(tmp_path)
def fake_getaddrinfo(hostname, *_args, **_kwargs):
addresses = {
"example.com": "93.184.216.34",
"127.0.0.1": "127.0.0.1",
}
return [(None, None, None, None, (addresses[hostname], 0))]
class RedirectResponse:
is_redirect = True
headers = {"location": "http://127.0.0.1/cover.jpg"}
def close(self):
return None
calls = []
def fake_get(url, **_kwargs):
calls.append(url)
return RedirectResponse()
monkeypatch.setattr("shelfmark.core.image_cache.socket.getaddrinfo", fake_getaddrinfo)
monkeypatch.setattr("shelfmark.core.image_cache.requests.get", fake_get)
assert cache.fetch_and_cache("cover-redirect", "https://example.com/cover.jpg") is None
assert calls == ["https://example.com/cover.jpg"]
assert "cover-redirect" not in cache._index
def test_fetch_and_cache_returns_none_on_request_exception(tmp_path, monkeypatch) -> None:
cache = ImageCacheService(tmp_path)
monkeypatch.setattr(cache, "_is_safe_url", lambda _url: True)
def fake_get(*_args, **_kwargs):
def fake_get(*args, **kwargs):
raise requests.exceptions.TooManyRedirects("too many redirects")
monkeypatch.setattr("shelfmark.core.image_cache.requests.get", fake_get)
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"
)
-27
View File
@@ -2,8 +2,6 @@
import sqlite3
import pytest
from shelfmark.core.auth_modes import (
determine_auth_mode,
get_auth_check_admin_status,
@@ -65,31 +63,6 @@ class TestDetermineAuthMode:
}
assert determine_auth_mode(config, cwa_db_path=None, has_local_admin=False) == "none"
@pytest.mark.parametrize(
("auth_mode", "config"),
[
("builtin", {"AUTH_METHOD": "builtin"}),
(
"oidc",
{
"AUTH_METHOD": "oidc",
"OIDC_DISCOVERY_URL": "https://auth.example.com/.well-known/openid-configuration",
"OIDC_CLIENT_ID": "shelfmark",
},
),
],
)
def test_disable_local_auth_keeps_configured_mode_without_admin(self, auth_mode, config):
assert (
determine_auth_mode(
config,
cwa_db_path=None,
has_local_admin=False,
disable_local_auth=True,
)
== auth_mode
)
def test_load_active_auth_mode_reads_env_backed_cwa_setting(self, monkeypatch, tmp_path):
from shelfmark.core.config import config as app_config
-57
View File
@@ -445,7 +445,6 @@ class TestOIDCCallbackEndpoint:
"userinfo": {
"sub": "oidc-alice-sub",
"email": "alice@example.com",
"email_verified": True,
"preferred_username": "alice_oidc",
"groups": [],
}
@@ -549,7 +548,6 @@ class TestOIDCCallbackEndpoint:
"userinfo": {
"sub": "oidc-new-sub",
"email": "shared@example.com",
"email_verified": True,
"preferred_username": "oidcuser",
"groups": [],
}
@@ -576,7 +574,6 @@ class TestOIDCCallbackEndpoint:
"userinfo": {
"sub": "oidc-nomatch",
"email": "different@example.com",
"email_verified": True,
"preferred_username": "newuser",
"groups": [],
}
@@ -592,60 +589,6 @@ class TestOIDCCallbackEndpoint:
original = user_db.get_user(username="existing")
assert original["oidc_subject"] is None
@patch("shelfmark.core.oidc_routes._get_oidc_client")
def test_callback_does_not_link_with_unverified_email(self, mock_get_client, client, user_db):
"""OIDC login should not link by email when email_verified is false."""
user_db.create_user(username="existing", email="shared@example.com", password_hash="hash")
fake_client = Mock()
fake_client.authorize_access_token.return_value = {
"userinfo": {
"sub": "oidc-unverified",
"email": "shared@example.com",
"email_verified": False,
"preferred_username": "attackeruser",
"groups": [],
}
}
mock_get_client.return_value = (fake_client, MOCK_OIDC_CONFIG)
resp = client.get("/api/auth/oidc/callback?code=abc123&state=test-state")
assert resp.status_code == 302
with client.session_transaction() as sess:
assert sess["user_id"] == "attackeruser"
original = user_db.get_user(username="existing")
assert original["oidc_subject"] is None
@patch("shelfmark.core.oidc_routes._get_oidc_client")
def test_callback_rejects_unverified_email_link_when_no_provision(
self, mock_get_client, client, user_db
):
"""OIDC login should not link by unverified email when creation is disabled."""
config = {**MOCK_OIDC_CONFIG, "OIDC_AUTO_PROVISION": False}
user_db.create_user(username="existing", email="shared@example.com", password_hash="hash")
fake_client = Mock()
fake_client.authorize_access_token.return_value = {
"userinfo": {
"sub": "oidc-unverified-no-provision",
"email": "shared@example.com",
"email_verified": False,
"preferred_username": "attackeruser",
"groups": [],
}
}
mock_get_client.return_value = (fake_client, config)
resp = client.get("/api/auth/oidc/callback?code=abc123&state=test-state")
error = _get_oidc_error(resp)
assert error is not None
assert "Account not found" in error
original = user_db.get_user(username="existing")
assert original["oidc_subject"] is None
@patch("shelfmark.core.oidc_routes._get_oidc_client")
def test_callback_no_email_link_when_oidc_has_no_email(self, mock_get_client, client, user_db):
"""OIDC login without email in claims should not attempt email linking."""
-59
View File
@@ -1,59 +0,0 @@
from shelfmark.core.path_mappings import (
RemotePathMapping,
remap_remote_to_local_with_match,
)
def test_remap_rejects_parent_directory_remainder(tmp_path):
mapping = RemotePathMapping(
host="qbittorrent",
remote_path="/remote/downloads",
local_path=str(tmp_path / "local" / "downloads"),
)
remote_path = "/remote/downloads/../outside/book.epub"
remapped, matched = remap_remote_to_local_with_match(
mappings=[mapping],
host="qbittorrent",
remote_path=remote_path,
)
assert matched is True
assert remapped is None
def test_remap_rejects_path_that_resolves_outside_local_prefix(tmp_path):
local_prefix = tmp_path / "local" / "downloads"
mapping = RemotePathMapping(
host="qbittorrent",
remote_path="/remote/downloads",
local_path=str(local_prefix),
)
remote_path = "/remote/downloads/subdir/../../outside/book.epub"
remapped, matched = remap_remote_to_local_with_match(
mappings=[mapping],
host="qbittorrent",
remote_path=remote_path,
)
assert matched is True
assert remapped is None
def test_remap_allows_normal_child_path_under_local_prefix(tmp_path):
local_prefix = tmp_path / "local" / "downloads"
mapping = RemotePathMapping(
host="qbittorrent",
remote_path="/remote/downloads",
local_path=str(local_prefix),
)
remapped, matched = remap_remote_to_local_with_match(
mappings=[mapping],
host="qbittorrent",
remote_path="/remote/downloads/author/book.epub",
)
assert matched is True
assert remapped == local_prefix / "author" / "book.epub"
+12 -21
View File
@@ -1,9 +1,7 @@
"""Integration tests for real filesystem processing flows."""
import errno
import os
import zipfile
from contextlib import nullcontext
from pathlib import Path
from threading import Event
from unittest.mock import MagicMock, patch
@@ -48,15 +46,6 @@ def _sync_config(mock_config, mock_core):
mock_core.CUSTOM_SCRIPT = mock_config.CUSTOM_SCRIPT
def _hardlink_support_patch(supported: bool):
if supported:
return nullcontext()
return patch(
"shelfmark.download.fs.os.link",
side_effect=OSError(errno.EXDEV, "Invalid cross-device link"),
)
def test_direct_download_rename_moves_file(tmp_path):
from shelfmark.download.postprocess.router import (
post_process_download as _post_process_download,
@@ -377,7 +366,7 @@ def test_torrent_hardlink_enabled_copy_fallback_does_not_extract_archives(tmp_pa
with (
patch("shelfmark.core.config.config") as mock_config,
patch("shelfmark.config.env.TMP_DIR", staging),
_hardlink_support_patch(False),
patch("shelfmark.download.postprocess.transfer.same_filesystem", return_value=False),
):
mock_config.get = _build_config(ingest, organization="none", hardlink=True)
mock_config.CUSTOM_SCRIPT = None
@@ -396,7 +385,7 @@ def test_torrent_hardlink_enabled_copy_fallback_does_not_extract_archives(tmp_pa
# Most importantly: hardlink-setting-enabled fallback to copy should NOT extract.
assert list(ingest.glob("*.epub")) == []
assert any(msg.startswith("Hardlinking") for _, msg in statuses)
assert any(msg.startswith("Copying") for _, msg in statuses)
def test_torrent_hardlink_enabled_copy_fallback_directory_archive_kept_when_zip_supported(tmp_path):
@@ -433,7 +422,7 @@ def test_torrent_hardlink_enabled_copy_fallback_directory_archive_kept_when_zip_
with (
patch("shelfmark.core.config.config") as mock_config,
patch("shelfmark.config.env.TMP_DIR", staging),
_hardlink_support_patch(False),
patch("shelfmark.download.postprocess.transfer.same_filesystem", return_value=False),
):
mock_config.get = _build_config(
ingest,
@@ -1039,20 +1028,20 @@ def test_postprocess_folder_blackbox_matrix(
@pytest.mark.parametrize("content_kind", ["book", "audiobook"])
@pytest.mark.parametrize("organization", ["none", "organize"])
@pytest.mark.parametrize("hardlink_enabled", [False, True])
@pytest.mark.parametrize("hardlink_supported", [True, False])
@pytest.mark.parametrize("same_filesystem", [True, False])
def test_postprocess_torrent_blackbox_matrix(
tmp_path,
input_kind: str,
content_kind: str,
organization: str,
hardlink_enabled: bool,
hardlink_supported: bool,
same_filesystem: bool,
):
"""Torrent-like (original_download_path set) black-box test matrix.
This exercises:
- hardlink enabled/disabled
- successful hardlink vs copy fallback
- same-filesystem hardlink vs copy fallback
- content type differences (book vs audiobook)
Assertions focus on invariants:
@@ -1098,7 +1087,7 @@ def test_postprocess_torrent_blackbox_matrix(
source_file.write_text("content")
task = DownloadTask(
task_id=f"torrent-matrix-{input_kind}-{content_kind}-{organization}-{hardlink_enabled}-{hardlink_supported}",
task_id=f"torrent-matrix-{input_kind}-{content_kind}-{organization}-{hardlink_enabled}-{same_filesystem}",
source="prowlarr",
title=title,
author=author,
@@ -1111,7 +1100,9 @@ def test_postprocess_torrent_blackbox_matrix(
with (
patch("shelfmark.core.config.config") as mock_config,
patch("shelfmark.config.env.TMP_DIR", staging),
_hardlink_support_patch(hardlink_supported),
patch(
"shelfmark.download.postprocess.transfer.same_filesystem", return_value=same_filesystem
),
):
mock_config.get = _build_config(
ingest,
@@ -1140,8 +1131,8 @@ def test_postprocess_torrent_blackbox_matrix(
assert result_path.parent == ingest
assert result_path.name == f"random.{extension}"
# Hardlink only when enabled and supported by the filesystem.
if hardlink_enabled and hardlink_supported:
# Hardlink only when enabled and same filesystem.
if hardlink_enabled and same_filesystem:
assert os.stat(source_file).st_ino == os.stat(result_path).st_ino
else:
assert os.stat(source_file).st_ino != os.stat(result_path).st_ino
+6 -182
View File
@@ -354,166 +354,6 @@ class TestRequestRoutes:
mock_notify_admin.assert_not_called()
mock_notify_user.assert_not_called()
def test_download_policy_rejects_mismatched_context_and_release_source(
self, main_module, client
):
user = _create_user(main_module, prefix="reader")
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
policy = _policy(
default_ebook="download",
rules=[{"source": "direct_download", "content_type": "*", "mode": "blocked"}],
)
payload = {
"book_data": {
"title": "Policy Source Mismatch",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "policy-source-mismatch-1",
},
"context": {
"source": "prowlarr",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "direct_download",
"source_id": "blocked-release-1",
"title": "Blocked Release.epub",
},
}
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module, "load_users_request_policy_settings", return_value=policy
):
with patch(
"shelfmark.core.request_routes.load_users_request_policy_settings",
return_value=policy,
):
with patch.object(main_module.backend, "queue_release") as mock_queue:
resp = client.post("/api/requests", json=payload)
assert resp.status_code == 400
assert resp.json["code"] == "policy_source_mismatch"
assert resp.json["error"] == "Policy context source must match release_data.source"
assert main_module.user_db.list_requests(user_id=user["id"]) == []
mock_queue.assert_not_called()
def test_release_result_source_rejects_mismatch_before_normalization(self, main_module, client):
user = _create_user(main_module, prefix="reader")
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
policy = _policy(
default_ebook="download",
rules=[{"source": "prowlarr", "content_type": "*", "mode": "blocked"}],
)
payload = {
"book_data": {
"title": "Release Result Source Mismatch",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "release-result-mismatch-1",
},
"context": {
"source": "direct_download",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "prowlarr",
"source_id": "blocked-prowlarr-release-1",
"title": "Blocked Prowlarr Release.epub",
},
}
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module, "load_users_request_policy_settings", return_value=policy
):
with patch(
"shelfmark.core.request_routes.load_users_request_policy_settings",
return_value=policy,
):
with patch.object(main_module.backend, "queue_release") as mock_queue:
resp = client.post("/api/requests", json=payload)
assert resp.status_code == 400
assert resp.json["code"] == "policy_source_mismatch"
assert resp.json["error"] == "Policy context source must match release_data.source"
assert main_module.user_db.list_requests(user_id=user["id"]) == []
mock_queue.assert_not_called()
def test_batch_rejects_release_result_source_mismatch_before_creating_any_requests(
self, main_module, client
):
user = _create_user(main_module, prefix="reader")
_set_session(client, user_id=user["username"], db_user_id=user["id"], is_admin=False)
policy = _policy(
default_ebook="request_release",
rules=[{"source": "prowlarr", "content_type": "*", "mode": "blocked"}],
)
payloads = [
{
"book_data": {
"title": "Batch Valid Direct",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "batch-valid-direct-1",
},
"context": {
"source": "direct_download",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "direct_download",
"source_id": "batch-valid-direct-release-1",
"title": "Batch Valid Direct.epub",
},
},
{
"book_data": {
"title": "Batch Release Result Mismatch",
"author": "Shelfmark",
"content_type": "ebook",
"provider": "openlibrary",
"provider_id": "batch-release-result-mismatch-1",
},
"context": {
"source": "direct_download",
"content_type": "ebook",
"request_level": "release",
},
"release_data": {
"source": "prowlarr",
"source_id": "batch-blocked-prowlarr-release-1",
"title": "Batch Blocked Prowlarr.epub",
},
},
]
with patch.object(main_module, "get_auth_mode", return_value="builtin"):
with patch.object(
main_module, "load_users_request_policy_settings", return_value=policy
):
with patch(
"shelfmark.core.request_routes.load_users_request_policy_settings",
return_value=policy,
):
with patch.object(main_module.backend, "queue_release") as mock_queue:
resp = client.post("/api/requests/batch", json={"requests": payloads})
assert resp.status_code == 400
assert resp.json["code"] == "policy_source_mismatch"
assert resp.json["error"] == "Policy context source must match release_data.source"
assert main_module.user_db.list_requests(user_id=user["id"]) == []
mock_queue.assert_not_called()
def test_batch_download_policy_queues_releases_without_creating_requests(
self, main_module, client
):
@@ -1692,12 +1532,6 @@ class TestRequestRoutes:
def test_admin_fulfil_uses_real_queue_and_preserves_requesting_identity(
self, main_module, client
):
class AvailableSource:
display_name = "Direct Download"
def is_available(self):
return True
user = _create_user(main_module, prefix="reader")
other_user = _create_user(main_module, prefix="reader")
admin = _create_user(main_module, prefix="admin", role="admin")
@@ -1733,23 +1567,13 @@ class TestRequestRoutes:
"shelfmark.core.request_routes.load_users_request_policy_settings",
return_value=policy,
):
with patch.object(
main_module.backend,
"get_source",
return_value=AvailableSource(),
):
create_resp = client.post("/api/requests", json=create_payload)
request_id = create_resp.json["id"]
create_resp = client.post("/api/requests", json=create_payload)
request_id = create_resp.json["id"]
_set_session(
client,
user_id=admin["username"],
db_user_id=admin["id"],
is_admin=True,
)
fulfil_resp = client.post(
f"/api/admin/requests/{request_id}/fulfil", json={}
)
_set_session(
client, user_id=admin["username"], db_user_id=admin["id"], is_admin=True
)
fulfil_resp = client.post(f"/api/admin/requests/{request_id}/fulfil", json={})
assert fulfil_resp.status_code == 200
assert fulfil_resp.json["status"] == "fulfilled"
-78
View File
@@ -1,78 +0,0 @@
from __future__ import annotations
from pathlib import Path
from shelfmark.download import archive as archive_mod
class _FakeZipInfo:
filename = "book.epub"
flag_bits = 0
def is_dir(self) -> bool:
return False
class _ChunkOnlyStream:
def __init__(self, content: bytes) -> None:
self._content = content
self._offset = 0
self.whole_read_called = False
def __enter__(self) -> _ChunkOnlyStream:
return self
def __exit__(self, *_args: object) -> None:
return None
def read(self, size: int = -1) -> bytes:
if size < 0:
self.whole_read_called = True
msg = "archive member was read into memory"
raise AssertionError(msg)
chunk = self._content[self._offset : self._offset + size]
self._offset += len(chunk)
return chunk
class _FakeZipFile:
stream: _ChunkOnlyStream
def __init__(self, _path: Path, _mode: str) -> None:
self.stream = _ChunkOnlyStream(b"streamed archive content")
def __enter__(self) -> _FakeZipFile:
return self
def __exit__(self, *_args: object) -> None:
return None
def infolist(self) -> list[_FakeZipInfo]:
return [_FakeZipInfo()]
def testzip(self) -> None:
return None
def open(self, _info: _FakeZipInfo) -> _ChunkOnlyStream:
return self.stream
def test_extract_archive_raw_streams_members_without_whole_read(
tmp_path: Path, monkeypatch
) -> None:
fake_archives: list[_FakeZipFile] = []
def fake_zip_file(path: Path, mode: str) -> _FakeZipFile:
archive = _FakeZipFile(path, mode)
fake_archives.append(archive)
return archive
monkeypatch.setattr(archive_mod.zipfile, "ZipFile", fake_zip_file)
extracted_files, warnings = archive_mod.extract_archive_raw(tmp_path / "book.zip", tmp_path)
assert warnings == []
assert [path.name for path in extracted_files] == ["book.epub"]
assert extracted_files[0].read_bytes() == b"streamed archive content"
assert fake_archives[0].stream.whole_read_called is False
-16
View File
@@ -4,26 +4,10 @@ from pathlib import Path
from threading import Event
from unittest.mock import MagicMock
import pytest
from shelfmark.core.models import DownloadTask, QueueStatus
from shelfmark.core.queue import BookQueue
class _AvailableSource:
display_name = "Test Source"
def is_available(self):
return True
@pytest.fixture(autouse=True)
def source_available_by_default(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
monkeypatch.setattr(orchestrator, "get_source", lambda _source: _AvailableSource())
def test_retry_download_requeues_error_task(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
@@ -1,40 +1,6 @@
from threading import Event
from unittest.mock import MagicMock
import pytest
from shelfmark.core.models import DownloadTask, SearchMode
class _AvailableSource:
display_name = "Test Source"
def is_available(self):
return True
class _UnavailableSource:
display_name = "Direct Download"
def is_available(self):
return False
@pytest.fixture(autouse=True)
def source_available_by_default(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
monkeypatch.setattr(orchestrator, "get_source", lambda _source: _AvailableSource())
def enable_prowlarr_seed_preferences(monkeypatch, orchestrator):
monkeypatch.setattr(
orchestrator.config,
"get",
lambda key, default=None, user_id=None: (
True if key == "PROWLARR_USE_SEED_PREFERENCES" else default
),
)
from shelfmark.core.models import SearchMode
def test_queue_release_uses_user_specific_books_output_mode(monkeypatch):
@@ -110,28 +76,6 @@ def test_queue_release_preserves_direct_search_mode_from_payload(monkeypatch):
assert captured["task"].search_mode == SearchMode.DIRECT
def test_queue_release_rejects_unavailable_source(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
monkeypatch.setattr(orchestrator, "get_source", lambda _source: _UnavailableSource())
monkeypatch.setattr(orchestrator.book_queue, "add", MagicMock())
success, error = orchestrator.queue_release(
{
"source": "direct_download",
"source_id": "release-disabled-direct",
"title": "Disabled Direct Release",
"content_type": "ebook",
},
user_id=42,
username="alice",
)
assert success is False
assert error == "Direct Download is unavailable. Enable and configure the source in Settings."
orchestrator.book_queue.add.assert_not_called()
def test_queue_release_email_mode_without_recipient_is_queued(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
@@ -170,41 +114,6 @@ def test_queue_release_email_mode_without_recipient_is_queued(monkeypatch):
assert task.output_args == {}
def test_download_task_rejects_unavailable_source_before_handler(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
task = DownloadTask(
task_id="disabled-task",
source="direct_download",
title="Disabled Direct Release",
)
status_messages: list[tuple[str, str]] = []
monkeypatch.setattr(orchestrator, "get_source", lambda _source: _UnavailableSource())
monkeypatch.setattr(orchestrator, "get_handler", MagicMock())
monkeypatch.setattr(orchestrator.book_queue, "get_task", lambda _task_id: task)
monkeypatch.setattr(
orchestrator.book_queue,
"update_status_message",
lambda task_id, message: status_messages.append((task_id, message)),
)
result = orchestrator._download_task("disabled-task", Event())
assert result is None
assert task.last_error_type == "SourceUnavailable"
assert task.last_error_message == (
"Direct Download is unavailable. Enable and configure the source in Settings."
)
assert status_messages == [
(
"disabled-task",
"Direct Download is unavailable. Enable and configure the source in Settings.",
)
]
orchestrator.get_handler.assert_not_called()
def test_queue_release_persists_generic_retry_resolution_fields(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
@@ -216,7 +125,6 @@ def test_queue_release_persists_generic_retry_resolution_fields(monkeypatch):
monkeypatch.setattr(orchestrator.book_queue, "add", fake_add)
monkeypatch.setattr(orchestrator, "ws_manager", None)
enable_prowlarr_seed_preferences(monkeypatch, orchestrator)
success, error = orchestrator.queue_release(
{
@@ -227,8 +135,8 @@ def test_queue_release_persists_generic_retry_resolution_fields(monkeypatch):
"protocol": "torrent",
"indexer": "MyIndexer",
"extra": {
"configured_ratio_limit": 1.25,
"configured_seed_time_minutes": 90,
"minimum_ratio": 1.25,
"minimum_seed_time": 5400,
"info_hash": "ABC123",
},
},
@@ -248,114 +156,6 @@ def test_queue_release_persists_generic_retry_resolution_fields(monkeypatch):
assert task.can_retry_without_staged_source is True
def test_queue_release_prefers_configured_seed_time_minutes_for_retry(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
captured: dict[str, object] = {}
def fake_add(task):
captured["task"] = task
return True
monkeypatch.setattr(orchestrator.book_queue, "add", fake_add)
monkeypatch.setattr(orchestrator, "ws_manager", None)
enable_prowlarr_seed_preferences(monkeypatch, orchestrator)
success, error = orchestrator.queue_release(
{
"source": "prowlarr",
"source_id": "prowlarr-release-configured-seed-time",
"title": "Queued Prowlarr Release",
"download_url": "magnet:?xt=urn:btih:abc123",
"protocol": "torrent",
"extra": {
"configured_ratio_limit": 2,
"configured_seed_time_minutes": 7200,
"minimum_ratio": 1,
"minimum_seed_time": 259200,
},
},
user_id=42,
username="alice",
)
assert success is True
assert error is None
task = captured["task"]
assert task.retry_ratio_limit == 2.0
assert task.retry_seeding_time_limit_minutes == 7200
def test_queue_release_ignores_configured_seed_time_when_disabled_for_retry(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
captured: dict[str, object] = {}
def fake_add(task):
captured["task"] = task
return True
monkeypatch.setattr(orchestrator.book_queue, "add", fake_add)
monkeypatch.setattr(orchestrator, "ws_manager", None)
success, error = orchestrator.queue_release(
{
"source": "prowlarr",
"source_id": "prowlarr-release-configured-seed-time-disabled",
"title": "Queued Prowlarr Release",
"download_url": "magnet:?xt=urn:btih:abc123",
"protocol": "torrent",
"extra": {
"configured_ratio_limit": 2,
"configured_seed_time_minutes": 7200,
},
},
user_id=42,
username="alice",
)
assert success is True
assert error is None
task = captured["task"]
assert task.retry_ratio_limit is None
assert task.retry_seeding_time_limit_minutes is None
def test_queue_release_ignores_torznab_minimum_seed_criteria_for_retry(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
captured: dict[str, object] = {}
def fake_add(task):
captured["task"] = task
return True
monkeypatch.setattr(orchestrator.book_queue, "add", fake_add)
monkeypatch.setattr(orchestrator, "ws_manager", None)
success, error = orchestrator.queue_release(
{
"source": "prowlarr",
"source_id": "prowlarr-release-minimum-only",
"title": "Queued Prowlarr Release",
"download_url": "magnet:?xt=urn:btih:abc123",
"protocol": "torrent",
"extra": {
"minimum_ratio": 1,
"minimum_seed_time": 259200,
},
},
user_id=42,
username="alice",
)
assert success is True
assert error is None
task = captured["task"]
assert task.retry_ratio_limit is None
assert task.retry_seeding_time_limit_minutes is None
def test_queue_release_returns_error_for_operational_queue_failure(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
-84
View File
@@ -1,84 +0,0 @@
from ipaddress import IPv4Address
from shelfmark.release_sources.irc.client import IRCClient, IRCEvent, IRCMessage
def _dcc_raw(sender: str, filename: str, ip: str, port: int = 443) -> str:
ip_int = int(IPv4Address(ip))
return f':{sender}!user@example.test PRIVMSG reader :\x01DCC SEND "{filename}" {ip_int} {port} 1\x01'
def test_wait_for_dcc_ignores_unexpected_sender(monkeypatch) -> None:
client = IRCClient(nick="reader", server="irc.example.test", port=6697)
client.online_servers = {"BookBot"}
messages = [
IRCMessage(
raw=_dcc_raw("Mallory", "evil.epub", "8.8.8.8"),
prefix="Mallory!user@example.test",
event=IRCEvent.BOOK_RESULT,
),
IRCMessage(
raw=_dcc_raw("BookBot", "book.epub", "8.8.8.8"),
prefix="BookBot!user@example.test",
event=IRCEvent.BOOK_RESULT,
),
]
monkeypatch.setattr(client, "read_messages", lambda: iter(messages))
offer = client.wait_for_dcc(timeout=1.0, result_type=False)
assert offer is not None
assert offer.filename == "book.epub"
def test_wait_for_dcc_uses_expected_sender_over_online_server_list(monkeypatch) -> None:
client = IRCClient(nick="reader", server="irc.example.test", port=6697)
client.online_servers = {"OtherBot"}
messages = [
IRCMessage(
raw=_dcc_raw("BookBot", "book.epub", "8.8.8.8"),
prefix="BookBot!user@example.test",
event=IRCEvent.BOOK_RESULT,
)
]
monkeypatch.setattr(client, "read_messages", lambda: iter(messages))
offer = client.wait_for_dcc(
timeout=1.0,
result_type=False,
expected_senders={"BookBot"},
)
assert offer is not None
assert offer.filename == "book.epub"
def test_wait_for_dcc_ignores_unsafe_offer_and_keeps_waiting(monkeypatch) -> None:
client = IRCClient(nick="reader", server="irc.example.test", port=6697)
client.online_servers = {"BookBot"}
messages = [
IRCMessage(
raw=_dcc_raw("BookBot", "../outside.epub", "8.8.8.8"),
prefix="BookBot!user@example.test",
event=IRCEvent.BOOK_RESULT,
),
IRCMessage(
raw=_dcc_raw("BookBot", "internal.epub", "127.0.0.1"),
prefix="BookBot!user@example.test",
event=IRCEvent.BOOK_RESULT,
),
IRCMessage(
raw=_dcc_raw("BookBot", "book.epub", "8.8.8.8"),
prefix="BookBot!user@example.test",
event=IRCEvent.BOOK_RESULT,
),
]
monkeypatch.setattr(client, "read_messages", lambda: iter(messages))
offer = client.wait_for_dcc(timeout=1.0, result_type=False)
assert offer is not None
assert offer.filename == "book.epub"
-82
View File
@@ -1,82 +0,0 @@
import socket
import pytest
from shelfmark.release_sources.irc.dcc import (
DCCOffer,
DCCParseError,
DCCSecurityError,
download_dcc,
parse_dcc_send,
safe_dcc_filename,
validate_dcc_endpoint,
)
def test_safe_dcc_filename_allows_plain_filenames() -> None:
assert safe_dcc_filename("results.txt") == "results.txt"
assert safe_dcc_filename("Author - Title.epub") == "Author - Title.epub"
@pytest.mark.parametrize(
"filename",
[
"",
".",
"..",
"../outside.txt",
"/tmp/outside.txt",
r"..\outside.txt",
r"C:\temp\outside.txt",
],
)
def test_safe_dcc_filename_rejects_paths(filename: str) -> None:
with pytest.raises(DCCSecurityError):
safe_dcc_filename(filename)
@pytest.mark.parametrize(
"ip",
[
"127.0.0.1",
"10.0.0.1",
"172.16.0.1",
"192.168.1.1",
"169.254.169.254",
"0.0.0.0",
],
)
def test_validate_dcc_endpoint_rejects_non_public_ips(ip: str) -> None:
with pytest.raises(DCCSecurityError):
validate_dcc_endpoint(DCCOffer(filename="book.epub", ip=ip, port=1234, size=1))
@pytest.mark.parametrize("port", [0, 65536])
def test_validate_dcc_endpoint_rejects_invalid_ports(port: int) -> None:
with pytest.raises(DCCSecurityError):
validate_dcc_endpoint(DCCOffer(filename="book.epub", ip="8.8.8.8", port=port, size=1))
def test_validate_dcc_endpoint_allows_public_endpoint() -> None:
validate_dcc_endpoint(DCCOffer(filename="book.epub", ip="8.8.8.8", port=443, size=1))
def test_parse_dcc_send_rejects_out_of_range_ip_integer() -> None:
with pytest.raises(DCCParseError):
parse_dcc_send('DCC SEND "book.epub" 999999999999999999 443 1')
def test_download_dcc_rejects_private_endpoint_before_socket_connect(
monkeypatch,
tmp_path,
) -> None:
def fail_socket(*_args: object, **_kwargs: object) -> socket.socket:
raise AssertionError("download should reject the endpoint before opening a socket")
monkeypatch.setattr(socket, "socket", fail_socket)
with pytest.raises(DCCSecurityError):
download_dcc(
DCCOffer(filename="book.epub", ip="127.0.0.1", port=1234, size=1),
tmp_path / "book.epub",
)
-53
View File
@@ -1,59 +1,6 @@
import requests
from shelfmark.core.cache import get_metadata_cache
from shelfmark.metadata_providers import MetadataSearchOptions
from shelfmark.metadata_providers.googlebooks import GoogleBooksProvider
class _GoogleBooksResponse:
def __init__(self, payload):
self._payload = payload
def raise_for_status(self):
return None
def json(self):
return self._payload
class _FlakyGoogleBooksSession:
def __init__(self):
self.calls = 0
def get(self, *args, **kwargs):
self.calls += 1
if self.calls == 1:
raise requests.Timeout
return _GoogleBooksResponse(
{
"items": [
{
"id": "volume-1",
"volumeInfo": {
"title": "Recovered Book",
"authors": ["Alice Author"],
},
}
]
}
)
def test_googlebooks_search_does_not_cache_request_failures():
get_metadata_cache().clear()
provider = GoogleBooksProvider(api_key="test-key")
session = _FlakyGoogleBooksSession()
provider.session = session
options = MetadataSearchOptions(query="Recovered Book")
assert provider.search(options) == []
result = provider.search(options)
assert session.calls == 2
assert [book.title for book in result] == ["Recovered Book"]
class TestGoogleBooksParseVolume:
def test_parse_volume_returns_metadata_for_valid_payload(self):
provider = GoogleBooksProvider(api_key="test-key")
-22
View File
@@ -96,28 +96,6 @@ class TestGetDownloadUrl:
class TestHandlerErrors:
def test_cache_miss_uses_persisted_retry_fields(self):
with patch("shelfmark.release_sources.newznab.handler.get_release", return_value=None):
handler = NewznabHandler()
task = DownloadTask(
task_id="retryable",
source="newznab",
title="Book",
retry_download_url="https://indexer.example.com/nzb/42?apikey=secret",
retry_download_protocol="usenet",
retry_release_name="Book Release",
retry_expected_hash="abc123",
)
recorder = ProgressRecorder()
result = handler._resolve_download(task, recorder.status_callback)
assert result is not None
assert result.url == "https://indexer.example.com/nzb/42?apikey=secret"
assert result.protocol == "usenet"
assert result.release_name == "Book Release"
assert result.expected_hash == "abc123"
assert recorder.status_updates == []
def test_cache_miss_returns_error(self):
with patch("shelfmark.release_sources.newznab.handler.get_release", return_value=None):
handler = NewznabHandler()
+1 -44
View File
@@ -2,8 +2,6 @@
from unittest.mock import MagicMock
import pytest
from shelfmark.core.search_plan import ReleaseSearchPlan, ReleaseSearchVariant
from shelfmark.metadata_providers import BookMetadata
from shelfmark.release_sources import ReleaseProtocol
@@ -15,20 +13,6 @@ from shelfmark.release_sources.newznab.source import (
# ── fixtures / helpers ─────────────────────────────────────────────────────────
class _AvailableSource:
display_name = "Newznab"
def is_available(self):
return True
@pytest.fixture(autouse=True)
def source_available_by_default(monkeypatch):
import shelfmark.download.orchestrator as orchestrator
monkeypatch.setattr(orchestrator, "get_source", lambda _source: _AvailableSource())
def _make_book(**kwargs) -> BookMetadata:
defaults = {
"provider": "hardcover",
@@ -76,7 +60,7 @@ class TestResultToRelease:
assert r.size_bytes == 2097152
assert r.indexer == "MyIndexer"
assert r.source_id == "https://indexer.example.com/nzb/42"
assert r.download_url is None
assert r.download_url == "https://indexer.example.com/nzb/42?apikey=secret"
def test_torrent_result_has_torrent_protocol(self):
r = _newznab_result_to_release(
@@ -150,33 +134,6 @@ class TestResultToRelease:
assert r.extra["book_title"] == "Dune"
assert r.extra["info_hash"] == "abc123"
def test_redacted_result_still_builds_private_retry_payload(self, monkeypatch):
import shelfmark.download.orchestrator as orchestrator
secret_download_url = "https://indexer.example.com/nzb/42?apikey=secret"
release = _newznab_result_to_release(_make_result(downloadUrl=secret_download_url))
assert release.download_url is None
captured_tasks = []
def fake_add(task):
captured_tasks.append(task)
return True
monkeypatch.setattr(orchestrator.book_queue, "add", fake_add)
monkeypatch.setattr(
orchestrator.config, "get", lambda key, default=None, user_id=None: default
)
success, error = orchestrator.queue_release(release.__dict__)
assert success is True
assert error is None
assert captured_tasks[0].retry_download_url == secret_download_url
assert captured_tasks[0].retry_download_protocol == "usenet"
assert "retry_download_url" not in orchestrator._task_to_dict(captured_tasks[0])
# ── NewznabSource.is_available ─────────────────────────────────────────────────
+16 -53
View File
@@ -236,7 +236,7 @@ class TestProwlarrHandlerDownloadErrors:
class TestProwlarrHandlerSeedCriteria:
"""Tests for seed criteria passed through from Prowlarr."""
def test_resolve_download_ignores_torznab_minimum_seed_criteria(self):
def test_resolve_download_converts_seed_time_seconds_to_minutes(self):
with patch(
"shelfmark.release_sources.prowlarr.handler.get_release",
return_value={
@@ -257,28 +257,22 @@ class TestProwlarrHandlerSeedCriteria:
request = handler._resolve_download(task, lambda *_: None)
assert request is not None
assert request.seeding_time_limit is None
assert request.ratio_limit is None
assert request.seeding_time_limit == 4320
assert request.ratio_limit == 1.0
def test_resolve_download_uses_configured_seed_time_minutes(self):
with (
patch(
"shelfmark.release_sources.prowlarr.handler.get_release",
return_value={
"protocol": "torrent",
"title": "Test Release",
"magnetUrl": "magnet:?xt=urn:btih:abc123",
"configuredSeedTimeMinutes": 7200,
"configuredRatioLimit": 2,
"minimumSeedTime": 259200,
"minimumRatio": 1,
},
),
patch("shelfmark.release_sources.prowlarr.handler.config.get", return_value=True),
def test_resolve_download_rounds_seed_time_up_to_next_minute(self):
with patch(
"shelfmark.release_sources.prowlarr.handler.get_release",
return_value={
"protocol": "torrent",
"title": "Test Release",
"magnetUrl": "magnet:?xt=urn:btih:abc123",
"minimumSeedTime": 61,
},
):
handler = ProwlarrHandler()
task = DownloadTask(
task_id="configured-seed-time",
task_id="seed-time-round-up",
source="prowlarr",
title="Test Book",
)
@@ -286,35 +280,7 @@ class TestProwlarrHandlerSeedCriteria:
request = handler._resolve_download(task, lambda *_: None)
assert request is not None
assert request.seeding_time_limit == 7200
assert request.ratio_limit == 2.0
def test_resolve_download_ignores_configured_seed_time_when_disabled(self):
with (
patch(
"shelfmark.release_sources.prowlarr.handler.get_release",
return_value={
"protocol": "torrent",
"title": "Test Release",
"magnetUrl": "magnet:?xt=urn:btih:abc123",
"configuredSeedTimeMinutes": 7200,
"configuredRatioLimit": 2,
},
),
patch("shelfmark.release_sources.prowlarr.handler.config.get", return_value=False),
):
handler = ProwlarrHandler()
task = DownloadTask(
task_id="configured-seed-time-disabled",
source="prowlarr",
title="Test Book",
)
request = handler._resolve_download(task, lambda *_: None)
assert request is not None
assert request.seeding_time_limit is None
assert request.ratio_limit is None
assert request.seeding_time_limit == 2
def test_download_passes_seed_limits_to_client(self):
mock_client = MagicMock()
@@ -329,10 +295,8 @@ class TestProwlarrHandlerSeedCriteria:
"protocol": "torrent",
"title": "Test Release",
"magnetUrl": "magnet:?xt=urn:btih:abc123",
"configuredSeedTimeMinutes": 7200,
"configuredRatioLimit": 1.25,
"minimumSeedTime": 259200,
"minimumRatio": 1,
"minimumRatio": 1.25,
},
),
patch(
@@ -342,7 +306,6 @@ class TestProwlarrHandlerSeedCriteria:
patch(
"shelfmark.release_sources.prowlarr.handler.remove_release",
),
patch("shelfmark.release_sources.prowlarr.handler.config.get", return_value=True),
patch.object(
ProwlarrHandler,
"_poll_and_complete",
@@ -366,7 +329,7 @@ class TestProwlarrHandlerSeedCriteria:
)
call_kwargs = mock_client.add_download.call_args.kwargs
assert call_kwargs["seeding_time_limit"] == 7200
assert call_kwargs["seeding_time_limit"] == 4320
assert call_kwargs["ratio_limit"] == 1.25

Some files were not shown because too many files have changed in this diff Show More