Dymas/abogen
1
0
Fork 0
mirror of https://github.com/denizsafak/abogen.git synced 2026-09-20 11:40:57 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactor: centralize integration config in domain layer, fix OutputFormat enum serialization

Browse Source 
- Move stored_integration_config(), build_audiobookshelf_config() to domain/settings_core.py
- Remove legacy fallback from stored_integration_config() (only config[integrations])
- Add load_audiobookshelf_config() as combined entry point
- PostConversionHooks reads config directly via stored_integration_config()
- WebUI imports from domain/settings_core instead of webui/routes/utils/settings
- Remove duplicate _build_abs_config() from PostConversionHooks
- Remove dead audiobookshelf code from infrastructure/exporters.py
- Remove duplicate audiobookshelf functions from webui/service.py
- Fix OutputFormat enum serialization in output_layout_service.py and conversion_executor.py
  (f'{enum}' gave 'OutputFormat.WAV' instead of '.wav')
- Mock spaCy in test_returns_at_least_one_segment instead of loading real model
- Add 28 tests for PostConversionHooks and build_audiobookshelf_config
This commit is contained in:
Artem Akymenko
2026-07-29 13:11:56 +00:00
parent c293cc90f6
commit f802fb2af6
13 changed files with 781 additions and 321 deletions
Download Patch File Download Diff File Expand all files Collapse all files
abogen/application/conversion_executor.py
+1 -1
View File
@@ -229,7 +229,7 @@ def execute_conversion(
audio_sink: Optional[AudioSink] = None
audio_path = None
if merge_chapters:
audio_path = output_layout.audio_dir / f"{_base_name(request)}.{request.output_format}"
audio_path = output_layout.audio_dir / f"{_base_name(request)}{request.output_format.dot_ext}"
meta = plan.metadata if plan.metadata else None
audio_sink = stack.enter_context(
open_audio_sink(
abogen/application/conversion_service.py
+6
View File
@@ -240,5 +240,11 @@ def _finalize(
except Exception as exc:
events.log(f"Failed to record override usage: {exc}", level="debug")
# Post-conversion hooks (Audiobookshelf, etc.)
from abogen.application.integration_hooks import PostConversionHooks
hooks = PostConversionHooks()
hooks.run(request, result, events)
abogen/application/integration_hooks.py
+164
View File
@@ -0,0 +1,164 @@
"""Post-conversion integration hooks.
Called by ConversionService after finalization.
Each integration is a method on PostConversionHooks — isolated, testable,
and easy to extend with new hooks (Plex, Navidrome, etc.).
The service NEVER imports from PyQt or WebUI.
"""
from __future__ import annotations
import logging
from pathlib import Path
from typing import Any, Mapping, Optional
from abogen.application.conversion_ports import ConversionEvents
from abogen.application.conversion_request import ConversionRequest
from abogen.application.conversion_result import ConversionResult
from abogen.domain.metadata_helpers import (
build_audiobookshelf_metadata as _build_abs_metadata,
load_audiobookshelf_chapters as _load_abs_chapters,
)
from abogen.domain.settings_core import (
build_audiobookshelf_config,
coerce_bool,
load_audiobookshelf_config,
stored_integration_config,
)
from abogen.integrations.audiobookshelf import (
AudiobookshelfClient,
AudiobookshelfUploadError,
)
logger = logging.getLogger(__name__)
class PostConversionHooks:
"""Runs post-conversion integrations (Audiobookshelf, etc.).
Usage::
hooks = PostConversionHooks()
hooks.run(request, result, events)
"""
def run(
self,
request: ConversionRequest,
result: ConversionResult,
events: ConversionEvents,
) -> None:
"""Run all registered post-conversion hooks."""
self._maybe_send_to_audiobookshelf(request, result, events)
# ------------------------------------------------------------------
# Audiobookshelf
# ------------------------------------------------------------------
def _maybe_send_to_audiobookshelf(
self,
request: ConversionRequest,
result: ConversionResult,
events: ConversionEvents,
) -> None:
"""Upload finished audiobook to Audiobookshelf if enabled."""
abs_settings = stored_integration_config("audiobookshelf")
if not abs_settings:
return
enabled = coerce_bool(abs_settings.get("enabled"), False)
auto_send = coerce_bool(abs_settings.get("auto_send"), False)
if not (enabled and auto_send):
return
config = build_audiobookshelf_config(abs_settings)
if config is None:
events.log(
"Audiobookshelf upload skipped: configure base URL, API token, "
"library ID, and folder ID first.",
level="warning",
)
return
audio_path = result.audio_path
if not audio_path or not audio_path.exists():
events.log(
"Audiobookshelf upload skipped: audio output not found.",
level="warning",
)
return
# Build metadata
filename = request.original_filename or "Audiobook"
lang = request.language.value if hasattr(request.language, "value") else str(request.language)
metadata = _build_abs_metadata(
result.metadata or {},
language=lang,
filename=Path(filename).stem,
)
# Load chapters from metadata artifact
chapters = None
if config.send_chapters:
metadata_artifact = result.artifacts.get("metadata")
if metadata_artifact:
metadata_path = (
metadata_artifact
if isinstance(metadata_artifact, Path)
else Path(str(metadata_artifact))
)
chapters = _load_abs_chapters(metadata_path)
# Resolve cover
cover_path = None
if config.send_cover and request.cover and request.cover.path:
candidate = request.cover.path
if isinstance(candidate, Path) and candidate.exists():
cover_path = candidate
# Resolve subtitles
subtitles = None
if config.send_subtitles and result.subtitle_paths:
subtitles = [
p for p in result.subtitle_paths
if isinstance(p, Path) and p.exists()
]
# Upload
client = AudiobookshelfClient(config)
display_title = metadata.get("title") or audio_path.stem
try:
existing_items = client.find_existing_items(
display_title, folder_id=config.folder_id,
)
except AudiobookshelfUploadError as exc:
events.log(f"Audiobookshelf lookup failed: {exc}", level="error")
return
if existing_items:
events.log(
f"Removing existing Audiobookshelf item(s) for '{display_title}'.",
level="info",
)
try:
client.delete_items(existing_items)
except Exception as exc:
events.log(
f"Failed to remove existing item(s): {exc}", level="warning",
)
try:
client.upload_audiobook(
audio_path,
metadata=metadata,
cover_path=cover_path,
chapters=chapters,
subtitles=subtitles,
)
events.log("Audiobookshelf upload queued.", level="info")
except AudiobookshelfUploadError as exc:
events.log(f"Audiobookshelf upload failed: {exc}", level="error")
except Exception as exc:
events.log(f"Audiobookshelf integration error: {exc}", level="error")
abogen/application/output_layout_service.py
+1 -1
View File
@@ -98,7 +98,7 @@ def resolve_merged_path(
base_name = sanitize_output_stem(
request.original_filename or "output"
)
return layout.audio_dir / f"{base_name}.{request.output_format}"
return layout.audio_dir / f"{base_name}{request.output_format.dot_ext}"
def resolve_chapter_path(
abogen/domain/settings_core.py
+62 -1
View File
@@ -10,7 +10,7 @@ from __future__ import annotations
import os
import re
from dataclasses import dataclass
from typing import Any, Callable, Dict, Mapping
from typing import Any, Callable, Dict, Mapping, Optional
from abogen.constants import (
KOKORO_CODE_LABELS,
@@ -578,3 +578,64 @@ def integration_defaults() -> Dict[str, Dict[str, Any]]:
"timeout": 30.0,
},
}
def stored_integration_config(name: str) -> Dict[str, Any]:
"""Read raw integration config from config.json.
Reads ``config["integrations"][name]``.
"""
from abogen.utils import load_config
cfg = load_config() or {}
integrations = cfg.get("integrations")
if isinstance(integrations, Mapping):
entry = integrations.get(name)
if isinstance(entry, Mapping):
return dict(entry)
return {}
def load_audiobookshelf_config() -> Optional["AudiobookshelfConfig"]:
"""Read Audiobookshelf settings from config.json and build typed config.
Returns ``None`` when the integration is not configured or required
fields are missing.
"""
raw = stored_integration_config("audiobookshelf")
if not raw:
return None
return build_audiobookshelf_config(raw)
def build_audiobookshelf_config(
settings: Mapping[str, Any],
) -> Optional["AudiobookshelfConfig"]:
"""Build :class:`AudiobookshelfConfig` from a settings dict.
Returns ``None`` when required fields (base_url, api_token, library_id)
are missing.
"""
from abogen.integrations.audiobookshelf import AudiobookshelfConfig
base_url = str(settings.get("base_url") or "").strip()
api_token = str(settings.get("api_token") or "").strip()
library_id = str(settings.get("library_id") or "").strip()
if not (base_url and api_token and library_id):
return None
try:
timeout = float(settings.get("timeout", 3600.0))
except (TypeError, ValueError):
timeout = 3600.0
return AudiobookshelfConfig(
base_url=base_url,
api_token=api_token,
library_id=library_id,
collection_id=(str(settings.get("collection_id") or "").strip() or None),
folder_id=(str(settings.get("folder_id") or "").strip() or None),
verify_ssl=coerce_bool(settings.get("verify_ssl"), True),
send_cover=coerce_bool(settings.get("send_cover"), True),
send_chapters=coerce_bool(settings.get("send_chapters"), True),
send_subtitles=coerce_bool(settings.get("send_subtitles"), False),
timeout=timeout,
)
abogen/infrastructure/exporters.py
-132
View File
@@ -16,16 +16,9 @@ from abogen.domain.metadata_helpers import (
first_nonempty,
extract_year,
normalize_series_sequence,
build_audiobookshelf_metadata as _build_abs_metadata,
load_audiobookshelf_chapters as _load_abs_chapters,
_SERIES_SEQUENCE_TAG_KEYS,
)
from abogen.epub3.exporter import build_epub3_package
from abogen.integrations.audiobookshelf import (
AudiobookshelfClient,
AudiobookshelfConfig,
AudiobookshelfUploadError,
)
from abogen.utils import create_process
logger = logging.getLogger(__name__)
@@ -320,132 +313,7 @@ class ExportService:
cover_image_mime=cover_mime,
)
# ----------------------------------------------------------------------
# Audiobookshelf Integration
# ----------------------------------------------------------------------
def build_audiobookshelf_metadata(self, job: Any) -> Dict[str, Any]:
"""Build Audiobookshelf metadata from job."""
filename = Path(getattr(job, "original_filename", "") or "").stem or "Audiobook"
return _build_abs_metadata(
getattr(job, "metadata_tags", {}),
language=getattr(job, "language", "") or "",
filename=filename,
)
def load_audiobookshelf_chapters(self, job: Any) -> Optional[List[Dict[str, Any]]]:
"""Load chapters from job artifacts for Audiobookshelf."""
metadata_ref = job.result.artifacts.get("metadata") if getattr(job, "result", None) else None
if not metadata_ref:
return None
metadata_path = metadata_ref if isinstance(metadata_ref, Path) else Path(str(metadata_ref))
return _load_abs_chapters(metadata_path)
def upload_audiobookshelf(
self,
job: Any,
audio_path: Path,
subtitle_paths: List[Path],
chapters: List[Dict[str, Any]],
metadata: Dict[str, Any],
cover_path: Optional[Path] = None,
config: Optional[AudiobookshelfConfig] = None,
log_callback: Optional[callable] = None,
) -> None:
"""Upload to Audiobookshelf."""
if config is None:
cfg = getattr(job, "_abs_config", None)
if cfg is None:
from abogen.utils import load_config
global_cfg = load_config() or {}
abs_cfg = global_cfg.get("audiobookshelf")
if isinstance(abs_cfg, Mapping):
config = AudiobookshelfConfig(
base_url=str(abs_cfg.get("base_url") or "").strip(),
api_token=str(abs_cfg.get("api_token") or "").strip(),
library_id=str(abs_cfg.get("library_id") or "").strip(),
collection_id=(str(abs_cfg.get("collection_id") or "").strip() or None),
folder_id=str(abs_cfg.get("folder_id") or "").strip(),
verify_ssl=self._coerce_bool(abs_cfg.get("verify_ssl"), True),
send_cover=self._coerce_bool(abs_cfg.get("send_cover"), True),
send_chapters=self._coerce_bool(abs_cfg.get("send_chapters"), True),
send_subtitles=self._coerce_bool(abs_cfg.get("send_subtitles"), False),
timeout=float(abs_cfg.get("timeout", 3600.0)),
)
else:
if log_callback:
log_callback("Audiobookshelf upload skipped: not configured", "warning")
return
if not config.base_url or not config.api_token or not config.library_id:
if log_callback:
log_callback("Audiobookshelf upload skipped: configure base URL, API token, and library ID first", "warning")
return
if not config.folder_id:
if log_callback:
log_callback("Audiobookshelf upload skipped: enter folder name or ID in settings", "warning")
return
if not audio_path.exists():
if log_callback:
log_callback("Audiobookshelf upload skipped: audio output not found", "warning")
return
existing_subtitles = [p for p in subtitle_paths if p.exists()] if config.send_subtitles else None
chapters_to_send = chapters if config.send_chapters else None
client = AudiobookshelfClient(config)
display_title = metadata.get("title") or audio_path.stem
try:
existing_items = client.find_existing_items(display_title, folder_id=config.folder_id)
except AudiobookshelfUploadError as exc:
if log_callback:
log_callback(f"Audiobookshelf lookup failed: {exc}", "error")
return
if existing_items:
if log_callback:
log_callback(f"Removing existing Audiobookshelf item(s) for '{display_title}' before upload.", "info")
try:
client.delete_items(existing_items)
except Exception as exc:
if log_callback:
log_callback(f"Failed to remove existing item(s): {exc}", "warning")
cover_to_send = cover_path
if config.send_cover and cover_to_send:
if isinstance(cover_to_send, str):
cover_to_send = Path(cover_to_send)
if not cover_to_send.exists():
cover_to_send = None
client.upload_audiobook(
audio_path,
metadata=metadata,
cover_path=cover_to_send,
chapters=chapters_to_send,
subtitles=existing_subtitles,
)
if log_callback:
log_callback("Audiobookshelf upload queued.", "info")
# ----------------------------------------------------------------------
# Helpers
# ----------------------------------------------------------------------
@staticmethod
def _coerce_bool(value: Any, default: bool = True) -> bool:
if isinstance(value, bool):
return value
if isinstance(value, str):
lowered = value.strip().lower()
if lowered in {"true", "1", "yes", "on"}:
return True
if lowered in {"false", "0", "no", "off"}:
return False
return default
if value is None:
return default
return bool(value)
abogen/webui/routes/jobs.py
+4 -2
View File
@@ -8,6 +8,8 @@ from flask.typing import ResponseReturnValue
from abogen.webui.service import (
JobStatus,
)
from abogen.domain.metadata_helpers import (
build_audiobookshelf_metadata,
load_audiobookshelf_chapters,
)
@@ -19,9 +21,9 @@ from abogen.webui.routes.utils.epub import (
locate_job_epub,
locate_job_audio,
)
from abogen.webui.routes.utils.settings import (
stored_integration_config,
from abogen.domain.settings_core import (
build_audiobookshelf_config,
stored_integration_config,
)
from abogen.webui.routes.utils.common import existing_paths
from abogen.infrastructure.exporters import ExportService
abogen/webui/routes/utils/settings.py
+4 -39
View File
@@ -2,7 +2,6 @@ import os
from typing import Any, Dict, Mapping, Optional
from abogen.integrations.calibre_opds import CalibreOPDSClient
from abogen.integrations.audiobookshelf import AudiobookshelfConfig
from abogen.utils import load_config, save_config
from abogen.domain.settings_core import (
CHUNK_LEVEL_OPTIONS,
@@ -11,6 +10,7 @@ from abogen.domain.settings_core import (
SAVE_MODE_LABELS,
_NORMALIZATION_BOOLEAN_KEYS,
_NORMALIZATION_STRING_KEYS,
build_audiobookshelf_config,
coerce_bool,
coerce_float,
coerce_int,
@@ -18,6 +18,7 @@ from abogen.domain.settings_core import (
load_settings,
llm_ready,
settings_defaults,
stored_integration_config,
)
_NORMALIZATION_GROUPS = [
@@ -124,20 +125,8 @@ def load_integration_settings() -> Dict[str, Dict[str, Any]]:
return integrations
def stored_integration_config(name: str) -> Dict[str, Any]:
cfg = load_config() or {}
# Check under "integrations" first (new structure)
integrations = cfg.get("integrations")
if isinstance(integrations, Mapping):
entry = integrations.get(name)
if isinstance(entry, Mapping):
return dict(entry)
# Fallback to top-level (legacy structure)
entry = cfg.get(name)
if isinstance(entry, Mapping):
return dict(entry)
return {}
# stored_integration_config and build_audiobookshelf_config are imported from
# abogen.domain.settings_core — single source of truth for integration config.
def calibre_settings_from_payload(payload: Mapping[str, Any]) -> Dict[str, Any]:
@@ -305,30 +294,6 @@ def audiobookshelf_settings_from_payload(payload: Mapping[str, Any]) -> Dict[str
}
def build_audiobookshelf_config(settings: Mapping[str, Any]) -> Optional[AudiobookshelfConfig]:
base_url = str(settings.get("base_url") or "").strip()
api_token = str(settings.get("api_token") or "").strip()
library_id = str(settings.get("library_id") or "").strip()
if not (base_url and api_token and library_id):
return None
try:
timeout = float(settings.get("timeout", 3600.0))
except (TypeError, ValueError):
timeout = 3600.0
return AudiobookshelfConfig(
base_url=base_url,
api_token=api_token,
library_id=library_id,
collection_id=(str(settings.get("collection_id") or "").strip() or None),
folder_id=(str(settings.get("folder_id") or "").strip() or None),
verify_ssl=coerce_bool(settings.get("verify_ssl"), True),
send_cover=coerce_bool(settings.get("send_cover"), True),
send_chapters=coerce_bool(settings.get("send_chapters"), True),
send_subtitles=coerce_bool(settings.get("send_subtitles"), False),
timeout=timeout,
)
def calibre_integration_enabled(
integrations: Optional[Mapping[str, Any]] = None,
) -> bool:
abogen/webui/service.py
+1 -134
View File
@@ -15,24 +15,8 @@ from pathlib import Path
from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping
from abogen.domain.enums import Language
from abogen.utils import get_internal_cache_path, get_user_settings_dir, load_config
from abogen.utils import get_internal_cache_path, get_user_settings_dir
from abogen.voice_cache import bootstrap_voice_cache
from abogen.integrations.audiobookshelf import (
AudiobookshelfClient,
AudiobookshelfConfig,
AudiobookshelfUploadError,
)
from abogen.domain.metadata_helpers import (
normalize_metadata_casefold as _normalize_metadata_casefold,
split_people_field as _split_people_field,
split_simple_list as _split_simple_list,
first_nonempty as _first_nonempty,
extract_year as _extract_year,
normalize_series_sequence as _normalize_series_sequence,
build_audiobookshelf_metadata as _build_abs_metadata,
load_audiobookshelf_chapters as _load_abs_chapters,
_SERIES_SEQUENCE_TAG_KEYS,
)
def _create_set_event() -> threading.Event:
@@ -266,23 +250,6 @@ class Job:
}
def build_audiobookshelf_metadata(job: Job) -> Dict[str, Any]:
filename = Path(job.original_filename or "").stem or job.original_filename or "Audiobook"
return _build_abs_metadata(
job.metadata_tags,
language=job.language or "",
filename=filename,
)
def load_audiobookshelf_chapters(job: Job) -> Optional[List[Dict[str, Any]]]:
metadata_ref = job.result.artifacts.get("metadata")
if not metadata_ref:
return None
metadata_path = metadata_ref if isinstance(metadata_ref, Path) else Path(str(metadata_ref))
return _load_abs_chapters(metadata_path)
def _existing_paths(paths: Iterable[Any]) -> List[Path]:
resolved: List[Path] = []
for item in paths:
@@ -781,7 +748,6 @@ class ConversionService:
elif job.status != JobStatus.FAILED:
job.status = JobStatus.COMPLETED
job.add_log("Job completed", level="success")
self._post_completion_hooks(job)
job.finished_at = time.time()
finally:
job.pause_event.set()
@@ -802,105 +768,6 @@ class ConversionService:
self._queue.remove(job_id)
self._update_queue_positions_locked()
def _post_completion_hooks(self, job: Job) -> None:
try:
self._maybe_send_to_audiobookshelf(job)
except AudiobookshelfUploadError as exc:
job.add_log(f"Audiobookshelf upload failed: {exc}", level="error")
except Exception as exc: # pragma: no cover - defensive guard
job.add_log(f"Audiobookshelf integration error: {exc}", level="error")
def _maybe_send_to_audiobookshelf(self, job: Job) -> None:
cfg = load_config() or {}
integration_cfg = cfg.get("audiobookshelf")
if not isinstance(integration_cfg, Mapping):
return
enabled = self._coerce_bool(integration_cfg.get("enabled"), False)
auto_send = self._coerce_bool(integration_cfg.get("auto_send"), False)
if not (enabled and auto_send):
return
base_url = str(integration_cfg.get("base_url") or "").strip()
api_token = str(integration_cfg.get("api_token") or "").strip()
library_id = str(integration_cfg.get("library_id") or "").strip()
folder_id = str(integration_cfg.get("folder_id") or "").strip()
if not base_url or not api_token or not library_id:
job.add_log(
"Audiobookshelf upload skipped: configure base URL, API token, and library ID first.",
level="warning",
)
return
if not folder_id:
job.add_log(
"Audiobookshelf upload skipped: enter the folder name or ID in the Audiobookshelf settings.",
level="warning",
)
return
audio_ref = job.result.audio_path
audio_path = audio_ref if isinstance(audio_ref, Path) else Path(str(audio_ref)) if audio_ref else None
if not audio_path or not audio_path.exists():
job.add_log("Audiobookshelf upload skipped: audio output not found.", level="warning")
return
timeout_raw = integration_cfg.get("timeout", 3600.0)
try:
timeout_value = float(timeout_raw)
except (TypeError, ValueError):
timeout_value = 3600.0
config = AudiobookshelfConfig(
base_url=base_url,
api_token=api_token,
library_id=library_id,
collection_id=(str(integration_cfg.get("collection_id") or "").strip() or None),
folder_id=folder_id,
verify_ssl=self._coerce_bool(integration_cfg.get("verify_ssl"), True),
send_cover=self._coerce_bool(integration_cfg.get("send_cover"), True),
send_chapters=self._coerce_bool(integration_cfg.get("send_chapters"), True),
send_subtitles=self._coerce_bool(integration_cfg.get("send_subtitles"), False),
timeout=timeout_value,
)
cover_ref = job.cover_image_path
cover_path = None
if config.send_cover and cover_ref:
cover_candidate = cover_ref if isinstance(cover_ref, Path) else Path(str(cover_ref))
if cover_candidate.exists():
cover_path = cover_candidate
subtitles = _existing_paths(job.result.subtitle_paths) if config.send_subtitles else None
chapters = load_audiobookshelf_chapters(job) if config.send_chapters else None
metadata = build_audiobookshelf_metadata(job)
client = AudiobookshelfClient(config)
display_title = metadata.get("title") or audio_path.stem
try:
existing_items = client.find_existing_items(display_title, folder_id=config.folder_id)
except AudiobookshelfUploadError as exc:
job.add_log(f"Audiobookshelf lookup failed: {exc}", level="error")
return
if existing_items:
job.add_log(
f"Removing existing Audiobookshelf item(s) for '{display_title}' before upload.",
level="info",
)
try:
client.delete_items(existing_items)
except Exception as exc:
job.add_log(f"Failed to remove existing item(s): {exc}", level="warning")
client.upload_audiobook(
audio_path,
metadata=metadata,
cover_path=cover_path,
chapters=chapters,
subtitles=subtitles,
)
job.add_log("Audiobookshelf upload queued.", level="info")
# Persistence ------------------------------------------------------
def _serialize_job(self, job: Job) -> Dict[str, Any]:
result_audio = str(job.result.audio_path) if job.result.audio_path else None
abogen/webui/services/settings_service.py
+1 -1
View File
@@ -29,8 +29,8 @@ def apply_form_to_settings(current: dict, form: Mapping[str, Any]) -> dict:
DEFAULT_ANALYSIS_THRESHOLD,
_NORMALIZATION_BOOLEAN_KEYS,
_NORMALIZATION_STRING_KEYS,
stored_integration_config,
)
from abogen.webui.routes.utils.settings import stored_integration_config
from abogen.webui.routes.utils.common import extract_checkbox
from abogen.utils import load_config
# General settings
tests/test_domain_conversion_pipeline.py
+2
View File
@@ -251,9 +251,11 @@ class TestSpacyPreTtsSegmentation:
assert len(segments) == 1
def test_returns_at_least_one_segment(self):
from unittest.mock import patch
from abogen.domain.conversion_pipeline import spacy_pre_tts_segmentation
from abogen.domain.enums import Language
with patch("abogen.spacy_utils.segment_sentences", return_value=None):
segments, split = spacy_pre_tts_segmentation(
"",
Language.FR,
tests/test_integration_hooks.py
+525
View File
@@ -0,0 +1,525 @@
"""Tests for application/integration_hooks.py — PostConversionHooks
and domain/settings_core.py — build_audiobookshelf_config."""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, List
from unittest.mock import MagicMock, patch
import pytest
from abogen.application.conversion_request import ConversionRequest
from abogen.application.conversion_result import ConversionResult
from abogen.application.integration_hooks import PostConversionHooks
from abogen.domain.enums import Language
from abogen.domain.settings_core import build_audiobookshelf_config
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_request(**overrides: Any) -> ConversionRequest:
defaults = dict(
source_path=Path("/tmp/test.txt"),
original_filename="test.txt",
language=Language.EN_US,
voice="M1",
speed=1.0,
use_gpu=False,
)
defaults.update(overrides)
return ConversionRequest(**defaults)
def _make_result(**overrides: Any) -> ConversionResult:
defaults: Dict[str, Any] = dict(
metadata={"title": "Test Book"},
)
defaults.update(overrides)
return ConversionResult(**defaults)
class _FakeEvents:
def __init__(self) -> None:
self.logs: List[tuple[str, str]] = []
def log(self, message: str, level: str = "info") -> None:
self.logs.append((message, level))
def progress(self, pct: int, etr: str) -> None:
pass
def check_cancelled(self) -> None:
pass
def _abs_settings(**overrides: Any) -> Dict[str, Any]:
"""Build a minimal Audiobookshelf settings dict."""
settings: Dict[str, Any] = {
"enabled": True,
"auto_send": True,
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
}
settings.update(overrides)
return settings
# ---------------------------------------------------------------------------
# build_audiobookshelf_config tests (domain layer)
# ---------------------------------------------------------------------------
class TestBuildAbsConfig:
"""build_audiobookshelf_config from domain.settings_core."""
def test_returns_none_when_base_url_missing(self) -> None:
result = build_audiobookshelf_config({
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
})
assert result is None
def test_returns_none_when_api_token_missing(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"library_id": "lib",
"folder_id": "fld",
})
assert result is None
def test_returns_none_when_library_id_missing(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"folder_id": "fld",
})
assert result is None
def test_returns_none_when_folder_id_missing(self) -> None:
# folder_id is optional in AudiobookshelfConfig, so this should succeed
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
})
assert result is not None
def test_returns_config_when_all_required_fields_present(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
})
assert result is not None
assert result.base_url == "https://example.com"
assert result.api_token == "tok"
assert result.library_id == "lib"
assert result.folder_id == "fld"
def test_preserves_trailing_slash_in_base_url(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com/",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
})
assert result is not None
# normalization is done by AudiobookshelfClient, not config
assert result.base_url == "https://example.com/"
def test_preserves_api_suffix_in_base_url(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com/api",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
})
assert result is not None
# normalization is done by AudiobookshelfClient, not config
assert result.base_url == "https://example.com/api"
def test_applies_default_timeout(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
})
assert result is not None
assert result.timeout == 3600.0
def test_applies_custom_timeout(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
"timeout": 7200.0,
})
assert result is not None
assert result.timeout == 7200.0
def test_invalid_timeout_falls_back_to_default(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
"timeout": "invalid",
})
assert result is not None
assert result.timeout == 3600.0
def test_collection_id_is_optional(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
})
assert result is not None
assert result.collection_id is None
def test_collection_id_when_provided(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
"collection_id": "col123",
})
assert result is not None
assert result.collection_id == "col123"
def test_boolean_flags_default(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
})
assert result is not None
assert result.verify_ssl is True
assert result.send_cover is True
assert result.send_chapters is True
assert result.send_subtitles is False
def test_boolean_flags_custom(self) -> None:
result = build_audiobookshelf_config({
"base_url": "https://example.com",
"api_token": "tok",
"library_id": "lib",
"folder_id": "fld",
"verify_ssl": False,
"send_cover": False,
"send_chapters": False,
"send_subtitles": True,
})
assert result is not None
assert result.verify_ssl is False
assert result.send_cover is False
assert result.send_chapters is False
assert result.send_subtitles is True
# ---------------------------------------------------------------------------
# Hook skipping tests
# ---------------------------------------------------------------------------
class TestPostConversionHooks:
"""PostConversionHooks.run() skipping logic."""
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_skip_when_no_audiobookshelf_config(self, mock_stored: MagicMock) -> None:
mock_stored.return_value = {}
hooks = PostConversionHooks()
request = _make_request()
result = _make_result()
events = _FakeEvents()
hooks.run(request, result, events)
assert events.logs == []
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_skip_when_enabled_false(self, mock_stored: MagicMock) -> None:
mock_stored.return_value = _abs_settings(enabled=False)
hooks = PostConversionHooks()
request = _make_request()
result = _make_result()
events = _FakeEvents()
hooks.run(request, result, events)
assert events.logs == []
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_skip_when_auto_send_false(self, mock_stored: MagicMock) -> None:
mock_stored.return_value = _abs_settings(auto_send=False)
hooks = PostConversionHooks()
request = _make_request()
result = _make_result()
events = _FakeEvents()
hooks.run(request, result, events)
assert events.logs == []
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_skip_when_config_incomplete(self, mock_stored: MagicMock) -> None:
mock_stored.return_value = _abs_settings(base_url="")
hooks = PostConversionHooks()
request = _make_request()
result = _make_result()
events = _FakeEvents()
hooks.run(request, result, events)
assert len(events.logs) == 1
assert "configure" in events.logs[0][0].lower()
assert events.logs[0][1] == "warning"
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_skip_when_audio_path_missing(self, mock_stored: MagicMock) -> None:
mock_stored.return_value = _abs_settings()
hooks = PostConversionHooks()
request = _make_request()
result = _make_result(audio_path=None)
events = _FakeEvents()
hooks.run(request, result, events)
assert len(events.logs) == 1
assert "audio output not found" in events.logs[0][0].lower()
assert events.logs[0][1] == "warning"
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_skip_when_audio_file_does_not_exist(self, mock_stored: MagicMock, tmp_path: Path) -> None:
mock_stored.return_value = _abs_settings()
hooks = PostConversionHooks()
request = _make_request()
result = _make_result(audio_path=tmp_path / "nonexistent.mp3")
events = _FakeEvents()
hooks.run(request, result, events)
assert len(events.logs) == 1
assert "audio output not found" in events.logs[0][0].lower()
# ---------------------------------------------------------------------------
# Upload flow tests (mocked client)
# ---------------------------------------------------------------------------
class TestAudiobookshelfUpload:
"""Test the upload flow with mocked AudiobookshelfClient."""
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_successful_upload(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
mock_client = MagicMock()
mock_client.find_existing_items.return_value = []
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings()
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
result = _make_result(audio_path=audio_path)
events = _FakeEvents()
hooks.run(request, result, events)
mock_client.find_existing_items.assert_called_once()
mock_client.upload_audiobook.assert_called_once()
assert any("upload queued" in msg.lower() for msg, _ in events.logs)
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_deletes_existing_items_before_upload(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
mock_client = MagicMock()
mock_client.find_existing_items.return_value = [{"id": "existing-1"}]
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings()
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
result = _make_result(audio_path=audio_path)
events = _FakeEvents()
hooks.run(request, result, events)
mock_client.delete_items.assert_called_once_with([{"id": "existing-1"}])
mock_client.upload_audiobook.assert_called_once()
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_lookup_error_logged_not_raised(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
from abogen.integrations.audiobookshelf import AudiobookshelfUploadError
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
mock_client = MagicMock()
mock_client.find_existing_items.side_effect = AudiobookshelfUploadError("connection refused")
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings()
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
result = _make_result(audio_path=audio_path)
events = _FakeEvents()
hooks.run(request, result, events)
assert any("lookup failed" in msg.lower() for msg, _ in events.logs)
mock_client.upload_audiobook.assert_not_called()
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_upload_error_logged_not_raised(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
from abogen.integrations.audiobookshelf import AudiobookshelfUploadError
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
mock_client = MagicMock()
mock_client.find_existing_items.return_value = []
mock_client.upload_audiobook.side_effect = AudiobookshelfUploadError("timeout")
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings()
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
result = _make_result(audio_path=audio_path)
events = _FakeEvents()
hooks.run(request, result, events)
assert any("upload failed" in msg.lower() for msg, _ in events.logs)
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_delete_error_logged_not_raised(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
mock_client = MagicMock()
mock_client.find_existing_items.return_value = [{"id": "old-item"}]
mock_client.delete_items.side_effect = Exception("network error")
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings()
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
result = _make_result(audio_path=audio_path)
events = _FakeEvents()
hooks.run(request, result, events)
assert any("failed to remove" in msg.lower() for msg, _ in events.logs)
mock_client.upload_audiobook.assert_called_once()
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_cover_included_when_exists(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
cover_path = tmp_path / "cover.jpg"
cover_path.write_bytes(b"jpeg-content")
mock_client = MagicMock()
mock_client.find_existing_items.return_value = []
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings(send_cover=True)
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
from abogen.application.conversion_config import CoverConfig
request.cover = CoverConfig(path=cover_path, mime="image/jpeg")
result = _make_result(audio_path=audio_path)
events = _FakeEvents()
hooks.run(request, result, events)
call_kwargs = mock_client.upload_audiobook.call_args
assert call_kwargs[1]["cover_path"] == cover_path
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_subtitles_included_when_enabled(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
subtitle_path = tmp_path / "book.srt"
subtitle_path.write_bytes(b"srt-content")
mock_client = MagicMock()
mock_client.find_existing_items.return_value = []
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings(send_subtitles=True)
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
result = _make_result(audio_path=audio_path)
result.subtitle_paths = [subtitle_path]
events = _FakeEvents()
hooks.run(request, result, events)
call_kwargs = mock_client.upload_audiobook.call_args
assert call_kwargs[1]["subtitles"] == [subtitle_path]
@patch("abogen.application.integration_hooks.AudiobookshelfClient")
@patch("abogen.application.integration_hooks.stored_integration_config")
def test_subtitles_skipped_when_disabled(
self, mock_stored: MagicMock, mock_client_cls: MagicMock, tmp_path: Path,
) -> None:
audio_path = tmp_path / "book.mp3"
audio_path.write_bytes(b"audio-content")
subtitle_path = tmp_path / "book.srt"
subtitle_path.write_bytes(b"srt-content")
mock_client = MagicMock()
mock_client.find_existing_items.return_value = []
mock_client_cls.return_value = mock_client
mock_stored.return_value = _abs_settings(send_subtitles=False)
hooks = PostConversionHooks()
request = _make_request(original_filename="book.mp3")
result = _make_result(audio_path=audio_path)
result.subtitle_paths = [subtitle_path]
events = _FakeEvents()
hooks.run(request, result, events)
call_kwargs = mock_client.upload_audiobook.call_args
assert call_kwargs[1]["subtitles"] is None
tests/test_service.py
+4 -4
View File
@@ -7,8 +7,8 @@ from abogen.webui.service import (
JobStatus,
build_service,
_JOB_LOGGER,
build_audiobookshelf_metadata,
)
from abogen.domain.metadata_helpers import build_audiobookshelf_metadata
def test_service_processes_job(tmp_path):
@@ -233,7 +233,7 @@ def test_audiobookshelf_metadata_uses_book_number(tmp_path):
},
)
metadata = build_audiobookshelf_metadata(job)
metadata = build_audiobookshelf_metadata(job.metadata_tags)
assert metadata["seriesName"] == "Example Saga"
assert metadata["seriesSequence"] == "7"
@@ -264,7 +264,7 @@ def test_audiobookshelf_metadata_normalizes_sequence_value(tmp_path):
},
)
metadata = build_audiobookshelf_metadata(job)
metadata = build_audiobookshelf_metadata(job.metadata_tags)
assert metadata["seriesName"] == "Example Saga"
assert metadata["seriesSequence"] == "7"
@@ -295,6 +295,6 @@ def test_audiobookshelf_metadata_allows_decimal_sequence(tmp_path):
},
)
metadata = build_audiobookshelf_metadata(job)
metadata = build_audiobookshelf_metadata(job.metadata_tags)
assert metadata["seriesSequence"] == "4.5"