feat: Add subtitle support in OPDSEntry and enhance metadata handling in API routes

This commit is contained in:
JB
2025-12-20 08:27:47 -08:00
parent eabfe87ffb
commit 19e98c3ad6
7 changed files with 293 additions and 60 deletions
+11
View File
@@ -77,6 +77,7 @@ class OPDSEntry:
title: str title: str
position: Optional[int] = None position: Optional[int] = None
authors: List[str] = field(default_factory=list) authors: List[str] = field(default_factory=list)
subtitle: Optional[str] = None
updated: Optional[str] = None updated: Optional[str] = None
published: Optional[str] = None published: Optional[str] = None
summary: Optional[str] = None summary: Optional[str] = None
@@ -96,6 +97,7 @@ class OPDSEntry:
"title": self.title, "title": self.title,
"position": self.position, "position": self.position,
"authors": list(self.authors), "authors": list(self.authors),
"subtitle": self.subtitle,
"updated": self.updated, "updated": self.updated,
"published": self.published, "published": self.published,
"summary": self.summary, "summary": self.summary,
@@ -599,6 +601,14 @@ class CalibreOPDSClient:
def _parse_entry(self, node: ET.Element, base_url: str) -> OPDSEntry: def _parse_entry(self, node: ET.Element, base_url: str) -> OPDSEntry:
entry_id = node.findtext("atom:id", default="", namespaces=NS).strip() entry_id = node.findtext("atom:id", default="", namespaces=NS).strip()
title = node.findtext("atom:title", default="Untitled", namespaces=NS).strip() or "Untitled" title = node.findtext("atom:title", default="Untitled", namespaces=NS).strip() or "Untitled"
subtitle = (
node.findtext("calibre_md:subtitle", default=None, namespaces=NS)
or node.findtext("calibre:subtitle", default=None, namespaces=NS)
or node.findtext("atom:subtitle", default=None, namespaces=NS)
)
subtitle = self._strip_html(subtitle.strip()) if subtitle else None
position_value = self._extract_position(node) position_value = self._extract_position(node)
updated = node.findtext("atom:updated", default=None, namespaces=NS) updated = node.findtext("atom:updated", default=None, namespaces=NS)
published = ( published = (
@@ -691,6 +701,7 @@ class CalibreOPDSClient:
title=title, title=title,
position=position_value, position=position_value,
authors=authors, authors=authors,
subtitle=subtitle,
updated=updated, updated=updated,
published=published, published=published,
summary=cleaned_summary, summary=cleaned_summary,
+80 -58
View File
@@ -260,6 +260,85 @@ def api_speaker_preview() -> ResponseReturnValue:
# --- Integration Routes --- # --- Integration Routes ---
def _opds_metadata_overrides(metadata_payload: Mapping[str, Any]) -> Dict[str, Any]:
metadata_overrides: Dict[str, Any] = {}
def _stringify_metadata_value(value: Any) -> str:
if value is None:
return ""
if isinstance(value, (list, tuple, set)):
parts = [str(item).strip() for item in value if item is not None]
parts = [part for part in parts if part]
return ", ".join(parts)
return str(value).strip()
raw_series = metadata_payload.get("series") or metadata_payload.get("series_name")
series_name = str(raw_series or "").strip()
if series_name:
metadata_overrides["series"] = series_name
metadata_overrides.setdefault("series_name", series_name)
series_index_value = (
metadata_payload.get("series_index")
or metadata_payload.get("series_position")
or metadata_payload.get("series_sequence")
or metadata_payload.get("book_number")
)
if series_index_value is not None:
series_index_text = str(series_index_value).strip()
if series_index_text:
metadata_overrides.setdefault("series_index", series_index_text)
metadata_overrides.setdefault("series_position", series_index_text)
metadata_overrides.setdefault("series_sequence", series_index_text)
metadata_overrides.setdefault("book_number", series_index_text)
tags_value = metadata_payload.get("tags") or metadata_payload.get("keywords")
if tags_value:
tags_text = _stringify_metadata_value(tags_value)
if tags_text:
metadata_overrides.setdefault("tags", tags_text)
metadata_overrides.setdefault("keywords", tags_text)
metadata_overrides.setdefault("genre", tags_text)
description_value = metadata_payload.get("description") or metadata_payload.get("summary")
if description_value:
description_text = _stringify_metadata_value(description_value)
if description_text:
metadata_overrides.setdefault("description", description_text)
metadata_overrides.setdefault("summary", description_text)
subtitle_value = (
metadata_payload.get("subtitle")
or metadata_payload.get("sub_title")
or metadata_payload.get("calibre_subtitle")
)
if subtitle_value:
subtitle_text = _stringify_metadata_value(subtitle_value)
if subtitle_text:
metadata_overrides.setdefault("subtitle", subtitle_text)
publisher_value = metadata_payload.get("publisher")
if publisher_value:
publisher_text = _stringify_metadata_value(publisher_value)
if publisher_text:
metadata_overrides.setdefault("publisher", publisher_text)
# Author mapping: Abogen templates look for either 'authors' or 'author'.
authors_value = (
metadata_payload.get("authors")
or metadata_payload.get("author")
or metadata_payload.get("creator")
or metadata_payload.get("dc_creator")
)
if authors_value:
authors_text = _stringify_metadata_value(authors_value)
if authors_text:
metadata_overrides.setdefault("authors", authors_text)
metadata_overrides.setdefault("author", authors_text)
return metadata_overrides
@api_bp.get("/integrations/calibre-opds/feed") @api_bp.get("/integrations/calibre-opds/feed")
def api_calibre_opds_feed() -> ResponseReturnValue: def api_calibre_opds_feed() -> ResponseReturnValue:
integrations = load_integration_settings() integrations = load_integration_settings()
@@ -383,65 +462,8 @@ def api_calibre_opds_import() -> ResponseReturnValue:
metadata_payload = data.get("metadata") if isinstance(data, Mapping) else None metadata_payload = data.get("metadata") if isinstance(data, Mapping) else None
metadata_overrides: Dict[str, Any] = {} metadata_overrides: Dict[str, Any] = {}
if isinstance(metadata_payload, Mapping): if isinstance(metadata_payload, Mapping):
def _stringify_metadata_value(value: Any) -> str: metadata_overrides = _opds_metadata_overrides(metadata_payload)
if value is None:
return ""
if isinstance(value, (list, tuple, set)):
parts = [str(item).strip() for item in value if item is not None]
parts = [part for part in parts if part]
return ", ".join(parts)
text = str(value).strip()
return text
raw_series = metadata_payload.get("series") or metadata_payload.get("series_name")
series_name = str(raw_series or "").strip()
if series_name:
metadata_overrides["series"] = series_name
metadata_overrides.setdefault("series_name", series_name)
series_index_value = (
metadata_payload.get("series_index")
or metadata_payload.get("series_position")
or metadata_payload.get("series_sequence")
or metadata_payload.get("book_number")
)
if series_index_value is not None:
series_index_text = str(series_index_value).strip()
if series_index_text:
metadata_overrides.setdefault("series_index", series_index_text)
metadata_overrides.setdefault("series_position", series_index_text)
metadata_overrides.setdefault("series_sequence", series_index_text)
metadata_overrides.setdefault("book_number", series_index_text)
tags_value = metadata_payload.get("tags") or metadata_payload.get("keywords")
if tags_value:
tags_text = _stringify_metadata_value(tags_value)
if tags_text:
metadata_overrides.setdefault("tags", tags_text)
metadata_overrides.setdefault("keywords", tags_text)
metadata_overrides.setdefault("genre", tags_text)
description_value = metadata_payload.get("description") or metadata_payload.get("summary")
if description_value:
description_text = _stringify_metadata_value(description_value)
if description_text:
metadata_overrides.setdefault("description", description_text)
metadata_overrides.setdefault("summary", description_text)
subtitle_value = metadata_payload.get("subtitle")
if subtitle_value:
subtitle_text = _stringify_metadata_value(subtitle_value)
if subtitle_text:
metadata_overrides.setdefault("subtitle", subtitle_text)
publisher_value = metadata_payload.get("publisher")
if publisher_value:
publisher_text = _stringify_metadata_value(publisher_value)
if publisher_text:
metadata_overrides.setdefault("publisher", publisher_text)
settings = load_settings() settings = load_settings()
integrations = load_integration_settings() integrations = load_integration_settings()
+21 -2
View File
@@ -1,5 +1,8 @@
from pathlib import Path from pathlib import Path
from collections.abc import Mapping
from typing import Any
from flask import Blueprint, current_app, render_template, request, redirect, url_for, flash, send_file, abort from flask import Blueprint, current_app, render_template, request, redirect, url_for, flash, send_file, abort
from flask.typing import ResponseReturnValue from flask.typing import ResponseReturnValue
@@ -7,6 +10,7 @@ from abogen.web.routes.utils.settings import (
load_settings, load_settings,
load_integration_settings, load_integration_settings,
save_settings, save_settings,
stored_integration_config,
coerce_bool, coerce_bool,
coerce_int, coerce_int,
SAVE_MODE_LABELS, SAVE_MODE_LABELS,
@@ -18,7 +22,7 @@ from abogen.web.routes.utils.settings import (
from abogen.web.routes.utils.voice import template_options from abogen.web.routes.utils.voice import template_options
from abogen.web.debug_tts_runner import run_debug_tts_wavs from abogen.web.debug_tts_runner import run_debug_tts_wavs
from abogen.debug_tts_samples import DEBUG_TTS_SAMPLES from abogen.debug_tts_samples import DEBUG_TTS_SAMPLES
from abogen.utils import get_user_output_path from abogen.utils import get_user_output_path, load_config
settings_bp = Blueprint("settings", __name__) settings_bp = Blueprint("settings", __name__)
@@ -105,7 +109,22 @@ def update_settings() -> ResponseReturnValue:
current[key] = (form.get(key) or "").strip() current[key] = (form.get(key) or "").strip()
# Integrations # Integrations
current["integrations"] = current.get("integrations", {}) # `load_settings()` returns only the general settings subset and intentionally
# does not include stored integrations. Seed them from the stored config so
# saving unrelated settings cannot wipe credentials/tokens.
current_integrations: dict[str, dict[str, Any]] = {}
cfg = load_config() or {}
stored_integrations = cfg.get("integrations")
if isinstance(stored_integrations, Mapping):
for name, payload in stored_integrations.items():
if isinstance(name, str) and isinstance(payload, Mapping):
current_integrations[name] = dict(payload)
# Ensure known integrations are loaded even if the config is still in legacy format.
for name in ("audiobookshelf", "calibre_opds"):
stored = stored_integration_config(name)
if stored and name not in current_integrations:
current_integrations[name] = dict(stored)
current["integrations"] = current_integrations
# Audiobookshelf # Audiobookshelf
abs_enabled = coerce_bool(form.get("audiobookshelf_enabled"), False) abs_enabled = coerce_bool(form.get("audiobookshelf_enabled"), False)
+12
View File
@@ -759,6 +759,18 @@ if (modal && browser) {
metadata.description = entry.summary; metadata.description = entry.summary;
metadata.summary = entry.summary; metadata.summary = entry.summary;
} }
if (Array.isArray(entry.authors) && entry.authors.length > 0) {
const authorsText = entry.authors.map((name) => String(name || '').trim()).filter(Boolean).join(', ');
if (authorsText) {
metadata.authors = authorsText;
metadata.author = authorsText;
}
}
if (typeof entry.subtitle === 'string' && entry.subtitle.trim()) {
metadata.subtitle = entry.subtitle.trim();
}
if (entry.rating !== null && entry.rating !== undefined && entry.rating !== '') { if (entry.rating !== null && entry.rating !== undefined && entry.rating !== '') {
metadata.rating = String(entry.rating); metadata.rating = String(entry.rating);
} }
+26
View File
@@ -33,6 +33,32 @@ def test_calibre_opds_feed_exposes_series_metadata() -> None:
assert feed_dict["entries"][0]["series_index"] == 4.0 assert feed_dict["entries"][0]["series_index"] == 4.0
def test_calibre_opds_feed_exposes_subtitle_metadata() -> None:
client = CalibreOPDSClient("http://example.com/catalog")
xml_payload = """<?xml version=\"1.0\" encoding=\"UTF-8\"?>
<feed xmlns=\"http://www.w3.org/2005/Atom\"
xmlns:calibre_md=\"http://calibre.kovidgoyal.net/2009/metadata\">
<id>catalog</id>
<title>Example Catalog</title>
<entry>
<id>book-1</id>
<title>Sample Book</title>
<calibre_md:subtitle>A Novel</calibre_md:subtitle>
<link rel=\"http://opds-spec.org/acquisition\"
href=\"books/sample.epub\"
type=\"application/epub+zip\" />
</entry>
</feed>
"""
feed = client._parse_feed(xml_payload, base_url="http://example.com/catalog")
assert feed.entries
assert feed.entries[0].subtitle == "A Novel"
feed_dict = feed_to_dict(feed)
assert feed_dict["entries"][0]["subtitle"] == "A Novel"
def test_calibre_opds_feed_extracts_series_from_categories() -> None: def test_calibre_opds_feed_extracts_series_from_categories() -> None:
client = CalibreOPDSClient("http://example.com/catalog") client = CalibreOPDSClient("http://example.com/catalog")
xml_payload = """<?xml version=\"1.0\" encoding=\"UTF-8\"?> xml_payload = """<?xml version=\"1.0\" encoding=\"UTF-8\"?>
@@ -0,0 +1,32 @@
from __future__ import annotations
from abogen.web.routes.api import _opds_metadata_overrides
def test_opds_metadata_overrides_maps_author_and_subtitle() -> None:
overrides = _opds_metadata_overrides(
{
"authors": ["Alexandre Dumas"],
"subtitle": "Unabridged",
"series": "Example",
"series_index": 2,
"tags": ["Fiction", "Classic"],
"summary": "Summary text",
}
)
assert overrides["authors"] == "Alexandre Dumas"
assert overrides["author"] == "Alexandre Dumas"
assert overrides["subtitle"] == "Unabridged"
# Existing behavior still present
assert overrides["series"] == "Example"
assert overrides["series_index"] == "2"
assert overrides["tags"] == "Fiction, Classic"
assert overrides["description"] == "Summary text"
def test_opds_metadata_overrides_accepts_author_string() -> None:
overrides = _opds_metadata_overrides({"author": "Mary Shelley"})
assert overrides["authors"] == "Mary Shelley"
assert overrides["author"] == "Mary Shelley"
+111
View File
@@ -0,0 +1,111 @@
from __future__ import annotations
from pathlib import Path
from abogen.utils import load_config, save_config
from abogen.web.app import create_app
def test_settings_update_preserves_abs_api_token_when_blank(tmp_path):
# Seed config with stored integration secret.
save_config(
{
"language": "en",
"integrations": {
"audiobookshelf": {
"enabled": True,
"base_url": "https://abs.example",
"api_token": "SECRET_TOKEN",
"library_id": "lib1",
"folder_id": "fold1",
"verify_ssl": True,
},
"calibre_opds": {
"enabled": True,
"base_url": "https://opds.example",
"username": "user",
"password": "SECRET_PASS",
"verify_ssl": True,
},
},
}
)
app = create_app(
{
"TESTING": True,
"SECRET_KEY": "test",
"OUTPUT_FOLDER": str(tmp_path),
"UPLOAD_FOLDER": str(tmp_path / "uploads"),
}
)
with app.test_client() as client:
# Emulate saving settings where integrations are present but secrets are blank
# (typical of masked password/token inputs).
resp = client.post(
"/settings/update",
data={
"language": "en",
"output_format": "mp3",
# ABS integration fields (token blank)
"audiobookshelf_enabled": "on",
"audiobookshelf_base_url": "https://abs.example",
"audiobookshelf_api_token": "",
"audiobookshelf_library_id": "lib1",
"audiobookshelf_folder_id": "fold1",
"audiobookshelf_verify_ssl": "on",
# Calibre OPDS integration fields (password blank)
"calibre_opds_enabled": "on",
"calibre_opds_base_url": "https://opds.example",
"calibre_opds_username": "user",
"calibre_opds_password": "",
"calibre_opds_verify_ssl": "on",
},
follow_redirects=False,
)
assert resp.status_code in {302, 303}
cfg = load_config() or {}
integrations = cfg.get("integrations") or {}
assert integrations["audiobookshelf"]["api_token"] == "SECRET_TOKEN"
assert integrations["calibre_opds"]["password"] == "SECRET_PASS"
def test_settings_update_preserves_secrets_when_fields_missing(tmp_path):
save_config(
{
"language": "en",
"integrations": {
"audiobookshelf": {"api_token": "SECRET_TOKEN"},
"calibre_opds": {"password": "SECRET_PASS"},
},
}
)
app = create_app(
{
"TESTING": True,
"SECRET_KEY": "test",
"OUTPUT_FOLDER": str(tmp_path),
"UPLOAD_FOLDER": str(tmp_path / "uploads"),
}
)
with app.test_client() as client:
# Post unrelated changes; omit integration fields completely.
resp = client.post(
"/settings/update",
data={
"language": "en",
"output_format": "wav",
},
follow_redirects=False,
)
assert resp.status_code in {302, 303}
cfg = load_config() or {}
integrations = cfg.get("integrations") or {}
assert integrations["audiobookshelf"]["api_token"] == "SECRET_TOKEN"
assert integrations["calibre_opds"]["password"] == "SECRET_PASS"