playwish/player/auth.py

#!/usr/bin/env python3
"""Audiofy – einmaliger Spotify-Login (Authorization Code Flow).

Holt einen Refresh-Token für das eigene Spotify-Konto und schreibt ihn in die
.env-Datei. Funktioniert ohne Browser auf dem Gerät selbst:

  1. Die ausgegebene Login-Adresse in einem beliebigen Browser öffnen.
  2. Bei Spotify anmelden und zustimmen.
  3. Der Browser landet auf http://127.0.0.1:8888/callback?... – die Seite lädt
     nicht, das ist normal. Die komplette Adresse aus der Adresszeile kopieren
     und hier einfügen.

Aufruf:
    python3 auth.py --env /opt/audiofy/config/.env                  # Player (Wiedergabe steuern)
    python3 auth.py --profil server --env /opt/audiofy/.env         # Server (nur Suche)

Benötigte Einträge in der .env: SPOTIFY_CLIENT_ID, SPOTIFY_CLIENT_SECRET,
optional SPOTIFY_REDIRECT_URI. Nur Python-Standardbibliothek, kein venv nötig.
"""

import argparse
import base64
import getpass
import json
import os
import secrets
import stat
import sys
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path

AUTHORIZE_URL = "https://accounts.spotify.com/authorize"
TOKEN_URL = "https://accounts.spotify.com/api/token"
ME_URL = "https://api.spotify.com/v1/me"
STANDARD_REDIRECT = "http://127.0.0.1:8888/callback"

# Player: Wiedergabe steuern + Playlist lesen; user-read-private nur, damit /me das
# Feld "product" (Premium-Check) liefert.
# Server: Suche braucht keinen Scope. ("pi"/"vps" = alte Namen)
SCOPES = {
    "player": "user-read-playback-state user-modify-playback-state "
          "user-read-currently-playing playlist-read-private user-read-private",
    "server": "",
    "pi": None,     # alter Name für "player"
    "vps": None,    # alter Name für "server"
}


def env_lesen(pfad: Path) -> dict:
    """Einfacher KEY=VALUE-Parser (Kommentare und Anführungszeichen erlaubt)."""
    werte = {}
    for zeile in pfad.read_text().splitlines():
        zeile = zeile.strip()
        if not zeile or zeile.startswith("#") or "=" not in zeile:
            continue
        key, _, wert = zeile.partition("=")
        werte[key.strip()] = wert.strip().strip('"').strip("'")
    return werte


def env_setzen(pfad: Path, key: str, wert: str) -> None:
    """Setzt KEY=wert in der .env (ersetzt oder hängt an), atomar und mit Rechten 600."""
    zeilen = pfad.read_text().splitlines()
    neu, gefunden = [], False
    for zeile in zeilen:
        if zeile.strip().startswith(f"{key}="):
            neu.append(f"{key}={wert}")
            gefunden = True
        else:
            neu.append(zeile)
    if not gefunden:
        neu.append(f"{key}={wert}")

    inhalt = "\n".join(neu) + "\n"
    if not os.access(pfad.parent, os.W_OK):
        # z. B. Server: /opt/audiofy gehört root, nur die .env selbst dem Dienst-Benutzer
        fd = os.open(pfad, os.O_WRONLY | os.O_TRUNC)
        with os.fdopen(fd, "w") as f:
            f.write(inhalt)
        return
    tmp = pfad.with_name(pfad.name + ".tmp")
    fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(fd, "w") as f:
        f.write(inhalt)
    os.replace(tmp, pfad)


def token_anfragen(client_id: str, client_secret: str, daten: dict) -> dict:
    body = urllib.parse.urlencode(daten).encode()
    basic = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
    req = urllib.request.Request(TOKEN_URL, data=body, headers={
        "Authorization": f"Basic {basic}",
        "Content-Type": "application/x-www-form-urlencoded",
    })
    try:
        with urllib.request.urlopen(req, timeout=15) as antwort:
            return json.load(antwort)
    except urllib.error.HTTPError as e:
        # Spotify-Fehlertexte enthalten keine Geheimnisse
        sys.exit(f"Token-Anfrage fehlgeschlagen ({e.code}): {e.read().decode(errors='replace')[:300]}")


