hermes-agent/fast-paths/patterns/spotify.py

"""Fast-Path-Skill: Spotify.

Deckt reine Steuerbefehle (pause/weiter/next/previous/lauter/leiser/
Lautstaerke setzen/aktueller Titel) OHNE LLM-Roundtrip ab, plus als
Kuer "spiel Playlist X [auf Geraet Y] ab" per Fuzzy-Match (kein LLM
noetig fuer den Freitext-Playlistnamen -- difflib reicht).

Nutzt bewusst NICHT die injizierten <tool_call>-Tools (die laufen ueber
den Proxy -> Claude -> zurueck, genau der Roundtrip den wir hier
umgehen wollen), sondern importiert Hermes' eigenen SpotifyClient direkt
-- gleiche OAuth-Session, gleicher Auto-Refresh, kein Code doppelt
gepflegt (siehe plugins/spotify/client.py im Hermes-Sourcecode).

Sicherheitsnetz: jeder Handler gibt None zurueck, sobald er sich nicht
sicher genug ist (z.B. Playlist-Name mehrdeutig) -- dann uebernimmt ganz
normal das LLM, es wird NIE geraten.
"""

from __future__ import annotations

import difflib
import logging
from typing import Any, Dict, List, Optional

from ._base import pattern

logger = logging.getLogger("hermes_plugins.fast_paths.spotify")


def _client():
    # Lazy-Import: erst bei tatsaechlichem Bedarf, damit ein Import-Fehler
    # hier (z.B. Spotify-Plugin bei einem Hermes-Update umbenannt) nicht das
    # Laden ALLER Fast-Paths verhindert -- siehe ../__init__.py, der jede
    # Skill-Datei einzeln try/except importiert.
    from plugins.spotify.client import SpotifyClient
    return SpotifyClient()


def _friendly_error(exc: Exception) -> str:
    # str(exc) ist bei SpotifyAuthRequiredError/SpotifyAPIError bereits die
    # aufbereitete, menschenlesbare Meldung (siehe _friendly_spotify_error_message
    # in plugins/spotify/client.py) -- 1:1 zitieren, nicht neu formulieren
    # oder raten (siehe ARIAs eigene Anti-Halluzinations-Regel).
    return f"⚠️ Spotify: {exc}"


def _devices(payload: Dict[str, Any]) -> List[Dict[str, Any]]:
    return list((payload or {}).get("devices") or [])


def _active_device_id(payload: Dict[str, Any]) -> Optional[str]:
    for d in _devices(payload):
        if d.get("is_active"):
            return d.get("id")
    return None


def _best_match(query: str, candidates: List[str], cutoff: float = 0.55) -> Optional[int]:
    """Index des besten Fuzzy-Treffers, oder None wenn zu unsicher.

    "Zu unsicher" heisst: kein Treffer ueber cutoff, ODER die besten zwei
    Treffer liegen zu nah beieinander (< 0.08 Differenz) -- dann lieber ans
    LLM abgeben statt eine von zwei aehnlich benannten Playlists zu raten.
    Exakt das Verhalten, das ARIA bei "mehrere Substring-Matches" auch
    einfordert: nachfragen statt raten.
    """
    if not candidates:
        return None
    scored = sorted(
        (
            (i, difflib.SequenceMatcher(None, query.lower(), (c or "").lower()).ratio())
            for i, c in enumerate(candidates)
        ),
        key=lambda pair: pair[1],
        reverse=True,
    )
    if not scored or scored[0][1] < cutoff:
        return None
    if len(scored) > 1 and (scored[0][1] - scored[1][1]) < 0.08:
        return None
    return scored[0][0]


# ---------------------------------------------------------------------------
# Reine Steuerbefehle
# ---------------------------------------------------------------------------

def _handle_pause(match, event, gateway) -> Optional[str]:
    try:
        _client().pause_playback()
    except Exception as exc:
        return _friendly_error(exc)
    return "⏸ Pausiert."


def _handle_resume(match, event, gateway) -> Optional[str]:
    try:
        _client().start_playback()
    except Exception as exc:
        return _friendly_error(exc)
    return "▶️ Weiter."


def _handle_next(match, event, gateway) -> Optional[str]:
    try:
        _client().skip_next()
    except Exception as exc:
        return _friendly_error(exc)
    return "⏭ Nächster Titel."


def _handle_previous(match, event, gateway) -> Optional[str]:
    try:
        _client().skip_previous()
    except Exception as exc:
        return _friendly_error(exc)
    return "⏮ Vorheriger Titel."


def _handle_volume_relative(match, event, gateway) -> Optional[str]:
    direction = (match.group("dir") or "").lower()
    delta = 10 if direction.startswith("laut") else -10
    client = _client()
    try:
        devices_payload = client.get_devices()
        device_id = _active_device_id(devices_payload)
        current = 50
        for d in _devices(devices_payload):
            if d.get("id") == device_id:
                current = int(d.get("volume_percent") or 50)
                break
        new_volume = max(0, min(100, current + delta))
        client.set_volume(volume_percent=new_volume, device_id=device_id)
    except Exception as exc:
        return _friendly_error(exc)
    return f"🔊 Lautstärke: {new_volume}%"


def _handle_volume_set(match, event, gateway) -> Optional[str]:
    try:
        percent = max(0, min(100, int(match.group("pct"))))
    except (TypeError, ValueError):
        return None  # Zahl nicht sauber parsebar -> lieber ans LLM
    try:
        _client().set_volume(volume_percent=percent)
    except Exception as exc:
        return _friendly_error(exc)
    return f"🔊 Lautstärke: {percent}%"


