tessl/pypi-beets

A comprehensive music library management system and command-line application for organizing and maintaining digital music collections

—

Pending

Quality

Pending

Does it follow best practices?

Impact

Pending

No eval scenarios have been run

Overview

Eval results

Files

Import and Autotag System

Name: tessl/pypi-beets
Author: tessl

Automated music import with metadata matching, user interaction for ambiguous matches, and integration with multiple metadata sources for comprehensive tagging. The import system is the primary way music gets into a beets library with correct metadata.

Capabilities

Import Session Management

Central class that controls the import process, handles user interaction, and manages import decisions.

class ImportSession:
    def __init__(self, lib: Library, loghandler: logging.Handler = None, config: dict = None):
        """
        Initialize an import session.
        
        Parameters:
        - lib: Library instance to import into
        - loghandler: Optional logging handler for import messages
        - config: Optional configuration overrides
        """
    
    def run(self) -> None:
        """
        Execute the import process for all configured paths.
        Processes files, matches metadata, and handles user interaction.
        """
    
    def choose_match(self, task: ImportTask) -> action:
        """
        Handle match selection for an import task.
        
        Parameters:
        - task: ImportTask with potential matches
        
        Returns:
        Action constant (APPLY, ASIS, SKIP, etc.)
        """
    
    def choose_item(self, task: ImportTask, item: Item) -> action:
        """
        Handle individual item decisions during import.
        
        Parameters:
        - task: ImportTask containing the item
        - item: Specific Item being processed
        
        Returns:
        Action constant for this item
        """
    
    def resolve_duplicate(self, task: ImportTask, found_duplicates: List[Item]) -> action:
        """
        Handle duplicate resolution during import.
        
        Parameters:
        - task: ImportTask that found duplicates
        - found_duplicates: List of existing items that match
        
        Returns:
        Action to take for the duplicates
        """

Import Task Types

Different task types for handling various import scenarios.

class ImportTask:
    """Base class for import tasks representing album imports."""
    
    def __init__(self, toppath: str, paths: List[str], items: List[Item]):
        """
        Initialize import task.
        
        Parameters:
        - toppath: Top-level directory being imported
        - paths: List of file paths in the album
        - items: List of Item objects created from files
        """
    
    def lookup_candidates(self) -> List[AlbumInfo]:
        """
        Look up metadata candidates for this album.
        
        Returns:
        List of AlbumInfo objects with potential matches
        """
    
    def match_candidates(self, candidates: List[AlbumInfo]) -> List[Tuple[AlbumInfo, Distance]]:
        """
        Score metadata candidates against the actual items.
        
        Parameters:
        - candidates: List of AlbumInfo objects to score
        
        Returns:
        List of (AlbumInfo, Distance) tuples sorted by match quality
        """

class SingletonImportTask(ImportTask):
    """Import task for individual tracks (not part of an album)."""
    
    def lookup_candidates(self) -> List[TrackInfo]:
        """
        Look up metadata candidates for this single track.
        
        Returns:
        List of TrackInfo objects with potential matches
        """

class ArchiveImportTask(ImportTask):
    """Import task for compressed archives (ZIP, RAR, etc.)."""
    
    def extract(self, dest: str) -> str:
        """
        Extract archive contents to destination directory.
        
        Parameters:
        - dest: Destination directory for extraction
        
        Returns:
        Path to extracted contents
        """

Import Actions

Constants defining possible actions during import process.

# Import action constants
SKIP = 'skip'           # Skip importing this item/album
ASIS = 'asis'           # Import as-is without metadata changes  
TRACKS = 'tracks'       # Import as individual tracks, not album
APPLY = 'apply'         # Apply the selected metadata match
ALBUMS = 'albums'       # Group items into albums during import
RETAG = 'retag'         # Re-tag existing items with new metadata

Core Autotag Functions

High-level functions for automated metadata tagging.

