Create zotify.py

zotify/zotify.py

@@ -0,0 +1,110 @@
+import json
+from pathlib import Path
+from pwinput import pwinput
+import time
+import requests
+from librespot.audio.decoders import VorbisOnlyAudioQuality
+from librespot.core import Session
+
+from zotify.const import TYPE, \
+    PREMIUM, USER_READ_EMAIL, OFFSET, LIMIT, \
+    PLAYLIST_READ_PRIVATE, USER_LIBRARY_READ, USER_FOLLOW_READ
+from zotify.config import Config
+
+class Zotify:    
+    SESSION: Session = None
+    DOWNLOAD_QUALITY = None
+    CONFIG: Config = Config()
+
+    def __init__(self, args):
+        Zotify.CONFIG.load(args)
+        Zotify.login(args)
+
+    @classmethod
+    def login(cls, args):
+        """ Authenticates with Spotify and saves credentials to a file """
+
+        cred_location = Config.get_credentials_location()
+
+        if Path(cred_location).is_file():
+            try:
+                conf = Session.Configuration.Builder().set_store_credentials(False).build()
+                cls.SESSION = Session.Builder(conf).stored_file(cred_location).create()
+                return
+            except RuntimeError:
+                pass
+        while True:
+            user_name = args.username if args.username else ''
+            while len(user_name) == 0:
+                user_name = input('Username: ')
+            password = args.password if args.password else pwinput(prompt='Password: ', mask='*')
+            try:
+                if Config.get_save_credentials():
+                    conf = Session.Configuration.Builder().set_stored_credential_file(cred_location).build()
+                else:
+                    conf = Session.Configuration.Builder().set_store_credentials(False).build()
+                cls.SESSION = Session.Builder(conf).user_pass(user_name, password).create()
+                return
+            except RuntimeError:
+                pass
+
+    @classmethod
+    def get_content_stream(cls, content_id, quality):
+        return cls.SESSION.content_feeder().load(content_id, VorbisOnlyAudioQuality(quality), False, None)
+
+    @classmethod
+    def __get_auth_token(cls):
+        return cls.SESSION.tokens().get_token(
+            USER_READ_EMAIL, PLAYLIST_READ_PRIVATE, USER_LIBRARY_READ, USER_FOLLOW_READ
+        ).access_token
+
+    @classmethod
+    def get_auth_header(cls):
+        return {
+            'Authorization': f'Bearer {cls.__get_auth_token()}',
+            'Accept-Language': f'{cls.CONFIG.get_language()}',
+            'Accept': 'application/json',
+            'app-platform': 'WebPlayer'
+        }
+
+    @classmethod
+    def get_auth_header_and_params(cls, limit, offset):
+        return {
+            'Authorization': f'Bearer {cls.__get_auth_token()}',
+            'Accept-Language': f'{cls.CONFIG.get_language()}',
+            'Accept': 'application/json',
+            'app-platform': 'WebPlayer'
+        }, {LIMIT: limit, OFFSET: offset}
+
+    @classmethod
+    def invoke_url_with_params(cls, url, limit, offset, **kwargs):
+        headers, params = cls.get_auth_header_and_params(limit=limit, offset=offset)
+        params.update(kwargs)
+        return requests.get(url, headers=headers, params=params).json()
+
+    @classmethod
+    def invoke_url(cls, url, tryCount=0):
+        # we need to import that here, otherwise we will get circular imports!
+        from zotify.termoutput import Printer, PrintChannel
+        headers = cls.get_auth_header()
+        response = requests.get(url, headers=headers)
+        responsetext = response.text
+        try:
+            responsejson = response.json()
+        except json.decoder.JSONDecodeError:
+            responsejson = {"error": {"status": "unknown", "message": "received an empty response"}}
+
+        if not responsejson or 'error' in responsejson:
+            if tryCount < (cls.CONFIG.get_retry_attempts() - 1):
+                Printer.print(PrintChannel.WARNINGS, f"Spotify API Error (try {tryCount + 1}) ({responsejson['error']['status']}): {responsejson['error']['message']}")
+                time.sleep(5)
+                return cls.invoke_url(url, tryCount + 1)
+
+            Printer.print(PrintChannel.API_ERRORS, f"Spotify API Error ({responsejson['error']['status']}): {responsejson['error']['message']}")
+
+        return responsetext, responsejson
+
+    @classmethod
+    def check_premium(cls) -> bool:
+        """ If user has spotify premium return true """
+        return (cls.SESSION.get_user_attribute(TYPE) == PREMIUM)