def _handle_current(match, event, gateway) -> Optional[str]:
    try:
        payload = _client().get_currently_playing()
    except Exception as exc:
        return _friendly_error(exc)
    if not payload or payload.get("empty") or not payload.get("item"):
        return "⏸ Aktuell läuft nichts auf Spotify."
    item = payload["item"]
    name = item.get("name", "?")
    artists = ", ".join(a.get("name", "?") for a in item.get("artists", []) or [])
    device = (payload.get("device") or {}).get("name")
    suffix = f" ({device})" if device else ""
    return f"🎵 Läuft gerade: {name} – {artists}{suffix}"


# ---------------------------------------------------------------------------
# "spiel Playlist X [auf Geraet Y] ab" -- Freitext-Namen per Fuzzy-Match,
# kein LLM noetig. Sicherheitsnetz: bei Unsicherheit -> None -> LLM uebernimmt.
# ---------------------------------------------------------------------------

def _handle_play_playlist(match, event, gateway) -> Optional[str]:
    playlist_query = (match.group("playlist") or "").strip().strip("\"“”'")
    device_query = ""
    if "device" in match.groupdict():
        device_query = (match.group("device") or "").strip()
    if not playlist_query:
        return None

    client = _client()

    # ALLE Playlists einsammeln bevor gematcht wird -- nicht auf
    # Teilergebnissen raten (ARIAs eigene Pagination-Regel).
    try:
        items: List[Dict[str, Any]] = []
        offset = 0
        while len(items) < 300:
            page = client.get_my_playlists(limit=50, offset=offset)
            page_items = page.get("items") or []
            items.extend(page_items)
            if not page.get("next") or not page_items:
                break
            offset += len(page_items)
    except Exception as exc:
        return _friendly_error(exc)

    names = [p.get("name", "") for p in items]
    idx = _best_match(playlist_query, names)
    if idx is None:
        logger.info(
            "fast_paths/spotify: Playlist '%s' nicht eindeutig unter %d Playlists -> LLM uebernimmt",
            playlist_query, len(names),
        )
        return None
    playlist = items[idx]

    try:
        devices_payload = client.get_devices()
    except Exception as exc:
        return _friendly_error(exc)
    devices = _devices(devices_payload)
    if not devices:
        return "⚠️ Spotify: kein Gerät gefunden — öffne Spotify auf irgendeinem Gerät und versuch's nochmal."

    device_id = None
    if device_query:
        device_names = [d.get("name", "") for d in devices]
        d_idx = _best_match(device_query, device_names, cutoff=0.4)
        if d_idx is None:
            logger.info(
                "fast_paths/spotify: Geraet '%s' nicht eindeutig unter %s -> LLM uebernimmt",
                device_query, device_names,
            )
            return None
        device_id = devices[d_idx].get("id")
    else:
        device_id = _active_device_id(devices_payload) or devices[0].get("id")

    try:
        client.start_playback(device_id=device_id, context_uri=playlist.get("uri"))
    except Exception as exc:
        return _friendly_error(exc)

    device_name = next((d.get("name") for d in devices if d.get("id") == device_id), "?")
    return f"▶️ Spiele Playlist „{playlist.get('name')}“ auf {device_name}."


PATTERNS = [
    pattern(
        "pause",
        r"^(?:spotify\s+)?(?:pause|pausier(?:e|en)?|stop|stopp|halt|anhalten)"
        # * statt ? -- deckt auch mehrere aneinandergereihte Fuellwoerter ab
        # ("pause bitte spotify"), nicht nur genau ein einzelnes.
        r"(?:\s+(?:mal|bitte|spotify|die\s+musik))*$",
        _handle_pause,
    ),
    pattern(
        "resume",
        r"^(?:spotify\s+)?(?:weiter|weiterspielen|fortsetzen|play|abspielen|los)"
        r"(?:\s+(?:mal|bitte|spotify|die\s+musik))*$",
        _handle_resume,
    ),
    pattern(
        "next",
        r"^(?:spotify\s+)?(?:n(?:ä|ae)chste[rs]?(?:\s+(?:lied|song|titel))?"
        r"|weiter(?:er)?\s+(?:lied|song|titel)|skip|next)(?:\s+(?:mal|bitte|spotify))*$",
        _handle_next,
    ),
    pattern(
        "previous",
        r"^(?:spotify\s+)?(?:vorherige[rs]?(?:\s+(?:lied|song|titel))?"
        r"|zur(?:ü|ue)ck(?:\s+zum\s+vorherigen)?|previous)(?:\s+(?:mal|bitte|spotify))*$",
        _handle_previous,
    ),
    pattern(
        "volume_relative",
        r"^(?:spotify\s+)?(?:mach\s+(?:es\s+)?)?(?P<dir>lauter|leiser)(?:\s+(?:mal|bitte|spotify))*$",
        _handle_volume_relative,
    ),
    pattern(
        "volume_set",
        r"^(?:spotify\s+)?(?:lautst(?:ä|ae)rke|volume)\s+(?:auf\s+)?(?P<pct>\d{1,3})\s*%?$",
        _handle_volume_set,
    ),
    pattern(
        "current",
        r"^(?:spotify\s+)?(?:was\s+l(?:ä|ae)uft(?:\s+(?:gerade|jetzt))?"
        r"|welches\s+lied\s+l(?:ä|ae)uft|aktueller\s+(?:song|titel)|current(?:ly\s+playing)?)"
        r"(?:\s+(?:mal|bitte))*$",
        _handle_current,
    ),
    pattern(
        "play_playlist",
        r"^spiel(?:e)?\s+(?:mal\s+|bitte\s+)?(?:meine\s+)?playlist\s+"
        r"[\"“]?(?P<playlist>.+?)[\"”]?(?:\s+auf\s+(?:mein(?:em|er)?\s+)?(?P<device>.+?))?"
        r"\s*(?:ab)?$",
        _handle_play_playlist,
    ),
]