track_cleaning

Track Cleaning Service.

Handles metadata cleaning operations for tracks (removing promotional text, normalizing names, etc.).

TrackCleaningService

TrackCleaningService(
    track_processor, config, console_logger, error_logger
)

Service for cleaning track and album metadata.

Removes promotional text, normalizes names, and handles batch cleaning operations with proper change logging.

Initialize the cleaning service.

Parameters:

Name	Type	Description	Default
track_processor	TrackProcessor	Processor for updating tracks.	required
config	AppConfig	Typed application configuration.	required
console_logger	Logger	Logger for console output.	required
error_logger	Logger	Logger for error output.	required

Source code in src/app/track_cleaning.py

def __init__(
    self,
    track_processor: TrackProcessor,
    config: AppConfig,
    console_logger: logging.Logger,
    error_logger: logging.Logger,
) -> None:
    """Initialize the cleaning service.

    Args:
        track_processor: Processor for updating tracks.
        config: Typed application configuration.
        console_logger: Logger for console output.
        error_logger: Logger for error output.
    """
    self._track_processor = track_processor
    self._config = config
    self._console_logger = console_logger
    self._error_logger = error_logger

extract_and_clean_metadata

extract_and_clean_metadata(track)

Extract and clean track metadata.

Parameters:

Name	Type	Description	Default
track	TrackDict	Track data to process.	required

Returns:

Type	Description
tuple[TrackFieldValue, str, TrackFieldValue, TrackFieldValue, str, str]	Tuple of (track_id, artist_name, track_name, album_name, cleaned_track_name, cleaned_album_name).

Source code in src/app/track_cleaning.py

def extract_and_clean_metadata(self, track: TrackDict) -> tuple[TrackFieldValue, str, TrackFieldValue, TrackFieldValue, str, str]:
    """Extract and clean track metadata.

    Args:
        track: Track data to process.

    Returns:
        Tuple of (track_id, artist_name, track_name, album_name,
                 cleaned_track_name, cleaned_album_name).
    """
    artist_name = str(track.get("artist", ""))
    track_name = track.get("name", "")
    album_name = track.get("album", "")
    track_id = track.get("id", "")

    cleaned_track_name, cleaned_album_name = clean_names(
        artist=artist_name,
        track_name=str(track_name),
        album_name=str(album_name),
        config=self._config,
        console_logger=self._console_logger,
        error_logger=self._error_logger,
    )

    return track_id, artist_name, track_name, album_name, cleaned_track_name, cleaned_album_name

process_single_track async

process_single_track(track, artist_override=None)

Process a single track for cleaning.

This unified method handles both standalone cleaning and pipeline cleaning.

Parameters:

Name	Type	Description	Default
track	TrackDict	Track data to process.	required
artist_override	str | None	Optional artist name override for logging. If None, uses artist from track metadata.	None

Returns:

Type	Description
tuple[TrackDict | None, ChangeLogEntry | None]	Tuple of (updated_track, change_entry) or (None, None) if no update needed.

Source code in src/app/track_cleaning.py

async def process_single_track(
    self,
    track: TrackDict,
    artist_override: str | None = None,
) -> tuple[TrackDict | None, ChangeLogEntry | None]:
    """Process a single track for cleaning.

    This unified method handles both standalone cleaning and pipeline cleaning.

    Args:
        track: Track data to process.
        artist_override: Optional artist name override for logging.
                        If None, uses artist from track metadata.

    Returns:
        Tuple of (updated_track, change_entry) or (None, None) if no update needed.
    """
    # Extract and clean track metadata
    track_id, artist_name, track_name, album_name, cleaned_track_name, cleaned_album_name = self.extract_and_clean_metadata(track)

    if not track_id:
        return None, None

    # Normalize originals for comparison (collapse whitespace to match clean_names behavior)
    track_name_normalized = _normalize_whitespace(str(track_name)) if track_name else ""
    album_name_normalized = _normalize_whitespace(str(album_name)) if album_name else ""

    # Check if update needed (compare against normalized values)
    if cleaned_track_name == track_name_normalized and cleaned_album_name == album_name_normalized:
        return None, None

    # Update track
    success = await self._track_processor.update_track_async(
        track_id=str(track_id),
        new_track_name=(cleaned_track_name if cleaned_track_name != track_name_normalized else None),
        new_album_name=(cleaned_album_name if cleaned_album_name != album_name_normalized else None),
        track_status=track.track_status,
        original_artist=artist_name,
        original_album=str(album_name) if album_name is not None else None,
        original_track=str(track_name) if track_name is not None else None,
    )

    if not success:
        return None, None

    # Create updated track
    updated_track = track.copy()
    updated_track.name = cleaned_track_name
    updated_track.album = cleaned_album_name

    # Create change entry (use override if provided, otherwise use artist from track)
    change_entry = self._create_change_log_entry(
        track_id=str(track_id),
        artist=artist_override or artist_name,
        original_track_name=str(track_name) if track_name is not None else "",
        original_album_name=str(album_name) if album_name is not None else "",
        cleaned_track_name=cleaned_track_name,
        cleaned_album_name=cleaned_album_name,
    )

    return updated_track, change_entry

process_all_tracks async

process_all_tracks(tracks, artist)

Process multiple tracks for cleaning with change logging.

Parameters:

Name	Type	Description	Default
tracks	list[TrackDict]	List of tracks to process.	required
artist	str	Artist name for logging.	required

Returns:

Type	Description
tuple[list[TrackDict], list[ChangeLogEntry]]	Tuple of (updated_tracks, changes_log).

Source code in src/app/track_cleaning.py

async def process_all_tracks(
    self,
    tracks: list[TrackDict],
    artist: str,
) -> tuple[list[TrackDict], list[ChangeLogEntry]]:
    """Process multiple tracks for cleaning with change logging.

    Args:
        tracks: List of tracks to process.
        artist: Artist name for logging.

    Returns:
        Tuple of (updated_tracks, changes_log).
    """
    updated_tracks: list[TrackDict] = []
    changes_log: list[ChangeLogEntry] = []

    for track in tracks:
        updated_track, change_entry = await self.process_single_track(track, artist_override=artist)
        if updated_track:
            updated_tracks.append(updated_track)
        if change_entry:
            changes_log.append(change_entry)

    return updated_tracks, changes_log

clean_all_metadata_with_logs async

clean_all_metadata_with_logs(tracks)

Clean metadata for all tracks with change logging.

Parameters:

Name	Type	Description	Default
tracks	list[TrackDict]	List of tracks to clean.	required

Returns:

Type	Description
list[ChangeLogEntry]	List of change log entries for cleaned tracks.

Source code in src/app/track_cleaning.py

async def clean_all_metadata_with_logs(self, tracks: list[TrackDict]) -> list[ChangeLogEntry]:
    """Clean metadata for all tracks with change logging.

    Args:
        tracks: List of tracks to clean.

    Returns:
        List of change log entries for cleaned tracks.
    """
    changes_log: list[ChangeLogEntry] = []

    for track in tracks:
        _, change_entry = await self.process_single_track(track)
        if change_entry:
            changes_log.append(change_entry)

    if changes_log:
        self._console_logger.info("Cleaned metadata for %d tracks", len(changes_log))

    return changes_log