diff --git a/abogen/application/conversion_executor.py b/abogen/application/conversion_executor.py index 1041dd0..462ff5f 100644 --- a/abogen/application/conversion_executor.py +++ b/abogen/application/conversion_executor.py @@ -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( diff --git a/abogen/application/conversion_service.py b/abogen/application/conversion_service.py index 7b654c8..ef46576 100644 --- a/abogen/application/conversion_service.py +++ b/abogen/application/conversion_service.py @@ -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) + diff --git a/abogen/application/integration_hooks.py b/abogen/application/integration_hooks.py new file mode 100644 index 0000000..eea76f6 --- /dev/null +++ b/abogen/application/integration_hooks.py @@ -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") diff --git a/abogen/application/output_layout_service.py b/abogen/application/output_layout_service.py index 89408cd..9b70bd1 100644 --- a/abogen/application/output_layout_service.py +++ b/abogen/application/output_layout_service.py @@ -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( diff --git a/abogen/domain/settings_core.py b/abogen/domain/settings_core.py index 8e638f3..7cc0ee7 100644 --- a/abogen/domain/settings_core.py +++ b/abogen/domain/settings_core.py @@ -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, + ) diff --git a/abogen/infrastructure/exporters.py b/abogen/infrastructure/exporters.py index 7e43d7b..1f1126a 100644 --- a/abogen/infrastructure/exporters.py +++ b/abogen/infrastructure/exporters.py @@ -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) diff --git a/abogen/webui/routes/jobs.py b/abogen/webui/routes/jobs.py index 1e28ed7..febade6 100644 --- a/abogen/webui/routes/jobs.py +++ b/abogen/webui/routes/jobs.py @@ -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 diff --git a/abogen/webui/routes/utils/settings.py b/abogen/webui/routes/utils/settings.py index d022a5a..6682251 100644 --- a/abogen/webui/routes/utils/settings.py +++ b/abogen/webui/routes/utils/settings.py @@ -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: diff --git a/abogen/webui/service.py b/abogen/webui/service.py index d1eeb96..e160dbc 100644 --- a/abogen/webui/service.py +++ b/abogen/webui/service.py @@ -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 diff --git a/abogen/webui/services/settings_service.py b/abogen/webui/services/settings_service.py index e73b64b..12cd8bf 100644 --- a/abogen/webui/services/settings_service.py +++ b/abogen/webui/services/settings_service.py @@ -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 diff --git a/tests/test_domain_conversion_pipeline.py b/tests/test_domain_conversion_pipeline.py index 5970e14..ca92137 100644 --- a/tests/test_domain_conversion_pipeline.py +++ b/tests/test_domain_conversion_pipeline.py @@ -251,13 +251,15 @@ 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 - segments, split = spacy_pre_tts_segmentation( - "", - Language.FR, - "Sentence", - use_spacy_segmentation=True, - ) + with patch("abogen.spacy_utils.segment_sentences", return_value=None): + segments, split = spacy_pre_tts_segmentation( + "", + Language.FR, + "Sentence", + use_spacy_segmentation=True, + ) assert len(segments) >= 1 diff --git a/tests/test_integration_hooks.py b/tests/test_integration_hooks.py new file mode 100644 index 0000000..0f8b07b --- /dev/null +++ b/tests/test_integration_hooks.py @@ -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 diff --git a/tests/test_service.py b/tests/test_service.py index afa56c5..cffd5c8 100644 --- a/tests/test_service.py +++ b/tests/test_service.py @@ -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"