abhin4v
/
python-lastfm
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Implementation of all read-only API methods is complete.

Abhinav Sarkar 2008-09-02 14:24:38 +00:00
commit 6cc59b39c5
21 changed files with 3859 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
setup.py Normal file
View File

@ -0,0 +1,55 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
METADATA = dict(
name='lastfm',
version='0.1',
description="a pure python interface to the Last.fm Webservices API",
long_description="""a pure python interface to the Last.fm Webservices API version 2.0,
located at http://ws.audioscrobbler.com/2.0/ .""",
author="Abhinav Sarkar",
author_email="abhinav.sarkar@gmail.com",
maintainer="Abhinav Sarkar",
maintainer_email="abhinav.sarkar@gmail.com",
url="http://python-lastfm.googlecode.com/svn/trunk/dist/",
package_dir = {'lastfm':'src'},
packages=['lastfm'],
license="GNU Lesser General Public License",
keywords="audioscrobbler webservice api last.fm",
)
SETUPTOOLS_METADATA = dict(
install_requires = ['setuptools'],
include_package_data = True,
classifiers = [
'Development Status :: 3 - Alpha',
'Intended Audience :: Developers',
'License :: OSI Approved :: GNU Library or Lesser General Public License (LGPL)',
'Operating System :: OS Independent',
'Programming Language :: Python',
'Topic :: Software Development :: Libraries :: Python Modules',
'Topic :: Multimedia :: Sound/Audio',
'Topic :: Internet',
],
)
import sys
if sys.version < '2.5':
SETUPTOOLS_METADATA['install_requires'].append('ElementTree')
SETUPTOOLS_METADATA['install_requires'].append('cElementTree')
def Main():
# Use setuptools if available, otherwise fallback and use distutils
try:
import setuptools
METADATA.update(SETUPTOOLS_METADATA)
setuptools.setup(**METADATA)
except ImportError:
import distutils.core
distutils.core.setup(**METADATA)
if __name__ == '__main__':
Main()

24
src/__init__.py Normal file
View File

@ -0,0 +1,24 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from album import Album
from api import Api
from artist import Artist
from base import LastfmBase
from error import LastfmError
from event import Event
from geo import Location, Country
from group import Group
from playlist import Playlist
from registry import Registry
from tag import Tag
from tasteometer import Tasteometer
from track import Track
from user import User
__all__ = ['LastfmError', 'Api', 'Album', 'Artist', 'Event',
'Location', 'Country', 'Group', 'Playlist', 'Tag',
'Tasteometer', 'Track', 'User', 'Registry']

212
src/album.py Normal file
View File

