Updated beets library to 1.0b11

This commit is contained in:
Remy Varma
2011-11-03 22:03:03 +00:00
parent 0ba1e578ab
commit 5cf995b3dc
27 changed files with 8689 additions and 1059 deletions
+46 -535
View File
@@ -16,80 +16,23 @@
"""
import os
import logging
from collections import defaultdict
import re
from lib.munkres import Munkres
# from lib.unidecode import unidecode
from lib.beets.autotag import mb
from lib.beets import library, mediafile, plugins
from lib.beets.util import levenshtein, sorted_walk
from lib.beets import library, mediafile
from lib.beets.util import sorted_walk
# Try 5 releases. In the future, this should be more dynamic: let the
# probability of continuing to the next release be inversely
# proportional to how good our current best is and how long we've
# already taken.
MAX_CANDIDATES = 5
# Distance parameters.
# Text distance weights: proportions on the normalized intuitive edit
# distance.
ARTIST_WEIGHT = 3.0
ALBUM_WEIGHT = 3.0
# The weight of the entire distance calculated for a given track.
TRACK_WEIGHT = 1.0
# 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).
TRACK_TITLE_WEIGHT = 3.0
# Used instead of a global artist penalty for various-artist matches.
TRACK_ARTIST_WEIGHT = 2.0
# Added when the indices of tracks don't match.
TRACK_INDEX_WEIGHT = 1.0
# Track length weights: no penalty before GRACE, maximum (WEIGHT)
# penalty at GRACE+MAX discrepancy.
TRACK_LENGTH_GRACE = 10
TRACK_LENGTH_MAX = 30
TRACK_LENGTH_WEIGHT = 2.0
# MusicBrainz track ID matches.
TRACK_ID_WEIGHT = 5.0
# Recommendation constants.
RECOMMEND_STRONG = 'RECOMMEND_STRONG'
RECOMMEND_MEDIUM = 'RECOMMEND_MEDIUM'
RECOMMEND_NONE = 'RECOMMEND_NONE'
# Thresholds for recommendations.
STRONG_REC_THRESH = 0.04
MEDIUM_REC_THRESH = 0.25
REC_GAP_THRESH = 0.25
# Parameters for string distance function.
# Words that can be moved to the end of a string using a comma.
SD_END_WORDS = ['the', 'a', 'an']
# Reduced weights for certain portions of the string.
SD_PATTERNS = [
(r'^the ', 0.1),
(r'[\[\(]?(ep|single)[\]\)]?', 0.0),
(r'[\[\(]?(featuring|feat|ft)[\. :].+', 0.1),
(r'\(.*?\)', 0.3),
(r'\[.*?\]', 0.3),
(r'(, )?(pt\.|part) .+', 0.2),
]
# Replacements to use before testing distance.
SD_REPLACE = [
(r'&', 'and'),
]
# Artist signals that indicate "various artists".
VA_ARTISTS = (u'', u'various artists', u'va', u'unknown')
# Autotagging exceptions.
class AutotagError(Exception):
pass
# Parts of external interface.
from .hooks import AlbumInfo, TrackInfo
from .match import AutotagError
from .match import tag_item, tag_album
from .match import RECOMMEND_STRONG, RECOMMEND_MEDIUM, RECOMMEND_NONE
from .match import STRONG_REC_THRESH, MEDIUM_REC_THRESH, REC_GAP_THRESH
# Global logger.
log = logging.getLogger('beets')
# Additional utilities for the main interface.
def albums_in_dir(path):
"""Recursively searches the given directory and returns an iterable
of (path, items) where path is a containing directory and items is
@@ -113,488 +56,56 @@ def albums_in_dir(path):
if items:
yield root, items
def _string_dist_basic(str1, str2):
"""Basic edit distance between two strings, ignoring
non-alphanumeric characters and case. Comparisons are based on a
transliteration/lowering to ASCII characters. Normalized by string
length.
def apply_item_metadata(item, track_info):
"""Set an item's metadata from its matched TrackInfo object.
"""
# str1 = unidecode(str1)
# str2 = unidecode(str2)
str1 = re.sub(r'[^a-z0-9]', '', str1.lower())
str2 = re.sub(r'[^a-z0-9]', '', str2.lower())
if not str1 and not str2:
return 0.0
return levenshtein(str1, str2) / float(max(len(str1), len(str2)))
def string_dist(str1, str2):
"""Gives an "intuitive" edit distance between two strings. This is
an edit distance, normalized by the string length, with a number of
tweaks that reflect intuition about text.
"""
str1 = str1.lower()
str2 = str2.lower()
# Don't penalize strings that move certain words to the end. For
# example, "the something" should be considered equal to
# "something, the".
for word in SD_END_WORDS:
if str1.endswith(', %s' % word):
str1 = '%s %s' % (word, str1[:-len(word)-2])
if str2.endswith(', %s' % word):
str2 = '%s %s' % (word, str2[:-len(word)-2])
# Perform a couple of basic normalizing substitutions.
for pat, repl in SD_REPLACE:
str1 = re.sub(pat, repl, str1)
str2 = re.sub(pat, repl, str2)
# Change the weight for certain string portions matched by a set
# of regular expressions. We gradually change the strings and build
# up penalties associated with parts of the string that were
# deleted.
base_dist = _string_dist_basic(str1, str2)
penalty = 0.0
for pat, weight in SD_PATTERNS:
# Get strings that drop the pattern.
case_str1 = re.sub(pat, '', str1)
case_str2 = re.sub(pat, '', str2)
if case_str1 != str1 or case_str2 != str2:
# If the pattern was present (i.e., it is deleted in the
# the current case), recalculate the distances for the
# modified strings.
case_dist = _string_dist_basic(case_str1, case_str2)
case_delta = max(0.0, base_dist - case_dist)
if case_delta == 0.0:
continue
# Shift our baseline strings down (to avoid rematching the
# same part of the string) and add a scaled distance
# amount to the penalties.
str1 = case_str1
str2 = case_str2
base_dist = case_dist
penalty += weight * case_delta
dist = base_dist + penalty
return dist
def _plurality(objs):
"""Given a sequence of comparable objects, returns the object that
is most common in the set and if it is the only object is the set.
"""
# Calculate frequencies.
freqs = defaultdict(int)
for obj in objs:
freqs[obj] += 1
# Find object with maximum frequency.
max_freq = 0
res = None
for obj, freq in freqs.items():
if freq > max_freq:
max_freq = freq
res = obj
return res, len(freqs) <= 1
def current_metadata(items):
"""Returns the most likely artist and album for a set of Items.
Each is determined by tag reflected by the plurality of the Items.
"""
keys = 'artist', 'album'
likelies = {}
consensus = {}
for key in keys:
values = [getattr(item, key) for item in items]
likelies[key], consensus[key] = _plurality(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.
"""
# Make sure lengths match.
if len(items) != len(trackinfo):
return None
# Construct the cost matrix.
costs = []
for cur_item in items:
row = []
for i, canon_item in enumerate(trackinfo):
row.append(track_distance(cur_item, canon_item, i+1))
costs.append(row)
# Find a minimum-cost bipartite matching.
matching = Munkres().compute(costs)
# Order items based on the matching.
ordered_items = [None]*len(items)
for cur_idx, canon_idx in matching:
ordered_items[canon_idx] = items[cur_idx]
return ordered_items
def track_distance(item, track_data, track_index=None, incl_artist=False):
"""Determines the significance of a track metadata change. Returns
a float in [0.0,1.0]. `track_index` is the track number of the
`track_data` metadata set. If `track_index` is provided and
item.track is set, then these indices are used as a component of
the distance calculation. `incl_artist` indicates that a distance
component should be included for the track artist (i.e., for
various-artist releases).
"""
# Distance and normalization accumulators.
dist, dist_max = 0.0, 0.0
# Check track length.
if 'length' not in track_data:
# If there's no length to check, assume the worst.
dist += TRACK_LENGTH_WEIGHT
else:
diff = abs(item.length - track_data['length'])
diff = max(diff - TRACK_LENGTH_GRACE, 0.0)
diff = min(diff, TRACK_LENGTH_MAX)
dist += (diff / TRACK_LENGTH_MAX) * TRACK_LENGTH_WEIGHT
dist_max += TRACK_LENGTH_WEIGHT
# Track title.
dist += string_dist(item.title, track_data['title']) * TRACK_TITLE_WEIGHT
dist_max += TRACK_TITLE_WEIGHT
# Track artist, if included.
# 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 'artist' in track_data:
dist += string_dist(item.artist, track_data['artist']) * \
TRACK_ARTIST_WEIGHT
dist_max += TRACK_ARTIST_WEIGHT
# Track index.
if track_index and item.track:
if track_index != item.track:
dist += TRACK_INDEX_WEIGHT
dist_max += TRACK_INDEX_WEIGHT
# MusicBrainz track ID.
if item.mb_trackid:
if item.mb_trackid != track_data['id']:
dist += TRACK_ID_WEIGHT
dist_max += TRACK_ID_WEIGHT
# Plugin distances.
plugin_d, plugin_dm = plugins.track_distance(item, track_data)
dist += plugin_d
dist_max += plugin_dm
return dist / dist_max
def distance(items, info):
"""Determines how "significant" an album metadata change would be.
Returns a float in [0.0,1.0]. The list of items must be ordered.
"""
cur_artist, cur_album, _ = current_metadata(items)
cur_artist = cur_artist or ''
cur_album = cur_album or ''
# These accumulate the possible distance components. The final
# distance will be dist/dist_max.
dist = 0.0
dist_max = 0.0
# Artist/album metadata.
if not info['va']:
dist += string_dist(cur_artist, info['artist']) * ARTIST_WEIGHT
dist_max += ARTIST_WEIGHT
dist += string_dist(cur_album, info['album']) * ALBUM_WEIGHT
dist_max += ALBUM_WEIGHT
# Track distances.
for i, (item, track_data) in enumerate(zip(items, info['tracks'])):
dist += track_distance(item, track_data, i+1, info['va']) * \
TRACK_WEIGHT
dist_max += TRACK_WEIGHT
# Plugin distances.
plugin_d, plugin_dm = plugins.album_distance(items, info)
dist += plugin_d
dist_max += plugin_dm
# Normalize distance, avoiding divide-by-zero.
if dist_max == 0.0:
return 0.0
else:
return dist/dist_max
def apply_item_metadata(item, track_data):
"""Set an item's metadata from its matched info dictionary.
"""
item.artist = track_data['artist']
item.title = track_data['title']
item.mb_trackid = track_data['id']
if 'artist_id' in track_data:
item.mb_artistid = track_data['artist_id']
item.artist = track_info.artist
item.title = track_info.title
item.mb_trackid = track_info.track_id
if track_info.artist_id:
item.mb_artistid = track_info.artist_id
# At the moment, the other metadata is left intact (including album
# and track number). Perhaps these should be emptied?
def apply_metadata(items, info):
"""Set the items' metadata to match the data given in info. The
list of items must be ordered.
def apply_metadata(items, album_info):
"""Set the items' metadata to match an AlbumInfo object. The list
of items must be ordered.
"""
for index, (item, track_data) in enumerate(zip(items, info['tracks'])):
for index, (item, track_info) in enumerate(zip(items, album_info.tracks)):
# Album, artist, track count.
if 'artist' in track_data:
item.artist = track_data['artist']
if track_info.artist:
item.artist = track_info.artist
else:
item.artist = info['artist']
item.albumartist = info['artist']
item.album = info['album']
item.artist = album_info.artist
item.albumartist = album_info.artist
item.album = album_info.album
item.tracktotal = len(items)
# Release date.
if 'year' in info:
item.year = info['year']
if 'month' in info:
item.month = info['month']
if 'day' in info:
item.day = info['day']
if album_info.year:
item.year = album_info.year
if album_info.month:
item.month = album_info.month
if album_info.day:
item.day = album_info.day
# Title and track index.
item.title = track_data['title']
item.title = track_info.title
item.track = index + 1
# MusicBrainz IDs.
item.mb_trackid = track_data['id']
item.mb_albumid = info['album_id']
if 'artist_id' in track_data:
item.mb_artistid = track_data['artist_id']
item.mb_trackid = track_info.track_id
item.mb_albumid = album_info.album_id
if track_info.artist_id:
item.mb_artistid = track_info.artist_id
else:
item.mb_artistid = info['artist_id']
item.mb_albumartistid = info['artist_id']
item.albumtype = info['albumtype']
item.mb_artistid = album_info.artist_id
item.mb_albumartistid = album_info.artist_id
item.albumtype = album_info.albumtype
if album_info.label:
item.label = album_info.label
# Compilation flag.
item.comp = info['va']
item.comments = 'tagged by headphones/beets'
def match_by_id(items):
"""If the items are tagged with a MusicBrainz album ID, returns an
info dict for the corresponding album. Otherwise, returns None.
"""
# Is there a consensus on the MB album ID?
albumids = [item.mb_albumid for item in items if item.mb_albumid]
if not albumids:
log.debug('No album IDs found.')
return None
# If all album IDs are equal, look up the album.
if bool(reduce(lambda x,y: x if x==y else (), albumids)):
albumid = albumids[0]
log.debug('Searching for discovered album ID: ' + albumid)
return mb.album_for_id(albumid)
else:
log.debug('No album ID consensus.')
return None
#fixme In the future, at the expense of performance, we could use
# other IDs (i.e., track and artist) in case the album tag isn't
# present, but that event seems very unlikely.
def recommendation(results):
"""Given a sorted list of result tuples, returns a recommendation
flag (RECOMMEND_STRONG, RECOMMEND_MEDIUM, RECOMMEND_NONE) based
on the results' distances.
"""
if not results:
# No candidates: no recommendation.
rec = RECOMMEND_NONE
else:
min_dist = results[0][0]
if min_dist < STRONG_REC_THRESH:
# Strong recommendation level.
rec = RECOMMEND_STRONG
elif len(results) == 1:
# Only a single candidate. Medium recommendation.
rec = RECOMMEND_MEDIUM
elif min_dist <= MEDIUM_REC_THRESH:
# Medium recommendation level.
rec = RECOMMEND_MEDIUM
elif results[1][0] - min_dist >= REC_GAP_THRESH:
# Gap between first two candidates is large.
rec = RECOMMEND_MEDIUM
else:
# No conclusion.
rec = RECOMMEND_NONE
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
the track count, ordering the items, checking for duplicates, and
calculating the distance.
"""
log.debug('Candidate: %s - %s' % (info['artist'], info['album']))
# Don't duplicate.
if info['album_id'] in tuple_dict:
log.debug('Duplicate.')
return
# Make sure the album has the correct number of tracks.
if len(items) != len(info['tracks']):
log.debug('Track count mismatch.')
return
# Put items in order.
ordered = order_items(items, info['tracks'])
if not ordered:
log.debug('Not orderable.')
return
# Get the change distance.
dist = distance(ordered, info)
log.debug('Success. Distance: %f' % dist)
tuple_dict[info['album_id']] = dist, ordered, info
def tag_album(items, timid=False, search_artist=None, search_album=None,
search_id=None):
"""Bundles together the functionality used to infer tags for a
set of items comprised by an album. Returns everything relevant:
- The current artist.
- The current album.
- A list of (distance, items, info) tuples where info is a
dictionary containing the inferred tags and items is a
reordered version of the input items list. The candidates are
sorted by distance (i.e., best match first).
- A recommendation, one of RECOMMEND_STRONG, RECOMMEND_MEDIUM,
or RECOMMEND_NONE; indicating that the first candidate is
very likely, it is somewhat likely, or no conclusion could
be reached.
If search_artist and search_album or search_id are provided, then
they are used as search terms in place of the current metadata.
May raise an AutotagError if existing metadata is insufficient.
"""
# Get current metadata.
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 = {}
# Try to find album indicated by MusicBrainz IDs.
if search_id:
log.debug('Searching for album ID: ' + search_id)
id_info = mb.album_for_id(search_id)
else:
id_info = match_by_id(items)
if id_info:
validate_candidate(items, out_tuples, id_info)
rec = recommendation(out_tuples.values())
log.debug('Album ID match recommendation is ' + str(rec))
if out_tuples 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
# 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
else:
return cur_artist, cur_album, [], RECOMMEND_NONE
# Search terms.
if not (search_artist and search_album):
# No explicit search terms -- use current metadata.
search_artist, search_album = cur_artist, cur_album
log.debug(u'Search terms: %s - %s' % (search_artist, search_album))
# Get candidate metadata from search.
if search_artist and search_album:
candidates = mb.match_album(search_artist, search_album,
len(items), MAX_CANDIDATES)
candidates = list(candidates)
else:
candidates = []
# Possibly add "various artists" search.
if search_album and ((not artist_consensus) or \
(search_artist.lower() in VA_ARTISTS) or \
any(item.comp for item in items)):
log.debug(u'Possibly Various Artists; adding matches.')
candidates.extend(mb.match_album(None, search_album, len(items),
MAX_CANDIDATES))
# Get candidates from plugins.
candidates.extend(plugins.candidates(items))
# 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
def tag_item(item, timid=False, search_artist=None, search_title=None,
search_id=None):
"""Attempts to find metadata for a single track. Returns a
`(candidates, recommendation)` pair where `candidates` is a list
of `(distance, track_info)` pairs. `search_artist` and
`search_title` may be used to override the current metadata for
the purposes of the MusicBrainz title; likewise `search_id`.
"""
candidates = []
# First, try matching by MusicBrainz ID.
trackid = search_id or item.mb_trackid
if trackid:
log.debug('Searching for track ID: ' + trackid)
track_info = mb.track_for_id(trackid)
if track_info:
dist = track_distance(item, track_info, incl_artist=True)
candidates.append((dist, track_info))
# If this is a good match, then don't keep searching.
rec = recommendation(candidates)
if rec == RECOMMEND_STRONG and not timid:
log.debug('Track ID match.')
return candidates, rec
# If we're searching by ID, don't proceed.
if search_id is not None:
if candidates:
return candidates, rec
else:
return [], RECOMMEND_NONE
# Search terms.
if not (search_artist and search_title):
search_artist, search_title = item.artist, item.title
log.debug(u'Item search terms: %s - %s' % (search_artist, search_title))
# Candidate metadata from search.
for track_info in mb.match_track(search_artist, search_title):
dist = track_distance(item, track_info, incl_artist=True)
candidates.append((dist, track_info))
# Add candidates from plugins.
for track_info in plugins.item_candidates(item):
dist = track_distance(item, track_info, incl_artist=True)
candidates.append((dist, track_info))
# Sort by distance and return with recommendation.
log.debug('Found %i candidates.' % len(candidates))
candidates.sort()
rec = recommendation(candidates)
return candidates, rec
item.comp = album_info.va
# Uncomment to get rid of comments tag
item.comments = 'tagged by headphones/beets'
+109
View File
@@ -0,0 +1,109 @@
# 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.
"""Facilities for automatically determining files' correct metadata.
"""
import os
import logging
from lib.beets import library, mediafile
from lib.beets.util import sorted_walk
# Parts of external interface.
from .hooks import AlbumInfo, TrackInfo
from .match import AutotagError
from .match import tag_item, tag_album
from .match import RECOMMEND_STRONG, RECOMMEND_MEDIUM, RECOMMEND_NONE
from .match import STRONG_REC_THRESH, MEDIUM_REC_THRESH, REC_GAP_THRESH
# Global logger.
log = logging.getLogger('beets')
# Additional utilities for the main interface.
def albums_in_dir(path):
"""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.
"""
for root, dirs, files in sorted_walk(path):
# Get a list of items in the directory.
items = []
for filename in files:
try:
i = library.Item.from_path(os.path.join(root, filename))
except mediafile.FileTypeError:
pass
except mediafile.UnreadableFileError:
log.warn('unreadable file: ' + filename)
else:
items.append(i)
# If it's nonempty, yield it.
if items:
yield root, items
def apply_item_metadata(item, track_info):
"""Set an item's metadata from its matched TrackInfo object.
"""
item.artist = track_info.artist
item.title = track_info.title
item.mb_trackid = track_info.track_id
if track_info.artist_id:
item.mb_artistid = track_info.artist_id
# At the moment, the other metadata is left intact (including album
# and track number). Perhaps these should be emptied?
def apply_metadata(items, album_info):
"""Set the items' metadata to match an AlbumInfo object. The list
of items must be ordered.
"""
for index, (item, track_info) in enumerate(zip(items, album_info.tracks)):
# Album, artist, track count.
if track_info.artist:
item.artist = track_info.artist
else:
item.artist = album_info.artist
item.albumartist = album_info.artist
item.album = album_info.album
item.tracktotal = len(items)
# Release date.
if album_info.year:
item.year = album_info.year
if album_info.month:
item.month = album_info.month
if album_info.day:
item.day = album_info.day
# Title and track index.
item.title = track_info.title
item.track = index + 1
# MusicBrainz IDs.
item.mb_trackid = track_info.track_id
item.mb_albumid = album_info.album_id
if track_info.artist_id:
item.mb_artistid = track_info.artist_id
else:
item.mb_artistid = album_info.artist_id
item.mb_albumartistid = album_info.artist_id
item.albumtype = album_info.albumtype
if album_info.label:
item.label = album_info.label
# Compilation flag.
item.comp = album_info.va
+42 -6
View File
@@ -1,5 +1,5 @@
# This file is part of beets.
# Copyright 2010, Adrian Sampson.
# 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
@@ -17,9 +17,13 @@
import urllib
import sys
import logging
import os
from lib.beets.autotag.mb import album_for_id
IMAGE_EXTENSIONS = ['png', 'jpg', 'jpeg']
COVER_NAMES = ['cover', 'front', 'art', 'album', 'folder']
# The common logger.
log = logging.getLogger('beets')
@@ -47,15 +51,47 @@ def art_for_asin(asin):
return fn
# Art from the filesystem.
def art_in_path(path):
"""Look for album art files in a specified directory."""
if not os.path.isdir(path):
return
# Find all files that look like images in the directory.
images = []
for fn in os.listdir(path):
for ext in IMAGE_EXTENSIONS:
if fn.lower().endswith('.' + ext):
images.append(fn)
# Look for "preferred" filenames.
for fn in images:
for name in COVER_NAMES:
if fn.lower().startswith(name):
log.debug('Using well-named art file %s' % fn)
return os.path.join(path, fn)
# Fall back to any image in the folder.
if images:
log.debug('Using fallback art file %s' % images[0])
return os.path.join(path, images[0])
# Main interface.
def art_for_album(album):
def art_for_album(album, path):
"""Given an album info dictionary from MusicBrainz, returns a path
to downloaded art for the album (or None if no art is found).
"""
if album['asin']:
log.debug('Fetching album art for ASIN %s.' % album['asin'])
return art_for_asin(album['asin'])
if isinstance(path, basestring):
out = art_in_path(path)
if out:
return out
if album.asin:
log.debug('Fetching album art for ASIN %s.' % album.asin)
return art_for_asin(album.asin)
else:
log.debug('No ASIN available: no art found.')
return None
@@ -69,7 +105,7 @@ if __name__ == '__main__':
if not album:
print 'album not found'
else:
fn = art_for_album(album)
fn = art_for_album(album, None)
if fn:
print fn
print len(open(fn).read())/1024
+113
View File
@@ -0,0 +1,113 @@
# 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.
"""Finding album art for tagged albums."""
import urllib
import sys
import logging
import os
from beets.autotag.mb import album_for_id
IMAGE_EXTENSIONS = ['png', 'jpg', 'jpeg']
COVER_NAMES = ['cover', 'front', 'art', 'album', 'folder']
# The common logger.
log = logging.getLogger('beets')
# 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)
return fn
# Art from the filesystem.
def art_in_path(path):
"""Look for album art files in a specified directory."""
if not os.path.isdir(path):
return
# Find all files that look like images in the directory.
images = []
for fn in os.listdir(path):
for ext in IMAGE_EXTENSIONS:
if fn.lower().endswith('.' + ext):
images.append(fn)
# Look for "preferred" filenames.
for fn in images:
for name in COVER_NAMES:
if fn.lower().startswith(name):
log.debug('Using well-named art file %s' % fn)
return os.path.join(path, fn)
# Fall back to any image in the folder.
if images:
log.debug('Using fallback art file %s' % images[0])
return os.path.join(path, images[0])
# Main interface.
def art_for_album(album, path):
"""Given an album info dictionary from MusicBrainz, returns a path
to downloaded art for the album (or None if no art is found).
"""
if isinstance(path, basestring):
out = art_in_path(path)
if out:
return out
if album.asin:
log.debug('Fetching album art for ASIN %s.' % album.asin)
return art_for_asin(album.asin)
else:
log.debug('No ASIN available: no art found.')
return None
# Smoke test.
if __name__ == '__main__':
aid = sys.argv[1]
album = album_for_id(aid)
if not album:
print 'album not found'
else:
fn = art_for_album(album, None)
if fn:
print fn
print len(open(fn).read())/1024
else:
print 'no art found'
+125
View File
@@ -0,0 +1,125 @@
# 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.
"""Glue between metadata sources and the matching logic."""
from lib.beets import plugins
from lib.beets.autotag import mb
# Classes used to represent candidate options.
class AlbumInfo(object):
"""Describes a canonical release that may be used to match a release
in the library. Consists of these data members:
- ``album``: the release title
- ``album_id``: MusicBrainz ID; UUID fragment only
- ``artist``: name of the release's primary artist
- ``artist_id``
- ``tracks``: list of TrackInfo objects making up the release
- ``asin``: Amazon ASIN
- ``albumtype``: string describing the kind of release
- ``va``: boolean: whether the release has "various artists"
- ``year``: release year
- ``month``: release month
- ``day``: release day
- ``label``: music label responsible for the 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):
self.album = album
self.album_id = album_id
self.artist = artist
self.artist_id = artist_id
self.tracks = tracks
self.asin = asin
self.albumtype = albumtype
self.va = va
self.year = year
self.month = month
self.day = day
self.label = label
class TrackInfo(object):
"""Describes a canonical track present on a release. Appears as part
of an AlbumInfo's ``tracks`` list. Consists of these data members:
- ``title``: name of the track
- ``track_id``: MusicBrainz ID; UUID fragment only
- ``artist``: individual track artist name
- ``artist_id``
- ``length``: float: duration of the track in seconds
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):
self.title = title
self.track_id = track_id
self.artist = artist
self.artist_id = artist_id
self.length = length
# Aggregation of sources.
def _album_for_id(album_id):
"""Get an album corresponding to a MusicBrainz release ID."""
return mb.album_for_id(album_id)
def _track_for_id(track_id):
"""Get an item for a recording MBID."""
return mb.track_for_id(track_id)
def _album_candidates(items, artist, album, va_likely):
"""Search for album matches. ``items`` is a list of Item objects
that make up the album. ``artist`` and ``album`` are the respective
names (strings), which may be derived from the item list or may be
entered by the user. ``va_likely`` is a boolean indicating whether
the album is likely to be a "various artists" release.
"""
out = []
# Base candidates if we have album and artist to match.
if artist and album:
out.extend(mb.match_album(artist, album, len(items)))
# Also add VA matches from MusicBrainz where appropriate.
if va_likely and album:
out.extend(mb.match_album(None, album, len(items)))
# Candidates from plugins.
out.extend(plugins.candidates(items))
return out
def _item_candidates(item, artist, title):
"""Search for item matches. ``item`` is the Item to be matched.
``artist`` and ``title`` are strings and either reflect the item or
are specified by the user.
"""
out = []
# MusicBrainz candidates.
out.extend(mb.match_track(artist, title))
# Plugin candidates.
out.extend(plugins.item_candidates(item))
return out
+125
View File
@@ -0,0 +1,125 @@
# 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.
"""Glue between metadata sources and the matching logic."""
from beets import plugins
from beets.autotag import mb
# Classes used to represent candidate options.
class AlbumInfo(object):
"""Describes a canonical release that may be used to match a release
in the library. Consists of these data members:
- ``album``: the release title
- ``album_id``: MusicBrainz ID; UUID fragment only
- ``artist``: name of the release's primary artist
- ``artist_id``
- ``tracks``: list of TrackInfo objects making up the release
- ``asin``: Amazon ASIN
- ``albumtype``: string describing the kind of release
- ``va``: boolean: whether the release has "various artists"
- ``year``: release year
- ``month``: release month
- ``day``: release day
- ``label``: music label responsible for the 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):
self.album = album
self.album_id = album_id
self.artist = artist
self.artist_id = artist_id
self.tracks = tracks
self.asin = asin
self.albumtype = albumtype
self.va = va
self.year = year
self.month = month
self.day = day
self.label = label
class TrackInfo(object):
"""Describes a canonical track present on a release. Appears as part
of an AlbumInfo's ``tracks`` list. Consists of these data members:
- ``title``: name of the track
- ``track_id``: MusicBrainz ID; UUID fragment only
- ``artist``: individual track artist name
- ``artist_id``
- ``length``: float: duration of the track in seconds
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):
self.title = title
self.track_id = track_id
self.artist = artist
self.artist_id = artist_id
self.length = length
# Aggregation of sources.
def _album_for_id(album_id):
"""Get an album corresponding to a MusicBrainz release ID."""
return mb.album_for_id(album_id)
def _track_for_id(track_id):
"""Get an item for a recording MBID."""
return mb.track_for_id(track_id)
def _album_candidates(items, artist, album, va_likely):
"""Search for album matches. ``items`` is a list of Item objects
that make up the album. ``artist`` and ``album`` are the respective
names (strings), which may be derived from the item list or may be
entered by the user. ``va_likely`` is a boolean indicating whether
the album is likely to be a "various artists" release.
"""
out = []
# Base candidates if we have album and artist to match.
if artist and album:
out.extend(mb.match_album(artist, album, len(items)))
# Also add VA matches from MusicBrainz where appropriate.
if va_likely and album:
out.extend(mb.match_album(None, album, len(items)))
# Candidates from plugins.
out.extend(plugins.candidates(items))
return out
def _item_candidates(item, artist, title):
"""Search for item matches. ``item`` is the Item to be matched.
``artist`` and ``title`` are strings and either reflect the item or
are specified by the user.
"""
out = []
# MusicBrainz candidates.
out.extend(mb.match_track(artist, title))
# Plugin candidates.
out.extend(plugins.item_candidates(item))
return out
+490
View File
@@ -0,0 +1,490 @@
# 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.
"""Matches existing metadata with canonical information to identify
releases and tracks.
"""
import logging
import re
from lib.munkres import Munkres
#from unidecode import unidecode
from lib.beets import plugins
from lib.beets.util import levenshtein, plurality
from lib.beets.autotag import hooks
# Distance parameters.
# Text distance weights: proportions on the normalized intuitive edit
# distance.
ARTIST_WEIGHT = 3.0
ALBUM_WEIGHT = 3.0
# The weight of the entire distance calculated for a given track.
TRACK_WEIGHT = 1.0
# 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).
TRACK_TITLE_WEIGHT = 3.0
# Used instead of a global artist penalty for various-artist matches.
TRACK_ARTIST_WEIGHT = 2.0
# Added when the indices of tracks don't match.
TRACK_INDEX_WEIGHT = 1.0
# Track length weights: no penalty before GRACE, maximum (WEIGHT)
# penalty at GRACE+MAX discrepancy.
TRACK_LENGTH_GRACE = 10
TRACK_LENGTH_MAX = 30
TRACK_LENGTH_WEIGHT = 2.0
# MusicBrainz track ID matches.
TRACK_ID_WEIGHT = 5.0
# Parameters for string distance function.
# Words that can be moved to the end of a string using a comma.
SD_END_WORDS = ['the', 'a', 'an']
# Reduced weights for certain portions of the string.
SD_PATTERNS = [
(r'^the ', 0.1),
(r'[\[\(]?(ep|single)[\]\)]?', 0.0),
(r'[\[\(]?(featuring|feat|ft)[\. :].+', 0.1),
(r'\(.*?\)', 0.3),
(r'\[.*?\]', 0.3),
(r'(, )?(pt\.|part) .+', 0.2),
]
# Replacements to use before testing distance.
SD_REPLACE = [
(r'&', 'and'),
]
# Recommendation constants.
RECOMMEND_STRONG = 'RECOMMEND_STRONG'
RECOMMEND_MEDIUM = 'RECOMMEND_MEDIUM'
RECOMMEND_NONE = 'RECOMMEND_NONE'
# Thresholds for recommendations.
STRONG_REC_THRESH = 0.04
MEDIUM_REC_THRESH = 0.25
REC_GAP_THRESH = 0.25
# Artist signals that indicate "various artists".
VA_ARTISTS = (u'', u'various artists', u'va', u'unknown')
# Autotagging exceptions.
class AutotagError(Exception):
pass
# Global logger.
log = logging.getLogger('beets')
# Primary matching functionality.
def _string_dist_basic(str1, str2):
"""Basic edit distance between two strings, ignoring
non-alphanumeric characters and case. Comparisons are based on a
transliteration/lowering to ASCII characters. Normalized by string
length.
"""
#str1 = unidecode(str1)
#str2 = unidecode(str2)
str1 = re.sub(r'[^a-z0-9]', '', str1.lower())
str2 = re.sub(r'[^a-z0-9]', '', str2.lower())
if not str1 and not str2:
return 0.0
return levenshtein(str1, str2) / float(max(len(str1), len(str2)))
def string_dist(str1, str2):
"""Gives an "intuitive" edit distance between two strings. This is
an edit distance, normalized by the string length, with a number of
tweaks that reflect intuition about text.
"""
str1 = str1.lower()
str2 = str2.lower()
# Don't penalize strings that move certain words to the end. For
# example, "the something" should be considered equal to
# "something, the".
for word in SD_END_WORDS:
if str1.endswith(', %s' % word):
str1 = '%s %s' % (word, str1[:-len(word)-2])
if str2.endswith(', %s' % word):
str2 = '%s %s' % (word, str2[:-len(word)-2])
# Perform a couple of basic normalizing substitutions.
for pat, repl in SD_REPLACE:
str1 = re.sub(pat, repl, str1)
str2 = re.sub(pat, repl, str2)
# Change the weight for certain string portions matched by a set
# of regular expressions. We gradually change the strings and build
# up penalties associated with parts of the string that were
# deleted.
base_dist = _string_dist_basic(str1, str2)
penalty = 0.0
for pat, weight in SD_PATTERNS:
# Get strings that drop the pattern.
case_str1 = re.sub(pat, '', str1)
case_str2 = re.sub(pat, '', str2)
if case_str1 != str1 or case_str2 != str2:
# If the pattern was present (i.e., it is deleted in the
# the current case), recalculate the distances for the
# modified strings.
case_dist = _string_dist_basic(case_str1, case_str2)
case_delta = max(0.0, base_dist - case_dist)
if case_delta == 0.0:
continue
# Shift our baseline strings down (to avoid rematching the
# same part of the string) and add a scaled distance
# amount to the penalties.
str1 = case_str1
str2 = case_str2
base_dist = case_dist
penalty += weight * case_delta
dist = base_dist + penalty
return dist
def current_metadata(items):
"""Returns the most likely artist and album for a set of Items.
Each is determined by tag reflected by the plurality of the Items.
"""
keys = 'artist', 'album'
likelies = {}
consensus = {}
for key in keys:
values = [getattr(item, key) for item in items]
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.
"""
# Make sure lengths match.
if len(items) != len(trackinfo):
return None
# Construct the cost matrix.
costs = []
for cur_item in items:
row = []
for i, canon_item in enumerate(trackinfo):
row.append(track_distance(cur_item, canon_item, i+1))
costs.append(row)
# Find a minimum-cost bipartite matching.
matching = Munkres().compute(costs)
# Order items based on the matching.
ordered_items = [None]*len(items)
for cur_idx, canon_idx in matching:
ordered_items[canon_idx] = items[cur_idx]
return ordered_items
def track_distance(item, track_info, track_index=None, incl_artist=False):
"""Determines the significance of a track metadata change. Returns
a float in [0.0,1.0]. `track_index` is the track number of the
`track_info` metadata set. If `track_index` is provided and
item.track is set, then these indices are used as a component of
the distance calculation. `incl_artist` indicates that a distance
component should be included for the track artist (i.e., for
various-artist releases).
"""
# Distance and normalization accumulators.
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:
diff = abs(item.length - track_info.length)
diff = max(diff - TRACK_LENGTH_GRACE, 0.0)
diff = min(diff, TRACK_LENGTH_MAX)
dist += (diff / TRACK_LENGTH_MAX) * TRACK_LENGTH_WEIGHT
dist_max += TRACK_LENGTH_WEIGHT
# Track title.
dist += string_dist(item.title, track_info.title) * TRACK_TITLE_WEIGHT
dist_max += TRACK_TITLE_WEIGHT
# Track artist, if included.
# 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:
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:
dist += TRACK_INDEX_WEIGHT
dist_max += TRACK_INDEX_WEIGHT
# MusicBrainz track ID.
if item.mb_trackid:
if item.mb_trackid != track_info.track_id:
dist += TRACK_ID_WEIGHT
dist_max += TRACK_ID_WEIGHT
# Plugin distances.
plugin_d, plugin_dm = plugins.track_distance(item, track_info)
dist += plugin_d
dist_max += plugin_dm
return dist / dist_max
def distance(items, album_info):
"""Determines how "significant" an album metadata change would be.
Returns a float in [0.0,1.0]. The list of items must be ordered.
"""
cur_artist, cur_album, _ = current_metadata(items)
cur_artist = cur_artist or ''
cur_album = cur_album or ''
# These accumulate the possible distance components. The final
# distance will be dist/dist_max.
dist = 0.0
dist_max = 0.0
# Artist/album metadata.
if not album_info.va:
dist += string_dist(cur_artist, album_info.artist) * ARTIST_WEIGHT
dist_max += ARTIST_WEIGHT
dist += string_dist(cur_album, album_info.album) * ALBUM_WEIGHT
dist_max += ALBUM_WEIGHT
# 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
# Plugin distances.
plugin_d, plugin_dm = plugins.album_distance(items, album_info)
dist += plugin_d
dist_max += plugin_dm
# Normalize distance, avoiding divide-by-zero.
if dist_max == 0.0:
return 0.0
else:
return dist/dist_max
def match_by_id(items):
"""If the items are tagged with a MusicBrainz album ID, returns an
info dict for the corresponding album. Otherwise, returns None.
"""
# Is there a consensus on the MB album ID?
albumids = [item.mb_albumid for item in items if item.mb_albumid]
if not albumids:
log.debug('No album IDs found.')
return None
# If all album IDs are equal, look up the album.
if bool(reduce(lambda x,y: x if x==y else (), albumids)):
albumid = albumids[0]
log.debug('Searching for discovered album ID: ' + albumid)
return hooks._album_for_id(albumid)
else:
log.debug('No album ID consensus.')
return None
#fixme In the future, at the expense of performance, we could use
# other IDs (i.e., track and artist) in case the album tag isn't
# present, but that event seems very unlikely.
def recommendation(results):
"""Given a sorted list of result tuples, returns a recommendation
flag (RECOMMEND_STRONG, RECOMMEND_MEDIUM, RECOMMEND_NONE) based
on the results' distances.
"""
if not results:
# No candidates: no recommendation.
rec = RECOMMEND_NONE
else:
min_dist = results[0][0]
if min_dist < STRONG_REC_THRESH:
# Strong recommendation level.
rec = RECOMMEND_STRONG
elif len(results) == 1:
# Only a single candidate. Medium recommendation.
rec = RECOMMEND_MEDIUM
elif min_dist <= MEDIUM_REC_THRESH:
# Medium recommendation level.
rec = RECOMMEND_MEDIUM
elif results[1][0] - min_dist >= REC_GAP_THRESH:
# Gap between first two candidates is large.
rec = RECOMMEND_MEDIUM
else:
# No conclusion.
rec = RECOMMEND_NONE
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
the track count, ordering the items, checking for duplicates, and
calculating the distance.
"""
log.debug('Candidate: %s - %s' % (info.artist, info.album))
# Don't duplicate.
if info.album_id in tuple_dict:
log.debug('Duplicate.')
return
# Make sure the album has the correct number of tracks.
if len(items) != len(info.tracks):
log.debug('Track count mismatch.')
return
# Put items in order.
ordered = order_items(items, info.tracks)
if not ordered:
log.debug('Not orderable.')
return
# Get the change distance.
dist = distance(ordered, info)
log.debug('Success. Distance: %f' % dist)
tuple_dict[info.album_id] = dist, ordered, info
def tag_album(items, timid=False, search_artist=None, search_album=None,
search_id=None):
"""Bundles together the functionality used to infer tags for a
set of items comprised by an album. Returns everything relevant:
- The current artist.
- The current album.
- A list of (distance, items, info) tuples where info is a
dictionary containing the inferred tags and items is a
reordered version of the input items list. The candidates are
sorted by distance (i.e., best match first).
- A recommendation, one of RECOMMEND_STRONG, RECOMMEND_MEDIUM,
or RECOMMEND_NONE; indicating that the first candidate is
very likely, it is somewhat likely, or no conclusion could
be reached.
If search_artist and search_album or search_id are provided, then
they are used as search terms in place of the current metadata.
May raise an AutotagError if existing metadata is insufficient.
"""
# Get current metadata.
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 = {}
# Try to find album indicated by MusicBrainz IDs.
if search_id:
log.debug('Searching for album ID: ' + search_id)
id_info = hooks._album_for_id(search_id)
else:
id_info = match_by_id(items)
if id_info:
validate_candidate(items, out_tuples, id_info)
rec = recommendation(out_tuples.values())
log.debug('Album ID match recommendation is ' + str(rec))
if out_tuples 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
# 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
else:
return cur_artist, cur_album, [], RECOMMEND_NONE
# Search terms.
if not (search_artist and search_album):
# No explicit search terms -- use current metadata.
search_artist, search_album = cur_artist, cur_album
log.debug(u'Search terms: %s - %s' % (search_artist, search_album))
# Is this album likely to be a "various artist" release?
va_likely = ((not artist_consensus) or
(search_artist.lower() in VA_ARTISTS) or
any(item.comp for item in items))
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)
# 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
def tag_item(item, timid=False, search_artist=None, search_title=None,
search_id=None):
"""Attempts to find metadata for a single track. Returns a
`(candidates, recommendation)` pair where `candidates` is a list
of `(distance, track_info)` pairs. `search_artist` and
`search_title` may be used to override the current metadata for
the purposes of the MusicBrainz title; likewise `search_id`.
"""
candidates = []
# First, try matching by MusicBrainz ID.
trackid = search_id or item.mb_trackid
if trackid:
log.debug('Searching for track ID: ' + trackid)
track_info = hooks._track_for_id(trackid)
if track_info:
dist = track_distance(item, track_info, incl_artist=True)
candidates.append((dist, track_info))
# If this is a good match, then don't keep searching.
rec = recommendation(candidates)
if rec == RECOMMEND_STRONG and not timid:
log.debug('Track ID match.')
return candidates, rec
# If we're searching by ID, don't proceed.
if search_id is not None:
if candidates:
return candidates, rec
else:
return [], RECOMMEND_NONE
# Search terms.
if not (search_artist and search_title):
search_artist, search_title = item.artist, item.title
log.debug(u'Item search terms: %s - %s' % (search_artist, search_title))
# 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))
# Sort by distance and return with recommendation.
log.debug('Found %i candidates.' % len(candidates))
candidates.sort()
rec = recommendation(candidates)
return candidates, rec
+490
View File
@@ -0,0 +1,490 @@
# 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.
"""Matches existing metadata with canonical information to identify
releases and tracks.
"""
import logging
import re
from munkres import Munkres
from unidecode import unidecode
from beets import plugins
from beets.util import levenshtein, plurality
from beets.autotag import hooks
# Distance parameters.
# Text distance weights: proportions on the normalized intuitive edit
# distance.
ARTIST_WEIGHT = 3.0
ALBUM_WEIGHT = 3.0
# The weight of the entire distance calculated for a given track.
TRACK_WEIGHT = 1.0
# 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).
TRACK_TITLE_WEIGHT = 3.0
# Used instead of a global artist penalty for various-artist matches.
TRACK_ARTIST_WEIGHT = 2.0
# Added when the indices of tracks don't match.
TRACK_INDEX_WEIGHT = 1.0
# Track length weights: no penalty before GRACE, maximum (WEIGHT)
# penalty at GRACE+MAX discrepancy.
TRACK_LENGTH_GRACE = 10
TRACK_LENGTH_MAX = 30
TRACK_LENGTH_WEIGHT = 2.0
# MusicBrainz track ID matches.
TRACK_ID_WEIGHT = 5.0
# Parameters for string distance function.
# Words that can be moved to the end of a string using a comma.
SD_END_WORDS = ['the', 'a', 'an']
# Reduced weights for certain portions of the string.
SD_PATTERNS = [
(r'^the ', 0.1),
(r'[\[\(]?(ep|single)[\]\)]?', 0.0),
(r'[\[\(]?(featuring|feat|ft)[\. :].+', 0.1),
(r'\(.*?\)', 0.3),
(r'\[.*?\]', 0.3),
(r'(, )?(pt\.|part) .+', 0.2),
]
# Replacements to use before testing distance.
SD_REPLACE = [
(r'&', 'and'),
]
# Recommendation constants.
RECOMMEND_STRONG = 'RECOMMEND_STRONG'
RECOMMEND_MEDIUM = 'RECOMMEND_MEDIUM'
RECOMMEND_NONE = 'RECOMMEND_NONE'
# Thresholds for recommendations.
STRONG_REC_THRESH = 0.04
MEDIUM_REC_THRESH = 0.25
REC_GAP_THRESH = 0.25
# Artist signals that indicate "various artists".
VA_ARTISTS = (u'', u'various artists', u'va', u'unknown')
# Autotagging exceptions.
class AutotagError(Exception):
pass
# Global logger.
log = logging.getLogger('beets')
# Primary matching functionality.
def _string_dist_basic(str1, str2):
"""Basic edit distance between two strings, ignoring
non-alphanumeric characters and case. Comparisons are based on a
transliteration/lowering to ASCII characters. Normalized by string
length.
"""
str1 = unidecode(str1)
str2 = unidecode(str2)
str1 = re.sub(r'[^a-z0-9]', '', str1.lower())
str2 = re.sub(r'[^a-z0-9]', '', str2.lower())
if not str1 and not str2:
return 0.0
return levenshtein(str1, str2) / float(max(len(str1), len(str2)))
def string_dist(str1, str2):
"""Gives an "intuitive" edit distance between two strings. This is
an edit distance, normalized by the string length, with a number of
tweaks that reflect intuition about text.
"""
str1 = str1.lower()
str2 = str2.lower()
# Don't penalize strings that move certain words to the end. For
# example, "the something" should be considered equal to
# "something, the".
for word in SD_END_WORDS:
if str1.endswith(', %s' % word):
str1 = '%s %s' % (word, str1[:-len(word)-2])
if str2.endswith(', %s' % word):
str2 = '%s %s' % (word, str2[:-len(word)-2])
# Perform a couple of basic normalizing substitutions.
for pat, repl in SD_REPLACE:
str1 = re.sub(pat, repl, str1)
str2 = re.sub(pat, repl, str2)
# Change the weight for certain string portions matched by a set
# of regular expressions. We gradually change the strings and build
# up penalties associated with parts of the string that were
# deleted.
base_dist = _string_dist_basic(str1, str2)
penalty = 0.0
for pat, weight in SD_PATTERNS:
# Get strings that drop the pattern.
case_str1 = re.sub(pat, '', str1)
case_str2 = re.sub(pat, '', str2)
if case_str1 != str1 or case_str2 != str2:
# If the pattern was present (i.e., it is deleted in the
# the current case), recalculate the distances for the
# modified strings.
case_dist = _string_dist_basic(case_str1, case_str2)
case_delta = max(0.0, base_dist - case_dist)
if case_delta == 0.0:
continue
# Shift our baseline strings down (to avoid rematching the
# same part of the string) and add a scaled distance
# amount to the penalties.
str1 = case_str1
str2 = case_str2
base_dist = case_dist
penalty += weight * case_delta
dist = base_dist + penalty
return dist
def current_metadata(items):
"""Returns the most likely artist and album for a set of Items.
Each is determined by tag reflected by the plurality of the Items.
"""
keys = 'artist', 'album'
likelies = {}
consensus = {}
for key in keys:
values = [getattr(item, key) for item in items]
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.
"""
# Make sure lengths match.
if len(items) != len(trackinfo):
return None
# Construct the cost matrix.
costs = []
for cur_item in items:
row = []
for i, canon_item in enumerate(trackinfo):
row.append(track_distance(cur_item, canon_item, i+1))
costs.append(row)
# Find a minimum-cost bipartite matching.
matching = Munkres().compute(costs)
# Order items based on the matching.
ordered_items = [None]*len(items)
for cur_idx, canon_idx in matching:
ordered_items[canon_idx] = items[cur_idx]
return ordered_items
def track_distance(item, track_info, track_index=None, incl_artist=False):
"""Determines the significance of a track metadata change. Returns
a float in [0.0,1.0]. `track_index` is the track number of the
`track_info` metadata set. If `track_index` is provided and
item.track is set, then these indices are used as a component of
the distance calculation. `incl_artist` indicates that a distance
component should be included for the track artist (i.e., for
various-artist releases).
"""
# Distance and normalization accumulators.
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:
diff = abs(item.length - track_info.length)
diff = max(diff - TRACK_LENGTH_GRACE, 0.0)
diff = min(diff, TRACK_LENGTH_MAX)
dist += (diff / TRACK_LENGTH_MAX) * TRACK_LENGTH_WEIGHT
dist_max += TRACK_LENGTH_WEIGHT
# Track title.
dist += string_dist(item.title, track_info.title) * TRACK_TITLE_WEIGHT
dist_max += TRACK_TITLE_WEIGHT
# Track artist, if included.
# 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:
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:
dist += TRACK_INDEX_WEIGHT
dist_max += TRACK_INDEX_WEIGHT
# MusicBrainz track ID.
if item.mb_trackid:
if item.mb_trackid != track_info.track_id:
dist += TRACK_ID_WEIGHT
dist_max += TRACK_ID_WEIGHT
# Plugin distances.
plugin_d, plugin_dm = plugins.track_distance(item, track_info)
dist += plugin_d
dist_max += plugin_dm
return dist / dist_max
def distance(items, album_info):
"""Determines how "significant" an album metadata change would be.
Returns a float in [0.0,1.0]. The list of items must be ordered.
"""
cur_artist, cur_album, _ = current_metadata(items)
cur_artist = cur_artist or ''
cur_album = cur_album or ''
# These accumulate the possible distance components. The final
# distance will be dist/dist_max.
dist = 0.0
dist_max = 0.0
# Artist/album metadata.
if not album_info.va:
dist += string_dist(cur_artist, album_info.artist) * ARTIST_WEIGHT
dist_max += ARTIST_WEIGHT
dist += string_dist(cur_album, album_info.album) * ALBUM_WEIGHT
dist_max += ALBUM_WEIGHT
# 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
# Plugin distances.
plugin_d, plugin_dm = plugins.album_distance(items, album_info)
dist += plugin_d
dist_max += plugin_dm
# Normalize distance, avoiding divide-by-zero.
if dist_max == 0.0:
return 0.0
else:
return dist/dist_max
def match_by_id(items):
"""If the items are tagged with a MusicBrainz album ID, returns an
info dict for the corresponding album. Otherwise, returns None.
"""
# Is there a consensus on the MB album ID?
albumids = [item.mb_albumid for item in items if item.mb_albumid]
if not albumids:
log.debug('No album IDs found.')
return None
# If all album IDs are equal, look up the album.
if bool(reduce(lambda x,y: x if x==y else (), albumids)):
albumid = albumids[0]
log.debug('Searching for discovered album ID: ' + albumid)
return hooks._album_for_id(albumid)
else:
log.debug('No album ID consensus.')
return None
#fixme In the future, at the expense of performance, we could use
# other IDs (i.e., track and artist) in case the album tag isn't
# present, but that event seems very unlikely.
def recommendation(results):
"""Given a sorted list of result tuples, returns a recommendation
flag (RECOMMEND_STRONG, RECOMMEND_MEDIUM, RECOMMEND_NONE) based
on the results' distances.
"""
if not results:
# No candidates: no recommendation.
rec = RECOMMEND_NONE
else:
min_dist = results[0][0]
if min_dist < STRONG_REC_THRESH:
# Strong recommendation level.
rec = RECOMMEND_STRONG
elif len(results) == 1:
# Only a single candidate. Medium recommendation.
rec = RECOMMEND_MEDIUM
elif min_dist <= MEDIUM_REC_THRESH:
# Medium recommendation level.
rec = RECOMMEND_MEDIUM
elif results[1][0] - min_dist >= REC_GAP_THRESH:
# Gap between first two candidates is large.
rec = RECOMMEND_MEDIUM
else:
# No conclusion.
rec = RECOMMEND_NONE
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
the track count, ordering the items, checking for duplicates, and
calculating the distance.
"""
log.debug('Candidate: %s - %s' % (info.artist, info.album))
# Don't duplicate.
if info.album_id in tuple_dict:
log.debug('Duplicate.')
return
# Make sure the album has the correct number of tracks.
if len(items) != len(info.tracks):
log.debug('Track count mismatch.')
return
# Put items in order.
ordered = order_items(items, info.tracks)
if not ordered:
log.debug('Not orderable.')
return
# Get the change distance.
dist = distance(ordered, info)
log.debug('Success. Distance: %f' % dist)
tuple_dict[info.album_id] = dist, ordered, info
def tag_album(items, timid=False, search_artist=None, search_album=None,
search_id=None):
"""Bundles together the functionality used to infer tags for a
set of items comprised by an album. Returns everything relevant:
- The current artist.
- The current album.
- A list of (distance, items, info) tuples where info is a
dictionary containing the inferred tags and items is a
reordered version of the input items list. The candidates are
sorted by distance (i.e., best match first).
- A recommendation, one of RECOMMEND_STRONG, RECOMMEND_MEDIUM,
or RECOMMEND_NONE; indicating that the first candidate is
very likely, it is somewhat likely, or no conclusion could
be reached.
If search_artist and search_album or search_id are provided, then
they are used as search terms in place of the current metadata.
May raise an AutotagError if existing metadata is insufficient.
"""
# Get current metadata.
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 = {}
# Try to find album indicated by MusicBrainz IDs.
if search_id:
log.debug('Searching for album ID: ' + search_id)
id_info = hooks._album_for_id(search_id)
else:
id_info = match_by_id(items)
if id_info:
validate_candidate(items, out_tuples, id_info)
rec = recommendation(out_tuples.values())
log.debug('Album ID match recommendation is ' + str(rec))
if out_tuples 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
# 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
else:
return cur_artist, cur_album, [], RECOMMEND_NONE
# Search terms.
if not (search_artist and search_album):
# No explicit search terms -- use current metadata.
search_artist, search_album = cur_artist, cur_album
log.debug(u'Search terms: %s - %s' % (search_artist, search_album))
# Is this album likely to be a "various artist" release?
va_likely = ((not artist_consensus) or
(search_artist.lower() in VA_ARTISTS) or
any(item.comp for item in items))
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)
# 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
def tag_item(item, timid=False, search_artist=None, search_title=None,
search_id=None):
"""Attempts to find metadata for a single track. Returns a
`(candidates, recommendation)` pair where `candidates` is a list
of `(distance, track_info)` pairs. `search_artist` and
`search_title` may be used to override the current metadata for
the purposes of the MusicBrainz title; likewise `search_id`.
"""
candidates = []
# First, try matching by MusicBrainz ID.
trackid = search_id or item.mb_trackid
if trackid:
log.debug('Searching for track ID: ' + trackid)
track_info = hooks._track_for_id(trackid)
if track_info:
dist = track_distance(item, track_info, incl_artist=True)
candidates.append((dist, track_info))
# If this is a good match, then don't keep searching.
rec = recommendation(candidates)
if rec == RECOMMEND_STRONG and not timid:
log.debug('Track ID match.')
return candidates, rec
# If we're searching by ID, don't proceed.
if search_id is not None:
if candidates:
return candidates, rec
else:
return [], RECOMMEND_NONE
# Search terms.
if not (search_artist and search_title):
search_artist, search_title = item.artist, item.title
log.debug(u'Item search terms: %s - %s' % (search_artist, search_title))
# 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))
# Sort by distance and return with recommendation.
log.debug('Found %i candidates.' % len(candidates))
candidates.sort()
rec = recommendation(candidates)
return candidates, rec
+98 -260
View File
@@ -13,24 +13,17 @@
# included in all copies or substantial portions of the Software.
"""Searches for albums in the MusicBrainz database.
This is a thin layer over the official `python-musicbrainz2` module. It
abstracts away that module's object model, the server's Lucene query
syntax, and other uninteresting parts of using musicbrainz2. The
principal interface is the function `match_album`.
"""
from __future__ import with_statement # for Python 2.5
import re
import time
import logging
import lib.musicbrainz2.webservice as mbws
from lib.musicbrainz2.model import Release
from threading import Lock
from lib.musicbrainz2.model import VARIOUS_ARTISTS_ID
SEARCH_LIMIT = 10
VARIOUS_ARTISTS_ID = VARIOUS_ARTISTS_ID.rsplit('/', 1)[1]
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__
class ServerBusyError(Exception): pass
class BadResponseError(Exception): pass
@@ -42,242 +35,84 @@ SPECIAL_CASE_ARTISTS = {
'!!!': 'f26c72d3-e52c-467b-b651-679c73d8e1a7',
}
RELEASE_TYPES = [
Release.TYPE_ALBUM,
Release.TYPE_SINGLE,
Release.TYPE_EP,
Release.TYPE_COMPILATION,
Release.TYPE_SOUNDTRACK,
Release.TYPE_SPOKENWORD,
Release.TYPE_INTERVIEW,
Release.TYPE_AUDIOBOOK,
Release.TYPE_LIVE,
Release.TYPE_REMIX,
Release.TYPE_OTHER
]
RELEASE_INCLUDES = ['artists', 'media', 'recordings', 'release-groups',
'labels']
TRACK_INCLUDES = ['artists']
RELEASE_INCLUDES = mbws.ReleaseIncludes(artist=True, tracks=True,
releaseEvents=True, labels=True,
releaseGroup=True)
TRACK_INCLUDES = mbws.TrackIncludes(artist=True)
# MusicBrainz requires that a client does not query the server more
# than once a second. This function enforces that limit using a
# module-global variable to keep track of the last time a query was
# sent.
MAX_QUERY_RETRY = 8
QUERY_WAIT_TIME = 1.0
last_query_time = 0.0
mb_lock = Lock()
def _query_wrap(fun, *args, **kwargs):
"""Wait until at least `QUERY_WAIT_TIME` seconds have passed since
the last invocation of this function. Then call
fun(*args, **kwargs). If it fails due to a "server busy" message,
then try again. Tries up to `MAX_QUERY_RETRY` times before
giving up.
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.
"""
with mb_lock:
global last_query_time
for i in range(MAX_QUERY_RETRY):
since_last_query = time.time() - last_query_time
if since_last_query < QUERY_WAIT_TIME:
time.sleep(QUERY_WAIT_TIME - since_last_query)
last_query_time = time.time()
try:
# Try the function.
res = fun(*args, **kwargs)
except mbws.WebServiceError, e:
# Server busy. Retry.
message = str(e.reason)
for errnum in (503, 504):
if 'Error %i' % errnum in message:
break
else:
# This is not the error we're looking for.
raise
except mbws.ConnectionError:
# Typically a timeout.
pass
except mbws.ResponseError, exc:
# Malformed response from server.
log.error('Bad response from MusicBrainz: ' + str(exc))
raise BadResponseError()
else:
# Success. Return the result.
return res
# Gave up.
raise ServerBusyError()
# FIXME exponential backoff?
def get_releases(**params):
"""Given a list of parameters to ReleaseFilter, executes the
query and yields release dicts (complete with tracks).
"""
# Replace special cases.
if 'artistName' in params:
artist = params['artistName']
if artist in SPECIAL_CASE_ARTISTS:
del params['artistName']
params['artistId'] = SPECIAL_CASE_ARTISTS[artist]
# Issue query.
filt = mbws.ReleaseFilter(**params)
try:
results = _query_wrap(mbws.Query().getReleases, filter=filt)
except BadResponseError:
results = ()
# Construct results.
for result in results:
release = result.release
tracks, _ = release_info(release.id)
yield release_dict(release, tracks)
def release_info(release_id):
"""Given a MusicBrainz release ID, fetch a list of tracks on the
release and the release group ID. If the release is not found,
returns None.
"""
try:
release = _query_wrap(mbws.Query().getReleaseById, release_id,
RELEASE_INCLUDES)
except BadResponseError:
release = None
if release:
return release.getTracks(), release.getReleaseGroup().getId()
else:
return None
def _lucene_escape(text):
"""Escapes a string so it may be used verbatim in a Lucene query
string.
"""
# Regex stolen from MusicBrainz Picard.
out = re.sub(r'([+\-&|!(){}\[\]\^"~*?:\\])', r'\\\1', text)
return out.replace('\x00', '')
def _lucene_query(criteria):
"""Given a dictionary containing search criteria, produce a string
that may be used as a MusicBrainz search query.
"""
query_parts = []
for name, value in criteria.items():
value = _lucene_escape(value).strip().lower()
if value:
query_parts.append(u'%s:(%s)' % (name, value))
return u' '.join(query_parts)
def find_releases(criteria, limit=SEARCH_LIMIT):
"""Get a list of release dictionaries from the MusicBrainz
database that match `criteria`. The latter is a dictionary whose
keys are MusicBrainz field names and whose values are search terms
for those fields.
The field names are from MusicBrainz's Lucene query syntax, which
is detailed here:
http://wiki.musicbrainz.org/Text_Search_Syntax
"""
# Replace special cases.
if 'artist' in criteria:
artist = criteria['artist']
if artist in SPECIAL_CASE_ARTISTS:
del criteria['artist']
criteria['arid'] = SPECIAL_CASE_ARTISTS[artist]
# Build the filter and send the query.
if any(criteria.itervalues()):
query = _lucene_query(criteria)
log.debug('album query: %s' % query)
return get_releases(limit=limit, query=query)
for artist, artist_id in SPECIAL_CASE_ARTISTS.items():
if criteria['artist'] == artist:
criteria['arid'] = artist_id
del criteria['artist']
break
def find_tracks(criteria, limit=SEARCH_LIMIT):
"""Get a sequence of track dictionaries from MusicBrainz that match
`criteria`, a search term dictionary similar to the one passed to
`find_releases`.
def track_info(recording):
"""Translates a MusicBrainz recording result dictionary into a beets
``TrackInfo`` object.
"""
if any(criteria.itervalues()):
query = _lucene_query(criteria)
log.debug('track query: %s' % query)
filt = mbws.TrackFilter(limit=limit, query=query)
try:
results = _query_wrap(mbws.Query().getTracks, filter=filt)
except BadResponseError:
results = ()
for result in results:
track = result.track
yield track_dict(track)
info = lib.beets.autotag.hooks.TrackInfo(recording['title'],
recording['id'])
def track_dict(track):
"""Produces a dictionary summarizing a MusicBrainz `Track` object.
"""
t = {'title': track.title,
'id': track.id.rsplit('/', 1)[1]}
if track.artist is not None:
# Track artists will only be present for releases with
# multiple artists.
t['artist'] = track.artist.name
t['artist_id'] = track.artist.id.rsplit('/', 1)[1]
if track.duration is not None:
# Duration not always present.
t['length'] = track.duration/(1000.0)
return t
if 'artist-credit' in recording: # XXX: when is this not included?
artist = recording['artist-credit'][0]['artist']
info.artist = artist['name']
info.artist_id = artist['id']
def release_dict(release, tracks=None):
"""Takes a MusicBrainz `Release` object and returns a dictionary
containing the interesting data about that release. A list of
`Track` objects may also be provided as `tracks`; they are then
included in the resulting dictionary.
if recording.get('length'):
info.length = int(recording['length'])/(1000.0)
return info
def album_info(release):
"""Takes a MusicBrainz release result dictionary and returns a beets
AlbumInfo object containing the interesting data about that release.
"""
# Basic info.
out = {'album': release.title,
'album_id': release.id.rsplit('/', 1)[1],
'artist': release.artist.name,
'artist_id': release.artist.id.rsplit('/', 1)[1],
'asin': release.asin,
'albumtype': '',
}
out['va'] = out['artist_id'] == VARIOUS_ARTISTS_ID
artist = release['artist-credit'][0]['artist']
tracks = []
for medium in release['medium-list']:
tracks.extend(i['recording'] for i in medium['track-list'])
info = lib.beets.autotag.hooks.AlbumInfo(
release['title'],
release['id'],
artist['name'],
artist['id'],
[track_info(track) for track in tracks],
)
info.va = info.artist_id == VARIOUS_ARTISTS_ID
if 'asin' in release:
info.asin = release['asin']
# Release type not always populated.
for releasetype in release.types:
if releasetype in RELEASE_TYPES:
out['albumtype'] = releasetype.split('#')[1].lower()
break
reltype = release['release-group']['type']
if reltype:
info.albumtype = reltype.lower()
# Release date and label.
try:
event = release.getEarliestReleaseEvent()
except:
# The python-musicbrainz2 module has a bug that will raise an
# exception when there is no release date to be found. In this
# case, we just skip adding a release date to the dict.
pass
else:
if event:
# Release date.
date_str = event.getDate()
if date_str:
date_parts = date_str.split('-')
for key in ('year', 'month', 'day'):
if date_parts:
out[key] = int(date_parts.pop(0))
# 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)))
# Label name.
label = event.getLabel()
if label:
out['label'] = label.getName()
# Label name.
if release.get('label-info-list'):
label = release['label-info-list'][0]['label']['name']
if label != '[no label]':
info.label = label
# Tracks.
if tracks is not None:
out['tracks'] = map(track_dict, tracks)
return out
return info
def match_album(artist, album, tracks=None, limit=SEARCH_LIMIT):
"""Searches for a single album ("release" in MusicBrainz parlance)
and returns an iterator over dictionaries of information (as
returned by `release_dict`).
and returns an iterator over AlbumInfo objects.
The query consists of an artist name, an album name, and,
optionally, a number of tracks on the album.
@@ -292,42 +127,45 @@ def match_album(artist, album, tracks=None, limit=SEARCH_LIMIT):
if tracks is not None:
criteria['tracks'] = str(tracks)
# Search for the release.
return find_releases(criteria)
_adapt_criteria(criteria)
res = musicbrainz3.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'])
def match_track(artist, title):
"""Searches for a single track and returns an iterable of track
info dictionaries (as returned by `track_dict`).
def match_track(artist, title, limit=SEARCH_LIMIT):
"""Searches for a single track and returns an iterable of TrackInfo
objects.
"""
return find_tracks({
criteria = {
'artist': artist,
'track': title,
})
'recording': title,
}
_adapt_criteria(criteria)
res = musicbrainz3.recording_search(limit=limit, **criteria)
for recording in res['recording-list']:
yield track_info(recording)
def album_for_id(albumid):
"""Fetches an album by its MusicBrainz ID and returns an
information dictionary. If no match is found, returns None.
"""Fetches an album by its MusicBrainz ID and returns an AlbumInfo
object or None if the album is not found.
"""
query = mbws.Query()
try:
album = _query_wrap(query.getReleaseById, albumid, RELEASE_INCLUDES)
except BadResponseError:
res = musicbrainz3.get_release_by_id(albumid, RELEASE_INCLUDES)
except musicbrainz3.ResponseError:
log.debug('Album ID match failed.')
return None
except (mbws.ResourceNotFoundError, mbws.RequestError), exc:
log.debug('Album ID match failed: ' + str(exc))
return None
return release_dict(album, album.tracks)
return album_info(res['release'])
def track_for_id(trackid):
"""Fetches a track by its MusicBrainz ID. Returns a track info
dictionary or None if no track is found.
"""Fetches a track by its MusicBrainz ID. Returns a TrackInfo object
or None if no track is found.
"""
query = mbws.Query()
try:
track = _query_wrap(query.getTrackById, trackid, TRACK_INCLUDES)
except BadResponseError:
res = musicbrainz3.get_recording_by_id(trackid, TRACK_INCLUDES)
except musicbrainz3.ResponseError:
log.debug('Track ID match failed.')
return None
except (mbws.ResourceNotFoundError, mbws.RequestError), exc:
log.debug('Track ID match failed: ' + str(exc))
return None
return track_dict(track)
return track_info(res['recording'])
+171
View File
@@ -0,0 +1,171 @@
# 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.
"""Searches for albums in the MusicBrainz database.
"""
import logging
from . import musicbrainz3
import beets.autotag.hooks
import beets
SEARCH_LIMIT = 5
VARIOUS_ARTISTS_ID = '89ad4ac3-39f7-470e-963a-56509c546377'
musicbrainz3._useragent = 'beets/%s' % beets.__version__
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']
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
def track_info(recording):
"""Translates a MusicBrainz recording result dictionary into a beets
``TrackInfo`` object.
"""
info = beets.autotag.hooks.TrackInfo(recording['title'],
recording['id'])
if 'artist-credit' in recording: # XXX: when is this not included?
artist = recording['artist-credit'][0]['artist']
info.artist = artist['name']
info.artist_id = artist['id']
if recording.get('length'):
info.length = int(recording['length'])/(1000.0)
return info
def album_info(release):
"""Takes a MusicBrainz release result dictionary and returns a beets
AlbumInfo object containing the interesting data about that release.
"""
# Basic info.
artist = release['artist-credit'][0]['artist']
tracks = []
for medium in release['medium-list']:
tracks.extend(i['recording'] for i in medium['track-list'])
info = beets.autotag.hooks.AlbumInfo(
release['title'],
release['id'],
artist['name'],
artist['id'],
[track_info(track) for track in tracks],
)
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()
# 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)))
# Label name.
if release.get('label-info-list'):
label = release['label-info-list'][0]['label']['name']
if label != '[no label]':
info.label = label
return info
def match_album(artist, album, tracks=None, limit=SEARCH_LIMIT):
"""Searches for a single album ("release" in MusicBrainz parlance)
and returns an iterator over AlbumInfo objects.
The query consists of an artist name, an album name, and,
optionally, a number of tracks on the album.
"""
# Build search criteria.
criteria = {'release': album}
if artist is not None:
criteria['artist'] = artist
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)
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'])
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,
}
_adapt_criteria(criteria)
res = musicbrainz3.recording_search(limit=limit, **criteria)
for recording in res['recording-list']:
yield track_info(recording)
def album_for_id(albumid):
"""Fetches an album by its MusicBrainz ID and returns an AlbumInfo
object or None if the album is not found.
"""
try:
res = musicbrainz3.get_release_by_id(albumid, RELEASE_INCLUDES)
except musicbrainz3.ResponseError:
log.debug('Album ID match failed.')
return None
return album_info(res['release'])
def track_for_id(trackid):
"""Fetches a track by its MusicBrainz ID. Returns a TrackInfo object
or None if no track is found.
"""
try:
res = musicbrainz3.get_recording_by_id(trackid, TRACK_INCLUDES)
except musicbrainz3.ResponseError:
log.debug('Track ID match failed.')
return None
return track_info(res['recording'])
+744
View File
@@ -0,0 +1,744 @@
# 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
import urlparse
import urllib2
import urllib
import re
import threading
import time
import logging
import httplib
import xml.etree.ElementTree as etree
from . import mbxml
_useragent = "pythonmusicbrainzngs-0.1"
_log = logging.getLogger("python-musicbrainz-ngs")
# Constants for validation.
VALID_INCLUDES = {
'artist': [
"recordings", "releases", "release-groups", "works", # Subqueries
"various-artists", "discids", "media",
"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",
"aliases", "tags", "user-tags", "ratings", "user-ratings", # misc
"artist-rels", "label-rels", "recording-rels", "release-rels",
"release-group-rels", "url-rels", "work-rels"
],
'recording': [
"artists", "releases", # Subqueries
"discids", "media", "artist-credits",
"tags", "user-tags", "ratings", "user-ratings", # misc
"artist-rels", "label-rels", "recording-rels", "release-rels",
"release-group-rels", "url-rels", "work-rels"
],
'release': [
"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"
],
'release-group': [
"artists", "releases", "discids", "media",
"artist-credits", "tags", "user-tags", "ratings", "user-ratings", # misc
"artist-rels", "label-rels", "recording-rels", "release-rels",
"release-group-rels", "url-rels", "work-rels"
],
'work': [
"artists", # Subqueries
"aliases", "tags", "user-tags", "ratings", "user-ratings", # misc
"artist-rels", "label-rels", "recording-rels", "release-rels",
"release-group-rels", "url-rels", "work-rels"
],
'discid': [
"artists", "labels", "recordings", "release-groups", "puids",
"echoprints", "isrcs"
],
'echoprint': ["artists", "releases"],
'puid': ["artists", "releases", "puids", "echoprints", "isrcs"],
'isrc': ["artists", "releases", "puids", "echoprints", "isrcs"],
'iswc': ["artists"],
}
VALID_RELEASE_TYPES = [
"nat", "album", "single", "ep", "compilation", "soundtrack", "spokenword",
"interview", "audiobook", "live", "remix", "other"
]
VALID_RELEASE_STATUSES = ["official", "promotion", "bootleg", "pseudo-release"]
VALID_SEARCH_FIELDS = {
'artist': [
'arid', 'artist', 'sortname', 'type', 'begin', 'end', 'comment',
'alias', 'country', 'gender', 'tag'
],
'release-group': [
'rgid', 'releasegroup', 'reid', 'release', 'arid', 'artist',
'artistname', 'creditname', 'type', 'tag'
],
'release': [
'reid', 'release', 'arid', 'artist', 'artistname', 'creditname',
'type', 'status', 'tracks', 'tracksmedium', 'discids',
'discidsmedium', 'mediums', 'date', 'asin', 'lang', 'script',
'country', 'date', 'label', 'catno', 'barcode', 'puid'
],
'recording': [
'rid', 'recording', 'isrc', 'arid', 'artist', 'artistname',
'creditname', 'reid', 'release', 'type', 'status', 'tracks',
'tracksrelease', 'dur', 'qdur', 'tnum', 'position', 'tag'
],
'label': [
'laid', 'label', 'sortname', 'type', 'code', 'country', 'begin',
'end', 'comment', 'alias', 'tag'
],
'work': [
'wid', 'work', 'iswc', 'type', 'arid', 'artist', 'alias', 'tag'
],
}
# Exceptions.
class MusicBrainzError(Exception):
"""Base class for all exceptions related to MusicBrainz."""
pass
class UsageError(MusicBrainzError):
"""Error related to misuse of the module API."""
pass
class InvalidSearchFieldError(UsageError):
pass
class InvalidIncludeError(UsageError):
def __init__(self, msg='Invalid Includes', reason=None):
super(InvalidIncludeError, self).__init__(self)
self.msg = msg
self.reason = reason
def __str__(self):
return self.msg
class InvalidFilterError(UsageError):
def __init__(self, msg='Invalid Includes', reason=None):
super(InvalidFilterError, self).__init__(self)
self.msg = msg
self.reason = reason
def __str__(self):
return self.msg
class WebServiceError(MusicBrainzError):
"""Error related to MusicBrainz API requests."""
def __init__(self, message=None, cause=None):
"""Pass ``cause`` if this exception was caused by another
exception.
"""
self.message = message
self.cause = cause
def __str__(self):
if self.message:
msg = "%s, " % self.message
else:
msg = ""
msg += "caused by: %s" % str(self.cause)
return msg
class NetworkError(WebServiceError):
"""Problem communicating with the MB server."""
pass
class ResponseError(WebServiceError):
"""Bad response sent by the MB server."""
pass
# Helpers for validating and formatting allowed sets.
def _check_includes_impl(includes, valid_includes):
for i in includes:
if i not in valid_includes:
raise InvalidIncludeError("Bad includes", "%s is not a valid include" % i)
def _check_includes(entity, inc):
_check_includes_impl(inc, VALID_INCLUDES[entity])
def _check_filter(values, valid):
for v in values:
if v not in valid:
raise InvalidFilterError(v)
def _check_filter_and_make_params(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.
"""
if isinstance(release_status, basestring):
release_status = [release_status]
if isinstance(release_type, basestring):
release_type = [release_type]
_check_filter(release_status, VALID_RELEASE_STATUSES)
_check_filter(release_type, VALID_RELEASE_TYPES)
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):
raise InvalidFilterError("Can't have a release type with no "
"release-group include")
# Build parameters.
params = {}
if len(release_status):
params["status"] = "|".join(release_status)
if len(release_type):
params["type"] = "|".join(release_type)
return params
# Global authentication and endpoint details.
user = password = ""
hostname = "musicbrainz.org"
_client = ""
def auth(u, p):
"""Set the username and password to be used in subsequent queries to
the MusicBrainz XML API that require authentication.
"""
global user, password
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
# Rate limiting.
limit_interval = 1.0
limit_requests = 1
def set_rate_limit(new_interval=1.0, new_requests=1):
"""Sets the rate limiting behavior of the module. Must be invoked
before the first Web service call. Specify the number of requests
(`new_requests`) that may be made per given interval
(`new_interval`).
"""
global limit_interval
global limit_requests
limit_interval = new_interval
limit_requests = new_requests
class _rate_limit(object):
"""A decorator that limits the rate at which the function may be
called. The rate is controlled by the `limit_interval` and
`limit_requests` global variables. The limiting is thread-safe;
only one thread may be in the function at a time (acts like a
monitor in this sense). The globals must be set before the first
call to the limited function.
"""
def __init__(self, fun):
self.fun = fun
self.last_call = 0.0
self.lock = threading.Lock()
self.remaining_requests = None # Set on first invocation.
def _update_remaining(self):
"""Update remaining requests based on the elapsed time since
they were last calculated.
"""
# On first invocation, we have the maximum number of requests
# available.
if self.remaining_requests is None:
self.remaining_requests = float(limit_requests)
else:
since_last_call = time.time() - self.last_call
self.remaining_requests += since_last_call * \
(limit_requests / limit_interval)
self.remaining_requests = min(self.remaining_requests,
float(limit_requests))
self.last_call = time.time()
def __call__(self, *args, **kwargs):
with self.lock:
self._update_remaining()
# Delay if necessary.
while self.remaining_requests < 0.999:
time.sleep((1.0 - self.remaining_requests) *
(limit_requests / limit_interval))
self._update_remaining()
# Call the original function, "paying" for this call.
self.remaining_requests -= 1.0
return self.fun(*args, **kwargs)
# Generic support for making HTTP requests.
# From pymb2
class _RedirectPasswordMgr(urllib2.HTTPPasswordMgr):
def __init__(self):
self._realms = { }
def find_user_password(self, realm, uri):
# ignoring the uri parameter intentionally
try:
return self._realms[realm]
except KeyError:
return (None, None)
def add_password(self, realm, uri, username, password):
# ignoring the uri parameter intentionally
self._realms[realm] = (username, password)
class _DigestAuthHandler(urllib2.HTTPDigestAuthHandler):
def get_authorization (self, req, chal):
qop = chal.get ('qop', None)
if qop and ',' in qop and 'auth' in qop.split (','):
chal['qop'] = 'auth'
return urllib2.HTTPDigestAuthHandler.get_authorization (self, req, chal)
class _MusicbrainzHttpRequest(urllib2.Request):
""" A custom request handler that allows DELETE and PUT"""
def __init__(self, method, url, data=None):
urllib2.Request.__init__(self, url, data)
allowed_m = ["GET", "POST", "DELETE", "PUT"]
if method not in allowed_m:
raise ValueError("invalid method: %s" % method)
self.method = method
def get_method(self):
return self.method
# Core (internal) functions for calling the MB API.
def _safe_open(opener, req, body=None, max_retries=8, retry_delay_delta=2.0):
"""Open an HTTP request with a given URL opener and (optionally) a
request body. Transient errors lead to retries. Permanent errors
and repeated errors are translated into a small set of handleable
exceptions. Returns a file-like object.
"""
last_exc = None
for retry_num in range(max_retries):
if retry_num: # Not the first try: delay an increasing amount.
_log.debug("retrying after delay (#%i)" % retry_num)
time.sleep(retry_num * retry_delay_delta)
try:
if body:
f = opener.open(req, body)
else:
f = opener.open(req)
except urllib2.HTTPError, exc:
if exc.code in (400, 404):
# Bad request, not found, etc.
raise ResponseError(cause=exc)
elif exc.code in (503, 502, 500):
# Rate limiting, internal overloading...
_log.debug("HTTP error %i" % exc.code)
else:
# Other, unknown error. Should handle more cases, but
# retrying for now.
_log.debug("unknown HTTP error %i" % exc.code)
last_exc = exc
except httplib.BadStatusLine, exc:
_log.debug("bad status line")
last_exc = exc
except httplib.HTTPException, exc:
_log.debug("miscellaneous HTTP exception: %s" % str(exc))
last_exc = exc
except urllib2.URLError, exc:
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)
@_rate_limit
def _mb_request(path, method='GET', auth_required=False, client_required=False,
args=None, data=None, body=None):
"""Makes a request for the specified `path` (endpoint) on /ws/2 on
the globally-specified hostname. Parses the responses and returns
the resulting object. `auth_required` and `client_required` control
whether exceptions should be raised if the client and
username/password are left unspecified, respectively.
"""
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:
args["client"] = _client
# Construct the full URL for the request, including hostname and
# query string.
url = urlparse.urlunparse((
'http',
hostname,
'/ws/2/%s' % path,
'',
urllib.urlencode(args),
''
))
_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:
if not user:
raise UsageError("authorization required; "
"use musicbrainz.auth(u, p) first")
passwordMgr = _RedirectPasswordMgr()
authHandler = _DigestAuthHandler(passwordMgr)
authHandler.add_password("musicbrainz.org", (), user, password)
handlers.append(authHandler)
# Make request.
req = _MusicbrainzHttpRequest(method, url, data)
req.add_header('User-Agent', _useragent)
if body:
req.add_header('Content-Type', 'application/xml; charset=UTF-8')
f = _safe_open(opener, req, body)
# Parse the response.
try:
return mbxml.parse_message(f)
except etree.ParseError, exc:
raise ResponseError(cause=exc)
except UnicodeError, exc:
raise ResponseError(cause=exc)
def _is_auth_required(entity, includes):
""" Some calls require authentication. This returns
True if a call does, False otherwise
"""
if "user-tags" in includes or "user-ratings" in includes:
return True
elif entity.startswith("collection"):
return True
else:
return False
def _do_mb_query(entity, id, includes=[], params={}):
"""Make a single GET call to the MusicBrainz XML API. `entity` is a
string indicated the type of object to be retrieved. The id may be
empty, in which case the query is a search. `includes` is a list
of strings that must be valid includes for the entity type. `params`
is a dictionary of additional parameters for the API call. The
response is parsed and returned.
"""
# Build arguments.
_check_includes(entity, includes)
auth_required = _is_auth_required(entity, includes)
args = dict(params)
if len(includes) > 0:
inc = " ".join(includes)
args["inc"] = inc
# Build the endpoint components.
path = '%s/%s' % (entity, id)
return _mb_request(path, 'GET', auth_required, args=args)
def _do_mb_search(entity, query='', fields={}, limit=None, offset=None):
"""Perform a full-text search on the MusicBrainz search server.
`query` is a free-form query string and `fields` is a dictionary
of key/value query parameters. They keys in `fields` must be valid
for the given entity type.
"""
# Encode the query terms as a Lucene query string.
query_parts = [query.replace('\x00', '').strip()]
for key, value in fields.iteritems():
# Ensure this is a valid search field.
if key not in VALID_SEARCH_FIELDS[entity]:
raise InvalidSearchFieldError(
'%s is not a valid search field for %s' % (key, entity)
)
# Escape Lucene's special characters.
value = re.sub(r'([+\-&|!(){}\[\]\^"~*?:\\])', r'\\\1', value)
value = value.replace('\x00', '').strip()
if value:
query_parts.append(u'%s:(%s)' % (key, value))
full_query = u' '.join(query_parts).strip()
if not full_query:
raise ValueError('at least one query term is required')
# Additional parameters to the search.
params = {'query': full_query}
if limit:
params['limit'] = str(limit)
if offset:
params['offset'] = str(offset)
return _do_mb_query(entity, '', [], params)
def _do_mb_delete(path):
"""Send a DELETE request for the specified object.
"""
return _mb_request(path, 'DELETE', True, True)
def _do_mb_put(path):
"""Send a PUT request for the specified object.
"""
return _mb_request(path, 'PUT', True, True)
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)
# The main interface!
# Single entity by ID
def get_artist_by_id(id, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("artist", id, includes, params)
def get_label_by_id(id, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("label", id, includes, params)
def get_recording_by_id(id, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("recording", id, includes, params)
def get_release_by_id(id, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("release", id, includes, params)
def get_release_group_by_id(id, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("release-group", id, includes, params)
def get_work_by_id(id, includes=[]):
return _do_mb_query("work", id, includes)
# Searching
def artist_search(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,
gender, tag
"""
return _do_mb_search('artist', query, fields, limit, offset)
def label_search(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,
alias, tag
"""
return _do_mb_search('label', query, fields, limit, offset)
def recording_search(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,
release, type, status, tracks, tracksrelease, dur, qdur, tnum,
position, tag
"""
return _do_mb_search('recording', query, fields, limit, offset)
def release_search(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,
tracks, tracksmedium, discids, discidsmedium, mediums, date, asin,
lang, script, country, date, label, catno, barcode, puid
"""
return _do_mb_search('release', query, fields, limit, offset)
def release_group_search(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,
creditname, type, tag
"""
return _do_mb_search('release-group', query, fields, limit, offset)
def work_search(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
"""
return _do_mb_search('work', query, fields, limit, offset)
# Lists of entities
def get_releases_by_discid(id, includes=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_type=release_type)
return _do_mb_query("discid", id, includes, params)
def get_recordings_by_echoprint(echoprint, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("echoprint", echoprint, includes, params)
def get_recordings_by_puid(puid, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("puid", puid, includes, params)
def get_recordings_by_isrc(isrc, includes=[], release_status=[], release_type=[]):
params = _check_filter_and_make_params(includes, release_status, release_type)
return _do_mb_query("isrc", isrc, includes, params)
def get_works_by_iswc(iswc, includes=[]):
return _do_mb_query("iswc", iswc, includes)
# 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):
# 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)
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_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_release(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)
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)
# browse_work is defined in the docs but has no browse criteria
# Collections
def get_all_collections():
# Missing <release-list count="n"> the count in the reply
return _do_mb_query("collection", '')
def get_releases_in_collection(collection):
return _do_mb_query("collection", "%s/releases" % collection)
# Submission methods
def submit_barcodes(barcodes):
"""
Submits a set of {release1: barcode1, release2:barcode2}
Must call auth(user, pass) first
"""
query = mbxml.make_barcode_request(barcodes)
return _do_mb_post("release", query)
def submit_puids(puids):
query = mbxml.make_puid_request(puids)
return _do_mb_post("recording", query)
def submit_echoprints(echoprints):
query = mbxml.make_echoprint_request(echoprints)
return _do_mb_post("recording", query)
def submit_isrcs(isrcs):
raise NotImplementedError
def submit_tags(artist_tags={}, recording_tags={}):
""" Submit user tags.
Artist or recording parameters are of the form:
{'entityid': [taglist]}
"""
query = mbxml.make_tag_request(artist_tags, recording_tags)
return _do_mb_post("tag", query)
def submit_ratings(artist_ratings={}, recording_ratings={}):
""" Submit user ratings.
Artist or recording parameters are of the form:
{'entityid': rating}
"""
query = mbxml.make_rating_request(artist_ratings, recording_ratings)
return _do_mb_post("rating", query)
def add_releases_to_collection(collection, releases=[]):
# XXX: Maximum URI length of 16kb means we should only allow ~400 releases
releaselist = ";".join(releases)
_do_mb_put("collection/%s/releases/%s" % (collection, releaselist))
def remove_releases_from_collection(collection, releases=[]):
releaselist = ";".join(releases)
_do_mb_delete("collection/%s/releases/%s" % (collection, releaselist))
+545
View File
@@ -0,0 +1,545 @@
import xml.etree.ElementTree as ET
import string
import StringIO
import logging
try:
from ET import fixtag
except:
# Python < 2.7
def fixtag(tag, namespaces):
# given a decorated tag (of the form {uri}tag), return prefixed
# tag and namespace declaration, if any
if isinstance(tag, ET.QName):
tag = tag.text
namespace_uri, tag = string.split(tag[1:], "}", 1)
prefix = namespaces.get(namespace_uri)
if prefix is None:
prefix = "ns%d" % len(namespaces)
namespaces[namespace_uri] = prefix
if prefix == "xml":
xmlns = None
else:
xmlns = ("xmlns:%s" % prefix, namespace_uri)
else:
xmlns = None
return "%s:%s" % (prefix, tag), xmlns
NS_MAP = {"http://musicbrainz.org/ns/mmd-2.0#": "ws2"}
def make_artist_credit(artists):
names = []
for artist in artists:
if isinstance(artist, dict):
names.append(artist.get("artist", {}).get("name", ""))
else:
names.append(artist)
return "".join(names)
def parse_elements(valid_els, element):
""" Extract single level subelements from an element.
For example, given the element:
<element>
<subelement>Text</subelement>
</element>
and a list valid_els that contains "subelement",
return a dict {'subelement': 'Text'}
"""
result = {}
for sub in element:
t = fixtag(sub.tag, NS_MAP)[0]
if ":" in t:
t = t.split(":")[1]
if t in valid_els:
result[t] = sub.text
else:
logging.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
def parse_inner(inner_els, element):
""" Delegate the parsing of a subelement to another function.
For example, given the element:
<element>
<subelement>
<a>Foo</a><b>Bar</b>
</subelement>
</element>
and a dictionary {'subelement': parse_subelement},
call parse_subelement(<subelement>) and
return a dict {'subelement': <result>}
if parse_subelement returns a tuple of the form
('subelement-key', <result>) then return a dict
{'subelement-key': <result>} instead
"""
result = {}
for sub in element:
t = fixtag(sub.tag, NS_MAP)[0]
if ":" in t:
t = t.split(":")[1]
if t in inner_els.keys():
inner_result = inner_els[t](sub)
if isinstance(inner_result, tuple):
result[inner_result[0]] = inner_result[1]
else:
result[t] = inner_result
else:
logging.debug("in <%s>, not delegating <%s>", fixtag(element.tag, NS_MAP)[0], t)
return result
def parse_message(message):
s = message.read()
f = StringIO.StringIO(s)
tree = ET.ElementTree(file=f)
root = tree.getroot()
result = {}
valid_elements = {"artist": parse_artist,
"label": parse_label,
"release": parse_release,
"release-group": parse_release_group,
"recording": parse_recording,
"work": parse_work,
"disc": parse_disc,
"puid": parse_puid,
"echoprint": parse_puid,
"artist-list": parse_artist_list,
"label-list": parse_label_list,
"release-list": parse_release_list,
"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,
"message": parse_response_message
}
result.update(parse_inner(valid_elements, root))
return result
def parse_response_message(message):
return parse_elements(["text"], message)
def parse_collection_list(cl):
return [parse_collection(c) for c in cl]
def parse_collection(collection):
result = {}
attribs = ["id"]
elements = ["name", "editor"]
inner_els = {"release-list": parse_release_list}
result.update(parse_attributes(attribs, collection))
result.update(parse_elements(elements, collection))
result.update(parse_inner(inner_els, collection))
return result
def parse_collection_release_list(rl):
attribs = ["count"]
return parse_attributes(attribs, rl)
def parse_artist_lifespan(lifespan):
parts = parse_elements(["begin", "end"], lifespan)
beginval = parts.get("begin", "")
endval = parts.get("end", "")
return (beginval, endval)
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"]
inner_els = {"life-span": parse_artist_lifespan,
"recording-list": parse_recording_list,
"release-list": parse_release_list,
"release-group-list": parse_release_group_list,
"work-list": parse_work_list,
"tag-list": parse_tag_list,
"user-tag-list": parse_tag_list,
"rating": parse_rating,
"alias-list": parse_alias_list}
result.update(parse_attributes(attribs, artist))
result.update(parse_elements(elements, artist))
result.update(parse_inner(inner_els, artist))
return result
def parse_label_list(ll):
return [parse_label(l) for l in ll]
def parse_label(label):
result = {}
attribs = ["id", "type"]
elements = ["name", "sort-name", "country", "label-code", "user-rating"]
inner_els = {"life-span": parse_artist_lifespan,
"release-list": parse_release_list,
"tag-list": parse_tag_list,
"user-tag-list": parse_tag_list,
"rating": parse_rating,
"alias-list": parse_alias_list}
result.update(parse_attributes(attribs, label))
result.update(parse_elements(elements, label))
result.update(parse_inner(inner_els, label))
return result
def parse_attribute_list(al):
return [parse_attribute_tag(a) for a in al]
def parse_attribute_tag(attribute):
return attribute.text
def parse_relation_list(rl):
attribs = ["target-type"]
ttype = parse_attributes(attribs, rl)
key = "%s-relation-list" % ttype["target-type"]
return (key, [parse_relation(r) for r in rl])
def parse_relation(relation):
result = {}
attribs = ["type"]
elements = ["target", "direction"]
inner_els = {"artist": parse_artist,
"label": parse_label,
"recording": parse_recording,
"release": parse_release,
"release-group": parse_release_group,
"attribute-list": parse_attribute_list,
"work": parse_work
}
result.update(parse_attributes(attribs, relation))
result.update(parse_elements(elements, relation))
result.update(parse_inner(inner_els, relation))
return result
def parse_release(release):
result = {}
attribs = ["id"]
elements = ["title", "status", "disambiguation", "quality", "country", "barcode", "date", "packaging", "asin"]
inner_els = {"text-representation": parse_text_representation,
"artist-credit": parse_artist_credit,
"label-info-list": parse_label_info_list,
"medium-list": parse_medium_list,
"release-group": parse_release_group,
"relation-list": parse_relation_list}
result.update(parse_attributes(attribs, release))
result.update(parse_elements(elements, release))
result.update(parse_inner(inner_els, release))
if "artist-credit" in result:
result["artist-credit-phrase"] = make_artist_credit(result["artist-credit"])
return result
def parse_medium_list(ml):
return [parse_medium(m) for m in ml]
def parse_medium(medium):
result = {}
elements = ["position", "format", "title"]
inner_els = {"disc-list": parse_disc_list,
"track-list": parse_track_list}
result.update(parse_elements(elements, medium))
result.update(parse_inner(inner_els, medium))
return result
def parse_disc_list(dl):
return [parse_disc(d) for d in dl]
def parse_text_representation(textr):
return parse_elements(["language", "script"], textr)
def parse_release_group(rg):
result = {}
attribs = ["id", "type"]
elements = ["title", "user-rating", "first-release-date"]
inner_els = {"artist-credit": parse_artist_credit,
"release-list": parse_release_list,
"tag-list": parse_tag_list,
"user-tag-list": parse_tag_list,
"rating": parse_rating}
result.update(parse_attributes(attribs, rg))
result.update(parse_elements(elements, rg))
result.update(parse_inner(inner_els, rg))
if "artist-credit" in result:
result["artist-credit-phrase"] = make_artist_credit(result["artist-credit"])
return result
def parse_recording(recording):
result = {}
attribs = ["id"]
elements = ["title", "length", "user-rating"]
inner_els = {"artist-credit": parse_artist_credit,
"release-list": parse_release_list,
"tag-list": parse_tag_list,
"user-tag-list": parse_tag_list,
"rating": parse_rating,
"puid-list": parse_external_id_list,
"isrc-list": parse_external_id_list,
"echoprint-list": parse_external_id_list}
result.update(parse_attributes(attribs, recording))
result.update(parse_elements(elements, recording))
result.update(parse_inner(inner_els, recording))
if "artist-credit" in result:
result["artist-credit-phrase"] = make_artist_credit(result["artist-credit"])
return result
def parse_external_id_list(pl):
return [parse_attributes(["id"], p)["id"] for p in pl]
def parse_work_list(wl):
result = []
for w in wl:
result.append(parse_work(w))
return result
def parse_work(work):
result = {}
attribs = ["id"]
elements = ["title", "user-rating"]
inner_els = {"tag-list": parse_tag_list,
"user-tag-list": parse_tag_list,
"rating": parse_rating,
"alias-list": parse_alias_list}
result.update(parse_attributes(attribs, work))
result.update(parse_elements(elements, work))
result.update(parse_inner(inner_els, work))
return result
def parse_disc(disc):
result = {}
attribs = ["id"]
elements = ["sectors"]
inner_els = {"release-list": parse_release_list}
result.update(parse_attributes(attribs, disc))
result.update(parse_elements(elements, disc))
result.update(parse_inner(inner_els, disc))
return result
def parse_release_list(rl):
result = []
for r in rl:
result.append(parse_release(r))
return result
def parse_release_group_list(rgl):
result = []
for rg in rgl:
result.append(parse_release_group(rg))
return result
def parse_puid(puid):
result = {}
attribs = ["id"]
inner_els = {"recording-list": parse_recording_list}
result.update(parse_attributes(attribs, puid))
result.update(parse_inner(inner_els, puid))
return result
def parse_recording_list(recs):
result = []
for r in recs:
result.append(parse_recording(r))
return result
def parse_artist_credit(ac):
result = []
for namecredit in ac:
result.append(parse_name_credit(namecredit))
join = parse_attributes(["joinphrase"], namecredit)
if "joinphrase" in join:
result.append(join["joinphrase"])
return result
def parse_name_credit(nc):
result = {}
elements = ["name"]
inner_els = {"artist": parse_artist}
result.update(parse_elements(elements, nc))
result.update(parse_inner(inner_els, nc))
return result
def parse_label_info_list(lil):
result = []
for li in lil:
result.append(parse_label_info(li))
return result
def parse_label_info(li):
result = {}
elements = ["catalog-number"]
inner_els = {"label": parse_label}
result.update(parse_elements(elements, li))
result.update(parse_inner(inner_els, li))
return result
def parse_track_list(tl):
result = []
for t in tl:
result.append(parse_track(t))
return result
def parse_track(track):
result = {}
elements = ["position"]
inner_els = {"recording": parse_recording}
result.update(parse_elements(elements, track))
result.update(parse_inner(inner_els, track))
return result
def parse_tag_list(tl):
result = []
for t in tl:
result.append(parse_tag(t))
return result
def parse_tag(tag):
result = {}
attribs = ["count"]
elements = ["name"]
result.update(parse_attributes(attribs, tag))
result.update(parse_elements(elements, tag))
return result
def parse_rating(rating):
result = {}
attribs = ["votes-count"]
result.update(parse_attributes(attribs, rating))
result["rating"] = rating.text
return result
def parse_alias_list(al):
result = []
for a in al:
result.append(a.text)
return result
###
def make_barcode_request(barcodes):
NS = "http://musicbrainz.org/ns/mmd-2.0#"
root = ET.Element("{%s}metadata" % NS)
rel_list = ET.SubElement(root, "{%s}release-list" % NS)
for release, barcode in barcodes.items():
rel_xml = ET.SubElement(rel_list, "{%s}release" % NS)
bar_xml = ET.SubElement(rel_xml, "{%s}barcode" % NS)
rel_xml.set("{%s}id" % NS, release)
bar_xml.text = barcode
return ET.tostring(root, "utf-8")
def make_puid_request(puids):
NS = "http://musicbrainz.org/ns/mmd-2.0#"
root = ET.Element("{%s}metadata" % NS)
rec_list = ET.SubElement(root, "{%s}recording-list" % NS)
for recording, puid_list in puids.items():
rec_xml = ET.SubElement(rec_list, "{%s}recording" % NS)
rec_xml.set("id", recording)
p_list_xml = ET.SubElement(rec_xml, "{%s}puid-list" % NS)
l = puid_list if isinstance(puid_list, list) else [puid_list]
for p in l:
p_xml = ET.SubElement(p_list_xml, "{%s}puid" % NS)
p_xml.set("id", p)
return ET.tostring(root, "utf-8")
def make_echoprint_request(echoprints):
NS = "http://musicbrainz.org/ns/mmd-2.0#"
root = ET.Element("{%s}metadata" % NS)
rec_list = ET.SubElement(root, "{%s}recording-list" % NS)
for recording, echoprint_list in echoprints.items():
rec_xml = ET.SubElement(rec_list, "{%s}recording" % NS)
rec_xml.set("id", recording)
e_list_xml = ET.SubElement(rec_xml, "{%s}echoprint-list" % NS)
l = echoprint_list if isinstance(echoprint_list, list) else [echoprint_list]
for e in l:
e_xml = ET.SubElement(e_list_xml, "{%s}echoprint" % NS)
e_xml.set("id", e)
return ET.tostring(root, "utf-8")
def make_tag_request(artist_tags, recording_tags):
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, tags in recording_tags.items():
rec_xml = ET.SubElement(rec_list, "{%s}recording" % NS)
rec_xml.set("{%s}id" % NS, rec)
taglist = ET.SubElement(rec_xml, "{%s}user-tag-list" % NS)
for t in tags:
usertag_xml = ET.SubElement(taglist, "{%s}user-tag" % NS)
name_xml = ET.SubElement(usertag_xml, "{%s}name" % NS)
name_xml.text = t
art_list = ET.SubElement(root, "{%s}artist-list" % NS)
for art, tags in artist_tags.items():
art_xml = ET.SubElement(art_list, "{%s}artist" % NS)
art_xml.set("{%s}id" % NS, art)
taglist = ET.SubElement(art_xml, "{%s}user-tag-list" % NS)
for t in tags:
usertag_xml = ET.SubElement(taglist, "{%s}user-tag" % NS)
name_xml = ET.SubElement(usertag_xml, "{%s}name" % NS)
name_xml.text = t
return ET.tostring(root, "utf-8")
def make_rating_request(artist_ratings, recording_ratings):
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, rating in recording_ratings.items():
rec_xml = ET.SubElement(rec_list, "{%s}recording" % NS)
rec_xml.set("{%s}id" % NS, rec)
rating_xml = ET.SubElement(rec_xml, "{%s}user-rating" % NS)
if isinstance(rating, int):
rating = "%d" % rating
rating_xml.text = rating
art_list = ET.SubElement(root, "{%s}artist-list" % NS)
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)
if isinstance(rating, int):
rating = "%d" % rating
rating_xml.text = rating
return ET.tostring(root, "utf-8")