SpotiClub Patch v0.2.0
Some checks failed
CodeQL / Analyze (python) (push) Has been cancelled

This commit is contained in:
unknown
2025-12-18 03:13:50 +01:00
parent 6c05cf4915
commit 8a7d0fa3c8
2 changed files with 54 additions and 8 deletions

View File

@@ -600,12 +600,29 @@ class ApResolver:
"""
response = requests.get("{}?type={}".format(ApResolver.base_url,
service_type))
# If ApResolve responds with a non-200, treat this as a clear,
# high-level error instead of bubbling up JSON parsing
# exceptions from HTML error pages.
if response.status_code != 200:
if response.status_code == 502:
raise RuntimeError(
f"ApResolve request failed with the following return value: {response.content}. Servers might be down!"
"Failed to contact Spotify ApResolve (502). "
"Servers might be down or unreachable."
)
return response.json()
raise RuntimeError(
f"Failed to contact Spotify ApResolve (status {response.status_code}). "
"This is usually a network, DNS, or firewall issue."
)
try:
return response.json()
except ValueError as exc:
# Response wasn't valid JSON; surface a friendly error
# instead of a long JSONDecodeError traceback.
raise RuntimeError(
"Spotify ApResolve returned invalid data. "
"This is likely a temporary server or network problem."
) from exc
@staticmethod
def get_random_of(service_type: str) -> str:
@@ -1290,6 +1307,18 @@ class Session(Closeable, MessageListener, SubListener):
for attempt in range(1, max_attempts + 1):
try:
# Inform the user about each connection attempt so it is
# visible in the console (e.g. when called from Zotify).
# Only show attempt counters after the first failure; the
# initial attempt is shown without numbering.
if attempt == 1:
connect_msg = "Connecting to Spotify..."
else:
connect_msg = (
f"Connecting to Spotify (attempt {attempt}/{max_attempts})..."
)
self.logger.info(connect_msg)
print(connect_msg)
acc = Session.Accumulator()
# Send ClientHello
nonce = Random.get_random_bytes(0x10)
@@ -1414,6 +1443,12 @@ class Session(Closeable, MessageListener, SubListener):
max_attempts,
exc,
)
if attempt == 1:
print(f"Connecting to Spotify failed: {exc}")
else:
print(
f"Connecting to Spotify (attempt {attempt}/{max_attempts}) failed: {exc}"
)
# Close current connection; a new access point will be
# selected on the next attempt.
if self.connection is not None:
@@ -1430,6 +1465,10 @@ class Session(Closeable, MessageListener, SubListener):
self.logger.info(
"Retrying connection, new access point: %s", address
)
print(
"Retrying connection to Spotify with new access point: "
f"{address} (next attempt {attempt + 1}/{max_attempts})"
)
self.connection = Session.ConnectionHolder.create(
address, None
)
@@ -1742,9 +1781,16 @@ class Session(Closeable, MessageListener, SubListener):
else:
self.logger.warning("Login5 authentication failed: {}".format(login5_response.error))
else:
self.logger.warning("Login5 request failed with status: {}".format(response.status_code))
# Login5 is best-effort; if it fails (e.g. 403 or
# region restrictions), we silently skip it to
# avoid confusing the user when the main
# connection has already failed.
self.logger.debug(
"Login5 request failed with status: %s", response.status_code
)
except Exception as e:
self.logger.warning("Failed to authenticate with Login5: {}".format(e))
# Also treat unexpected Login5 issues as debug-only noise.
self.logger.debug("Failed to authenticate with Login5: %s", e)
def get_login5_token(self) -> typing.Union[str, None]:
"""Get the Login5 access token if available and not expired"""