gronod/headphones
1
0
Fork 1
mirror of https://github.com/rembo10/headphones.git synced 2026-05-15 16:19:28 +01:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'importer'

Browse Source
This commit is contained in:
rembo10
2012-03-23 16:44:58 +00:00
parent ab8744e30d d645b02d4f
commit aafe49b9c3
19 changed files with 1908 additions and 596 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
headphones/postprocessor.py
View File

@@ -408,7 +408,10 @@ def correctMetadata(albumid, release, downloaded_track_list):
autotag.apply_metadata(items, info)
for item in items:
item.write()
try:
item.write()
except Exception, e:
logger.warn('Error writing metadata to track: %s' % e)
def embedLyrics(downloaded_track_list):
logger.info('Adding lyrics')

8
lib/beets/__init__.py
View File

@@ -1,5 +1,5 @@
# This file is part of beets.
# Copyright 2011, Adrian Sampson.
# Copyright 2012, Adrian Sampson.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
@@ -12,7 +12,11 @@
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
__version__ = '1.0b11'
#
# MODIFIED TO WORK WITH HEADPHONES!!
#
__version__ = '1.0b14'
__author__ = 'Adrian Sampson <adrian@radbox.org>'
from lib.beets.library import Library

65
lib/beets/autotag/__init__.py
View File

@@ -16,9 +16,10 @@
"""
import os
import logging
import re
from lib.beets import library, mediafile
from lib.beets.util import sorted_walk
from lib.beets.util import sorted_walk, ancestry
# Parts of external interface.
from .hooks import AlbumInfo, TrackInfo
@@ -30,16 +31,24 @@ from .match import STRONG_REC_THRESH, MEDIUM_REC_THRESH, REC_GAP_THRESH
# Global logger.
log = logging.getLogger('beets')
# Constants for directory walker.
MULTIDISC_MARKERS = (r'part', r'volume', r'vol\.', r'disc', r'cd')
MULTIDISC_PAT_FMT = r'%s\s*\d'
# Additional utilities for the main interface.
def albums_in_dir(path):
def albums_in_dir(path, ignore=()):
"""Recursively searches the given directory and returns an iterable
of (path, items) where path is a containing directory and items is
a list of Items that is probably an album. Specifically, any folder
containing any media files is an album.
containing any media files is an album. Directories and file names
that match the glob patterns in ``ignore`` are skipped.
"""
for root, dirs, files in sorted_walk(path):
collapse_root = None
collapse_items = None
for root, dirs, files in sorted_walk(path, ignore):
# Get a list of items in the directory.
items = []
for filename in files:
@@ -51,11 +60,48 @@ def albums_in_dir(path):
log.warn('unreadable file: ' + filename)
else:
items.append(i)
# If we're collapsing, test to see whether we should continue to
# collapse. If so, just add to the collapsed item set;
# otherwise, end the collapse and continue as normal.
if collapse_root is not None:
if collapse_root in ancestry(root):
# Still collapsing.
collapse_items += items
continue
else:
# Collapse finished. Yield the collapsed directory and
# proceed to process the current one.
if collapse_items:
yield collapse_root, collapse_items
collapse_root = collapse_items = None
# Does the current directory look like a multi-disc album? If
# so, begin collapsing here.
if dirs and not items: # Must be only directories.
multidisc = False
for marker in MULTIDISC_MARKERS:
pat = MULTIDISC_PAT_FMT % marker
if all(re.search(pat, dirname, re.I) for dirname in dirs):
multidisc = True
break
# This becomes True only when all directories match a
# pattern for a single marker.
if multidisc:
# Start collapsing; continue to the next iteration.
collapse_root = root
collapse_items = []
continue
# If it's nonempty, yield it.
if items:
yield root, items
# Clear out any unfinished collapse.
if collapse_root is not None and collapse_items:
yield collapse_root, collapse_items
def apply_item_metadata(item, track_info):
"""Set an item's metadata from its matched TrackInfo object.
"""
@@ -73,6 +119,8 @@ def apply_metadata(items, album_info):
"""
for index, (item, track_info) in enumerate(zip(items, album_info.tracks)):
# Album, artist, track count.
if not item:
continue
if track_info.artist:
item.artist = track_info.artist
else:
@@ -92,6 +140,10 @@ def apply_metadata(items, album_info):
# Title and track index.
item.title = track_info.title
item.track = index + 1
# Disc and disc count.
item.disc = track_info.medium
item.disctotal = album_info.mediums
# MusicBrainz IDs.
item.mb_trackid = track_info.track_id
@@ -107,5 +159,6 @@ def apply_metadata(items, album_info):
# Compilation flag.
item.comp = album_info.va
# Uncomment to get rid of comments tag
item.comments = 'tagged by headphones/beets'
# Headphones seal of approval
item.comments = 'tagged by headphones/beets'

63
lib/beets/autotag/art.py
View File

@@ -18,6 +18,7 @@ import urllib
import sys
import logging
import os
import re
from lib.beets.autotag.mb import album_for_id
@@ -28,29 +29,64 @@ COVER_NAMES = ['cover', 'front', 'art', 'album', 'folder']
log = logging.getLogger('beets')
CONTENT_TYPES = ('image/jpeg',)
def _fetch_image(url):
"""Downloads an image from a URL and checks whether it seems to
actually be an image. If so, returns a path to the downloaded image.
Otherwise, returns None.
"""
log.debug('Downloading art: %s' % url)
try:
fn, headers = urllib.urlretrieve(url)
except IOError:
log.debug('error fetching art')
return
# Make sure it's actually an image.
if headers.gettype() in CONTENT_TYPES:
log.debug('Downloaded art to: %s' % fn)
return fn
else:
log.debug('Not an image.')
# Art from Amazon.
AMAZON_URL = 'http://images.amazon.com/images/P/%s.%02i.LZZZZZZZ.jpg'
AMAZON_INDICES = (1,2)
AMAZON_CONTENT_TYPE = 'image/jpeg'
def art_for_asin(asin):
"""Fetches art for an Amazon ID (ASIN) string."""
for index in AMAZON_INDICES:
# Fetch the image.
url = AMAZON_URL % (asin, index)
try:
log.debug('Downloading art: %s' % url)
fn, headers = urllib.urlretrieve(url)
except IOError:
log.debug('error fetching art at URL %s' % url)
continue
# Make sure it's actually an image.
if headers.gettype() == AMAZON_CONTENT_TYPE:
log.debug('Downloaded art to: %s' % fn)
fn = _fetch_image(url)
if fn:
return fn
# AlbumArt.org scraper.
AAO_URL = 'http://www.albumart.org/index_detail.php'
AAO_PAT = r'href\s*=\s*"([^>"]*)"[^>]*title\s*=\s*"View larger image"'
def aao_art(asin):
# Get the page from albumart.org.
url = '%s?%s' % (AAO_URL, urllib.urlencode({'asin': asin}))
try:
log.debug('Scraping art URL: %s' % url)
page = urllib.urlopen(url).read()
except IOError:
log.debug('Error scraping art page')
return
# Search the page for the image URL.
m = re.search(AAO_PAT, page)
if m:
image_url = m.group(1)
return _fetch_image(image_url)
else:
log.debug('No image found on page')
# Art from the filesystem.
def art_in_path(path):
@@ -91,7 +127,10 @@ def art_for_album(album, path):
if album.asin:
log.debug('Fetching album art for ASIN %s.' % album.asin)
return art_for_asin(album.asin)
out = art_for_asin(album.asin)
if out:
return out
return aao_art(album.asin)
else:
log.debug('No ASIN available: no art found.')
return None

13
lib/beets/autotag/hooks.py
View File

