pylast._Network

Complexity

Total Complexity

98

Size/Duplication

Total Lines	657
Duplicated Lines	0 %

# -*- coding: utf-8 -*-
SCROBBLE_MODE_PLAYED = ""
SCROBBLE_MODE_LOVED = "L"
SCROBBLE_MODE_BANNED = "B"
SCROBBLE_MODE_SKIPPED = "S"

# From http://boodebr.org/main/python/all-about-python-and-unicode#UNI_XML
RE_XML_ILLEGAL = (u'([\u0000-\u0008\u000b-\u000c\u000e-\u001f\ufffe-\uffff])' +
                  u'|' +
                  u'([%s-%s][^%s-%s])|([^%s-%s][%s-%s])|([%s-%s]$)|(^[%s-%s])'
                  %
                  (unichr(0xd800), unichr(0xdbff), unichr(0xdc00),
                   unichr(0xdfff), unichr(0xd800), unichr(0xdbff),
                   unichr(0xdc00), unichr(0xdfff), unichr(0xd800),
                   unichr(0xdbff), unichr(0xdc00), unichr(0xdfff)))

XML_ILLEGAL = re.compile(RE_XML_ILLEGAL)

# Python <=3.3 doesn't support create_default_context()
# <2.7.9 and <3.2 never did any SSL verification
# FIXME This can be removed after 2017-09 when 3.3 is no longer supported and
# pypy3 uses 3.4 or later, see
# https://en.wikipedia.org/wiki/CPython#Version_history
if sys.version_info[0] == 3 and sys.version_info[1] == 3:
    import certifi
    SSL_CONTEXT = ssl.SSLContext(ssl.PROTOCOL_TLSv1)
    SSL_CONTEXT.verify_mode = ssl.CERT_REQUIRED
    SSL_CONTEXT.options |= ssl.OP_NO_COMPRESSION
    # Intermediate from https://wiki.mozilla.org/Security/Server_Side_TLS
    # Create the cipher string
    cipher_string = """
    ECDHE-ECDSA-CHACHA20-POLY1305
    ECDHE-RSA-CHACHA20-POLY1305
    ECDHE-ECDSA-AES128-GCM-SHA256
    ECDHE-RSA-AES128-GCM-SHA256
    ECDHE-ECDSA-AES256-GCM-SHA384
    ECDHE-RSA-AES256-GCM-SHA384
    DHE-RSA-AES128-GCM-SHA256
    DHE-RSA-AES256-GCM-SHA384
    ECDHE-ECDSA-AES128-SHA256
    ECDHE-RSA-AES128-SHA256
    ECDHE-ECDSA-AES128-SHA
    ECDHE-RSA-AES256-SHA384
    ECDHE-RSA-AES128-SHA
    ECDHE-ECDSA-AES256-SHA384
    ECDHE-ECDSA-AES256-SHA
    ECDHE-RSA-AES256-SHA
    DHE-RSA-AES128-SHA256
    DHE-RSA-AES128-SHA
    DHE-RSA-AES256-SHA256
    DHE-RSA-AES256-SHA
    ECDHE-ECDSA-DES-CBC3-SHA
    ECDHE-RSA-DES-CBC3-SHA
    EDH-RSA-DES-CBC3-SHA
    AES128-GCM-SHA256
    AES256-GCM-SHA384
    AES128-SHA256
    AES256-SHA256
    AES128-SHA
    AES256-SHA
    DES-CBC3-SHA
    !DSS
    """
    cipher_string = ' '.join(cipher_string.split())
    SSL_CONTEXT.set_ciphers(cipher_string)
    SSL_CONTEXT.load_verify_locations(certifi.where())

# Python >3.4 and >2.7.9 has sane defaults
elif sys.version_info > (3, 4) or ((2, 7, 9) < sys.version_info < (3, 0)):
    SSL_CONTEXT = ssl.create_default_context()