@ -0,0 +1,212 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
class Album(LastfmBase):
"""A class representing an album."""
def init(self,
api,
name = None,
artist = None,
id = None,
mbid = None,
url = None,
releaseDate = None,
image = None,
stats = None,
topTags = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__name = name
self.__artist = artist
self.__id = id
self.__mbid = mbid
self.__url = url
self.__releaseDate = releaseDate
self.__image = image
self.__stats = stats and Stats(
subject = self,
listeners = stats.listeners,
playcount = stats.playcount,
match = stats.match,
rank = stats.rank
)
self.__topTags = topTags
@property
def name(self):
"""name of the album"""
return self.__name
@property
def artist(self):
"""artist of the album"""
return self.__artist
@property
def id(self):
"""id of the album"""
if self.__id is None:
self._fillInfo()
return self.__id
@property
def mbid(self):
"""mbid of the album"""
if self.__mbid is None:
self._fillInfo()
return self.__mbid
@property
def url(self):
"""url of the album's page"""
if self.__url is None:
self._fillInfo()
return self.__url
@property
def releaseDate(self):
"""release date of the album"""
if self.__releaseDate is None:
self._fillInfo()
return self.__releaseDate
@property
def image(self):
"""cover images of the album"""
if self.__image is None:
self._fillInfo()
return self.__image
@property
def stats(self):
"""stats related to the album"""
if self.__stats is None:
self._fillInfo()
return self.__stats
@LastfmBase.cachedProperty
def topTags(self):
"""top tags for the album"""
params = {'method': 'album.getinfo'}
if self.artist and self.name:
params.update({'artist': self.artist.name, 'album': self.name})
elif self.mbid:
params.update({'mbid': self.mbid})
data = self.__api._fetchData(params).find('album')
return [
Tag(
self.__api,
subject = self,
name = t.findtext('name'),
url = t.findtext('url')
)
for t in data.findall('toptags/tag')
]
@LastfmBase.topProperty("topTags")
def topTag(self):
"""top tag for the album"""
pass
@LastfmBase.cachedProperty
def playlist(self):
return Playlist.fetch(self.__api, "lastfm://playlist/album/%s" % self.id)
@staticmethod
def _fetchData(api,
artist = None,
album = None,
mbid = None):
params = {'method': 'album.getinfo'}
if not ((artist and album) or mbid):
raise LastfmInvalidParametersError("either (artist and album) or mbid has to be given as argument.")
if artist and album:
params.update({'artist': artist, 'album': album})
elif mbid:
params.update({'mbid': mbid})
return api._fetchData(params).find('album')
def _fillInfo(self):
data = Album._fetchData(self.__api, self.artist.name, self.name)
self.__id = int(data.findtext('id'))
self.__mbid = data.findtext('mbid')
self.__url = data.findtext('url')
self.__releaseDate = data.findtext('releasedate') and data.findtext('releasedate').strip() and \
datetime(*(time.strptime(data.findtext('releasedate').strip(), '%d %b %Y, 00:00')[0:6]))
self.__image = dict([(i.get('size'), i.text) for i in data.findall('image')])
self.__stats = Stats(
subject = self,
listeners = int(data.findtext('listeners')),
playcount = int(data.findtext('playcount')),
)
self.__topTags = [
Tag(
self.__api,
subject = self,
name = t.findtext('name'),
url = t.findtext('url')
)
for t in data.findall('toptags/tag')
]
@staticmethod
def getInfo(api,
artist = None,
album = None,
mbid = None):
data = Album._fetchData(api, artist, album, mbid)
a = Album(
api,
name = data.findtext('name'),
artist = Artist(
api,
name = data.findtext('artist'),
),
)
if a.id is None:
a._fillInfo()
return a
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash("%s%s" % (kwds['name'], hash(kwds['artist'])))
except KeyError:
raise LastfmInvalidParametersError("name and artist have to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(name = self.name, artist = self.artist)
def __eq__(self, other):
if self.id and other.id:
return self.id == other.id
if self.mbid and other.mbid:
return self.mbid == other.mbid
if self.url and other.url:
return self.url == other.url
if (self.name and self.artist) and (other.name and other.artist):
return (self.name == other.name) and (self.artist == other.artist)
return super(Album, self).__eq__(other)
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.Album: '%s' by %s>" % (self.name, self.artist.name)
from datetime import datetime
import time
from api import Api
from artist import Artist
from error import LastfmInvalidParametersError
from playlist import Playlist
from stats import Stats
from tag import Tag

316
src/api.py Normal file
View File

@ -0,0 +1,316 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
class Api(object):
"""The class representing the last.fm web services API."""
DEFAULT_CACHE_TIMEOUT = 3600 # cache for 1 hour
API_ROOT_URL = "http://ws.audioscrobbler.com/2.0/"
FETCH_INTERVAL = 1
SEARCH_XMLNS = "http://a9.com/-/spec/opensearch/1.1/"
def __init__(self,
apiKey,
input_encoding=None,
request_headers=None,
no_cache = False,
debug = False):
self.__apiKey = apiKey
self._cache = FileCache()
self._urllib = urllib2
self._cache_timeout = Api.DEFAULT_CACHE_TIMEOUT
self._InitializeRequestHeaders(request_headers)
self._InitializeUserAgent()
self._input_encoding = input_encoding
self._no_cache = no_cache
self._debug = debug
self._lastFetchTime = datetime.now()
def getApiKey(self):
return self.__apiKey
def setCache(self, cache):
'''Override the default cache. Set to None to prevent caching.
Args:
cache: an instance that supports the same API as the audioscrobblerws.FileCache
'''
self._cache = cache
def setUrllib(self, urllib):
'''Override the default urllib implementation.
Args:
urllib: an instance that supports the same API as the urllib2 module
'''
self._urllib = urllib
def setCacheTimeout(self, cache_timeout):
'''Override the default cache timeout.
Args:
cache_timeout: time, in seconds, that responses should be reused.
'''
self._cache_timeout = cache_timeout
def setUserAgent(self, user_agent):
'''Override the default user agent
Args:
user_agent: a string that should be send to the server as the User-agent
'''
self._request_headers['User-Agent'] = user_agent
def _BuildUrl(self, url, path_elements=None, extra_params=None):
# Break url into consituent parts
(scheme, netloc, path, params, query, fragment) = urlparse.urlparse(url)
path = path.replace(' ', '+')
# Add any additional path elements to the path
if path_elements:
# Filter out the path elements that have a value of None
p = [i for i in path_elements if i]
if not path.endswith('/'):
path += '/'
path += '/'.join(p)
# Add any additional query parameters to the query string
if extra_params and len(extra_params) > 0:
extra_query = self._EncodeParameters(extra_params)
# Add it to the existing query
if query:
query += '&' + extra_query
else:
query = extra_query
# Return the rebuilt URL
return urlparse.urlunparse((scheme, netloc, path, params, query, fragment))
def _InitializeRequestHeaders(self, request_headers):
if request_headers:
self._request_headers = request_headers
else:
self._request_headers = {}
def _InitializeUserAgent(self):
user_agent = 'Python-urllib/%s (python-lastfm/%s)' % \
(self._urllib.__version__, __version__)
self.setUserAgent(user_agent)
def _GetOpener(self, url):
opener = self._urllib.build_opener()
opener.addheaders = self._request_headers.items()
return opener
def _Encode(self, s):
if self._input_encoding:
return unicode(s, self._input_encoding).encode('utf-8')
else:
return unicode(s).encode('utf-8')
def _EncodeParameters(self, parameters):
'''Return a string in key=value&key=value form
Values of None are not included in the output string.
Args:
parameters:
A dict of (key, value) tuples, where value is encoded as
specified by self._encoding
Returns:
A URL-encoded string in "key=value&key=value" form
'''
if parameters is None:
return None
else:
return urllib.urlencode(dict([(k, self._Encode(v)) for k, v in parameters.items() if v is not None]))
def getAlbum(self,
artist = None,
album = None,
mbid = None):
if isinstance(artist, Artist):
artist = artist.name
return Album.getInfo(self, artist, album, mbid)
def getArtist(self,
artist = None,
mbid = None):
return Artist.getInfo(self, artist, mbid)
def searchArtist(self,
artist,
limit = None):
return Artist.search(self, artist, limit)
def getEvent(self, event):
return Event.getInfo(self, event)
def getLocation(self, name):
return Location(self, name = name)
def getCountry(self, name):
return Country(self, name = name)
def getGroup(self, name):
return Group(self, name = name)
def fetchPlaylist(self, url):
return Playlist.fetch(self, url)
def getTag(self, name):
return Tag(self, name = name)
def getGlobalTopTags(self):
return Tag.getTopTags(self)
def searchTag(self,
tag,
limit = None):
return Tag.search(self, tag, limit)
def compareTaste(self,
type1, type2,
value1, value2,
limit = None):
return Tasteometer.compare(self, type1, type2, value1, value2, limit)
def getTrack(self, track, artist):
if isinstance(artist, Artist):
artist = artist.name
result = Track.search(self, track, artist)
if len(result.matches) == 0:
raise LastfmInvalidResourceError("'%s' by %s: no such track found" % (track, artist))
return result.matches[0]
def searchTrack(self,
track,
artist = None,
limit = None):
if isinstance(artist, Artist):
artist = artist.name
return Track.search(self, track, artist, limit)
def getUser(self, name):
user = None
try:
user = User(self, name = name)
user.friends
except LastfmError, e:
raise e
return user
def _fetchUrl(self,
url,
parameters = None,
no_cache = False):
'''Fetch a URL, optionally caching for a specified time.
Args:
url: The URL to retrieve
parameters: A dict of key/value pairs that should added to
the query string. [OPTIONAL]
no_cache: If true, overrides the cache on the current request
Returns:
A string containing the body of the response.
'''
# Add key/value parameters to the query string of the url
url = self._BuildUrl(url, extra_params=parameters)
if self._debug:
print url
# Get a url opener that can handle basic auth
opener = self._GetOpener(url)
def readUrlData():
now = datetime.now()
delta = now - self._lastFetchTime
delta = delta.seconds + float(delta.microseconds)/1000000
if delta < Api.FETCH_INTERVAL:
time.sleep(Api.FETCH_INTERVAL - delta)
url_data = opener.open(url).read()
self._lastFetchTime = datetime.now()
return url_data
# Open and return the URL immediately if we're not going to cache
if no_cache or not self._cache or not self._cache_timeout:
try:
url_data = readUrlData()
except urllib2.HTTPError, e:
url_data = e.read()
else:
# Unique keys are a combination of the url and the username
key = url.encode('utf-8')
# See if it has been cached before
last_cached = self._cache.GetCachedTime(key)
# If the cached version is outdated then fetch another and store it
if not last_cached or time.time() >= last_cached + self._cache_timeout:
try:
url_data = readUrlData()
except urllib2.HTTPError, e:
url_data = e.read()
self._cache.Set(key, url_data)
else:
url_data = self._cache.Get(key)
# Always return the latest version
return url_data
def _fetchData(self,
params,
no_cache = False):
params.update({'api_key': self.__apiKey})
xml = self._fetchUrl(Api.API_ROOT_URL, params, no_cache = self._no_cache or no_cache)
#print xml
try:
data = ElementTree.XML(xml)
except SyntaxError, e:
raise LastfmOperationFailedError("Error in parsing XML: %s" % e)
if data.get('status') != "ok":
code = int(data.find("error").get('code'))
message = data.findtext('error')
if code in errorMap.keys():
raise errorMap[code](message, code)
else:
raise LastfmError(message, code)
return data
def __repr__(self):
return "<lastfm.Api: %s>" % self.__apiKey
from datetime import datetime
import sys
import time
import urllib
import urllib2
import urlparse
from album import Album
from artist import Artist
from error import errorMap, LastfmError, LastfmOperationFailedError, LastfmInvalidResourceError
from event import Event
from filecache import FileCache
from geo import Location, Country
from group import Group
from playlist import Playlist
from tag import Tag
from tasteometer import Tasteometer
from track import Track
from user import User
if sys.version.startswith('2.5'):
import xml.etree.cElementTree as ElementTree
else:
try:
import cElementTree as ElementTree
except ImportError:
try:
import ElementTree
except ImportError:
raise LastfmError("Install ElementTree package for using python-lastfm")

419
src/artist.py Normal file
View File

@ -0,0 +1,419 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
from lazylist import lazylist
class Artist(LastfmBase):
"""A class representing an artist."""
def init(self,
api,
name = None,
mbid = None,
url = None,
image = None,
streamable = None,
stats = None,
similar = None,
topTags = None,
bio = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__name = name
self.__mbid = mbid
self.__url = url
self.__image = image
self.__streamable = streamable
self.__stats = stats and Stats(
subject = self,
listeners = stats.listeners,
playcount = stats.playcount,
match = stats.match,
rank = stats.rank
)
self.__similar = similar
self.__topTags = topTags
self.__bio = bio and Artist.Bio(
artist = self,
published = bio.published,
summary = bio.summary,
content = bio.content
)
@property
def name(self):
"""name of the artist"""
return self.__name
@property
def mbid(self):
"""mbid of the artist"""
if self.__mbid is None:
self._fillInfo()
return self.__mbid
@property
def url(self):
"""url of the artist's page"""
if self.__url is None:
self._fillInfo()
return self.__url
@property
def image(self):
"""images of the artist"""
if self.__image is None:
self._fillInfo()
return self.__image
@property
def streamable(self):
"""is the artist streamable"""
if self.__streamable is None:
self._fillInfo()
return self.__streamable
@property
def stats(self):
"""stats for the artist"""
if self.__stats is None:
self._fillInfo()
return self.__stats
def getSimilar(self, limit = None):
params = {'method': 'artist.getsimilar', 'artist': self.__name}
if limit is not None:
params.update({'limit': limit})
data = self.__api._fetchData(params).find('similarartists')
self.__similar = [
Artist(
self.__api,
subject = self,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
stats = Stats(
subject = a.findtext('name'),
match = float(a.findtext('match')),
),
url = 'http://' + a.findtext('url'),
image = {'large': a.findtext('image')}
)
for a in data.findall('artist')
]
return self.__similar
@property
def similar(self):
"""artists similar to this artist"""
if self.__similar is None or len(self.__similar) < 6:
return self.getSimilar()
return self.__similar
@LastfmBase.topProperty("similar")
def mostSimilar(self):
"""artist most similar to this artist"""
pass
@property
def topTags(self):
"""top tags for the artist"""
if self.__topTags is None or len(self.__topTags) < 6:
params = {
'method': 'artist.gettoptags',
'artist': self.__name
}
data = self.__api._fetchData(params).find('toptags')
self.__topTags = [
Tag(
self.__api,
subject = self,
name = t.findtext('name'),
url = t.findtext('url')
)
for t in data.findall('tag')
]
return self.__topTags
@LastfmBase.topProperty("topTags")
def topTag(self):
"""top tag for the artist"""
pass
@property
def bio(self):
"""biography of the artist"""
if self.__bio is None:
self._fillInfo()
return self.__bio
@LastfmBase.cachedProperty
def events(self):
"""events for the artist"""
params = {'method': 'artist.getevents', 'artist': self.name}
data = self.__api._fetchData(params).find('events')
return [
Event.createFromData(self.__api, e)
for e in data.findall('event')
]
@LastfmBase.cachedProperty
def topAlbums(self):
"""top albums of the artist"""
params = {'method': 'artist.gettopalbums', 'artist': self.name}
data = self.__api._fetchData(params).find('topalbums')
return [
Album(
self.__api,
subject = self,
name = a.findtext('name'),
artist = self,
mbid = a.findtext('mbid'),
url = a.findtext('url'),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
stats = Stats(
subject = a.findtext('name'),
playcount = int(a.findtext('playcount')),
rank = int(a.attrib['rank'])
)
)
for a in data.findall('album')
]
@LastfmBase.topProperty("topAlbums")
def topAlbum(self):
"""top album of the artist"""
pass
@LastfmBase.cachedProperty
def topFans(self):
"""top fans of the artist"""
params = {'method': 'artist.gettopfans', 'artist': self.name}
data = self.__api._fetchData(params).find('topfans')
return [
User(
self.__api,
subject = self,
name = u.findtext('name'),
url = u.findtext('url'),
image = dict([(i.get('size'), i.text) for i in u.findall('image')]),
stats = Stats(
subject = u.findtext('name'),
weight = int(u.findtext('weight'))
)
)
for u in data.findall('user')
]
@LastfmBase.topProperty("topFans")
def topFan(self):
"""top fan of the artist"""
pass
@LastfmBase.cachedProperty
def topTracks(self):
"""top tracks of the artist"""
params = {'method': 'artist.gettoptracks', 'artist': self.name}
data = self.__api._fetchData(params).find('toptracks')
return [
Track(
self.__api,
subject = self,
name = t.findtext('name'),
artist = self,
mbid = t.findtext('mbid'),
stats = Stats(
subject = t.findtext('name'),
playcount = int(t.findtext('playcount')),
rank = int(t.attrib['rank'])
),
streamable = (t.findtext('streamable') == '1'),
fullTrack = (t.find('streamable').attrib['fulltrack'] == '1'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
)
for t in data.findall('track')
]
@LastfmBase.topProperty("topTracks")
def topTrack(self):
"""topmost fan of the artist"""
pass
@staticmethod
def search(api,
artist,
limit = None):
params = {'method': 'artist.search', 'artist': artist}
if limit:
params.update({'limit': limit})
@lazylist
def gen(lst):
data = api._fetchData(params).find('results')
totalPages = int(data.findtext("{%s}totalResults" % Api.SEARCH_XMLNS))/ \
int(data.findtext("{%s}itemsPerPage" % Api.SEARCH_XMLNS)) + 1
@lazylist
def gen2(lst, data):
for a in data.findall('artistmatches/artist'):
yield Artist(
api,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
url = a.findtext('url'),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
streamable = (a.findtext('streamable') == '1'),
)
for a in gen2(data):
yield a
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = api._fetchData(params).find('results')
for a in gen2(data):
yield a
return gen()
@staticmethod
def _fetchData(api,
artist = None,
mbid = None):
params = {'method': 'artist.getinfo'}
if not (artist or mbid):
raise LastfmInvalidParametersError("either artist or mbid has to be given as argument.")
if artist:
params.update({'artist': artist})
elif mbid:
params.update({'mbid': mbid})
return api._fetchData(params).find('artist')
def _fillInfo(self):
data = Artist._fetchData(self.__api, self.name)
self.__name = data.findtext('name')
self.__mbid = data.findtext('mbid')
self.__url = data.findtext('url')
self.__image = dict([(i.get('size'), i.text) for i in data.findall('image')])
self.__streamable = (data.findtext('streamable') == 1)
self.__stats = Stats(
subject = self,
listeners = int(data.findtext('stats/listeners')),
playcount = int(data.findtext('stats/playcount'))
)
self.__similar = [
Artist(
self.__api,
subject = self,
name = a.findtext('name'),
url = a.findtext('url'),
image = dict([(i.get('size'), i.text) for i in a.findall('image')])
)
for a in data.findall('similar/artist')
]
self.__topTags = [
Tag(
self.__api,
subject = self,
name = t.findtext('name'),
url = t.findtext('url')
)
for t in data.findall('tags/tag')
]
self.__bio = Artist.Bio(
self,
published = datetime(*(time.strptime(
data.findtext('bio/published').strip(),
'%a, %d %b %Y %H:%M:%S +0000'
)[0:6])),
summary = data.findtext('bio/summary'),
content = data.findtext('bio/content')
)
@staticmethod
def getInfo(api,
artist = None,
mbid = None):
data = Artist._fetchData(api, artist, mbid)
a = Artist(api, name = data.findtext('name'))
if a.bio is None:
a._fillInfo()
return a
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['name'].lower())
except KeyError:
try:
return hash(args[1].lower())
except IndexError:
raise LastfmInvalidParametersError("name has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(name = self.name)
def __eq__(self, other):
if self.mbid and other.mbid:
return self.mbid == other.mbid
if self.url and other.url:
return self.url == other.url
return self.name == other.name
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.Artist: %s>" % self.__name
class Bio(object):
"""A class representing the biography of an artist."""
def __init__(self,
artist,
published = None,
summary = None,
content = None):
self.__artist = artist
self.__published = published
self.__summary = summary
self.__content = content
@property
def artist(self):
"""artist for which the biography is"""
return self.__artist
@property
def published(self):
"""publication time of the biography"""
return self.__published
@property
def summary(self):
"""summary of the biography"""
return self.__summary
@property
def content(self):
"""content of the biography"""
return self.__content
def __repr__(self):
return "<lastfm.artist.Bio: for artist '%s'>" % self.__artist.name
from datetime import datetime
import time
from album import Album
from api import Api
from error import LastfmInvalidParametersError
from event import Event
from stats import Stats
from tag import Tag
from track import Track
from user import User

5
src/auth.py Normal file
View File

@ -0,0 +1,5 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"

93
src/base.py Normal file
View File

@ -0,0 +1,93 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
try:
from threading import Lock
except ImportError:
from dummy_threading import Lock
class LastfmBase(object):
"""Base class for all the classes in this package"""
registry = {}
_lock = Lock()
def __new__(cls, *args, **kwds):
subject = None
if 'subject' in kwds and not cls.__name__.startswith('Weekly'):
subject = kwds['subject']
del kwds['subject']
if 'bypassRegistry' in kwds:
del kwds['bypassRegistry']
inst = object.__new__(cls)
inst.init(*args, **kwds)
return inst
key = cls.hashFunc(*args, **kwds)
if subject is not None:
key = (hash(subject), key)
LastfmBase._lock.acquire()
try:
inst, alreadyRegistered = LastfmBase.register(object.__new__(cls), key)
if not alreadyRegistered:
inst.init(*args, **kwds)
finally:
LastfmBase._lock.release()
return inst
@staticmethod
def register(ob, key):
if not ob.__class__ in LastfmBase.registry:
LastfmBase.registry[ob.__class__] = {}
if key in LastfmBase.registry[ob.__class__]:
ob = LastfmBase.registry[ob.__class__][key]
#print "already registered: %s" % repr(ob)
return (ob, True)
else:
#print "not already registered: %s" % ob.__class__
LastfmBase.registry[ob.__class__][key] = ob
return (ob, False)
@staticmethod
def topProperty(listPropertyName):
def decorator(func):
def wrapper(ob):
topList = getattr(ob, listPropertyName)
return (len(topList) and topList[0] or None)
return property(fget = wrapper, doc = func.__doc__)
return decorator
@staticmethod
def cachedProperty(func):
frame = sys._getframe(1)
classname = frame.f_code.co_name
funcName = func.func_code.co_name
attributeName = "_%s__%s" % (classname, funcName)
def wrapper(ob):
cacheAttribute = getattr(ob, attributeName, None)
if cacheAttribute is None:
cacheAttribute = func(ob)
setattr(ob, attributeName, cacheAttribute)
return cacheAttribute
return property(fget = wrapper, doc = func.__doc__)
def __gt__(self, other):
return not (self.__lt__(other) or self.__eq(other))
def __ne__(self, other):
return not self.__eq__(other)
def __ge__(self, other):
return not self.__lt__(other)
def __le__(self, other):
return not self.__gt__(other)
import sys

80
src/error.py Normal file
View File

@ -0,0 +1,80 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
class LastfmError(Exception):
"""Base class for Lastfm errors"""
def __init__(self,
message = None,
code = None):
self.__code = code
self.__message = message
@property
def code(self):
return self.__code
@property
def message(self):
return self.__message
def __str__(self):
return "%s" % self.message
class LastfmInvalidServiceError(LastfmError):#2
pass
class LastfmInvalidMethodError(LastfmError):#3
pass
class LastfmAuthenticationFailedError(LastfmError):#4
pass
class LastfmInvalidFormatError(LastfmError):#5
pass
class LastfmInvalidParametersError(LastfmError):#6
pass
class LastfmInvalidResourceError(LastfmError):#7
pass
class LastfmOperationFailedError(LastfmError):#8
pass
class LastfmInvalidSessionKeyError(LastfmError):#9
pass
class LastfmInvalidApiKeyError(LastfmError):#10
pass
class LastfmServiceOfflineError(LastfmError):#11
pass
class LastfmSubscribersOnlyError(LastfmError):#12
pass
class LastfmTokenNotAuthorizedError(LastfmError):#14
pass
class LastfmTokenExpiredError(LastfmError):#15
pass
errorMap = {
1: LastfmError,
2: LastfmInvalidServiceError,
3: LastfmInvalidMethodError,
4: LastfmAuthenticationFailedError,
5: LastfmInvalidFormatError,
6: LastfmInvalidParametersError,
7: LastfmInvalidResourceError,
8: LastfmOperationFailedError,
9: LastfmInvalidSessionKeyError,
10: LastfmInvalidApiKeyError,
11: LastfmServiceOfflineError,
12: LastfmSubscribersOnlyError,
14: LastfmTokenNotAuthorizedError,
15: LastfmTokenExpiredError
}

204
src/event.py Normal file
View File

@ -0,0 +1,204 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
class Event(LastfmBase):
"""A class representing an event."""
def init(self,
api,
id = None,
title = None,
artists = None,
headliner = None,
venue = None,
startDate = None,
startTime = None,
description = None,
image = None,
url = None,
stats = None,
tag = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__id = id
self.__title = title
self.__artists = artists
self.__headliner = headliner
self.__venue = venue
self.__startDate = startDate
self.__description = description
self.__image = image
self.__url = url
self.__stats = stats and Stats(
subject = self,
attendance = stats.attendance,
reviews = stats.reviews
)
self.__tag = tag
@property
def id(self):
"""id of the event"""
return self.__id
@property
def title(self):
"""title of the event"""
return self.__title
@property
def artists(self):
"""artists performing in the event"""
return self.__artists
@property
def headliner(self):
"""headliner artist of the event"""
return self.__headliner
@property
def venue(self):
"""venue of the event"""
return self.__venue
@property
def startDate(self):
"""start date of the event"""
return self.__startDate
@property
def description(self):
"""description of the event"""
return self.__description
@property
def image(self):
"""poster of the event"""
return self.__image
@property
def url(self):
"""url of the event's page"""
return self.__url
@property
def stats(self):
"""stats of the event"""
return self.__stats
@property
def tag(self):
"""tags for the event"""
return self.__tag
@staticmethod
def getInfo(api, event):
params = {'method': 'event.getinfo', 'event': event}
data = api._fetchData(params).find('event')
return Event.createFromData(api, data)
@staticmethod
def createFromData(api, data):
startDate = None
if data.findtext('startTime') is not None:
startDate = datetime(*(
time.strptime(
"%s %s" % (
data.findtext('startDate').strip(),
data.findtext('startTime').strip()
),
'%a, %d %b %Y %H:%M'
)[0:6])
)
else:
try:
startDate = datetime(*(
time.strptime(
data.findtext('startDate').strip(),
'%a, %d %b %Y %H:%M:%S'
)[0:6])
)
except ValueError:
try:
startDate = datetime(*(
time.strptime(
data.findtext('startDate').strip(),
'%a, %d %b %Y'
)[0:6])
)
except ValueError:
pass
return Event(
api,
id = int(data.findtext('id')),
title = data.findtext('title'),
artists = [Artist(api, name = a.text) for a in data.findall('artists/artist')],
headliner = Artist(api, name = data.findtext('artists/headliner')),
venue = Venue(
name = data.findtext('venue/name'),
location = Location(
api,
city = data.findtext('venue/location/city'),
country = Country(
api,
name = data.findtext('venue/location/country')
),
street = data.findtext('venue/location/street'),
postalCode = data.findtext('venue/location/postalcode'),
latitude = float(data.findtext(
'venue/location/{%s}point/{%s}lat' % ((Location.xmlns,)*2)
)),
longitude = float(data.findtext(
'venue/location/{%s}point/{%s}long' % ((Location.xmlns,)*2)
)),
timezone = data.findtext('venue/location/timezone')
),
url = data.findtext('venue/url')
),
startDate = startDate,
description = data.findtext('description'),
image = dict([(i.get('size'), i.text) for i in data.findall('image')]),
url = data.findtext('url'),
stats = Stats(
subject = int(data.findtext('id')),
attendance = int(data.findtext('attendance')),
reviews = int(data.findtext('reviews')),
),
tag = data.findtext('tag')
)
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['id'])
except KeyError:
raise LastfmInvalidParametersError("id has to be provided for hashing")
def __hash__(self):
return Event.hashFunc(id = self.id)
def __eq__(self, other):
return self.id == other.id
def __lt__(self, other):
return self.startDate < other.startDate
def __repr__(self):
return "<lastfm.Event: %s at %s on %s>" % (self.title, self.venue.name, self.startDate.strftime("%x"))
from datetime import datetime
import time
from api import Api
from artist import Artist
from error import LastfmInvalidParametersError
from geo import Venue, Location, Country
from stats import Stats

92
src/filecache.py Normal file
View File

@ -0,0 +1,92 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
import md5
import os
import tempfile
class _FileCacheError(Exception):
'''Base exception class for FileCache related errors'''
class FileCache(object):
DEPTH = 3
def __init__(self,root_directory=None):
self._InitializeRootDirectory(root_directory)
def Get(self,key):
path = self._GetPath(key)
if os.path.exists(path):
return open(path).read()
else:
return None
def Set(self,key,data):
path = self._GetPath(key)
directory = os.path.dirname(path)
if not os.path.exists(directory):
os.makedirs(directory)
if not os.path.isdir(directory):
raise _FileCacheError('%s exists but is not a directory' % directory)
temp_fd, temp_path = tempfile.mkstemp()
temp_fp = os.fdopen(temp_fd, 'w')
temp_fp.write(data)
temp_fp.close()
if not path.startswith(self._root_directory):
raise _FileCacheError('%s does not appear to live under %s' %
(path, self._root_directory))
if os.path.exists(path):
os.remove(path)
os.rename(temp_path, path)
def Remove(self,key):
path = self._GetPath(key)
if not path.startswith(self._root_directory):
raise _FileCacheError('%s does not appear to live under %s' %
(path, self._root_directory ))
if os.path.exists(path):
os.remove(path)
def GetCachedTime(self,key):
path = self._GetPath(key)
if os.path.exists(path):
return os.path.getmtime(path)
else:
return None
def _GetUsername(self):
'''Attempt to find the username in a cross-platform fashion.'''
return os.getenv('USER') or \
os.getenv('LOGNAME') or \
os.getenv('USERNAME') or \
os.getlogin() or \
'nobody'
def _GetTmpCachePath(self):
username = self._GetUsername()
cache_directory = 'python.cache_' + username
return os.path.join(tempfile.gettempdir(), cache_directory)
def _InitializeRootDirectory(self, root_directory):
if not root_directory:
root_directory = self._GetTmpCachePath()
root_directory = os.path.abspath(root_directory)
if not os.path.exists(root_directory):
os.mkdir(root_directory)
if not os.path.isdir(root_directory):
raise _FileCacheError('%s exists but is not a directory' %
root_directory)
self._root_directory = root_directory
def _GetPath(self,key):
hashed_key = md5.new(key).hexdigest()
return os.path.join(self._root_directory,
self._GetPrefix(hashed_key),
hashed_key)
def _GetPrefix(self,hashed_key):
return os.path.sep.join(hashed_key[0:FileCache.DEPTH])

324
src/geo.py Normal file
View File

@ -0,0 +1,324 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
from lazylist import lazylist
class Geo(object):
"""A class representing an geographic location."""
@staticmethod
def getEvents(api,
location,
latitude = None,
longitude = None,
distance = None):
params = {'method': 'geo.getevents', 'location': location}
if distance is not None:
params.update({'distance': distance})
if latitude is not None and longitude is not None:
params.update({'latitude': latitude, 'longitude': longitude})
@lazylist
def gen(lst):
data = api._fetchData(params).find('events')
totalPages = int(data.attrib['totalpages'])
@lazylist
def gen2(lst, data):
for e in data.findall('event'):
yield Event.createFromData(api, e)
for e in gen2(data):
yield e
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = api._fetchData(params).find('events')
for e in gen2(data):
yield e
return gen()
@staticmethod
def getTopArtists(api, country):
params = {'method': 'geo.gettopartists', 'country': country}
data = api._fetchData(params).find('topartists')
return [
Artist(
api,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
stats = Stats(
subject = a.findtext('name'),
rank = int(a.attrib['rank']),
playcount = int(a.findtext('playcount'))
),
url = 'http://' + a.findtext('url'),
image = {'large': a.findtext('image')}
)
for a in data.findall('artist')
]
@staticmethod
def getTopTracks(api, country, location = None):
params = {'method': 'geo.gettoptracks', 'country': country}
if location is not None:
params.update({'location': location})
data = api._fetchData(params).find('toptracks')
return [
Track(
api,
name = t.findtext('name'),
mbid = t.findtext('mbid'),
artist = Artist(
api,
name = t.findtext('artist/name'),
mbid = t.findtext('artist/mbid'),
url = t.findtext('artist/url')
),
stats = Stats(
subject = t.findtext('name'),
rank = int(t.attrib['rank']),
playcount = int(t.findtext('playcount'))
),
streamable = (t.findtext('streamable') == '1'),
fullTrack = (t.find('streamable').attrib['fulltrack'] == '1'),
url = 'http://' + t.findtext('url'),
image = {'large': t.findtext('image')}
)
for t in data.findall('track')
]
class Venue(LastfmBase):
"""A class representing a venue of an event"""
def init(self,
name = None,
location = None,
url = None):
self.__name = name
self.__location = location
self.__url = url
@property
def name(self):
"""name of the venue"""
return self.__name
@property
def location(self):
"""location of the event"""
return self.__location
@property
def url(self):
"""url of the event's page"""
return self.__url
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['url'])
except KeyError:
raise LastfmInvalidParametersError("url has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(url = self.url)
def __eq__(self, other):
return self.url == other.url
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.geo.Venue: %s, %s>" % (self.name, self.location.city)
class Location(LastfmBase):
"""A class representing a location of an event"""
xmlns = "http://www.w3.org/2003/01/geo/wgs84_pos#"
def init(self,
api,
city = None,
country = None,
street = None,
postalCode = None,
latitude = None,
longitude = None,
timezone = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__city = city
self.__country = country
self.__street = street
self.__postalCode = postalCode
self.__latitude = latitude
self.__longitude = longitude
self.__timezone = timezone
@property
def city(self):
"""city in which the location is situated"""
return self.__city
@property
def country(self):
"""country in which the location is situated"""
return self.__country
@property
def street(self):
"""street in which the location is situated"""
return self.__street
@property
def postalCode(self):
"""postal code of the location"""
return self.__postalCode
@property
def latitude(self):
"""latitude of the location"""
return self.__latitude
@property
def longitude(self):
"""longitude of the location"""
return self.__longitude
@property
def timezone(self):
"""timezone in which the location is situated"""
return self.__timezone
@LastfmBase.cachedProperty
def topTracks(self):
"""top tracks of the location"""
if self.country is None or self.city is None:
raise LastfmInvalidParametersError("country and city of this location are required for calling this method")
return Geo.getTopTracks(self.__api, self.country.name, self.city)
@LastfmBase.topProperty("topTracks")
def topTrack(self):
"""top track of the location"""
pass
def getEvents(self,
distance = None):
return Geo.getEvents(self.__api,
self.city,
self.latitude,
self.longitude,
distance)
@LastfmBase.cachedProperty
def events(self):
"""events taking place at/around the location"""
return self.getEvents()
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash("latlong%s%s" % (kwds['latitude'], kwds['longitude']))
except KeyError:
try:
return hash("name%s" % kwds['city'])
except KeyError:
raise LastfmInvalidParametersError("either latitude and longitude or city has to be provided for hashing")
def __hash__(self):
if not self.name:
return self.__class__.hashFunc(
latitude = self.latitude,
longitude = self.longitude)
else:
return self.__class__.hashFunc(name = self.city)
def __eq__(self, other):
return self.latitude == other.latitude and self.longitude == other.longitude
def __lt__(self, other):
if self.country != other.country:
return self.country < other.country
else:
return self.city < other.city
def __repr__(self):
if self.city is None:
return "<lastfm.geo.Location: (%s, %s)>" % (self.latitude, self.longitude)
else:
return "<lastfm.geo.Location: %s>" % self.city
class Country(LastfmBase):
"""A class representing a country."""
def init(self,
api,
name = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__name = name
@property
def name(self):
"""name of the country"""
return self.__name
@LastfmBase.cachedProperty
def topArtists(self):
"""top artists of the country"""
return Geo.getTopArtists(self.__api, self.name)
@LastfmBase.topProperty("topArtists")
def topArtist(self):
"""top artist of the country"""
pass
def getTopTracks(self, location = None):
return Geo.getTopTracks(self.__api, self.name, location)
@LastfmBase.cachedProperty
def topTracks(self):
"""top tracks of the country"""
return self.getTopTracks()
@LastfmBase.topProperty("topTracks")
def topTrack(self):
"""top track of the country"""
pass
@LastfmBase.cachedProperty
def events(self):
"""events taking place at/around the location"""
return Geo.getEvents(self.__api, self.name)
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['name'])
except KeyError:
raise LastfmInvalidParametersError("name has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(name = self.name)
def __eq__(self, other):
return self.name == other.name
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.geo.Country: %s>" % self.name
from api import Api
from artist import Artist
from error import LastfmInvalidParametersError
from event import Event
from stats import Stats
from track import Track

120
src/group.py Normal file
View File

@ -0,0 +1,120 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
from lazylist import lazylist
class Group(LastfmBase):
"""A class representing a group on last.fm."""
def init(self,
api,
name = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__name = name
@property
def name(self):
return self.__name
@LastfmBase.cachedProperty
def weeklyChartList(self):
params = {'method': 'group.getweeklychartlist', 'group': self.name}
data = self.__api._fetchData(params).find('weeklychartlist')
return [
WeeklyChart.createFromData(self.__api, self, c)
for c in data.findall('chart')
]
def getWeeklyAlbumChart(self,
start = None,
end = None):
params = {'method': 'group.getweeklyalbumchart', 'group': self.name}
params = WeeklyChart._checkWeeklyChartParams(params, start, end)
data = self.__api._fetchData(params).find('weeklyalbumchart')
return WeeklyAlbumChart.createFromData(self.__api, self, data)
@LastfmBase.cachedProperty
def recentWeeklyAlbumChart(self):
return self.getWeeklyAlbumChart()
@LastfmBase.cachedProperty
def weeklyAlbumChartList(self):
wcl = list(self.weeklyChartList)
wcl.reverse()
@lazylist
def gen(lst):
for wc in wcl:
yield self.getWeeklyAlbumChart(wc.start, wc.end)
return gen()
def getWeeklyArtistChart(self,
start = None,
end = None):
params = {'method': 'group.getweeklyartistchart', 'group': self.name}
params = WeeklyChart._checkWeeklyChartParams(params, start, end)
data = self.__api._fetchData(params).find('weeklyartistchart')
return WeeklyArtistChart.createFromData(self.__api, self, data)
@LastfmBase.cachedProperty
def recentWeeklyArtistChart(self):
return self.getWeeklyArtistChart()
@LastfmBase.cachedProperty
def weeklyArtistChartList(self):
wcl = list(self.weeklyChartList)
wcl.reverse()
@lazylist
def gen(lst):
for wc in wcl:
yield self.getWeeklyArtistChart(wc.start, wc.end)
return gen()
def getWeeklyTrackChart(self,
start = None,
end = None):
params = {'method': 'group.getweeklytrackchart', 'group': self.name}
params = WeeklyChart._checkWeeklyChartParams(params, start, end)
data = self.__api._fetchData(params).find('weeklytrackchart')
return WeeklyTrackChart.createFromData(self.__api, self, data)
@LastfmBase.cachedProperty
def recentWeeklyTrackChart(self):
return self.getWeeklyTrackChart()
@LastfmBase.cachedProperty
def weeklyTrackChartList(self):
wcl = list(self.weeklyChartList)
wcl.reverse()
@lazylist
def gen(lst):
for wc in wcl:
yield self.getWeeklyTrackChart(wc.start, wc.end)
return gen()
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['name'])
except KeyError:
raise LastfmInvalidParametersError("name has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(name = self.name)
def __eq__(self, other):
return self.name == other.name
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.Group: %s>" % self.name
from api import Api
from error import LastfmInvalidParametersError
from weeklychart import WeeklyChart, WeeklyAlbumChart, WeeklyArtistChart, WeeklyTrackChart

141
src/lazylist.py Normal file
View File

@ -0,0 +1,141 @@
"""Module for the creation and use of iterator-based lazy lists.
this module defines a class LazyList which can be used to represent sequences
of values generated lazily. One can also create recursively defined lazy lists
that generate their values based on ones previously generated.
Backport to python 2.5 by Michael Pust
"""
__author__ = 'Dan Spitz'
__all__ = ('LazyList', 'RecursiveLazyList', 'lazylist')
import itertools
class LazyList(object):
"""A Sequence whose values are computed lazily by an iterator.
"""
def __init__(self, iterable):
self._exhausted = False
self._iterator = iter(iterable)
self._data = []
def __len__(self):
"""Get the length of a LazyList's computed data."""
return len(self._data)
def __getitem__(self, i):
"""Get an item from a LazyList.
i should be a positive integer or a slice object."""
if isinstance(i, int):
#index has not yet been yielded by iterator (or iterator exhausted
#before reaching that index)
if i >= len(self):
self.exhaust(i)
elif i < 0:
raise ValueError('cannot index LazyList with negative number')
return self._data[i]
#LazyList slices are iterators over a portion of the list.
elif isinstance(i, slice):
start, stop, step = i.start, i.stop, i.step
if any(x is not None and x < 0 for x in (start, stop, step)):
raise ValueError('cannot index or step through a LazyList with'
'a negative number')
#set start and step to their integer defaults if they are None.
if start is None:
start = 0
if step is None:
step = 1
def LazyListIterator():
count = start
predicate = (stop is None) and (lambda: True) or (lambda: count < stop)
while predicate():
try:
yield self[count]
#slices can go out of actual index range without raising an
#error
except IndexError:
break
count += step
return LazyListIterator()
raise TypeError('i must be an integer or slice')
def __iter__(self):
"""return an iterator over each value in the sequence,
whether it has been computed yet or not."""
return self[:]
def computed(self):
"""Return an iterator over the values in a LazyList that have
already been computed."""
return self[:len(self)]
def exhaust(self, index = None):
"""Exhaust the iterator generating this LazyList's values.
if index is None, this will exhaust the iterator completely.
Otherwise, it will iterate over the iterator until either the list
has a value for index or the iterator is exhausted.
"""
if self._exhausted:
return
if index is None:
ind_range = itertools.count(len(self))
else:
ind_range = range(len(self), index + 1)
for ind in ind_range:
try:
self._data.append(self._iterator.next())
except StopIteration: #iterator is fully exhausted
self._exhausted = True
break
class RecursiveLazyList(LazyList):
def __init__(self, prod, *args, **kwds):
super(RecursiveLazyList,self).__init__(prod(self,*args, **kwds))
def __repr__(self):
return "<lastfm.lazylist>"
class RecursiveLazyListFactory:
def __init__(self, producer):
self._gen = producer
def __call__(self,*a,**kw):
return RecursiveLazyList(self._gen,*a,**kw)
def lazylist(gen):
"""Decorator for creating a RecursiveLazyList subclass.
This should decorate a generator function taking the LazyList object as its
first argument which yields the contents of the list in order.
"""
return RecursiveLazyListFactory(gen)
#two examples
if __name__ == '__main__':
#fibonnacci sequence in a lazy list.
@lazylist
def fibgen(lst):
yield 0
yield 1
for a, b in itertools.izip(lst, lst[1:]):
yield a + b
fibs = fibgen() #now fibs can be indexed or iterated over as if it were
#an infinitely long list containing the fibonnaci sequence
#prime numbers in a lazy list.
@lazylist
def primegen(lst):
yield 2
for candidate in itertools.count(3): #start at next number after 2
#if candidate is not divisible by any smaller prime numbers,
#it is a prime.
if all(candidate % p for p in lst.computed()):
yield candidate
primes = primegen() #same for primes- treat it like an infinitely long list
#containing all prime numbers.
print fibs[0], fibs[1], fibs[2], primes[0], primes[1], primes[2]
print list(fibs[:10]), list(primes[:10])

65
src/playlist.py Normal file
View File

@ -0,0 +1,65 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
class Playlist(LastfmBase):
"""A class representing an XPSF playlist."""
def init(self, api, url):
self.__api = api
self.__data = None
self.__url = url
@LastfmBase.cachedProperty
def data(self):
"""playlist's data"""
params = {'method': 'playlist.fetch', 'playlistURL': self.__url}
tmp = StringIO.StringIO()
ElementTree.ElementTree(self.__api._fetchData(params)[0]).write(tmp)
return tmp.getvalue()
@property
def url(self):
"""url of the playlist"""
return self.__url
@staticmethod
def fetch(api, url):
return Playlist(api, url = url)
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['url'])
except KeyError:
raise LastfmInvalidParametersError("url has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(url = self.url)
def __eq__(self, other):
return self.url == other.url
def __lt__(self, other):
return self.url < other.url
def __repr__(self):
return "<lastfm.Playlist: %s>" % self.url
import StringIO
import sys
from error import LastfmInvalidParametersError
if sys.version.startswith('2.5'):
import xml.etree.cElementTree as ElementTree
else:
try:
import cElementTree as ElementTree
except ImportError:
try:
import ElementTree
except ImportError:
raise LastfmError("Install ElementTree package for using python-lastfm")

37
src/registry.py Normal file
View File

@ -0,0 +1,37 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from album import Album
from artist import Artist
from base import LastfmBase
from error import LastfmInvalidParametersError
from event import Event
from geo import Location, Country
from group import Group
from playlist import Playlist
from tag import Tag
from track import Track
from user import User
from weeklychart import WeeklyAlbumChart, WeeklyArtistChart, WeeklyTrackChart
class Registry(object):
"""The registry to contain all the entities"""
keys = [c.__name__ for c in [Album, Artist, Event, Location, Country, Group,
Playlist, Tag, Track, User, WeeklyAlbumChart, WeeklyArtistChart, WeeklyTrackChart]]
def __getitem__(self, name):
if name not in Registry.keys:
raise LastfmInvalidParametersError("Key does not correspond to a valid class")
else:
try:
vals = LastfmBase.registry[eval(name)].values()
vals.sort()
return vals
except KeyError:
return []
def __repr__(self):
return "<lastfm.Registry>"

83
src/stats.py Normal file
View File

@ -0,0 +1,83 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
class Stats(object):
"""A class representing the stats of an artist."""
def __init__(self,
subject,
listeners = None,
playcount = None,
tagcount = None,
count = None,
match = None,
rank = None,
weight = None,
attendance = None,
reviews = None,):
self.__subject = subject
self.__listeners = listeners
self.__playcount = playcount
self.__tagcount = tagcount
self.__count = count
self.__match = match
self.__rank = rank
self.__weight = weight
self.__attendance = attendance
self.__reviews = reviews
@property
def subject(self):
"""subject of the stats"""
return self.__subject
@property
def rank(self):
"""rank of the subject"""
return self.__rank
@property
def listeners(self):
"""number of listeners of the subject"""
return self.__listeners
@property
def playcount(self):
"""playcount of the subject"""
return self.__playcount
@property
def tagcount(self):
"""tagcount of the subject"""
return self.__tagcount
@property
def count(self):
"""count of the subject"""
return self.__count
@property
def match(self):
"""match of the subject"""
return self.__match
@property
def weight(self):
"""weight of the subject"""
return self.__weight
@property
def attendance(self):
"""attendance of the subject"""
return self.__attendance
@property
def reviews(self):
"""reviews of the subject"""
return self.__reviews
def __repr__(self):
return "<lastfm.Stats: for '%s'>" % self.__subject.name

250
src/tag.py Normal file
View File

@ -0,0 +1,250 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
from lazylist import lazylist
class Tag(LastfmBase):
""""A class representing a tag."""
def init(self,
api,
name = None,
url = None,
streamable = None,
stats = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__name = name
self.__url = url
self.__streamable = streamable
self.__stats = stats and Stats(
subject = self,
count = stats.count
)
@property
def name(self):
"""name of the tag"""
return self.__name
@property
def url(self):
"""url of the tag's page"""
return self.__url
@property
def streamable(self):
"""is the tag streamable"""
return self.__streamable
@property
def stats(self):
return self.__stats
@LastfmBase.cachedProperty
def similar(self):
"""tags similar to this tag"""
params = {'method': 'tag.getsimilar', 'tag': self.name}
data = self.__api._fetchData(params).find('similartags')
return [
Tag(
self.__api,
subject = self,
name = t.findtext('name'),
url = t.findtext('url'),
streamable = (t.findtext('streamable') == "1"),
)
for t in data.findall('tag')
]
@LastfmBase.topProperty("similar")
def mostSimilar(self):
"""most similar tag to this tag"""
pass
@LastfmBase.cachedProperty
def topAlbums(self):
"""top albums for the tag"""
params = {'method': 'tag.gettopalbums', 'tag': self.name}
data = self.__api._fetchData(params).find('topalbums')
return [
Album(
self.__api,
subject = self,
name = a.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = a.findtext('artist/name'),
mbid = a.findtext('artist/mbid'),
url = a.findtext('artist/url'),
),
mbid = a.findtext('mbid'),
url = a.findtext('url'),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
stats = Stats(
subject = a.findtext('name'),
tagcount = a.findtext('tagcount') and int(a.findtext('tagcount')) or None,
rank = a.attrib['rank'].strip() and int(a.attrib['rank']) or None
)
)
for a in data.findall('album')
]
@LastfmBase.topProperty("topAlbums")
def topAlbum(self):
"""top album for the tag"""
pass
@LastfmBase.cachedProperty
def topArtists(self):
"""top artists for the tag"""
params = {'method': 'tag.gettopartists', 'tag': self.name}
data = self.__api._fetchData(params).find('topartists')
return [
Artist(
self.__api,
subject = self,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
stats = Stats(
subject = a.findtext('name'),
rank = a.attrib['rank'].strip() and int(a.attrib['rank']) or None,
tagcount = a.findtext('tagcount') and int(a.findtext('tagcount')) or None
),
url = a.findtext('url'),
streamable = (a.findtext('streamable') == "1"),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
)
for a in data.findall('artist')
]
@LastfmBase.topProperty("topArtists")
def topArtist(self):
"""top artist for the tag"""
pass
@LastfmBase.cachedProperty
def topTracks(self):
"""top tracks for the tag"""
params = {'method': 'tag.gettoptracks', 'tag': self.name}
data = self.__api._fetchData(params).find('toptracks')
return [
Track(
self.__api,
subject = self,
name = t.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = t.findtext('artist/name'),
mbid = t.findtext('artist/mbid'),
url = t.findtext('artist/url'),
),
mbid = t.findtext('mbid'),
stats = Stats(
subject = t.findtext('name'),
rank = t.attrib['rank'].strip() and int(t.attrib['rank']) or None,
tagcount = t.findtext('tagcount') and int(t.findtext('tagcount')) or None
),
streamable = (t.findtext('streamable') == '1'),
fullTrack = (t.find('streamable').attrib['fulltrack'] == '1'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
)
for t in data.findall('track')
]
@LastfmBase.topProperty("topTracks")
def topTrack(self):
"""top track for the tag"""
pass
@LastfmBase.cachedProperty
def playlist(self):
return Playlist.fetch(self.__api,
"lastfm://playlist/tag/%s/freetracks" % self.name)
@staticmethod
def getTopTags(api):
params = {'method': 'tag.getTopTags'}
data = api._fetchData(params).find('toptags')
return [
Tag(
api,
name = t.findtext('name'),
url = t.findtext('url'),
stats = Stats(
subject = t.findtext('name'),
count = int(t.findtext('count')),
)
)
for t in data.findall('tag')
]
@staticmethod
def search(api,
tag,
limit = None):
params = {'method': 'tag.search', 'tag': tag}
if limit:
params.update({'limit': limit})
@lazylist
def gen(lst):
data = api._fetchData(params).find('results')
totalPages = int(data.findtext("{%s}totalResults" % Api.SEARCH_XMLNS))/ \
int(data.findtext("{%s}itemsPerPage" % Api.SEARCH_XMLNS)) + 1
@lazylist
def gen2(lst, data):
for t in data.findall('tagmatches/tag'):
yield Tag(
api,
name = t.findtext('name'),
url = t.findtext('url'),
stats = Stats(
subject = t.findtext('name'),
count = int(t.findtext('count')),
)
)
for t in gen2(data):
yield t
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = api._fetchData(params).find('results')
for t in gen2(data):
yield t
return gen()
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['name'])
except KeyError:
raise LastfmInvalidParametersError("name has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(name = self.name)
def __eq__(self, other):
return self.name == other.name
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.Tag: %s>" % self.name
from album import Album
from api import Api
from artist import Artist
from error import LastfmInvalidParametersError
from playlist import Playlist
from stats import Stats
from track import Track

66
src/tasteometer.py Normal file
View File

@ -0,0 +1,66 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
class Tasteometer(object):
"""A class representing a tasteometer."""
def __init__(self,
score = None,
matches = None,
artists = None):
self.__score = score
self.__matches = matches
self.__artists = artists
@property
def score(self):
"""score of the comparison"""
return self.__score
@property
def matches(self):
"""matches for the comparison"""
return self.__matches
@property
def artists(self):
"""artists for the comparison"""
return self.__artists
@staticmethod
def compare(api,
type1, type2,
value1, value2,
limit = None):
params = {
'method': 'tasteometer.compare',
'type1': type1,
'type2': type2,
'value1': value1,
'value2': value2
}
if limit is not None:
params.update({'limit': limit})
data = api._fetchData(params).find('comparison/result')
return Tasteometer(
score = float(data.findtext('score')),
matches = int(data.find('artists').attrib['matches']),
artists = [
Artist(
api,
name = a.findtext('name'),
url = a.findtext('url'),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
)
for a in data.findall('artists/artist')
]
)
def __repr__(self):
return "<lastfm.Tasteometer: %s%% match>" % (self.score*100)
from artist import Artist

290
src/track.py Normal file
View File

@ -0,0 +1,290 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
from lazylist import lazylist
class Track(LastfmBase):
"""A class representing a track."""
def init(self,
api,
name = None,
mbid = None,
url = None,
streamable = None,
artist = None,
album = None,
image = None,
stats = None,
fullTrack = None,
playedOn = None,
lovedOn = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__name = name
self.__mbid = mbid
self.__url = url
self.__streamable = streamable
self.__artist = artist
self.__album = album
self.__image = image
self.__stats = stats and Stats(
subject = self,
match = stats.match,
playcount = stats.playcount,
rank = stats.rank,
listeners = stats.listeners,
)
self.__fullTrack = fullTrack
self.__playedOn = playedOn
self.__lovedOn = lovedOn
@property
def name(self):
"""name of the track"""
return self.__name
@property
def mbid(self):
"""mbid of the track"""
return self.__mbid
@property
def url(self):
"""url of the tracks's page"""
return self.__url
@property
def streamable(self):
"""is the track streamable"""
return self.__streamable
@property
def artist(self):
"""artist of the track"""
return self.__artist
@property
def album(self):
"""artist of the track"""
return self.__album
@property
def image(self):
"""image of the track's album cover"""
return self.__image
@property
def stats(self):
"""stats of the track"""
return self.__stats
@property
def fullTrack(self):
"""is the full track streamable"""
return self.__fullTrack
@property
def playedOn(self):
"""datetime the track was last played"""
return self.__playedOn
@property
def lovedOn(self):
"""datetime the track was marked 'loved'"""
return self.__lovedOn
def __checkParams(self,
params,
artist = None,
track = None,
mbid = None):
if not ((artist and track) or mbid):
raise LastfmInvalidParametersError("either (artist and track) or mbid has to be given as argument.")
if artist and track:
params.update({'artist': artist, 'track': track})
elif mbid:
params.update({'mbid': mbid})
return params
@LastfmBase.cachedProperty
def similar(self):
"""tracks similar to this track"""
params = self.__checkParams(
{'method': 'track.getsimilar'},
self.artist.name,
self.name,
self.mbid
)
data = self.__api._fetchData(params).find('similartracks')
return [
Track(
self.__api,
subject = self,
name = t.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = t.findtext('artist/name'),
mbid = t.findtext('artist/mbid'),
url = t.findtext('artist/url')
),
mbid = t.findtext('mbid'),
stats = Stats(
subject = t.findtext('name'),
match = float(t.findtext('match'))
),
streamable = (t.findtext('streamable') == '1'),
fullTrack = (t.find('streamable').attrib['fulltrack'] == '1'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
)
for t in data.findall('track')
]
@LastfmBase.topProperty("similar")
def mostSimilar(self):
"""track most similar to this track"""
pass
@LastfmBase.cachedProperty
def topFans(self):
"""top fans of the track"""
params = self.__checkParams(
{'method': 'track.gettopfans'},
self.artist.name,
self.name,
self.mbid
)
data = self.__api._fetchData(params).find('topfans')
return [
User(
self.__api,
subject = self,
name = u.findtext('name'),
url = u.findtext('url'),
image = dict([(i.get('size'), i.text) for i in u.findall('image')]),
stats = Stats(
subject = u.findtext('name'),
weight = int(u.findtext('weight'))
)
)
for u in data.findall('user')
]
@LastfmBase.topProperty("topFans")
def topFan(self):
"""topmost fan of the track"""
pass
@LastfmBase.cachedProperty
def topTags(self):
"""top tags for the track"""
params = self.__checkParams(
{'method': 'track.gettoptags'},
self.artist.name,
self.name,
self.mbid
)
data = self.__api._fetchData(params).find('toptags')
return [
Tag(
self.__api,
subject = self,
name = t.findtext('name'),
url = t.findtext('url'),
stats = Stats(
subject = t.findtext('name'),
count = int(t.findtext('count')),
)
)
for t in data.findall('tag')
]
@LastfmBase.topProperty("topTags")
def topTag(self):
"""topmost tag for the track"""
pass
@staticmethod
def search(api,
track,
artist = None,
limit = None):
params = {'method': 'track.search', 'track': track}
if artist is not None:
params.update({'artist': artist})
if limit is not None:
params.update({'limit': limit})
@lazylist
def gen(lst):
data = api._fetchData(params).find('results')
totalPages = int(data.findtext("{%s}totalResults" % Api.SEARCH_XMLNS))/ \
int(data.findtext("{%s}itemsPerPage" % Api.SEARCH_XMLNS)) + 1
@lazylist
def gen2(lst, data):
for t in data.findall('trackmatches/track'):
yield Track(
api,
name = t.findtext('name'),
artist = Artist(
api,
name = t.findtext('artist')
),
url = t.findtext('url'),
stats = Stats(
subject = t.findtext('name'),
listeners = int(t.findtext('listeners'))
),
streamable = (t.findtext('streamable') == '1'),
fullTrack = (t.find('streamable').attrib['fulltrack'] == '1'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
)
for t in gen2(data):
yield t
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = api._fetchData(params).find('results')
for t in gen2(data):
yield t
return gen()
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash("%s%s" % (kwds['name'], hash(kwds['artist'])))
except KeyError:
raise LastfmInvalidParametersError("name and artist have to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(name = self.name, artist = self.artist)
def __eq__(self, other):
if self.mbid and other.mbid:
return self.mbid == other.mbid
if self.url and other.url:
return self.url == other.url
if (self.name and self.artist) and (other.name and other.artist):
return (self.name == other.name) and (self.artist == other.artist)
return super(Track, self).__eq__(other)
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.Track: '%s' by %s>" % (self.name, self.artist.name)
from api import Api
from artist import Artist
from error import LastfmInvalidParametersError
from stats import Stats
from tag import Tag
from user import User

720
src/user.py Normal file
View File

@ -0,0 +1,720 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
from lazylist import lazylist
import playlist
class User(LastfmBase):
"""A class representing an user."""
def init(self,
api,
name = None,
url = None,
image = None,
stats = None):
if not isinstance(api, Api):
raise LastfmInvalidParametersError("api reference must be supplied as an argument")
self.__api = api
self.__name = name
self.__url = url
self.__image = image
self.__stats = stats and Stats(
subject = self,
match = stats.match,
weight = stats.weight
)
self.__lirary = User.Library(api, self)
@property
def name(self):
"""name of the user"""
return self.__name
@property
def url(self):
"""url of the user's page"""
return self.__url
@property
def image(self):
"""image of the user"""
return self.__image
@property
def stats(self):
"""stats for the user"""
return self.__stats
@LastfmBase.cachedProperty
def events(self):
params = {'method': 'user.getevents', 'user': self.name}
data = self.__api._fetchData(params).find('events')
return [
Event.createFromData(self.__api, e)
for e in data.findall('event')
]
def getPastEvents(self,
limit = None):
params = {'method': 'user.getpastevents', 'user': self.name}
if limit is not None:
params.update({'limit': limit})
@lazylist
def gen(lst):
data = self.__api._fetchData(params).find('events')
totalPages = int(data.attrib['totalPages'])
@lazylist
def gen2(lst, data):
for e in data.findall('event'):
yield Event.createFromData(self.__api, e)
for e in gen2(data):
yield e
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = self.__api._fetchData(params).find('events')
for e in gen2(data):
yield e
return gen()
@LastfmBase.cachedProperty
def pastEvents(self):
return self.getPastEvents()
def getFriends(self,
limit = None):
params = {'method': 'user.getfriends', 'user': self.name}
if limit is not None:
params.update({'limit': limit})
data = self.__api._fetchData(params).find('friends')
return [
User(
self.__api,
subject = self,
name = u.findtext('name'),
image = dict([(i.get('size'), i.text) for i in u.findall('image')]),
url = u.findtext('url'),
)
for u in data.findall('user')
]
@LastfmBase.cachedProperty
def friends(self):
"""friends of the user"""
return self.getFriends()
def getNeighbours(self, limit = None):
params = {'method': 'user.getneighbours', 'user': self.name}
if limit is not None:
params.update({'limit': limit})
data = self.__api._fetchData(params).find('neighbours')
return [
User(
self.__api,
subject = self,
name = u.findtext('name'),
image = {'medium': u.findtext('image')},
url = u.findtext('url'),
stats = Stats(
subject = u.findtext('name'),
match = float(u.findtext('match')),
),
)
for u in data.findall('user')
]
@LastfmBase.cachedProperty
def neighbours(self):
"""neighbours of the user"""
return self.getNeighbours()
@LastfmBase.topProperty("neighbours")
def nearestNeighbour(self):
"""nearest neightbour of the user"""
pass
@LastfmBase.cachedProperty
def playlists(self):
"""playlists of the user"""
params = {'method': 'user.getPlaylists', 'user': self.name}
data = self.__api._fetchData(params).find('playlists')
return [
User.Playlist(
self.__api,
id = int(p.findtext('id')),
title = p.findtext('title'),
date = datetime(*(
time.strptime(
p.findtext('date').strip(),
'%Y-%m-%dT%H:%M:%S'
)[0:6])
),
size = int(p.findtext('size')),
creator = self
)
for p in data.findall('playlist')
]
@LastfmBase.cachedProperty
def lovedTracks(self):
params = {'method': 'user.getlovedtracks', 'user': self.name}
data = self.__api._fetchData(params).find('lovedtracks')
return [
Track(
self.__api,
subject = self,
name = t.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = t.findtext('artist/name'),
mbid = t.findtext('artist/mbid'),
url = t.findtext('artist/url'),
),
mbid = t.findtext('mbid'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
lovedOn = datetime(*(
time.strptime(
t.findtext('date').strip(),
'%d %b %Y, %H:%M'
)[0:6])
)
)
for t in data.findall('track')
]
def getRecentTracks(self, limit = None):
params = {'method': 'user.getrecenttracks', 'user': self.name}
data = self.__api._fetchData(params, no_cache = True).find('recenttracks')
return [
Track(
self.__api,
subject = self,
name = t.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = t.findtext('artist'),
mbid = t.find('artist').attrib['mbid'],
),
album = Album(
self.__api,
subject = self,
name = t.findtext('album'),
artist = Artist(
self.__api,
subject = self,
name = t.findtext('artist'),
mbid = t.find('artist').attrib['mbid'],
),
mbid = t.find('album').attrib['mbid'],
),
mbid = t.findtext('mbid'),
streamable = (t.findtext('streamable') == '1'),
url = t.findtext('url'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
playedOn = datetime(*(
time.strptime(
t.findtext('date').strip(),
'%d %b %Y, %H:%M'
)[0:6])
)
)
for t in data.findall('track')
]
@property
def recentTracks(self):
"""recent tracks played by the user"""
return self.getRecentTracks()
@LastfmBase.topProperty("recentTracks")
def mostRecentTrack(self):
"""most recent track played by the user"""
pass
def getTopAlbums(self, period = None):
params = {'method': 'user.gettopalbums', 'user': self.name}
if period is not None:
params.update({'period': period})
data = self.__api._fetchData(params).find('topalbums')
return [
Album(
self.__api,
subject = self,
name = a.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = a.findtext('artist/name'),
mbid = a.findtext('artist/mbid'),
url = a.findtext('artist/url'),
),
mbid = a.findtext('mbid'),
url = a.findtext('url'),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
stats = Stats(
subject = a.findtext('name'),
playcount = int(a.findtext('playcount')),
rank = int(a.attrib['rank'])
)
)
for a in data.findall('album')
]
@LastfmBase.cachedProperty
def topAlbums(self):
"""overall top albums of the user"""
return self.getTopAlbums()
@LastfmBase.topProperty("topAlbums")
def topAlbum(self):
"""overall top most album of the user"""
pass
def getTopArtists(self, period = None):
params = {'method': 'user.gettopartists', 'user': self.name}
if period is not None:
params.update({'period': period})
data = self.__api._fetchData(params).find('topartists')
return [
Artist(
self.__api,
subject = self,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
stats = Stats(
subject = a.findtext('name'),
rank = a.attrib['rank'].strip() and int(a.attrib['rank']) or None,
playcount = a.findtext('playcount') and int(a.findtext('playcount')) or None
),
url = a.findtext('url'),
streamable = (a.findtext('streamable') == "1"),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
)
for a in data.findall('artist')
]
@LastfmBase.cachedProperty
def topArtists(self):
"""top artists of the user"""
return self.getTopArtists()
@LastfmBase.topProperty("topArtists")
def topArtist(self):
"""top artist of the user"""
pass
def getTopTracks(self, period = None):
params = {'method': 'user.gettoptracks', 'user': self.name}
if period is not None:
params.update({'period': period})
data = self.__api._fetchData(params).find('toptracks')
return [
Track(
self.__api,
subject = self,
name = t.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = t.findtext('artist/name'),
mbid = t.findtext('artist/mbid'),
url = t.findtext('artist/url'),
),
mbid = t.findtext('mbid'),
stats = Stats(
subject = t.findtext('name'),
rank = t.attrib['rank'].strip() and int(t.attrib['rank']) or None,
playcount = t.findtext('playcount') and int(t.findtext('playcount')) or None
),
streamable = (t.findtext('streamable') == '1'),
fullTrack = (t.find('streamable').attrib['fulltrack'] == '1'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
)
for t in data.findall('track')
]
@LastfmBase.cachedProperty
def topTracks(self):
"""top tracks of the user"""
return self.getTopTracks()
@LastfmBase.topProperty("topTracks")
def topTrack(self):
"""top track of the user"""
return (len(self.topTracks) and self.topTracks[0] or None)
def getTopTags(self, limit = None):
params = {'method': 'user.gettoptags', 'user': self.name}
if limit is not None:
params.update({'limit': limit})
data = self.__api._fetchData(params).find('toptags')
return [
Tag(
self.__api,
subject = self,
name = t.findtext('name'),
url = t.findtext('url'),
stats = Stats(
subject = t.findtext('name'),
count = int(t.findtext('count'))
)
)
for t in data.findall('tag')
]
@LastfmBase.cachedProperty
def topTags(self):
"""top tags of the user"""
return self.getTopTags()
@LastfmBase.topProperty("topTags")
def topTag(self):
"""top tag of the user"""
pass
@LastfmBase.cachedProperty
def weeklyChartList(self):
params = {'method': 'user.getweeklychartlist', 'user': self.name}
data = self.__api._fetchData(params).find('weeklychartlist')
return [
WeeklyChart.createFromData(self.__api, self, c)
for c in data.findall('chart')
]
def getWeeklyAlbumChart(self,
start = None,
end = None):
params = {'method': 'user.getweeklyalbumchart', 'user': self.name}
params = WeeklyChart._checkWeeklyChartParams(params, start, end)
data = self.__api._fetchData(params).find('weeklyalbumchart')
return WeeklyAlbumChart.createFromData(self.__api, self, data)
@LastfmBase.cachedProperty
def recentWeeklyAlbumChart(self):
return self.getWeeklyAlbumChart()
@LastfmBase.cachedProperty
def weeklyAlbumChartList(self):
wcl = list(self.weeklyChartList)
wcl.reverse()
@lazylist
def gen(lst):
for wc in wcl:
try:
yield self.getWeeklyAlbumChart(wc.start, wc.end)
except LastfmError:
pass
return gen()
def getWeeklyArtistChart(self,
start = None,
end = None):
params = {'method': 'user.getweeklyartistchart', 'user': self.name}
params = WeeklyChart._checkWeeklyChartParams(params, start, end)
data = self.__api._fetchData(params).find('weeklyartistchart')
return WeeklyArtistChart.createFromData(self.__api, self, data)
@LastfmBase.cachedProperty
def recentWeeklyArtistChart(self):
return self.getWeeklyArtistChart()
@LastfmBase.cachedProperty
def weeklyArtistChartList(self):
wcl = list(self.weeklyChartList)
wcl.reverse()
@lazylist
def gen(lst):
for wc in wcl:
try:
yield self.getWeeklyArtistChart(wc.start, wc.end)
except LastfmError:
pass
return gen()
def getWeeklyTrackChart(self,
start = None,
end = None):
params = {'method': 'user.getweeklytrackchart', 'user': self.name}
params = WeeklyChart._checkWeeklyChartParams(params, start, end)
data = self.__api._fetchData(params).find('weeklytrackchart')
return WeeklyTrackChart.createFromData(self.__api, self, data)
@LastfmBase.cachedProperty
def recentWeeklyTrackChart(self):
return self.getWeeklyTrackChart()
@LastfmBase.cachedProperty
def weeklyTrackChartList(self):
wcl = list(self.weeklyChartList)
wcl.reverse()
@lazylist
def gen(lst):
for wc in wcl:
try:
yield self.getWeeklyTrackChart(wc.start, wc.end)
except LastfmError:
pass
return gen()
def compare(self, other, limit = None):
return Tasteometer.compare(self.__api,
'user', 'user',
self.name, other.name,
limit)
@property
def library(self):
return self.__lirary
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['name'])
except KeyError:
raise LastfmInvalidParametersError("name has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(name = self.name)
def __eq__(self, other):
return self.name == other.name
def __lt__(self, other):
return self.name < other.name
def __repr__(self):
return "<lastfm.User: %s>" % self.name
class Playlist(playlist.Playlist):
"""A class representing a playlist belonging to the user."""
def init(self, api, id, title, date, size, creator):
super(User.Playlist, self).init(api, "lastfm://playlist/%s" % id)
self.__id = id
self.__title = title
self.__date = date
self.__size = size
self.__creator = creator
@property
def id(self):
return self.__id
@property
def title(self):
return self.__title
@property
def date(self):
return self.__date
@property
def size(self):
return self.__size
@property
def creator(self):
return self.__creator
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['id'])
except KeyError:
raise LastfmInvalidParametersError("id has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(id = self.id)
def __repr__(self):
return "<lastfm.User.Playlist: %s>" % self.title
class Library(object):
"""A class representing the music library of the user."""
def __init__(self, api, user):
self.__api = api
self.__user = user
@property
def user(self):
return self.__user
def getAlbums(self,
limit = None):
params = {'method': 'library.getalbums', 'user': self.user.name}
if limit is not None:
params.update({'limit': limit})
@lazylist
def gen(lst):
data = self.__api._fetchData(params).find('albums')
totalPages = int(data.attrib['totalPages'])
@lazylist
def gen2(lst, data):
for a in data.findall('album'):
yield Album(
self.__api,
subject = self,
name = a.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = a.findtext('artist/name'),
mbid = a.findtext('artist/mbid'),
url = a.findtext('artist/url'),
),
mbid = a.findtext('mbid'),
url = a.findtext('url'),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
stats = Stats(
subject = a.findtext('name'),
playcount = int(a.findtext('playcount')),
)
)
for a in gen2(data):
yield a
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = self.__api._fetchData(params).find('albums')
for a in gen2(data):
yield a
return gen()
@LastfmBase.cachedProperty
def albums(self):
return self.getAlbums()
def getArtists(self,
limit = None):
params = {'method': 'library.getartists', 'user': self.user.name}
if limit is not None:
params.update({'limit': limit})
@lazylist
def gen(lst):
data = self.__api._fetchData(params).find('artists')
totalPages = int(data.attrib['totalPages'])
@lazylist
def gen2(lst, data):
for a in data.findall('artist'):
yield Artist(
self.__api,
subject = self,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
stats = Stats(
subject = a.findtext('name'),
playcount = a.findtext('playcount') and int(a.findtext('playcount')) or None,
tagcount = a.findtext('tagcount') and int(a.findtext('tagcount')) or None
),
url = a.findtext('url'),
streamable = (a.findtext('streamable') == "1"),
image = dict([(i.get('size'), i.text) for i in a.findall('image')]),
)
for a in gen2(data):
yield a
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = self.__api._fetchData(params).find('artists')
for a in gen2(data):
yield a
return gen()
@LastfmBase.cachedProperty
def artists(self):
return self.getArtists()
def getTracks(self,
limit = None):
params = {'method': 'library.gettracks', 'user': self.user.name}
if limit is not None:
params.update({'limit': limit})
@lazylist
def gen(lst):
data = self.__api._fetchData(params).find('tracks')
totalPages = int(data.attrib['totalPages'])
@lazylist
def gen2(lst, data):
for t in data.findall('track'):
yield Track(
self.__api,
subject = self,
name = t.findtext('name'),
artist = Artist(
self.__api,
subject = self,
name = t.findtext('artist/name'),
mbid = t.findtext('artist/mbid'),
url = t.findtext('artist/url'),
),
mbid = t.findtext('mbid'),
stats = Stats(
subject = t.findtext('name'),
playcount = t.findtext('playcount') and int(t.findtext('playcount')) or None,
tagcount = t.findtext('tagcount') and int(t.findtext('tagcount')) or None
),
streamable = (t.findtext('streamable') == '1'),
fullTrack = (t.find('streamable').attrib['fulltrack'] == '1'),
image = dict([(i.get('size'), i.text) for i in t.findall('image')]),
)
for t in gen2(data):
yield t
for page in xrange(2, totalPages+1):
params.update({'page': page})
data = self.__api._fetchData(params).find('tracks')
for t in gen2(data):
yield t
return gen()
@LastfmBase.cachedProperty
def tracks(self):
return self.getTracks()
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash(kwds['user'])
except KeyError:
raise LastfmInvalidParametersError("user has to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(user = self.user)
def __repr__(self):
return "<lastfm.User.Library: for user '%s'>" % self.user.name
from datetime import datetime
import time
from api import Api
from artist import Artist
from album import Album
from error import LastfmError, LastfmInvalidParametersError
from event import Event
from stats import Stats
from tag import Tag
from tasteometer import Tasteometer
from track import Track
from weeklychart import WeeklyChart, WeeklyAlbumChart, WeeklyArtistChart, WeeklyTrackChart

263
src/weeklychart.py Normal file
View File

@ -0,0 +1,263 @@
#!/usr/bin/env python
__author__ = "Abhinav Sarkar <abhinav@abhinavsarkar.net>"
__version__ = "0.1"
__license__ = "GNU Lesser General Public License"
from base import LastfmBase
class WeeklyChart(LastfmBase):
"""A class for representing the weekly charts"""
def init(self, subject, start, end,
stats = None):
self.__subject = subject
self.__start = start
self.__end = end
self.__stats = stats
@property
def subject(self):
return self.__subject
@property
def start(self):
return self.__start
@property
def end(self):
return self.__end
@property
def stats(self):
return self.__stats
@staticmethod
def createFromData(api, subject, data):
return WeeklyChart(
subject = subject,
start = datetime.utcfromtimestamp(int(data.attrib['from'])),
end = datetime.utcfromtimestamp(int(data.attrib['to']))
)
@staticmethod
def _checkWeeklyChartParams(params, start = None, end = None):
if (start is not None and end is None) or (start is None and end is not None):
raise LastfmInvalidParametersError("both start and end have to be provided.")
if start is not None and end is not None:
if isinstance(start, datetime) and isinstance(end, datetime):
params.update({
'from': int(calendar.timegm(start.timetuple())),
'to': int(calendar.timegm(end.timetuple()))
})
else:
raise LastfmInvalidParametersError("start and end must be datetime.datetime instances")
return params
@staticmethod
def hashFunc(*args, **kwds):
try:
return hash("%s%s%s%s" % (
kwds['subject'].__class__.__name__,
kwds['subject'].name,
kwds['start'],
kwds['end']
))
except KeyError:
raise LastfmInvalidParametersError("subject, start and end have to be provided for hashing")
def __hash__(self):
return self.__class__.hashFunc(
subject = self.subject,
start = self.start,
end = self.end
)
def __eq__(self, other):
return self.subject == other.subject and \
self.start == other.start and \
self.end == other.end
def __lt__(self, other):
if self.subject == other.subject:
if self.start == other.start:
return self.end < other.end
else:
return self.start < other.start
else:
return self.subject < other.subject
def __repr__(self):
return "<lastfm.%s: for %s:%s from %s to %s>" % \
(
self.__class__.__name__,
self.subject.__class__.__name__,
self.subject.name,
self.start.strftime("%x"),
self.end.strftime("%x"),
)
class WeeklyAlbumChart(WeeklyChart):
"""A class for representing the weekly album charts"""
def init(self, subject, start, end, stats, albums):
super(WeeklyAlbumChart, self).init(subject, start, end, stats)
self.__albums = albums
@property
def albums(self):
return self.__albums
@staticmethod
def createFromData(api, subject, data):
w = WeeklyChart(
subject = subject,
start = datetime.utcfromtimestamp(int(data.attrib['from'])),
end = datetime.utcfromtimestamp(int(data.attrib['to'])),
)
return WeeklyAlbumChart(
subject = subject,
start = datetime.utcfromtimestamp(int(data.attrib['from'])),
end = datetime.utcfromtimestamp(int(data.attrib['to'])),
stats = Stats(
subject = subject,
playcount = reduce(
lambda x,y:(
x + int(y.findtext('playcount'))
),
data.findall('album'),
0
)
),
albums = [
Album(
api,
subject = w,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
artist = Artist(
api,
subject = w,
name = a.findtext('artist'),
mbid = a.find('artist').attrib['mbid'],
),
stats = Stats(
subject = a.findtext('name'),
rank = int(a.attrib['rank']),
playcount = int(a.findtext('playcount')),
),
url = a.findtext('url'),
)
for a in data.findall('album')
]
)
class WeeklyArtistChart(WeeklyChart):
"""A class for representing the weekly artist charts"""
def init(self, subject, start, end, stats, artists):
super(WeeklyArtistChart, self).init(subject, start, end, stats)
self.__artists = artists
@property
def artists(self):
return self.__artists
@staticmethod
def createFromData(api, subject, data):
w = WeeklyChart(
subject = subject,
start = datetime.utcfromtimestamp(int(data.attrib['from'])),
end = datetime.utcfromtimestamp(int(data.attrib['to'])),
)
return WeeklyArtistChart(
subject = subject,
start = datetime.utcfromtimestamp(int(data.attrib['from'])),
end = datetime.utcfromtimestamp(int(data.attrib['to'])),
stats = Stats(
subject = subject,
playcount = reduce(
lambda x,y:(
x + int(y.findtext('playcount'))
),
data.findall('artist'),
0
)
),
artists = [
Artist(
api,
subject = w,
name = a.findtext('name'),
mbid = a.findtext('mbid'),
stats = Stats(
subject = a.findtext('name'),
rank = int(a.attrib['rank']),
playcount = int(a.findtext('playcount')),
),
url = a.findtext('url'),
)
for a in data.findall('artist')
]
)
class WeeklyTrackChart(WeeklyChart):
"""A class for representing the weekly track charts"""
def init(self, subject, start, end, tracks, stats):
super(WeeklyTrackChart, self).init(subject, start, end, stats)
self.__tracks = tracks
@property
def tracks(self):
return self.__tracks
@staticmethod
def createFromData(api, subject, data):
w = WeeklyChart(
subject = subject,
start = datetime.utcfromtimestamp(int(data.attrib['from'])),
end = datetime.utcfromtimestamp(int(data.attrib['to'])),
)
return WeeklyTrackChart(
subject = subject,
start = datetime.utcfromtimestamp(int(data.attrib['from'])),
end = datetime.utcfromtimestamp(int(data.attrib['to'])),
stats = Stats(
subject = subject,
playcount = reduce(
lambda x,y:(
x + int(y.findtext('playcount'))
),
data.findall('track'),
0
)
),
tracks = [
Track(
api,
subject = w,
name = t.findtext('name'),
mbid = t.findtext('mbid'),
artist = Artist(
api,
name = t.findtext('artist'),
mbid = t.find('artist').attrib['mbid'],
),
stats = Stats(
subject = t.findtext('name'),
rank = int(t.attrib['rank']),
playcount = int(t.findtext('playcount')),
),
url = t.findtext('url'),
)
for t in data.findall('track')
]
)
from datetime import datetime
import calendar
from album import Album
from artist import Artist
from error import LastfmInvalidParametersError
from stats import Stats
from track import Track