"music/genres/genres_for_media_item",
self.get_genres_for_media_item,
)
+ self.mass.register_api_command(
+ "music/genres/merge",
+ self.merge_genres,
+ required_role="admin",
+ )
# Run genre mapping scanner after library sync completes
self.mass.subscribe(self._on_sync_tasks_updated, EventType.SYNC_TASKS_UPDATED)
order_by: str = "sort_name",
provider: str | list[str] | None = None,
genre: int | list[int] | None = None,
+ hide_empty: bool = True,
**kwargs: Any,
) -> list[Genre]:
"""Get genres in the library.
:param genre: NOT SUPPORTED - Filtering genres by genres doesn't make sense.
+ :param hide_empty: If True (default), only return genres that have media mappings.
+ Set to False to return all genres including unmapped ones.
"""
if genre is not None:
msg = "genre parameter is not supported for Genre.library_items()"
# Pass raw lowered search for alias matching (search_raw),
# since the normalized :search param strips spaces/special chars.
extra_params: dict[str, Any] | None = None
+ extra_parts: list[str] | None = None
if search:
extra_params = {"search_raw": f"%{search.strip().lower()}%"}
+ if hide_empty:
+ gm = DB_TABLE_GENRE_MEDIA_ITEM_MAPPING
+ extra_parts = [
+ f"EXISTS(SELECT 1 FROM {gm} gm WHERE gm.genre_id = {self.db_table}.item_id)"
+ ]
return await self.get_library_items_by_query(
favorite=favorite,
search=search,
offset=offset,
order_by=order_by,
extra_query_params=extra_params,
+ extra_query_parts=extra_parts,
)
async def radio_mode_base_tracks(
return await self.get_library_item(new_genre_id)
+ async def merge_genres(self, genre_ids: list[str | int], target_genre_id: str | int) -> Genre:
+ """Merge one or more genres into a target genre.
+
+ Transfers all aliases and media mappings from the source genres to the
+ target, then deletes the source genres. Aliases and mappings are
+ deduplicated so no duplicates are created on the target.
+
+ :param genre_ids: List of genre IDs to merge into the target.
+ :param target_genre_id: Database ID of the genre to merge into.
+ """
+ target_id = int(target_genre_id)
+ source_ids = [int(gid) for gid in genre_ids]
+
+ if target_id in source_ids:
+ msg = "Target genre cannot be in the list of genres to merge"
+ raise ValueError(msg)
+ if not source_ids:
+ msg = "No genre IDs provided to merge"
+ raise ValueError(msg)
+
+ target_genre = await self.get_library_item(target_id)
+ db = self.mass.music.database
+ gm = DB_TABLE_GENRE_MEDIA_ITEM_MAPPING
+
+ # Collect and merge aliases from all source genres into the target
+ all_new_aliases: list[str] = []
+ for source_id in source_ids:
+ source_genre = await self.get_library_item(source_id)
+ if source_genre.genre_aliases:
+ all_new_aliases.extend(source_genre.genre_aliases)
+
+ existing_aliases = list(target_genre.genre_aliases) if target_genre.genre_aliases else []
+ merged_aliases = self._dedup_aliases(existing_aliases, all_new_aliases)
+ await db.update(
+ self.db_table,
+ {"item_id": target_id},
+ {"genre_aliases": serialize_to_json(merged_aliases)},
+ )
+
+ # Transfer media mappings from source genres to target (deduplicated)
+ placeholders = ", ".join(str(sid) for sid in source_ids)
+ await db.execute(
+ f"INSERT OR IGNORE INTO {gm} (genre_id, media_id, media_type, alias) "
+ f"SELECT :target_id, media_id, media_type, alias FROM {gm} "
+ f"WHERE genre_id IN ({placeholders})",
+ {"target_id": target_id},
+ )
+
+ # Delete source genres (remove_item_from_library cleans up remaining mappings)
+ for source_id in source_ids:
+ await self.remove_item_from_library(source_id)
+
+ updated = await self.get_library_item(target_id)
+ self.mass.signal_event(EventType.MEDIA_ITEM_UPDATED, updated.uri, updated)
+ return updated
+
async def sync_media_item_genres(
self, media_type: MediaType, media_id: str | int, genre_names: set[str]
) -> None:
"""Add 3 genres, returns all 3."""
for name in ("Alpha", "Beta", "Gamma"):
await genre_ctrl.add_item_to_library(_make_genre(name))
- items = await genre_ctrl.library_items()
+ items = await genre_ctrl.library_items(hide_empty=False)
names = {g.name for g in items}
assert {"Alpha", "Beta", "Gamma"}.issubset(names)
"""Search 'country' returns only matching genres."""
await genre_ctrl.add_item_to_library(_make_genre("Country"))
await genre_ctrl.add_item_to_library(_make_genre("Metal"))
- items = await genre_ctrl.library_items(search="country")
+ items = await genre_ctrl.library_items(search="country", hide_empty=False)
assert all("country" in g.name.lower() for g in items)
async def test_library_items_rejects_genre_param(self, genre_ctrl: GenreController) -> None:
assert "PromComplete" in updated_parent.genre_aliases
+# ===================================================================
+# Group F2: merge_genres (7 tests)
+# ===================================================================
+
+
+class TestMergeGenres:
+ """Tests for merge_genres."""
+
+ async def test_merge_transfers_aliases(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """Aliases from source genres are added to the target."""
+ target = await genre_ctrl.add_item_to_library(_make_genre("MergeTarget"))
+ source = await genre_ctrl.add_item_to_library(_make_genre("MergeSource"))
+ await genre_ctrl.add_alias(source.item_id, "SourceAlias")
+
+ result = await genre_ctrl.merge_genres([source.item_id], target.item_id)
+ assert result.genre_aliases is not None
+ assert "MergeTarget" in result.genre_aliases
+ assert "MergeSource" in result.genre_aliases
+ assert "SourceAlias" in result.genre_aliases
+
+ async def test_merge_transfers_media_mappings(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """Media mappings from source genres are moved to the target."""
+ target = await genre_ctrl.add_item_to_library(_make_genre("MergeMapTarget"))
+ source = await genre_ctrl.add_item_to_library(_make_genre("MergeMapSource"))
+ track = await _add_test_track(mass, "Merge Track")
+ await genre_ctrl.add_media_mapping(
+ source.item_id, MediaType.TRACK, track.item_id, "MergeMapSource"
+ )
+
+ await genre_ctrl.merge_genres([source.item_id], target.item_id)
+ rows = await mass.music.database.get_rows_from_query(
+ f"SELECT * FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} "
+ "WHERE genre_id = :gid AND media_type = 'track' AND media_id = :mid",
+ {"gid": int(target.item_id), "mid": int(track.item_id)},
+ limit=1,
+ )
+ assert len(rows) == 1
+
+ async def test_merge_deletes_source_genres(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """Source genres are deleted after merge."""
+ target = await genre_ctrl.add_item_to_library(_make_genre("MergeDelTarget"))
+ source = await genre_ctrl.add_item_to_library(_make_genre("MergeDelSource"))
+
+ await genre_ctrl.merge_genres([source.item_id], target.item_id)
+ with pytest.raises(MediaNotFoundError):
+ await genre_ctrl.get_library_item(int(source.item_id))
+
+ async def test_merge_deduplicates_aliases(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """Overlapping aliases are not duplicated on the target."""
+ target = await genre_ctrl.add_item_to_library(_make_genre("MergeDedupTarget"))
+ await genre_ctrl.add_alias(target.item_id, "SharedAlias")
+ source = await genre_ctrl.add_item_to_library(_make_genre("MergeDedupSource"))
+ await genre_ctrl.add_alias(source.item_id, "SharedAlias")
+
+ result = await genre_ctrl.merge_genres([source.item_id], target.item_id)
+ assert result.genre_aliases is not None
+ alias_list = list(result.genre_aliases)
+ norm_aliases = [a for a in alias_list if a.lower().replace(" ", "") == "sharedalias"]
+ assert len(norm_aliases) == 1
+
+ async def test_merge_deduplicates_media_mappings(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """Overlapping media mappings do not create duplicates."""
+ target = await genre_ctrl.add_item_to_library(_make_genre("MergeDedupMapTarget"))
+ source = await genre_ctrl.add_item_to_library(_make_genre("MergeDedupMapSource"))
+ track = await _add_test_track(mass, "Merge Dedup Track")
+ # Both genres map the same track
+ await genre_ctrl.add_media_mapping(
+ target.item_id, MediaType.TRACK, track.item_id, "MergeDedupMapTarget"
+ )
+ await genre_ctrl.add_media_mapping(
+ source.item_id, MediaType.TRACK, track.item_id, "MergeDedupMapSource"
+ )
+
+ await genre_ctrl.merge_genres([source.item_id], target.item_id)
+ rows = await mass.music.database.get_rows_from_query(
+ f"SELECT * FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} "
+ "WHERE genre_id = :gid AND media_type = 'track' AND media_id = :mid",
+ {"gid": int(target.item_id), "mid": int(track.item_id)},
+ limit=0,
+ )
+ assert len(rows) == 1
+
+ async def test_merge_multiple_sources(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """Multiple source genres can be merged at once."""
+ target = await genre_ctrl.add_item_to_library(_make_genre("MergeMultiTarget"))
+ source1 = await genre_ctrl.add_item_to_library(_make_genre("MergeMultiSrc1"))
+ source2 = await genre_ctrl.add_item_to_library(_make_genre("MergeMultiSrc2"))
+ track1 = await _add_test_track(mass, "Multi Merge Track 1")
+ track2 = await _add_test_track(mass, "Multi Merge Track 2")
+ await genre_ctrl.add_media_mapping(
+ source1.item_id, MediaType.TRACK, track1.item_id, "MergeMultiSrc1"
+ )
+ await genre_ctrl.add_media_mapping(
+ source2.item_id, MediaType.TRACK, track2.item_id, "MergeMultiSrc2"
+ )
+
+ result = await genre_ctrl.merge_genres([source1.item_id, source2.item_id], target.item_id)
+ assert result.genre_aliases is not None
+ assert "MergeMultiSrc1" in result.genre_aliases
+ assert "MergeMultiSrc2" in result.genre_aliases
+
+ # Both tracks mapped to target
+ rows = await mass.music.database.get_rows_from_query(
+ f"SELECT * FROM {DB_TABLE_GENRE_MEDIA_ITEM_MAPPING} "
+ "WHERE genre_id = :gid AND media_type = 'track'",
+ {"gid": int(target.item_id)},
+ limit=0,
+ )
+ assert len(rows) == 2
+
+ # Both sources deleted
+ for src in (source1, source2):
+ with pytest.raises(MediaNotFoundError):
+ await genre_ctrl.get_library_item(int(src.item_id))
+
+ async def test_merge_target_in_source_raises(self, genre_ctrl: GenreController) -> None:
+ """Raises ValueError when target is in the source list."""
+ genre = await genre_ctrl.add_item_to_library(_make_genre("MergeSelfTarget"))
+ with pytest.raises(ValueError, match="Target genre cannot be in the list"):
+ await genre_ctrl.merge_genres([genre.item_id], genre.item_id)
+
+ async def test_merge_empty_source_raises(self, genre_ctrl: GenreController) -> None:
+ """Raises ValueError when source list is empty."""
+ target = await genre_ctrl.add_item_to_library(_make_genre("MergeEmptyTarget"))
+ with pytest.raises(ValueError, match="No genre IDs provided"):
+ await genre_ctrl.merge_genres([], target.item_id)
+
+
# ===================================================================
# Group G: restore_default_genres (5 tests)
# ===================================================================
"""Full restore: custom genres gone, only defaults remain."""
await genre_ctrl.add_item_to_library(_make_genre("MyCustomGenre"))
await genre_ctrl.restore_default_genres(full_restore=True)
- items = await genre_ctrl.library_items(limit=0)
+ items = await genre_ctrl.library_items(limit=0, hide_empty=False)
names = {g.name for g in items}
assert "MyCustomGenre" not in names
assert len(items) == len(DEFAULT_GENRE_MAPPING)
if not entries_with_aliases:
pytest.skip("No default genres with aliases configured")
entry = entries_with_aliases[0]
- items = await genre_ctrl.library_items(search=entry["genre"])
+ items = await genre_ctrl.library_items(search=entry["genre"], hide_empty=False)
assert len(items) > 0
genre = items[0]
assert genre.genre_aliases is not None
genres = await genre_ctrl.get_genres_for_media_item(MediaType.TRACK, track.item_id)
assert genres == []
+ async def test_library_items_hide_empty_true(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """hide_empty=True returns only genres with mappings."""
+ mapped = await genre_ctrl.add_item_to_library(_make_genre("HasMappingGenre"))
+ unmapped = await genre_ctrl.add_item_to_library(_make_genre("NoMappingGenre"))
+ track = await _add_test_track(mass, "HasMapping Track")
+ await genre_ctrl.add_media_mapping(
+ mapped.item_id, MediaType.TRACK, track.item_id, "HasMappingGenre"
+ )
+ results = await genre_ctrl.library_items(hide_empty=True)
+ result_ids = {int(g.item_id) for g in results}
+ assert int(mapped.item_id) in result_ids
+ assert int(unmapped.item_id) not in result_ids
+
+ async def test_library_items_hide_empty_default(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """Default (hide_empty=True) excludes unmapped genres."""
+ mapped = await genre_ctrl.add_item_to_library(_make_genre("DefaultFilterMapped"))
+ unmapped = await genre_ctrl.add_item_to_library(_make_genre("DefaultFilterUnmapped"))
+ track = await _add_test_track(mass, "DefaultFilter Track")
+ await genre_ctrl.add_media_mapping(
+ mapped.item_id, MediaType.TRACK, track.item_id, "DefaultFilterMapped"
+ )
+ results = await genre_ctrl.library_items()
+ result_ids = {int(g.item_id) for g in results}
+ assert int(mapped.item_id) in result_ids
+ assert int(unmapped.item_id) not in result_ids
+
+ async def test_library_items_show_all(
+ self, mass: MusicAssistant, genre_ctrl: GenreController
+ ) -> None:
+ """hide_empty=False returns all genres including unmapped."""
+ mapped = await genre_ctrl.add_item_to_library(_make_genre("ShowAllMapped"))
+ unmapped = await genre_ctrl.add_item_to_library(_make_genre("ShowAllUnmapped"))
+ track = await _add_test_track(mass, "ShowAll Track")
+ await genre_ctrl.add_media_mapping(
+ mapped.item_id, MediaType.TRACK, track.item_id, "ShowAllMapped"
+ )
+ results = await genre_ctrl.library_items(hide_empty=False)
+ result_ids = {int(g.item_id) for g in results}
+ assert int(mapped.item_id) in result_ids
+ assert int(unmapped.item_id) in result_ids
+
# ===================================================================
# Group I: Genre Lookup & Scanner (5 tests)
genre = await genre_ctrl.add_item_to_library(_make_genre("InlineTest"))
await genre_ctrl.add_alias(genre.item_id, "Inline Alias")
# Fetch via library_items (uses base_query)
- items = await genre_ctrl.library_items(search="InlineTest")
+ items = await genre_ctrl.library_items(search="InlineTest", hide_empty=False)
assert len(items) >= 1
fetched = items[0]
assert fetched.genre_aliases is not None
"""limit/offset work correctly."""
for i in range(5):
await genre_ctrl.add_item_to_library(_make_genre(f"Page{i}"))
- page1 = await genre_ctrl.library_items(limit=2, offset=0, order_by="name")
- page2 = await genre_ctrl.library_items(limit=2, offset=2, order_by="name")
+ page1 = await genre_ctrl.library_items(limit=2, offset=0, order_by="name", hide_empty=False)
+ page2 = await genre_ctrl.library_items(limit=2, offset=2, order_by="name", hide_empty=False)
assert len(page1) == 2
assert len(page2) == 2
ids1 = {g.item_id for g in page1}
"""favorite=True filters correctly."""
await genre_ctrl.add_item_to_library(_make_genre("FavYes", favorite=True))
await genre_ctrl.add_item_to_library(_make_genre("FavNo", favorite=False))
- favs = await genre_ctrl.library_items(favorite=True)
+ favs = await genre_ctrl.library_items(favorite=True, hide_empty=False)
assert all(g.favorite for g in favs)
assert any(g.name == "FavYes" for g in favs)
projects / music-assistant-server.git / commitdiff