Scrameupeutchi/zotify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Prepare V0.2

Browse Source
This commit is contained in:
unknown
2025-12-17 20:03:46 +01:00
parent cc56e1aa8e
commit b2d05fa051
2 changed files with 106 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
librespot/audio/__init__.py
View File

@@ -7,17 +7,7 @@ from librespot.cache import CacheManager
from librespot.crypto import Packet
from librespot.metadata import EpisodeId, PlayableId, TrackId
from librespot.proto import Metadata_pb2 as Metadata, StorageResolve_pb2 as StorageResolve
from librespot.structure import (
AudioDecrypt,
AudioQualityPicker,
Closeable,
FeederException,
GeneralAudioStream,
GeneralWritableStream,
HaltListener,
NoopAudioDecrypt,
PacketsReceiver,
)
from librespot.structure import GeneralAudioStream, AudioDecrypt, AudioQualityPicker, Closeable, FeederException, GeneralAudioStream, GeneralWritableStream, HaltListener, NoopAudioDecrypt, PacketsReceiver
from pathlib import Path
import concurrent.futures
import io
@@ -37,36 +27,41 @@ if typing.TYPE_CHECKING:
from librespot.core import Session
"""
PATCH : SpotiClub Audio Key Fetching
PATCH : SpotiClub Audio Key Fetching (v0.2.0)
Fetches the audio decryption key from the SpotiClub Audio Key API instead of Spotify directly.
This is a workaround for Spotify's tightened restrictions on Audio Key access (only Premium Tier now).
This is a workaround for Spotify's tightened restrictions on Audio Key access (they allow only Premium Tier now).
There are 3 importants parameters to provide, and one is already filled in:
If you are using our fork, there is no reason for you to complete this section, as upon first run, Zotify will ask you for the logins and save them for future use.
But if needed somehow or by using this single patch file, there are 3 importants parameters to provide, and one is already filled in:
- server_url: The URL of the SpotiClub Audio Key API endpoint. You should not need to change this, except if a dev instructs you to do so.
- spoticlub_user : Your SpotiClub FTP username. You can get this by using our Padoru Asssistant once.
- spoticlub_password : Your SpotiClub FTP password, also obtainable via the Padoru Assistant.
Alternatively, you can use the file `spoticlub_credentials.json` in the Zotify project root with the following structure:
{
"spoticlub_user": "spoticlub-...",
"spoticlub_password": "..."
}
Using the fork's assistant is the recommended way to get register your credentials, as overwriting this file during the beta phase will need you to put them here over and over again.
"""
##### WRITE YOUR LOGINS DOWN HERE #####
#######################################
server_url = "http://api.spoticlub.zip:4277/get_audio_key"
spoticlub_user = ""
spoticlub_password = ""
spoticlub_user = "anonymous"
spoticlub_password = "IfWeFeelLikeEnablingThis"
########################################
##### END OF USER INPUT AREA ###########
SPOTICLUB_DISABLE = os.getenv("SPOTICLUB_DISABLE", "").lower() in {"1", "true", "yes"}
SPOTICLUB_FALLBACK_TO_SPOTIFY = os.getenv("SPOTICLUB_FALLBACK_SPOTIFY", "1").lower() not in {
"0",
"false",
"no",
}
### SPOTICLUB CLIENT SERIAL TRACKING (DO NOT EDIT) ###
spoticlub_client_serial: typing.Optional[str] = None
spoticlub_loaded_logged: bool = False
########################################
class LoadedStream(GeneralAudioStream):
def __init__(self, data: bytes):
super().__init__()
self._buffer = io.BytesIO(data)
def read(self, n: int = -1) -> bytes:
return self._buffer.read(n)
def close(self) -> None:
self._buffer.close()
class AbsChunkedInputStream(io.BytesIO, HaltListener):
chunk_exception = None
@@ -275,6 +270,7 @@ class AudioKeyManager(PacketsReceiver, Closeable):
__seq_holder_lock = threading.Condition()
__session: Session
__zero_short = b"\x00\x00"
_spoticlub_current_country: typing.Optional[str] = None
def __init__(self, session: Session):
self.__session = session
@@ -298,88 +294,47 @@ class AudioKeyManager(PacketsReceiver, Closeable):
"Couldn't handle packet, cmd: {}, length: {}".format(
packet.cmd, len(packet.payload)))
def get_audio_key(
self,
def get_audio_key(self,
gid: bytes,
file_id: bytes,
retry: bool = True,
) -> bytes:
"""Retrieve an audio key via SpotiClub API or fall back to Spotify."""
if self._spoticlub_available():
try:
return self._get_audio_key_via_spoticlub(gid, file_id, retry)
except Exception as exc: # noqa: BLE001
self.logger.warning(
"SpotiClub API failed for gid=%s file_id=%s: %s",
util.bytes_to_hex(gid),
util.bytes_to_hex(file_id),
exc,
)
if not SPOTICLUB_FALLBACK_TO_SPOTIFY:
raise
self.logger.info("Falling back to Spotify audio-key pipeline")
return self._get_audio_key_via_spotify(gid, file_id, retry)
def _spoticlub_available(self) -> bool:
if SPOTICLUB_DISABLE or not server_url:
return False
return self._ensure_spoticlub_credentials()
def _ensure_spoticlub_credentials(self) -> bool:
global spoticlub_user, spoticlub_password
if spoticlub_user and spoticlub_password:
return True
retry: bool = True) -> bytes:
global spoticlub_user, spoticlub_password, spoticlub_client_serial, spoticlub_loaded_logged
if not spoticlub_user or not spoticlub_password or spoticlub_user == "anonymous":
try:
# To verify : Do all forks look for the same path ?
cfg_path = Path.home() / "AppData\\Roaming\\Zotify\\spoticlub_credentials.json"
print(f"\n[SpotiClub API] Looking for credentials file at: {cfg_path}")
if cfg_path.is_file():
print("[SpotiClub API] Found credentials file")
print(f"\n[SpotiClub API] Loading credentials...")
with open(cfg_path, "r", encoding="utf-8") as f:
cfg = json.load(f)
spoticlub_user = spoticlub_user or cfg.get("spoticlub_user")
spoticlub_password = spoticlub_password or cfg.get("spoticlub_password")
spoticlub_user = cfg.get("spoticlub_user")
spoticlub_password = cfg.get("spoticlub_password")
else:
print(
"[SpotiClub API] Credentials file NOT found at: %s. Will proceed with hardcoded credentials if any.\n"
% cfg_path
)
except Exception as exc: # noqa: BLE001
print(f"[SpotiClub API] Credentials file NOT found at: {cfg_path}. We will proceed with hardcoded credentials if any...\n")
except Exception as exc:
print(f"[SpotiClub API] Error while loading credentials file: {exc}\n")
has_credentials = bool(spoticlub_user and spoticlub_password)
if not has_credentials:
if not spoticlub_user or not spoticlub_password or not server_url:
cfg_path = Path.home() / "AppData\\Roaming\\Zotify\\spoticlub_credentials.json"
msg = (
"Missing SpotiClub credentials: please set spoticlub_user & spoticlub_password inside the top of the __init__.py file,"
" or use the spoticlub_credentials.json in your Zotify config folder [C:\\Users\\USERNAME\\AppData\\Roaming\\Zotify\\]."
"Missing SpotiClub credentials: please set the appropriates values inside your spoticlub_credentials.json,"
f"located in the Zotify config folder [{cfg_path}] (Or delete it and restart Zotify to be prompted for credentials)."
)
if SPOTICLUB_FALLBACK_TO_SPOTIFY:
self.logger.warning("%s Falling back to Spotify audio keys.", msg)
return False
print(f"[SpotiClub API][FATAL] {msg}")
print(f"[SpotiClub API][ERROR]\n{msg}")
raise SystemExit(1)
return True
def _get_audio_key_via_spoticlub(self, gid: bytes, file_id: bytes, retry: bool) -> bytes:
global spoticlub_user, spoticlub_password
if not spoticlub_loaded_logged:
spoticlub_loaded_logged = True
print(f"\n[SpotiClub API] Plugin Loaded! Welcome {spoticlub_user}\n")
payload = {
"gid": util.bytes_to_hex(gid),
"file_id": util.bytes_to_hex(file_id),
"ftpUser": spoticlub_user,
"user": spoticlub_user,
"password": spoticlub_password,
}
if not server_url:
msg = (
"The SpotiClub server_url is not set! It should be defined at the top of __init__.py or via SPOTICLUB_SERVER_URL."
)
print(f"[SpotiClub API][FATAL] {msg}")
raise SystemExit(1)
print(f"[SpotiClub API] Loaded credentials - Welcome {spoticlub_user}\n")
if spoticlub_client_serial:
payload["client_serial"] = spoticlub_client_serial
tries = 0
last_err: typing.Optional[Exception] = None
@@ -387,82 +342,70 @@ class AudioKeyManager(PacketsReceiver, Closeable):
while True:
tries += 1
try:
resp = requests.post(
server_url,
json=payload,
timeout=AudioKeyManager.audio_key_request_timeout,
resp = requests.post(server_url, json=payload, timeout=AudioKeyManager.audio_key_request_timeout)
# If another client instance is already active for this
# SpotiClub user, the server will reply with HTTP 423 and
# instruct this client to wait before retrying.
if resp.status_code == 423:
try:
data = resp.json()
except Exception: # noqa: BLE001
data = {}
retry_after = data.get("retry_after", 60)
if not isinstance(retry_after, (int, float)):
retry_after = 10
print(
f"[SpotiClub API] Another client is already using this account. Waiting {int(retry_after)}s before retrying..."
)
self.logger.info(
"[SpotiClub API] Queued client for user %s; waiting %ds before retry",
spoticlub_user,
int(retry_after),
)
time.sleep(float(retry_after))
# Do NOT count this as a failure towards the max retries.
continue
# Explicit handling for bad logins so we don't just retry.
if resp.status_code == 401:
print(
"[SpotiClub API][BAD_LOGIN] It seems your credentials aren't recognized by the API. Please ensure you have entered them correctly, or contact a DEV if you are absolutely certain of their validity."
)
raise SystemExit(1)
if resp.status_code != 200:
raise RuntimeError(
f"[SpotiClub API] Unexpected response {resp.status_code}: {resp.text}"
)
raise RuntimeError(f"[SpotiClub API] Sorry, the API returned the unexpected code {resp.status_code}: {resp.text}")
data = resp.json()
key_hex = data.get("key")
if not isinstance(key_hex, str):
raise RuntimeError("[SpotiClub API] Response missing 'key'")
raise RuntimeError("[SpotiClub API] Sorry, API response missing 'key'")
country = data.get("country")
if isinstance(country, str):
if AudioKeyManager._spoticlub_current_country != country:
AudioKeyManager._spoticlub_current_country = country
print(f"[SpotiClub API] Received {country} as the download country\n\n")
new_serial = data.get("client_serial")
if isinstance(new_serial, str) and new_serial:
spoticlub_client_serial = new_serial
key_bytes = util.hex_to_bytes(key_hex)
if len(key_bytes) != 16:
raise RuntimeError("[SpotiClub API] Audio key must be 16 bytes long")
raise RuntimeError("[SpotiClub API] Woops, received Audio Key must be 16 bytes long")
return key_bytes
except Exception as exc: # noqa: BLE001
last_err = exc
self.logger.warning(
"[SpotiClub API] Retrying... (try %d/%d): %s",
tries,
3,
exc,
)
self.logger.warning("[SpotiClub API] Retrying the contact... (try %d): %s", tries, exc)
if not retry or tries >= 3:
break
time.sleep(5)
raise RuntimeError(
"Failed fetching Audio Key from API for gid: {}, fileId: {} (last error: {})".format(
util.bytes_to_hex(gid), util.bytes_to_hex(file_id), last_err
)
)
def _get_audio_key_via_spotify(self, gid: bytes, file_id: bytes, retry: bool) -> bytes:
callback = AudioKeyManager.SyncCallback(self)
seq = self._next_sequence()
self.__callbacks[seq] = callback
try:
self._send_audio_key_request(seq, gid, file_id)
key = callback.wait_response()
finally:
self.__callbacks.pop(seq, None)
if key is None:
if retry:
self.logger.warning(
"Spotify audio key request returned no data for gid=%s file_id=%s; retrying once",
util.bytes_to_hex(gid),
util.bytes_to_hex(file_id),
)
time.sleep(0.5)
return self._get_audio_key_via_spotify(gid, file_id, False)
raise RuntimeError(
"Failed retrieving audio key from Spotify for gid: {}, fileId: {}".format(
util.bytes_to_hex(gid), util.bytes_to_hex(file_id)
)
)
return key
def _send_audio_key_request(self, seq: int, gid: bytes, file_id: bytes) -> None:
buffer = io.BytesIO()
buffer.write(struct.pack(">i", seq))
buffer.write(gid)
buffer.write(self.__zero_short)
buffer.write(file_id)
buffer.write(self.__zero_short)
self.__session.send(Packet.Type.aes_key, buffer.getvalue())
def _next_sequence(self) -> int:
with self.__seq_holder_lock:
self.__seq_holder = (self.__seq_holder + 1) & 0x7FFFFFFF
if self.__seq_holder == 0:
self.__seq_holder = 1
return self.__seq_holder
util.bytes_to_hex(gid), util.bytes_to_hex(file_id), last_err))
class Callback:
@@ -492,14 +435,11 @@ class AudioKeyManager(PacketsReceiver, Closeable):
self.__reference.put(None)
self.__reference_lock.notify_all()
def wait_response(self) -> typing.Optional[bytes]:
def wait_response(self) -> bytes:
with self.__reference_lock:
self.__reference_lock.wait(
AudioKeyManager.audio_key_request_timeout)
try:
return self.__reference.get(block=False)
except queue.Empty:
return None
class CdnFeedHelper:
@@ -956,7 +896,7 @@ class PlayableContentFeeder:
def load_episode(self, episode_id: EpisodeId,
audio_quality_picker: AudioQualityPicker, preload: bool,
halt_listener: HaltListener) -> "LoadedStream":
halt_listener: HaltListener) -> LoadedStream:
episode = self.__session.api().get_metadata_4_episode(episode_id)
if episode.external_url:
return CdnFeedHelper.load_episode_external(self.__session, episode,

3
zotify/track.py
View File

@@ -440,7 +440,8 @@ def convert_audio_format(filename) -> None:
if file_codec != 'copy':
bitrate = Zotify.CONFIG.get_transcode_bitrate()
bitrates = {
'auto': '320k' if Zotify.check_premium() else '160k',
#SpotiClub API permit the use of '320k' for free users, so we map 'auto' to that value.
'auto': '320k',
'normal': '96k',
'high': '160k',
'very_high': '320k'