import os
import subprocess
from base64 import b64encode
import audio_metadata
import pendulum
from attr import attrib, attrs
from .models import MusicManagerCall
from .pb import download_pb2, locker_pb2, upload_pb2
from .utils import generate_client_id, get_album_art, transcode_to_mp3
from ..models import Call, JSONCall
[docs]@attrs(slots=True)
class ClientState(MusicManagerCall):
"""Get information about the state of a Google Music account.
Note:
This provides things like the quota for uploaded songs.
Parameters:
uploader_id (str):
A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
"""
endpoint = 'clientstate'
method = 'POST'
request_type = upload_pb2.ClientStateRequest
uploader_id = attrib()
def __attrs_post_init__(self):
super().__attrs_post_init__(self.uploader_id)
[docs]@attrs(slots=True)
class Export(Call):
"""Download a song from a Google Music library.
Parameters:
uploader_id (str):
A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
song_id (str): A song ID.
"""
base_url = 'https://music.google.com/music/export'
follow_redirects = True
method = 'GET'
uploader_id = attrib()
song_id = attrib()
def __attrs_post_init__(self):
self._headers = {'X-Device-ID': self.uploader_id}
self._params.update(
{'songid': self.song_id}
)
self._url = Export.base_url
[docs]@attrs(slots=True)
class ExportIDs(Call):
"""Get a listing of uploaded and purchased library tracks.
Note:
The track list is paged.
Getting all tracks will require looping through all pages.
Parameters:
uploader_id (str):
A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
continuation_token (str, Optional):
The token of the page to return.
Default: Not sent to get first page.
export_type (int, Optional):
The type of tracks to return:
1 for all tracks, 2 for promotional and purchased.
Default: ``1``
updated_min (int, Optional):
List changes since the given Unix epoch time in microseconds.
Default lists all changes.
"""
base_url = 'https://music.google.com/music/exportids'
method = 'POST'
request_type = download_pb2.GetTracksToExportRequest
response_type = download_pb2.GetTracksToExportResponse
uploader_id = attrib()
continuation_token = attrib(default=None)
export_type = attrib(default=1)
updated_min = attrib(default=-1)
def __attrs_post_init__(self):
self._data = self.request_type()
self._data.client_id = self.uploader_id
self._data.export_type = self.export_type
self._data.updated_min = self.updated_min
self._headers.update(
{
'Content-Type': 'application/x-google-protobuf',
'X-Device-ID': self.uploader_id,
}
)
self._params.update(
{'version': 1}
)
if self.continuation_token is not None:
self._data.continuation_token = self.continuation_token
self._url = ExportIDs.base_url
@property
def body(self):
"""Binary-encoded body of the HTTP request."""
return self._data.SerializeToString() if self._data else b''
@staticmethod
def check_success(response_body):
return response_body.status == download_pb2.GetTracksToExportResponse.OK
parse_response = MusicManagerCall.parse_response
[docs]@attrs(slots=True)
class GetJobs(MusicManagerCall):
"""Get a listing of pending upload jobs.
Parameters:
uploader_id (str):
A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
"""
endpoint = 'getjobs'
method = 'POST'
request_type = upload_pb2.GetJobsRequest
uploader_id = attrib()
def __attrs_post_init__(self):
super().__attrs_post_init__(self.uploader_id)
@staticmethod
def check_success(response):
return response.get_tracks_success
[docs]@attrs(slots=True)
class Sample(MusicManagerCall):
"""Send samples of audio files to Google Music.
Parameters:
uploader_id (str):
A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
track_samples (list):
A list of track samples in the form of :class:`upload_pb2.TrackSample`.
Use :meth:`Sample.generate_sample` to generate
a track sample from an audio file.
"""
endpoint = 'sample'
method = 'POST'
request_type = upload_pb2.UploadSampleRequest
uploader_id = attrib()
track_samples = attrib()
def __attrs_post_init__(self):
super().__attrs_post_init__(self.uploader_id)
self._data.track_sample.extend(self.track_samples)
# TODO: album art documentation.
# TODO: Improved album art API?
[docs] @staticmethod
def generate_sample(
song, track, sample_request, *, external_art=None, no_sample=False
):
"""Generate a track sample from an audio file.
Parameters:
track (locker_pb2.Track):
A locker track of the audio file as created by :meth:`Metadata.get_track_info`.
sample_request (upload_pb2.SignedChallengeInfo):
The ``'signed_challenge_info'`` portion for the
audio file from the :class:`Metadata` response.
external_art(bytes, Optional):
The binary data of an external album art image.
If not provided, embedded album art will be used, if present.
no_sample(bool, Optional):
Don't generate an audio sample from song;
send empty audio sample.
Default: Create an audio sample using ffmpeg/avconv.
"""
track_sample = upload_pb2.TrackSample()
track_sample.track.CopyFrom(track)
track_sample.signed_challenge_info.CopyFrom(sample_request)
try:
if no_sample:
track_sample.sample = b''
else:
track_sample.sample = transcode_to_mp3(
song,
slice_start=sample_request.challenge_info.start_millis // 1000,
slice_duration=sample_request.challenge_info.duration_millis // 1000,
quality='128k',
)
album_art = external_art or get_album_art(song)
if album_art:
album_art_image = upload_pb2.ImageUnion()
album_art_image.user_album_art = album_art
track_sample.user_album_art.CopyFrom(album_art_image)
except (OSError, ValueError, subprocess.CalledProcessError):
raise
return track_sample
# TODO: Album art.
# TODO: contentType for title and external.
# TODO: FLAC seems to be put as 'audio/mpeg'?
# TODO: XingHeaderLength.
# TODO: AlbumArtLength/AlbumArtStart.
[docs]@attrs(slots=True)
class ScottyAgentPost(JSONCall):
"""Request an upload URL for a track from Google Music.
Parameters:
uploader_id (str):
A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
server_track_id (str):
The server ID of the audio file to upload as given
in the response of :class:`Metadata` or :class:`Sample`.
track (locker_pb2.Track):
A locker track of the audio file as created by :meth:`Metadata.get_track_info`.
song (os.PathLike or str or audio_metadata.Format):
The path to an audio file or an instance of :class:`audio_metadata.Format`.
external_art(bytes, Optional):
The binary data of an external album art image.
If not provided, embedded album art will be used, if present.
total_song_count (int, Optional):
Total number of songs to be uploaded in this session.
Default: 1
total_uploaded_count (int, Optional):
Number of songs uploaded in this session.
Default: 0
"""
base_url = 'https://uploadsj.clients.google.com/uploadsj/scottyagent'
method = 'POST'
uploader_id = attrib()
server_track_id = attrib()
track = attrib()
song = attrib()
external_art = attrib(default=None)
total_song_count = attrib(default=1)
total_uploaded_count = attrib(default=0)
def __attrs_post_init__(self):
super().__attrs_post_init__()
inlined = {
'title': 'jumper-uploader-title-42',
'ClientId': self.track.client_id,
'ClientTotalSongCount': str(self.total_song_count),
'CurrentTotalUploadedCount': str(self.total_uploaded_count),
'CurrentUploadingTrackArtist': self.track.artist,
'CurrentUploadingTrack': self.track.title,
'ServerId': self.server_track_id,
'SyncNow': 'true',
'TrackBitRate': str(self.track.original_bit_rate),
'TrackDoNotRematch': 'false',
'UploaderId': self.uploader_id,
}
if not isinstance(self.song, audio_metadata.Format):
self.song = audio_metadata.load(self.song)
album_art = self.external_art or get_album_art(self.song)
if album_art:
inlined['AlbumArt'] = b64encode(album_art).decode()
self._data.update(
{
'clientId': 'Jumper Uploader',
'createSessionRequest': {
'fields': [
{
'external': {
'filename': os.path.basename(self.song.filepath),
'name': os.path.abspath(self.song.filepath),
'put': {},
# Size seems to be sent when uploading MP3, but not FLAC.
# In fact, uploading FLAC directly fails when this is given.
# Leaving it out works for everything.
# 'size': self.track.estimated_size
}
}
]
},
'protocolVersion': '0.8',
}
)
for field, value in inlined.items():
self._data['createSessionRequest']['fields'].append(
{
'inlined': {
'content': value,
'name': field,
}
}
)
self._url = ScottyAgentPost.base_url
[docs]@attrs(slots=True)
class ScottyAgentPut(Call):
"""Upload a file to a Google Music library.
Parameters:
upload_url (str):
The upload URL given by :class:`ScottyAgentPost` response.
audio_file (os.PathLike or str or bytes):
An audio file as :class:`os.PathLike`,
a file/bytes-like object, or binary data.
content_type (str):
The mime type to be sent in the ContentType header field.
Default: ``'audio/mpeg'``
"""
method = 'PUT'
upload_url = attrib()
audio_file = attrib()
content_type = attrib(default='audio/mpeg')
def __attrs_post_init__(self):
if hasattr(self.audio_file, 'read'):
self._data = self.audio_file.read()
elif isinstance(self.audio_file, (os.PathLike, str)):
with open(self.audio_file, 'rb') as f:
self._data = f.read()
elif isinstance(self.audio_file, bytes):
self._data = self.audio_file
else:
raise ValueError(
"'audio_file' must be os.PathLike, filepath string, a file/bytes-like object, or binary data."
)
if len(self._data) >= 300 * 1024 * 1024:
raise ValueError("Maximum allowed file size is 300 MiB.")
self._headers.update({'ContentType': self.content_type})
self._url = self.upload_url
parse_response = JSONCall.parse_response
[docs]@attrs(slots=True)
class UpAuth(MusicManagerCall):
"""Authenticate device as a Music Manager uploader.
Parameters:
uploader_id (str):
A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
uploader_name (str):
The name given to the device in the Google Music device listing.
"""
endpoint = 'upauth'
method = 'POST'
request_type = upload_pb2.UpAuthRequest
uploader_id = attrib()
uploader_name = attrib()
def __attrs_post_init__(self):
super().__attrs_post_init__(self.uploader_id)
self._data.friendly_name = self.uploader_name
@staticmethod
def check_success(response):
return (
response.HasField('auth_status')
and response.auth_status == upload_pb2.UploadResponse.OK
)
[docs]@attrs(slots=True)
class UploadState(MusicManagerCall):
"""Notify Google Music of the state of an upload.
Parameters:
uploader_id (str): A unique ID given as a MAC address.
Only one Music Manager client may use a given uploader ID.
state (str): Can be one of ``'START'``, ``'PAUSED'``, ``'STOPPED'``.
Will be uppercased if lowercase is given.
"""
endpoint = 'uploadstate'
method = 'POST'
request_type = upload_pb2.UpdateUploadStateRequest
uploader_id = attrib()
state = attrib()
def __attrs_post_init__(self):
super().__attrs_post_init__(self.uploader_id)
state = self.state.upper()
try:
self._data.state = getattr(upload_pb2.UpdateUploadStateRequest, state)
except AttributeError as e:
raise ValueError from e