mirror of
https://github.com/denizsafak/abogen.git
synced 2026-07-18 13:40:27 +02:00
refactor(domain): deduplicate metadata helpers across 3 layers
Extracted 8 metadata functions from service.py, exporters.py, and audiobookshelf.py into domain/metadata_helpers.py: - normalize_metadata_casefold, split_people_field, split_simple_list - first_nonempty, extract_year, normalize_series_sequence - build_audiobookshelf_metadata, load_audiobookshelf_chapters service.py, exporters.py, and audiobookshelf.py now import from domain instead of maintaining separate copies. Thin wrappers adapt to layer interfaces (Job objects, etc.). Net -231 lines. 1006 tests pass.
This commit is contained in:
Artem Akymenko
@@ -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
|
||||
|
||||
@@ -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,95 +316,12 @@ 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."""
|
||||
@@ -401,29 +329,7 @@ class ExportService:
|
||||
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):
|
||||
|
||||
@@ -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"
|
||||
|
||||
+17
-257
@@ -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]:
|
||||
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user