spotify-cli/spotify_cli/auth.py

import hashlib
import base64
import secrets
import webbrowser
import time
import json
from urllib.parse import urlencode, urlparse, parse_qs
import requests

from .config import load_config, save_token, load_token, clear_token

SPOTIFY_AUTH_URL = "https://accounts.spotify.com/authorize"
SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token"
REDIRECT_URI = "http://127.0.0.1:8888/callback"
SCOPES = [
    "user-read-playback-state",
    "user-modify-playback-state",
    "user-read-currently-playing",
    "playlist-read-private",
    "playlist-read-collaborative",
    "user-library-read",
    "user-read-recently-played",
    "user-top-read",
    "streaming",
]


def _generate_pkce():
    code_verifier = secrets.token_urlsafe(64)[:128]
    digest = hashlib.sha256(code_verifier.encode()).digest()
    code_challenge = base64.urlsafe_b64encode(digest).rstrip(b"=").decode()
    return code_verifier, code_challenge


class SpotifyAuth:
    def __init__(self):
        self.cfg = load_config()
        self.client_id = self.cfg.get("client_id", "")
        self.token_data = load_token()
        self._code_verifier = None

    def _is_token_valid(self):
        if not self.token_data:
            return False
        return time.time() < self.token_data.get("expires_at", 0) - 60

    def get_token(self):
        if self._is_token_valid():
            return self.token_data["access_token"]

        if self.token_data and self.token_data.get("refresh_token"):
            refreshed = self._refresh(self.token_data["refresh_token"])
            if refreshed:
                return self.token_data["access_token"]

        return self._full_auth()

    def _refresh(self, refresh_token):
        resp = requests.post(SPOTIFY_TOKEN_URL, data={
            "grant_type": "refresh_token",
            "refresh_token": refresh_token,
            "client_id": self.client_id,
        })
        if resp.status_code == 200:
            data = resp.json()
            self.token_data = {
                "access_token": data["access_token"],
                "refresh_token": data.get("refresh_token", refresh_token),
                "expires_at": time.time() + data["expires_in"],
            }
            save_token(self.token_data)
            return True
        clear_token()
        self.token_data = None
        return False

    def _full_auth(self):
        if not self.client_id:
            self.client_id = input("Spotify Client ID: ").strip()
            self.cfg["client_id"] = self.client_id
            from .config import save_config
            save_config(self.cfg)

        code_verifier, code_challenge = _generate_pkce()
        self._code_verifier = code_verifier
        state = secrets.token_urlsafe(16)

        params = {
            "client_id": self.client_id,
            "response_type": "code",
            "redirect_uri": REDIRECT_URI,
            "code_challenge_method": "S256",
            "code_challenge": code_challenge,
            "state": state,
            "scope": " ".join(SCOPES),
        }

        auth_url = f"{SPOTIFY_AUTH_URL}?{urlencode(params)}"
        print(f"\n{'='*60}")
        print(f"Oeffne Browser fuer Spotify Login...")
        print(f"Falls der Browser nicht öffnet, nutze diese URL:")
        print(f"\n{auth_url}\n")
        print(f"{'='*60}")
        print(f"\nNach der Authentifizierung wirst du weitergeleitet.")
        print(f"Der Redirect wird fehlschlagen - das ist normal!")
        print(f"\nKopiere die URL aus der Browser-Adresszeile")
        print(f"und fuege sie hier ein.\n")

        try:
            webbrowser.open(auth_url)
        except Exception:
            pass

        code = self._wait_for_code()
        if not code:
            print("Authentifizierung fehlgeschlagen.")
            return None

        return self._exchange_code(code, code_verifier)

    def _wait_for_code(self):
        for attempt in range(5):
            url_input = input("URL einfügen: ").strip()

            if not url_input:
                if attempt < 4:
                    print("Bitte URL einfügen.")
                continue

            code = self._extract_code(url_input)
            if code:
                return code

            print("Code konnte nicht extrahiert werden. Versuche es erneut.")

        return None

    def _extract_code(self, url_or_code):
        if "code=" in url_or_code:
            parsed = urlparse(url_or_code)
            params = parse_qs(parsed.query)
            return params.get("code", [None])[0]

        if url_or_code.startswith("AQ"):
            return url_or_code

        return url_or_code if len(url_or_code) > 20 else None

    def _exchange_code(self, code, code_verifier):
        resp = requests.post(SPOTIFY_TOKEN_URL, data={
            "grant_type": "authorization_code",
            "code": code,
            "redirect_uri": REDIRECT_URI,
            "client_id": self.client_id,
            "code_verifier": code_verifier,
        })

        if resp.status_code == 200:
            data = resp.json()
            self.token_data = {
                "access_token": data["access_token"],
                "refresh_token": data["refresh_token"],
                "expires_at": time.time() + data["expires_in"],
            }
            save_token(self.token_data)
            print("Erfolgreich eingelogged!")
            return data["access_token"]

        print(f"Token-Austausch fehlgeschlagen: {resp.status_code}")
        print(resp.text)
        return None

    def logout(self):
        clear_token()
        self.token_data = None
        print("Ausgelogged.")