music-assistant-server

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