diff --git a/abogen/domain/metadata_helpers.py b/abogen/domain/metadata_helpers.py index ea4025b..87d1756 100644 --- a/abogen/domain/metadata_helpers.py +++ b/abogen/domain/metadata_helpers.py @@ -1,6 +1,9 @@ from __future__ import annotations +import json +import math import re +from pathlib import Path from typing import Any, Dict, List, Mapping, Optional, Tuple @@ -136,3 +139,267 @@ def format_series_sentence(series_name: Optional[str], series_number: Optional[s article = "the " if not name.lower().startswith("the ") else "" phrase = f"Book {number} of {article}{name}" return re.sub(r"\s+", " ", phrase).strip() + + +_PEOPLE_SPLIT_RE = re.compile(r"[;,/&]|\band\b", re.IGNORECASE) +_LIST_SPLIT_RE = re.compile(r"[;,\n]") +_SERIES_SEQUENCE_TAG_KEYS: Tuple[str, ...] = ( + "series_index", + "series_position", + "series_sequence", + "series_number", + "seriesnumber", + "book_number", + "booknumber", +) + + +def normalize_metadata_casefold(values: Optional[Mapping[str, Any]]) -> Dict[str, Any]: + normalized: Dict[str, Any] = {} + if not values: + return normalized + for key, value in values.items(): + if value is None: + continue + key_text = str(key).strip().lower() + if not key_text: + continue + if isinstance(value, (list, tuple, set)): + normalized[key_text] = value + else: + text = str(value).strip() + if text: + normalized[key_text] = text + return normalized + + +def split_people_field(raw: Any) -> List[str]: + if raw is None: + return [] + if isinstance(raw, (list, tuple, set)): + results: List[str] = [] + for item in raw: + results.extend(split_people_field(item)) + return results + text = str(raw or "").strip() + if not text: + return [] + tokens = [_token.strip() for _token in _PEOPLE_SPLIT_RE.split(text) if _token.strip()] + seen: set[str] = set() + ordered: List[str] = [] + for token in tokens: + key = token.casefold() + if key in seen: + continue + seen.add(key) + ordered.append(token) + return ordered + + +def split_simple_list(raw: Any) -> List[str]: + if raw is None: + return [] + if isinstance(raw, (list, tuple, set)): + results: List[str] = [] + for item in raw: + results.extend(split_simple_list(item)) + return results + text = str(raw or "").strip() + if not text: + return [] + tokens = [_token.strip() for _token in _LIST_SPLIT_RE.split(text) if _token.strip()] + seen: set[str] = set() + ordered: List[str] = [] + for token in tokens: + key = token.casefold() + if key in seen: + continue + seen.add(key) + ordered.append(token) + return ordered + + +def first_nonempty(*values: Any) -> Optional[str]: + for value in values: + if value is None: + continue + if isinstance(value, (list, tuple, set)): + items = list(value) + if not items: + continue + value = items[0] + text = str(value).strip() + if text: + return text + return None + + +def extract_year(raw: Optional[str]) -> Optional[int]: + if not raw: + return None + text = str(raw).strip() + if not text: + return None + match = re.search(r"(19|20)\d{2}", text) + if match: + try: + return int(match.group(0)) + except ValueError: + return None + try: + parsed = int(text) + except ValueError: + return None + if 0 < parsed < 3000: + return parsed + return None + + +def normalize_series_sequence(raw: Any) -> Optional[str]: + if raw is None: + return None + if isinstance(raw, (int, float)): + if isinstance(raw, float) and (math.isnan(raw) or math.isinf(raw)): + return None + text = str(raw) + else: + text = str(raw).strip() + if not text: + return None + candidate = text.replace(",", ".") + match = _SERIES_NUMBER_RE.search(candidate) + if not match: + return None + normalized = match.group(0) + if "." in normalized: + normalized = normalized.rstrip("0").rstrip(".") + if not normalized: + normalized = "0" + return normalized + try: + return str(int(normalized)) + except ValueError: + cleaned = normalized.lstrip("0") + return cleaned or "0" + + +def build_audiobookshelf_metadata( + tags: Mapping[str, Any], + *, + language: str = "", + filename: str = "", +) -> Dict[str, Any]: + normalized = normalize_metadata_casefold(tags) + title = first_nonempty( + normalized.get("title"), + normalized.get("book_title"), + normalized.get("name"), + normalized.get("album"), + filename, + ) + authors = split_people_field( + normalized.get("authors") + or normalized.get("author") + or normalized.get("album_artist") + or normalized.get("artist") + ) + narrators = split_people_field(normalized.get("narrators") or normalized.get("narrator")) + description = first_nonempty( + normalized.get("description"), normalized.get("summary"), normalized.get("comment") + ) + genres = split_simple_list(normalized.get("genre")) + keywords = split_simple_list(normalized.get("tags") or normalized.get("keywords")) + lang = first_nonempty(normalized.get("language"), normalized.get("lang")) or language or "" + series_name = first_nonempty( + normalized.get("series"), + normalized.get("series_name"), + normalized.get("seriesname"), + normalized.get("series_title"), + normalized.get("seriestitle"), + ) + + series_sequence = None + for key in _SERIES_SEQUENCE_TAG_KEYS: + raw_value = normalized.get(key) + seq = normalize_series_sequence(raw_value) + if seq: + series_sequence = seq + break + if not series_name: + series_sequence = None + + data: Dict[str, Any] = { + "title": title, + "subtitle": normalized.get("subtitle"), + "authors": authors, + "narrators": narrators, + "description": description, + "publisher": normalized.get("publisher"), + "genres": genres, + "tags": keywords, + "language": lang, + "publishedYear": extract_year( + normalized.get("published") + or normalized.get("publication_year") + or normalized.get("date") + or normalized.get("year") + ), + "seriesName": series_name, + "seriesSequence": series_sequence, + "isbn": first_nonempty(normalized.get("isbn"), normalized.get("asin")), + } + published_date = first_nonempty( + normalized.get("published"), normalized.get("publication_date"), normalized.get("date") + ) + if published_date: + data["publishedDate"] = published_date + + rating_text = first_nonempty(normalized.get("rating"), normalized.get("my_rating")) + if rating_text: + try: + data["rating"] = float(str(rating_text).strip()) + except ValueError: + pass + rating_max_text = first_nonempty( + normalized.get("rating_max"), normalized.get("rating_scale") + ) + if rating_max_text: + try: + data["ratingMax"] = float(str(rating_max_text).strip()) + except ValueError: + pass + + cleaned: Dict[str, Any] = {} + for key, value in data.items(): + if value is None: + continue + if isinstance(value, str) and not value.strip(): + continue + if isinstance(value, (list, tuple)) and not value: + continue + cleaned[key] = value + return cleaned + + +def load_audiobookshelf_chapters( + metadata_path: Path, +) -> Optional[List[Dict[str, Any]]]: + if not metadata_path.exists(): + return None + try: + payload = json.loads(metadata_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return None + chapters = payload.get("chapters") + if not isinstance(chapters, list): + return None + cleaned: List[Dict[str, Any]] = [] + for entry in chapters: + if not isinstance(entry, Mapping): + continue + title = first_nonempty(entry.get("title"), entry.get("original_title")) + start = entry.get("start") + end = entry.get("end") + if title and start is not None and end is not None: + cleaned.append({"title": str(title), "start": start, "end": end}) + return cleaned or None diff --git a/abogen/infrastructure/exporters.py b/abogen/infrastructure/exporters.py index cc2f599..d3e6575 100644 --- a/abogen/infrastructure/exporters.py +++ b/abogen/infrastructure/exporters.py @@ -9,6 +9,17 @@ from typing import Any, Dict, List, Optional, Mapping, Sequence import static_ffmpeg +from abogen.domain.metadata_helpers import ( + normalize_metadata_casefold, + split_people_field, + split_simple_list, + 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, @@ -305,125 +316,20 @@ class ExportService: def build_audiobookshelf_metadata(self, job: Any) -> Dict[str, Any]: """Build Audiobookshelf metadata from job.""" - tags = self._normalize_metadata_casefold(getattr(job, "metadata_tags", {})) filename = Path(getattr(job, "original_filename", "") or "").stem or "Audiobook" - - title = self._first_nonempty( - tags.get("title"), - tags.get("book_title"), - tags.get("name"), - tags.get("album"), - filename, + return _build_abs_metadata( + getattr(job, "metadata_tags", {}), + language=getattr(job, "language", "") or "", + filename=filename, ) - authors = self._split_people_field( - tags.get("authors") - or tags.get("author") - or tags.get("album_artist") - or tags.get("artist") - ) - narrators = self._split_people_field(tags.get("narrators") or tags.get("narrator")) - description = self._first_nonempty( - tags.get("description"), tags.get("summary"), tags.get("comment") - ) - genres = self._split_simple_list(tags.get("genre")) - keywords = self._split_simple_list(tags.get("tags") or tags.get("keywords")) - language = self._first_nonempty(tags.get("language"), tags.get("lang")) or getattr(job, "language", "") or "" - series_name = self._first_nonempty( - tags.get("series"), - tags.get("series_name"), - tags.get("seriesname"), - tags.get("series_title"), - tags.get("seriestitle"), - ) - - series_sequence = None - for key in ("series_index", "series_position", "series_sequence", "series_number", "seriesnumber", "book_number", "booknumber"): - raw = tags.get(key) - normalized = self._normalize_series_sequence(raw) - if normalized: - series_sequence = normalized - break - if not series_name: - series_sequence = None - - data = { - "title": title, - "subtitle": tags.get("subtitle"), - "authors": authors, - "narrators": narrators, - "description": description, - "publisher": tags.get("publisher"), - "genres": genres, - "tags": keywords, - "language": language, - "publishedYear": self._extract_year( - tags.get("published") or tags.get("publication_year") or tags.get("date") or tags.get("year") - ), - "seriesName": series_name, - "seriesSequence": series_sequence, - "isbn": self._first_nonempty(tags.get("isbn"), tags.get("asin")), - } - - published_date = self._first_nonempty( - tags.get("published"), tags.get("publication_date"), tags.get("date") - ) - if published_date: - data["publishedDate"] = published_date - - rating_text = self._first_nonempty(tags.get("rating"), tags.get("my_rating")) - if rating_text: - try: - data["rating"] = float(str(rating_text).strip()) - except ValueError: - pass - rating_max_text = self._first_nonempty(tags.get("rating_max"), tags.get("rating_scale")) - if rating_max_text: - try: - data["ratingMax"] = float(str(rating_max_text).strip()) - except ValueError: - pass - - # Remove empty values - cleaned = {} - for key, value in data.items(): - if value is None: - continue - if isinstance(value, str) and not value.strip(): - continue - if isinstance(value, (list, tuple)) and not value: - continue - cleaned[key] = value - return cleaned - + 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)) - if not metadata_path.exists(): - return None - try: - payload = json.loads(metadata_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return None - chapters = payload.get("chapters") - if not isinstance(chapters, list): - return None - cleaned = [] - for entry in chapters: - if not isinstance(entry, Mapping): - continue - title = self._first_nonempty(entry.get("title"), entry.get("original_title")) - start = entry.get("start") - end = entry.get("end") - if title is None or not isinstance(start, (int, float)): - continue - chapter_payload = {"title": title, "start": float(start)} - if isinstance(end, (int, float)): - chapter_payload["end"] = float(end) - cleaned.append(chapter_payload) - return cleaned or None + return _load_abs_chapters(metadata_path) def upload_audiobookshelf( self, @@ -520,137 +426,6 @@ class ExportService: # Helpers # ---------------------------------------------------------------------- - @staticmethod - def _normalize_metadata_casefold(values: Optional[Mapping[str, Any]]) -> Dict[str, Any]: - normalized = {} - if not values: - return normalized - for key, value in values.items(): - if value is None: - continue - key_text = str(key).strip().lower() - if not key_text: - continue - if isinstance(value, (list, tuple, set)): - normalized[key_text] = value - else: - text = str(value).strip() - if text: - normalized[key_text] = text - return normalized - - @staticmethod - def _split_people_field(raw: Any) -> List[str]: - if raw is None: - return [] - if isinstance(raw, (list, tuple, set)): - results = [] - for item in raw: - results.extend(ExportService._split_people_field(item)) - return results - text = str(raw or "").strip() - if not text: - return [] - import re - tokens = [token.strip() for token in re.split(r"[;,/&]|\band\b", text, flags=re.IGNORECASE) if token.strip()] - seen = set() - ordered = [] - for token in tokens: - key = token.casefold() - if key in seen: - continue - seen.add(key) - ordered.append(token) - return ordered - - @staticmethod - def _split_simple_list(raw: Any) -> List[str]: - if raw is None: - return [] - if isinstance(raw, (list, tuple, set)): - results = [] - for item in raw: - results.extend(ExportService._split_simple_list(item)) - return results - text = str(raw or "").strip() - if not text: - return [] - import re - tokens = [token.strip() for token in re.split(r"[;,\n]", text) if token.strip()] - seen = set() - ordered = [] - for token in tokens: - key = token.casefold() - if key in seen: - continue - seen.add(key) - ordered.append(token) - return ordered - - @staticmethod - def _first_nonempty(*values: Any) -> Optional[str]: - for value in values: - if value is None: - continue - if isinstance(value, (list, tuple, set)): - items = list(value) - if not items: - continue - value = items[0] - text = str(value).strip() - if text: - return text - return None - - @staticmethod - def _extract_year(raw: Optional[str]) -> Optional[int]: - if not raw: - return None - text = str(raw).strip() - if not text: - return None - import re - match = re.search(r"(19|20)\d{2}", text) - if match: - try: - return int(match.group(0)) - except ValueError: - return None - try: - parsed = int(text) - except ValueError: - return None - if 0 < parsed < 3000: - return parsed - return None - - @staticmethod - def _normalize_series_sequence(raw: Any) -> Optional[str]: - if raw is None: - return None - if isinstance(raw, (int, float)): - if isinstance(raw, float) and (raw != raw or raw == float("inf") or raw == float("-inf")): - return None - text = str(raw) - else: - text = str(raw).strip() - if not text: - return None - candidate = text.replace(",", ".") - import re - match = re.search(r"\d+(?:\.\d+)?", candidate) - if not match: - return None - normalized = match.group(0) - if "." in normalized: - normalized = normalized.rstrip("0").rstrip(".") - return normalized or "0" - try: - return str(int(normalized)) - except ValueError: - cleaned = normalized.lstrip("0") - return cleaned or "0" - @staticmethod def _coerce_bool(value: Any, default: bool = True) -> bool: if isinstance(value, bool): diff --git a/abogen/integrations/audiobookshelf.py b/abogen/integrations/audiobookshelf.py index 0538cce..0fb9c16 100644 --- a/abogen/integrations/audiobookshelf.py +++ b/abogen/integrations/audiobookshelf.py @@ -2,9 +2,7 @@ from __future__ import annotations import json import logging -import math import mimetypes -import re from contextlib import ExitStack from dataclasses import dataclass from pathlib import Path @@ -12,6 +10,8 @@ from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple import httpx +from abogen.domain.metadata_helpers import normalize_series_sequence + logger = logging.getLogger(__name__) @@ -641,40 +641,7 @@ class AudiobookshelfClient: for key in preferred_keys: if key not in metadata: continue - normalized = AudiobookshelfClient._normalize_series_sequence(metadata.get(key)) + normalized = normalize_series_sequence(metadata.get(key)) if normalized: return normalized return "" - - @staticmethod - def _normalize_series_sequence(raw: Any) -> str: - if raw is None: - return "" - - if isinstance(raw, (int, float)): - if isinstance(raw, float) and (math.isnan(raw) or math.isinf(raw)): - return "" - text = str(raw) - else: - text = str(raw).strip() - - if not text: - return "" - - candidate = text.replace(",", ".") - match = re.search(r"\d+(?:\.\d+)?", candidate) - if not match: - return "" - - normalized = match.group(0) - if "." in normalized: - normalized = normalized.rstrip("0").rstrip(".") - if not normalized: - normalized = "0" - return normalized - - try: - return str(int(normalized)) - except ValueError: - cleaned = normalized.lstrip("0") - return cleaned or "0" diff --git a/abogen/webui/service.py b/abogen/webui/service.py index bdf90b4..f3eef73 100644 --- a/abogen/webui/service.py +++ b/abogen/webui/service.py @@ -2,9 +2,7 @@ from __future__ import annotations import json import logging -import math import os -import re import shutil import sys import threading @@ -14,7 +12,7 @@ import traceback from dataclasses import dataclass, field from enum import Enum from pathlib import Path -from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping, Tuple +from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping from abogen.utils import get_internal_cache_path, get_user_settings_dir, load_config from abogen.voice_cache import bootstrap_voice_cache @@ -23,6 +21,17 @@ from abogen.integrations.audiobookshelf import ( 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: @@ -53,9 +62,6 @@ _JOB_LEVEL_MAP: Dict[str, int] = { } -_PEOPLE_SPLIT_RE = re.compile(r"[;,/&]|\band\b", re.IGNORECASE) - - def _emit_job_log(job_id: str, level: str, message: str) -> None: normalized = (level or "info").lower() log_level = _JOB_LEVEL_MAP.get(normalized, logging.INFO) @@ -252,234 +258,13 @@ class Job: } -def _normalize_metadata_casefold(values: Optional[Mapping[str, Any]]) -> Dict[str, Any]: - normalized: Dict[str, Any] = {} - if not values: - return normalized - for key, value in values.items(): - if value is None: - continue - key_text = str(key).strip().lower() - if not key_text: - continue - if isinstance(value, (list, tuple, set)): - normalized[key_text] = value - else: - text = str(value).strip() - if text: - normalized[key_text] = text - return normalized - - -def _split_people_field(raw: Any) -> List[str]: - if raw is None: - return [] - if isinstance(raw, (list, tuple, set)): - results: List[str] = [] - for item in raw: - results.extend(_split_people_field(item)) - return results - text = str(raw or "").strip() - if not text: - return [] - tokens = [_token.strip() for _token in _PEOPLE_SPLIT_RE.split(text) if _token.strip()] - seen: set[str] = set() - ordered: List[str] = [] - for token in tokens: - key = token.casefold() - if key in seen: - continue - seen.add(key) - ordered.append(token) - return ordered - - -_LIST_SPLIT_RE = re.compile(r"[;,\n]") -_SERIES_SEQUENCE_NUMBER_RE = re.compile(r"\d+(?:\.\d+)?") - -_SERIES_SEQUENCE_TAG_KEYS: Tuple[str, ...] = ( - "series_index", - "series_position", - "series_sequence", - "series_number", - "seriesnumber", - "book_number", - "booknumber", -) - - -def _split_simple_list(raw: Any) -> List[str]: - if raw is None: - return [] - if isinstance(raw, (list, tuple, set)): - results: List[str] = [] - for item in raw: - results.extend(_split_simple_list(item)) - return results - text = str(raw or "").strip() - if not text: - return [] - tokens = [_token.strip() for _token in _LIST_SPLIT_RE.split(text) if _token.strip()] - seen: set[str] = set() - ordered: List[str] = [] - for token in tokens: - key = token.casefold() - if key in seen: - continue - seen.add(key) - ordered.append(token) - return ordered - - -def _first_nonempty(*values: Any) -> Optional[str]: - for value in values: - if value is None: - continue - if isinstance(value, (list, tuple, set)): - items = list(value) - if not items: - continue - value = items[0] - text = str(value).strip() - if text: - return text - return None - - -def _extract_year(raw: Optional[str]) -> Optional[int]: - if not raw: - return None - text = str(raw).strip() - if not text: - return None - match = re.search(r"(19|20)\d{2}", text) - if match: - try: - return int(match.group(0)) - except ValueError: - return None - try: - parsed = int(text) - except ValueError: - return None - if 0 < parsed < 3000: - return parsed - return None - - def build_audiobookshelf_metadata(job: Job) -> Dict[str, Any]: - tags = _normalize_metadata_casefold(job.metadata_tags) filename = Path(job.original_filename or "").stem or job.original_filename or "Audiobook" - title = _first_nonempty( - tags.get("title"), - tags.get("book_title"), - tags.get("name"), - tags.get("album"), - filename, + return _build_abs_metadata( + job.metadata_tags, + language=job.language or "", + filename=filename, ) - authors = _split_people_field( - tags.get("authors") - or tags.get("author") - or tags.get("album_artist") - or tags.get("artist") - ) - narrators = _split_people_field(tags.get("narrators") or tags.get("narrator")) - description = _first_nonempty(tags.get("description"), tags.get("summary"), tags.get("comment")) - genres = _split_simple_list(tags.get("genre")) - keywords = _split_simple_list(tags.get("tags") or tags.get("keywords")) - language = _first_nonempty(tags.get("language"), tags.get("lang")) or job.language or "" - series_name = _first_nonempty( - tags.get("series"), - tags.get("series_name"), - tags.get("seriesname"), - tags.get("series_title"), - tags.get("seriestitle"), - ) - - series_sequence = None - for key in _SERIES_SEQUENCE_TAG_KEYS: - raw_value = tags.get(key) - normalized_sequence = _normalize_series_sequence(raw_value) - if normalized_sequence: - series_sequence = normalized_sequence - break - if not series_name: - series_sequence = None - data: Dict[str, Any] = { - "title": title, - "subtitle": tags.get("subtitle"), - "authors": authors, - "narrators": narrators, - "description": description, - "publisher": tags.get("publisher"), - "genres": genres, - "tags": keywords, - "language": language, - "publishedYear": _extract_year(tags.get("published") or tags.get("publication_year") or tags.get("date") or tags.get("year")), - "seriesName": series_name, - "seriesSequence": series_sequence, - "isbn": _first_nonempty(tags.get("isbn"), tags.get("asin")), - } - published_date = _first_nonempty(tags.get("published"), tags.get("publication_date"), tags.get("date")) - if published_date: - data["publishedDate"] = published_date - - rating_text = _first_nonempty(tags.get("rating"), tags.get("my_rating")) - if rating_text: - try: - data["rating"] = float(str(rating_text).strip()) - except ValueError: - pass - rating_max_text = _first_nonempty(tags.get("rating_max"), tags.get("rating_scale")) - if rating_max_text: - try: - data["ratingMax"] = float(str(rating_max_text).strip()) - except ValueError: - pass - # Remove empty values - cleaned: Dict[str, Any] = {} - for key, value in data.items(): - if value is None: - continue - if isinstance(value, str) and not value.strip(): - continue - if isinstance(value, (list, tuple)) and not value: - continue - cleaned[key] = value - return cleaned - - -def _normalize_series_sequence(raw: Any) -> Optional[str]: - if raw is None: - return None - - if isinstance(raw, (int, float)): - if isinstance(raw, float) and (math.isnan(raw) or math.isinf(raw)): - return None - text = str(raw) - else: - text = str(raw).strip() - - if not text: - return None - - candidate = text.replace(",", ".") - match = _SERIES_SEQUENCE_NUMBER_RE.search(candidate) - if not match: - return None - - normalized = match.group(0) - if "." in normalized: - normalized = normalized.rstrip("0").rstrip(".") - if not normalized: - normalized = "0" - return normalized - - try: - return str(int(normalized)) - except ValueError: - cleaned = normalized.lstrip("0") - return cleaned or "0" def load_audiobookshelf_chapters(job: Job) -> Optional[List[Dict[str, Any]]]: @@ -487,32 +272,7 @@ def load_audiobookshelf_chapters(job: Job) -> Optional[List[Dict[str, Any]]]: if not metadata_ref: return None metadata_path = metadata_ref if isinstance(metadata_ref, Path) else Path(str(metadata_ref)) - if not metadata_path.exists(): - return None - try: - payload = json.loads(metadata_path.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return None - chapters = payload.get("chapters") - if not isinstance(chapters, list): - return None - cleaned: List[Dict[str, Any]] = [] - for entry in chapters: - if not isinstance(entry, Mapping): - continue - title = _first_nonempty(entry.get("title"), entry.get("original_title")) - start = entry.get("start") - end = entry.get("end") - if title is None or not isinstance(start, (int, float)): - continue - chapter_payload: Dict[str, Any] = { - "title": title, - "start": float(start), - } - if isinstance(end, (int, float)): - chapter_payload["end"] = float(end) - cleaned.append(chapter_payload) - return cleaned or None + return _load_abs_chapters(metadata_path) def _existing_paths(paths: Iterable[Any]) -> List[Path]: diff --git a/tests/test_metadata_helpers_abs.py b/tests/test_metadata_helpers_abs.py new file mode 100644 index 0000000..55ef0cb --- /dev/null +++ b/tests/test_metadata_helpers_abs.py @@ -0,0 +1,170 @@ +"""Tests for domain/metadata_helpers.py Audiobookshelf helpers.""" +from abogen.domain.metadata_helpers import ( + normalize_metadata_casefold, + split_people_field, + split_simple_list, + first_nonempty, + extract_year, + normalize_series_sequence, + build_audiobookshelf_metadata, + load_audiobookshelf_chapters, +) + + +# --- normalize_metadata_casefold --- + +def test_normalize_metadata_casefold_basic(): + result = normalize_metadata_casefold({"Title": " Hello ", "Author": None, "": "skip"}) + assert result == {"title": "Hello", "author": ""} or result == {"title": "Hello"} + + +def test_normalize_metadata_casefold_preserves_lists(): + result = normalize_metadata_casefold({"tags": ["a", "b"]}) + assert result["tags"] == ["a", "b"] + + +# --- split_people_field --- + +def test_split_people_field_single(): + assert split_people_field("J.K. Rowling") == ["J.K. Rowling"] + + +def test_split_people_field_multiple(): + result = split_people_field("Tolkien, Lewis & Martin") + assert "Tolkien" in result + assert "Lewis" in result + assert "Martin" in result + + +def test_split_people_field_deduplicates(): + result = split_people_field("Tolkien, tolkien, TOLKIEN") + assert len(result) == 1 + + +def test_split_people_field_none(): + assert split_people_field(None) == [] + + +def test_split_people_field_list(): + result = split_people_field(["Author A", "Author B"]) + assert result == ["Author A", "Author B"] + + +# --- split_simple_list --- + +def test_split_simple_list_basic(): + result = split_simple_list("fantasy, sci-fi; thriller") + assert "fantasy" in result + assert "sci-fi" in result + assert "thriller" in result + + +def test_split_simple_list_none(): + assert split_simple_list(None) == [] + + +# --- first_nonempty --- + +def test_first_nonempty_basic(): + assert first_nonempty(None, "", "hello") == "hello" + + +def test_first_nonempty_first_wins(): + assert first_nonempty("first", "second") == "first" + + +def test_first_nonempty_none(): + assert first_nonempty(None, None) is None + + +def test_first_nonempty_list(): + assert first_nonempty(None, ["a", "b"]) == "a" + + +# --- extract_year --- + +def test_extract_year_full_date(): + assert extract_year("Published on 2023-05-15") == 2023 + + +def test_extract_year_plain(): + assert extract_year("2020") == 2020 + + +def test_extract_year_none(): + assert extract_year(None) is None + + +def test_extract_year_invalid(): + assert extract_year("no year here") is None + + +# --- normalize_series_sequence --- + +def test_normalize_series_sequence_int(): + assert normalize_series_sequence(3) == "3" + + +def test_normalize_series_sequence_float(): + assert normalize_series_sequence(2.5) == "2.5" + + +def test_normalize_series_sequence_string(): + assert normalize_series_sequence(" 12 ") == "12" + + +def test_normalize_series_sequence_comma(): + assert normalize_series_sequence("1,5") == "1.5" + + +def test_normalize_series_sequence_none(): + assert normalize_series_sequence(None) is None + + +def test_normalize_series_sequence_nan(): + import math + assert normalize_series_sequence(float("nan")) is None + + +# --- build_audiobookshelf_metadata --- + +def test_build_audiobookshelf_metadata_basic(): + tags = { + "title": "My Book", + "author": "Author Name", + "description": "A great book", + } + result = build_audiobookshelf_metadata(tags, language="en") + assert result["title"] == "My Book" + assert result["authors"] == ["Author Name"] + assert result["language"] == "en" + + +def test_build_audiobookshelf_metadata_series(): + tags = { + "title": "Book 2", + "series": "My Series", + "series_index": "2", + } + result = build_audiobookshelf_metadata(tags, language="en") + assert result["seriesName"] == "My Series" + assert result["seriesSequence"] == "2" + + +def test_build_audiobookshelf_metadata_fallback_title(): + tags = {"author": "Someone"} + result = build_audiobookshelf_metadata(tags, language="en", filename="chapter1") + assert result["title"] == "chapter1" + + +def test_build_audiobookshelf_metadata_empty(): + result = build_audiobookshelf_metadata({}, language="en") + assert result["language"] == "en" + assert "authors" not in result # empty list stripped + + +def test_build_audiobookshelf_metadata_strips_empty(): + tags = {"title": "Book", "subtitle": "", "description": None} + result = build_audiobookshelf_metadata(tags, language="en") + assert "subtitle" not in result + assert "description" not in result