/
/
/
1"""Helper utilities for the Pandora provider."""
2
3from __future__ import annotations
4
5import secrets
6from typing import Any
7
8import aiohttp
9from music_assistant_models.errors import (
10 InvalidDataError,
11 LoginFailed,
12 MediaNotFoundError,
13 ProviderUnavailableError,
14 ResourceTemporarilyUnavailable,
15)
16
17from .constants import AUTH_ERRORS, NOT_FOUND_ERRORS, UNAVAILABLE_ERRORS
18
19
20def generate_csrf_token() -> str:
21 """Generate a random CSRF token."""
22 return secrets.token_hex(16)
23
24
25def handle_pandora_error(response_data: dict[str, Any]) -> None:
26 """
27 Handle Pandora API error responses.
28
29 Maps Pandora API error codes to appropriate Music Assistant exceptions.
30
31 Raises:
32 LoginFailed: For authentication errors
33 MediaNotFoundError: For missing stations/tracks
34 ResourceTemporarilyUnavailable: For service availability issues
35 InvalidDataError: For other API errors
36 """
37 if (error_code := response_data.get("errorCode")) is None:
38 return
39
40 message = response_data.get("message", response_data.get("errorString", "Unknown error"))
41
42 # Use the categorized sets for cleaner logic
43 if error_code in AUTH_ERRORS:
44 raise LoginFailed(f"Authentication failed: {message}")
45
46 if error_code in NOT_FOUND_ERRORS:
47 raise MediaNotFoundError(f"The requested resource was not found: {message}")
48
49 if error_code in UNAVAILABLE_ERRORS:
50 raise ResourceTemporarilyUnavailable(f"Pandora service issue: {message}")
51
52 # Fallback for any other API error
53 raise InvalidDataError(f"Pandora API Error [{error_code}]: {message}")
54
55
56async def get_csrf_token(session: aiohttp.ClientSession) -> str:
57 """
58 Get CSRF token from Pandora website.
59
60 Attempts to retrieve CSRF token from Pandora cookies.
61
62 Args:
63 session: aiohttp client session
64
65 Returns:
66 CSRF token string
67
68 Raises:
69 ProviderUnavailableError: If network request fails
70 ResourceTemporarilyUnavailable: If no token available
71 """
72 try:
73 # Use a more specific timeout for this initial handshake
74 async with session.head(
75 "https://www.pandora.com/",
76 timeout=aiohttp.ClientTimeout(total=10),
77 ) as response:
78 if "csrftoken" in response.cookies:
79 return str(response.cookies["csrftoken"].value)
80 except aiohttp.ClientError as err:
81 # Catch network issues at the source and wrap in MA error
82 raise ProviderUnavailableError(f"Network error while reaching Pandora: {err}") from err
83
84 raise ResourceTemporarilyUnavailable("Pandora web session failed to provide a CSRF token.")
85
86
87def create_auth_headers(csrf_token: str, auth_token: str | None = None) -> dict[str, str]:
88 """
89 Create authentication headers for Pandora API requests.
90
91 Args:
92 csrf_token: CSRF token for request validation
93 auth_token: Optional authentication token for authenticated requests
94
95 Returns:
96 Dictionary of HTTP headers
97 """
98 headers = {
99 "Content-Type": "application/json;charset=utf-8",
100 "X-CsrfToken": csrf_token,
101 "Cookie": f"csrftoken={csrf_token}",
102 "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:148.0) Gecko/20100101 Firefox/148.0",
103 "Accept": "application/json, text/plain, */*",
104 "Accept-Language": "en-US,en;q=0.9",
105 "Origin": "https://www.pandora.com",
106 "Referer": "https://www.pandora.com/",
107 }
108
109 if auth_token:
110 headers["X-AuthToken"] = auth_token
111
112 return headers
113