class _Network(object):
    """
    A music social network website such as Last.fm or
    one with a Last.fm-compatible API.
    """

    def __init__(
            self, name, homepage, ws_server, api_key, api_secret, session_key,
            submission_server, username, password_hash, domain_names, urls):
        """
            name: the name of the network
            homepage: the homepage URL
            ws_server: the URL of the webservices server
            api_key: a provided API_KEY
            api_secret: a provided API_SECRET
            session_key: a generated session_key or None
            submission_server: the URL of the server to which tracks are
                submitted (scrobbled)
            username: a username of a valid user
            password_hash: the output of pylast.md5(password) where password is
                the user's password
            domain_names: a dict mapping each DOMAIN_* value to a string domain
                name
            urls: a dict mapping types to URLs

            if username and password_hash were provided and not session_key,
            session_key will be generated automatically when needed.

            Either a valid session_key or a combination of username and
            password_hash must be present for scrobbling.

            You should use a preconfigured network object through a
            get_*_network(...) method instead of creating an object
            of this class, unless you know what you're doing.
        """

        self.name = name
        self.homepage = homepage
        self.ws_server = ws_server
        self.api_key = api_key
        self.api_secret = api_secret
        self.session_key = session_key
        self.submission_server = submission_server
        self.username = username
        self.password_hash = password_hash
        self.domain_names = domain_names
        self.urls = urls

        self.cache_backend = None
        self.proxy_enabled = False
        self.proxy = None
        self.last_call_time = 0
        self.limit_rate = False

        # Generate a session_key if necessary
        if ((self.api_key and self.api_secret) and not self.session_key and
           (self.username and self.password_hash)):
            sk_gen = SessionKeyGenerator(self)
            self.session_key = sk_gen.get_session_key(
                self.username, self.password_hash)

    def __str__(self):
        return "%s Network" % self.name

    def get_artist(self, artist_name):
        """
            Return an Artist object
        """

        return Artist(artist_name, self)

    def get_track(self, artist, title):
        """
            Return a Track object
        """

        return Track(artist, title, self)

    def get_album(self, artist, title):
        """
            Return an Album object
        """

        return Album(artist, title, self)

    def get_authenticated_user(self):
        """
            Returns the authenticated user
        """

        return AuthenticatedUser(self)

    def get_country(self, country_name):
        """
            Returns a country object
        """

        return Country(country_name, self)

    def get_metro(self, metro_name, country_name):
        """
            Returns a metro object
        """

        return Metro(metro_name, country_name, self)

    def get_group(self, name):
        """
            Returns a Group object
        """

        return Group(name, self)

    def get_user(self, username):
        """
            Returns a user object
        """

        return User(username, self)

    def get_tag(self, name):
        """
            Returns a tag object
        """

        return Tag(name, self)

    def get_scrobbler(self, client_id, client_version):
        """
            Returns a Scrobbler object used for submitting tracks to the server

            Quote from http://www.last.fm/api/submissions:
            ========
            Client identifiers are used to provide a centrally managed database
            of the client versions, allowing clients to be banned if they are
            found to be behaving undesirably. The client ID is associated with
            a version number on the server, however these are only incremented
            if a client is banned and do not have to reflect the version of the
            actual client application.

            During development, clients which have not been allocated an
            identifier should use the identifier tst, with a version number of
            1.0. Do not distribute code or client implementations which use
            this test identifier. Do not use the identifiers used by other
            clients.
            =========

            To obtain a new client identifier please contact:
                * Last.fm: submissions@last.fm
                * # TODO: list others

            ...and provide us with the name of your client and its homepage
            address.
        """

