Add Pandora provider

music_assistant/providers/pandora/__init__.py

@@ -0,0 +1,73 @@
+"""Pandora music provider support for Music Assistant."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from music_assistant_models.config_entries import ConfigEntry, ConfigValueType
+from music_assistant_models.enums import ConfigEntryType, ProviderFeature
+from music_assistant_models.errors import SetupFailedError
+
+from music_assistant.constants import CONF_PASSWORD, CONF_USERNAME
+
+from .provider import PandoraProvider
+
+if TYPE_CHECKING:
+    from music_assistant_models.config_entries import ProviderConfig
+    from music_assistant_models.provider import ProviderManifest
+
+    from music_assistant import MusicAssistant
+    from music_assistant.models import ProviderInstanceType
+
+# Supported Features - Pandora is primarily a radio service
+SUPPORTED_FEATURES = {
+    ProviderFeature.BROWSE,
+    ProviderFeature.LIBRARY_RADIOS,
+}
+
+
+async def setup(
+    mass: MusicAssistant, manifest: ProviderManifest, config: ProviderConfig
+) -> ProviderInstanceType:
+    """Initialize provider instance with given configuration."""
+    username = config.get_value(CONF_USERNAME)
+    password = config.get_value(CONF_PASSWORD)
+
+    # Type-safe validation
+    if (
+        not username
+        or not password
+        or not isinstance(username, str)
+        or not isinstance(password, str)
+        or not username.strip()
+        or not password.strip()
+    ):
+        raise SetupFailedError("Username and password are required")
+
+    return PandoraProvider(mass, manifest, config, SUPPORTED_FEATURES)
+
+
+async def get_config_entries(
+    mass: MusicAssistant,
+    instance_id: str | None = None,
+    action: str | None = None,
+    values: dict[str, ConfigValueType] | None = None,
+) -> tuple[ConfigEntry, ...]:
+    """Return configuration entries for this provider."""
+    # ruff: noqa: ARG001
+    return (
+        ConfigEntry(
+            key=CONF_USERNAME,
+            type=ConfigEntryType.STRING,
+            label="Username",
+            description="Your Pandora username or email address",
+            required=True,
+        ),
+        ConfigEntry(
+            key=CONF_PASSWORD,
+            type=ConfigEntryType.SECURE_STRING,
+            label="Password",
+            description="Your Pandora password",
+            required=True,
+        ),
+    )

music_assistant/providers/pandora/constants.py

@@ -0,0 +1,43 @@
+"""Constants for the Pandora provider."""
+
+from __future__ import annotations
+
+# API Endpoints
+API_BASE_URL = "https://www.pandora.com/api/v1"
+LOGIN_ENDPOINT = f"{API_BASE_URL}/auth/login"
+STATIONS_ENDPOINT = f"{API_BASE_URL}/station/getStations"
+PLAYLIST_FRAGMENT_ENDPOINT = f"{API_BASE_URL}/playlist/getFragment"
+
+# Error Codes from Pandora API
+PANDORA_ERROR_CODES = {
+    0: "INVALID_REQUEST",
+    1: "INVALID_PARTNER",
+    2: "LISTENER_NOT_AUTHORIZED",
+    3: "USER_NOT_AUTHORIZED",
+    4: "STATION_DOES_NOT_EXIST",
+    5: "TRACK_NOT_FOUND",
+    9: "PANDORA_NOT_AVAILABLE",
+    10: "SYSTEM_NOT_AVAILABLE",
+    11: "CALL_NOT_ALLOWED",
+    12: "INVALID_USERNAME",
+    13: "INVALID_PASSWORD",
+    14: "DEVICE_NOT_FOUND",
+    15: "PARTNER_NOT_AUTHORIZED",
+    1000: "READ_ONLY_MODE",
+    1001: "INVALID_AUTH_TOKEN",
+    1002: "INVALID_LOGIN",
+    1003: "LISTENER_NOT_AUTHORIZED",
+    1004: "USER_ALREADY_EXISTS",
+    1005: "DEVICE_ALREADY_ASSOCIATED_TO_ACCOUNT",
+    1006: "UPGRADE_DEVICE_MODEL_INVALID",
+    1009: "DEVICE_MODEL_INVALID",
+    1010: "INVALID_SPONSOR",
+    1018: "EXPLICIT_PIN_INCORRECT",
+    1020: "EXPLICIT_PIN_MALFORMED",
+    1023: "DEVICE_DISABLED",
+    1024: "DAILY_TRIAL_LIMIT_REACHED",
+    1025: "INVALID_SPONSOR_USERNAME",
+    1026: "SPONSOR_CANNOT_SKIP_ADS",
+    1027: "INSUFFICIENT_CONNECTIVITY",
+    1034: "GEOLOCATION_REQUIRED",
+}

music_assistant/providers/pandora/helpers.py

@@ -0,0 +1,105 @@
+"""Helper utilities for the Pandora provider."""
+
+from __future__ import annotations
+
+import secrets
+from typing import Any
+
+import aiohttp
+from music_assistant_models.errors import (
+    LoginFailed,
+    MediaNotFoundError,
+    ProviderUnavailableError,
+    ResourceTemporarilyUnavailable,
+)
+
+from .constants import PANDORA_ERROR_CODES
+
+
+def generate_csrf_token() -> str:
+    """Generate a random CSRF token."""
+    return secrets.token_hex(16)
+
+
+def handle_pandora_error(response_data: dict[str, Any]) -> None:
+    """Handle Pandora API error responses.
+
+    Maps Pandora API error codes to appropriate Music Assistant exceptions.
+
+    Raises:
+        LoginFailed: For authentication errors
+        MediaNotFoundError: For missing stations/tracks
+        ResourceTemporarilyUnavailable: For service availability issues
+        ProviderUnavailableError: For other API errors
+    """
+    if response_data.get("errorCode") is not None:
+        error_code = response_data["errorCode"]
+        error_string = response_data.get("errorString", "UNKNOWN_ERROR")
+        message = response_data.get("message", "An unknown error occurred")
+
+        # Map specific error codes to Music Assistant exceptions
+        if error_code in (12, 13, 1002):  # Invalid username/password/login
+            raise LoginFailed(f"Login failed: {message}")
+
+        if error_code in (4, 5):  # Station/track not found
+            raise MediaNotFoundError(f"Media not found: {message}")
+
+        if error_code in (9, 10):  # Service unavailable
+            raise ResourceTemporarilyUnavailable(f"Service unavailable: {message}")
+
+        if error_code in (1001, 1003):  # Auth token issues
+            raise LoginFailed(f"Authentication error: {message}")
+
+        # Get error description from our mapping
+        error_desc = PANDORA_ERROR_CODES.get(error_code, error_string)
+        raise ProviderUnavailableError(f"Pandora API error {error_code} ({error_desc}): {message}")
+
+
+async def get_csrf_token(session: aiohttp.ClientSession) -> str:
+    """Get CSRF token from Pandora website.
+
+    Attempts to retrieve CSRF token from Pandora cookies.
+
+    Args:
+        session: aiohttp client session
+
+    Returns:
+        CSRF token string
+
+    Raises:
+        ResourceTemporarilyUnavailable: If network request fails or no token available
+    """
+    try:
+        async with session.head("https://www.pandora.com/") as response:
+            if "csrftoken" in response.cookies:
+                return str(response.cookies["csrftoken"].value)
+    except aiohttp.ClientError as e:
+        raise ResourceTemporarilyUnavailable(f"Failed to get CSRF token from Pandora: {e}") from e
+
+    # No token found - service may be unavailable
+    raise ResourceTemporarilyUnavailable(
+        "Pandora did not provide a CSRF token. Service may be unavailable."
+    )
+
+
+def create_auth_headers(csrf_token: str, auth_token: str | None = None) -> dict[str, str]:
+    """Create authentication headers for Pandora API requests.
+
+    Args:
+        csrf_token: CSRF token for request validation
+        auth_token: Optional authentication token for authenticated requests
+
+    Returns:
+        Dictionary of HTTP headers
+    """
+    headers = {
+        "Content-Type": "application/json;charset=utf-8",
+        "X-CsrfToken": csrf_token,
+        "Cookie": f"csrftoken={csrf_token}",
+        "User-Agent": "Music Assistant Pandora Provider/1.0",
+    }
+
+    if auth_token:
+        headers["X-AuthToken"] = auth_token
+
+    return headers

music_assistant/providers/pandora/manifest.json

@@ -0,0 +1,11 @@
+{
+  "domain": "pandora",
+  "name": "Pandora",
+  "description": "Pandora is a music and podcast streaming service that creates personalized radio stations based on your favorite songs and artists.",
+  "documentation": "https://music-assistant.io/music-providers/pandora/",
+  "type": "music",
+  "requirements": [],
+  "codeowners": "@ozgav",
+  "multi_instance": false,
+  "stage": "beta"
+}