@@ -35,13 +35,14 @@ class AlbumInfo(object):
- ``month``: release month
- ``day``: release day
- ``label``: music label responsible for the release
- ``mediums``: the number of discs in this release
The fields up through ``tracks`` are required. The others are
optional and may be None.
"""
def __init__(self, album, album_id, artist, artist_id, tracks, asin=None,
albumtype=None, va=False, year=None, month=None, day=None,
label=None):
label=None, mediums=None):
self.album = album
self.album_id = album_id
self.artist = artist
@@ -54,6 +55,7 @@ class AlbumInfo(object):
self.month = month
self.day = day
self.label = label
self.mediums = mediums
class TrackInfo(object):
"""Describes a canonical track present on a release. Appears as part
@@ -64,17 +66,21 @@ class TrackInfo(object):
- ``artist``: individual track artist name
- ``artist_id``
- ``length``: float: duration of the track in seconds
- ``medium``: the disc number this track appears on in the album
- ``medium_index``: the track's position on the disc
Only ``title`` and ``track_id`` are required. The rest of the fields
may be None.
"""
def __init__(self, title, track_id, artist=None, artist_id=None,
length=None):
length=None, medium=None, medium_index=None):
self.title = title
self.track_id = track_id
self.artist = artist
self.artist_id = artist_id
self.length = length
self.medium = medium
self.medium_index = medium_index
# Aggregation of sources.
@@ -117,7 +123,8 @@ def _item_candidates(item, artist, title):
out = []
# MusicBrainz candidates.
out.extend(mb.match_track(artist, title))
if artist and title:
out.extend(mb.match_track(artist, title))
# Plugin candidates.
out.extend(plugins.item_candidates(item))

108
lib/beets/autotag/match.py
View File

@@ -31,6 +31,8 @@ ARTIST_WEIGHT = 3.0
ALBUM_WEIGHT = 3.0
# The weight of the entire distance calculated for a given track.
TRACK_WEIGHT = 1.0
# The weight of a missing track.
MISSING_WEIGHT = 0.9
# These distances are components of the track distance (that is, they
# compete against each other but not ARTIST_WEIGHT and ALBUM_WEIGHT;
# the overall TRACK_WEIGHT does that).
@@ -73,7 +75,10 @@ STRONG_REC_THRESH = 0.04
MEDIUM_REC_THRESH = 0.25
REC_GAP_THRESH = 0.25
# Artist signals that indicate "various artists".
# Artist signals that indicate "various artists". These are used at the
# album level to determine whether a given release is likely a VA
# release and also on the track level to to remove the penalty for
# differing artists.
VA_ARTISTS = (u'', u'various artists', u'va', u'unknown')
# Autotagging exceptions.
@@ -161,18 +166,22 @@ def current_metadata(items):
likelies = {}
consensus = {}
for key in keys:
values = [getattr(item, key) for item in items]
values = [getattr(item, key) for item in items if item]
likelies[key], freq = plurality(values)
consensus[key] = (freq == len(values))
return likelies['artist'], likelies['album'], consensus['artist']
def order_items(items, trackinfo):
"""Orders the items based on how they match some canonical track
information. This always produces a result if the numbers of tracks
match.
information. Returns a list of Items whose length is equal to the
length of ``trackinfo``. This always produces a result if the
numbers of items is at most the number of TrackInfo objects
(otherwise, returns None). In the case of a partial match, the
returned list may contain None in some positions.
"""
# Make sure lengths match.
if len(items) != len(trackinfo):
# Make sure lengths match: If there is less items, it might just be that
# there is some tracks missing.
if len(items) > len(trackinfo):
return None
# Construct the cost matrix.
@@ -187,7 +196,7 @@ def order_items(items, trackinfo):
matching = Munkres().compute(costs)
# Order items based on the matching.
ordered_items = [None]*len(items)
ordered_items = [None]*len(trackinfo)
for cur_idx, canon_idx in matching:
ordered_items[canon_idx] = items[cur_idx]
return ordered_items
@@ -205,10 +214,8 @@ def track_distance(item, track_info, track_index=None, incl_artist=False):
dist, dist_max = 0.0, 0.0
# Check track length.
if not track_info.length:
# If there's no length to check, assume the worst.
dist += TRACK_LENGTH_WEIGHT
else:
# If there's no length to check, apply no penalty.
if track_info.length:
diff = abs(item.length - track_info.length)
diff = max(diff - TRACK_LENGTH_GRACE, 0.0)
diff = min(diff, TRACK_LENGTH_MAX)
@@ -223,14 +230,15 @@ def track_distance(item, track_info, track_index=None, incl_artist=False):
# Attention: MB DB does not have artist info for all compilations,
# so only check artist distance if there is actually an artist in
# the MB track data.
if incl_artist and track_info.artist:
if incl_artist and track_info.artist and \
item.artist.lower() not in VA_ARTISTS:
dist += string_dist(item.artist, track_info.artist) * \
TRACK_ARTIST_WEIGHT
dist_max += TRACK_ARTIST_WEIGHT
# Track index.
if track_index and item.track:
if track_index != item.track:
if item.track not in (track_index, track_info.medium_index):
dist += TRACK_INDEX_WEIGHT
dist_max += TRACK_INDEX_WEIGHT
@@ -269,9 +277,13 @@ def distance(items, album_info):
# Track distances.
for i, (item, track_info) in enumerate(zip(items, album_info.tracks)):
dist += track_distance(item, track_info, i+1, album_info.va) * \
TRACK_WEIGHT
dist_max += TRACK_WEIGHT
if item:
dist += track_distance(item, track_info, i+1, album_info.va) * \
TRACK_WEIGHT
dist_max += TRACK_WEIGHT
else:
dist += MISSING_WEIGHT
dist_max += MISSING_WEIGHT
# Plugin distances.
plugin_d, plugin_dm = plugins.album_distance(items, album_info)
@@ -335,8 +347,8 @@ def recommendation(results):
return rec
def validate_candidate(items, tuple_dict, info):
"""Given a candidate info dict, attempt to add the candidate to
the output dictionary of result tuples. This involves checking
"""Given a candidate AlbumInfo object, attempt to add the candidate
to the output dictionary of result tuples. This involves checking
the track count, ordering the items, checking for duplicates, and
calculating the distance.
"""
@@ -348,8 +360,9 @@ def validate_candidate(items, tuple_dict, info):
return
# Make sure the album has the correct number of tracks.
if len(items) != len(info.tracks):
log.debug('Track count mismatch.')
if len(items) > len(info.tracks):
log.debug('Too many items to match: %i > %i.' %
(len(items), len(info.tracks)))
return
# Put items in order.
@@ -386,8 +399,9 @@ def tag_album(items, timid=False, search_artist=None, search_album=None,
cur_artist, cur_album, artist_consensus = current_metadata(items)
log.debug('Tagging %s - %s' % (cur_artist, cur_album))
# The output result tuples (keyed by MB album ID).
out_tuples = {}
# The output result (distance, AlbumInfo) tuples (keyed by MB album
# ID).
candidates = {}
# Try to find album indicated by MusicBrainz IDs.
if search_id:
@@ -396,21 +410,21 @@ def tag_album(items, timid=False, search_artist=None, search_album=None,
else:
id_info = match_by_id(items)
if id_info:
validate_candidate(items, out_tuples, id_info)
rec = recommendation(out_tuples.values())
validate_candidate(items, candidates, id_info)
rec = recommendation(candidates.values())
log.debug('Album ID match recommendation is ' + str(rec))
if out_tuples and not timid:
if candidates and not timid:
# If we have a very good MBID match, return immediately.
# Otherwise, this match will compete against metadata-based
# matches.
if rec == RECOMMEND_STRONG:
log.debug('ID match.')
return cur_artist, cur_album, out_tuples.values(), rec
return cur_artist, cur_album, candidates.values(), rec
# If searching by ID, don't continue to metadata search.
if search_id is not None:
if out_tuples:
return cur_artist, cur_album, out_tuples.values(), rec
if candidates:
return cur_artist, cur_album, candidates.values(), rec
else:
return cur_artist, cur_album, [], RECOMMEND_NONE
@@ -427,20 +441,16 @@ def tag_album(items, timid=False, search_artist=None, search_album=None,
log.debug(u'Album might be VA: %s' % str(va_likely))
# Get the results from the data sources.
candidates = hooks._album_candidates(items, search_artist, search_album,
va_likely)
search_cands = hooks._album_candidates(items, search_artist, search_album,
va_likely)
log.debug(u'Evaluating %i candidates.' % len(search_cands))
for info in search_cands:
validate_candidate(items, candidates, info)
# Get the distance to each candidate.
log.debug(u'Evaluating %i candidates.' % len(candidates))
for info in candidates:
validate_candidate(items, out_tuples, info)
# Sort by distance.
out_tuples = out_tuples.values()
out_tuples.sort()
rec = recommendation(out_tuples)
return cur_artist, cur_album, out_tuples, rec
# Sort and get the recommendation.
candidates = sorted(candidates.itervalues())
rec = recommendation(candidates)
return cur_artist, cur_album, candidates, rec
def tag_item(item, timid=False, search_artist=None, search_title=None,
search_id=None):
@@ -450,7 +460,9 @@ def tag_item(item, timid=False, search_artist=None, search_title=None,
`search_title` may be used to override the current metadata for
the purposes of the MusicBrainz title; likewise `search_id`.
"""
candidates = []
# Holds candidates found so far: keys are MBIDs; values are
# (distance, TrackInfo) pairs.
candidates = {}
# First, try matching by MusicBrainz ID.
trackid = search_id or item.mb_trackid
@@ -459,17 +471,17 @@ def tag_item(item, timid=False, search_artist=None, search_title=None,
track_info = hooks._track_for_id(trackid)
if track_info:
dist = track_distance(item, track_info, incl_artist=True)
candidates.append((dist, track_info))
candidates[track_info.track_id] = (dist, track_info)
# If this is a good match, then don't keep searching.
rec = recommendation(candidates)
rec = recommendation(candidates.values())
if rec == RECOMMEND_STRONG and not timid:
log.debug('Track ID match.')
return candidates, rec
return candidates.values(), rec
# If we're searching by ID, don't proceed.
if search_id is not None:
if candidates:
return candidates, rec
return candidates.values(), rec
else:
return [], RECOMMEND_NONE
@@ -481,10 +493,10 @@ def tag_item(item, timid=False, search_artist=None, search_title=None,
# Get and evaluate candidate metadata.
for track_info in hooks._item_candidates(item, search_artist, search_title):
dist = track_distance(item, track_info, incl_artist=True)
candidates.append((dist, track_info))
candidates[track_info.track_id] = (dist, track_info)
# Sort by distance and return with recommendation.
log.debug('Found %i candidates.' % len(candidates))
candidates.sort()
candidates = sorted(candidates.itervalues())
rec = recommendation(candidates)
return candidates, rec

149
lib/beets/autotag/mb.py
View File

@@ -15,52 +15,53 @@
"""Searches for albums in the MusicBrainz database.
"""
import logging
import lib.musicbrainzngs as musicbrainzngs
from . import musicbrainz3
import lib.beets.autotag.hooks
import lib.beets
SEARCH_LIMIT = 5
VARIOUS_ARTISTS_ID = '89ad4ac3-39f7-470e-963a-56509c546377'
musicbrainz3._useragent = 'beets/%s' % lib.beets.__version__
musicbrainzngs.set_useragent('beets', lib.beets.__version__,
'http://beets.radbox.org/')
class ServerBusyError(Exception): pass
class BadResponseError(Exception): pass
log = logging.getLogger('beets')
# We hard-code IDs for artists that can't easily be searched for.
SPECIAL_CASE_ARTISTS = {
'!!!': 'f26c72d3-e52c-467b-b651-679c73d8e1a7',
}
RELEASE_INCLUDES = ['artists', 'media', 'recordings', 'release-groups',
'labels']
'labels', 'artist-credits']
TRACK_INCLUDES = ['artists']
def _adapt_criteria(criteria):
"""Special-case artists in a criteria dictionary before it is passed
to the MusicBrainz search server. The dictionary supplied is
mutated; nothing is returned.
"""
if 'artist' in criteria:
for artist, artist_id in SPECIAL_CASE_ARTISTS.items():
if criteria['artist'] == artist:
criteria['arid'] = artist_id
del criteria['artist']
break
# python-musicbrainz-ngs search functions: tolerate different API versions.
if hasattr(musicbrainzngs, 'release_search'):
# Old API names.
_mb_release_search = musicbrainzngs.release_search
_mb_recording_search = musicbrainzngs.recording_search
else:
# New API names.
_mb_release_search = musicbrainzngs.search_releases
_mb_recording_search = musicbrainzngs.search_recordings
def track_info(recording):
def track_info(recording, medium=None, medium_index=None):
"""Translates a MusicBrainz recording result dictionary into a beets
``TrackInfo`` object.
``TrackInfo`` object. ``medium_index``, if provided, is the track's
index (1-based) on its medium.
"""
info = lib.beets.autotag.hooks.TrackInfo(recording['title'],
recording['id'])
recording['id'],
medium=medium,
medium_index=medium_index)
if 'artist-credit' in recording: # XXX: when is this not included?
# Get the name of the track artist.
if recording.get('artist-credit-phrase'):
info.artist = recording['artist-credit-phrase']
# Get the ID of the first artist.
if 'artist-credit' in recording:
artist = recording['artist-credit'][0]['artist']
info.artist = artist['name']
info.artist_id = artist['id']
if recording.get('length'):
@@ -68,45 +69,74 @@ def track_info(recording):
return info
def _set_date_str(info, date_str):
"""Given a (possibly partial) YYYY-MM-DD string and an AlbumInfo
object, set the object's release date fields appropriately.
"""
if date_str:
date_parts = date_str.split('-')
for key in ('year', 'month', 'day'):
if date_parts:
setattr(info, key, int(date_parts.pop(0)))
def album_info(release):
"""Takes a MusicBrainz release result dictionary and returns a beets
AlbumInfo object containing the interesting data about that release.
"""
# Get artist name using join phrases.
artist_parts = []
for el in release['artist-credit']:
if isinstance(el, basestring):
artist_parts.append(el)
else:
artist_parts.append(el['artist']['name'])
artist_name = ''.join(artist_parts)
# Basic info.
artist = release['artist-credit'][0]['artist']
tracks = []
track_infos = []
for medium in release['medium-list']:
tracks.extend(i['recording'] for i in medium['track-list'])
for track in medium['track-list']:
ti = track_info(track['recording'],
int(medium['position']),
int(track['position']))
if track.get('title'):
# Track title may be distinct from underling recording
# title.
ti.title = track['title']
track_infos.append(ti)
info = lib.beets.autotag.hooks.AlbumInfo(
release['title'],
release['id'],
artist['name'],
artist['id'],
[track_info(track) for track in tracks],
artist_name,
release['artist-credit'][0]['artist']['id'],
track_infos,
mediums=len(release['medium-list']),
)
info.va = info.artist_id == VARIOUS_ARTISTS_ID
if 'asin' in release:
info.asin = release['asin']
# Release type not always populated.
reltype = release['release-group']['type']
if reltype:
info.albumtype = reltype.lower()
if 'type' in release['release-group']:
reltype = release['release-group']['type']
if reltype:
info.albumtype = reltype.lower()
# Release date.
if 'date' in release: # XXX: when is this not included?
date_str = release['date']
if date_str:
date_parts = date_str.split('-')
for key in ('year', 'month', 'day'):
if date_parts:
setattr(info, key, int(date_parts.pop(0)))
if 'first-release-date' in release['release-group']:
# Try earliest release date for the entire group first.
_set_date_str(info, release['release-group']['first-release-date'])
elif 'date' in release:
# Fall back to release-specific date.
_set_date_str(info, release['date'])
# Label name.
if release.get('label-info-list'):
label = release['label-info-list'][0]['label']['name']
if label != '[no label]':
info.label = label
label_info = release['label-info-list'][0]
if label_info.get('label'):
label = label_info['label']['name']
if label != '[no label]':
info.label = label
return info
@@ -118,33 +148,40 @@ def match_album(artist, album, tracks=None, limit=SEARCH_LIMIT):
optionally, a number of tracks on the album.
"""
# Build search criteria.
criteria = {'release': album}
criteria = {'release': album.lower()}
if artist is not None:
criteria['artist'] = artist
criteria['artist'] = artist.lower()
else:
# Various Artists search.
criteria['arid'] = VARIOUS_ARTISTS_ID
if tracks is not None:
criteria['tracks'] = str(tracks)
_adapt_criteria(criteria)
res = musicbrainz3.release_search(limit=limit, **criteria)
# Abort if we have no search terms.
if not any(criteria.itervalues()):
return
res = _mb_release_search(limit=limit, **criteria)
for release in res['release-list']:
# The search result is missing some data (namely, the tracks),
# so we just use the ID and fetch the rest of the information.
yield album_for_id(release['id'])
albuminfo = album_for_id(release['id'])
assert albuminfo is not None
yield albuminfo
def match_track(artist, title, limit=SEARCH_LIMIT):
"""Searches for a single track and returns an iterable of TrackInfo
objects.
"""
criteria = {
'artist': artist,
'recording': title,
'artist': artist.lower(),
'recording': title.lower(),
}
_adapt_criteria(criteria)
res = musicbrainz3.recording_search(limit=limit, **criteria)
if not any(criteria.itervalues()):
return
res = _mb_recording_search(limit=limit, **criteria)
for recording in res['recording-list']:
yield track_info(recording)
@@ -153,8 +190,8 @@ def album_for_id(albumid):
object or None if the album is not found.
"""
try:
res = musicbrainz3.get_release_by_id(albumid, RELEASE_INCLUDES)
except musicbrainz3.ResponseError:
res = musicbrainzngs.get_release_by_id(albumid, RELEASE_INCLUDES)
except musicbrainzngs.ResponseError:
log.debug('Album ID match failed.')
return None
return album_info(res['release'])
@@ -164,8 +201,8 @@ def track_for_id(trackid):
or None if no track is found.
"""
try:
res = musicbrainz3.get_recording_by_id(trackid, TRACK_INCLUDES)
except musicbrainz3.ResponseError:
res = musicbrainzngs.get_recording_by_id(trackid, TRACK_INCLUDES)
except musicbrainzngs.ResponseError:
log.debug('Track ID match failed.')
return None
return track_info(res['recording'])

263
lib/beets/importer.py
View File

@@ -25,8 +25,9 @@ from lib.beets import autotag
from lib.beets import library
import lib.beets.autotag.art
from lib.beets import plugins
from lib.beets import util
from lib.beets.util import pipeline
from lib.beets.util import syspath, normpath, plurality
from lib.beets.util import syspath, normpath, displayable_path
from lib.beets.util.enumeration import enum
action = enum(
@@ -56,15 +57,28 @@ def tag_log(logfile, status, path):
"""
if logfile:
print >>logfile, '%s %s' % (status, path)
logfile.flush()
def log_choice(config, task):
"""Logs the task's current choice if it should be logged.
def log_choice(config, task, duplicate=False):
"""Logs the task's current choice if it should be logged. If
``duplicate``, then this is a secondary choice after a duplicate was
detected and a decision was made.
"""
path = task.path if task.is_album else task.item.path
if task.choice_flag is action.ASIS:
tag_log(config.logfile, 'asis', path)
elif task.choice_flag is action.SKIP:
tag_log(config.logfile, 'skip', path)
if duplicate:
# Duplicate: log all three choices (skip, keep both, and trump).
if task.remove_duplicates:
tag_log(config.logfile, 'duplicate-replace', path)
elif task.choice_flag in (action.ASIS, action.APPLY):
tag_log(config.logfile, 'duplicate-keep', path)
elif task.choice_flag is (action.SKIP):
tag_log(config.logfile, 'duplicate-skip', path)
else:
# Non-duplicate: log "skip" and "asis" choices.
if task.choice_flag is action.ASIS:
tag_log(config.logfile, 'asis', path)
elif task.choice_flag is action.SKIP:
tag_log(config.logfile, 'skip', path)
def _reopen_lib(lib):
"""Because of limitations in SQLite, a given Library is bound to
@@ -77,37 +91,25 @@ def _reopen_lib(lib):
lib.directory,
lib.path_formats,
lib.art_filename,
lib.timeout,
lib.replacements,
)
else:
return lib
def _duplicate_check(lib, task, recent=None):
"""Check whether an album already exists in the library. `recent`
should be a set of (artist, album) pairs that will be built up
with every call to this function and checked along with the
library.
def _duplicate_check(lib, task):
"""Check whether an album already exists in the library. Returns a
list of Album objects (empty if no duplicates are found).
"""
if task.choice_flag is action.ASIS:
artist = task.cur_artist
album = task.cur_album
elif task.choice_flag is action.APPLY:
artist = task.info.artist
album = task.info.album
else:
return False
assert task.choice_flag in (action.ASIS, action.APPLY)
artist, album = task.chosen_ident()
if artist is None:
# As-is import with no artist. Skip check.
return False
return []
# Try the recent albums.
if recent is not None:
if (artist, album) in recent:
return True
recent.add((artist, album))
# Look in the library.
cur_paths = set(i.path for i in task.items)
found_albums = []
cur_paths = set(i.path for i in task.items if i)
for album_cand in lib.albums(artist=artist):
if album_cand.album == album:
# Check whether the album is identical in contents, in which
@@ -115,38 +117,23 @@ def _duplicate_check(lib, task, recent=None):
other_paths = set(i.path for i in album_cand.items())
if other_paths == cur_paths:
continue
return True
found_albums.append(album_cand)
return found_albums
return False
def _item_duplicate_check(lib, task):
"""Check whether an item already exists in the library. Returns a
list of Item objects.
"""
assert task.choice_flag in (action.ASIS, action.APPLY)
artist, title = task.chosen_ident()
def _item_duplicate_check(lib, task, recent=None):
"""Check whether an item already exists in the library."""
if task.choice_flag is action.ASIS:
artist = task.item.artist
title = task.item.title
elif task.choice_flag is action.APPLY:
artist = task.info.artist
title = task.info.title
else:
return False
# Try recent items.
if recent is not None:
if (artist, title) in recent:
return True
recent.add((artist, title))
# Check the library.
item_iter = lib.items(artist=artist, title=title)
try:
for other_item in item_iter:
# Existing items not considered duplicates.
if other_item.path == task.item.path:
continue
return True
finally:
item_iter.close()
return False
found_items = []
for other_item in lib.items(artist=artist, title=title):
# Existing items not considered duplicates.
if other_item.path == task.item.path:
continue
found_items.append(other_item)
return found_items
def _infer_album_fields(task):
"""Given an album and an associated import task, massage the
@@ -160,7 +147,7 @@ def _infer_album_fields(task):
if task.choice_flag == action.ASIS:
# Taking metadata "as-is". Guess whether this album is VA.
plur_artist, freq = plurality([i.artist for i in task.items])
plur_artist, freq = util.plurality([i.artist for i in task.items])
if freq == len(task.items) or (freq > 1 and
float(freq) / len(task.items) >= SINGLE_ARTIST_THRESH):
# Single-artist album.
@@ -174,30 +161,40 @@ def _infer_album_fields(task):
elif task.choice_flag == action.APPLY:
# Applying autotagged metadata. Just get AA from the first
# item.
if not task.items[0].albumartist:
changes['albumartist'] = task.items[0].artist
if not task.items[0].mb_albumartistid:
changes['mb_albumartistid'] = task.items[0].mb_artistid
for item in task.items:
if item is not None:
first_item = item
break
else:
assert False, "all items are None"
if not first_item.albumartist:
changes['albumartist'] = first_item.artist
if not first_item.mb_albumartistid:
changes['mb_albumartistid'] = first_item.mb_artistid
else:
assert False
# Apply new metadata.
for item in task.items:
for k, v in changes.iteritems():
setattr(item, k, v)
if item is not None:
for k, v in changes.iteritems():
setattr(item, k, v)
def _open_state():
"""Reads the state file, returning a dictionary."""
try:
with open(STATE_FILE) as f:
return pickle.load(f)
except IOError:
except (IOError, EOFError):
return {}
def _save_state(state):
"""Writes the state dictionary out to disk."""
with open(STATE_FILE, 'w') as f:
pickle.dump(state, f)
try:
with open(STATE_FILE, 'w') as f:
pickle.dump(state, f)
except IOError, exc:
log.error(u'state file could not be written: %s' % unicode(exc))
# Utilities for reading and writing the beets progress file, which
@@ -265,7 +262,8 @@ class ImportConfig(object):
'quiet_fallback', 'copy', 'write', 'art', 'delete',
'choose_match_func', 'should_resume_func', 'threaded',
'autot', 'singletons', 'timid', 'choose_item_func',
'query', 'incremental']
'query', 'incremental', 'ignore',
'resolve_duplicate_func']
def __init__(self, **kwargs):
for slot in self._fields:
setattr(self, slot, kwargs[slot])
@@ -297,6 +295,7 @@ class ImportTask(object):
self.path = path
self.items = items
self.sentinel = False
self.remove_duplicates = False
@classmethod
def done_sentinel(cls, toppath):
@@ -412,6 +411,26 @@ class ImportTask(object):
"""
return self.sentinel or self.choice_flag == action.SKIP
# Useful data.
def chosen_ident(self):
"""Returns identifying metadata about the current choice. For
albums, this is an (artist, album) pair. For items, this is
(artist, title). May only be called when the choice flag is ASIS
(in which case the data comes from the files' current metadata)
or APPLY (data comes from the choice).
"""
assert self.choice_flag in (action.ASIS, action.APPLY)
if self.is_album:
if self.choice_flag is action.ASIS:
return (self.cur_artist, self.cur_album)
elif self.choice_flag is action.APPLY:
return (self.info.artist, self.info.album)
else:
if self.choice_flag is action.ASIS:
return (self.item.artist, self.item.title)
elif self.choice_flag is action.APPLY:
return (self.info.artist, self.info.title)
# Full-album pipeline stages.
@@ -443,6 +462,7 @@ def read_tasks(config):
# Look for saved incremental directories.
if config.incremental:
incremental_skipped = 0
history_dirs = history_get()
for toppath in config.paths:
@@ -455,7 +475,7 @@ def read_tasks(config):
# Produce paths under this directory.
if progress:
resume_dir = resume_dirs.get(toppath)
for path, items in autotag.albums_in_dir(toppath):
for path, items in autotag.albums_in_dir(toppath, config.ignore):
# Skip according to progress.
if progress and resume_dir:
# We're fast-forwarding to resume a previous tagging.
@@ -467,6 +487,9 @@ def read_tasks(config):
# When incremental, skip paths in the history.
if config.incremental and path in history_dirs:
log.debug(u'Skipping previously-imported path: %s' %
displayable_path(path))
incremental_skipped += 1
continue
# Yield all the necessary tasks.
@@ -480,6 +503,11 @@ def read_tasks(config):
# Indicate the directory is finished.
yield ImportTask.done_sentinel(toppath)
# Show skipped directories.
if config.incremental and incremental_skipped:
log.info(u'Incremental import: skipped %i directories.' %
incremental_skipped)
def query_tasks(config):
"""A generator that works as a drop-in-replacement for read_tasks.
Instead of finding files from the filesystem, a query is used to
@@ -489,14 +517,12 @@ def query_tasks(config):
if config.singletons:
# Search for items.
items = list(lib.items(config.query))
for item in items:
for item in lib.items(config.query):
yield ImportTask.item_task(item)
else:
# Search for albums.
albums = lib.albums(config.query)
for album in albums:
for album in lib.albums(config.query):
log.debug('yielding album %i: %s - %s' %
(album.id, album.albumartist, album.album))
items = list(album.items())
@@ -558,10 +584,15 @@ def user_query(config):
continue
# Check for duplicates if we have a match (or ASIS).
if _duplicate_check(lib, task, recent):
tag_log(config.logfile, 'duplicate', task.path)
log.warn("This album is already in the library!")
task.set_choice(action.SKIP)
if task.choice_flag in (action.ASIS, action.APPLY):
ident = task.chosen_ident()
# The "recent" set keeps track of identifiers for recently
# imported albums -- those that haven't reached the database
# yet.
if ident in recent or _duplicate_check(lib, task):
config.resolve_duplicate_func(task, config)
log_choice(config, task, True)
recent.add(ident)
def show_progress(config):
"""This stage replaces the initial_lookup and user_query stages
@@ -591,7 +622,7 @@ def apply_choices(config):
if task.should_skip():
continue
items = task.items if task.is_album else [task.item]
items = [i for i in task.items if i] if task.is_album else [task.item]
# Clear IDs in case the items are being re-tagged.
for item in items:
item.id = None
@@ -608,28 +639,54 @@ def apply_choices(config):
if task.is_album:
_infer_album_fields(task)
# Find existing item entries that these are replacing. Old
# album structures are automatically cleaned up when the
# last item is removed.
# Find existing item entries that these are replacing (for
# re-imports). Old album structures are automatically cleaned up
# when the last item is removed.
replaced_items = defaultdict(list)
for item in items:
dup_items = list(lib.items(
library.MatchQuery('path', item.path)
))
dup_items = lib.items(library.MatchQuery('path', item.path))
for dup_item in dup_items:
replaced_items[item].append(dup_item)
log.debug('replacing item %i: %s' % (dup_item.id, item.path))
log.debug('replacing item %i: %s' %
(dup_item.id, displayable_path(item.path)))
log.debug('%i of %i items replaced' % (len(replaced_items),
len(items)))
# Find old items that should be replaced as part of a duplicate
# resolution.
duplicate_items = []
if task.remove_duplicates:
if task.is_album:
for album in _duplicate_check(lib, task):
duplicate_items += album.items()
else:
duplicate_items = _item_duplicate_check(lib, task)
log.debug('removing %i old duplicated items' %
len(duplicate_items))
# Delete duplicate files that are located inside the library
# directory.
for duplicate_path in [i.path for i in duplicate_items]:
if lib.directory in util.ancestry(duplicate_path):
log.debug(u'deleting replaced duplicate %s' %
util.displayable_path(duplicate_path))
util.soft_remove(duplicate_path)
util.prune_dirs(os.path.dirname(duplicate_path),
lib.directory)
# Move/copy files.
task.old_paths = [item.path for item in items]
for item in items:
if config.copy:
# If we're replacing an item, then move rather than
# copying.
old_path = item.path
do_copy = not bool(replaced_items[item])
lib.move(item, do_copy, task.is_album)
if not do_copy:
# If we moved the item, remove the now-nonexistent
# file from old_paths.
task.old_paths.remove(old_path)
if config.write and task.should_write_tags():
item.write()
@@ -640,11 +697,13 @@ def apply_choices(config):
for replaced in replaced_items.itervalues():
for item in replaced:
lib.remove(item)
for item in duplicate_items:
lib.remove(item)
# Add new ones.
if task.is_album:
# Add an album.
album = lib.add_album(task.items)
album = lib.add_album(items)
task.album_id = album.id
else:
# Add tracks.
@@ -672,6 +731,10 @@ def fetch_art(config):
try:
album = lib.get_album(task.album_id)
album.set_art(artpath)
if config.delete and not util.samefile(artpath,
album.artpath):
# Delete the original file after it's imported.
os.remove(artpath)
finally:
lib.save(False)
@@ -690,23 +753,27 @@ def finalize(config):
task.save_history()
continue
items = task.items if task.is_album else [task.item]
items = [i for i in task.items if i] if task.is_album else [task.item]
# Announce that we've added an album.
if task.is_album:
album = lib.get_album(task.album_id)
plugins.send('album_imported', lib=lib, album=album)
plugins.send('album_imported', lib=lib, album=album, config=config)
else:
for item in items:
plugins.send('item_imported', lib=lib, item=item)
plugins.send('item_imported', lib=lib, item=item, config=config)
# Finally, delete old files.
if config.copy and config.delete:
new_paths = [os.path.realpath(item.path) for item in items]
for old_path in task.old_paths:
# Only delete files that were actually moved.
# Only delete files that were actually copied.
if old_path not in new_paths:
os.remove(syspath(old_path))
# Clean up directory if it is emptied.
if task.toppath:
util.prune_dirs(os.path.dirname(old_path),
task.toppath)
# Update progress.
if config.resume is not False:
@@ -746,10 +813,12 @@ def item_query(config):
log_choice(config, task)
# Duplicate check.
if _item_duplicate_check(lib, task, recent):
tag_log(config.logfile, 'duplicate', task.item.path)
log.warn("This item is already in the library!")
task.set_choice(action.SKIP)
if task.choice_flag in (action.ASIS, action.APPLY):
ident = task.chosen_ident()
if ident in recent or _item_duplicate_check(lib, task):
config.resolve_duplicate_func(task, config)
log_choice(config, task, True)
recent.add(ident)
def item_progress(config):
"""Skips the lookup and query stages in a non-autotagged singleton
@@ -762,7 +831,7 @@ def item_progress(config):
if task.sentinel:
continue
log.info(task.item.path)
log.info(displayable_path(task.item.path))
task.set_null_item_match()
task.set_choice(action.ASIS)

342
lib/beets/library.py
View File

@@ -16,12 +16,14 @@ import sqlite3
import os
import re
import sys
from string import Template
import logging
import shlex
#from unidecode import unidecode
from lib.beets.mediafile import MediaFile
from lib.beets import plugins
from lib.beets import util
from lib.beets.util import bytestring_path, syspath, normpath, samefile
from lib.beets.util.functemplate import Template
MAX_FILENAME_LENGTH = 200
@@ -66,6 +68,10 @@ ITEM_FIELDS = [
('length', 'real', False, True),
('bitrate', 'int', False, True),
('format', 'text', False, True),
('samplerate', 'int', False, True),
('bitdepth', 'int', False, True),
('channels', 'int', False, True),
('mtime', 'int', False, False),
]
ITEM_KEYS_WRITABLE = [f[0] for f in ITEM_FIELDS if f[3] and f[2]]
ITEM_KEYS_META = [f[0] for f in ITEM_FIELDS if f[3]]
@@ -101,6 +107,9 @@ ALBUM_DEFAULT_FIELDS = ('album', 'albumartist', 'genre')
ITEM_DEFAULT_FIELDS = ARTIST_DEFAULT_FIELDS + ALBUM_DEFAULT_FIELDS + \
('title', 'comments')
# Special path format key.
PF_KEY_DEFAULT = 'default'
# Logger.
log = logging.getLogger('beets')
if not log.handlers:
@@ -130,6 +139,7 @@ class Item(object):
'album_id': None,
})
i.read(path)
i.mtime = i.current_mtime() # Initial mtime.
return i
def _fill_record(self, values):
@@ -176,10 +186,12 @@ class Item(object):
value = str(value)
if key in ITEM_KEYS:
# If the value changed, mark the field as dirty.
if (not (key in self.record)) or (self.record[key] != value):
# don't dirty if value unchanged
self.record[key] = value
self.dirty[key] = True
if key in ITEM_KEYS_WRITABLE:
self.mtime = 0 # Reset mtime on dirty.
else:
super(Item, self).__setattr__(key, value)
@@ -199,22 +211,33 @@ class Item(object):
for key in ITEM_KEYS_META:
setattr(self, key, getattr(f, key))
self.path = read_path
# Database's mtime should now reflect the on-disk value.
if read_path == self.path:
self.mtime = self.current_mtime()
def write(self):
"""Writes the item's metadata to the associated file.
"""
f = MediaFile(syspath(self.path))
plugins.send('write', item=self, mf=f)
for key in ITEM_KEYS_WRITABLE:
setattr(f, key, getattr(self, key))
f.save()
# The file has a new mtime.
self.mtime = self.current_mtime()
# Files themselves.
def move(self, dest, copy=False):
"""Moves or copies the item's file, updating the path value if
the move succeeds.
the move succeeds. If a file exists at ``dest``, then it is
slightly modified to be unique.
"""
if not util.samefile(self.path, dest):
dest = util.unique_path(dest)
if copy:
util.copy(self.path, dest)
else:
@@ -223,6 +246,12 @@ class Item(object):
# Either copying or moving succeeded, so update the stored path.
self.path = dest
def current_mtime(self):
"""Returns the current mtime of the file, rounded to the nearest
integer.
"""
return int(os.path.getmtime(syspath(self.path)))
# Library queries.
@@ -262,16 +291,6 @@ class Query(object):
c.close()
return (result[0], result[1] or 0.0)
def execute(self, library):
"""Runs the query in the specified library, returning a
ResultIterator.
"""
c = library.conn.cursor()
stmt, subs = self.statement()
log.debug('Executing query: %s' % stmt)
c.execute(stmt, subs)
return ResultIterator(c, library)
class FieldQuery(Query):
"""An abstract query that searches in a specific field for a
pattern.
@@ -303,7 +322,8 @@ class SubstringQuery(FieldQuery):
return clause, subvals
def match(self, item):
return self.pattern.lower() in getattr(item, self.field).lower()
value = getattr(item, self.field) or ''
return self.pattern.lower() in value.lower()
class BooleanQuery(MatchQuery):
"""Matches a boolean field. Pattern should either be a boolean or a
@@ -333,14 +353,14 @@ class CollectionQuery(Query):
"""An abstract query class that aggregates other queries. Can be
indexed like a list to access the sub-queries.
"""
def __init__(self, subqueries = ()):
def __init__(self, subqueries=()):
self.subqueries = subqueries
# is there a better way to do this?
def __len__(self): return len(self.subqueries)
def __getitem__(self, key): return self.subqueries[key]
def __iter__(self): iter(self.subqueries)
def __contains__(self, item): item in self.subqueries
def __iter__(self): return iter(self.subqueries)
def __contains__(self, item): return item in self.subqueries
def clause_with_joiner(self, joiner):
"""Returns a clause created by joining together the clauses of
@@ -395,7 +415,7 @@ class CollectionQuery(Query):
continue
key, pattern = res
if key is None: # No key specified.
if os.sep in pattern:
if os.sep in pattern and 'path' in all_keys:
# This looks like a path.
subqueries.append(PathQuery(pattern))
else:
@@ -404,7 +424,7 @@ class CollectionQuery(Query):
default_fields))
elif key.lower() == 'comp': # a boolean field
subqueries.append(BooleanQuery(key.lower(), pattern))
elif key.lower() == 'path':
elif key.lower() == 'path' and 'path' in all_keys:
subqueries.append(PathQuery(pattern))
elif key.lower() in all_keys: # ignore unrecognized keys
subqueries.append(SubstringQuery(key.lower(), pattern))
@@ -414,6 +434,13 @@ class CollectionQuery(Query):
subqueries = [TrueQuery()]
return cls(subqueries)
@classmethod
def from_string(cls, query, default_fields=None, all_keys=ITEM_KEYS):
"""Creates a query based on a single string. The string is split
into query parts using shell-style syntax.
"""
return cls.from_strings(shlex.split(query))
class AnySubstringQuery(CollectionQuery):
"""A query that matches a substring in any of a list of metadata
fields.
@@ -468,6 +495,14 @@ class TrueQuery(Query):
def match(self, item):
return True
class FalseQuery(Query):
"""A query that never matches."""
def clause(self):
return '0', ()
def match(self, item):
return False
class PathQuery(Query):
"""A query that matches all items under a given path."""
def __init__(self, path):
@@ -486,28 +521,23 @@ class PathQuery(Query):
return '(path = ?) || (path LIKE ?)', (file_blob, dir_pat)
class ResultIterator(object):
"""An iterator into an item query result set."""
"""An iterator into an item query result set. The iterator eagerly
fetches all of the results from the cursor but lazily constructs
Item objects that reflect them.
"""
def __init__(self, cursor):
# Fetch all of the rows, closing the cursor (and unlocking the
# database).
self.rows = cursor.fetchall()
self.rowiter = iter(self.rows)
def __init__(self, cursor, library):
self.cursor = cursor
self.library = library
def __iter__(self): return self
def __iter__(self):
return self
def next(self):
try:
row = self.cursor.next()
except StopIteration:
self.cursor.close()
raise
row = self.rowiter.next() # May raise StopIteration.
return Item(row)
def close(self):
self.cursor.close()
def __del__(self):
self.close()
# An abstract library.
@@ -696,8 +726,11 @@ class Library(BaseLibrary):
"""A music library using an SQLite database as a metadata store."""
def __init__(self, path='library.blb',
directory='~/Music',
path_formats=None,
path_formats=((PF_KEY_DEFAULT,
'$artist/$album/$track $title'),),
art_filename='cover',
timeout=5.0,
replacements=None,
item_fields=ITEM_FIELDS,
album_fields=ALBUM_FIELDS):
if path == ':memory:':
@@ -705,14 +738,12 @@ class Library(BaseLibrary):
else:
self.path = bytestring_path(normpath(path))
self.directory = bytestring_path(normpath(directory))
if path_formats is None:
path_formats = {'default': '$artist/$album/$track $title'}
elif isinstance(path_formats, basestring):
path_formats = {'default': path_formats}
self.path_formats = path_formats
self.art_filename = bytestring_path(art_filename)
self.replacements = replacements
self.conn = sqlite3.connect(self.path)
self.timeout = timeout
self.conn = sqlite3.connect(self.path, timeout)
self.conn.row_factory = sqlite3.Row
# this way we can access our SELECT results like dictionaries
@@ -758,10 +789,10 @@ class Library(BaseLibrary):
if table == 'albums' and 'artist' in current_fields and \
'albumartist' not in current_fields:
setup_sql += "UPDATE ALBUMS SET albumartist=artist;\n"
self.conn.executescript(setup_sql)
self.conn.commit()
def destination(self, item, pathmod=None, in_album=False,
fragment=False, basedir=None):
"""Returns the path in the library directory designated for item
@@ -773,25 +804,37 @@ class Library(BaseLibrary):
directory for the destination.
"""
pathmod = pathmod or os.path
# Use a path format based on the album type, if available.
if not item.album_id and not in_album:
# Singleton track. Never use the "album" formats.
if 'singleton' in self.path_formats:
path_format = self.path_formats['singleton']
else:
path_format = self.path_formats['default']
elif item.albumtype and item.albumtype in self.path_formats:
path_format = self.path_formats[item.albumtype]
elif item.comp and 'comp' in self.path_formats:
path_format = self.path_formats['comp']
# Use a path format based on a query, falling back on the
# default.
for query, path_format in self.path_formats:
if query == PF_KEY_DEFAULT:
continue
query = AndQuery.from_string(query)
if in_album:
# If we're treating this item as a member of the item,
# hack the query so that singleton queries always
# observe the item to be non-singleton.
for i, subquery in enumerate(query):
if isinstance(subquery, SingletonQuery):
query[i] = FalseQuery() if subquery.sense \
else TrueQuery()
if query.match(item):
# The query matches the item! Use the corresponding path
# format.
break
else:
path_format = self.path_formats['default']
# No query matched; fall back to default.
for query, path_format in self.path_formats:
if query == PF_KEY_DEFAULT:
break
else:
assert False, "no default path format"
subpath_tmpl = Template(path_format)
# Get the item's Album if it has one.
album = self.get_album(item)
# Build the mapping for substitution in the path template,
# beginning with the values from the database.
mapping = {}
@@ -804,29 +847,35 @@ class Library(BaseLibrary):
# From Item.
value = getattr(item, key)
mapping[key] = util.sanitize_for_path(value, pathmod, key)
# Use the album artist if the track artist is not set and
# vice-versa.
if not mapping['artist']:
mapping['artist'] = mapping['albumartist']
if not mapping['albumartist']:
mapping['albumartist'] = mapping['artist']
# Get values from plugins.
for key, value in plugins.template_values(item).iteritems():
mapping[key] = util.sanitize_for_path(value, pathmod, key)
# Perform substitution.
subpath = subpath_tmpl.substitute(mapping)
funcs = DefaultTemplateFunctions(self, item).functions()
funcs.update(plugins.template_funcs())
subpath = subpath_tmpl.substitute(mapping, funcs)
# Encode for the filesystem, dropping unencodable characters.
if isinstance(subpath, unicode) and not fragment:
encoding = sys.getfilesystemencoding() or sys.getdefaultencoding()
subpath = subpath.encode(encoding, 'replace')
# Truncate components and remove forbidden characters.
subpath = util.sanitize_path(subpath)
subpath = util.sanitize_path(subpath, pathmod, self.replacements)
# Preserve extension.
_, extension = pathmod.splitext(item.path)
subpath += extension
subpath += extension.lower()
if fragment:
return subpath
else:
@@ -837,7 +886,6 @@ class Library(BaseLibrary):
# Item manipulation.
def add(self, item, copy=False):
#FIXME make a deep copy of the item?
item.library = self
if copy:
self.move(item, copy=True)
@@ -852,18 +900,18 @@ class Library(BaseLibrary):
if key == 'path' and isinstance(value, str):
value = buffer(value)
subvars.append(value)
# issue query
c = self.conn.cursor()
query = 'INSERT INTO items (' + columns + ') VALUES (' + values + ')'
c.execute(query, subvars)
new_id = c.lastrowid
c.close()
item._clear_dirty()
item.id = new_id
return new_id
def save(self, event=True):
"""Writes the library to disk (completing an sqlite
transaction).
@@ -875,7 +923,7 @@ class Library(BaseLibrary):
def load(self, item, load_id=None):
if load_id is None:
load_id = item.id
c = self.conn.execute(
'SELECT * FROM items WHERE id=?', (load_id,) )
item._fill_record(c.fetchone())
@@ -885,7 +933,7 @@ class Library(BaseLibrary):
def store(self, item, store_id=None, store_all=False):
if store_id is None:
store_id = item.id
# build assignments for query
assignments = ''
subvars = []
@@ -898,7 +946,7 @@ class Library(BaseLibrary):
if key == 'path' and isinstance(value, str):
value = buffer(value)
subvars.append(value)
if not assignments:
# nothing to store (i.e., nothing was dirty)
return
@@ -1011,7 +1059,7 @@ class Library(BaseLibrary):
" ORDER BY artist, album, disc, track"
log.debug('Getting items with SQL: %s' % sql)
c = self.conn.execute(sql, subvals)
return ResultIterator(c, self)
return ResultIterator(c)
# Convenience accessors.
@@ -1020,13 +1068,11 @@ class Library(BaseLibrary):
"""Fetch an Item by its ID. Returns None if no match is found.
"""
c = self.conn.execute("SELECT * FROM items WHERE id=?", (id,))
it = ResultIterator(c, self)
it = ResultIterator(c)
try:
return it.next()
except StopIteration:
return None
finally:
it.close()
def get_album(self, item_or_id):
"""Given an album ID or an item associated with an album,
@@ -1147,7 +1193,7 @@ class Album(BaseAlbum):
'SELECT * FROM items WHERE album_id=?',
(self.id,)
)
return ResultIterator(c, self._library)
return ResultIterator(c)
def remove(self, delete=False, with_items=True):
"""Removes this album and all its associated items from the
@@ -1256,5 +1302,141 @@ class Album(BaseAlbum):
# Normal operation.
if oldart == artdest:
util.soft_remove(oldart)
artdest = util.unique_path(artdest)
util.copy(path, artdest)
self.artpath = artdest
# Default path template resources.
def _int_arg(s):
"""Convert a string argument to an integer for use in a template
function. May raise a ValueError.
"""
return int(s.strip())
class DefaultTemplateFunctions(object):
"""A container class for the default functions provided to path
templates. These functions are contained in an object to provide
additional context to the functions -- specifically, the Item being
evaluated.
"""
def __init__(self, lib, item):
self.lib = lib
self.item = item
_prefix = 'tmpl_'
def functions(self):
"""Returns a dictionary containing the functions defined in this
object. The keys are function names (as exposed in templates)
and the values are Python functions.
"""
out = {}
for key in dir(self):
if key.startswith(self._prefix):
out[key[len(self._prefix):]] = getattr(self, key)
return out
@staticmethod
def tmpl_lower(s):
"""Convert a string to lower case."""
return s.lower()
@staticmethod
def tmpl_upper(s):
"""Covert a string to upper case."""
return s.upper()
@staticmethod
def tmpl_title(s):
"""Convert a string to title case."""
return s.title()
@staticmethod
def tmpl_left(s, chars):
"""Get the leftmost characters of a string."""
return s[0:_int_arg(chars)]
@staticmethod
def tmpl_right(s, chars):
"""Get the rightmost characters of a string."""
return s[-_int_arg(chars):]
@staticmethod
def tmpl_if(condition, trueval, falseval=u''):
"""If ``condition`` is nonempty and nonzero, emit ``trueval``;
otherwise, emit ``falseval`` (if provided).
"""
try:
condition = _int_arg(condition)
except ValueError:
condition = condition.strip()
if condition:
return trueval
else:
return falseval
@staticmethod
def tmpl_asciify(s):
"""Translate non-ASCII characters to their ASCII equivalents.
"""
#return unidecode(s)
return s
def tmpl_unique(self, keys, disam):
"""Generate a string that is guaranteed to be unique among all
albums in the library who share the same set of keys. Fields
from "disam" are used in the string if they are sufficient to
disambiguate the albums. Otherwise, a fallback opaque value is
used. Both "keys" and "disam" should be given as
whitespace-separated lists of field names.
"""
keys = keys.split()
disam = disam.split()
album = self.lib.get_album(self.item)
if not album:
# Do nothing for singletons.
return u''
# Find matching albums to disambiguate with.
subqueries = []
for key in keys:
value = getattr(album, key)
subqueries.append(MatchQuery(key, value))
albums = self.lib.albums(query=AndQuery(subqueries))
# If there's only one album to matching these details, then do
# nothing.
if len(albums) == 1:
return u''
# Find the minimum number of fields necessary to disambiguate
# the set of albums.
disambiguators = []
for field in disam:
disambiguators.append(field)
# Get the value tuple for each album for these
# disambiguators.
disam_values = set()
for a in albums:
values = [getattr(a, f) for f in disambiguators]
disam_values.add(tuple(values))
# If the set of unique tuples is equal to the number of
# albums in the disambiguation set, we're done -- this is
# sufficient disambiguation.
if len(disam_values) == len(albums):
break
else:
# Even when using all of the disambiguating fields, we
# could not separate all the albums. Fall back to the unique
# album ID.
return u' {}'.format(album.id)
# Flatten disambiguation values into a string.
values = [unicode(getattr(album, f)) for f in disambiguators]
return u' [{}]'.format(u' '.join(values))

160
lib/beets/mediafile.py
View File

@@ -94,7 +94,7 @@ def _safe_cast(out_type, val):
if not isinstance(val, basestring):
val = unicode(val)
# Get a number from the front of the string.
val = re.match('[0-9]*', val.strip()).group(0)
val = re.match(r'[0-9]*', val.strip()).group(0)
if not val:
return 0
else:
@@ -114,7 +114,26 @@ def _safe_cast(out_type, val):
if val is None:
return u''
else:
return unicode(val)
if isinstance(val, str):
return val.decode('utf8', 'ignore')
elif isinstance(val, unicode):
return val
else:
return unicode(val)
elif out_type == float:
if val is None:
return 0.0
elif isinstance(val, int) or isinstance(val, float):
return float(val)
else:
if not isinstance(val, basestring):
val = unicode(val)
val = re.match(r'[\+-]?[0-9\.]*', val.strip()).group(0)
if not val:
return 0.0
else:
return float(val)
else:
return val
@@ -275,7 +294,7 @@ class MediaField(object):
frames = obj.mgfile.tags.getall(style.key)
entry = None
for frame in frames:
if frame.desc == style.id3_desc:
if frame.desc.lower() == style.id3_desc.lower():
entry = getattr(frame, style.id3_frame_field)
break
if entry is None: # no desc match
@@ -298,7 +317,13 @@ class MediaField(object):
# possibly index the list
if style.list_elem:
if entry: # List must have at least one value.
return entry[0]
# Handle Mutagen bugs when reading values (#356).
try:
return entry[0]
except:
log.error('Mutagen exception when reading field: %s' %
traceback.format_exc)
return None
else:
return None
else:
@@ -321,7 +346,7 @@ class MediaField(object):
# try modifying in place
found = False
for frame in frames:
if frame.desc == style.id3_desc:
if frame.desc.lower() == style.id3_desc.lower():
setattr(frame, style.id3_frame_field, out)
found = True
break
@@ -380,19 +405,22 @@ class MediaField(object):
"""
# Fetch the data using the various StorageStyles.
styles = self._styles(obj)
for style in styles:
# Use the first style that returns a reasonable value.
out = self._fetchdata(obj, style)
if out:
break
if styles is None:
out = None
else:
for style in styles:
# Use the first style that returns a reasonable value.
out = self._fetchdata(obj, style)
if out:
break
if style.packing:
out = Packed(out, style.packing)[style.pack_pos]
if style.packing:
out = Packed(out, style.packing)[style.pack_pos]
# MPEG-4 freeform frames are (should be?) encoded as UTF-8.
if obj.type == 'mp4' and style.key.startswith('----:') and \
isinstance(out, str):
out = out.decode('utf8')
# MPEG-4 freeform frames are (should be?) encoded as UTF-8.
if obj.type == 'mp4' and style.key.startswith('----:') and \
isinstance(out, str):
out = out.decode('utf8')
return _safe_cast(self.out_type, out)
@@ -401,6 +429,9 @@ class MediaField(object):
"""
# Store using every StorageStyle available.
styles = self._styles(obj)
if styles is None:
return
for style in styles:
if style.packing:
@@ -430,6 +461,8 @@ class MediaField(object):
if self.out_type == bool:
# store bools as 1,0 instead of True,False
out = unicode(int(out))
elif isinstance(out, str):
out = out.decode('utf8', 'ignore')
else:
out = unicode(out)
elif style.as_type == int:
@@ -612,6 +645,30 @@ class ImageField(object):
base64.b64encode(pic.write())
]
class FloatValueField(MediaField):
"""A field that stores a floating-point number as a string."""
def __init__(self, places=2, suffix=None, **kwargs):
"""Make a field that stores ``places`` digits after the decimal
point and appends ``suffix`` (if specified) when encoding as a
string.
"""
super(FloatValueField, self).__init__(unicode, **kwargs)
fmt = ['%.', str(places), 'f']
if suffix:
fmt += [' ', suffix]
self.fmt = ''.join(fmt)
def __get__(self, obj, owner):
valstr = super(FloatValueField, self).__get__(obj, owner)
return _safe_cast(float, valstr)
def __set__(self, obj, val):
if not val:
val = 0.0
valstr = self.fmt % val
super(FloatValueField, self).__set__(obj, valstr)
# The file (a collection of fields).
@@ -865,21 +922,90 @@ class MediaFile(object):
etc = StorageStyle('musicbrainz_albumartistid')
)
# ReplayGain fields.
rg_track_gain = FloatValueField(2, 'dB',
mp3 = StorageStyle('TXXX',
id3_desc=u'REPLAYGAIN_TRACK_GAIN'),
mp4 = None,
etc = StorageStyle(u'REPLAYGAIN_TRACK_GAIN')
)
rg_album_gain = FloatValueField(2, 'dB',
mp3 = StorageStyle('TXXX',
id3_desc=u'REPLAYGAIN_ALBUM_GAIN'),
mp4 = None,
etc = StorageStyle(u'REPLAYGAIN_ALBUM_GAIN')
)
rg_track_peak = FloatValueField(6, None,
mp3 = StorageStyle('TXXX',
id3_desc=u'REPLAYGAIN_TRACK_PEAK'),
mp4 = None,
etc = StorageStyle(u'REPLAYGAIN_TRACK_PEAK')
)
rg_album_peak = FloatValueField(6, None,
mp3 = StorageStyle('TXXX',
id3_desc=u'REPLAYGAIN_ALBUM_PEAK'),
mp4 = None,
etc = StorageStyle(u'REPLAYGAIN_ALBUM_PEAK')
)
@property
def length(self):
"""The duration of the audio in seconds (a float)."""
return self.mgfile.info.length
@property
def samplerate(self):
"""The audio's sample rate (an int)."""
if hasattr(self.mgfile.info, 'sample_rate'):
return self.mgfile.info.sample_rate
return 0
@property
def bitdepth(self):
"""The number of bits per sample in the audio encoding (an int).
Only available for certain file formats (zero where
unavailable).
"""
if hasattr(self.mgfile.info, 'bits_per_sample'):
return self.mgfile.info.bits_per_sample
return 0
@property
def channels(self):
"""The number of channels in the audio (an int)."""
if isinstance(self.mgfile.info, lib.mutagen.mp3.MPEGInfo):
return {
lib.mutagen.mp3.STEREO: 2,
lib.mutagen.mp3.JOINTSTEREO: 2,
lib.mutagen.mp3.DUALCHANNEL: 2,
lib.mutagen.mp3.MONO: 1,
}[self.mgfile.info.mode]
if hasattr(self.mgfile.info, 'channels'):
return self.mgfile.info.channels
return 0
@property
def bitrate(self):
if hasattr(self.mgfile.info, 'bitrate'):
"""The number of bits per seconds used in the audio coding (an
int). If this is provided explicitly by the compressed file
format, this is a precise reflection of the encoding. Otherwise,
it is estimated from the on-disk file size. In this case, some
imprecision is possible because the file header is incorporated
in the file size.
"""
if hasattr(self.mgfile.info, 'bitrate') and self.mgfile.info.bitrate:
# Many formats provide it explicitly.
return self.mgfile.info.bitrate
else:
# Otherwise, we calculate bitrate from the file size. (This
# is the case for all of the lossless formats.)
if not self.length:
# Avoid division by zero if length is not available.
return 0
size = os.path.getsize(self.path)
return int(size * 8 / self.length)
@property
def format(self):
"""A string describing the file format/codec."""
return TYPES[self.type]

74
lib/beets/plugins.py
View File

@@ -19,6 +19,8 @@ import itertools
import traceback
from collections import defaultdict
from lib.beets import mediafile
PLUGIN_NAMESPACE = 'beetsplug'
DEFAULT_PLUGINS = []
@@ -36,6 +38,12 @@ class BeetsPlugin(object):
functionality by defining a subclass of BeetsPlugin and overriding
the abstract methods defined here.
"""
def __init__(self):
"""Perform one-time plugin setup. There is probably no reason to
override this method.
"""
_add_media_fields(self.item_fields())
def commands(self):
"""Should return a list of beets.ui.Subcommand objects for
commands that should be added to beets' CLI.
@@ -72,6 +80,14 @@ class BeetsPlugin(object):
"""
pass
def item_fields(self):
"""Returns field descriptors to be added to the MediaFile class,
in the form of a dictionary whose keys are field names and whose
values are descriptor (e.g., MediaField) instances. The Library
database schema is not (currently) extended.
"""
return {}
listeners = None
@classmethod
@@ -104,6 +120,36 @@ class BeetsPlugin(object):
return func
return helper
template_funcs = None
template_fields = None
@classmethod
def template_func(cls, name):
"""Decorator that registers a path template function. The
function will be invoked as ``%name{}`` from path format
strings.
"""
def helper(func):
if cls.template_funcs is None:
cls.template_funcs = {}
cls.template_funcs[name] = func
return func
return helper
@classmethod
def template_field(cls, name):
"""Decorator that registers a path template field computation.
The value will be referenced as ``$name`` from path format
strings. The function must accept a single parameter, the Item
being formatted.
"""
def helper(func):
if cls.template_fields is None:
cls.template_fields = {}
cls.template_fields[name] = func
return func
return helper
def load_plugins(names=()):
"""Imports the modules for a sequence of plugin names. Each name
must be the name of a Python module under the "beetsplug" namespace
@@ -195,6 +241,34 @@ def configure(config):
for plugin in find_plugins():
plugin.configure(config)
def template_funcs():
"""Get all the template functions declared by plugins as a
dictionary.
"""
funcs = {}
for plugin in find_plugins():
if plugin.template_funcs:
funcs.update(plugin.template_funcs)
return funcs
def template_values(item):
"""Get all the template values computed for a given Item by
registered field computations.
"""
values = {}
for plugin in find_plugins():
if plugin.template_fields:
for name, func in plugin.template_fields.iteritems():
values[name] = unicode(func(item))
return values
def _add_media_fields(fields):
"""Adds a {name: descriptor} dictionary of fields to the MediaFile
class. Called during the plugin initialization.
"""
for key, value in fields.iteritems():
setattr(mediafile.MediaFile, key, value)
# Event dispatch.

172
lib/beets/ui/__init__.py
View File

@@ -26,23 +26,36 @@ from difflib import SequenceMatcher
import logging
import sqlite3
import errno
import re
from lib.beets import library
from lib.beets import plugins
from lib.beets import util
if sys.platform == 'win32':
import colorama
colorama.init()
# Constants.
CONFIG_PATH_VAR = 'BEETSCONFIG'
DEFAULT_CONFIG_FILE = os.path.expanduser('~/.beetsconfig')
DEFAULT_LIBRARY = '~/.beetsmusic.blb'
DEFAULT_DIRECTORY = '~/Music'
DEFAULT_PATH_FORMATS = {
'default': '$albumartist/$album/$track $title',
'comp': 'Compilations/$album/$track $title',
'singleton': 'Non-Album/$artist/$title',
DEFAULT_CONFIG_FILENAME_UNIX = '.beetsconfig'
DEFAULT_CONFIG_FILENAME_WINDOWS = 'beetsconfig.ini'
DEFAULT_LIBRARY_FILENAME_UNIX = '.beetsmusic.blb'
DEFAULT_LIBRARY_FILENAME_WINDOWS = 'beetsmusic.blb'
DEFAULT_DIRECTORY_NAME = 'Music'
WINDOWS_BASEDIR = os.environ.get('APPDATA') or '~'
PF_KEY_QUERIES = {
'comp': 'comp:true',
'singleton': 'singleton:true',
}
DEFAULT_PATH_FORMATS = [
(library.PF_KEY_DEFAULT, '$albumartist/$album/$track $title'),
(PF_KEY_QUERIES['singleton'], 'Non-Album/$artist/$title'),
(PF_KEY_QUERIES['comp'], 'Compilations/$album/$track $title'),
]
DEFAULT_ART_FILENAME = 'cover'
DEFAULT_TIMEOUT = 5.0
NULL_REPLACE = '<strip>'
# UI exception. Commands should throw this in order to display
# nonrecoverable errors to the user.
@@ -177,7 +190,7 @@ def input_options(options, require=False, prompt=None, fallback_prompt=None,
prompt_part_lengths.append(len(tmpl % str(default)))
else:
prompt_parts.append('# selection')
prompt_part_lengths.append(prompt_parts[-1])
prompt_part_lengths.append(len(prompt_parts[-1]))
prompt_parts += capitalized
prompt_part_lengths += [len(s) for s in options]
@@ -257,10 +270,10 @@ def input_yn(prompt, require=False, color=False):
return sel == 'y'
def config_val(config, section, name, default, vtype=None):
"""Queries the configuration file for a value (given by the
section and name). If no value is present, returns default.
vtype optionally specifies the return type (although only bool
is supported for now).
"""Queries the configuration file for a value (given by the section
and name). If no value is present, returns default. vtype
optionally specifies the return type (although only ``bool`` and
``list`` are supported for now).
"""
if not config.has_section(section):
config.add_section(section)
@@ -268,8 +281,12 @@ def config_val(config, section, name, default, vtype=None):
try:
if vtype is bool:
return config.getboolean(section, name)
elif vtype is list:
# Whitespace-separated strings.
strval = config.get(section, name, True)
return strval.split()
else:
return config.get(section, name)
return config.get(section, name, True)
except ConfigParser.NoOptionError:
return default
@@ -284,7 +301,7 @@ def human_bytes(size):
def human_seconds(interval):
"""Formats interval, a number of seconds, as a human-readable time
interval.
interval using English words.
"""
units = [
(1, 'second'),
@@ -308,6 +325,13 @@ def human_seconds(interval):
return "%3.1f %ss" % (interval, suffix)
def human_seconds_short(interval):
"""Formats a number of seconds as a short human-readable M:SS
string.
"""
interval = int(interval)
return u'%i:%02i' % (interval // 60, interval % 60)
# ANSI terminal colorization code heavily inspired by pygments:
# http://dev.pocoo.org/hg/pygments-main/file/b2deea5b5030/pygments/console.py
# (pygments is by Tim Hatch, Armin Ronacher, et al.)
@@ -369,6 +393,84 @@ def colordiff(a, b, highlight='red'):
return u''.join(a_out), u''.join(b_out)
def default_paths(pathmod=None):
"""Produces the appropriate default config, library, and directory
paths for the current system. On Unix, this is always in ~. On
Windows, tries ~ first and then $APPDATA for the config and library
files (for backwards compatibility).
"""
pathmod = pathmod or os.path
windows = pathmod.__name__ == 'ntpath'
if windows:
windata = os.environ.get('APPDATA') or '~'
# Shorthand for joining paths.
def exp(*vals):
return pathmod.expanduser(pathmod.join(*vals))
config = exp('~', DEFAULT_CONFIG_FILENAME_UNIX)
if windows and not pathmod.exists(config):
config = exp(windata, DEFAULT_CONFIG_FILENAME_WINDOWS)
libpath = exp('~', DEFAULT_LIBRARY_FILENAME_UNIX)
if windows and not pathmod.exists(libpath):
libpath = exp(windata, DEFAULT_LIBRARY_FILENAME_WINDOWS)
libdir = exp('~', DEFAULT_DIRECTORY_NAME)
return config, libpath, libdir
def _get_replacements(config):
"""Given a ConfigParser, get the list of replacement pairs. If no
replacements are specified, returns None. Otherwise, returns a list
of (compiled regex, replacement string) pairs.
"""
repl_string = config_val(config, 'beets', 'replace', None)
if not repl_string:
return
parts = repl_string.strip().split()
if not parts:
return
if len(parts) % 2 != 0:
# Must have an even number of parts.
raise UserError(u'"replace" config value must consist of'
u' pattern/replacement pairs')
out = []
for index in xrange(0, len(parts), 2):
pattern = parts[index]
replacement = parts[index+1]
if replacement.lower() == NULL_REPLACE:
replacement = ''
out.append((re.compile(pattern), replacement))
return out
def _get_path_formats(config):
"""Returns a list of path formats (query/template pairs); reflecting
the config's specified path formats.
"""
legacy_path_format = config_val(config, 'beets', 'path_format', None)
if legacy_path_format:
# Old path formats override the default values.
path_formats = [(library.PF_KEY_DEFAULT, legacy_path_format)]
else:
# If no legacy path format, use the defaults instead.
path_formats = DEFAULT_PATH_FORMATS
if config.has_section('paths'):
custom_path_formats = []
for key, value in config.items('paths', True):
if key in PF_KEY_QUERIES:
# Special values that indicate simple queries.
key = PF_KEY_QUERIES[key]
elif key != library.PF_KEY_DEFAULT:
# For non-special keys (literal queries), the _
# character denotes a :.
key = key.replace('_', ':')
custom_path_formats.append((key, value))
path_formats = custom_path_formats + path_formats
return path_formats
# Subcommand parsing infrastructure.
@@ -536,7 +638,10 @@ class SubcommandsOptionParser(optparse.OptionParser):
def main(args=None, configfh=None):
"""Run the main command-line interface for beets."""
# Get the default subcommands.
from beets.ui.commands import default_commands
from lib.beets.ui.commands import default_commands
# Get default file paths.
default_config, default_libpath, default_dir = default_paths()
# Read defaults from config file.
config = ConfigParser.SafeConfigParser()
@@ -545,7 +650,7 @@ def main(args=None, configfh=None):
elif CONFIG_PATH_VAR in os.environ:
configpath = os.path.expanduser(os.environ[CONFIG_PATH_VAR])
else:
configpath = DEFAULT_CONFIG_FILE
configpath = default_config
if configpath:
configpath = util.syspath(configpath)
if os.path.exists(util.syspath(configpath)):
@@ -574,8 +679,6 @@ def main(args=None, configfh=None):
help='library database file to use')
parser.add_option('-d', '--directory', dest='directory',
help="destination music directory")
parser.add_option('-p', '--pathformat', dest='path_format',
help="destination path format string")
parser.add_option('-v', '--verbose', dest='verbose', action='store_true',
help='print debugging information')
@@ -584,30 +687,26 @@ def main(args=None, configfh=None):
# Open library file.
libpath = options.libpath or \
config_val(config, 'beets', 'library', DEFAULT_LIBRARY)
config_val(config, 'beets', 'library', default_libpath)
directory = options.directory or \
config_val(config, 'beets', 'directory', DEFAULT_DIRECTORY)
legacy_path_format = config_val(config, 'beets', 'path_format', None)
if options.path_format:
# If given, -p overrides all path format settings
path_formats = {'default': options.path_format}
else:
if legacy_path_format:
# Old path formats override the default values.
path_formats = {'default': legacy_path_format}
else:
# If no legacy path format, use the defaults instead.
path_formats = DEFAULT_PATH_FORMATS
if config.has_section('paths'):
path_formats.update(config.items('paths'))
config_val(config, 'beets', 'directory', default_dir)
path_formats = _get_path_formats(config)
art_filename = \
config_val(config, 'beets', 'art_filename', DEFAULT_ART_FILENAME)
lib_timeout = config_val(config, 'beets', 'timeout', DEFAULT_TIMEOUT)
replacements = _get_replacements(config)
try:
lib_timeout = float(lib_timeout)
except ValueError:
lib_timeout = DEFAULT_TIMEOUT
db_path = os.path.expanduser(libpath)
try:
lib = library.Library(db_path,
directory,
path_formats,
art_filename)
art_filename,
lib_timeout,
replacements)
except sqlite3.OperationalError:
raise UserError("database file %s could not be opened" % db_path)
@@ -617,6 +716,9 @@ def main(args=None, configfh=None):
log.setLevel(logging.DEBUG)
else:
log.setLevel(logging.INFO)
log.debug(u'config file: %s' % util.displayable_path(configpath))
log.debug(u'library database: %s' % util.displayable_path(lib.path))
log.debug(u'library directory: %s' % util.displayable_path(lib.directory))
# Invoke the subcommand.
try:

246
lib/beets/ui/commands.py Executable file → Normal file
View File

@@ -29,7 +29,8 @@ from lib.beets import autotag
import lib.beets.autotag.art
from lib.beets import plugins
from lib.beets import importer
from lib.beets.util import syspath, normpath, ancestry
from lib.beets.util import syspath, normpath, ancestry, displayable_path
from lib.beets.util.functemplate import Template
from lib.beets import library
# Global logger.
@@ -97,9 +98,14 @@ DEFAULT_IMPORT_RESUME = None # "ask"
DEFAULT_IMPORT_INCREMENTAL = False
DEFAULT_THREADED = True
DEFAULT_COLOR = True
DEFAULT_IGNORE = [
'.*', '*~',
]
VARIOUS_ARTISTS = u'Various Artists'
PARTIAL_MATCH_MESSAGE = u'(partial match!)'
# Importer utilities and support.
def dist_string(dist, color):
@@ -121,13 +127,29 @@ def show_change(cur_artist, cur_album, items, info, dist, color=True):
tags are changed from (cur_artist, cur_album, items) to info with
distance dist.
"""
def show_album(artist, album):
def show_album(artist, album, partial=False):
if artist:
print_(' %s - %s' % (artist, album))
album_description = u' %s - %s' % (artist, album)
elif album:
print_(' %s' % album)
album_description = u' %s' % album
else:
print_(' (unknown album)')
album_description = u' (unknown album)'
# Add a suffix if this is a partial match.
if partial:
warning = PARTIAL_MATCH_MESSAGE
else:
warning = None
if color and warning:
warning = ui.colorize('yellow', warning)
out = album_description
if warning:
out += u' ' + warning
print_(out)
# Record if the match is partial or not.
partial_match = None in items
# Identify the album in question.
if cur_artist != info.artist or \
@@ -147,17 +169,35 @@ def show_change(cur_artist, cur_album, items, info, dist, color=True):
print_("To:")
show_album(artist_r, album_r)
else:
print_("Tagging: %s - %s" % (info.artist, info.album))
message = u"Tagging: %s - %s" % (info.artist, info.album)
if partial_match:
warning = PARTIAL_MATCH_MESSAGE
if color:
warning = ui.colorize('yellow', PARTIAL_MATCH_MESSAGE)
message += u' ' + warning
print_(message)
# Distance/similarity.
print_('(Similarity: %s)' % dist_string(dist, color))
# Tracks.
missing_tracks = []
for i, (item, track_info) in enumerate(zip(items, info.tracks)):
cur_track = str(item.track)
new_track = str(i+1)
if not item:
missing_tracks.append((i, track_info))
continue
# Get displayable LHS and RHS values.
cur_track = unicode(item.track)
new_track = unicode(i+1)
cur_title = item.title
new_title = track_info.title
if item.length and track_info.length:
cur_length = ui.human_seconds_short(item.length)
new_length = ui.human_seconds_short(track_info.length)
if color:
cur_length = ui.colorize('red', cur_length)
new_length = ui.colorize('red', new_length)
# Possibly colorize changes.
if color:
@@ -168,16 +208,31 @@ def show_change(cur_artist, cur_album, items, info, dist, color=True):
# Show filename (non-colorized) when title is not set.
if not item.title.strip():
cur_title = os.path.basename(item.path)
cur_title = displayable_path(os.path.basename(item.path))
if cur_title != new_title and cur_track != new_track:
print_(" * %s (%s) -> %s (%s)" % (
cur_title, cur_track, new_title, new_track
))
elif cur_title != new_title:
print_(" * %s -> %s" % (cur_title, new_title))
elif cur_track != new_track:
print_(" * %s (%s -> %s)" % (item.title, cur_track, new_track))
if cur_title != new_title:
lhs, rhs = cur_title, new_title
if cur_track != new_track:
lhs += u' (%s)' % cur_track
rhs += u' (%s)' % new_track
print_(u" * %s -> %s" % (lhs, rhs))
else:
line = u' * %s' % item.title
display = False
if cur_track != new_track:
display = True
line += u' (%s -> %s)' % (cur_track, new_track)
if item.length and track_info.length and \
abs(item.length - track_info.length) > 2.0:
display = True
line += u' (%s -> %s)' % (cur_length, new_length)
if display:
print_(line)
for i, track_info in missing_tracks:
line = u' * Missing track: %s (%d)' % (track_info.title, i+1)
if color:
line = ui.colorize('yellow', line)
print_(line)
def show_item_change(item, info, dist, color):
"""Print out the change that would occur by tagging `item` with the
@@ -281,22 +336,21 @@ def choose_candidate(candidates, singleton, rec, color, timid,
(item.artist, item.title))
print_('Candidates:')
for i, (dist, info) in enumerate(candidates):
print_('%i. %s - %s (%s)' % (i+1, info['artist'],
info['title'], dist_string(dist, color)))
print_('%i. %s - %s (%s)' % (i+1, info.artist,
info.title, dist_string(dist, color)))
else:
print_('Finding tags for album "%s - %s".' %
(cur_artist, cur_album))
print_('Candidates:')
for i, (dist, items, info) in enumerate(candidates):
line = '%i. %s - %s' % (i+1, info['artist'],
info['album'])
line = '%i. %s - %s' % (i+1, info.artist, info.album)
# Label and year disambiguation, if available.
label, year = None, None
if 'label' in info:
label = info['label']
if 'year' in info and info['year']:
year = unicode(info['year'])
if info.label:
label = info.label
if info.year:
year = unicode(info.year)
if label and year:
line += u' [%s, %s]' % (label, year)
elif label:
@@ -305,6 +359,14 @@ def choose_candidate(candidates, singleton, rec, color, timid,
line += u' [%s]' % year
line += ' (%s)' % dist_string(dist, color)
# Point out the partial matches.
if None in items:
warning = PARTIAL_MATCH_MESSAGE
if color:
warning = ui.colorize('yellow', warning)
line += u' %s' % warning
print_(line)
# Ask the user for a choice.
@@ -502,11 +564,40 @@ def choose_item(task, config):
assert not isinstance(choice, importer.action)
return choice
def resolve_duplicate(task, config):
"""Decide what to do when a new album or item seems similar to one
that's already in the library.
"""
log.warn("This %s is already in the library!" %
("album" if task.is_album else "item"))
if config.quiet:
# In quiet mode, don't prompt -- just skip.
log.info('Skipping.')
sel = 's'
else:
sel = ui.input_options(
('Skip new', 'Keep both', 'Remove old'),
color=config.color
)
if sel == 's':
# Skip new.
task.set_choice(importer.action.SKIP)
elif sel == 'k':
# Keep both. Do nothing; leave the choice intact.
pass
elif sel == 'r':
# Remove old.
task.remove_duplicates = True
else:
assert False
# The import command.
def import_files(lib, paths, copy, write, autot, logpath, art, threaded,
color, delete, quiet, resume, quiet_fallback, singletons,
timid, query, incremental):
timid, query, incremental, ignore):
"""Import the files in the given list of paths, tagging each leaf
directory as an album. If copy, then the files are copied into
the library folder. If write, then new metadata is written to the
@@ -537,7 +628,11 @@ def import_files(lib, paths, copy, write, autot, logpath, art, threaded,
# Open the log.
if logpath:
logpath = normpath(logpath)
logfile = open(syspath(logpath), 'a')
try:
logfile = open(syspath(logpath), 'a')
except IOError:
raise ui.UserError(u"could not open log file for writing: %s" %
displayable_path(logpath))
print >>logfile, 'import started', time.asctime()
else:
logfile = None
@@ -546,34 +641,38 @@ def import_files(lib, paths, copy, write, autot, logpath, art, threaded,
if resume is None and quiet:
resume = False
# Perform the import.
importer.run_import(
lib = lib,
paths = paths,
resume = resume,
logfile = logfile,
color = color,
quiet = quiet,
quiet_fallback = quiet_fallback,
copy = copy,
write = write,
art = art,
delete = delete,
threaded = threaded,
autot = autot,
choose_match_func = choose_match,
should_resume_func = should_resume,
singletons = singletons,
timid = timid,
choose_item_func = choose_item,
query = query,
incremental = incremental,
)
try:
# Perform the import.
importer.run_import(
lib = lib,
paths = paths,
resume = resume,
logfile = logfile,
color = color,
quiet = quiet,
quiet_fallback = quiet_fallback,
copy = copy,
write = write,
art = art,
delete = delete,
threaded = threaded,
autot = autot,
choose_match_func = choose_match,
should_resume_func = should_resume,
singletons = singletons,
timid = timid,
choose_item_func = choose_item,
query = query,
incremental = incremental,
ignore = ignore,
resolve_duplicate_func = resolve_duplicate,
)
# If we were logging, close the file.
if logfile:
print >>logfile, ''
logfile.close()
finally:
# If we were logging, close the file.
if logfile:
print >>logfile, ''
logfile.close()
# Emit event.
plugins.send('import', lib=lib, paths=paths)
@@ -641,6 +740,7 @@ def import_func(lib, config, opts, args):
incremental = opts.incremental if opts.incremental is not None else \
ui.config_val(config, 'beets', 'import_incremental',
DEFAULT_IMPORT_INCREMENTAL, bool)
ignore = ui.config_val(config, 'beets', 'ignore', DEFAULT_IGNORE, list)
# Resume has three options: yes, no, and "ask" (None).
resume = opts.resume if opts.resume is not None else \
@@ -667,38 +767,48 @@ def import_func(lib, config, opts, args):
import_files(lib, paths, copy, write, autot, logpath, art, threaded,
color, delete, quiet, resume, quiet_fallback, singletons,
timid, query, incremental)
timid, query, incremental, ignore)
import_cmd.func = import_func
default_commands.append(import_cmd)
# list: Query and show library contents.
def list_items(lib, query, album, path):
def list_items(lib, query, album, path, fmt):
"""Print out items in lib matching query. If album, then search for
albums instead of single items. If path, print the matched objects'
paths instead of human-readable information about them.
"""
if fmt is None:
# If no specific template is supplied, use a default.
if album:
fmt = u'$albumartist - $album'
else:
fmt = u'$artist - $album - $title'
template = Template(fmt)
if album:
for album in lib.albums(query):
if path:
print_(album.item_dir())
else:
print_(album.albumartist + u' - ' + album.album)
elif fmt is not None:
print_(template.substitute(album._record))
else:
for item in lib.items(query):
if path:
print_(item.path)
else:
print_(item.artist + u' - ' + item.album + u' - ' + item.title)
elif fmt is not None:
print_(template.substitute(item.record))
list_cmd = ui.Subcommand('list', help='query the library', aliases=('ls',))
list_cmd.parser.add_option('-a', '--album', action='store_true',
help='show matching albums instead of tracks')
list_cmd.parser.add_option('-p', '--path', action='store_true',
help='print paths for matched items or albums')
list_cmd.parser.add_option('-f', '--format', action='store',
help='print with custom format', default=None)
def list_func(lib, config, opts, args):
list_items(lib, decargs(args), opts.album, opts.path)
list_items(lib, decargs(args), opts.album, opts.path, opts.format)
list_cmd.func = list_func
default_commands.append(list_cmd)
@@ -722,6 +832,12 @@ def update_items(lib, query, album, move, color, pretend):
affected_albums.add(item.album_id)
continue
# Did the item change since last checked?
if item.current_mtime() <= item.mtime:
log.debug(u'skipping %s because mtime is up to date (%i)' %
(displayable_path(item.path), item.mtime))
continue
# Read new data.
old_data = dict(item.record)
item.read()
@@ -755,6 +871,12 @@ def update_items(lib, query, album, move, color, pretend):
lib.store(item)
affected_albums.add(item.album_id)
elif not pretend:
# The file's mtime was different, but there were no changes
# to the metadata. Store the new mtime, which is set in the
# call to read(), so we don't check this again in the
# future.
lib.store(item)
# Skip album changes while pretending.
if pretend:
@@ -885,7 +1007,7 @@ default_commands.append(stats_cmd)
# version: Show current beets version.
def show_version(lib, config, opts, args):
print 'beets version %s' % beets.__version__
print 'beets version %s' % lib.beets.__version__
# Show plugins.
names = []
for plugin in plugins.find_plugins():

156
lib/beets/util/__init__.py
View File

@@ -17,6 +17,7 @@ import os
import sys
import re
import shutil
import fnmatch
from collections import defaultdict
MAX_FILENAME_LENGTH = 200
@@ -47,9 +48,10 @@ def ancestry(path, pathmod=None):
out.insert(0, path)
return out
def sorted_walk(path):
"""Like os.walk, but yields things in sorted, breadth-first
order.
def sorted_walk(path, ignore=()):
"""Like ``os.walk``, but yields things in sorted, breadth-first
order. Directory and file names matching any glob pattern in
``ignore`` are skipped.
"""
# Make sure the path isn't a Unicode string.
path = bytestring_path(path)
@@ -58,6 +60,16 @@ def sorted_walk(path):
dirs = []
files = []
for base in os.listdir(path):
# Skip ignored filenames.
skip = False
for pat in ignore:
if fnmatch.fnmatch(base, pat):
skip = True
break
if skip:
continue
# Add to output as either a file or a directory.
cur = os.path.join(path, base)
if os.path.isdir(syspath(cur)):
dirs.append(base)
@@ -73,7 +85,7 @@ def sorted_walk(path):
for base in dirs:
cur = os.path.join(path, base)
# yield from _sorted_walk(cur)
for res in sorted_walk(cur):
for res in sorted_walk(cur, ignore):
yield res
def mkdirall(path):
@@ -84,38 +96,46 @@ def mkdirall(path):
if not os.path.isdir(syspath(ancestor)):
os.mkdir(syspath(ancestor))
def prune_dirs(path, root, clutter=('.DS_Store', 'Thumbs.db')):
"""If path is an empty directory, then remove it. Recursively
remove path's ancestry up to root (which is never removed) where
there are empty directories. If path is not contained in root, then
nothing is removed. Filenames in clutter are ignored when
determining emptiness.
def prune_dirs(path, root=None, clutter=('.DS_Store', 'Thumbs.db')):
"""If path is an empty directory, then remove it. Recursively remove
path's ancestry up to root (which is never removed) where there are
empty directories. If path is not contained in root, then nothing is
removed. Filenames in clutter are ignored when determining
emptiness. If root is not provided, then only path may be removed
(i.e., no recursive removal).
"""
path = normpath(path)
root = normpath(root)
if root is not None:
root = normpath(root)
ancestors = ancestry(path)
if root in ancestors:
if root is None:
# Only remove the top directory.
ancestors = []
elif root in ancestors:
# Only remove directories below the root.
ancestors = ancestors[ancestors.index(root)+1:]
else:
# Remove nothing.
return
# Traverse upward from path.
ancestors.append(path)
ancestors.reverse()
for directory in ancestors:
directory = syspath(directory)
if not os.path.exists(directory):
# Directory gone already.
continue
# Traverse upward from path.
ancestors.append(path)
ancestors.reverse()
for directory in ancestors:
directory = syspath(directory)
if not os.path.exists(directory):
# Directory gone already.
continue
if all(fn in clutter for fn in os.listdir(directory)):
# Directory contains only clutter (or nothing).
try:
shutil.rmtree(directory)
except OSError:
break
else:
if all(fn in clutter for fn in os.listdir(directory)):
# Directory contains only clutter (or nothing).
try:
shutil.rmtree(directory)
except OSError:
break
else:
break
def components(path, pathmod=None):
"""Return a list of the path components in path. For instance:
@@ -150,9 +170,25 @@ def bytestring_path(path):
encoding = sys.getfilesystemencoding() or sys.getdefaultencoding()
try:
return path.encode(encoding)
except UnicodeError:
except (UnicodeError, LookupError):
return path.encode('utf8')
def displayable_path(path):
"""Attempts to decode a bytestring path to a unicode object for the
purpose of displaying it to the user.
"""
if isinstance(path, unicode):
return path
elif not isinstance(path, str):
# A non-string object: just get its unicode representation.
return unicode(path)
encoding = sys.getfilesystemencoding() or sys.getdefaultencoding()
try:
return path.decode(encoding, 'ignore')
except (UnicodeError, LookupError):
return path.decode('utf8', 'ignore')
def syspath(path, pathmod=None):
"""Convert a path for use by the operating system. In particular,
paths on Windows must receive a magic prefix and must be converted
@@ -220,31 +256,60 @@ def move(path, dest, replace=False, pathmod=None):
_assert_not_exists(dest, pathmod)
return shutil.move(path, dest)
def unique_path(path):
"""Returns a version of ``path`` that does not exist on the
filesystem. Specifically, if ``path` itself already exists, then
something unique is appended to the path.
"""
if not os.path.exists(syspath(path)):
return path
base, ext = os.path.splitext(path)
match = re.search(r'\.(\d)+$', base)
if match:
num = int(match.group(1))
base = base[:match.start()]
else:
num = 0
while True:
num += 1
new_path = '%s.%i%s' % (base, num, ext)
if not os.path.exists(new_path):
return new_path
# Note: POSIX actually supports \ and : -- I just think they're
# a pain. And ? has caused problems for some.
CHAR_REPLACE = [
(re.compile(r'[\\/\?]|^\.'), '_'),
(re.compile(r'[\\/\?"]|^\.'), '_'),
(re.compile(r':'), '-'),
]
CHAR_REPLACE_WINDOWS = re.compile(r'["\*<>\|]|^\.|\.$| +$'), '_'
def sanitize_path(path, pathmod=None):
CHAR_REPLACE_WINDOWS = [
(re.compile(r'["\*<>\|]|^\.|\.$|\s+$'), '_'),
]
def sanitize_path(path, pathmod=None, replacements=None):
"""Takes a path and makes sure that it is legal. Returns a new path.
Only works with fragments; won't work reliably on Windows when a
path begins with a drive letter. Path separators (including altsep!)
should already be cleaned from the path components.
should already be cleaned from the path components. If replacements
is specified, it is used *instead* of the default set of
replacements for the platform; it must be a list of (compiled regex,
replacement string) pairs.
"""
pathmod = pathmod or os.path
windows = pathmod.__name__ == 'ntpath'
# Choose the appropriate replacements.
if not replacements:
replacements = list(CHAR_REPLACE)
if windows:
replacements += CHAR_REPLACE_WINDOWS
comps = components(path, pathmod)
if not comps:
return ''
for i, comp in enumerate(comps):
# Replace special characters.
for regex, repl in CHAR_REPLACE:
comp = regex.sub(repl, comp)
if windows:
regex, repl = CHAR_REPLACE_WINDOWS
for regex, repl in replacements:
comp = regex.sub(repl, comp)
# Truncate each component.
@@ -263,11 +328,18 @@ def sanitize_for_path(value, pathmod, key=None):
if sep:
value = value.replace(sep, u'_')
elif key in ('track', 'tracktotal', 'disc', 'disctotal'):
# pad with zeros
value = u'%02i' % value
# Pad indices with zeros.
value = u'%02i' % (value or 0)
elif key == 'year':
value = u'%04i' % (value or 0)
elif key in ('month', 'day'):
value = u'%02i' % (value or 0)
elif key == 'bitrate':
# Bitrate gets formatted as kbps.
value = u'%ikbps' % (value / 1000)
value = u'%ikbps' % ((value or 0) / 1000)
elif key == 'samplerate':
# Sample rate formatted as kHz.
value = u'%ikHz' % ((value or 0) / 1000)
else:
value = unicode(value)
return value
@@ -303,13 +375,17 @@ def levenshtein(s1, s2):
def plurality(objs):
"""Given a sequence of comparable objects, returns the object that
is most common in the set and the frequency of that object.
is most common in the set and the frequency of that object. The
sequence must contain at least one object.
"""
# Calculate frequencies.
freqs = defaultdict(int)
for obj in objs:
freqs[obj] += 1
if not freqs:
raise ValueError('sequence must be non-empty')
# Find object with maximum frequency.
max_freq = 0
res = None

353
lib/beets/util/functemplate.py Normal file
View File

@@ -0,0 +1,353 @@
# This file is part of beets.
# Copyright 2011, Adrian Sampson.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
"""This module implements a string formatter based on the standard PEP
292 string.Template class extended with function calls. Variables, as
with string.Template, are indicated with $ and functions are delimited
with %.
This module assumes that everything is Unicode: the template and the
substitution values. Bytestrings are not supported. Also, the templates
always behave like the ``safe_substitute`` method in the standard
library: unknown symbols are left intact.
This is sort of like a tiny, horrible degeneration of a real templating
engine like Jinja2 or Mustache.
"""
import re
SYMBOL_DELIM = u'$'
FUNC_DELIM = u'%'
GROUP_OPEN = u'{'
GROUP_CLOSE = u'}'
ARG_SEP = u','
ESCAPE_CHAR = u'$'
class Environment(object):
"""Contains the values and functions to be substituted into a
template.
"""
def __init__(self, values, functions):
self.values = values
self.functions = functions
class Symbol(object):
"""A variable-substitution symbol in a template."""
def __init__(self, ident, original):
self.ident = ident
self.original = original
def __repr__(self):
return u'Symbol(%s)' % repr(self.ident)
def evaluate(self, env):
"""Evaluate the symbol in the environment, returning a Unicode
string.
"""
if self.ident in env.values:
# Substitute for a value.
return env.values[self.ident]
else:
# Keep original text.
return self.original
class Call(object):
"""A function call in a template."""
def __init__(self, ident, args, original):
self.ident = ident
self.args = args
self.original = original
def __repr__(self):
return u'Call(%s, %s, %s)' % (repr(self.ident), repr(self.args),
repr(self.original))
def evaluate(self, env):
"""Evaluate the function call in the environment, returning a
Unicode string.
"""
if self.ident in env.functions:
arg_vals = [expr.evaluate(env) for expr in self.args]
try:
out = env.functions[self.ident](*arg_vals)
except Exception, exc:
# Function raised exception! Maybe inlining the name of
# the exception will help debug.
return u'<%s>' % unicode(exc)
return unicode(out)
else:
return self.original
class Expression(object):
"""Top-level template construct: contains a list of text blobs,
Symbols, and Calls.
"""
def __init__(self, parts):
self.parts = parts
def __repr__(self):
return u'Expression(%s)' % (repr(self.parts))
def evaluate(self, env):
"""Evaluate the entire expression in the environment, returning
a Unicode string.
"""
out = []
for part in self.parts:
if isinstance(part, basestring):
out.append(part)
else:
out.append(part.evaluate(env))
return u''.join(map(unicode, out))
class ParseError(Exception):
pass
class Parser(object):
"""Parses a template expression string. Instantiate the class with
the template source and call ``parse_expression``. The ``pos`` field
will indicate the character after the expression finished and
``parts`` will contain a list of Unicode strings, Symbols, and Calls
reflecting the concatenated portions of the expression.
This is a terrible, ad-hoc parser implementation based on a
left-to-right scan with no lexing step to speak of; it's probably
both inefficient and incorrect. Maybe this should eventually be
replaced with a real, accepted parsing technique (PEG, parser
generator, etc.).
"""
def __init__(self, string):
self.string = string
self.pos = 0
self.parts = []
# Common parsing resources.
special_chars = (SYMBOL_DELIM, FUNC_DELIM, GROUP_OPEN, GROUP_CLOSE,
ARG_SEP, ESCAPE_CHAR)
special_char_re = re.compile(ur'[%s]|$' %
u''.join(re.escape(c) for c in special_chars))
def parse_expression(self):
"""Parse a template expression starting at ``pos``. Resulting
components (Unicode strings, Symbols, and Calls) are added to
the ``parts`` field, a list. The ``pos`` field is updated to be
the next character after the expression.
"""
text_parts = []
while self.pos < len(self.string):
char = self.string[self.pos]
if char not in self.special_chars:
# A non-special character. Skip to the next special
# character, treating the interstice as literal text.
next_pos = (
self.special_char_re.search(self.string[self.pos:]).start()
+ self.pos
)
text_parts.append(self.string[self.pos:next_pos])
self.pos = next_pos
continue
if self.pos == len(self.string) - 1:
# The last character can never begin a structure, so we
# just interpret it as a literal character (unless it
# terminates the expression, as with , and }).
if char not in (GROUP_CLOSE, ARG_SEP):
text_parts.append(char)
self.pos += 1
break
next_char = self.string[self.pos + 1]
if char == ESCAPE_CHAR and next_char in \
(SYMBOL_DELIM, FUNC_DELIM, GROUP_CLOSE, ARG_SEP):
# An escaped special character ($$, $}, etc.). Note that
# ${ is not an escape sequence: this is ambiguous with
# the start of a symbol and it's not necessary (just
# using { suffices in all cases).
text_parts.append(next_char)
self.pos += 2 # Skip the next character.
continue
# Shift all characters collected so far into a single string.
if text_parts:
self.parts.append(u''.join(text_parts))
text_parts = []
if char == SYMBOL_DELIM:
# Parse a symbol.
self.parse_symbol()
elif char == FUNC_DELIM:
# Parse a function call.
self.parse_call()
elif char in (GROUP_CLOSE, ARG_SEP):
# Template terminated.
break
elif char == GROUP_OPEN:
# Start of a group has no meaning hear; just pass
# through the character.
text_parts.append(char)
self.pos += 1
else:
assert False
# If any parsed characters remain, shift them into a string.
if text_parts:
self.parts.append(u''.join(text_parts))
def parse_symbol(self):
"""Parse a variable reference (like ``$foo`` or ``${foo}``)
starting at ``pos``. Possibly appends a Symbol object (or,
failing that, text) to the ``parts`` field and updates ``pos``.
The character at ``pos`` must, as a precondition, be ``$``.
"""
assert self.pos < len(self.string)
assert self.string[self.pos] == SYMBOL_DELIM
if self.pos == len(self.string) - 1:
# Last character.
self.parts.append(SYMBOL_DELIM)
self.pos += 1
return
next_char = self.string[self.pos + 1]
start_pos = self.pos
self.pos += 1
if next_char == GROUP_OPEN:
# A symbol like ${this}.
self.pos += 1 # Skip opening.
closer = self.string.find(GROUP_CLOSE, self.pos)
if closer == -1 or closer == self.pos:
# No closing brace found or identifier is empty.
self.parts.append(self.string[start_pos:self.pos])
else:
# Closer found.
ident = self.string[self.pos:closer]
self.pos = closer + 1
self.parts.append(Symbol(ident,
self.string[start_pos:self.pos]))
else:
# A bare-word symbol.
ident = self._parse_ident()
if ident:
# Found a real symbol.
self.parts.append(Symbol(ident,
self.string[start_pos:self.pos]))
else:
# A standalone $.
self.parts.append(SYMBOL_DELIM)
def parse_call(self):
"""Parse a function call (like ``%foo{bar,baz}``) starting at
``pos``. Possibly appends a Call object to ``parts`` and update
``pos``. The character at ``pos`` must be ``%``.
"""
assert self.pos < len(self.string)
assert self.string[self.pos] == FUNC_DELIM
start_pos = self.pos
self.pos += 1
ident = self._parse_ident()
if not ident:
# No function name.
self.parts.append(FUNC_DELIM)
return
if self.pos >= len(self.string):
# Identifier terminates string.
self.parts.append(self.string[start_pos:self.pos])
return
if self.string[self.pos] != GROUP_OPEN:
# Argument list not opened.
self.parts.append(self.string[start_pos:self.pos])
return
# Skip past opening brace and try to parse an argument list.
self.pos += 1
args = self.parse_argument_list()
if self.pos >= len(self.string) or \
self.string[self.pos] != GROUP_CLOSE:
# Arguments unclosed.
self.parts.append(self.string[start_pos:self.pos])
return
self.pos += 1 # Move past closing brace.
self.parts.append(Call(ident, args, self.string[start_pos:self.pos]))
def parse_argument_list(self):
"""Parse a list of arguments starting at ``pos``, returning a
list of Expression objects. Does not modify ``parts``. Should
leave ``pos`` pointing to a } character or the end of the
string.
"""
# Try to parse a subexpression in a subparser.
expressions = []
while self.pos < len(self.string):
subparser = Parser(self.string[self.pos:])
subparser.parse_expression()
# Extract and advance past the parsed expression.
expressions.append(Expression(subparser.parts))
self.pos += subparser.pos
if self.pos >= len(self.string) or \
self.string[self.pos] == GROUP_CLOSE:
# Argument list terminated by EOF or closing brace.
break
# Only other way to terminate an expression is with ,.
# Continue to the next argument.
assert self.string[self.pos] == ARG_SEP
self.pos += 1
return expressions
def _parse_ident(self):
"""Parse an identifier and return it (possibly an empty string).
Updates ``pos``.
"""
remainder = self.string[self.pos:]
ident = re.match(ur'\w*', remainder).group(0)
self.pos += len(ident)
return ident
def _parse(template):
"""Parse a top-level template string Expression. Any extraneous text
is considered literal text.
"""
parser = Parser(template)
parser.parse_expression()
parts = parser.parts
remainder = parser.string[parser.pos:]
if remainder:
parts.append(remainder)
return Expression(parts)
class Template(object):
"""A string template, including text, Symbols, and Calls.
"""
def __init__(self, template):
self.expr = _parse(template)
self.original = template
def substitute(self, values={}, functions={}):
"""Evaluate the template given the values and functions.
"""
return self.expr.evaluate(Environment(values, functions))

4
lib/beets/util/pipeline.py
View File

@@ -157,9 +157,9 @@ class PipelineThread(Thread):
# Ensure that we are not blocking on a queue read or write.
if hasattr(self, 'in_queue'):
_invalidate_queue(self.in_queue)
_invalidate_queue(self.in_queue, POISON)
if hasattr(self, 'out_queue'):
_invalidate_queue(self.out_queue)
_invalidate_queue(self.out_queue, POISON)
def abort_all(self, exc_info):
"""Abort all other threads in the system for an exception.

1
lib/musicbrainzngs/__init__.py Normal file
View File

@@ -0,0 +1 @@
from musicbrainz import *

85
lib/beets/autotag/musicbrainz3/mbxml.py → lib/musicbrainzngs/mbxml.py
View File

@@ -1,3 +1,8 @@
# This file is part of the musicbrainzngs library
# Copyright (C) Alastair Porter, Adrian Sampson, and others
# This file is distributed under a BSD-2-Clause type license.
# See the COPYING file for more information.
import xml.etree.ElementTree as ET
import string
import StringIO
@@ -24,7 +29,9 @@ except:
xmlns = None
return "%s:%s" % (prefix, tag), xmlns
NS_MAP = {"http://musicbrainz.org/ns/mmd-2.0#": "ws2"}
NS_MAP = {"http://musicbrainz.org/ns/mmd-2.0#": "ws2",
"http://musicbrainz.org/ns/ext#-2.0": "ext"}
_log = logging.getLogger("python-musicbrainz-ngs")
def make_artist_credit(artists):
names = []
@@ -52,23 +59,28 @@ def parse_elements(valid_els, element):
if t in valid_els:
result[t] = sub.text
else:
logging.debug("in <%s>, uncaught <%s>", fixtag(element.tag, NS_MAP)[0], t)
_log.debug("in <%s>, uncaught <%s>", fixtag(element.tag, NS_MAP)[0], t)
return result
def parse_attributes(attributes, element):
""" Extract attributes from an element.
For example, given the element:
<element type="Group" />
and a list attributes that contains "type",
return a dict {'type': 'Group'}
"""
result = {}
for attr in attributes:
if attr in element.attrib:
result[attr] = element.attrib[attr]
else:
logging.debug("in <%s>, uncaught attribute %s", fixtag(element.tag, NS_MAP)[0], attr)
return result
""" Extract attributes from an element.
For example, given the element:
<element type="Group" />
and a list attributes that contains "type",
return a dict {'type': 'Group'}
"""
result = {}
for attr in element.attrib:
if "{" in attr:
a = fixtag(attr, NS_MAP)[0]
else:
a = attr
if a in attributes:
result[a] = element.attrib[attr]
else:
_log.debug("in <%s>, uncaught attribute %s", fixtag(element.tag, NS_MAP)[0], attr)
return result
def parse_inner(inner_els, element):
""" Delegate the parsing of a subelement to another function.
@@ -97,7 +109,7 @@ def parse_inner(inner_els, element):
else:
result[t] = inner_result
else:
logging.debug("in <%s>, not delegating <%s>", fixtag(element.tag, NS_MAP)[0], t)
_log.debug("in <%s>, not delegating <%s>", fixtag(element.tag, NS_MAP)[0], t)
return result
def parse_message(message):
@@ -123,7 +135,7 @@ def parse_message(message):
"release-group-list": parse_release_group_list,
"recording-list": parse_recording_list,
"work-list": parse_work_list,
"collection-list": parse_collection_list,
"collection": parse_collection,
@@ -155,18 +167,16 @@ def parse_collection_release_list(rl):
def parse_artist_lifespan(lifespan):
parts = parse_elements(["begin", "end"], lifespan)
beginval = parts.get("begin", "")
endval = parts.get("end", "")
return (beginval, endval)
return parts
def parse_artist_list(al):
return [parse_artist(a) for a in al]
def parse_artist(artist):
result = {}
attribs = ["id", "type"]
elements = ["name", "sort-name", "country", "user-rating"]
attribs = ["id", "type", "ext:score"]
elements = ["name", "sort-name", "country", "user-rating", "disambiguation"]
inner_els = {"life-span": parse_artist_lifespan,
"recording-list": parse_recording_list,
"release-list": parse_release_list,
@@ -188,7 +198,7 @@ def parse_label_list(ll):
def parse_label(label):
result = {}
attribs = ["id", "type"]
attribs = ["id", "type", "ext:score"]
elements = ["name", "sort-name", "country", "label-code", "user-rating"]
inner_els = {"life-span": parse_artist_lifespan,
"release-list": parse_release_list,
@@ -235,7 +245,7 @@ def parse_relation(relation):
def parse_release(release):
result = {}
attribs = ["id"]
attribs = ["id", "ext:score"]
elements = ["title", "status", "disambiguation", "quality", "country", "barcode", "date", "packaging", "asin"]
inner_els = {"text-representation": parse_text_representation,
"artist-credit": parse_artist_credit,
@@ -273,7 +283,7 @@ def parse_text_representation(textr):
def parse_release_group(rg):
result = {}
attribs = ["id", "type"]
attribs = ["id", "type", "ext:score"]
elements = ["title", "user-rating", "first-release-date"]
inner_els = {"artist-credit": parse_artist_credit,
"release-list": parse_release_list,
@@ -291,7 +301,7 @@ def parse_release_group(rg):
def parse_recording(recording):
result = {}
attribs = ["id"]
attribs = ["id", "ext:score"]
elements = ["title", "length", "user-rating"]
inner_els = {"artist-credit": parse_artist_credit,
"release-list": parse_release_list,
@@ -321,7 +331,7 @@ def parse_work_list(wl):
def parse_work(work):
result = {}
attribs = ["id"]
attribs = ["id", "ext:score"]
elements = ["title", "user-rating"]
inner_els = {"tag-list": parse_tag_list,
"user-tag-list": parse_tag_list,
@@ -417,7 +427,7 @@ def parse_track_list(tl):
def parse_track(track):
result = {}
elements = ["position"]
elements = ["position", "title"]
inner_els = {"recording": parse_recording}
result.update(parse_elements(elements, track))
@@ -537,9 +547,24 @@ def make_rating_request(artist_ratings, recording_ratings):
for art, rating in artist_ratings.items():
art_xml = ET.SubElement(art_list, "{%s}artist" % NS)
art_xml.set("{%s}id" % NS, art)
rating_xml = ET.SubElement(rec_xml, "{%s}user-rating" % NS)
rating_xml = ET.SubElement(art_xml, "{%s}user-rating" % NS)
if isinstance(rating, int):
rating = "%d" % rating
rating_xml.text = rating
return ET.tostring(root, "utf-8")
def make_isrc_request(recordings_isrcs):
NS = "http://musicbrainz.org/ns/mmd-2.0#"
root = ET.Element("{%s}metadata" % NS)
rec_list = ET.SubElement(root, "{%s}recording-list" % NS)
for rec, isrcs in recordings_isrcs.items():
if len(isrcs) > 0:
rec_xml = ET.SubElement(rec_list, "{%s}recording" % NS)
rec_xml.set("{%s}id" % NS, rec)
isrc_list_xml = ET.SubElement(rec_xml, "{%s}isrc-list" % NS)
isrc_list_xml.set("{%s}count" % NS, str(len(isrcs)))
for isrc in isrcs:
isrc_xml = ET.SubElement(isrc_list_xml, "{%s}isrc" % NS)
isrc_xml.set("{%s}id" % NS, isrc)
return ET.tostring(root, "utf-8")

237
lib/beets/autotag/musicbrainz3/__init__.py → lib/musicbrainzngs/musicbrainz.py
View File

@@ -1,22 +1,23 @@
# This is a copy of changeset e60b5af77 from the python-musicbrainz-ngs
# project:
# https://github.com/alastair/python-musicbrainz-ngs/
# MIT license; by Alastair Porter and Adrian Sampson
# This file is part of the musicbrainzngs library
# Copyright (C) Alastair Porter, Adrian Sampson, and others
# This file is distributed under a BSD-2-Clause type license.
# See the COPYING file for more information.
import urlparse
import urllib2
import urllib
import mbxml
import re
import threading
import time
import logging
import httplib
import socket
import xml.etree.ElementTree as etree
from xml.parsers import expat
from . import mbxml
_useragent = "pythonmusicbrainzngs-0.1"
_log = logging.getLogger("python-musicbrainz-ngs")
_version = "0.3dev"
_log = logging.getLogger("musicbrainzngs")
# Constants for validation.
@@ -28,7 +29,7 @@ VALID_INCLUDES = {
"aliases", "tags", "user-tags", "ratings", "user-ratings", # misc
"artist-rels", "label-rels", "recording-rels", "release-rels",
"release-group-rels", "url-rels", "work-rels"
],
],
'label': [
"releases", # Subqueries
"discids", "media",
@@ -63,8 +64,11 @@ VALID_INCLUDES = {
"release-group-rels", "url-rels", "work-rels"
],
'discid': [
"artists", "labels", "recordings", "release-groups", "puids",
"echoprints", "isrcs"
"artists", "labels", "recordings", "release-groups", "media",
"artist-credits", "discids", "puids", "echoprints", "isrcs",
"artist-rels", "label-rels", "recording-rels", "release-rels",
"release-group-rels", "url-rels", "work-rels", "recording-level-rels",
"work-level-rels"
],
'echoprint': ["artists", "releases"],
'puid': ["artists", "releases", "puids", "echoprints", "isrcs"],
@@ -145,7 +149,7 @@ class WebServiceError(MusicBrainzError):
"""
self.message = message
self.cause = cause
def __str__(self):
if self.message:
msg = "%s, " % self.message
@@ -177,7 +181,7 @@ def _check_filter(values, valid):
if v not in valid:
raise InvalidFilterError(v)
def _check_filter_and_make_params(includes, release_status=[], release_type=[]):
def _check_filter_and_make_params(entity, includes, release_status=[], release_type=[]):
"""Check that the status or type values are valid. Then, check that
the filters can be used with the given includes. Return a params
dict that can be passed to _do_mb_query.
@@ -192,7 +196,8 @@ def _check_filter_and_make_params(includes, release_status=[], release_type=[]):
if release_status and "releases" not in includes:
raise InvalidFilterError("Can't have a status with no release include")
if release_type and ("release-groups" not in includes and
"releases" not in includes):
"releases" not in includes and
entity != "release-group"):
raise InvalidFilterError("Can't have a release type with no "
"release-group include")
@@ -210,6 +215,7 @@ def _check_filter_and_make_params(includes, release_status=[], release_type=[]):
user = password = ""
hostname = "musicbrainz.org"
_client = ""
_useragent = ""
def auth(u, p):
"""Set the username and password to be used in subsequent queries to
@@ -219,13 +225,20 @@ def auth(u, p):
user = u
password = p
def set_client(c):
""" Set the client to be used in requests. This must be set before any
data submissions are made.
"""
global _client
_client = c
def set_useragent(app, version, contact=None):
""" Set the User-Agent to be used for requests to the MusicBrainz webservice.
This should be set before requests are made."""
global _useragent, _client
if contact is not None:
_useragent = "%s/%s python-musicbrainz-ngs/%s ( %s )" % (app, version, _version, contact)
else:
_useragent = "%s/%s python-musicbrainz-ngs/%s" % (app, version, _version)
_client = "%s-%s" % (app, version)
_log.debug("set user-agent to %s" % _useragent)
def set_hostname(new_hostname):
global hostname
hostname = new_hostname
# Rate limiting.
@@ -368,16 +381,27 @@ def _safe_open(opener, req, body=None, max_retries=8, retry_delay_delta=2.0):
_log.debug("miscellaneous HTTP exception: %s" % str(exc))
last_exc = exc
except urllib2.URLError, exc:
if isinstance(exc.reason, socket.error):
code = exc.reason.errno
if code == 104: # "Connection reset by peer."
continue
raise NetworkError(cause=exc)
except IOError, exc:
raise NetworkError(cause=exc)
else:
# No exception! Yay!
return f
# Out of retries!
raise NetworkError("retried %i times" % max_retries, last_exc)
# Get the XML parsing exceptions to catch. The behavior chnaged with Python 2.7
# and ElementTree 1.3.
if hasattr(etree, 'ParseError'):
ETREE_EXCEPTIONS = (etree.ParseError, expat.ExpatError)
else:
ETREE_EXCEPTIONS = (expat.ExpatError)
@_rate_limit
def _mb_request(path, method='GET', auth_required=False, client_required=False,
args=None, data=None, body=None):
@@ -387,15 +411,23 @@ def _mb_request(path, method='GET', auth_required=False, client_required=False,
whether exceptions should be raised if the client and
username/password are left unspecified, respectively.
"""
args = dict(args) or {}
if args is None:
args = {}
else:
args = dict(args) or {}
# Add client if required.
if client_required and _client == "":
raise UsageError("set a client name with "
"musicbrainz.set_client(\"client-version\")")
elif client_required:
if _useragent == "":
raise UsageError("set a proper user-agent with "
"set_useragent(\"application name\", \"application version\", \"contact info (preferably URL or email for your application)\")")
if client_required:
args["client"] = _client
# Encode Unicode arguments using UTF-8.
for key, value in args.items():
if isinstance(value, unicode):
args[key] = value.encode('utf8')
# Construct the full URL for the request, including hostname and
# query string.
url = urlparse.urlunparse((
@@ -407,25 +439,28 @@ def _mb_request(path, method='GET', auth_required=False, client_required=False,
''
))
_log.debug("%s request for %s" % (method, url))
# Set up HTTP request handler and URL opener.
httpHandler = urllib2.HTTPHandler(debuglevel=0)
handlers = [httpHandler]
opener = urllib2.build_opener(*handlers)
# Add credentials if required.
if auth_required:
_log.debug("Auth required for %s" % url)
if not user:
raise UsageError("authorization required; "
"use musicbrainz.auth(u, p) first")
"use auth(user, pass) first")
passwordMgr = _RedirectPasswordMgr()
authHandler = _DigestAuthHandler(passwordMgr)
authHandler.add_password("musicbrainz.org", (), user, password)
handlers.append(authHandler)
opener = urllib2.build_opener(*handlers)
# Make request.
req = _MusicbrainzHttpRequest(method, url, data)
req.add_header('User-Agent', _useragent)
_log.debug("requesting with UA %s" % _useragent)
if body:
req.add_header('Content-Type', 'application/xml; charset=UTF-8')
f = _safe_open(opener, req, body)
@@ -433,10 +468,13 @@ def _mb_request(path, method='GET', auth_required=False, client_required=False,
# Parse the response.
try:
return mbxml.parse_message(f)
except etree.ParseError, exc:
raise ResponseError(cause=exc)
except UnicodeError, exc:
except UnicodeError as exc:
raise ResponseError(cause=exc)
except Exception as exc:
if isinstance(exc, ETREE_EXCEPTIONS):
raise ResponseError(cause=exc)
else:
raise
def _is_auth_required(entity, includes):
""" Some calls require authentication. This returns
@@ -487,6 +525,7 @@ def _do_mb_search(entity, query='', fields={}, limit=None, offset=None):
# Escape Lucene's special characters.
value = re.sub(r'([+\-&|!(){}\[\]\^"~*?:\\])', r'\\\1', value)
value = value.replace('\x00', '').strip()
value = value.lower() # Avoid binary operators like OR.
if value:
query_parts.append(u'%s:(%s)' % (key, value))
full_query = u' '.join(query_parts).strip()
@@ -516,7 +555,7 @@ def _do_mb_post(path, body):
"""Perform a single POST call for an endpoint with a specified
request body.
"""
return _mb_request(path, 'PUT', True, True, body=body)
return _mb_request(path, 'POST', True, True, body=body)
# The main interface!
@@ -548,7 +587,7 @@ def get_work_by_id(id, includes=[]):
# Searching
def artist_search(query='', limit=None, offset=None, **fields):
def search_artists(query='', limit=None, offset=None, **fields):
"""Search for artists by a free-form `query` string and/or any of
the following keyword arguments specifying field queries:
arid, artist, sortname, type, begin, end, comment, alias, country,
@@ -556,7 +595,7 @@ def artist_search(query='', limit=None, offset=None, **fields):
"""
return _do_mb_search('artist', query, fields, limit, offset)
def label_search(query='', limit=None, offset=None, **fields):
def search_labels(query='', limit=None, offset=None, **fields):
"""Search for labels by a free-form `query` string and/or any of
the following keyword arguments specifying field queries:
laid, label, sortname, type, code, country, begin, end, comment,
@@ -564,7 +603,7 @@ def label_search(query='', limit=None, offset=None, **fields):
"""
return _do_mb_search('label', query, fields, limit, offset)
def recording_search(query='', limit=None, offset=None, **fields):
def search_recordings(query='', limit=None, offset=None, **fields):
"""Search for recordings by a free-form `query` string and/or any of
the following keyword arguments specifying field queries:
rid, recording, isrc, arid, artist, artistname, creditname, reid,
@@ -573,7 +612,7 @@ def recording_search(query='', limit=None, offset=None, **fields):
"""
return _do_mb_search('recording', query, fields, limit, offset)
def release_search(query='', limit=None, offset=None, **fields):
def search_releases(query='', limit=None, offset=None, **fields):
"""Search for releases by a free-form `query` string and/or any of
the following keyword arguments specifying field queries:
reid, release, arid, artist, artistname, creditname, type, status,
@@ -582,7 +621,7 @@ def release_search(query='', limit=None, offset=None, **fields):
"""
return _do_mb_search('release', query, fields, limit, offset)
def release_group_search(query='', limit=None, offset=None, **fields):
def search_release_groups(query='', limit=None, offset=None, **fields):
"""Search for release groups by a free-form `query` string and/or
any of the following keyword arguments specifying field queries:
rgid, releasegroup, reid, release, arid, artist, artistname,
@@ -590,7 +629,7 @@ def release_group_search(query='', limit=None, offset=None, **fields):
"""
return _do_mb_search('release-group', query, fields, limit, offset)
def work_search(query='', limit=None, offset=None, **fields):
def search_works(query='', limit=None, offset=None, **fields):
"""Search for works by a free-form `query` string and/or any of
the following keyword arguments specifying field queries:
wid, work, iswc, type, arid, artist, alias, tag
@@ -599,8 +638,8 @@ def work_search(query='', limit=None, offset=None, **fields):
# Lists of entities
def get_releases_by_discid(id, includes=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_type=release_type)
def get_releases_by_discid(id, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type=release_type)
return _do_mb_query("discid", id, includes, params)
def get_recordings_by_echoprint(echoprint, includes=[], release_status=[], release_type=[]):
@@ -618,79 +657,61 @@ def get_recordings_by_isrc(isrc, includes=[], release_status=[], release_type=[]
def get_works_by_iswc(iswc, includes=[]):
return _do_mb_query("iswc", iswc, includes)
def _browse_impl(entity, includes, valid_includes, limit, offset, params, release_status=[], release_type=[]):
_check_includes_impl(includes, valid_includes)
p = {}
for k,v in params.items():
if v:
p[k] = v
if len(p) > 1:
raise Exception("Can't have more than one of " + ", ".join(params.keys()))
if limit: p["limit"] = limit
if offset: p["offset"] = offset
filterp = _check_filter_and_make_params(entity, includes, release_status, release_type)
p.update(filterp)
return _do_mb_query(entity, "", includes, p)
# Browse methods
# Browse include are a subset of regular get includes, so we check them here
# and the test in _do_mb_query will pass anyway.
def browse_artist(recording=None, release=None, release_group=None, includes=[], limit=None, offset=None):
def browse_artists(recording=None, release=None, release_group=None, includes=[], limit=None, offset=None):
# optional parameter work?
_check_includes_impl(includes, ["aliases", "tags", "ratings", "user-tags", "user-ratings"])
p = {}
if recording: p["recording"] = recording
if release: p["release"] = release
if release_group: p["release-group"] = release_group
#if work: p["work"] = work
if len(p) > 1:
raise Exception("Can't have more than one of recording, release, release_group, work")
if limit: p["limit"] = limit
if offset: p["offset"] = offset
return _do_mb_query("artist", "", includes, p)
valid_includes = ["aliases", "tags", "ratings", "user-tags", "user-ratings"]
params = {"recording": recording,
"release": release,
"release-group": release_group}
return _browse_impl("artist", includes, valid_includes, limit, offset, params)
def browse_label(release=None, includes=[], limit=None, offset=None):
_check_includes_impl(includes, ["aliases", "tags", "ratings", "user-tags", "user-ratings"])
p = {"release": release}
if limit: p["limit"] = limit
if offset: p["offset"] = offset
return _do_mb_query("label", "", includes, p)
def browse_labels(release=None, includes=[], limit=None, offset=None):
valid_includes = ["aliases", "tags", "ratings", "user-tags", "user-ratings"]
params = {"release": release}
return _browse_impl("label", includes, valid_includes, limit, offset, params)
def browse_recording(artist=None, release=None, includes=[], limit=None, offset=None):
_check_includes_impl(includes, ["artist-credits", "tags", "ratings", "user-tags", "user-ratings"])
p = {}
if artist: p["artist"] = artist
if release: p["release"] = release
if len(p) > 1:
raise Exception("Can't have more than one of artist, release")
if limit: p["limit"] = limit
if offset: p["offset"] = offset
return _do_mb_query("recording", "", includes, p)
def browse_recordings(artist=None, release=None, includes=[], limit=None, offset=None):
valid_includes = ["artist-credits", "tags", "ratings", "user-tags", "user-ratings"]
params = {"artist": artist,
"release": release}
return _browse_impl("recording", includes, valid_includes, limit, offset, params)
def browse_release(artist=None, label=None, recording=None, release_group=None, release_status=[], release_type=[], includes=[], limit=None, offset=None):
def browse_releases(artist=None, label=None, recording=None, release_group=None, release_status=[], release_type=[], includes=[], limit=None, offset=None):
# track_artist param doesn't work yet
_check_includes_impl(includes, ["artist-credits", "labels", "recordings"])
p = {}
if artist: p["artist"] = artist
#if track_artist: p["track_artist"] = track_artist
if label: p["label"] = label
if recording: p["recording"] = recording
if release_group: p["release-group"] = release_group
if len(p) > 1:
raise Exception("Can't have more than one of artist, label, recording, release_group")
if limit: p["limit"] = limit
if offset: p["offset"] = offset
filterp = _check_filter_and_make_params("releases", release_status, release_type)
p.update(filterp)
if len(release_status) == 0 and len(release_type) == 0:
raise InvalidFilterError("Need at least one release status or type")
return _do_mb_query("release", "", includes, p)
valid_includes = ["artist-credits", "labels", "recordings", "release-groups"]
params = {"artist": artist,
"label": label,
"recording": recording,
"release-group": release_group}
return _browse_impl("release", includes, valid_includes, limit, offset, params, release_status, release_type)
def browse_release_group(artist=None, release=None, release_type=[], includes=[], limit=None, offset=None):
_check_includes_impl(includes, ["artist-credits", "tags", "ratings", "user-tags", "user-ratings"])
p = {}
if artist: p["artist"] = artist
if release: p["release"] = release
if len(p) > 1:
raise Exception("Can't have more than one of artist, release")
if limit: p["limit"] = limit
if offset: p["offset"] = offset
filterp = _check_filter_and_make_params("release-groups", [], release_type)
p.update(filterp)
if len(release_type) == 0:
raise InvalidFilterError("Need at least one release type")
return _do_mb_query("release-group", "", includes, p)
def browse_release_groups(artist=None, release=None, release_type=[], includes=[], limit=None, offset=None):
valid_includes = ["artist-credits", "tags", "ratings", "user-tags", "user-ratings"]
params = {"artist": artist,
"release": release}
return _browse_impl("release-group", includes, valid_includes, limit, offset, params, [], release_type)
# browse_work is defined in the docs but has no browse criteria
# Collections
def get_all_collections():
def get_collections():
# Missing <release-list count="n"> the count in the reply
return _do_mb_query("collection", '')
@@ -715,8 +736,14 @@ def submit_echoprints(echoprints):
query = mbxml.make_echoprint_request(echoprints)
return _do_mb_post("recording", query)
def submit_isrcs(isrcs):
raise NotImplementedError
def submit_isrcs(recordings_isrcs):
"""
Submit ISRCs.
Submits a set of {recording-id: [isrc1, irc1]}
Must call auth(user, pass) first
"""
query = mbxml.make_isrc_request(recordings_isrcs=recordings_isrcs)
return _do_mb_post("recording", query)
def submit_tags(artist_tags={}, recording_tags={}):
""" Submit user tags.