wolffd@0: # -*- coding: utf-8 -*- wolffd@0: # wolffd@0: # pylast - A Python interface to Last.fm (and other API compatible social networks) wolffd@0: # wolffd@0: # Copyright 2008-2010 Amr Hassan wolffd@0: # wolffd@0: # Licensed under the Apache License, Version 2.0 (the "License"); wolffd@0: # you may not use this file except in compliance with the License. wolffd@0: # You may obtain a copy of the License at wolffd@0: # wolffd@0: # http://www.apache.org/licenses/LICENSE-2.0 wolffd@0: # wolffd@0: # Unless required by applicable law or agreed to in writing, software wolffd@0: # distributed under the License is distributed on an "AS IS" BASIS, wolffd@0: # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. wolffd@0: # See the License for the specific language governing permissions and wolffd@0: # limitations under the License. wolffd@0: # wolffd@0: # http://code.google.com/p/pylast/ wolffd@0: wolffd@0: __version__ = '0.5' wolffd@0: __author__ = 'Amr Hassan' wolffd@0: __copyright__ = "Copyright (C) 2008-2010 Amr Hassan" wolffd@0: __license__ = "apache2" wolffd@0: __email__ = 'amr.hassan@gmail.com' wolffd@0: wolffd@0: import hashlib wolffd@0: from xml.dom import minidom wolffd@0: import xml.dom wolffd@0: import time wolffd@0: import shelve wolffd@0: import tempfile wolffd@0: import sys wolffd@0: import collections wolffd@0: import warnings wolffd@0: wolffd@0: def _deprecation_warning(message): wolffd@0: warnings.warn(message, DeprecationWarning) wolffd@0: wolffd@0: if sys.version_info[0] == 3: wolffd@0: from http.client import HTTPConnection wolffd@0: import html.entities as htmlentitydefs wolffd@0: from urllib.parse import splithost as url_split_host wolffd@0: from urllib.parse import quote_plus as url_quote_plus wolffd@0: wolffd@0: unichr = chr wolffd@0: wolffd@0: elif sys.version_info[0] == 2: wolffd@0: from httplib import HTTPConnection wolffd@0: import htmlentitydefs wolffd@0: from urllib import splithost as url_split_host wolffd@0: from urllib import quote_plus as url_quote_plus wolffd@0: wolffd@0: STATUS_INVALID_SERVICE = 2 wolffd@0: STATUS_INVALID_METHOD = 3 wolffd@0: STATUS_AUTH_FAILED = 4 wolffd@0: STATUS_INVALID_FORMAT = 5 wolffd@0: STATUS_INVALID_PARAMS = 6 wolffd@0: STATUS_INVALID_RESOURCE = 7 wolffd@0: STATUS_TOKEN_ERROR = 8 wolffd@0: STATUS_INVALID_SK = 9 wolffd@0: STATUS_INVALID_API_KEY = 10 wolffd@0: STATUS_OFFLINE = 11 wolffd@0: STATUS_SUBSCRIBERS_ONLY = 12 wolffd@0: STATUS_INVALID_SIGNATURE = 13 wolffd@0: STATUS_TOKEN_UNAUTHORIZED = 14 wolffd@0: STATUS_TOKEN_EXPIRED = 15 wolffd@0: wolffd@0: EVENT_ATTENDING = '0' wolffd@0: EVENT_MAYBE_ATTENDING = '1' wolffd@0: EVENT_NOT_ATTENDING = '2' wolffd@0: wolffd@0: PERIOD_OVERALL = 'overall' wolffd@0: PERIOD_7DAYS = "7day" wolffd@0: PERIOD_3MONTHS = '3month' wolffd@0: PERIOD_6MONTHS = '6month' wolffd@0: PERIOD_12MONTHS = '12month' wolffd@0: wolffd@0: DOMAIN_ENGLISH = 0 wolffd@0: DOMAIN_GERMAN = 1 wolffd@0: DOMAIN_SPANISH = 2 wolffd@0: DOMAIN_FRENCH = 3 wolffd@0: DOMAIN_ITALIAN = 4 wolffd@0: DOMAIN_POLISH = 5 wolffd@0: DOMAIN_PORTUGUESE = 6 wolffd@0: DOMAIN_SWEDISH = 7 wolffd@0: DOMAIN_TURKISH = 8 wolffd@0: DOMAIN_RUSSIAN = 9 wolffd@0: DOMAIN_JAPANESE = 10 wolffd@0: DOMAIN_CHINESE = 11 wolffd@0: wolffd@0: COVER_SMALL = 0 wolffd@0: COVER_MEDIUM = 1 wolffd@0: COVER_LARGE = 2 wolffd@0: COVER_EXTRA_LARGE = 3 wolffd@0: COVER_MEGA = 4 wolffd@0: wolffd@0: IMAGES_ORDER_POPULARITY = "popularity" wolffd@0: IMAGES_ORDER_DATE = "dateadded" wolffd@0: wolffd@0: wolffd@0: USER_MALE = 'Male' wolffd@0: USER_FEMALE = 'Female' wolffd@0: wolffd@0: SCROBBLE_SOURCE_USER = "P" wolffd@0: SCROBBLE_SOURCE_NON_PERSONALIZED_BROADCAST = "R" wolffd@0: SCROBBLE_SOURCE_PERSONALIZED_BROADCAST = "E" wolffd@0: SCROBBLE_SOURCE_LASTFM = "L" wolffd@0: SCROBBLE_SOURCE_UNKNOWN = "U" wolffd@0: wolffd@0: SCROBBLE_MODE_PLAYED = "" wolffd@0: SCROBBLE_MODE_LOVED = "L" wolffd@0: SCROBBLE_MODE_BANNED = "B" wolffd@0: SCROBBLE_MODE_SKIPPED = "S" wolffd@0: wolffd@0: class _Network(object): wolffd@0: """ wolffd@0: A music social network website that is Last.fm or one exposing a Last.fm compatible API wolffd@0: """ wolffd@0: wolffd@0: def __init__(self, name, homepage, ws_server, api_key, api_secret, session_key, submission_server, username, password_hash, wolffd@0: domain_names, urls): wolffd@0: """ wolffd@0: name: the name of the network wolffd@0: homepage: the homepage url wolffd@0: ws_server: the url of the webservices server wolffd@0: api_key: a provided API_KEY wolffd@0: api_secret: a provided API_SECRET wolffd@0: session_key: a generated session_key or None wolffd@0: submission_server: the url of the server to which tracks are submitted (scrobbled) wolffd@0: username: a username of a valid user wolffd@0: password_hash: the output of pylast.md5(password) where password is the user's password wolffd@0: domain_names: a dict mapping each DOMAIN_* value to a string domain name wolffd@0: urls: a dict mapping types to urls wolffd@0: wolffd@0: if username and password_hash were provided and not session_key, session_key will be wolffd@0: generated automatically when needed. wolffd@0: wolffd@0: Either a valid session_key or a combination of username and password_hash must be present for scrobbling. wolffd@0: wolffd@0: You should use a preconfigured network object through a get_*_network(...) method instead of creating an object wolffd@0: of this class, unless you know what you're doing. wolffd@0: """ wolffd@0: wolffd@0: self.name = name wolffd@0: self.homepage = homepage wolffd@0: self.ws_server = ws_server wolffd@0: self.api_key = api_key wolffd@0: self.api_secret = api_secret wolffd@0: self.session_key = session_key wolffd@0: self.submission_server = submission_server wolffd@0: self.username = username wolffd@0: self.password_hash = password_hash wolffd@0: self.domain_names = domain_names wolffd@0: self.urls = urls wolffd@0: wolffd@0: self.cache_backend = None wolffd@0: self.proxy_enabled = False wolffd@0: self.proxy = None wolffd@0: self.last_call_time = 0 wolffd@0: wolffd@0: #generate a session_key if necessary wolffd@0: if (self.api_key and self.api_secret) and not self.session_key and (self.username and self.password_hash): wolffd@0: sk_gen = SessionKeyGenerator(self) wolffd@0: self.session_key = sk_gen.get_session_key(self.username, self.password_hash) wolffd@0: wolffd@0: """def __repr__(self): wolffd@0: attributes = ("name", "homepage", "ws_server", "api_key", "api_secret", "session_key", "submission_server", wolffd@0: "username", "password_hash", "domain_names", "urls") wolffd@0: wolffd@0: text = "pylast._Network(%s)" wolffd@0: args = [] wolffd@0: for attr in attributes: wolffd@0: args.append("=".join((attr, repr(getattr(self, attr))))) wolffd@0: wolffd@0: return text % ", ".join(args) wolffd@0: """ wolffd@0: wolffd@0: def __str__(self): wolffd@0: return "The %s Network" %self.name wolffd@0: wolffd@0: def get_artist(self, artist_name): wolffd@0: """ wolffd@0: Return an Artist object wolffd@0: """ wolffd@0: wolffd@0: return Artist(artist_name, self) wolffd@0: wolffd@0: def get_track(self, artist, title): wolffd@0: """ wolffd@0: Return a Track object wolffd@0: """ wolffd@0: wolffd@0: return Track(artist, title, self) wolffd@0: wolffd@0: def get_album(self, artist, title): wolffd@0: """ wolffd@0: Return an Album object wolffd@0: """ wolffd@0: wolffd@0: return Album(artist, title, self) wolffd@0: wolffd@0: def get_authenticated_user(self): wolffd@0: """ wolffd@0: Returns the authenticated user wolffd@0: """ wolffd@0: wolffd@0: return AuthenticatedUser(self) wolffd@0: wolffd@0: def get_country(self, country_name): wolffd@0: """ wolffd@0: Returns a country object wolffd@0: """ wolffd@0: wolffd@0: return Country(country_name, self) wolffd@0: wolffd@0: def get_group(self, name): wolffd@0: """ wolffd@0: Returns a Group object wolffd@0: """ wolffd@0: wolffd@0: return Group(name, self) wolffd@0: wolffd@0: def get_user(self, username): wolffd@0: """ wolffd@0: Returns a user object wolffd@0: """ wolffd@0: wolffd@0: return User(username, self) wolffd@0: wolffd@0: def get_tag(self, name): wolffd@0: """ wolffd@0: Returns a tag object wolffd@0: """ wolffd@0: wolffd@0: return Tag(name, self) wolffd@0: wolffd@0: def get_scrobbler(self, client_id, client_version): wolffd@0: """ wolffd@0: Returns a Scrobbler object used for submitting tracks to the server wolffd@0: wolffd@0: Quote from http://www.last.fm/api/submissions: wolffd@0: ======== wolffd@0: Client identifiers are used to provide a centrally managed database of wolffd@0: the client versions, allowing clients to be banned if they are found to wolffd@0: be behaving undesirably. The client ID is associated with a version wolffd@0: number on the server, however these are only incremented if a client is wolffd@0: banned and do not have to reflect the version of the actual client application. wolffd@0: wolffd@0: During development, clients which have not been allocated an identifier should wolffd@0: use the identifier tst, with a version number of 1.0. Do not distribute code or wolffd@0: client implementations which use this test identifier. Do not use the identifiers wolffd@0: used by other clients. wolffd@0: ========= wolffd@0: wolffd@0: To obtain a new client identifier please contact: wolffd@0: * Last.fm: submissions@last.fm wolffd@0: * # TODO: list others wolffd@0: wolffd@0: ...and provide us with the name of your client and its homepage address. wolffd@0: """ wolffd@0: wolffd@0: _deprecation_warning("Use _Network.scrobble(...), _Network.scrobble_many(...), and Netowrk.update_now_playing(...) instead") wolffd@0: wolffd@0: return Scrobbler(self, client_id, client_version) wolffd@0: wolffd@0: def _get_language_domain(self, domain_language): wolffd@0: """ wolffd@0: Returns the mapped domain name of the network to a DOMAIN_* value wolffd@0: """ wolffd@0: wolffd@0: if domain_language in self.domain_names: wolffd@0: return self.domain_names[domain_language] wolffd@0: wolffd@0: def _get_url(self, domain, type): wolffd@0: return "http://%s/%s" %(self._get_language_domain(domain), self.urls[type]) wolffd@0: wolffd@0: def _get_ws_auth(self): wolffd@0: """ wolffd@0: Returns a (API_KEY, API_SECRET, SESSION_KEY) tuple. wolffd@0: """ wolffd@0: return (self.api_key, self.api_secret, self.session_key) wolffd@0: wolffd@0: def _delay_call(self): wolffd@0: """ wolffd@0: Makes sure that web service calls are at least a second apart wolffd@0: """ wolffd@0: wolffd@0: # delay time in seconds wolffd@0: DELAY_TIME = 1.0 wolffd@0: now = time.time() wolffd@0: wolffd@0: if (now - self.last_call_time) < DELAY_TIME: wolffd@0: time.sleep(1) wolffd@0: wolffd@0: self.last_call_time = now wolffd@0: wolffd@0: def create_new_playlist(self, title, description): wolffd@0: """ wolffd@0: Creates a playlist for the authenticated user and returns it wolffd@0: title: The title of the new playlist. wolffd@0: description: The description of the new playlist. wolffd@0: """ wolffd@0: wolffd@0: params = {} wolffd@0: params['title'] = title wolffd@0: params['description'] = description wolffd@0: wolffd@0: doc = _Request(self, 'playlist.create', params).execute(False) wolffd@0: wolffd@0: e_id = doc.getElementsByTagName("id")[0].firstChild.data wolffd@0: user = doc.getElementsByTagName('playlists')[0].getAttribute('user') wolffd@0: wolffd@0: return Playlist(user, e_id, self) wolffd@0: wolffd@0: def get_top_tags(self, limit=None): wolffd@0: """Returns a sequence of the most used tags as a sequence of TopItem objects.""" wolffd@0: wolffd@0: doc = _Request(self, "tag.getTopTags").execute(True) wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("tag"): wolffd@0: tag = Tag(_extract(node, "name"), self) wolffd@0: weight = _number(_extract(node, "count")) wolffd@0: wolffd@0: seq.append(TopItem(tag, weight)) wolffd@0: wolffd@0: if limit: wolffd@0: seq = seq[:limit] wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def enable_proxy(self, host, port): wolffd@0: """Enable a default web proxy""" wolffd@0: wolffd@0: self.proxy = [host, _number(port)] wolffd@0: self.proxy_enabled = True wolffd@0: wolffd@0: def disable_proxy(self): wolffd@0: """Disable using the web proxy""" wolffd@0: wolffd@0: self.proxy_enabled = False wolffd@0: wolffd@0: def is_proxy_enabled(self): wolffd@0: """Returns True if a web proxy is enabled.""" wolffd@0: wolffd@0: return self.proxy_enabled wolffd@0: wolffd@0: def _get_proxy(self): wolffd@0: """Returns proxy details.""" wolffd@0: wolffd@0: return self.proxy wolffd@0: wolffd@0: def enable_caching(self, file_path = None): wolffd@0: """Enables caching request-wide for all cachable calls. wolffd@0: In choosing the backend used for caching, it will try _SqliteCacheBackend first if wolffd@0: the module sqlite3 is present. If not, it will fallback to _ShelfCacheBackend which uses shelve.Shelf objects. wolffd@0: wolffd@0: * file_path: A file path for the backend storage file. If wolffd@0: None set, a temp file would probably be created, according the backend. wolffd@0: """ wolffd@0: wolffd@0: if not file_path: wolffd@0: file_path = tempfile.mktemp(prefix="pylast_tmp_") wolffd@0: wolffd@0: self.cache_backend = _ShelfCacheBackend(file_path) wolffd@0: wolffd@0: def disable_caching(self): wolffd@0: """Disables all caching features.""" wolffd@0: wolffd@0: self.cache_backend = None wolffd@0: wolffd@0: def is_caching_enabled(self): wolffd@0: """Returns True if caching is enabled.""" wolffd@0: wolffd@0: return not (self.cache_backend == None) wolffd@0: wolffd@0: def _get_cache_backend(self): wolffd@0: wolffd@0: return self.cache_backend wolffd@0: wolffd@0: def search_for_album(self, album_name): wolffd@0: """Searches for an album by its name. Returns a AlbumSearch object. wolffd@0: Use get_next_page() to retreive sequences of results.""" wolffd@0: wolffd@0: return AlbumSearch(album_name, self) wolffd@0: wolffd@0: def search_for_artist(self, artist_name): wolffd@0: """Searches of an artist by its name. Returns a ArtistSearch object. wolffd@0: Use get_next_page() to retreive sequences of results.""" wolffd@0: wolffd@0: return ArtistSearch(artist_name, self) wolffd@0: wolffd@0: def search_for_tag(self, tag_name): wolffd@0: """Searches of a tag by its name. Returns a TagSearch object. wolffd@0: Use get_next_page() to retreive sequences of results.""" wolffd@0: wolffd@0: return TagSearch(tag_name, self) wolffd@0: wolffd@0: def search_for_track(self, artist_name, track_name): wolffd@0: """Searches of a track by its name and its artist. Set artist to an empty string if not available. wolffd@0: Returns a TrackSearch object. wolffd@0: Use get_next_page() to retreive sequences of results.""" wolffd@0: wolffd@0: return TrackSearch(artist_name, track_name, self) wolffd@0: wolffd@0: def search_for_venue(self, venue_name, country_name): wolffd@0: """Searches of a venue by its name and its country. Set country_name to an empty string if not available. wolffd@0: Returns a VenueSearch object. wolffd@0: Use get_next_page() to retreive sequences of results.""" wolffd@0: wolffd@0: return VenueSearch(venue_name, country_name, self) wolffd@0: wolffd@0: def get_track_by_mbid(self, mbid): wolffd@0: """Looks up a track by its MusicBrainz ID""" wolffd@0: wolffd@0: params = {"mbid": mbid} wolffd@0: wolffd@0: doc = _Request(self, "track.getInfo", params).execute(True) wolffd@0: wolffd@0: return Track(_extract(doc, "name", 1), _extract(doc, "name"), self) wolffd@0: wolffd@0: def get_artist_by_mbid(self, mbid): wolffd@0: """Loooks up an artist by its MusicBrainz ID""" wolffd@0: wolffd@0: params = {"mbid": mbid} wolffd@0: wolffd@0: doc = _Request(self, "artist.getInfo", params).execute(True) wolffd@0: wolffd@0: return Artist(_extract(doc, "name"), self) wolffd@0: wolffd@0: def get_album_by_mbid(self, mbid): wolffd@0: """Looks up an album by its MusicBrainz ID""" wolffd@0: wolffd@0: params = {"mbid": mbid} wolffd@0: wolffd@0: doc = _Request(self, "album.getInfo", params).execute(True) wolffd@0: wolffd@0: return Album(_extract(doc, "artist"), _extract(doc, "name"), self) wolffd@0: wolffd@0: def update_now_playing(self, artist, title, album = None, album_artist = None, wolffd@0: duration = None, track_number = None, mbid = None, context = None): wolffd@0: """ wolffd@0: Used to notify Last.fm that a user has started listening to a track. wolffd@0: wolffd@0: Parameters: wolffd@0: artist (Required) : The artist name wolffd@0: title (Required) : The track title wolffd@0: album (Optional) : The album name. wolffd@0: album_artist (Optional) : The album artist - if this differs from the track artist. wolffd@0: duration (Optional) : The length of the track in seconds. wolffd@0: track_number (Optional) : The track number of the track on the album. wolffd@0: mbid (Optional) : The MusicBrainz Track ID. wolffd@0: context (Optional) : Sub-client version (not public, only enabled for certain API keys) wolffd@0: """ wolffd@0: wolffd@0: params = {"track": title, "artist": artist} wolffd@0: wolffd@0: if album: params["album"] = album wolffd@0: if album_artist: params["albumArtist"] = album_artist wolffd@0: if context: params["context"] = context wolffd@0: if track_number: params["trackNumber"] = track_number wolffd@0: if mbid: params["mbid"] = mbid wolffd@0: if duration: params["duration"] = duration wolffd@0: wolffd@0: _Request(self, "track.updateNowPlaying", params).execute() wolffd@0: wolffd@0: def scrobble(self, artist, title, timestamp, album = None, album_artist = None, track_number = None, wolffd@0: duration = None, stream_id = None, context = None, mbid = None): wolffd@0: wolffd@0: """Used to add a track-play to a user's profile. wolffd@0: wolffd@0: Parameters: wolffd@0: artist (Required) : The artist name. wolffd@0: title (Required) : The track name. wolffd@0: timestamp (Required) : The time the track started playing, in UNIX timestamp format (integer number of seconds since 00:00:00, January 1st 1970 UTC). This must be in the UTC time zone. wolffd@0: album (Optional) : The album name. wolffd@0: album_artist (Optional) : The album artist - if this differs from the track artist. wolffd@0: context (Optional) : Sub-client version (not public, only enabled for certain API keys) wolffd@0: stream_id (Optional) : The stream id for this track received from the radio.getPlaylist service. wolffd@0: track_number (Optional) : The track number of the track on the album. wolffd@0: mbid (Optional) : The MusicBrainz Track ID. wolffd@0: duration (Optional) : The length of the track in seconds. wolffd@0: """ wolffd@0: wolffd@0: return self.scrobble_many(({"artist": artist, "title": title, "timestamp": timestamp, "album": album, "album_artist": album_artist, wolffd@0: "track_number": track_number, "duration": duration, "stream_id": stream_id, "context": context, "mbid": mbid},)) wolffd@0: wolffd@0: def scrobble_many(self, tracks): wolffd@0: """ wolffd@0: Used to scrobble a batch of tracks at once. The parameter tracks is a sequence of dicts per wolffd@0: track containing the keyword arguments as if passed to the scrobble() method. wolffd@0: """ wolffd@0: wolffd@0: tracks_to_scrobble = tracks[:50] wolffd@0: if len(tracks) > 50: wolffd@0: remaining_tracks = tracks[50:] wolffd@0: else: wolffd@0: remaining_tracks = None wolffd@0: wolffd@0: params = {} wolffd@0: for i in range(len(tracks_to_scrobble)): wolffd@0: wolffd@0: params["artist[%d]" % i] = tracks_to_scrobble[i]["artist"] wolffd@0: params["track[%d]" % i] = tracks_to_scrobble[i]["title"] wolffd@0: wolffd@0: additional_args = ("timestamp", "album", "album_artist", "context", "stream_id", "track_number", "mbid", "duration") wolffd@0: args_map_to = {"album_artist": "albumArtist", "track_number": "trackNumber", "stream_id": "streamID"} # so friggin lazy wolffd@0: wolffd@0: for arg in additional_args: wolffd@0: wolffd@0: if arg in tracks_to_scrobble[i] and tracks_to_scrobble[i][arg]: wolffd@0: if arg in args_map_to: wolffd@0: maps_to = args_map_to[arg] wolffd@0: else: wolffd@0: maps_to = arg wolffd@0: wolffd@0: params["%s[%d]" %(maps_to, i)] = tracks_to_scrobble[i][arg] wolffd@0: wolffd@0: wolffd@0: _Request(self, "track.scrobble", params).execute() wolffd@0: wolffd@0: if remaining_tracks: wolffd@0: self.scrobble_many(remaining_tracks) wolffd@0: wolffd@0: class LastFMNetwork(_Network): wolffd@0: wolffd@0: """A Last.fm network object wolffd@0: wolffd@0: api_key: a provided API_KEY wolffd@0: api_secret: a provided API_SECRET wolffd@0: session_key: a generated session_key or None wolffd@0: username: a username of a valid user wolffd@0: password_hash: the output of pylast.md5(password) where password is the user's password wolffd@0: wolffd@0: if username and password_hash were provided and not session_key, session_key will be wolffd@0: generated automatically when needed. wolffd@0: wolffd@0: Either a valid session_key or a combination of username and password_hash must be present for scrobbling. wolffd@0: wolffd@0: Most read-only webservices only require an api_key and an api_secret, see about obtaining them from: wolffd@0: http://www.last.fm/api/account wolffd@0: """ wolffd@0: wolffd@0: def __init__(self, api_key="", api_secret="", session_key="", username="", password_hash=""): wolffd@0: _Network.__init__(self, wolffd@0: name = "Last.fm", wolffd@0: homepage = "http://last.fm", wolffd@0: ws_server = ("ws.audioscrobbler.com", "/2.0/"), wolffd@0: api_key = api_key, wolffd@0: api_secret = api_secret, wolffd@0: session_key = session_key, wolffd@0: submission_server = "http://post.audioscrobbler.com:80/", wolffd@0: username = username, wolffd@0: password_hash = password_hash, wolffd@0: domain_names = { wolffd@0: DOMAIN_ENGLISH: 'www.last.fm', wolffd@0: DOMAIN_GERMAN: 'www.lastfm.de', wolffd@0: DOMAIN_SPANISH: 'www.lastfm.es', wolffd@0: DOMAIN_FRENCH: 'www.lastfm.fr', wolffd@0: DOMAIN_ITALIAN: 'www.lastfm.it', wolffd@0: DOMAIN_POLISH: 'www.lastfm.pl', wolffd@0: DOMAIN_PORTUGUESE: 'www.lastfm.com.br', wolffd@0: DOMAIN_SWEDISH: 'www.lastfm.se', wolffd@0: DOMAIN_TURKISH: 'www.lastfm.com.tr', wolffd@0: DOMAIN_RUSSIAN: 'www.lastfm.ru', wolffd@0: DOMAIN_JAPANESE: 'www.lastfm.jp', wolffd@0: DOMAIN_CHINESE: 'cn.last.fm', wolffd@0: }, wolffd@0: urls = { wolffd@0: "album": "music/%(artist)s/%(album)s", wolffd@0: "artist": "music/%(artist)s", wolffd@0: "event": "event/%(id)s", wolffd@0: "country": "place/%(country_name)s", wolffd@0: "playlist": "user/%(user)s/library/playlists/%(appendix)s", wolffd@0: "tag": "tag/%(name)s", wolffd@0: "track": "music/%(artist)s/_/%(title)s", wolffd@0: "group": "group/%(name)s", wolffd@0: "user": "user/%(name)s", wolffd@0: } wolffd@0: ) wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.LastFMNetwork(%s)" %(", ".join(("'%s'" %self.api_key, "'%s'" %self.api_secret, "'%s'" %self.session_key, wolffd@0: "'%s'" %self.username, "'%s'" %self.password_hash))) wolffd@0: wolffd@0: def __str__(self): wolffd@0: return "LastFM Network" wolffd@0: wolffd@0: def get_lastfm_network(api_key="", api_secret="", session_key = "", username = "", password_hash = ""): wolffd@0: """ wolffd@0: Returns a preconfigured _Network object for Last.fm wolffd@0: wolffd@0: api_key: a provided API_KEY wolffd@0: api_secret: a provided API_SECRET wolffd@0: session_key: a generated session_key or None wolffd@0: username: a username of a valid user wolffd@0: password_hash: the output of pylast.md5(password) where password is the user's password wolffd@0: wolffd@0: if username and password_hash were provided and not session_key, session_key will be wolffd@0: generated automatically when needed. wolffd@0: wolffd@0: Either a valid session_key or a combination of username and password_hash must be present for scrobbling. wolffd@0: wolffd@0: Most read-only webservices only require an api_key and an api_secret, see about obtaining them from: wolffd@0: http://www.last.fm/api/account wolffd@0: """ wolffd@0: wolffd@0: _deprecation_warning("Create a LastFMNetwork object instead") wolffd@0: wolffd@0: return LastFMNetwork(api_key, api_secret, session_key, username, password_hash) wolffd@0: wolffd@0: class LibreFMNetwork(_Network): wolffd@0: """ wolffd@0: A preconfigured _Network object for Libre.fm wolffd@0: wolffd@0: api_key: a provided API_KEY wolffd@0: api_secret: a provided API_SECRET wolffd@0: session_key: a generated session_key or None wolffd@0: username: a username of a valid user wolffd@0: password_hash: the output of pylast.md5(password) where password is the user's password wolffd@0: wolffd@0: if username and password_hash were provided and not session_key, session_key will be wolffd@0: generated automatically when needed. wolffd@0: """ wolffd@0: wolffd@0: def __init__(self, api_key="", api_secret="", session_key = "", username = "", password_hash = ""): wolffd@0: wolffd@0: _Network.__init__(self, wolffd@0: name = "Libre.fm", wolffd@0: homepage = "http://alpha.dev.libre.fm", wolffd@0: ws_server = ("alpha.dev.libre.fm", "/2.0/"), wolffd@0: api_key = api_key, wolffd@0: api_secret = api_secret, wolffd@0: session_key = session_key, wolffd@0: submission_server = "http://turtle.libre.fm:80/", wolffd@0: username = username, wolffd@0: password_hash = password_hash, wolffd@0: domain_names = { wolffd@0: DOMAIN_ENGLISH: "alpha.dev.libre.fm", wolffd@0: DOMAIN_GERMAN: "alpha.dev.libre.fm", wolffd@0: DOMAIN_SPANISH: "alpha.dev.libre.fm", wolffd@0: DOMAIN_FRENCH: "alpha.dev.libre.fm", wolffd@0: DOMAIN_ITALIAN: "alpha.dev.libre.fm", wolffd@0: DOMAIN_POLISH: "alpha.dev.libre.fm", wolffd@0: DOMAIN_PORTUGUESE: "alpha.dev.libre.fm", wolffd@0: DOMAIN_SWEDISH: "alpha.dev.libre.fm", wolffd@0: DOMAIN_TURKISH: "alpha.dev.libre.fm", wolffd@0: DOMAIN_RUSSIAN: "alpha.dev.libre.fm", wolffd@0: DOMAIN_JAPANESE: "alpha.dev.libre.fm", wolffd@0: DOMAIN_CHINESE: "alpha.dev.libre.fm", wolffd@0: }, wolffd@0: urls = { wolffd@0: "album": "artist/%(artist)s/album/%(album)s", wolffd@0: "artist": "artist/%(artist)s", wolffd@0: "event": "event/%(id)s", wolffd@0: "country": "place/%(country_name)s", wolffd@0: "playlist": "user/%(user)s/library/playlists/%(appendix)s", wolffd@0: "tag": "tag/%(name)s", wolffd@0: "track": "music/%(artist)s/_/%(title)s", wolffd@0: "group": "group/%(name)s", wolffd@0: "user": "user/%(name)s", wolffd@0: } wolffd@0: ) wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.LibreFMNetwork(%s)" %(", ".join(("'%s'" %self.api_key, "'%s'" %self.api_secret, "'%s'" %self.session_key, wolffd@0: "'%s'" %self.username, "'%s'" %self.password_hash))) wolffd@0: wolffd@0: def __str__(self): wolffd@0: return "Libre.fm Network" wolffd@0: wolffd@0: def get_librefm_network(api_key="", api_secret="", session_key = "", username = "", password_hash = ""): wolffd@0: """ wolffd@0: Returns a preconfigured _Network object for Libre.fm wolffd@0: wolffd@0: api_key: a provided API_KEY wolffd@0: api_secret: a provided API_SECRET wolffd@0: session_key: a generated session_key or None wolffd@0: username: a username of a valid user wolffd@0: password_hash: the output of pylast.md5(password) where password is the user's password wolffd@0: wolffd@0: if username and password_hash were provided and not session_key, session_key will be wolffd@0: generated automatically when needed. wolffd@0: """ wolffd@0: wolffd@0: _deprecation_warning("DeprecationWarning: Create a LibreFMNetwork object instead") wolffd@0: wolffd@0: return LibreFMNetwork(api_key, api_secret, session_key, username, password_hash) wolffd@0: wolffd@0: class _ShelfCacheBackend(object): wolffd@0: """Used as a backend for caching cacheable requests.""" wolffd@0: def __init__(self, file_path = None): wolffd@0: self.shelf = shelve.open(file_path) wolffd@0: wolffd@0: def get_xml(self, key): wolffd@0: return self.shelf[key] wolffd@0: wolffd@0: def set_xml(self, key, xml_string): wolffd@0: self.shelf[key] = xml_string wolffd@0: wolffd@0: def has_key(self, key): wolffd@0: return key in self.shelf.keys() wolffd@0: wolffd@0: class _Request(object): wolffd@0: """Representing an abstract web service operation.""" wolffd@0: wolffd@0: def __init__(self, network, method_name, params = {}): wolffd@0: wolffd@0: self.network = network wolffd@0: self.params = {} wolffd@0: wolffd@0: for key in params: wolffd@0: self.params[key] = _unicode(params[key]) wolffd@0: wolffd@0: (self.api_key, self.api_secret, self.session_key) = network._get_ws_auth() wolffd@0: wolffd@0: self.params["api_key"] = self.api_key wolffd@0: self.params["method"] = method_name wolffd@0: wolffd@0: if network.is_caching_enabled(): wolffd@0: self.cache = network._get_cache_backend() wolffd@0: wolffd@0: if self.session_key: wolffd@0: self.params["sk"] = self.session_key wolffd@0: self.sign_it() wolffd@0: wolffd@0: def sign_it(self): wolffd@0: """Sign this request.""" wolffd@0: wolffd@0: if not "api_sig" in self.params.keys(): wolffd@0: self.params['api_sig'] = self._get_signature() wolffd@0: wolffd@0: def _get_signature(self): wolffd@0: """Returns a 32-character hexadecimal md5 hash of the signature string.""" wolffd@0: wolffd@0: keys = list(self.params.keys()) wolffd@0: wolffd@0: keys.sort() wolffd@0: wolffd@0: string = "" wolffd@0: wolffd@0: for name in keys: wolffd@0: string += name wolffd@0: string += self.params[name] wolffd@0: wolffd@0: string += self.api_secret wolffd@0: wolffd@0: return md5(string) wolffd@0: wolffd@0: def _get_cache_key(self): wolffd@0: """The cache key is a string of concatenated sorted names and values.""" wolffd@0: wolffd@0: keys = list(self.params.keys()) wolffd@0: keys.sort() wolffd@0: wolffd@0: cache_key = str() wolffd@0: wolffd@0: for key in keys: wolffd@0: if key != "api_sig" and key != "api_key" and key != "sk": wolffd@0: cache_key += key + _string(self.params[key]) wolffd@0: wolffd@0: return hashlib.sha1(cache_key).hexdigest() wolffd@0: wolffd@0: def _get_cached_response(self): wolffd@0: """Returns a file object of the cached response.""" wolffd@0: wolffd@0: if not self._is_cached(): wolffd@0: response = self._download_response() wolffd@0: self.cache.set_xml(self._get_cache_key(), response) wolffd@0: wolffd@0: return self.cache.get_xml(self._get_cache_key()) wolffd@0: wolffd@0: def _is_cached(self): wolffd@0: """Returns True if the request is already in cache.""" wolffd@0: wolffd@0: return self.cache.has_key(self._get_cache_key()) wolffd@0: wolffd@0: def _download_response(self): wolffd@0: """Returns a response body string from the server.""" wolffd@0: wolffd@0: # Delay the call if necessary wolffd@0: #self.network._delay_call() # enable it if you want. wolffd@0: wolffd@0: data = [] wolffd@0: for name in self.params.keys(): wolffd@0: data.append('='.join((name, url_quote_plus(_string(self.params[name]))))) wolffd@0: data = '&'.join(data) wolffd@0: wolffd@0: headers = { wolffd@0: "Content-type": "application/x-www-form-urlencoded", wolffd@0: 'Accept-Charset': 'utf-8', wolffd@0: 'User-Agent': "pylast" + '/' + __version__ wolffd@0: } wolffd@0: wolffd@0: (HOST_NAME, HOST_SUBDIR) = self.network.ws_server wolffd@0: wolffd@0: if self.network.is_proxy_enabled(): wolffd@0: conn = HTTPConnection(host = self._get_proxy()[0], port = self._get_proxy()[1]) wolffd@0: wolffd@0: try: wolffd@0: conn.request(method='POST', url="http://" + HOST_NAME + HOST_SUBDIR, wolffd@0: body=data, headers=headers) wolffd@0: except Exception as e: wolffd@0: raise NetworkError(self.network, e) wolffd@0: wolffd@0: else: wolffd@0: conn = HTTPConnection(host=HOST_NAME) wolffd@0: wolffd@0: try: wolffd@0: conn.request(method='POST', url=HOST_SUBDIR, body=data, headers=headers) wolffd@0: except Exception as e: wolffd@0: raise NetworkError(self.network, e) wolffd@0: wolffd@0: try: wolffd@0: response_text = _unicode(conn.getresponse().read()) wolffd@0: except Exception as e: wolffd@0: raise MalformedResponseError(self.network, e) wolffd@0: wolffd@0: self._check_response_for_errors(response_text) wolffd@0: return response_text wolffd@0: wolffd@0: def execute(self, cacheable = False): wolffd@0: """Returns the XML DOM response of the POST Request from the server""" wolffd@0: wolffd@0: if self.network.is_caching_enabled() and cacheable: wolffd@0: response = self._get_cached_response() wolffd@0: else: wolffd@0: response = self._download_response() wolffd@0: wolffd@0: return minidom.parseString(_string(response)) wolffd@0: wolffd@0: def _check_response_for_errors(self, response): wolffd@0: """Checks the response for errors and raises one if any exists.""" wolffd@0: wolffd@0: try: wolffd@0: doc = minidom.parseString(_string(response)) wolffd@0: except Exception as e: wolffd@0: raise MalformedResponseError(self.network, e) wolffd@0: wolffd@0: e = doc.getElementsByTagName('lfm')[0] wolffd@0: wolffd@0: if e.getAttribute('status') != "ok": wolffd@0: e = doc.getElementsByTagName('error')[0] wolffd@0: status = e.getAttribute('code') wolffd@0: details = e.firstChild.data.strip() wolffd@0: raise WSError(self.network, status, details) wolffd@0: wolffd@0: class SessionKeyGenerator(object): wolffd@0: """Methods of generating a session key: wolffd@0: 1) Web Authentication: wolffd@0: a. network = get_*_network(API_KEY, API_SECRET) wolffd@0: b. sg = SessionKeyGenerator(network) wolffd@0: c. url = sg.get_web_auth_url() wolffd@0: d. Ask the user to open the url and authorize you, and wait for it. wolffd@0: e. session_key = sg.get_web_auth_session_key(url) wolffd@0: 2) Username and Password Authentication: wolffd@0: a. network = get_*_network(API_KEY, API_SECRET) wolffd@0: b. username = raw_input("Please enter your username: ") wolffd@0: c. password_hash = pylast.md5(raw_input("Please enter your password: ") wolffd@0: d. session_key = SessionKeyGenerator(network).get_session_key(username, password_hash) wolffd@0: wolffd@0: A session key's lifetime is infinie, unless the user provokes the rights of the given API Key. wolffd@0: wolffd@0: If you create a Network object with just a API_KEY and API_SECRET and a username and a password_hash, a wolffd@0: SESSION_KEY will be automatically generated for that network and stored in it so you don't have to do this wolffd@0: manually, unless you want to. wolffd@0: """ wolffd@0: wolffd@0: def __init__(self, network): wolffd@0: self.network = network wolffd@0: self.web_auth_tokens = {} wolffd@0: wolffd@0: def _get_web_auth_token(self): wolffd@0: """Retrieves a token from the network for web authentication. wolffd@0: The token then has to be authorized from getAuthURL before creating session. wolffd@0: """ wolffd@0: wolffd@0: request = _Request(self.network, 'auth.getToken') wolffd@0: wolffd@0: # default action is that a request is signed only when wolffd@0: # a session key is provided. wolffd@0: request.sign_it() wolffd@0: wolffd@0: doc = request.execute() wolffd@0: wolffd@0: e = doc.getElementsByTagName('token')[0] wolffd@0: return e.firstChild.data wolffd@0: wolffd@0: def get_web_auth_url(self): wolffd@0: """The user must open this page, and you first, then call get_web_auth_session_key(url) after that.""" wolffd@0: wolffd@0: token = self._get_web_auth_token() wolffd@0: wolffd@0: url = '%(homepage)s/api/auth/?api_key=%(api)s&token=%(token)s' % \ wolffd@0: {"homepage": self.network.homepage, "api": self.network.api_key, "token": token} wolffd@0: wolffd@0: self.web_auth_tokens[url] = token wolffd@0: wolffd@0: return url wolffd@0: wolffd@0: def get_web_auth_session_key(self, url): wolffd@0: """Retrieves the session key of a web authorization process by its url.""" wolffd@0: wolffd@0: if url in self.web_auth_tokens.keys(): wolffd@0: token = self.web_auth_tokens[url] wolffd@0: else: wolffd@0: token = "" #that's gonna raise a WSError of an unauthorized token when the request is executed. wolffd@0: wolffd@0: request = _Request(self.network, 'auth.getSession', {'token': token}) wolffd@0: wolffd@0: # default action is that a request is signed only when wolffd@0: # a session key is provided. wolffd@0: request.sign_it() wolffd@0: wolffd@0: doc = request.execute() wolffd@0: wolffd@0: return doc.getElementsByTagName('key')[0].firstChild.data wolffd@0: wolffd@0: def get_session_key(self, username, password_hash): wolffd@0: """Retrieve a session key with a username and a md5 hash of the user's password.""" wolffd@0: wolffd@0: params = {"username": username, "authToken": md5(username + password_hash)} wolffd@0: request = _Request(self.network, "auth.getMobileSession", params) wolffd@0: wolffd@0: # default action is that a request is signed only when wolffd@0: # a session key is provided. wolffd@0: request.sign_it() wolffd@0: wolffd@0: doc = request.execute() wolffd@0: wolffd@0: return _extract(doc, "key") wolffd@0: wolffd@0: TopItem = collections.namedtuple("TopItem", ["item", "weight"]) wolffd@0: SimilarItem = collections.namedtuple("SimilarItem", ["item", "match"]) wolffd@0: LibraryItem = collections.namedtuple("LibraryItem", ["item", "playcount", "tagcount"]) wolffd@0: PlayedTrack = collections.namedtuple("PlayedTrack", ["track", "playback_date", "timestamp"]) wolffd@0: LovedTrack = collections.namedtuple("LovedTrack", ["track", "date", "timestamp"]) wolffd@0: ImageSizes = collections.namedtuple("ImageSizes", ["original", "large", "largesquare", "medium", "small", "extralarge"]) wolffd@0: Image = collections.namedtuple("Image", ["title", "url", "dateadded", "format", "owner", "sizes", "votes"]) wolffd@0: Shout = collections.namedtuple("Shout", ["body", "author", "date"]) wolffd@0: wolffd@0: def _string_output(funct): wolffd@0: def r(*args): wolffd@0: return _string(funct(*args)) wolffd@0: wolffd@0: return r wolffd@0: wolffd@0: def _pad_list(given_list, desired_length, padding = None): wolffd@0: """ wolffd@0: Pads a list to be of the desired_length. wolffd@0: """ wolffd@0: wolffd@0: while len(given_list) < desired_length: wolffd@0: given_list.append(padding) wolffd@0: wolffd@0: return given_list wolffd@0: wolffd@0: class _BaseObject(object): wolffd@0: """An abstract webservices object.""" wolffd@0: wolffd@0: network = None wolffd@0: wolffd@0: def __init__(self, network): wolffd@0: self.network = network wolffd@0: wolffd@0: def _request(self, method_name, cacheable = False, params = None): wolffd@0: if not params: wolffd@0: params = self._get_params() wolffd@0: wolffd@0: return _Request(self.network, method_name, params).execute(cacheable) wolffd@0: wolffd@0: def _get_params(self): wolffd@0: """Returns the most common set of parameters between all objects.""" wolffd@0: wolffd@0: return {} wolffd@0: wolffd@0: def __hash__(self): wolffd@0: return hash(self.network) + \ wolffd@0: hash(str(type(self)) + "".join(list(self._get_params().keys()) + list(self._get_params().values())).lower()) wolffd@0: wolffd@0: class _Taggable(object): wolffd@0: """Common functions for classes with tags.""" wolffd@0: wolffd@0: def __init__(self, ws_prefix): wolffd@0: self.ws_prefix = ws_prefix wolffd@0: wolffd@0: def add_tags(self, tags): wolffd@0: """Adds one or several tags. wolffd@0: * tags: A sequence of tag names or Tag objects. wolffd@0: """ wolffd@0: wolffd@0: for tag in tags: wolffd@0: self.add_tag(tag) wolffd@0: wolffd@0: def add_tag(self, tag): wolffd@0: """Adds one tag. wolffd@0: * tag: a tag name or a Tag object. wolffd@0: """ wolffd@0: wolffd@0: if isinstance(tag, Tag): wolffd@0: tag = tag.get_name() wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['tags'] = tag wolffd@0: wolffd@0: self._request(self.ws_prefix + '.addTags', False, params) wolffd@0: wolffd@0: def remove_tag(self, tag): wolffd@0: """Remove a user's tag from this object.""" wolffd@0: wolffd@0: if isinstance(tag, Tag): wolffd@0: tag = tag.get_name() wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['tag'] = tag wolffd@0: wolffd@0: self._request(self.ws_prefix + '.removeTag', False, params) wolffd@0: wolffd@0: def get_tags(self): wolffd@0: """Returns a list of the tags set by the user to this object.""" wolffd@0: wolffd@0: # Uncacheable because it can be dynamically changed by the user. wolffd@0: params = self._get_params() wolffd@0: wolffd@0: doc = self._request(self.ws_prefix + '.getTags', False, params) wolffd@0: tag_names = _extract_all(doc, 'name') wolffd@0: tags = [] wolffd@0: for tag in tag_names: wolffd@0: tags.append(Tag(tag, self.network)) wolffd@0: wolffd@0: return tags wolffd@0: wolffd@0: def remove_tags(self, tags): wolffd@0: """Removes one or several tags from this object. wolffd@0: * tags: a sequence of tag names or Tag objects. wolffd@0: """ wolffd@0: wolffd@0: for tag in tags: wolffd@0: self.remove_tag(tag) wolffd@0: wolffd@0: def clear_tags(self): wolffd@0: """Clears all the user-set tags. """ wolffd@0: wolffd@0: self.remove_tags(*(self.get_tags())) wolffd@0: wolffd@0: def set_tags(self, tags): wolffd@0: """Sets this object's tags to only those tags. wolffd@0: * tags: a sequence of tag names or Tag objects. wolffd@0: """ wolffd@0: wolffd@0: c_old_tags = [] wolffd@0: old_tags = [] wolffd@0: c_new_tags = [] wolffd@0: new_tags = [] wolffd@0: wolffd@0: to_remove = [] wolffd@0: to_add = [] wolffd@0: wolffd@0: tags_on_server = self.get_tags() wolffd@0: wolffd@0: for tag in tags_on_server: wolffd@0: c_old_tags.append(tag.get_name().lower()) wolffd@0: old_tags.append(tag.get_name()) wolffd@0: wolffd@0: for tag in tags: wolffd@0: c_new_tags.append(tag.lower()) wolffd@0: new_tags.append(tag) wolffd@0: wolffd@0: for i in range(0, len(old_tags)): wolffd@0: if not c_old_tags[i] in c_new_tags: wolffd@0: to_remove.append(old_tags[i]) wolffd@0: wolffd@0: for i in range(0, len(new_tags)): wolffd@0: if not c_new_tags[i] in c_old_tags: wolffd@0: to_add.append(new_tags[i]) wolffd@0: wolffd@0: self.remove_tags(to_remove) wolffd@0: self.add_tags(to_add) wolffd@0: wolffd@0: def get_top_tags(self, limit=None): wolffd@0: """Returns a list of the most frequently used Tags on this object.""" wolffd@0: wolffd@0: doc = self._request(self.ws_prefix + '.getTopTags', True) wolffd@0: wolffd@0: elements = doc.getElementsByTagName('tag') wolffd@0: seq = [] wolffd@0: wolffd@0: for element in elements: wolffd@0: tag_name = _extract(element, 'name') wolffd@0: tagcount = _extract(element, 'count') wolffd@0: wolffd@0: seq.append(TopItem(Tag(tag_name, self.network), tagcount)) wolffd@0: wolffd@0: if limit: wolffd@0: seq = seq[:limit] wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class WSError(Exception): wolffd@0: """Exception related to the Network web service""" wolffd@0: wolffd@0: def __init__(self, network, status, details): wolffd@0: self.status = status wolffd@0: self.details = details wolffd@0: self.network = network wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.details wolffd@0: wolffd@0: def get_id(self): wolffd@0: """Returns the exception ID, from one of the following: wolffd@0: STATUS_INVALID_SERVICE = 2 wolffd@0: STATUS_INVALID_METHOD = 3 wolffd@0: STATUS_AUTH_FAILED = 4 wolffd@0: STATUS_INVALID_FORMAT = 5 wolffd@0: STATUS_INVALID_PARAMS = 6 wolffd@0: STATUS_INVALID_RESOURCE = 7 wolffd@0: STATUS_TOKEN_ERROR = 8 wolffd@0: STATUS_INVALID_SK = 9 wolffd@0: STATUS_INVALID_API_KEY = 10 wolffd@0: STATUS_OFFLINE = 11 wolffd@0: STATUS_SUBSCRIBERS_ONLY = 12 wolffd@0: STATUS_TOKEN_UNAUTHORIZED = 14 wolffd@0: STATUS_TOKEN_EXPIRED = 15 wolffd@0: """ wolffd@0: wolffd@0: return self.status wolffd@0: wolffd@0: class MalformedResponseError(Exception): wolffd@0: """Exception conveying a malformed response from Last.fm.""" wolffd@0: wolffd@0: def __init__(self, network, underlying_error): wolffd@0: self.network = network wolffd@0: self.underlying_error = underlying_error wolffd@0: wolffd@0: def __str__(self): wolffd@0: return "Malformed response from Last.fm. Underlying error: %s" %str(self.underlying_error) wolffd@0: wolffd@0: class NetworkError(Exception): wolffd@0: """Exception conveying a problem in sending a request to Last.fm""" wolffd@0: wolffd@0: def __init__(self, network, underlying_error): wolffd@0: self.network = network wolffd@0: self.underlying_error = underlying_error wolffd@0: wolffd@0: def __str__(self): wolffd@0: return "NetworkError: %s" %str(self.underlying_error) wolffd@0: wolffd@0: class Album(_BaseObject, _Taggable): wolffd@0: """An album.""" wolffd@0: wolffd@0: title = None wolffd@0: artist = None wolffd@0: wolffd@0: def __init__(self, artist, title, network): wolffd@0: """ wolffd@0: Create an album instance. wolffd@0: # Parameters: wolffd@0: * artist: An artist name or an Artist object. wolffd@0: * title: The album title. wolffd@0: """ wolffd@0: wolffd@0: _BaseObject.__init__(self, network) wolffd@0: _Taggable.__init__(self, 'album') wolffd@0: wolffd@0: if isinstance(artist, Artist): wolffd@0: self.artist = artist wolffd@0: else: wolffd@0: self.artist = Artist(artist, self.network) wolffd@0: wolffd@0: self.title = title wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Album(%s, %s, %s)" %(repr(self.artist.name), repr(self.title), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return _unicode("%s - %s") %(self.get_artist().get_name(), self.get_title()) wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return (self.get_title().lower() == other.get_title().lower()) and (self.get_artist().get_name().lower() == other.get_artist().get_name().lower()) wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return (self.get_title().lower() != other.get_title().lower()) or (self.get_artist().get_name().lower() != other.get_artist().get_name().lower()) wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'artist': self.get_artist().get_name(), 'album': self.get_title(), } wolffd@0: wolffd@0: def get_artist(self): wolffd@0: """Returns the associated Artist object.""" wolffd@0: wolffd@0: return self.artist wolffd@0: wolffd@0: def get_title(self): wolffd@0: """Returns the album title.""" wolffd@0: wolffd@0: return self.title wolffd@0: wolffd@0: def get_name(self): wolffd@0: """Returns the album title (alias to Album.get_title).""" wolffd@0: wolffd@0: return self.get_title() wolffd@0: wolffd@0: def get_release_date(self): wolffd@0: """Retruns the release date of the album.""" wolffd@0: wolffd@0: return _extract(self._request("album.getInfo", cacheable = True), "releasedate") wolffd@0: wolffd@0: def get_cover_image(self, size = COVER_EXTRA_LARGE): wolffd@0: """ wolffd@0: Returns a uri to the cover image wolffd@0: size can be one of: wolffd@0: COVER_EXTRA_LARGE wolffd@0: COVER_LARGE wolffd@0: COVER_MEDIUM wolffd@0: COVER_SMALL wolffd@0: """ wolffd@0: wolffd@0: return _extract_all(self._request("album.getInfo", cacheable = True), 'image')[size] wolffd@0: wolffd@0: def get_id(self): wolffd@0: """Returns the ID""" wolffd@0: wolffd@0: return _extract(self._request("album.getInfo", cacheable = True), "id") wolffd@0: wolffd@0: def get_playcount(self): wolffd@0: """Returns the number of plays on the network""" wolffd@0: wolffd@0: return _number(_extract(self._request("album.getInfo", cacheable = True), "playcount")) wolffd@0: wolffd@0: def get_listener_count(self): wolffd@0: """Returns the number of liteners on the network""" wolffd@0: wolffd@0: return _number(_extract(self._request("album.getInfo", cacheable = True), "listeners")) wolffd@0: wolffd@0: def get_top_tags(self, limit=None): wolffd@0: """Returns a list of the most-applied tags to this album.""" wolffd@0: wolffd@0: doc = self._request("album.getInfo", True) wolffd@0: e = doc.getElementsByTagName("toptags")[0] wolffd@0: wolffd@0: seq = [] wolffd@0: for name in _extract_all(e, "name"): wolffd@0: seq.append(Tag(name, self.network)) wolffd@0: wolffd@0: if limit: wolffd@0: seq = seq[:limit] wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_tracks(self): wolffd@0: """Returns the list of Tracks on this album.""" wolffd@0: wolffd@0: uri = 'lastfm://playlist/album/%s' %self.get_id() wolffd@0: wolffd@0: return XSPF(uri, self.network).get_tracks() wolffd@0: wolffd@0: def get_mbid(self): wolffd@0: """Returns the MusicBrainz id of the album.""" wolffd@0: wolffd@0: return _extract(self._request("album.getInfo", cacheable = True), "mbid") wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the album page on the network. wolffd@0: # Parameters: wolffd@0: * domain_name str: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: artist = _url_safe(self.get_artist().get_name()) wolffd@0: album = _url_safe(self.get_title()) wolffd@0: wolffd@0: return self.network._get_url(domain_name, "album") %{'artist': artist, 'album': album} wolffd@0: wolffd@0: def get_wiki_published_date(self): wolffd@0: """Returns the date of publishing this version of the wiki.""" wolffd@0: wolffd@0: doc = self._request("album.getInfo", True) wolffd@0: wolffd@0: if len(doc.getElementsByTagName("wiki")) == 0: wolffd@0: return wolffd@0: wolffd@0: node = doc.getElementsByTagName("wiki")[0] wolffd@0: wolffd@0: return _extract(node, "published") wolffd@0: wolffd@0: def get_wiki_summary(self): wolffd@0: """Returns the summary of the wiki.""" wolffd@0: wolffd@0: doc = self._request("album.getInfo", True) wolffd@0: wolffd@0: if len(doc.getElementsByTagName("wiki")) == 0: wolffd@0: return wolffd@0: wolffd@0: node = doc.getElementsByTagName("wiki")[0] wolffd@0: wolffd@0: return _extract(node, "summary") wolffd@0: wolffd@0: def get_wiki_content(self): wolffd@0: """Returns the content of the wiki.""" wolffd@0: wolffd@0: doc = self._request("album.getInfo", True) wolffd@0: wolffd@0: if len(doc.getElementsByTagName("wiki")) == 0: wolffd@0: return wolffd@0: wolffd@0: node = doc.getElementsByTagName("wiki")[0] wolffd@0: wolffd@0: return _extract(node, "content") wolffd@0: wolffd@0: class Artist(_BaseObject, _Taggable): wolffd@0: """An artist.""" wolffd@0: wolffd@0: name = None wolffd@0: wolffd@0: def __init__(self, name, network): wolffd@0: """Create an artist object. wolffd@0: # Parameters: wolffd@0: * name str: The artist's name. wolffd@0: """ wolffd@0: wolffd@0: _BaseObject.__init__(self, network) wolffd@0: _Taggable.__init__(self, 'artist') wolffd@0: wolffd@0: self.name = name wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Artist(%s, %s)" %(repr(self.get_name()), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.get_name() wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return self.get_name().lower() == other.get_name().lower() wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return self.get_name().lower() != other.get_name().lower() wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'artist': self.get_name()} wolffd@0: wolffd@0: def get_name(self, properly_capitalized=False): wolffd@0: """Returns the name of the artist. wolffd@0: If properly_capitalized was asserted then the name would be downloaded wolffd@0: overwriting the given one.""" wolffd@0: wolffd@0: if properly_capitalized: wolffd@0: self.name = _extract(self._request("artist.getInfo", True), "name") wolffd@0: wolffd@0: return self.name wolffd@0: wolffd@0: def get_cover_image(self, size = COVER_MEGA): wolffd@0: """ wolffd@0: Returns a uri to the cover image wolffd@0: size can be one of: wolffd@0: COVER_MEGA wolffd@0: COVER_EXTRA_LARGE wolffd@0: COVER_LARGE wolffd@0: COVER_MEDIUM wolffd@0: COVER_SMALL wolffd@0: """ wolffd@0: wolffd@0: return _extract_all(self._request("artist.getInfo", True), "image")[size] wolffd@0: wolffd@0: def get_playcount(self): wolffd@0: """Returns the number of plays on the network.""" wolffd@0: wolffd@0: return _number(_extract(self._request("artist.getInfo", True), "playcount")) wolffd@0: wolffd@0: def get_mbid(self): wolffd@0: """Returns the MusicBrainz ID of this artist.""" wolffd@0: wolffd@0: doc = self._request("artist.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "mbid") wolffd@0: wolffd@0: def get_listener_count(self): wolffd@0: """Returns the number of liteners on the network.""" wolffd@0: wolffd@0: if hasattr(self, "listener_count"): wolffd@0: return self.listener_count wolffd@0: else: wolffd@0: self.listener_count = _number(_extract(self._request("artist.getInfo", True), "listeners")) wolffd@0: return self.listener_count wolffd@0: wolffd@0: def is_streamable(self): wolffd@0: """Returns True if the artist is streamable.""" wolffd@0: wolffd@0: return bool(_number(_extract(self._request("artist.getInfo", True), "streamable"))) wolffd@0: wolffd@0: def get_bio_published_date(self): wolffd@0: """Returns the date on which the artist's biography was published.""" wolffd@0: wolffd@0: return _extract(self._request("artist.getInfo", True), "published") wolffd@0: wolffd@0: def get_bio_summary(self): wolffd@0: """Returns the summary of the artist's biography.""" wolffd@0: wolffd@0: return _extract(self._request("artist.getInfo", True), "summary") wolffd@0: wolffd@0: def get_bio_content(self): wolffd@0: """Returns the content of the artist's biography.""" wolffd@0: wolffd@0: return _extract(self._request("artist.getInfo", True), "content") wolffd@0: wolffd@0: def get_upcoming_events(self): wolffd@0: """Returns a list of the upcoming Events for this artist.""" wolffd@0: wolffd@0: doc = self._request('artist.getEvents', True) wolffd@0: wolffd@0: ids = _extract_all(doc, 'id') wolffd@0: wolffd@0: events = [] wolffd@0: for e_id in ids: wolffd@0: events.append(Event(e_id, self.network)) wolffd@0: wolffd@0: return events wolffd@0: wolffd@0: def get_similar(self, limit = None): wolffd@0: """Returns the similar artists on the network.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if limit: wolffd@0: params['limit'] = limit wolffd@0: wolffd@0: doc = self._request('artist.getSimilar', True, params) wolffd@0: wolffd@0: names = _extract_all(doc, "name") wolffd@0: matches = _extract_all(doc, "match") wolffd@0: wolffd@0: artists = [] wolffd@0: for i in range(0, len(names)): wolffd@0: artists.append(SimilarItem(Artist(names[i], self.network), _number(matches[i]))) wolffd@0: wolffd@0: return artists wolffd@0: wolffd@0: def compare_with_artist(self, artist, shared_artists_limit = None): wolffd@0: """Compare this artist with another Last.fm artist. wolffd@0: Seems like the tasteometer only returns stats on subsets of artist sets... wolffd@0: Returns a sequence (tasteometer_score, (shared_artist1, shared_artist2, ...)) wolffd@0: artist: A artist object or a artistname string/unicode object. wolffd@0: """ wolffd@0: wolffd@0: if isinstance(artist, Artist): wolffd@0: artist = artist.get_name() wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if shared_artists_limit: wolffd@0: params['limit'] = shared_artists_limit wolffd@0: params['type1'] = 'artists' wolffd@0: params['type2'] = 'artists' wolffd@0: params['value1'] = self.get_name() wolffd@0: params['value2'] = artist wolffd@0: wolffd@0: doc = self._request('tasteometer.compare', False, params) wolffd@0: wolffd@0: score = _extract(doc, 'score') wolffd@0: wolffd@0: artists = doc.getElementsByTagName('artists')[0] wolffd@0: shared_artists_names = _extract_all(artists, 'name') wolffd@0: wolffd@0: shared_artists_seq = [] wolffd@0: wolffd@0: for name in shared_artists_names: wolffd@0: shared_artists_seq.append(Artist(name, self.network)) wolffd@0: wolffd@0: return (score, shared_artists_seq) wolffd@0: wolffd@0: def get_top_albums(self): wolffd@0: """Retuns a list of the top albums.""" wolffd@0: wolffd@0: doc = self._request('artist.getTopAlbums', True) wolffd@0: wolffd@0: seq = [] wolffd@0: wolffd@0: for node in doc.getElementsByTagName("album"): wolffd@0: name = _extract(node, "name") wolffd@0: artist = _extract(node, "name", 1) wolffd@0: playcount = _extract(node, "playcount") wolffd@0: wolffd@0: seq.append(TopItem(Album(artist, name, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_tracks(self): wolffd@0: """Returns a list of the most played Tracks by this artist.""" wolffd@0: wolffd@0: doc = self._request("artist.getTopTracks", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for track in doc.getElementsByTagName('track'): wolffd@0: wolffd@0: title = _extract(track, "name") wolffd@0: artist = _extract(track, "name", 1) wolffd@0: playcount = _number(_extract(track, "playcount")) wolffd@0: wolffd@0: seq.append( TopItem(Track(artist, title, self.network), playcount) ) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_fans(self, limit = None): wolffd@0: """Returns a list of the Users who played this artist the most. wolffd@0: # Parameters: wolffd@0: * limit int: Max elements. wolffd@0: """ wolffd@0: wolffd@0: doc = self._request('artist.getTopFans', True) wolffd@0: wolffd@0: seq = [] wolffd@0: wolffd@0: elements = doc.getElementsByTagName('user') wolffd@0: wolffd@0: for element in elements: wolffd@0: if limit and len(seq) >= limit: wolffd@0: break wolffd@0: wolffd@0: name = _extract(element, 'name') wolffd@0: weight = _number(_extract(element, 'weight')) wolffd@0: wolffd@0: seq.append(TopItem(User(name, self.network), weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def share(self, users, message = None): wolffd@0: """Shares this artist (sends out recommendations). wolffd@0: # Parameters: wolffd@0: * users [User|str,]: A list that can contain usernames, emails, User objects, or all of them. wolffd@0: * message str: A message to include in the recommendation message. wolffd@0: """ wolffd@0: wolffd@0: #last.fm currently accepts a max of 10 recipient at a time wolffd@0: while(len(users) > 10): wolffd@0: section = users[0:9] wolffd@0: users = users[9:] wolffd@0: self.share(section, message) wolffd@0: wolffd@0: nusers = [] wolffd@0: for user in users: wolffd@0: if isinstance(user, User): wolffd@0: nusers.append(user.get_name()) wolffd@0: else: wolffd@0: nusers.append(user) wolffd@0: wolffd@0: params = self._get_params() wolffd@0: recipients = ','.join(nusers) wolffd@0: params['recipient'] = recipients wolffd@0: if message: wolffd@0: params['message'] = message wolffd@0: wolffd@0: self._request('artist.share', False, params) wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the artist page on the network. wolffd@0: # Parameters: wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: artist = _url_safe(self.get_name()) wolffd@0: wolffd@0: return self.network._get_url(domain_name, "artist") %{'artist': artist} wolffd@0: wolffd@0: def get_images(self, order=IMAGES_ORDER_POPULARITY, limit=None): wolffd@0: """ wolffd@0: Returns a sequence of Image objects wolffd@0: if limit is None it will return all wolffd@0: order can be IMAGES_ORDER_POPULARITY or IMAGES_ORDER_DATE. wolffd@0: wolffd@0: If limit==None, it will try to pull all the available data. wolffd@0: """ wolffd@0: wolffd@0: images = [] wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["order"] = order wolffd@0: nodes = _collect_nodes(limit, self, "artist.getImages", True, params) wolffd@0: for e in nodes: wolffd@0: if _extract(e, "name"): wolffd@0: user = User(_extract(e, "name"), self.network) wolffd@0: else: wolffd@0: user = None wolffd@0: wolffd@0: images.append(Image( wolffd@0: _extract(e, "title"), wolffd@0: _extract(e, "url"), wolffd@0: _extract(e, "dateadded"), wolffd@0: _extract(e, "format"), wolffd@0: user, wolffd@0: ImageSizes(*_extract_all(e, "size")), wolffd@0: (_extract(e, "thumbsup"), _extract(e, "thumbsdown")) wolffd@0: ) wolffd@0: ) wolffd@0: return images wolffd@0: wolffd@0: def get_shouts(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequqence of Shout objects wolffd@0: """ wolffd@0: wolffd@0: shouts = [] wolffd@0: for node in _collect_nodes(limit, self, "artist.getShouts", False): wolffd@0: shouts.append(Shout( wolffd@0: _extract(node, "body"), wolffd@0: User(_extract(node, "author"), self.network), wolffd@0: _extract(node, "date") wolffd@0: ) wolffd@0: ) wolffd@0: return shouts wolffd@0: wolffd@0: def shout(self, message): wolffd@0: """ wolffd@0: Post a shout wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["message"] = message wolffd@0: wolffd@0: self._request("artist.Shout", False, params) wolffd@0: wolffd@0: wolffd@0: class Event(_BaseObject): wolffd@0: """An event.""" wolffd@0: wolffd@0: id = None wolffd@0: wolffd@0: def __init__(self, event_id, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self.id = event_id wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Event(%s, %s)" %(repr(self.id), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return "Event #" + self.get_id() wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return self.get_id() == other.get_id() wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return self.get_id() != other.get_id() wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'event': self.get_id()} wolffd@0: wolffd@0: def attend(self, attending_status): wolffd@0: """Sets the attending status. wolffd@0: * attending_status: The attending status. Possible values: wolffd@0: o EVENT_ATTENDING wolffd@0: o EVENT_MAYBE_ATTENDING wolffd@0: o EVENT_NOT_ATTENDING wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['status'] = attending_status wolffd@0: wolffd@0: self._request('event.attend', False, params) wolffd@0: wolffd@0: def get_attendees(self): wolffd@0: """ wolffd@0: Get a list of attendees for an event wolffd@0: """ wolffd@0: wolffd@0: doc = self._request("event.getAttendees", False) wolffd@0: wolffd@0: users = [] wolffd@0: for name in _extract_all(doc, "name"): wolffd@0: users.append(User(name, self.network)) wolffd@0: wolffd@0: return users wolffd@0: wolffd@0: def get_id(self): wolffd@0: """Returns the id of the event on the network. """ wolffd@0: wolffd@0: return self.id wolffd@0: wolffd@0: def get_title(self): wolffd@0: """Returns the title of the event. """ wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "title") wolffd@0: wolffd@0: def get_headliner(self): wolffd@0: """Returns the headliner of the event. """ wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: return Artist(_extract(doc, "headliner"), self.network) wolffd@0: wolffd@0: def get_artists(self): wolffd@0: """Returns a list of the participating Artists. """ wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: names = _extract_all(doc, "artist") wolffd@0: wolffd@0: artists = [] wolffd@0: for name in names: wolffd@0: artists.append(Artist(name, self.network)) wolffd@0: wolffd@0: return artists wolffd@0: wolffd@0: def get_venue(self): wolffd@0: """Returns the venue where the event is held.""" wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: v = doc.getElementsByTagName("venue")[0] wolffd@0: venue_id = _number(_extract(v, "id")) wolffd@0: wolffd@0: return Venue(venue_id, self.network) wolffd@0: wolffd@0: def get_start_date(self): wolffd@0: """Returns the date when the event starts.""" wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "startDate") wolffd@0: wolffd@0: def get_description(self): wolffd@0: """Returns the description of the event. """ wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "description") wolffd@0: wolffd@0: def get_cover_image(self, size = COVER_MEGA): wolffd@0: """ wolffd@0: Returns a uri to the cover image wolffd@0: size can be one of: wolffd@0: COVER_MEGA wolffd@0: COVER_EXTRA_LARGE wolffd@0: COVER_LARGE wolffd@0: COVER_MEDIUM wolffd@0: COVER_SMALL wolffd@0: """ wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: return _extract_all(doc, "image")[size] wolffd@0: wolffd@0: def get_attendance_count(self): wolffd@0: """Returns the number of attending people. """ wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: return _number(_extract(doc, "attendance")) wolffd@0: wolffd@0: def get_review_count(self): wolffd@0: """Returns the number of available reviews for this event. """ wolffd@0: wolffd@0: doc = self._request("event.getInfo", True) wolffd@0: wolffd@0: return _number(_extract(doc, "reviews")) wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the event page on the network. wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: return self.network._get_url(domain_name, "event") %{'id': self.get_id()} wolffd@0: wolffd@0: def share(self, users, message = None): wolffd@0: """Shares this event (sends out recommendations). wolffd@0: * users: A list that can contain usernames, emails, User objects, or all of them. wolffd@0: * message: A message to include in the recommendation message. wolffd@0: """ wolffd@0: wolffd@0: #last.fm currently accepts a max of 10 recipient at a time wolffd@0: while(len(users) > 10): wolffd@0: section = users[0:9] wolffd@0: users = users[9:] wolffd@0: self.share(section, message) wolffd@0: wolffd@0: nusers = [] wolffd@0: for user in users: wolffd@0: if isinstance(user, User): wolffd@0: nusers.append(user.get_name()) wolffd@0: else: wolffd@0: nusers.append(user) wolffd@0: wolffd@0: params = self._get_params() wolffd@0: recipients = ','.join(nusers) wolffd@0: params['recipient'] = recipients wolffd@0: if message: wolffd@0: params['message'] = message wolffd@0: wolffd@0: self._request('event.share', False, params) wolffd@0: wolffd@0: def get_shouts(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequqence of Shout objects wolffd@0: """ wolffd@0: wolffd@0: shouts = [] wolffd@0: for node in _collect_nodes(limit, self, "event.getShouts", False): wolffd@0: shouts.append(Shout( wolffd@0: _extract(node, "body"), wolffd@0: User(_extract(node, "author"), self.network), wolffd@0: _extract(node, "date") wolffd@0: ) wolffd@0: ) wolffd@0: return shouts wolffd@0: wolffd@0: def shout(self, message): wolffd@0: """ wolffd@0: Post a shout wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["message"] = message wolffd@0: wolffd@0: self._request("event.Shout", False, params) wolffd@0: wolffd@0: class Country(_BaseObject): wolffd@0: """A country at Last.fm.""" wolffd@0: wolffd@0: name = None wolffd@0: wolffd@0: def __init__(self, name, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self.name = name wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Country(%s, %s)" %(repr(self.name), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.get_name() wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return self.get_name().lower() == other.get_name().lower() wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return self.get_name() != other.get_name() wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'country': self.get_name()} wolffd@0: wolffd@0: def _get_name_from_code(self, alpha2code): wolffd@0: # TODO: Have this function lookup the alpha-2 code and return the country name. wolffd@0: wolffd@0: return alpha2code wolffd@0: wolffd@0: def get_name(self): wolffd@0: """Returns the country name. """ wolffd@0: wolffd@0: return self.name wolffd@0: wolffd@0: def get_top_artists(self): wolffd@0: """Returns a sequence of the most played artists.""" wolffd@0: wolffd@0: doc = self._request('geo.getTopArtists', True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("artist"): wolffd@0: name = _extract(node, 'name') wolffd@0: playcount = _extract(node, "playcount") wolffd@0: wolffd@0: seq.append(TopItem(Artist(name, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_tracks(self): wolffd@0: """Returns a sequence of the most played tracks""" wolffd@0: wolffd@0: doc = self._request("geo.getTopTracks", True) wolffd@0: wolffd@0: seq = [] wolffd@0: wolffd@0: for n in doc.getElementsByTagName('track'): wolffd@0: wolffd@0: title = _extract(n, 'name') wolffd@0: artist = _extract(n, 'name', 1) wolffd@0: playcount = _number(_extract(n, "playcount")) wolffd@0: wolffd@0: seq.append( TopItem(Track(artist, title, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the event page on the network. wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: country_name = _url_safe(self.get_name()) wolffd@0: wolffd@0: return self.network._get_url(domain_name, "country") %{'country_name': country_name} wolffd@0: wolffd@0: wolffd@0: class Library(_BaseObject): wolffd@0: """A user's Last.fm library.""" wolffd@0: wolffd@0: user = None wolffd@0: wolffd@0: def __init__(self, user, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: if isinstance(user, User): wolffd@0: self.user = user wolffd@0: else: wolffd@0: self.user = User(user, self.network) wolffd@0: wolffd@0: self._albums_index = 0 wolffd@0: self._artists_index = 0 wolffd@0: self._tracks_index = 0 wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Library(%s, %s)" %(repr(self.user), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return repr(self.get_user()) + "'s Library" wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'user': self.user.get_name()} wolffd@0: wolffd@0: def get_user(self): wolffd@0: """Returns the user who owns this library.""" wolffd@0: wolffd@0: return self.user wolffd@0: wolffd@0: def add_album(self, album): wolffd@0: """Add an album to this library.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["artist"] = album.get_artist.get_name() wolffd@0: params["album"] = album.get_name() wolffd@0: wolffd@0: self._request("library.addAlbum", False, params) wolffd@0: wolffd@0: def add_artist(self, artist): wolffd@0: """Add an artist to this library.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["artist"] = artist.get_name() wolffd@0: wolffd@0: self._request("library.addArtist", False, params) wolffd@0: wolffd@0: def add_track(self, track): wolffd@0: """Add a track to this library.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["track"] = track.get_title() wolffd@0: wolffd@0: self._request("library.addTrack", False, params) wolffd@0: wolffd@0: def get_albums(self, artist=None, limit=50): wolffd@0: """ wolffd@0: Returns a sequence of Album objects wolffd@0: If no artist is specified, it will return all, sorted by playcount descendingly. wolffd@0: If limit==None it will return all (may take a while) wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if artist: wolffd@0: params["artist"] = artist wolffd@0: wolffd@0: seq = [] wolffd@0: for node in _collect_nodes(limit, self, "library.getAlbums", True, params): wolffd@0: name = _extract(node, "name") wolffd@0: artist = _extract(node, "name", 1) wolffd@0: playcount = _number(_extract(node, "playcount")) wolffd@0: tagcount = _number(_extract(node, "tagcount")) wolffd@0: wolffd@0: seq.append(LibraryItem(Album(artist, name, self.network), playcount, tagcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_artists(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequence of Album objects wolffd@0: if limit==None it will return all (may take a while) wolffd@0: """ wolffd@0: wolffd@0: seq = [] wolffd@0: for node in _collect_nodes(limit, self, "library.getArtists", True): wolffd@0: name = _extract(node, "name") wolffd@0: wolffd@0: playcount = _number(_extract(node, "playcount")) wolffd@0: tagcount = _number(_extract(node, "tagcount")) wolffd@0: wolffd@0: seq.append(LibraryItem(Artist(name, self.network), playcount, tagcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_tracks(self, artist=None, album=None, limit=50): wolffd@0: """ wolffd@0: Returns a sequence of Album objects wolffd@0: If limit==None it will return all (may take a while) wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if artist: wolffd@0: params["artist"] = artist wolffd@0: if album: wolffd@0: params["album"] = album wolffd@0: wolffd@0: seq = [] wolffd@0: for node in _collect_nodes(limit, self, "library.getTracks", True, params): wolffd@0: name = _extract(node, "name") wolffd@0: artist = _extract(node, "name", 1) wolffd@0: playcount = _number(_extract(node, "playcount")) wolffd@0: tagcount = _number(_extract(node, "tagcount")) wolffd@0: wolffd@0: seq.append(LibraryItem(Track(artist, name, self.network), playcount, tagcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: wolffd@0: class Playlist(_BaseObject): wolffd@0: """A Last.fm user playlist.""" wolffd@0: wolffd@0: id = None wolffd@0: user = None wolffd@0: wolffd@0: def __init__(self, user, id, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: if isinstance(user, User): wolffd@0: self.user = user wolffd@0: else: wolffd@0: self.user = User(user, self.network) wolffd@0: wolffd@0: self.id = id wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return repr(self.user) + "'s playlist # " + repr(self.id) wolffd@0: wolffd@0: def _get_info_node(self): wolffd@0: """Returns the node from user.getPlaylists where this playlist's info is.""" wolffd@0: wolffd@0: doc = self._request("user.getPlaylists", True) wolffd@0: wolffd@0: for node in doc.getElementsByTagName("playlist"): wolffd@0: if _extract(node, "id") == str(self.get_id()): wolffd@0: return node wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'user': self.user.get_name(), 'playlistID': self.get_id()} wolffd@0: wolffd@0: def get_id(self): wolffd@0: """Returns the playlist id.""" wolffd@0: wolffd@0: return self.id wolffd@0: wolffd@0: def get_user(self): wolffd@0: """Returns the owner user of this playlist.""" wolffd@0: wolffd@0: return self.user wolffd@0: wolffd@0: def get_tracks(self): wolffd@0: """Returns a list of the tracks on this user playlist.""" wolffd@0: wolffd@0: uri = _unicode('lastfm://playlist/%s') %self.get_id() wolffd@0: wolffd@0: return XSPF(uri, self.network).get_tracks() wolffd@0: wolffd@0: def add_track(self, track): wolffd@0: """Adds a Track to this Playlist.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['artist'] = track.get_artist().get_name() wolffd@0: params['track'] = track.get_title() wolffd@0: wolffd@0: self._request('playlist.addTrack', False, params) wolffd@0: wolffd@0: def get_title(self): wolffd@0: """Returns the title of this playlist.""" wolffd@0: wolffd@0: return _extract(self._get_info_node(), "title") wolffd@0: wolffd@0: def get_creation_date(self): wolffd@0: """Returns the creation date of this playlist.""" wolffd@0: wolffd@0: return _extract(self._get_info_node(), "date") wolffd@0: wolffd@0: def get_size(self): wolffd@0: """Returns the number of tracks in this playlist.""" wolffd@0: wolffd@0: return _number(_extract(self._get_info_node(), "size")) wolffd@0: wolffd@0: def get_description(self): wolffd@0: """Returns the description of this playlist.""" wolffd@0: wolffd@0: return _extract(self._get_info_node(), "description") wolffd@0: wolffd@0: def get_duration(self): wolffd@0: """Returns the duration of this playlist in milliseconds.""" wolffd@0: wolffd@0: return _number(_extract(self._get_info_node(), "duration")) wolffd@0: wolffd@0: def is_streamable(self): wolffd@0: """Returns True if the playlist is streamable. wolffd@0: For a playlist to be streamable, it needs at least 45 tracks by 15 different artists.""" wolffd@0: wolffd@0: if _extract(self._get_info_node(), "streamable") == '1': wolffd@0: return True wolffd@0: else: wolffd@0: return False wolffd@0: wolffd@0: def has_track(self, track): wolffd@0: """Checks to see if track is already in the playlist. wolffd@0: * track: Any Track object. wolffd@0: """ wolffd@0: wolffd@0: return track in self.get_tracks() wolffd@0: wolffd@0: def get_cover_image(self, size = COVER_EXTRA_LARGE): wolffd@0: """ wolffd@0: Returns a uri to the cover image wolffd@0: size can be one of: wolffd@0: COVER_MEGA wolffd@0: COVER_EXTRA_LARGE wolffd@0: COVER_LARGE wolffd@0: COVER_MEDIUM wolffd@0: COVER_SMALL wolffd@0: """ wolffd@0: wolffd@0: return _extract(self._get_info_node(), "image")[size] wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the playlist on the network. wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: english_url = _extract(self._get_info_node(), "url") wolffd@0: appendix = english_url[english_url.rfind("/") + 1:] wolffd@0: wolffd@0: return self.network._get_url(domain_name, "playlist") %{'appendix': appendix, "user": self.get_user().get_name()} wolffd@0: wolffd@0: wolffd@0: class Tag(_BaseObject): wolffd@0: """A Last.fm object tag.""" wolffd@0: wolffd@0: # TODO: getWeeklyArtistChart (too lazy, i'll wait for when someone requests it) wolffd@0: wolffd@0: name = None wolffd@0: wolffd@0: def __init__(self, name, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self.name = name wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Tag(%s, %s)" %(repr(self.name), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.get_name() wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return self.get_name().lower() == other.get_name().lower() wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return self.get_name().lower() != other.get_name().lower() wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'tag': self.get_name()} wolffd@0: wolffd@0: def get_name(self, properly_capitalized=False): wolffd@0: """Returns the name of the tag. """ wolffd@0: wolffd@0: if properly_capitalized: wolffd@0: self.name = _extract(self._request("tag.getInfo", True), "name") wolffd@0: wolffd@0: return self.name wolffd@0: wolffd@0: def get_similar(self): wolffd@0: """Returns the tags similar to this one, ordered by similarity. """ wolffd@0: wolffd@0: doc = self._request('tag.getSimilar', True) wolffd@0: wolffd@0: seq = [] wolffd@0: names = _extract_all(doc, 'name') wolffd@0: for name in names: wolffd@0: seq.append(Tag(name, self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_albums(self): wolffd@0: """Retuns a list of the top albums.""" wolffd@0: wolffd@0: doc = self._request('tag.getTopAlbums', True) wolffd@0: wolffd@0: seq = [] wolffd@0: wolffd@0: for node in doc.getElementsByTagName("album"): wolffd@0: name = _extract(node, "name") wolffd@0: artist = _extract(node, "name", 1) wolffd@0: playcount = _extract(node, "playcount") wolffd@0: wolffd@0: seq.append(TopItem(Album(artist, name, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_tracks(self): wolffd@0: """Returns a list of the most played Tracks by this artist.""" wolffd@0: wolffd@0: doc = self._request("tag.getTopTracks", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for track in doc.getElementsByTagName('track'): wolffd@0: wolffd@0: title = _extract(track, "name") wolffd@0: artist = _extract(track, "name", 1) wolffd@0: playcount = _number(_extract(track, "playcount")) wolffd@0: wolffd@0: seq.append( TopItem(Track(artist, title, self.network), playcount) ) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_artists(self): wolffd@0: """Returns a sequence of the most played artists.""" wolffd@0: wolffd@0: doc = self._request('tag.getTopArtists', True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("artist"): wolffd@0: name = _extract(node, 'name') wolffd@0: playcount = _extract(node, "playcount") wolffd@0: wolffd@0: seq.append(TopItem(Artist(name, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_chart_dates(self): wolffd@0: """Returns a list of From and To tuples for the available charts.""" wolffd@0: wolffd@0: doc = self._request("tag.getWeeklyChartList", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("chart"): wolffd@0: seq.append( (node.getAttribute("from"), node.getAttribute("to")) ) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_artist_charts(self, from_date = None, to_date = None): wolffd@0: """Returns the weekly artist charts for the week starting from the from_date value to the to_date value.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if from_date and to_date: wolffd@0: params["from"] = from_date wolffd@0: params["to"] = to_date wolffd@0: wolffd@0: doc = self._request("tag.getWeeklyArtistChart", True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("artist"): wolffd@0: item = Artist(_extract(node, "name"), self.network) wolffd@0: weight = _number(_extract(node, "weight")) wolffd@0: seq.append(TopItem(item, weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the tag page on the network. wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: name = _url_safe(self.get_name()) wolffd@0: wolffd@0: return self.network._get_url(domain_name, "tag") %{'name': name} wolffd@0: wolffd@0: class Track(_BaseObject, _Taggable): wolffd@0: """A Last.fm track.""" wolffd@0: wolffd@0: artist = None wolffd@0: title = None wolffd@0: wolffd@0: def __init__(self, artist, title, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: _Taggable.__init__(self, 'track') wolffd@0: wolffd@0: if isinstance(artist, Artist): wolffd@0: self.artist = artist wolffd@0: else: wolffd@0: self.artist = Artist(artist, self.network) wolffd@0: wolffd@0: self.title = title wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Track(%s, %s, %s)" %(repr(self.artist.name), repr(self.title), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.get_artist().get_name() + ' - ' + self.get_title() wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return (self.get_title().lower() == other.get_title().lower()) and (self.get_artist().get_name().lower() == other.get_artist().get_name().lower()) wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return (self.get_title().lower() != other.get_title().lower()) or (self.get_artist().get_name().lower() != other.get_artist().get_name().lower()) wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'artist': self.get_artist().get_name(), 'track': self.get_title()} wolffd@0: wolffd@0: def get_artist(self): wolffd@0: """Returns the associated Artist object.""" wolffd@0: wolffd@0: return self.artist wolffd@0: wolffd@0: def get_title(self, properly_capitalized=False): wolffd@0: """Returns the track title.""" wolffd@0: wolffd@0: if properly_capitalized: wolffd@0: self.title = _extract(self._request("track.getInfo", True), "name") wolffd@0: wolffd@0: return self.title wolffd@0: wolffd@0: def get_name(self, properly_capitalized=False): wolffd@0: """Returns the track title (alias to Track.get_title).""" wolffd@0: wolffd@0: return self.get_title(properly_capitalized) wolffd@0: wolffd@0: def get_id(self): wolffd@0: """Returns the track id on the network.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "id") wolffd@0: wolffd@0: def get_duration(self): wolffd@0: """Returns the track duration.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: wolffd@0: return _number(_extract(doc, "duration")) wolffd@0: wolffd@0: def get_mbid(self): wolffd@0: """Returns the MusicBrainz ID of this track.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "mbid") wolffd@0: wolffd@0: def get_listener_count(self): wolffd@0: """Returns the listener count.""" wolffd@0: wolffd@0: if hasattr(self, "listener_count"): wolffd@0: return self.listener_count wolffd@0: else: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: self.listener_count = _number(_extract(doc, "listeners")) wolffd@0: return self.listener_count wolffd@0: wolffd@0: def get_playcount(self): wolffd@0: """Returns the play count.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: return _number(_extract(doc, "playcount")) wolffd@0: wolffd@0: def is_streamable(self): wolffd@0: """Returns True if the track is available at Last.fm.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: return _extract(doc, "streamable") == "1" wolffd@0: wolffd@0: def is_fulltrack_available(self): wolffd@0: """Returns True if the fulltrack is available for streaming.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: return doc.getElementsByTagName("streamable")[0].getAttribute("fulltrack") == "1" wolffd@0: wolffd@0: def get_album(self): wolffd@0: """Returns the album object of this track.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: wolffd@0: albums = doc.getElementsByTagName("album") wolffd@0: wolffd@0: if len(albums) == 0: wolffd@0: return wolffd@0: wolffd@0: node = doc.getElementsByTagName("album")[0] wolffd@0: return Album(_extract(node, "artist"), _extract(node, "title"), self.network) wolffd@0: wolffd@0: def get_wiki_published_date(self): wolffd@0: """Returns the date of publishing this version of the wiki.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: wolffd@0: if len(doc.getElementsByTagName("wiki")) == 0: wolffd@0: return wolffd@0: wolffd@0: node = doc.getElementsByTagName("wiki")[0] wolffd@0: wolffd@0: return _extract(node, "published") wolffd@0: wolffd@0: def get_wiki_summary(self): wolffd@0: """Returns the summary of the wiki.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: wolffd@0: if len(doc.getElementsByTagName("wiki")) == 0: wolffd@0: return wolffd@0: wolffd@0: node = doc.getElementsByTagName("wiki")[0] wolffd@0: wolffd@0: return _extract(node, "summary") wolffd@0: wolffd@0: def get_wiki_content(self): wolffd@0: """Returns the content of the wiki.""" wolffd@0: wolffd@0: doc = self._request("track.getInfo", True) wolffd@0: wolffd@0: if len(doc.getElementsByTagName("wiki")) == 0: wolffd@0: return wolffd@0: wolffd@0: node = doc.getElementsByTagName("wiki")[0] wolffd@0: wolffd@0: return _extract(node, "content") wolffd@0: wolffd@0: def love(self): wolffd@0: """Adds the track to the user's loved tracks. """ wolffd@0: wolffd@0: self._request('track.love') wolffd@0: wolffd@0: def ban(self): wolffd@0: """Ban this track from ever playing on the radio. """ wolffd@0: wolffd@0: self._request('track.ban') wolffd@0: wolffd@0: def get_similar(self): wolffd@0: """Returns similar tracks for this track on the network, based on listening data. """ wolffd@0: wolffd@0: doc = self._request('track.getSimilar', True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("track"): wolffd@0: title = _extract(node, 'name') wolffd@0: artist = _extract(node, 'name', 1) wolffd@0: match = _number(_extract(node, "match")) wolffd@0: wolffd@0: seq.append(SimilarItem(Track(artist, title, self.network), match)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_fans(self, limit = None): wolffd@0: """Returns a list of the Users who played this track.""" wolffd@0: wolffd@0: doc = self._request('track.getTopFans', True) wolffd@0: wolffd@0: seq = [] wolffd@0: wolffd@0: elements = doc.getElementsByTagName('user') wolffd@0: wolffd@0: for element in elements: wolffd@0: if limit and len(seq) >= limit: wolffd@0: break wolffd@0: wolffd@0: name = _extract(element, 'name') wolffd@0: weight = _number(_extract(element, 'weight')) wolffd@0: wolffd@0: seq.append(TopItem(User(name, self.network), weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def share(self, users, message = None): wolffd@0: """Shares this track (sends out recommendations). wolffd@0: * users: A list that can contain usernames, emails, User objects, or all of them. wolffd@0: * message: A message to include in the recommendation message. wolffd@0: """ wolffd@0: wolffd@0: #last.fm currently accepts a max of 10 recipient at a time wolffd@0: while(len(users) > 10): wolffd@0: section = users[0:9] wolffd@0: users = users[9:] wolffd@0: self.share(section, message) wolffd@0: wolffd@0: nusers = [] wolffd@0: for user in users: wolffd@0: if isinstance(user, User): wolffd@0: nusers.append(user.get_name()) wolffd@0: else: wolffd@0: nusers.append(user) wolffd@0: wolffd@0: params = self._get_params() wolffd@0: recipients = ','.join(nusers) wolffd@0: params['recipient'] = recipients wolffd@0: if message: wolffd@0: params['message'] = message wolffd@0: wolffd@0: self._request('track.share', False, params) wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the track page on the network. wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: artist = _url_safe(self.get_artist().get_name()) wolffd@0: title = _url_safe(self.get_title()) wolffd@0: wolffd@0: return self.network._get_url(domain_name, "track") %{'domain': self.network._get_language_domain(domain_name), 'artist': artist, 'title': title} wolffd@0: wolffd@0: def get_shouts(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequqence of Shout objects wolffd@0: """ wolffd@0: wolffd@0: shouts = [] wolffd@0: for node in _collect_nodes(limit, self, "track.getShouts", False): wolffd@0: shouts.append(Shout( wolffd@0: _extract(node, "body"), wolffd@0: User(_extract(node, "author"), self.network), wolffd@0: _extract(node, "date") wolffd@0: ) wolffd@0: ) wolffd@0: return shouts wolffd@0: wolffd@0: class Group(_BaseObject): wolffd@0: """A Last.fm group.""" wolffd@0: wolffd@0: name = None wolffd@0: wolffd@0: def __init__(self, group_name, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self.name = group_name wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Group(%s, %s)" %(repr(self.name), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.get_name() wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return self.get_name().lower() == other.get_name().lower() wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return self.get_name() != other.get_name() wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'group': self.get_name()} wolffd@0: wolffd@0: def get_name(self): wolffd@0: """Returns the group name. """ wolffd@0: return self.name wolffd@0: wolffd@0: def get_weekly_chart_dates(self): wolffd@0: """Returns a list of From and To tuples for the available charts.""" wolffd@0: wolffd@0: doc = self._request("group.getWeeklyChartList", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("chart"): wolffd@0: seq.append( (node.getAttribute("from"), node.getAttribute("to")) ) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_artist_charts(self, from_date = None, to_date = None): wolffd@0: """Returns the weekly artist charts for the week starting from the from_date value to the to_date value.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if from_date and to_date: wolffd@0: params["from"] = from_date wolffd@0: params["to"] = to_date wolffd@0: wolffd@0: doc = self._request("group.getWeeklyArtistChart", True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("artist"): wolffd@0: item = Artist(_extract(node, "name"), self.network) wolffd@0: weight = _number(_extract(node, "playcount")) wolffd@0: seq.append(TopItem(item, weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_album_charts(self, from_date = None, to_date = None): wolffd@0: """Returns the weekly album charts for the week starting from the from_date value to the to_date value.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if from_date and to_date: wolffd@0: params["from"] = from_date wolffd@0: params["to"] = to_date wolffd@0: wolffd@0: doc = self._request("group.getWeeklyAlbumChart", True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("album"): wolffd@0: item = Album(_extract(node, "artist"), _extract(node, "name"), self.network) wolffd@0: weight = _number(_extract(node, "playcount")) wolffd@0: seq.append(TopItem(item, weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_track_charts(self, from_date = None, to_date = None): wolffd@0: """Returns the weekly track charts for the week starting from the from_date value to the to_date value.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if from_date and to_date: wolffd@0: params["from"] = from_date wolffd@0: params["to"] = to_date wolffd@0: wolffd@0: doc = self._request("group.getWeeklyTrackChart", True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("track"): wolffd@0: item = Track(_extract(node, "artist"), _extract(node, "name"), self.network) wolffd@0: weight = _number(_extract(node, "playcount")) wolffd@0: seq.append(TopItem(item, weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the group page on the network. wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: name = _url_safe(self.get_name()) wolffd@0: wolffd@0: return self.network._get_url(domain_name, "group") %{'name': name} wolffd@0: wolffd@0: def get_members(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequence of User objects wolffd@0: if limit==None it will return all wolffd@0: """ wolffd@0: wolffd@0: nodes = _collect_nodes(limit, self, "group.getMembers", False) wolffd@0: wolffd@0: users = [] wolffd@0: wolffd@0: for node in nodes: wolffd@0: users.append(User(_extract(node, "name"), self.network)) wolffd@0: wolffd@0: return users wolffd@0: wolffd@0: class XSPF(_BaseObject): wolffd@0: "A Last.fm XSPF playlist.""" wolffd@0: wolffd@0: uri = None wolffd@0: wolffd@0: def __init__(self, uri, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self.uri = uri wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {'playlistURL': self.get_uri()} wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.get_uri() wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return self.get_uri() == other.get_uri() wolffd@0: wolffd@0: def __ne__(self, other): wolffd@0: return self.get_uri() != other.get_uri() wolffd@0: wolffd@0: def get_uri(self): wolffd@0: """Returns the Last.fm playlist URI. """ wolffd@0: wolffd@0: return self.uri wolffd@0: wolffd@0: def get_tracks(self): wolffd@0: """Returns the tracks on this playlist.""" wolffd@0: wolffd@0: doc = self._request('playlist.fetch', True) wolffd@0: wolffd@0: seq = [] wolffd@0: for n in doc.getElementsByTagName('track'): wolffd@0: title = _extract(n, 'title') wolffd@0: artist = _extract(n, 'creator') wolffd@0: wolffd@0: seq.append(Track(artist, title, self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class User(_BaseObject): wolffd@0: """A Last.fm user.""" wolffd@0: wolffd@0: name = None wolffd@0: wolffd@0: def __init__(self, user_name, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self.name = user_name wolffd@0: wolffd@0: self._past_events_index = 0 wolffd@0: self._recommended_events_index = 0 wolffd@0: self._recommended_artists_index = 0 wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.User(%s, %s)" %(repr(self.name), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.get_name() wolffd@0: wolffd@0: def __eq__(self, another): wolffd@0: return self.get_name() == another.get_name() wolffd@0: wolffd@0: def __ne__(self, another): wolffd@0: return self.get_name() != another.get_name() wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {"user": self.get_name()} wolffd@0: wolffd@0: def get_name(self, properly_capitalized=False): wolffd@0: """Returns the nuser name.""" wolffd@0: wolffd@0: if properly_capitalized: wolffd@0: self.name = _extract(self._request("user.getInfo", True), "name") wolffd@0: wolffd@0: return self.name wolffd@0: wolffd@0: def get_upcoming_events(self): wolffd@0: """Returns all the upcoming events for this user. """ wolffd@0: wolffd@0: doc = self._request('user.getEvents', True) wolffd@0: wolffd@0: ids = _extract_all(doc, 'id') wolffd@0: events = [] wolffd@0: wolffd@0: for e_id in ids: wolffd@0: events.append(Event(e_id, self.network)) wolffd@0: wolffd@0: return events wolffd@0: wolffd@0: def get_friends(self, limit = 50): wolffd@0: """Returns a list of the user's friends. """ wolffd@0: wolffd@0: seq = [] wolffd@0: for node in _collect_nodes(limit, self, "user.getFriends", False): wolffd@0: seq.append(User(_extract(node, "name"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_loved_tracks(self, limit=50): wolffd@0: """Returns this user's loved track as a sequence of LovedTrack objects wolffd@0: in reverse order of their timestamp, all the way back to the first track. wolffd@0: wolffd@0: If limit==None, it will try to pull all the available data. wolffd@0: wolffd@0: This method uses caching. Enable caching only if you're pulling a wolffd@0: large amount of data. wolffd@0: wolffd@0: Use extract_items() with the return of this function to wolffd@0: get only a sequence of Track objects with no playback dates. """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if limit: wolffd@0: params['limit'] = limit wolffd@0: wolffd@0: seq = [] wolffd@0: for track in _collect_nodes(limit, self, "user.getLovedTracks", True, params): wolffd@0: wolffd@0: title = _extract(track, "name") wolffd@0: artist = _extract(track, "name", 1) wolffd@0: date = _extract(track, "date") wolffd@0: timestamp = track.getElementsByTagName("date")[0].getAttribute("uts") wolffd@0: wolffd@0: seq.append(LovedTrack(Track(artist, title, self.network), date, timestamp)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_neighbours(self, limit = 50): wolffd@0: """Returns a list of the user's friends.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if limit: wolffd@0: params['limit'] = limit wolffd@0: wolffd@0: doc = self._request('user.getNeighbours', True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: names = _extract_all(doc, 'name') wolffd@0: wolffd@0: for name in names: wolffd@0: seq.append(User(name, self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_past_events(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequence of Event objects wolffd@0: if limit==None it will return all wolffd@0: """ wolffd@0: wolffd@0: seq = [] wolffd@0: for n in _collect_nodes(limit, self, "user.getPastEvents", False): wolffd@0: seq.append(Event(_extract(n, "id"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_playlists(self): wolffd@0: """Returns a list of Playlists that this user owns.""" wolffd@0: wolffd@0: doc = self._request("user.getPlaylists", True) wolffd@0: wolffd@0: playlists = [] wolffd@0: for playlist_id in _extract_all(doc, "id"): wolffd@0: playlists.append(Playlist(self.get_name(), playlist_id, self.network)) wolffd@0: wolffd@0: return playlists wolffd@0: wolffd@0: def get_now_playing(self): wolffd@0: """Returns the currently playing track, or None if nothing is playing. """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['limit'] = '1' wolffd@0: wolffd@0: doc = self._request('user.getRecentTracks', False, params) wolffd@0: wolffd@0: e = doc.getElementsByTagName('track')[0] wolffd@0: wolffd@0: if not e.hasAttribute('nowplaying'): wolffd@0: return None wolffd@0: wolffd@0: artist = _extract(e, 'artist') wolffd@0: title = _extract(e, 'name') wolffd@0: wolffd@0: return Track(artist, title, self.network) wolffd@0: wolffd@0: wolffd@0: def get_recent_tracks(self, limit = 10): wolffd@0: """Returns this user's played track as a sequence of PlayedTrack objects wolffd@0: in reverse order of their playtime, all the way back to the first track. wolffd@0: wolffd@0: If limit==None, it will try to pull all the available data. wolffd@0: wolffd@0: This method uses caching. Enable caching only if you're pulling a wolffd@0: large amount of data. wolffd@0: wolffd@0: Use extract_items() with the return of this function to wolffd@0: get only a sequence of Track objects with no playback dates. """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if limit: wolffd@0: params['limit'] = limit wolffd@0: wolffd@0: seq = [] wolffd@0: for track in _collect_nodes(limit, self, "user.getRecentTracks", True, params): wolffd@0: wolffd@0: if track.hasAttribute('nowplaying'): wolffd@0: continue #to prevent the now playing track from sneaking in here wolffd@0: wolffd@0: title = _extract(track, "name") wolffd@0: artist = _extract(track, "artist") wolffd@0: date = _extract(track, "date") wolffd@0: timestamp = track.getElementsByTagName("date")[0].getAttribute("uts") wolffd@0: wolffd@0: seq.append(PlayedTrack(Track(artist, title, self.network), date, timestamp)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_id(self): wolffd@0: """Returns the user id.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "id") wolffd@0: wolffd@0: def get_language(self): wolffd@0: """Returns the language code of the language used by the user.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "lang") wolffd@0: wolffd@0: def get_country(self): wolffd@0: """Returns the name of the country of the user.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: return Country(_extract(doc, "country"), self.network) wolffd@0: wolffd@0: def get_age(self): wolffd@0: """Returns the user's age.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: return _number(_extract(doc, "age")) wolffd@0: wolffd@0: def get_gender(self): wolffd@0: """Returns the user's gender. Either USER_MALE or USER_FEMALE.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: value = _extract(doc, "gender") wolffd@0: wolffd@0: if value == 'm': wolffd@0: return USER_MALE wolffd@0: elif value == 'f': wolffd@0: return USER_FEMALE wolffd@0: wolffd@0: return None wolffd@0: wolffd@0: def is_subscriber(self): wolffd@0: """Returns whether the user is a subscriber or not. True or False.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "subscriber") == "1" wolffd@0: wolffd@0: def get_playcount(self): wolffd@0: """Returns the user's playcount so far.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: return _number(_extract(doc, "playcount")) wolffd@0: wolffd@0: def get_top_albums(self, period = PERIOD_OVERALL): wolffd@0: """Returns the top albums played by a user. wolffd@0: * period: The period of time. Possible values: wolffd@0: o PERIOD_OVERALL wolffd@0: o PERIOD_7DAYS wolffd@0: o PERIOD_3MONTHS wolffd@0: o PERIOD_6MONTHS wolffd@0: o PERIOD_12MONTHS wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['period'] = period wolffd@0: wolffd@0: doc = self._request('user.getTopAlbums', True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for album in doc.getElementsByTagName('album'): wolffd@0: name = _extract(album, 'name') wolffd@0: artist = _extract(album, 'name', 1) wolffd@0: playcount = _extract(album, "playcount") wolffd@0: wolffd@0: seq.append(TopItem(Album(artist, name, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_artists(self, period = PERIOD_OVERALL): wolffd@0: """Returns the top artists played by a user. wolffd@0: * period: The period of time. Possible values: wolffd@0: o PERIOD_OVERALL wolffd@0: o PERIOD_7DAYS wolffd@0: o PERIOD_3MONTHS wolffd@0: o PERIOD_6MONTHS wolffd@0: o PERIOD_12MONTHS wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['period'] = period wolffd@0: wolffd@0: doc = self._request('user.getTopArtists', True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName('artist'): wolffd@0: name = _extract(node, 'name') wolffd@0: playcount = _extract(node, "playcount") wolffd@0: wolffd@0: seq.append(TopItem(Artist(name, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_tags(self, limit=None): wolffd@0: """Returns a sequence of the top tags used by this user with their counts as TopItem objects. wolffd@0: * limit: The limit of how many tags to return. wolffd@0: """ wolffd@0: wolffd@0: doc = self._request("user.getTopTags", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("tag"): wolffd@0: seq.append(TopItem(Tag(_extract(node, "name"), self.network), _extract(node, "count"))) wolffd@0: wolffd@0: if limit: wolffd@0: seq = seq[:limit] wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_top_tracks(self, period = PERIOD_OVERALL): wolffd@0: """Returns the top tracks played by a user. wolffd@0: * period: The period of time. Possible values: wolffd@0: o PERIOD_OVERALL wolffd@0: o PERIOD_7DAYS wolffd@0: o PERIOD_3MONTHS wolffd@0: o PERIOD_6MONTHS wolffd@0: o PERIOD_12MONTHS wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params['period'] = period wolffd@0: wolffd@0: doc = self._request('user.getTopTracks', True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for track in doc.getElementsByTagName('track'): wolffd@0: name = _extract(track, 'name') wolffd@0: artist = _extract(track, 'name', 1) wolffd@0: playcount = _extract(track, "playcount") wolffd@0: wolffd@0: seq.append(TopItem(Track(artist, name, self.network), playcount)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_chart_dates(self): wolffd@0: """Returns a list of From and To tuples for the available charts.""" wolffd@0: wolffd@0: doc = self._request("user.getWeeklyChartList", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("chart"): wolffd@0: seq.append( (node.getAttribute("from"), node.getAttribute("to")) ) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_artist_charts(self, from_date = None, to_date = None): wolffd@0: """Returns the weekly artist charts for the week starting from the from_date value to the to_date value.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if from_date and to_date: wolffd@0: params["from"] = from_date wolffd@0: params["to"] = to_date wolffd@0: wolffd@0: doc = self._request("user.getWeeklyArtistChart", True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("artist"): wolffd@0: item = Artist(_extract(node, "name"), self.network) wolffd@0: weight = _number(_extract(node, "playcount")) wolffd@0: seq.append(TopItem(item, weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_album_charts(self, from_date = None, to_date = None): wolffd@0: """Returns the weekly album charts for the week starting from the from_date value to the to_date value.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if from_date and to_date: wolffd@0: params["from"] = from_date wolffd@0: params["to"] = to_date wolffd@0: wolffd@0: doc = self._request("user.getWeeklyAlbumChart", True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("album"): wolffd@0: item = Album(_extract(node, "artist"), _extract(node, "name"), self.network) wolffd@0: weight = _number(_extract(node, "playcount")) wolffd@0: seq.append(TopItem(item, weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_weekly_track_charts(self, from_date = None, to_date = None): wolffd@0: """Returns the weekly track charts for the week starting from the from_date value to the to_date value.""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if from_date and to_date: wolffd@0: params["from"] = from_date wolffd@0: params["to"] = to_date wolffd@0: wolffd@0: doc = self._request("user.getWeeklyTrackChart", True, params) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("track"): wolffd@0: item = Track(_extract(node, "artist"), _extract(node, "name"), self.network) wolffd@0: weight = _number(_extract(node, "playcount")) wolffd@0: seq.append(TopItem(item, weight)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def compare_with_user(self, user, shared_artists_limit = None): wolffd@0: """Compare this user with another Last.fm user. wolffd@0: Returns a sequence (tasteometer_score, (shared_artist1, shared_artist2, ...)) wolffd@0: user: A User object or a username string/unicode object. wolffd@0: """ wolffd@0: wolffd@0: if isinstance(user, User): wolffd@0: user = user.get_name() wolffd@0: wolffd@0: params = self._get_params() wolffd@0: if shared_artists_limit: wolffd@0: params['limit'] = shared_artists_limit wolffd@0: params['type1'] = 'user' wolffd@0: params['type2'] = 'user' wolffd@0: params['value1'] = self.get_name() wolffd@0: params['value2'] = user wolffd@0: wolffd@0: doc = self._request('tasteometer.compare', False, params) wolffd@0: wolffd@0: score = _extract(doc, 'score') wolffd@0: wolffd@0: artists = doc.getElementsByTagName('artists')[0] wolffd@0: shared_artists_names = _extract_all(artists, 'name') wolffd@0: wolffd@0: shared_artists_seq = [] wolffd@0: wolffd@0: for name in shared_artists_names: wolffd@0: shared_artists_seq.append(Artist(name, self.network)) wolffd@0: wolffd@0: return (score, shared_artists_seq) wolffd@0: wolffd@0: def get_image(self): wolffd@0: """Returns the user's avatar.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True) wolffd@0: wolffd@0: return _extract(doc, "image") wolffd@0: wolffd@0: def get_url(self, domain_name = DOMAIN_ENGLISH): wolffd@0: """Returns the url of the user page on the network. wolffd@0: * domain_name: The network's language domain. Possible values: wolffd@0: o DOMAIN_ENGLISH wolffd@0: o DOMAIN_GERMAN wolffd@0: o DOMAIN_SPANISH wolffd@0: o DOMAIN_FRENCH wolffd@0: o DOMAIN_ITALIAN wolffd@0: o DOMAIN_POLISH wolffd@0: o DOMAIN_PORTUGUESE wolffd@0: o DOMAIN_SWEDISH wolffd@0: o DOMAIN_TURKISH wolffd@0: o DOMAIN_RUSSIAN wolffd@0: o DOMAIN_JAPANESE wolffd@0: o DOMAIN_CHINESE wolffd@0: """ wolffd@0: wolffd@0: name = _url_safe(self.get_name()) wolffd@0: wolffd@0: return self.network._get_url(domain_name, "user") %{'name': name} wolffd@0: wolffd@0: def get_library(self): wolffd@0: """Returns the associated Library object. """ wolffd@0: wolffd@0: return Library(self, self.network) wolffd@0: wolffd@0: def get_shouts(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequqence of Shout objects wolffd@0: """ wolffd@0: wolffd@0: shouts = [] wolffd@0: for node in _collect_nodes(limit, self, "user.getShouts", False): wolffd@0: shouts.append(Shout( wolffd@0: _extract(node, "body"), wolffd@0: User(_extract(node, "author"), self.network), wolffd@0: _extract(node, "date") wolffd@0: ) wolffd@0: ) wolffd@0: return shouts wolffd@0: wolffd@0: def shout(self, message): wolffd@0: """ wolffd@0: Post a shout wolffd@0: """ wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["message"] = message wolffd@0: wolffd@0: self._request("user.Shout", False, params) wolffd@0: wolffd@0: class AuthenticatedUser(User): wolffd@0: def __init__(self, network): wolffd@0: User.__init__(self, "", network); wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {"user": self.get_name()} wolffd@0: wolffd@0: def get_name(self): wolffd@0: """Returns the name of the authenticated user.""" wolffd@0: wolffd@0: doc = self._request("user.getInfo", True, {"user": ""}) # hack wolffd@0: wolffd@0: self.name = _extract(doc, "name") wolffd@0: return self.name wolffd@0: wolffd@0: def get_recommended_events(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequence of Event objects wolffd@0: if limit==None it will return all wolffd@0: """ wolffd@0: wolffd@0: seq = [] wolffd@0: for node in _collect_nodes(limit, self, "user.getRecommendedEvents", False): wolffd@0: seq.append(Event(_extract(node, "id"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_recommended_artists(self, limit=50): wolffd@0: """ wolffd@0: Returns a sequence of Event objects wolffd@0: if limit==None it will return all wolffd@0: """ wolffd@0: wolffd@0: seq = [] wolffd@0: for node in _collect_nodes(limit, self, "user.getRecommendedArtists", False): wolffd@0: seq.append(Artist(_extract(node, "name"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class _Search(_BaseObject): wolffd@0: """An abstract class. Use one of its derivatives.""" wolffd@0: wolffd@0: def __init__(self, ws_prefix, search_terms, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self._ws_prefix = ws_prefix wolffd@0: self.search_terms = search_terms wolffd@0: wolffd@0: self._last_page_index = 0 wolffd@0: wolffd@0: def _get_params(self): wolffd@0: params = {} wolffd@0: wolffd@0: for key in self.search_terms.keys(): wolffd@0: params[key] = self.search_terms[key] wolffd@0: wolffd@0: return params wolffd@0: wolffd@0: def get_total_result_count(self): wolffd@0: """Returns the total count of all the results.""" wolffd@0: wolffd@0: doc = self._request(self._ws_prefix + ".search", True) wolffd@0: wolffd@0: return _extract(doc, "opensearch:totalResults") wolffd@0: wolffd@0: def _retreive_page(self, page_index): wolffd@0: """Returns the node of matches to be processed""" wolffd@0: wolffd@0: params = self._get_params() wolffd@0: params["page"] = str(page_index) wolffd@0: doc = self._request(self._ws_prefix + ".search", True, params) wolffd@0: wolffd@0: return doc.getElementsByTagName(self._ws_prefix + "matches")[0] wolffd@0: wolffd@0: def _retrieve_next_page(self): wolffd@0: self._last_page_index += 1 wolffd@0: return self._retreive_page(self._last_page_index) wolffd@0: wolffd@0: class AlbumSearch(_Search): wolffd@0: """Search for an album by name.""" wolffd@0: wolffd@0: def __init__(self, album_name, network): wolffd@0: wolffd@0: _Search.__init__(self, "album", {"album": album_name}, network) wolffd@0: wolffd@0: def get_next_page(self): wolffd@0: """Returns the next page of results as a sequence of Album objects.""" wolffd@0: wolffd@0: master_node = self._retrieve_next_page() wolffd@0: wolffd@0: seq = [] wolffd@0: for node in master_node.getElementsByTagName("album"): wolffd@0: seq.append(Album(_extract(node, "artist"), _extract(node, "name"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class ArtistSearch(_Search): wolffd@0: """Search for an artist by artist name.""" wolffd@0: wolffd@0: def __init__(self, artist_name, network): wolffd@0: _Search.__init__(self, "artist", {"artist": artist_name}, network) wolffd@0: wolffd@0: def get_next_page(self): wolffd@0: """Returns the next page of results as a sequence of Artist objects.""" wolffd@0: wolffd@0: master_node = self._retrieve_next_page() wolffd@0: wolffd@0: seq = [] wolffd@0: for node in master_node.getElementsByTagName("artist"): wolffd@0: artist = Artist(_extract(node, "name"), self.network) wolffd@0: artist.listener_count = _number(_extract(node, "listeners")) wolffd@0: seq.append(artist) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class TagSearch(_Search): wolffd@0: """Search for a tag by tag name.""" wolffd@0: wolffd@0: def __init__(self, tag_name, network): wolffd@0: wolffd@0: _Search.__init__(self, "tag", {"tag": tag_name}, network) wolffd@0: wolffd@0: def get_next_page(self): wolffd@0: """Returns the next page of results as a sequence of Tag objects.""" wolffd@0: wolffd@0: master_node = self._retrieve_next_page() wolffd@0: wolffd@0: seq = [] wolffd@0: for node in master_node.getElementsByTagName("tag"): wolffd@0: tag = Tag(_extract(node, "name"), self.network) wolffd@0: tag.tag_count = _number(_extract(node, "count")) wolffd@0: seq.append(tag) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class TrackSearch(_Search): wolffd@0: """Search for a track by track title. If you don't wanna narrow the results down wolffd@0: by specifying the artist name, set it to empty string.""" wolffd@0: wolffd@0: def __init__(self, artist_name, track_title, network): wolffd@0: wolffd@0: _Search.__init__(self, "track", {"track": track_title, "artist": artist_name}, network) wolffd@0: wolffd@0: def get_next_page(self): wolffd@0: """Returns the next page of results as a sequence of Track objects.""" wolffd@0: wolffd@0: master_node = self._retrieve_next_page() wolffd@0: wolffd@0: seq = [] wolffd@0: for node in master_node.getElementsByTagName("track"): wolffd@0: track = Track(_extract(node, "artist"), _extract(node, "name"), self.network) wolffd@0: track.listener_count = _number(_extract(node, "listeners")) wolffd@0: seq.append(track) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class VenueSearch(_Search): wolffd@0: """Search for a venue by its name. If you don't wanna narrow the results down wolffd@0: by specifying a country, set it to empty string.""" wolffd@0: wolffd@0: def __init__(self, venue_name, country_name, network): wolffd@0: wolffd@0: _Search.__init__(self, "venue", {"venue": venue_name, "country": country_name}, network) wolffd@0: wolffd@0: def get_next_page(self): wolffd@0: """Returns the next page of results as a sequence of Track objects.""" wolffd@0: wolffd@0: master_node = self._retrieve_next_page() wolffd@0: wolffd@0: seq = [] wolffd@0: for node in master_node.getElementsByTagName("venue"): wolffd@0: seq.append(Venue(_extract(node, "id"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class Venue(_BaseObject): wolffd@0: """A venue where events are held.""" wolffd@0: wolffd@0: # TODO: waiting for a venue.getInfo web service to use. wolffd@0: wolffd@0: id = None wolffd@0: wolffd@0: def __init__(self, id, network): wolffd@0: _BaseObject.__init__(self, network) wolffd@0: wolffd@0: self.id = _number(id) wolffd@0: wolffd@0: def __repr__(self): wolffd@0: return "pylast.Venue(%s, %s)" %(repr(self.id), repr(self.network)) wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return "Venue #" + str(self.id) wolffd@0: wolffd@0: def __eq__(self, other): wolffd@0: return self.get_id() == other.get_id() wolffd@0: wolffd@0: def _get_params(self): wolffd@0: return {"venue": self.get_id()} wolffd@0: wolffd@0: def get_id(self): wolffd@0: """Returns the id of the venue.""" wolffd@0: wolffd@0: return self.id wolffd@0: wolffd@0: def get_upcoming_events(self): wolffd@0: """Returns the upcoming events in this venue.""" wolffd@0: wolffd@0: doc = self._request("venue.getEvents", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("event"): wolffd@0: seq.append(Event(_extract(node, "id"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def get_past_events(self): wolffd@0: """Returns the past events held in this venue.""" wolffd@0: wolffd@0: doc = self._request("venue.getEvents", True) wolffd@0: wolffd@0: seq = [] wolffd@0: for node in doc.getElementsByTagName("event"): wolffd@0: seq.append(Event(_extract(node, "id"), self.network)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def md5(text): wolffd@0: """Returns the md5 hash of a string.""" wolffd@0: wolffd@0: h = hashlib.md5() wolffd@0: h.update(_unicode(text).encode("utf-8")) wolffd@0: wolffd@0: return h.hexdigest() wolffd@0: wolffd@0: def _unicode(text): wolffd@0: if sys.version_info[0] == 3: wolffd@0: if type(text) in (bytes, bytearray): wolffd@0: return str(text, "utf-8") wolffd@0: elif type(text) == str: wolffd@0: return text wolffd@0: else: wolffd@0: return str(text) wolffd@0: wolffd@0: elif sys.version_info[0] ==2: wolffd@0: if type(text) in (str,): wolffd@0: return unicode(text, "utf-8") wolffd@0: elif type(text) == unicode: wolffd@0: return text wolffd@0: else: wolffd@0: return unicode(text) wolffd@0: wolffd@0: def _string(text): wolffd@0: """For Python2 routines that can only process str type.""" wolffd@0: wolffd@0: if sys.version_info[0] == 3: wolffd@0: if type(text) != str: wolffd@0: return str(text) wolffd@0: else: wolffd@0: return text wolffd@0: wolffd@0: elif sys.version_info[0] == 2: wolffd@0: if type(text) == str: wolffd@0: return text wolffd@0: wolffd@0: if type(text) == int: wolffd@0: return str(text) wolffd@0: wolffd@0: return text.encode("utf-8") wolffd@0: wolffd@0: def _collect_nodes(limit, sender, method_name, cacheable, params=None): wolffd@0: """ wolffd@0: Returns a sequqnce of dom.Node objects about as close to wolffd@0: limit as possible wolffd@0: """ wolffd@0: wolffd@0: if not params: wolffd@0: params = sender._get_params() wolffd@0: wolffd@0: nodes = [] wolffd@0: page = 1 wolffd@0: end_of_pages = False wolffd@0: wolffd@0: while not end_of_pages and (not limit or (limit and len(nodes) < limit)): wolffd@0: params["page"] = str(page) wolffd@0: doc = sender._request(method_name, cacheable, params) wolffd@0: wolffd@0: main = doc.documentElement.childNodes[1] wolffd@0: wolffd@0: if main.hasAttribute("totalPages"): wolffd@0: total_pages = _number(main.getAttribute("totalPages")) wolffd@0: elif main.hasAttribute("totalpages"): wolffd@0: total_pages = _number(main.getAttribute("totalpages")) wolffd@0: else: wolffd@0: raise Exception("No total pages attribute") wolffd@0: wolffd@0: for node in main.childNodes: wolffd@0: if not node.nodeType == xml.dom.Node.TEXT_NODE and len(nodes) < limit: wolffd@0: nodes.append(node) wolffd@0: wolffd@0: if page >= total_pages: wolffd@0: end_of_pages = True wolffd@0: wolffd@0: page += 1 wolffd@0: wolffd@0: return nodes wolffd@0: wolffd@0: def _extract(node, name, index = 0): wolffd@0: """Extracts a value from the xml string""" wolffd@0: wolffd@0: nodes = node.getElementsByTagName(name) wolffd@0: wolffd@0: if len(nodes): wolffd@0: if nodes[index].firstChild: wolffd@0: return _unescape_htmlentity(nodes[index].firstChild.data.strip()) wolffd@0: else: wolffd@0: return None wolffd@0: wolffd@0: def _extract_all(node, name, limit_count = None): wolffd@0: """Extracts all the values from the xml string. returning a list.""" wolffd@0: wolffd@0: seq = [] wolffd@0: wolffd@0: for i in range(0, len(node.getElementsByTagName(name))): wolffd@0: if len(seq) == limit_count: wolffd@0: break wolffd@0: wolffd@0: seq.append(_extract(node, name, i)) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: def _url_safe(text): wolffd@0: """Does all kinds of tricks on a text to make it safe to use in a url.""" wolffd@0: wolffd@0: return url_quote_plus(url_quote_plus(_string(text))).lower() wolffd@0: wolffd@0: def _number(string): wolffd@0: """ wolffd@0: Extracts an int from a string. Returns a 0 if None or an empty string was passed wolffd@0: """ wolffd@0: wolffd@0: if not string: wolffd@0: return 0 wolffd@0: elif string == "": wolffd@0: return 0 wolffd@0: else: wolffd@0: try: wolffd@0: return int(string) wolffd@0: except ValueError: wolffd@0: return float(string) wolffd@0: wolffd@0: def _unescape_htmlentity(string): wolffd@0: wolffd@0: #string = _unicode(string) wolffd@0: wolffd@0: mapping = htmlentitydefs.name2codepoint wolffd@0: for key in mapping: wolffd@0: string = string.replace("&%s;" %key, unichr(mapping[key])) wolffd@0: wolffd@0: return string wolffd@0: wolffd@0: def extract_items(topitems_or_libraryitems): wolffd@0: """Extracts a sequence of items from a sequence of TopItem or LibraryItem objects.""" wolffd@0: wolffd@0: seq = [] wolffd@0: for i in topitems_or_libraryitems: wolffd@0: seq.append(i.item) wolffd@0: wolffd@0: return seq wolffd@0: wolffd@0: class ScrobblingError(Exception): wolffd@0: def __init__(self, message): wolffd@0: Exception.__init__(self) wolffd@0: self.message = message wolffd@0: wolffd@0: @_string_output wolffd@0: def __str__(self): wolffd@0: return self.message wolffd@0: wolffd@0: class BannedClientError(ScrobblingError): wolffd@0: def __init__(self): wolffd@0: ScrobblingError.__init__(self, "This version of the client has been banned") wolffd@0: wolffd@0: class BadAuthenticationError(ScrobblingError): wolffd@0: def __init__(self): wolffd@0: ScrobblingError.__init__(self, "Bad authentication token") wolffd@0: wolffd@0: class BadTimeError(ScrobblingError): wolffd@0: def __init__(self): wolffd@0: ScrobblingError.__init__(self, "Time provided is not close enough to current time") wolffd@0: wolffd@0: class BadSessionError(ScrobblingError): wolffd@0: def __init__(self): wolffd@0: ScrobblingError.__init__(self, "Bad session id, consider re-handshaking") wolffd@0: wolffd@0: class _ScrobblerRequest(object): wolffd@0: wolffd@0: def __init__(self, url, params, network, type="POST"): wolffd@0: wolffd@0: for key in params: wolffd@0: params[key] = str(params[key]) wolffd@0: wolffd@0: self.params = params wolffd@0: self.type = type wolffd@0: (self.hostname, self.subdir) = url_split_host(url[len("http:"):]) wolffd@0: self.network = network wolffd@0: wolffd@0: def execute(self): wolffd@0: """Returns a string response of this request.""" wolffd@0: wolffd@0: connection = HTTPConnection(self.hostname) wolffd@0: wolffd@0: data = [] wolffd@0: for name in self.params.keys(): wolffd@0: value = url_quote_plus(self.params[name]) wolffd@0: data.append('='.join((name, value))) wolffd@0: data = "&".join(data) wolffd@0: wolffd@0: headers = { wolffd@0: "Content-type": "application/x-www-form-urlencoded", wolffd@0: "Accept-Charset": "utf-8", wolffd@0: "User-Agent": "pylast" + "/" + __version__, wolffd@0: "HOST": self.hostname wolffd@0: } wolffd@0: wolffd@0: if self.type == "GET": wolffd@0: connection.request("GET", self.subdir + "?" + data, headers = headers) wolffd@0: else: wolffd@0: connection.request("POST", self.subdir, data, headers) wolffd@0: response = _unicode(connection.getresponse().read()) wolffd@0: wolffd@0: self._check_response_for_errors(response) wolffd@0: wolffd@0: return response wolffd@0: wolffd@0: def _check_response_for_errors(self, response): wolffd@0: """When passed a string response it checks for erros, raising wolffd@0: any exceptions as necessary.""" wolffd@0: wolffd@0: lines = response.split("\n") wolffd@0: status_line = lines[0] wolffd@0: wolffd@0: if status_line == "OK": wolffd@0: return wolffd@0: elif status_line == "BANNED": wolffd@0: raise BannedClientError() wolffd@0: elif status_line == "BADAUTH": wolffd@0: raise BadAuthenticationError() wolffd@0: elif status_line == "BADTIME": wolffd@0: raise BadTimeError() wolffd@0: elif status_line == "BADSESSION": wolffd@0: raise BadSessionError() wolffd@0: elif status_line.startswith("FAILED "): wolffd@0: reason = status_line[status_line.find("FAILED ")+len("FAILED "):] wolffd@0: raise ScrobblingError(reason) wolffd@0: wolffd@0: class Scrobbler(object): wolffd@0: """A class for scrobbling tracks to Last.fm""" wolffd@0: wolffd@0: session_id = None wolffd@0: nowplaying_url = None wolffd@0: submissions_url = None wolffd@0: wolffd@0: def __init__(self, network, client_id, client_version): wolffd@0: self.client_id = client_id wolffd@0: self.client_version = client_version wolffd@0: self.username = network.username wolffd@0: self.password = network.password_hash wolffd@0: self.network = network wolffd@0: wolffd@0: def _do_handshake(self): wolffd@0: """Handshakes with the server""" wolffd@0: wolffd@0: timestamp = str(int(time.time())) wolffd@0: wolffd@0: if self.password and self.username: wolffd@0: token = md5(self.password + timestamp) wolffd@0: elif self.network.api_key and self.network.api_secret and self.network.session_key: wolffd@0: if not self.username: wolffd@0: self.username = self.network.get_authenticated_user().get_name() wolffd@0: token = md5(self.network.api_secret + timestamp) wolffd@0: wolffd@0: params = {"hs": "true", "p": "1.2.1", "c": self.client_id, wolffd@0: "v": self.client_version, "u": self.username, "t": timestamp, wolffd@0: "a": token} wolffd@0: wolffd@0: if self.network.session_key and self.network.api_key: wolffd@0: params["sk"] = self.network.session_key wolffd@0: params["api_key"] = self.network.api_key wolffd@0: wolffd@0: server = self.network.submission_server wolffd@0: response = _ScrobblerRequest(server, params, self.network, "GET").execute().split("\n") wolffd@0: wolffd@0: self.session_id = response[1] wolffd@0: self.nowplaying_url = response[2] wolffd@0: self.submissions_url = response[3] wolffd@0: wolffd@0: def _get_session_id(self, new = False): wolffd@0: """Returns a handshake. If new is true, then it will be requested from the server wolffd@0: even if one was cached.""" wolffd@0: wolffd@0: if not self.session_id or new: wolffd@0: self._do_handshake() wolffd@0: wolffd@0: return self.session_id wolffd@0: wolffd@0: def report_now_playing(self, artist, title, album = "", duration = "", track_number = "", mbid = ""): wolffd@0: wolffd@0: _deprecation_warning("DeprecationWarning: Use Netowrk.update_now_playing(...) instead") wolffd@0: wolffd@0: params = {"s": self._get_session_id(), "a": artist, "t": title, wolffd@0: "b": album, "l": duration, "n": track_number, "m": mbid} wolffd@0: wolffd@0: try: wolffd@0: _ScrobblerRequest(self.nowplaying_url, params, self.network).execute() wolffd@0: except BadSessionError: wolffd@0: self._do_handshake() wolffd@0: self.report_now_playing(artist, title, album, duration, track_number, mbid) wolffd@0: wolffd@0: def scrobble(self, artist, title, time_started, source, mode, duration, album="", track_number="", mbid=""): wolffd@0: """Scrobble a track. parameters: wolffd@0: artist: Artist name. wolffd@0: title: Track title. wolffd@0: time_started: UTC timestamp of when the track started playing. wolffd@0: source: The source of the track wolffd@0: SCROBBLE_SOURCE_USER: Chosen by the user (the most common value, unless you have a reason for choosing otherwise, use this). wolffd@0: SCROBBLE_SOURCE_NON_PERSONALIZED_BROADCAST: Non-personalised broadcast (e.g. Shoutcast, BBC Radio 1). wolffd@0: SCROBBLE_SOURCE_PERSONALIZED_BROADCAST: Personalised recommendation except Last.fm (e.g. Pandora, Launchcast). wolffd@0: SCROBBLE_SOURCE_LASTFM: ast.fm (any mode). In this case, the 5-digit recommendation_key value must be set. wolffd@0: SCROBBLE_SOURCE_UNKNOWN: Source unknown. wolffd@0: mode: The submission mode wolffd@0: SCROBBLE_MODE_PLAYED: The track was played. wolffd@0: SCROBBLE_MODE_LOVED: The user manually loved the track (implies a listen) wolffd@0: SCROBBLE_MODE_SKIPPED: The track was skipped (Only if source was Last.fm) wolffd@0: SCROBBLE_MODE_BANNED: The track was banned (Only if source was Last.fm) wolffd@0: duration: Track duration in seconds. wolffd@0: album: The album name. wolffd@0: track_number: The track number on the album. wolffd@0: mbid: MusicBrainz ID. wolffd@0: """ wolffd@0: wolffd@0: _deprecation_warning("DeprecationWarning: Use Network.scrobble(...) instead") wolffd@0: wolffd@0: params = {"s": self._get_session_id(), "a[0]": _string(artist), "t[0]": _string(title), wolffd@0: "i[0]": str(time_started), "o[0]": source, "r[0]": mode, "l[0]": str(duration), wolffd@0: "b[0]": _string(album), "n[0]": track_number, "m[0]": mbid} wolffd@0: wolffd@0: _ScrobblerRequest(self.submissions_url, params, self.network).execute() wolffd@0: wolffd@0: def scrobble_many(self, tracks): wolffd@0: """ wolffd@0: Scrobble several tracks at once. wolffd@0: wolffd@0: tracks: A sequence of a sequence of parameters for each trach. The order of parameters wolffd@0: is the same as if passed to the scrobble() method. wolffd@0: """ wolffd@0: wolffd@0: _deprecation_warning("DeprecationWarning: Use Network.scrobble_many(...) instead") wolffd@0: wolffd@0: remainder = [] wolffd@0: wolffd@0: if len(tracks) > 50: wolffd@0: remainder = tracks[50:] wolffd@0: tracks = tracks[:50] wolffd@0: wolffd@0: params = {"s": self._get_session_id()} wolffd@0: wolffd@0: i = 0 wolffd@0: for t in tracks: wolffd@0: _pad_list(t, 9, "") wolffd@0: params["a[%s]" % str(i)] = _string(t[0]) wolffd@0: params["t[%s]" % str(i)] = _string(t[1]) wolffd@0: params["i[%s]" % str(i)] = str(t[2]) wolffd@0: params["o[%s]" % str(i)] = t[3] wolffd@0: params["r[%s]" % str(i)] = t[4] wolffd@0: params["l[%s]" % str(i)] = str(t[5]) wolffd@0: params["b[%s]" % str(i)] = _string(t[6]) wolffd@0: params["n[%s]" % str(i)] = t[7] wolffd@0: params["m[%s]" % str(i)] = t[8] wolffd@0: wolffd@0: i += 1 wolffd@0: wolffd@0: _ScrobblerRequest(self.submissions_url, params, self.network).execute() wolffd@0: wolffd@0: if remainder: wolffd@0: self.scrobble_many(remainder)