<s> from __future__ import print_function , division , absolute_import , unicode_literals <EOL> from future import standard_library <EOL> standard_library . install_aliases ( ) <EOL> from past . builtins import basestring <EOL> from builtins import * <EOL> from collections import defaultdict <EOL> import datetime <EOL> from operator import itemgetter <EOL> import re <EOL> from uuid import getnode as getmac <EOL> from gmusicapi import session <EOL> from gmusicapi . clients . shared import _Base <EOL> from gmusicapi . exceptions import CallFailure , NotSubscribed <EOL> from gmusicapi . protocol import mobileclient <EOL> from gmusicapi . utils import utils <EOL> class Mobileclient ( _Base ) : <EOL> """<STR_LIT>""" <EOL> _session_class = session . Mobileclient <EOL> FROM_MAC_ADDRESS = object ( ) <EOL> def __init__ ( self , debug_logging = True , validate = True , verify_ssl = True ) : <EOL> super ( Mobileclient , self ) . __init__ ( self . __class__ . __name__ , <EOL> debug_logging , <EOL> validate , <EOL> verify_ssl ) <EOL> @ utils . cached_property ( ttl = <NUM_LIT> ) <EOL> def is_subscribed ( self ) : <EOL> """<STR_LIT>""" <EOL> res = self . _make_call ( mobileclient . Config ) <EOL> for item in res [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] : <EOL> if item [ '<STR_LIT:key>' ] == '<STR_LIT>' and item [ '<STR_LIT:value>' ] == '<STR_LIT:true>' : <EOL> return True <EOL> else : <EOL> return False <EOL> def login ( self , email , password , android_id ) : <EOL> """<STR_LIT>""" <EOL> if android_id is None : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> if android_id is self . FROM_MAC_ADDRESS : <EOL> mac_int = getmac ( ) <EOL> if ( mac_int >> <NUM_LIT> ) % <NUM_LIT:2> : <EOL> raise OSError ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> android_id = utils . create_mac_string ( mac_int ) <EOL> android_id = android_id . replace ( '<STR_LIT::>' , '<STR_LIT>' ) <EOL> if not self . session . login ( email , password , android_id ) : <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> return False <EOL> self . android_id = android_id <EOL> self . logger . info ( "<STR_LIT>" ) <EOL> return True <EOL> def get_all_songs ( self , incremental = False , include_deleted = False ) : <EOL> """<STR_LIT>""" <EOL> tracks = self . _get_all_items ( mobileclient . ListTracks , incremental , include_deleted ) <EOL> return tracks <EOL> @ utils . accept_singleton ( dict ) <EOL> @ utils . empty_arg_shortcircuit <EOL> def change_song_metadata ( self , songs ) : <EOL> """<STR_LIT>""" <EOL> mutate_call = mobileclient . BatchMutateTracks <EOL> mutations = [ { '<STR_LIT>' : s } for s in songs ] <EOL> self . _make_call ( mutate_call , mutations ) <EOL> return [ utils . id_or_nid ( d ) for d in songs ] <EOL> def increment_song_playcount ( self , song_id , plays = <NUM_LIT:1> , playtime = None ) : <EOL> """<STR_LIT>""" <EOL> if playtime is None : <EOL> playtime = datetime . datetime . now ( ) <EOL> self . _make_call ( mobileclient . IncrementPlayCount , song_id , plays , playtime ) <EOL> return song_id <EOL> @ utils . require_subscription <EOL> @ utils . enforce_id_param <EOL> def add_store_track ( self , store_song_id ) : <EOL> """<STR_LIT>""" <EOL> store_track_info = self . get_track_info ( store_song_id ) <EOL> mutate_call = mobileclient . BatchMutateTracks <EOL> add_mutation = mutate_call . build_track_add ( store_track_info ) <EOL> res = self . _make_call ( mutate_call , [ add_mutation ] ) <EOL> return res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> @ utils . accept_singleton ( basestring ) <EOL> @ utils . enforce_ids_param <EOL> @ utils . empty_arg_shortcircuit <EOL> def delete_songs ( self , library_song_ids ) : <EOL> """<STR_LIT>""" <EOL> mutate_call = mobileclient . BatchMutateTracks <EOL> del_mutations = mutate_call . build_track_deletes ( library_song_ids ) <EOL> res = self . _make_call ( mutate_call , del_mutations ) <EOL> return [ d [ '<STR_LIT:id>' ] for d in res [ '<STR_LIT>' ] ] <EOL> @ utils . enforce_id_param <EOL> def get_stream_url ( self , song_id , device_id = None , quality = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if song_id . startswith ( '<STR_LIT:T>' ) and not self . is_subscribed : <EOL> raise NotSubscribed ( "<STR_LIT>" ) <EOL> if device_id is None : <EOL> device_id = self . android_id <EOL> if len ( device_id ) == <NUM_LIT:16> and re . match ( '<STR_LIT>' , device_id ) : <EOL> device_id = str ( int ( device_id , <NUM_LIT:16> ) ) <EOL> return self . _make_call ( mobileclient . GetStreamUrl , song_id , device_id , quality ) <EOL> def get_all_playlists ( self , incremental = False , include_deleted = False ) : <EOL> """<STR_LIT>""" <EOL> playlists = self . _get_all_items ( mobileclient . ListPlaylists , incremental , include_deleted ) <EOL> return playlists <EOL> def create_playlist ( self , name , description = None , public = False ) : <EOL> """<STR_LIT>""" <EOL> share_state = '<STR_LIT>' if public else '<STR_LIT>' <EOL> mutate_call = mobileclient . BatchMutatePlaylists <EOL> add_mutations = mutate_call . build_playlist_adds ( [ { '<STR_LIT:name>' : name , <EOL> '<STR_LIT:description>' : description , <EOL> '<STR_LIT>' : share_state } ] ) <EOL> res = self . _make_call ( mutate_call , add_mutations ) <EOL> return res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> @ utils . enforce_id_param <EOL> def edit_playlist ( self , playlist_id , new_name = None , new_description = None , public = None ) : <EOL> """<STR_LIT>""" <EOL> if all ( value is None for value in ( new_name , new_description , public ) ) : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if public is None : <EOL> share_state = public <EOL> else : <EOL> share_state = '<STR_LIT>' if public else '<STR_LIT>' <EOL> mutate_call = mobileclient . BatchMutatePlaylists <EOL> update_mutations = mutate_call . build_playlist_updates ( [ <EOL> { '<STR_LIT:id>' : playlist_id , '<STR_LIT:name>' : new_name , <EOL> '<STR_LIT:description>' : new_description , '<STR_LIT>' : share_state } <EOL> ] ) <EOL> res = self . _make_call ( mutate_call , update_mutations ) <EOL> return res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> @ utils . enforce_id_param <EOL> def delete_playlist ( self , playlist_id ) : <EOL> """<STR_LIT>""" <EOL> mutate_call = mobileclient . BatchMutatePlaylists <EOL> del_mutations = mutate_call . build_playlist_deletes ( [ playlist_id ] ) <EOL> res = self . _make_call ( mutate_call , del_mutations ) <EOL> return res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> def get_all_user_playlist_contents ( self ) : <EOL> """<STR_LIT>""" <EOL> user_playlists = [ p for p in self . get_all_playlists ( ) <EOL> if ( p . get ( '<STR_LIT:type>' ) == '<STR_LIT>' or <EOL> p . get ( '<STR_LIT:type>' ) != '<STR_LIT>' or <EOL> '<STR_LIT:type>' not in p ) ] <EOL> all_entries = self . _get_all_items ( mobileclient . ListPlaylistEntries , <EOL> incremental = False , include_deleted = False , <EOL> updated_after = None ) <EOL> for playlist in user_playlists : <EOL> entries = [ e for e in all_entries <EOL> if e [ '<STR_LIT>' ] == playlist [ '<STR_LIT:id>' ] ] <EOL> entries . sort ( key = itemgetter ( '<STR_LIT>' ) ) <EOL> playlist [ '<STR_LIT>' ] = entries <EOL> return user_playlists <EOL> def get_shared_playlist_contents ( self , share_token ) : <EOL> """<STR_LIT>""" <EOL> res = self . _make_call ( mobileclient . ListSharedPlaylistEntries , <EOL> updated_after = None , share_token = share_token ) <EOL> entries = res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> entries . sort ( key = itemgetter ( '<STR_LIT>' ) ) <EOL> return entries <EOL> @ utils . accept_singleton ( basestring , <NUM_LIT:2> ) <EOL> @ utils . enforce_id_param <EOL> @ utils . enforce_ids_param ( position = <NUM_LIT:2> ) <EOL> @ utils . empty_arg_shortcircuit ( position = <NUM_LIT:2> ) <EOL> def add_songs_to_playlist ( self , playlist_id , song_ids ) : <EOL> """<STR_LIT>""" <EOL> mutate_call = mobileclient . BatchMutatePlaylistEntries <EOL> add_mutations = mutate_call . build_plentry_adds ( playlist_id , song_ids ) <EOL> res = self . _make_call ( mutate_call , add_mutations ) <EOL> return [ e [ '<STR_LIT:id>' ] for e in res [ '<STR_LIT>' ] ] <EOL> @ utils . accept_singleton ( basestring , <NUM_LIT:1> ) <EOL> @ utils . enforce_ids_param ( position = <NUM_LIT:1> ) <EOL> @ utils . empty_arg_shortcircuit ( position = <NUM_LIT:1> ) <EOL> def remove_entries_from_playlist ( self , entry_ids ) : <EOL> """<STR_LIT>""" <EOL> mutate_call = mobileclient . BatchMutatePlaylistEntries <EOL> del_mutations = mutate_call . build_plentry_deletes ( entry_ids ) <EOL> res = self . _make_call ( mutate_call , del_mutations ) <EOL> return [ e [ '<STR_LIT:id>' ] for e in res [ '<STR_LIT>' ] ] <EOL> def reorder_playlist_entry ( self , entry , to_follow_entry = None , to_precede_entry = None ) : <EOL> """<STR_LIT>""" <EOL> if to_follow_entry is None and to_precede_entry is None : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> mutate_call = mobileclient . BatchMutatePlaylistEntries <EOL> before = to_follow_entry [ '<STR_LIT>' ] if to_follow_entry else None <EOL> after = to_precede_entry [ '<STR_LIT>' ] if to_precede_entry else None <EOL> reorder_mutation = mutate_call . build_plentry_reorder ( entry , before , after ) <EOL> res = self . _make_call ( mutate_call , [ reorder_mutation ] ) <EOL> return [ e [ '<STR_LIT:id>' ] for e in res [ '<STR_LIT>' ] ] <EOL> def get_registered_devices ( self ) : <EOL> """<STR_LIT>""" <EOL> res = self . _make_call ( mobileclient . GetDeviceManagementInfo ) <EOL> return res [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] if '<STR_LIT:data>' in res else [ ] <EOL> def deauthorize_device ( self , device_id ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> self . _make_call ( mobileclient . DeauthDevice , device_id ) <EOL> except CallFailure : <EOL> self . logger . exception ( "<STR_LIT>" ) <EOL> return False <EOL> return True <EOL> def get_promoted_songs ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_all_items ( mobileclient . ListPromotedTracks , <EOL> incremental = False , include_deleted = False , <EOL> updated_after = None ) <EOL> def create_station ( self , name , <EOL> track_id = None , artist_id = None , album_id = None , <EOL> genre_id = None , playlist_token = None ) : <EOL> """<STR_LIT>""" <EOL> seed = { } <EOL> if track_id is not None : <EOL> if track_id [ <NUM_LIT:0> ] == '<STR_LIT:T>' : <EOL> seed [ '<STR_LIT>' ] = track_id <EOL> seed [ '<STR_LIT>' ] = <NUM_LIT:2> <EOL> else : <EOL> seed [ '<STR_LIT>' ] = track_id <EOL> seed [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> if artist_id is not None : <EOL> seed [ '<STR_LIT>' ] = artist_id <EOL> seed [ '<STR_LIT>' ] = <NUM_LIT:3> <EOL> if album_id is not None : <EOL> seed [ '<STR_LIT>' ] = album_id <EOL> seed [ '<STR_LIT>' ] = <NUM_LIT:4> <EOL> if genre_id is not None : <EOL> seed [ '<STR_LIT>' ] = genre_id <EOL> seed [ '<STR_LIT>' ] = <NUM_LIT:5> <EOL> if playlist_token is not None : <EOL> seed [ '<STR_LIT>' ] = playlist_token <EOL> seed [ '<STR_LIT>' ] = <NUM_LIT:8> <EOL> if len ( seed ) > <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> mutate_call = mobileclient . BatchMutateStations <EOL> add_mutation = mutate_call . build_add ( name , seed , include_tracks = False , num_tracks = <NUM_LIT:0> ) <EOL> res = self . _make_call ( mutate_call , [ add_mutation ] ) <EOL> return res [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:id>' ] <EOL> @ utils . accept_singleton ( basestring ) <EOL> @ utils . enforce_ids_param <EOL> @ utils . empty_arg_shortcircuit <EOL> def delete_stations ( self , station_ids ) : <EOL> """<STR_LIT>""" <EOL> mutate_call = mobileclient . BatchMutateStations <EOL> delete_mutations = mutate_call . build_deletes ( station_ids ) <EOL> res = self . _make_call ( mutate_call , delete_mutations ) <EOL> return [ s [ '<STR_LIT:id>' ] for s in res [ '<STR_LIT>' ] ] <EOL> def get_all_stations ( self , incremental = False , include_deleted = False , updated_after = None ) : <EOL> """<STR_LIT>""" <EOL> return self . _get_all_items ( mobileclient . ListStations , incremental , include_deleted , <EOL> updated_after = updated_after ) <EOL> def get_station_tracks ( self , station_id , num_tracks = <NUM_LIT> , recently_played_ids = None ) : <EOL> """<STR_LIT>""" <EOL> if recently_played_ids is None : <EOL> recently_played_ids = [ ] <EOL> def add_track_type ( track_id ) : <EOL> if track_id [ <NUM_LIT:0> ] == '<STR_LIT:T>' : <EOL> return { '<STR_LIT:id>' : track_id , '<STR_LIT:type>' : <NUM_LIT:1> } <EOL> else : <EOL> return { '<STR_LIT:id>' : track_id , '<STR_LIT:type>' : <NUM_LIT:0> } <EOL> recently_played = [ add_track_type ( track_id ) for track_id in recently_played_ids ] <EOL> res = self . _make_call ( mobileclient . ListStationTracks , <EOL> station_id , num_tracks , recently_played = recently_played ) <EOL> stations = res . get ( '<STR_LIT:data>' , { } ) . get ( '<STR_LIT>' ) <EOL> if not stations : <EOL> return [ ] <EOL> return stations [ <NUM_LIT:0> ] . get ( '<STR_LIT>' , [ ] ) <EOL> def search ( self , query , max_results = <NUM_LIT:50> ) : <EOL> """<STR_LIT>""" <EOL> res = self . _make_call ( mobileclient . Search , query , max_results ) <EOL> hits = res . get ( '<STR_LIT>' , [ ] ) <EOL> hits_by_type = defaultdict ( list ) <EOL> for hit in hits : <EOL> hits_by_type [ hit [ '<STR_LIT:type>' ] ] . append ( hit ) <EOL> return { '<STR_LIT>' : hits_by_type [ '<STR_LIT:3>' ] , <EOL> '<STR_LIT>' : hits_by_type [ '<STR_LIT:2>' ] , <EOL> '<STR_LIT>' : hits_by_type [ '<STR_LIT:4>' ] , <EOL> '<STR_LIT>' : hits_by_type [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : hits_by_type [ '<STR_LIT:1>' ] , <EOL> '<STR_LIT>' : hits_by_type [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : hits_by_type [ '<STR_LIT>' ] } <EOL> @ utils . enforce_id_param <EOL> def get_artist_info ( self , artist_id , include_albums = True , max_top_tracks = <NUM_LIT:5> , max_rel_artist = <NUM_LIT:5> ) : <EOL> """<STR_LIT>""" <EOL> res = self . _make_call ( mobileclient . GetArtist , <EOL> artist_id , include_albums , max_top_tracks , max_rel_artist ) <EOL> return res <EOL> def _get_all_items ( self , call , incremental , include_deleted , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if not incremental : <EOL> kwargs . setdefault ( '<STR_LIT>' , <NUM_LIT> ) <EOL> generator = self . _get_all_items_incremental ( call , include_deleted , ** kwargs ) <EOL> if incremental : <EOL> return generator <EOL> return [ s for chunk in generator for s in chunk ] <EOL> def _get_all_items_incremental ( self , call , include_deleted , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> get_next_chunk = True <EOL> lib_chunk = { '<STR_LIT>' : None } <EOL> while get_next_chunk : <EOL> lib_chunk = self . _make_call ( call , <EOL> start_token = lib_chunk [ '<STR_LIT>' ] , <EOL> ** kwargs ) <EOL> items = lib_chunk [ '<STR_LIT:data>' ] [ '<STR_LIT>' ] <EOL> if not include_deleted : <EOL> items = [ item for item in items <EOL> if not item . get ( '<STR_LIT>' , False ) ] <EOL> yield items <EOL> get_next_chunk = '<STR_LIT>' in lib_chunk <EOL> @ utils . enforce_id_param <EOL> def get_album_info ( self , album_id , include_tracks = True ) : <EOL> """<STR_LIT>""" <EOL> return self . _make_call ( mobileclient . GetAlbum , album_id , include_tracks ) <EOL> @ utils . enforce_id_param <EOL> def get_track_info ( self , store_track_id ) : <EOL> """<STR_LIT>""" <EOL> return self . _make_call ( mobileclient . GetStoreTrack , store_track_id ) <EOL> def get_genres ( self , parent_genre_id = None ) : <EOL> """<STR_LIT>""" <EOL> res = self . _make_call ( mobileclient . GetGenres , parent_genre_id ) <EOL> return res . get ( '<STR_LIT>' , [ ] ) </s>