This code seems to be duplicated in your project.

        _deprecation_warning(
            "Use _Network.scrobble(...), _Network.scrobble_many(...),"
            " and Network.update_now_playing(...) instead")

        return Scrobbler(self, client_id, client_version)

    def _get_language_domain(self, domain_language):
        """
            Returns the mapped domain name of the network to a DOMAIN_* value
        """

        if domain_language in self.domain_names:
            return self.domain_names[domain_language]

    def _get_url(self, domain, url_type):
        return "http://%s/%s" % (
            self._get_language_domain(domain), self.urls[url_type])

    def _get_ws_auth(self):
        """
            Returns an (API_KEY, API_SECRET, SESSION_KEY) tuple.
        """
        return (self.api_key, self.api_secret, self.session_key)

    def _delay_call(self):
        """
            Makes sure that web service calls are at least 0.2 seconds apart.
        """

        # Delay time in seconds from section 4.4 of http://www.last.fm/api/tos
        DELAY_TIME = 0.2
        now = time.time()

        time_since_last = now - self.last_call_time

        if time_since_last < DELAY_TIME:
            time.sleep(DELAY_TIME - time_since_last)

        self.last_call_time = now

    def create_new_playlist(self, title, description):
        """
            Creates a playlist for the authenticated user and returns it
                title: The title of the new playlist.
                description: The description of the new playlist.
        """

        params = {}
        params['title'] = title
        params['description'] = description

        doc = _Request(self, 'playlist.create', params).execute(False)

        e_id = doc.getElementsByTagName("id")[0].firstChild.data
        user = doc.getElementsByTagName('playlists')[0].getAttribute('user')

        return Playlist(user, e_id, self)

    def get_top_artists(self, limit=None, cacheable=True):
        """Returns the most played artists as a sequence of TopItem objects."""

        params = {}
        if limit:
            params["limit"] = limit

        doc = _Request(self, "chart.getTopArtists", params).execute(cacheable)

        return _extract_top_artists(doc, self)

    def get_top_tracks(self, limit=None, cacheable=True):
        """Returns the most played tracks as a sequence of TopItem objects."""

        params = {}
        if limit:
            params["limit"] = limit

        doc = _Request(self, "chart.getTopTracks", params).execute(cacheable)

        seq = []
        for node in doc.getElementsByTagName("track"):
            title = _extract(node, "name")
            artist = _extract(node, "name", 1)
            track = Track(artist, title, self)
            weight = _number(_extract(node, "playcount"))
            seq.append(TopItem(track, weight))

        return seq

    def get_top_tags(self, limit=None, cacheable=True):
        """Returns the most used tags as a sequence of TopItem objects."""

        # Last.fm has no "limit" parameter for tag.getTopTags
        # so we need to get all (250) and then limit locally
        doc = _Request(self, "tag.getTopTags").execute(cacheable)

        seq = []
        for node in doc.getElementsByTagName("tag"):
            if limit and len(seq) >= limit:
                break
            tag = Tag(_extract(node, "name"), self)
            weight = _number(_extract(node, "count"))
            seq.append(TopItem(tag, weight))

        return seq

    def get_geo_events(
            self, longitude=None, latitude=None, location=None, distance=None,
            tag=None, festivalsonly=None, limit=None, cacheable=True):
        """
        Returns all events in a specific location by country or city name.
        Parameters:
        longitude (Optional) : Specifies a longitude value to retrieve events
            for (service returns nearby events by default)
        latitude (Optional) : Specifies a latitude value to retrieve events for
            (service returns nearby events by default)
        location (Optional) : Specifies a location to retrieve events for
            (service returns nearby events by default)
        distance (Optional) : Find events within a specified radius
            (in kilometres)
        tag (Optional) : Specifies a tag to filter by.
        festivalsonly[0|1] (Optional) : Whether only festivals should be
            returned, or all events.
        limit (Optional) : The number of results to fetch per page.
            Defaults to 10.
        """

        params = {}

        if longitude:
            params["long"] = longitude
        if latitude:
            params["lat"] = latitude
        if location:
            params["location"] = location
        if limit:
            params["limit"] = limit

