Prepare V0.2

librespot/audio/__init__.py

@@ -322,10 +322,83 @@ class AudioKeyManager(PacketsReceiver, Closeable):
                 "Couldn't handle packet, cmd: {}, length: {}".format(
                     packet.cmd, len(packet.payload)))
 
+    def _is_premium_user(self) -> bool:
+        """Best-effort check whether the current Spotify account is Premium.
+
+        We rely on Session user attributes populated from the AP product_info packet.
+        Historically, the attribute named "type" contains values like "premium", "free",
+        "open", etc.
+        """
+        try:
+            raw = (
+                self.__session.get_user_attribute("type")
+                or self.__session.get_user_attribute("product")
+                or self.__session.get_user_attribute("product_type")
+                or self.__session.get_user_attribute("subscription")
+                or self.__session.get_user_attribute("account_type")
+                or ""
+            )
+            product = str(raw).strip().lower()
+            if not product:
+                return False
+
+            if "premium" in product:
+                return True
+
+            # Legacy/alt plan names occasionally seen.
+            return product in {"unlimited"}
+        except Exception:
+            return False
+
+    def _get_spotify_audio_key(self, gid: bytes, file_id: bytes, retry: bool = True) -> bytes:
+        """Request the audio key directly from Spotify (Premium accounts)."""
+        tries = 0
+        last_err: typing.Optional[Exception] = None
+
+        while True:
+            tries += 1
+            callback = AudioKeyManager.SyncCallback(self)
+
+            with self.__seq_holder_lock:
+                AudioKeyManager.__seq_holder += 1
+                seq = AudioKeyManager.__seq_holder
+                AudioKeyManager.__callbacks[seq] = callback
+
+            try:
+                payload = struct.pack(">i", seq) + gid + file_id
+                self.__session.send(Packet.Type.request_key, payload)
+
+                key = callback.wait_response()
+                if key is None:
+                    raise RuntimeError("Audio key request failed")
+                return key
+            except Exception as exc:  # noqa: BLE001
+                last_err = exc
+                self.logger.warning("Spotify audio key request failed (try %d): %s", tries, exc)
+                if not retry or tries >= 3:
+                    break
+                time.sleep(1)
+            finally:
+                try:
+                    AudioKeyManager.__callbacks.pop(seq, None)
+                except Exception:
+                    pass
+
+        raise RuntimeError(
+            "Failed fetching Audio Key from Spotify for gid: {}, fileId: {} (last error: {})".format(
+                util.bytes_to_hex(gid), util.bytes_to_hex(file_id), last_err
+            )
+        )
+
     def get_audio_key(self,
                       gid: bytes,
                       file_id: bytes,
                       retry: bool = True) -> bytes:
+        # If the user is Premium, Spotify will return audio keys directly.
+        # In that case, do not use the SpotiClub API.
+        if self._is_premium_user():
+            return self._get_spotify_audio_key(gid, file_id, retry=retry)
+
         global spoticlub_user, spoticlub_password, spoticlub_client_serial, spoticlub_loaded_logged
         if not spoticlub_user or not spoticlub_password or spoticlub_user == "anonymous":
             try: