diff --git a/abogen/domain/metadata_helpers.py b/abogen/domain/metadata_helpers.py index fe05863..0e9ab31 100644 --- a/abogen/domain/metadata_helpers.py +++ b/abogen/domain/metadata_helpers.py @@ -21,6 +21,60 @@ _SERIES_NUMBER_KEYS = ( ) _SERIES_NUMBER_RE = re.compile(r"\d+(?:\.\d+)?") +_SERIES_NAME_ALIASES = ("series", "series_name", "seriesname", "series_title", "seriestitle") +_SERIES_INDEX_ALIASES = ("series_index", "series_sequence", "series_position", "book_number") +_AUTHOR_ALIASES = ("author", "authors") +_DESCRIPTION_ALIASES = ("description", "summary") +_TAGS_ALIASES = ("tags", "keywords", "genre") + + +def expand_metadata_aliases(tags: Mapping[str, Any]) -> Dict[str, Any]: + """Expand concept aliases so each concept has all canonical keys set. + + One input concept fans out to multiple keys so that downstream consumers + can look up any variant and find the value. + + Expanded concepts: + series -> series, series_name, seriesname, series_title, seriestitle + series_index -> series_index, series_sequence, series_position, book_number + author -> author, authors + description -> description, summary + tags -> tags, keywords, genre + """ + if not tags: + return {} + + result: Dict[str, Any] = {} + for key, value in tags.items(): + if value is None: + continue + text = str(value).strip() if not isinstance(value, (list, tuple, set)) else value + if not text: + continue + key_lower = str(key).strip().lower() + if not key_lower: + continue + + if key_lower in _SERIES_NAME_ALIASES: + for alias in _SERIES_NAME_ALIASES: + result[alias] = text + elif key_lower in _SERIES_INDEX_ALIASES: + for alias in _SERIES_INDEX_ALIASES: + result[alias] = text + elif key_lower in _AUTHOR_ALIASES: + for alias in _AUTHOR_ALIASES: + result[alias] = text + elif key_lower in _DESCRIPTION_ALIASES: + for alias in _DESCRIPTION_ALIASES: + result[alias] = text + elif key_lower in _TAGS_ALIASES: + for alias in _TAGS_ALIASES: + result[alias] = text + else: + result[key_lower] = text + + return result + def normalize_metadata_map(values: Optional[Mapping[str, Any]]) -> Dict[str, str]: normalized: Dict[str, str] = {} diff --git a/abogen/domain/metadata_overrides.py b/abogen/domain/metadata_overrides.py index bd5b9fa..9c1baa7 100644 --- a/abogen/domain/metadata_overrides.py +++ b/abogen/domain/metadata_overrides.py @@ -8,23 +8,22 @@ from __future__ import annotations from typing import Any, Dict, Mapping +from abogen.domain.metadata_helpers import expand_metadata_aliases + def normalize_opds_metadata(metadata_payload: Mapping[str, Any]) -> Dict[str, Any]: """Normalize OPDS/Calibre metadata into canonical override keys. Takes a metadata payload with various key aliases (e.g. 'series'/'series_name', - 'tags'/'keywords', 'authors'/'creator') and returns a dict with canonical - keys set. + 'tags'/'keywords', 'authors'/'creator') and returns a dict with all + concept aliases expanded. Args: metadata_payload: Raw metadata dict from OPDS/Calibre import. Returns: - Dict with canonical metadata keys (series, series_index, tags, - description, subtitle, publisher, authors). + Dict with all canonical metadata key aliases expanded. """ - metadata_overrides: Dict[str, Any] = {} - def _stringify(value: Any) -> str: if value is None: return "" @@ -33,67 +32,25 @@ def normalize_opds_metadata(metadata_payload: Mapping[str, Any]) -> Dict[str, An return ", ".join(part for part in parts if part) return str(value).strip() - raw_series = metadata_payload.get("series") or metadata_payload.get("series_name") - series_name = str(raw_series or "").strip() - if series_name: - metadata_overrides["series"] = series_name - metadata_overrides.setdefault("series_name", series_name) + # Map OPDS-specific keys to common concept keys before expansion + normalized_input: Dict[str, Any] = {} + for key, value in metadata_payload.items(): + if value is None: + continue + key_lower = str(key).strip().lower() + if not key_lower: + continue + text = _stringify(value) + if not text: + continue - series_index_value = ( - metadata_payload.get("series_index") - or metadata_payload.get("series_position") - or metadata_payload.get("series_sequence") - or metadata_payload.get("book_number") - ) - if series_index_value is not None: - series_index_text = str(series_index_value).strip() - if series_index_text: - metadata_overrides.setdefault("series_index", series_index_text) - metadata_overrides.setdefault("series_position", series_index_text) - metadata_overrides.setdefault("series_sequence", series_index_text) - metadata_overrides.setdefault("book_number", series_index_text) + # Map OPDS-specific author aliases + if key_lower in ("creator", "dc_creator"): + normalized_input["author"] = text + # Map OPDS-specific subtitle aliases + elif key_lower in ("sub_title", "calibre_subtitle"): + normalized_input["subtitle"] = text + else: + normalized_input[key_lower] = text - tags_value = metadata_payload.get("tags") or metadata_payload.get("keywords") - if tags_value: - tags_text = _stringify(tags_value) - if tags_text: - metadata_overrides.setdefault("tags", tags_text) - metadata_overrides.setdefault("keywords", tags_text) - metadata_overrides.setdefault("genre", tags_text) - - description_value = metadata_payload.get("description") or metadata_payload.get("summary") - if description_value: - description_text = _stringify(description_value) - if description_text: - metadata_overrides.setdefault("description", description_text) - metadata_overrides.setdefault("summary", description_text) - - subtitle_value = ( - metadata_payload.get("subtitle") - or metadata_payload.get("sub_title") - or metadata_payload.get("calibre_subtitle") - ) - if subtitle_value: - subtitle_text = _stringify(subtitle_value) - if subtitle_text: - metadata_overrides.setdefault("subtitle", subtitle_text) - - publisher_value = metadata_payload.get("publisher") - if publisher_value: - publisher_text = _stringify(publisher_value) - if publisher_text: - metadata_overrides.setdefault("publisher", publisher_text) - - authors_value = ( - metadata_payload.get("authors") - or metadata_payload.get("author") - or metadata_payload.get("creator") - or metadata_payload.get("dc_creator") - ) - if authors_value: - authors_text = _stringify(authors_value) - if authors_text: - metadata_overrides.setdefault("authors", authors_text) - metadata_overrides.setdefault("author", authors_text) - - return metadata_overrides + return expand_metadata_aliases(normalized_input) diff --git a/abogen/epub3/exporter.py b/abogen/epub3/exporter.py index ce1a736..a354e3d 100644 --- a/abogen/epub3/exporter.py +++ b/abogen/epub3/exporter.py @@ -12,6 +12,7 @@ from typing import Any, Dict, Iterable, List, Optional, Pattern, Sequence, Tuple import zipfile from abogen.text_extractor import ExtractedChapter, ExtractionResult +from abogen.domain.metadata_helpers import normalize_metadata_map @dataclass(slots=True) @@ -59,7 +60,7 @@ class EPUB3PackageBuilder: self.output_path = output_path self.book_id = book_id or str(uuid.uuid4()) self.extraction = extraction - self.metadata_tags = _normalize_metadata(metadata_tags) + self.metadata_tags = normalize_metadata_map(metadata_tags) self.chapter_markers = list(chapter_markers or []) self.chunk_markers = list(chunk_markers or []) self.chunks = list(chunks or []) @@ -550,15 +551,6 @@ class ChunkLookup: by_chapter: Dict[int, List[Dict[str, Any]]] -def _normalize_metadata(metadata: Optional[Dict[str, Any]]) -> Dict[str, str]: - normalized: Dict[str, str] = {} - for key, value in (metadata or {}).items(): - if value is None: - continue - normalized[str(key).lower()] = str(value) - return normalized - - def _combine_metadata(*sources: Dict[str, Any]) -> Dict[str, str]: combined: Dict[str, str] = {} for source in sources: diff --git a/abogen/infrastructure/exporters.py b/abogen/infrastructure/exporters.py index 1f1126a..16ca716 100644 --- a/abogen/infrastructure/exporters.py +++ b/abogen/infrastructure/exporters.py @@ -10,7 +10,6 @@ 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, diff --git a/abogen/webui/routes/utils/form.py b/abogen/webui/routes/utils/form.py index 77398f2..8966c24 100644 --- a/abogen/webui/routes/utils/form.py +++ b/abogen/webui/routes/utils/form.py @@ -31,6 +31,7 @@ from abogen.webui.routes.utils.voice import ( template_options, ) from abogen.domain.speaker_metadata import prepare_speaker_metadata +from abogen.domain.metadata_helpers import expand_metadata_aliases from abogen.domain.voice_resolution import ( formula_from_profile, resolve_voice_setting, @@ -546,35 +547,27 @@ def apply_book_step_form( if "meta_subtitle" in form: pending.metadata_tags["subtitle"] = str(form.get("meta_subtitle", "")).strip() + # Collect user-editable metadata fields that have concept aliases + user_metadata: Dict[str, str] = {} if "meta_author" in form: - authors = str(form.get("meta_author", "")).strip() - pending.metadata_tags["authors"] = authors - pending.metadata_tags["author"] = authors - + user_metadata["author"] = str(form.get("meta_author", "")).strip() if "meta_series" in form: - series = str(form.get("meta_series", "")).strip() - pending.metadata_tags["series"] = series - pending.metadata_tags["series_name"] = series - pending.metadata_tags["seriesname"] = series - pending.metadata_tags["series_title"] = series - pending.metadata_tags["seriestitle"] = series - # If user manually edits series, update opds_series too so it persists - if "opds_series" in pending.metadata_tags: - pending.metadata_tags["opds_series"] = series - + user_metadata["series"] = str(form.get("meta_series", "")).strip() if "meta_series_index" in form: - idx = str(form.get("meta_series_index", "")).strip() - pending.metadata_tags["series_index"] = idx - pending.metadata_tags["series_sequence"] = idx + user_metadata["series_index"] = str(form.get("meta_series_index", "")).strip() + if "meta_description" in form: + user_metadata["description"] = str(form.get("meta_description", "")).strip() + + if user_metadata: + expanded = expand_metadata_aliases(user_metadata) + pending.metadata_tags.update(expanded) + # If user manually edits series, update opds_series too so it persists + if "meta_series" in form and "opds_series" in pending.metadata_tags: + pending.metadata_tags["opds_series"] = expanded.get("series", "") if "meta_publisher" in form: pending.metadata_tags["publisher"] = str(form.get("meta_publisher", "")).strip() - if "meta_description" in form: - desc = str(form.get("meta_description", "")).strip() - pending.metadata_tags["description"] = desc - pending.metadata_tags["summary"] = desc - if coerce_bool(form.get("remove_cover"), False): pending.cover_image_path = None pending.cover_image_mime = None diff --git a/abogen/webui/service.py b/abogen/webui/service.py index e160dbc..dd5fa0c 100644 --- a/abogen/webui/service.py +++ b/abogen/webui/service.py @@ -14,6 +14,8 @@ from enum import Enum from pathlib import Path from typing import Any, Callable, Dict, Iterable, List, Optional, Mapping +from abogen.domain.metadata_helpers import normalize_metadata_map + from abogen.domain.enums import Language from abogen.utils import get_internal_cache_path, get_user_settings_dir from abogen.voice_cache import bootstrap_voice_cache @@ -396,7 +398,7 @@ class ConversionService: normalization_overrides: Optional[Mapping[str, Any]] = None, ) -> Job: job_id = uuid.uuid4().hex - normalized_metadata = self._normalize_metadata_tags(metadata_tags) + normalized_metadata = normalize_metadata_map(metadata_tags) normalized_chapters = self._normalize_chapters(chapters) normalized_chunks = self._normalize_chunks(chunks) if total_characters <= 0 and normalized_chapters: @@ -1053,20 +1055,6 @@ class ConversionService: except (TypeError, ValueError): return None - @staticmethod - def _normalize_metadata_tags(values: Optional[Mapping[str, Any]]) -> Dict[str, str]: - if not values: - return {} - normalized: Dict[str, str] = {} - for key, raw_value in values.items(): - if raw_value is None: - continue - key_str = str(key).strip() - if not key_str: - continue - normalized[key_str] = str(raw_value) - return normalized - @classmethod def _normalize_chapters(cls, chapters: Optional[Iterable[Any]]) -> List[Dict[str, Any]]: if not chapters: @@ -1143,7 +1131,7 @@ class ConversionService: entry["enabled"] = enabled metadata_payload = raw_dict.get("metadata") or raw_dict.get("metadata_tags") - normalized_metadata = cls._normalize_metadata_tags(metadata_payload) + normalized_metadata = normalize_metadata_map(metadata_payload) if normalized_metadata: entry["metadata"] = normalized_metadata diff --git a/tests/test_metadata_helpers.py b/tests/test_metadata_helpers.py index dd6d962..3da47d7 100644 --- a/tests/test_metadata_helpers.py +++ b/tests/test_metadata_helpers.py @@ -5,6 +5,7 @@ sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..") from abogen.domain.metadata_helpers import ( normalize_metadata_map, + expand_metadata_aliases, format_author_sentence, ensure_sentence, normalize_series_number, @@ -185,3 +186,114 @@ class TestBuildMetadataPayload: result = build_metadata_payload(speakers=original) result["speakers"]["narrator"] = "M2" assert original["narrator"] == "M1" + + +class TestExpandMetadataAliases: + def test_empty(self): + assert expand_metadata_aliases({}) == {} + + def test_none(self): + assert expand_metadata_aliases(None) == {} # type: ignore + + def test_series_expansion(self): + result = expand_metadata_aliases({"series": "My Series"}) + assert result["series"] == "My Series" + assert result["series_name"] == "My Series" + assert result["seriesname"] == "My Series" + assert result["series_title"] == "My Series" + assert result["seriestitle"] == "My Series" + + def test_series_name_variant(self): + result = expand_metadata_aliases({"series_name": "Test"}) + assert result["series"] == "Test" + assert result["series_name"] == "Test" + assert result["series_title"] == "Test" + + def test_series_index_expansion(self): + result = expand_metadata_aliases({"series_index": "3"}) + assert result["series_index"] == "3" + assert result["series_sequence"] == "3" + assert result["series_position"] == "3" + assert result["book_number"] == "3" + + def test_series_index_variant(self): + result = expand_metadata_aliases({"series_sequence": "5"}) + assert result["series_index"] == "5" + assert result["series_sequence"] == "5" + + def test_author_expansion(self): + result = expand_metadata_aliases({"author": "John"}) + assert result["author"] == "John" + assert result["authors"] == "John" + + def test_authors_variant(self): + result = expand_metadata_aliases({"authors": "Jane"}) + assert result["author"] == "Jane" + assert result["authors"] == "Jane" + + def test_description_expansion(self): + result = expand_metadata_aliases({"description": "A book"}) + assert result["description"] == "A book" + assert result["summary"] == "A book" + + def test_summary_variant(self): + result = expand_metadata_aliases({"summary": "Summary text"}) + assert result["description"] == "Summary text" + assert result["summary"] == "Summary text" + + def test_tags_expansion(self): + result = expand_metadata_aliases({"tags": "sci-fi"}) + assert result["tags"] == "sci-fi" + assert result["keywords"] == "sci-fi" + assert result["genre"] == "sci-fi" + + def test_keywords_variant(self): + result = expand_metadata_aliases({"keywords": "fantasy"}) + assert result["tags"] == "fantasy" + assert result["keywords"] == "fantasy" + assert result["genre"] == "fantasy" + + def test_passthrough_non_alias_keys(self): + result = expand_metadata_aliases({"title": "My Book", "publisher": "Pub"}) + assert result["title"] == "My Book" + assert result["publisher"] == "Pub" + + def test_mixed_concepts(self): + result = expand_metadata_aliases({ + "series": "My Series", + "series_index": "3", + "author": "John", + "title": "Book", + }) + assert result["series"] == "My Series" + assert result["series_name"] == "My Series" + assert result["series_index"] == "3" + assert result["series_sequence"] == "3" + assert result["author"] == "John" + assert result["authors"] == "John" + assert result["title"] == "Book" + + def test_skips_none_values(self): + result = expand_metadata_aliases({"series": None, "author": "John"}) + assert "series" not in result + assert result["author"] == "John" + + def test_skips_empty_values(self): + result = expand_metadata_aliases({"series": "", "author": "John"}) + assert "series" not in result + assert result["author"] == "John" + + def test_none_key_converted_to_string(self): + result = expand_metadata_aliases({None: "value"}) + assert result["none"] == "value" + + def test_case_insensitive_keys(self): + result = expand_metadata_aliases({"Series": "Test", "AUTHOR": "John"}) + assert result["series"] == "Test" + assert result["author"] == "John" + + def test_list_value_preserved(self): + result = expand_metadata_aliases({"tags": ["a", "b"]}) + assert result["tags"] == ["a", "b"] + assert result["keywords"] == ["a", "b"] + assert result["genre"] == ["a", "b"]