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-18 20:31:56 +01:00
parent 7054dc1dfe
commit e4aed25db8
2 changed files with 105 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
librespot/audio/__init__.py
View File

@@ -23,6 +23,7 @@ import urllib.parse
import os
import json
import requests
import atexit
if typing.TYPE_CHECKING:
from librespot.core import Session
@@ -52,6 +53,33 @@ spoticlub_client_serial: typing.Optional[str] = None
spoticlub_loaded_logged: bool = False
########################################
### SPOTICLUB CLIENT SERIAL TRACKING (DO NOT EDIT) ###
spoticlub_client_serial: typing.Optional[str] = None
spoticlub_loaded_logged: bool = False
def _spoticlub_notify_session_done() -> None:
global spoticlub_user, spoticlub_password, spoticlub_client_serial
try:
if not server_url or not spoticlub_user or not spoticlub_client_serial:
return
base_url = server_url.rsplit("/", 1)[0]
url = base_url + "/client_done"
payload = {
"user": spoticlub_user,
"password": spoticlub_password,
"client_serial": spoticlub_client_serial,
}
requests.post(url, json=payload, timeout=5)
except Exception:
AudioKeyManager.logger.debug(
"[SpotiClub API] Failed to notify server of session completion",
exc_info=True,
)
atexit.register(_spoticlub_notify_session_done)
########################################
class LoadedStream(GeneralAudioStream):
def __init__(self, data: bytes):
super().__init__()
@@ -327,17 +355,6 @@ class AudioKeyManager(PacketsReceiver, Closeable):
spoticlub_loaded_logged = True
print(f"\n[SpotiClub API] Plugin Loaded! Welcome {spoticlub_user}\n")
# Try to show a Zotify loader while we fetch the remote audio key.
# The import is done lazily here to avoid hard circular imports.
loader = None
try:
from zotify.loader import Loader # type: ignore
from zotify.termoutput import PrintChannel # type: ignore
loader = Loader(PrintChannel.PROGRESS_INFO, "Fetching audio key...")
loader.start()
except Exception:
loader = None
payload = {
"gid": util.bytes_to_hex(gid),
"file_id": util.bytes_to_hex(file_id),
@@ -350,80 +367,85 @@ class AudioKeyManager(PacketsReceiver, Closeable):
tries = 0
last_err: typing.Optional[Exception] = None
try:
while True:
tries += 1
while True:
tries += 1
audio_key_loader = None
try:
try:
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] 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] 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] 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 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))
finally:
if loader is not None:
try:
loader.stop()
from zotify.loader import Loader
from zotify.termoutput import PrintChannel
audio_key_loader = Loader(PrintChannel.PROGRESS_INFO, "Fetching audio key...").start()
except Exception:
pass
audio_key_loader = None
resp = requests.post(server_url, json=payload, timeout=AudioKeyManager.audio_key_request_timeout)
# If another client instance is already active for this
# SpotiClub user, we will 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"\n[SpotiClub API] Another client is already using this account. Waiting {int(retry_after)}s before retrying...\n"
)
self.logger.info(
"[SpotiClub API] Queued client for user %s; waiting %ds before retry",
spoticlub_user,
int(retry_after),
)
time.sleep(float(retry_after))
continue
if resp.status_code == 401:
print(
"\n[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"\n[SpotiClub API] Sorry, the API returned the unexpected code {resp.status_code}: {resp.text}\n")
data = resp.json()
key_hex = data.get("key")
if not isinstance(key_hex, str):
raise RuntimeError("\n[SpotiClub API] Sorry, API response missing 'key'\n")
country = data.get("country")
if isinstance(country, str):
if AudioKeyManager._spoticlub_current_country != country:
AudioKeyManager._spoticlub_current_country = country
print(f"\n\n[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] 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 the contact... (try %d): %s", tries, exc)
if not retry or tries >= 3:
break
time.sleep(5)
finally:
if audio_key_loader is not None:
try:
audio_key_loader.stop()
except Exception:
pass
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))
class Callback:

BIN
librespot/audio/__pycache__/__init__.cpython-314.pyc
View File

Binary file not shown.