def tag_album(items: List[Item], search_artist: str = None, search_album: str = None) -> Tuple[List[AlbumInfo], List[TrackInfo]]:
    """
    Automatically tag an album by searching metadata sources.
    
    Parameters:
    - items: List of Item objects representing album tracks
    - search_artist: Optional artist hint for search
    - search_album: Optional album title hint for search
    
    Returns:
    Tuple of (album_matches, track_matches) with scoring information
    """

def tag_item(item: Item, search_artist: str = None, search_title: str = None) -> Tuple[List[TrackInfo], Distance]:
    """
    Automatically tag a single item by searching metadata sources.
    
    Parameters:
    - item: Item object to tag
    - search_artist: Optional artist hint for search
    - search_title: Optional title hint for search
    
    Returns:
    Tuple of (track_matches, best_distance) with match information
    """

def apply_metadata(info: Union[AlbumInfo, TrackInfo], items: List[Item]) -> None:
    """
    Apply metadata from AlbumInfo or TrackInfo to items.
    
    Parameters:
    - info: AlbumInfo or TrackInfo object with metadata
    - items: List of Item objects to update
    """

Metadata Information Classes

Classes representing metadata retrieved from various sources.

class AlbumInfo:
    """Album metadata information from external sources."""
    
    # Core album fields
    album: str              # Album title
    artist: str             # Album artist
    albumtype: str          # Album type (album, single, EP, etc.)
    albumtypes: List[str]   # List of album types
    va: bool                # Various artists album
    year: int               # Release year
    month: int              # Release month
    day: int                # Release day
    country: str            # Release country
    label: str              # Record label
    catalognum: str         # Catalog number
    asin: str               # Amazon ASIN
    albumdisambig: str      # Disambiguation string
    releasegroupid: str     # Release group ID
    albumid: str            # Album ID from source
    albumstatus: str        # Release status
    media: str              # Media format
    albumdisambig: str      # Album disambiguation
    
    # Track information
    tracks: List[TrackInfo] # List of track metadata
    
    def __init__(self, **kwargs):
        """Initialize with metadata fields as keyword arguments."""

class TrackInfo:
    """Track metadata information from external sources."""
    
    # Core track fields
    title: str              # Track title
    artist: str             # Track artist
    length: float           # Duration in seconds
    track: int              # Track number
    disc: int               # Disc number
    medium: int             # Medium number
    medium_index: int       # Index within medium
    medium_total: int       # Total tracks on medium
    artist_sort: str        # Artist sort name
    disctitle: str          # Disc title
    trackid: str            # Track ID from source
    artistid: str           # Artist ID from source
    
    def __init__(self, **kwargs):
        """Initialize with metadata fields as keyword arguments."""

class Distance:
    """Represents similarity measurement between items and metadata."""
    
    def __init__(self):
        self.album_distance: float = 0.0    # Album-level distance
        self.tracks: List[float] = []       # Per-track distances  
        self.unmatched_tracks: int = 0      # Number of unmatched tracks
        self.extra_tracks: int = 0          # Number of extra tracks
        
    @property 
    def distance(self) -> float:
        """Overall distance score (lower is better match)."""
    
    @property
    def max_distance(self) -> float:
        """Maximum possible distance for normalization."""

Match Classes

Classes representing matching results with scoring information.

class AlbumMatch:
    """Represents a potential album match with distance scoring."""
    
    def __init__(self, info: AlbumInfo, distance: Distance, items: List[Item]):
        """
        Initialize album match.
        
        Parameters:
        - info: AlbumInfo object with metadata
        - distance: Distance object with match scoring
        - items: List of Item objects being matched
        """
        
    @property
    def goodness(self) -> float:
        """Match quality score (higher is better)."""

class TrackMatch:
    """Represents a potential track match with distance scoring."""
    
    def __init__(self, info: TrackInfo, distance: float, item: Item):
        """
        Initialize track match.
        
        Parameters:
        - info: TrackInfo object with metadata
        - distance: Distance score for this match
        - item: Item object being matched
        """