This code seems to be duplicated in your project.

        if distance:
            params["distance"] = distance
        if tag:
            params["tag"] = tag
        if festivalsonly:
            params["festivalsonly"] = 1
        elif not festivalsonly:
            params["festivalsonly"] = 0

        doc = _Request(self, "geo.getEvents", params).execute(cacheable)

        return _extract_events_from_doc(doc, self)

    def get_metro_weekly_chart_dates(self, cacheable=True):
        """
        Returns a list of From and To tuples for the available metro charts.
        """

        doc = _Request(self, "geo.getMetroWeeklyChartlist").execute(cacheable)

        seq = []
        for node in doc.getElementsByTagName("chart"):
            seq.append((node.getAttribute("from"), node.getAttribute("to")))

        return seq

    def get_metros(self, country=None, cacheable=True):
        """
        Get a list of valid countries and metros for use in the other
        webservices.
        Parameters:
        country (Optional) : Optionally restrict the results to those Metros
            from a particular country, as defined by the ISO 3166-1 country
            names standard.
        """
        params = {}

        if country:
            params["country"] = country

        doc = _Request(self, "geo.getMetros", params).execute(cacheable)

        metros = doc.getElementsByTagName("metro")
        seq = []

        for metro in metros:
            name = _extract(metro, "name")
            country = _extract(metro, "country")

            seq.append(Metro(name, country, self))

        return seq

    def get_geo_top_artists(self, country, limit=None, cacheable=True):
        """Get the most popular artists on Last.fm by country.
        Parameters:
        country (Required) : A country name, as defined by the ISO 3166-1
            country names standard.
        limit (Optional) : The number of results to fetch per page.
            Defaults to 50.
        """
        params = {"country": country}

        if limit:
            params["limit"] = limit

        doc = _Request(self, "geo.getTopArtists", params).execute(cacheable)

        return _extract_top_artists(doc, self)

    def get_geo_top_tracks(
            self, country, location=None, limit=None, cacheable=True):
        """Get the most popular tracks on Last.fm last week by country.
        Parameters:
        country (Required) : A country name, as defined by the ISO 3166-1
            country names standard
        location (Optional) : A metro name, to fetch the charts for
            (must be within the country specified)
        limit (Optional) : The number of results to fetch per page.
            Defaults to 50.
        """
        params = {"country": country}

        if location:
            params["location"] = location
        if limit:
            params["limit"] = limit

        doc = _Request(self, "geo.getTopTracks", params).execute(cacheable)

        tracks = doc.getElementsByTagName("track")
        seq = []

        for track in tracks:
            title = _extract(track, "name")
            artist = _extract(track, "name", 1)
            listeners = _extract(track, "listeners")

            seq.append(TopItem(Track(artist, title, self), listeners))

        return seq

    def enable_proxy(self, host, port):
        """Enable a default web proxy"""

        self.proxy = [host, _number(port)]
        self.proxy_enabled = True

    def disable_proxy(self):
        """Disable using the web proxy"""

        self.proxy_enabled = False

    def is_proxy_enabled(self):
        """Returns True if a web proxy is enabled."""

        return self.proxy_enabled

    def _get_proxy(self):
        """Returns proxy details."""

        return self.proxy

    def enable_rate_limit(self):
        """Enables rate limiting for this network"""
        self.limit_rate = True

    def disable_rate_limit(self):
        """Disables rate limiting for this network"""
        self.limit_rate = False

    def is_rate_limited(self):
        """Return True if web service calls are rate limited"""
        return self.limit_rate

    def enable_caching(self, file_path=None):
        """Enables caching request-wide for all cacheable calls.

        * file_path: A file path for the backend storage file. If
        None set, a temp file would probably be created, according the backend.
        """

        if not file_path:
            file_path = tempfile.mktemp(prefix="pylast_tmp_")

        self.cache_backend = _ShelfCacheBackend(file_path)

    def disable_caching(self):
        """Disables all caching features."""

        self.cache_backend = None

    def is_caching_enabled(self):
        """Returns True if caching is enabled."""

        return not (self.cache_backend is None)

    def _get_cache_backend(self):

        return self.cache_backend

    def search_for_album(self, album_name):
        """Searches for an album by its name. Returns a AlbumSearch object.
        Use get_next_page() to retrieve sequences of results."""

        return AlbumSearch(album_name, self)

    def search_for_artist(self, artist_name):
        """Searches of an artist by its name. Returns a ArtistSearch object.
        Use get_next_page() to retrieve sequences of results."""

        return ArtistSearch(artist_name, self)

    def search_for_tag(self, tag_name):
        """Searches of a tag by its name. Returns a TagSearch object.
        Use get_next_page() to retrieve sequences of results."""

        return TagSearch(tag_name, self)

    def search_for_track(self, artist_name, track_name):
        """Searches of a track by its name and its artist. Set artist to an
        empty string if not available.
        Returns a TrackSearch object.
        Use get_next_page() to retrieve sequences of results."""

        return TrackSearch(artist_name, track_name, self)

    def search_for_venue(self, venue_name, country_name):
        """Searches of a venue by its name and its country. Set country_name to
        an empty string if not available.
        Returns a VenueSearch object.
        Use get_next_page() to retrieve sequences of results."""

        return VenueSearch(venue_name, country_name, self)

    def get_track_by_mbid(self, mbid):
        """Looks up a track by its MusicBrainz ID"""

        params = {"mbid": mbid}

        doc = _Request(self, "track.getInfo", params).execute(True)

        return Track(_extract(doc, "name", 1), _extract(doc, "name"), self)

    def get_artist_by_mbid(self, mbid):
        """Loooks up an artist by its MusicBrainz ID"""

        params = {"mbid": mbid}

        doc = _Request(self, "artist.getInfo", params).execute(True)

        return Artist(_extract(doc, "name"), self)

    def get_album_by_mbid(self, mbid):
        """Looks up an album by its MusicBrainz ID"""

        params = {"mbid": mbid}

        doc = _Request(self, "album.getInfo", params).execute(True)

        return Album(_extract(doc, "artist"), _extract(doc, "name"), self)

    def update_now_playing(
            self, artist, title, album=None, album_artist=None,
            duration=None, track_number=None, mbid=None, context=None):
        """
        Used to notify Last.fm that a user has started listening to a track.

            Parameters:
                artist (Required) : The artist name
                title (Required) : The track title
                album (Optional) : The album name.
                album_artist (Optional) : The album artist - if this differs
                    from the track artist.
                duration (Optional) : The length of the track in seconds.
                track_number (Optional) : The track number of the track on the
                    album.
                mbid (Optional) : The MusicBrainz Track ID.
                context (Optional) : Sub-client version
                    (not public, only enabled for certain API keys)
        """

        params = {"track": title, "artist": artist}

        if album:
            params["album"] = album
        if album_artist:
            params["albumArtist"] = album_artist
        if context:
            params["context"] = context
        if track_number:
            params["trackNumber"] = track_number
        if mbid:
            params["mbid"] = mbid
        if duration:
            params["duration"] = duration

        _Request(self, "track.updateNowPlaying", params).execute()

    def scrobble(
            self, artist, title, timestamp, album=None, album_artist=None,
            track_number=None, duration=None, stream_id=None, context=None,
            mbid=None):

        """Used to add a track-play to a user's profile.

        Parameters:
            artist (Required) : The artist name.
            title (Required) : The track name.
            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.
            album (Optional) : The album name.
            album_artist (Optional) : The album artist - if this differs from
                the track artist.
            context (Optional) : Sub-client version (not public, only enabled
                for certain API keys)
            stream_id (Optional) : The stream id for this track received from
                the radio.getPlaylist service.
            track_number (Optional) : The track number of the track on the
                album.
            mbid (Optional) : The MusicBrainz Track ID.
            duration (Optional) : The length of the track in seconds.
        """

        return self.scrobble_many(({
            "artist": artist, "title": title, "timestamp": timestamp,
            "album": album, "album_artist": album_artist,
            "track_number": track_number, "duration": duration,
            "stream_id": stream_id, "context": context, "mbid": mbid},))

    def scrobble_many(self, tracks):
        """
        Used to scrobble a batch of tracks at once. The parameter tracks is a
        sequence of dicts per track containing the keyword arguments as if
        passed to the scrobble() method.