def me_abfragen(access_token: str) -> dict:
    req = urllib.request.Request(ME_URL, headers={"Authorization": f"Bearer {access_token}"})
    try:
        with urllib.request.urlopen(req, timeout=15) as antwort:
            return json.load(antwort)
    except urllib.error.HTTPError as e:
        sys.exit(f"/me fehlgeschlagen ({e.code}): {e.read().decode(errors='replace')[:300]}")


def main() -> None:
    parser = argparse.ArgumentParser(description="Audiofy – Spotify-Login, schreibt den Refresh-Token in die .env")
    parser.add_argument("--profil", choices=SCOPES, default="player", help="Scope-Satz (Standard: player)")
    parser.add_argument("--env", type=Path, default=Path(__file__).resolve().parent / ".env",
                        help="Pfad zur .env (Standard: neben dem Skript)")
    args = parser.parse_args()
    args.profil = {"pi": "player", "vps": "server"}.get(args.profil, args.profil)   # alte Namen

    if not args.env.is_file():
        sys.exit(f"{args.env} fehlt. Bitte anlegen (Rechte 600) mit SPOTIFY_CLIENT_ID und SPOTIFY_CLIENT_SECRET.")
    if stat.S_IMODE(args.env.stat().st_mode) & 0o077:
        print(f"Warnung: {args.env} ist für andere lesbar – bitte 'chmod 600 {args.env}'.", file=sys.stderr)

    env = env_lesen(args.env)
    client_id = env.get("SPOTIFY_CLIENT_ID", "")
    client_secret = env.get("SPOTIFY_CLIENT_SECRET", "")
    redirect_uri = env.get("SPOTIFY_REDIRECT_URI") or STANDARD_REDIRECT
    if not client_id or not client_secret:
        sys.exit(f"SPOTIFY_CLIENT_ID und SPOTIFY_CLIENT_SECRET müssen in {args.env} gesetzt sein.")

    state = secrets.token_urlsafe(16)
    parameter = {
        "client_id": client_id,
        "response_type": "code",
        "redirect_uri": redirect_uri,
        "state": state,
        "show_dialog": "true",
    }
    if SCOPES[args.profil]:
        parameter["scope"] = SCOPES[args.profil]

    print(f"Profil: {args.profil}")
    print("\n1) Diese Adresse im Browser öffnen und bei Spotify anmelden:\n")
    print(f"   {AUTHORIZE_URL}?{urllib.parse.urlencode(parameter)}\n")
    print(f"2) Danach die komplette Adresse aus der Adresszeile ({redirect_uri}?...) einfügen.")
    print("   (Eingabe wird nicht angezeigt.)\n")
    eingabe = getpass.getpass("Adresse: ").strip()

    url = urllib.parse.urlparse(eingabe)
    query = urllib.parse.parse_qs(url.query)
    if "error" in query:
        sys.exit(f"Spotify meldet: {query['error'][0]}")
    if query.get("state", [""])[0] != state:
        sys.exit("State passt nicht – bitte die Adresse aus genau diesem Durchlauf einfügen.")
    code = query.get("code", [""])[0]
    if not code:
        sys.exit("Kein 'code' in der Adresse gefunden.")

    tokens = token_anfragen(client_id, client_secret, {
        "grant_type": "authorization_code",
        "code": code,
        "redirect_uri": redirect_uri,
    })
    refresh_token = tokens.get("refresh_token")
    if not refresh_token:
        sys.exit("Antwort enthält keinen Refresh-Token.")

    env_setzen(args.env, "SPOTIFY_REFRESH_TOKEN", refresh_token)
    print(f"\nRefresh-Token in {args.env} gespeichert.")

    erhalten = set(tokens.get("scope", "").split())
    fehlend = set(SCOPES[args.profil].split()) - erhalten
    if fehlend:
        print(f"Warnung: Diese Scopes wurden nicht gewährt: {' '.join(sorted(fehlend))}")

    me = me_abfragen(tokens["access_token"])
    print(f"Angemeldet als: {me.get('display_name') or me.get('id')}")
    if "product" in me:
        print(f"Abo: {me['product']}" + ("  ✅" if me["product"] == "premium" else "  ❌ (Premium nötig!)"))


if __name__ == "__main__":
    main()