# Match quality levels
class Recommendation:
    """Enumeration of match recommendation levels."""
    STRONG = 'strong'       # High confidence match
    MEDIUM = 'medium'       # Moderate confidence match  
    LOW = 'low'             # Low confidence match
    NONE = 'none'           # No good matches found

Import Workflow Examples

Basic Import Session

from beets.library import Library
from beets.importer import ImportSession

# Create library and import session
lib = Library('/path/to/library.db', '/music')
session = ImportSession(lib)

# Configure paths to import
session.paths = ['/path/to/new/music']

# Run import process (interactive)
session.run()

Programmatic Import Control

from beets.importer import ImportSession, action
from beets.library import Library

class AutoImportSession(ImportSession):
    """Custom import session with automatic decisions."""
    
    def choose_match(self, task):
        """Always apply the best match if confidence is high."""
        if task.rec == task.rec.STRONG:
            return action.APPLY
        else:
            return action.SKIP
    
    def choose_item(self, task, item):
        """Auto-apply strong track matches.""" 
        if task.item_match.goodness > 0.8:
            return action.APPLY
        else:
            return action.ASIS

# Use custom session
lib = Library('/path/to/library.db', '/music')
session = AutoImportSession(lib)
session.paths = ['/path/to/new/music']
session.run()

Manual Tagging

from beets.autotag import tag_album, tag_item, apply_metadata
from beets.library import Item

# Tag an album manually
items = [Item.from_path(p) for p in album_paths]
album_matches, track_matches = tag_album(items)

if album_matches:
    best_match = album_matches[0]  # Highest scoring match
    apply_metadata(best_match.info, items)
    
    # Add to library
    for item in items:
        lib.add(item)

# Tag individual track
item = Item.from_path('/path/to/track.mp3')
track_matches, distance = tag_item(item)

if track_matches and distance.distance < 0.1:  # Good match
    apply_metadata(track_matches[0], [item])
    lib.add(item)

Custom Metadata Sources

from beets.plugins import MetadataSourcePlugin
from beets.autotag.hooks import AlbumInfo, TrackInfo

class CustomMetadataSource(MetadataSourcePlugin):
    """Example custom metadata source plugin."""
    
    def get_albums(self, query: str) -> Iterator[AlbumInfo]:
        """Search for albums from custom source."""
        # Implement custom album search
        results = self.search_custom_api(query)
        
        for result in results:
            yield AlbumInfo(
                album=result['title'],
                artist=result['artist'], 
                year=result['year'],
                tracks=[
                    TrackInfo(
                        title=track['title'],
                        artist=track['artist'],
                        track=track['number']
                    ) for track in result['tracks']
                ]
            )
    
    def get_tracks(self, query: str) -> Iterator[TrackInfo]:
        """Search for individual tracks."""
        # Implement custom track search
        pass

Configuration Options

Import Configuration

# Access import configuration
from beets import config

# Common import settings
write_tags = config['import']['write'].get(bool)          # Write tags to files
copy_files = config['import']['copy'].get(bool)          # Copy vs move files
resume = config['import']['resume'].get(bool)            # Resume interrupted imports
incremental = config['import']['incremental'].get(bool)  # Skip unchanged directories
quiet_fallback = config['import']['quiet_fallback'].get(bool)  # Auto-accept on timeout

# Metadata source configuration  
sources = config['import']['sources'].get(list)          # Enabled metadata sources
search_ids = config['import']['search_ids'].get(list)    # IDs to search for

Path Configuration

# Path format configuration
path_formats = config['paths'].get(dict)

# Example path formats
default_format = config['paths']['default'].get(str)     # Default path template
album_format = config['paths']['albumtype:album'].get(str)  # Album-specific format
comp_format = config['paths']['comp'].get(str)           # Compilation format

Error Handling

class ImportAbortError(Exception):
    """Raised when import process is aborted by user."""

class ImportDuplicateAlbumError(Exception):
    """Raised when attempting to import duplicate album."""

Common import issues: