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