genre_update

Genre Update Service - standalone genre update operations.

GenreUpdateService

GenreUpdateService(
    track_processor,
    genre_manager,
    config,
    console_logger,
    error_logger,
    cleaning_service=None,
    artist_renamer=None,
)

Service for standalone genre update operations.

Initialize genre update service.

Parameters:

Name	Type	Description	Default
track_processor	TrackProcessor	Processor for fetching tracks.	required
genre_manager	GenreManager	Manager for genre updates.	required
config	AppConfig	Typed application configuration.	required
console_logger	Logger	Logger for console output.	required
error_logger	Logger	Logger for error output.	required
cleaning_service	TrackCleaningService | None	Optional service for metadata cleaning.	None
artist_renamer	ArtistRenamer | None	Optional service for artist renaming.	None

Source code in src/app/genre_update.py

def __init__(
    self,
    track_processor: TrackProcessor,
    genre_manager: GenreManager,
    config: AppConfig,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
    cleaning_service: TrackCleaningService | None = None,
    artist_renamer: ArtistRenamer | None = None,
) -> None:
    """Initialize genre update service.

    Args:
        track_processor: Processor for fetching tracks.
        genre_manager: Manager for genre updates.
        config: Typed application configuration.
        console_logger: Logger for console output.
        error_logger: Logger for error output.
        cleaning_service: Optional service for metadata cleaning.
        artist_renamer: Optional service for artist renaming.
    """
    self._track_processor = track_processor
    self._genre_manager = genre_manager
    self._config = config
    self._console_logger = console_logger
    self._error_logger = error_logger
    self._cleaning_service = cleaning_service
    self._artist_renamer = artist_renamer
    self._test_artists: set[str] | None = None

set_test_artists

set_test_artists(test_artists)

Set test artists for filtering.

Parameters:

Name	Type	Description	Default
test_artists	set[str] | None	Set of artist names to filter to, or None to process all.	required

Source code in src/app/genre_update.py

def set_test_artists(self, test_artists: set[str] | None) -> None:
    """Set test artists for filtering.

    Args:
        test_artists: Set of artist names to filter to, or None to process all.
    """
    self._test_artists = test_artists

get_tracks_for_genre_update async

get_tracks_for_genre_update(artist)

Get tracks for genre update based on artist filter.

Parameters:

Name	Type	Description	Default
artist	str | None	Optional artist filter.	required

Returns:

Type	Description
list[TrackDict] | None	List of tracks or None if not found.

Source code in src/app/genre_update.py

async def get_tracks_for_genre_update(self, artist: str | None) -> list[TrackDict] | None:
    """Get tracks for genre update based on artist filter.

    Args:
        artist: Optional artist filter.

    Returns:
        List of tracks or None if not found.
    """
    fetched_tracks: list[TrackDict] | None
    if artist is None:
        fetched_tracks = await self._track_processor.fetch_tracks_in_batches()
    else:
        fetched_tracks = await self._track_processor.fetch_tracks_async(artist=artist)

    # Filter by test_artists if in test mode
    if self._test_artists and fetched_tracks:
        fetched_tracks = [t for t in fetched_tracks if t.get("artist") in self._test_artists]
        self._console_logger.info(
            "Test mode: filtered to %d tracks for %d test artists",
            len(fetched_tracks),
            len(self._test_artists),
        )

    if not fetched_tracks:
        self._console_logger.warning(
            "No tracks found for genre update (artist=%s, test_mode=%s)",
            artist or "all",
            bool(self._test_artists),
        )
        return None
    return fetched_tracks

run_update_genres async

run_update_genres(artist, force)

Update genres for all or specific artist.

Parameters:

Name	Type	Description	Default
artist	str | None	Optional artist filter.	required
force	bool	Force update even if genre exists.	required

Source code in src/app/genre_update.py

async def run_update_genres(self, artist: str | None, force: bool) -> None:
    """Update genres for all or specific artist.

    Args:
        artist: Optional artist filter.
        force: Force update even if genre exists.
    """
    self._console_logger.info(
        "Starting genre update operation%s",
        f" for artist: {artist}" if artist else " for all artists",
    )

    tracks = await self.get_tracks_for_genre_update(artist)
    if not tracks:
        return

    # Preprocessing - clean metadata first
    if self._cleaning_service:
        self._console_logger.info("Preprocessing: Cleaning metadata...")
        await self._cleaning_service.clean_all_metadata_with_logs(tracks)

    # Preprocessing - rename artists
    if self._artist_renamer and self._artist_renamer.has_mapping:
        self._console_logger.info("Preprocessing: Renaming artists...")
        await self._artist_renamer.rename_tracks(tracks)

    # Update genres
    self._console_logger.info("Updating genres...")
    await self._genre_manager.update_genres_by_artist_async(tracks, force=force)

    self._console_logger.info("Genre update operation completed")