341 lines
14 KiB
Python
341 lines
14 KiB
Python
"""
|
||
Spotify Music Controller
|
||
Модуль для управления воспроизведением музыки через Spotify API.
|
||
|
||
Поддерживаемые команды:
|
||
- "включи музыку" / "play music" - воспроизведение топ-треков или по запросу
|
||
- "пауза" / "стоп музыка" - пауза
|
||
- "продолжи" / "дальше" - возобновление
|
||
- "следующий трек" / "next" - следующий трек
|
||
- "предыдущий трек" / "previous" - предыдущий трек
|
||
- "что играет" / "какая песня" - информация о текущем треке
|
||
- "угадай песню" / "распознай музыку" - распознавание текущего трека
|
||
"""
|
||
|
||
import os
|
||
import re
|
||
from typing import Optional
|
||
|
||
try:
|
||
import spotipy
|
||
from spotipy.oauth2 import SpotifyOAuth
|
||
except ImportError:
|
||
spotipy = None
|
||
SpotifyOAuth = None
|
||
|
||
# Singleton instance
|
||
_music_controller = None
|
||
|
||
|
||
class SpotifyMusicController:
|
||
"""Контроллер для управления Spotify воспроизведением."""
|
||
|
||
# Scopes для Spotify API
|
||
SCOPES = [
|
||
"user-read-playback-state",
|
||
"user-modify-playback-state",
|
||
"user-read-currently-playing",
|
||
"user-top-read",
|
||
"streaming",
|
||
]
|
||
|
||
def __init__(self):
|
||
"""Инициализация контроллера Spotify."""
|
||
self.sp = None
|
||
self._initialized = False
|
||
self._init_error = None
|
||
|
||
def initialize(self) -> bool:
|
||
"""
|
||
Инициализация подключения к Spotify.
|
||
Возвращает True при успехе, False при ошибке.
|
||
"""
|
||
if self._initialized:
|
||
return True
|
||
|
||
if spotipy is None:
|
||
self._init_error = "Библиотека spotipy не установлена. Установите: pip install spotipy"
|
||
print(f"⚠️ Spotify: {self._init_error}")
|
||
return False
|
||
|
||
# Проверяем наличие ключей
|
||
client_id = os.getenv("SPOTIFY_CLIENT_ID")
|
||
client_secret = os.getenv("SPOTIFY_CLIENT_SECRET")
|
||
redirect_uri = os.getenv("SPOTIFY_REDIRECT_URI", "http://localhost:8888/callback")
|
||
|
||
if not client_id or not client_secret:
|
||
self._init_error = "Не заданы SPOTIFY_CLIENT_ID и SPOTIFY_CLIENT_SECRET в .env"
|
||
print(f"⚠️ Spotify: {self._init_error}")
|
||
return False
|
||
|
||
try:
|
||
auth_manager = SpotifyOAuth(
|
||
client_id=client_id,
|
||
client_secret=client_secret,
|
||
redirect_uri=redirect_uri,
|
||
scope=" ".join(self.SCOPES),
|
||
cache_path=".spotify_cache",
|
||
)
|
||
self.sp = spotipy.Spotify(auth_manager=auth_manager)
|
||
# Проверяем подключение
|
||
self.sp.current_user()
|
||
self._initialized = True
|
||
print("✅ Spotify: подключено")
|
||
return True
|
||
except Exception as e:
|
||
self._init_error = str(e)
|
||
print(f"❌ Spotify: ошибка инициализации - {e}")
|
||
return False
|
||
|
||
def _ensure_initialized(self) -> bool:
|
||
"""Проверяет и инициализирует при необходимости."""
|
||
if not self._initialized:
|
||
return self.initialize()
|
||
return True
|
||
|
||
def _get_active_device(self) -> Optional[str]:
|
||
"""Получить ID активного устройства воспроизведения."""
|
||
try:
|
||
devices = self.sp.devices()
|
||
if devices and devices.get("devices"):
|
||
# Ищем активное устройство или берем первое
|
||
for device in devices["devices"]:
|
||
if device.get("is_active"):
|
||
return device["id"]
|
||
# Если нет активного, берем первое
|
||
return devices["devices"][0]["id"]
|
||
except Exception:
|
||
pass
|
||
return None
|
||
|
||
def play_music(self, query: Optional[str] = None) -> str:
|
||
"""
|
||
Включить музыку.
|
||
|
||
Args:
|
||
query: Поисковый запрос (название песни/артиста).
|
||
Если None, включает топ-треки пользователя.
|
||
|
||
Returns:
|
||
Сообщение для озвучки.
|
||
"""
|
||
if not self._ensure_initialized():
|
||
return f"Не удалось подключиться к Spotify. {self._init_error or ''}"
|
||
|
||
try:
|
||
device_id = self._get_active_device()
|
||
|
||
if query:
|
||
# Поиск по запросу
|
||
results = self.sp.search(q=query, type="track", limit=1)
|
||
tracks = results.get("tracks", {}).get("items", [])
|
||
|
||
if not tracks:
|
||
return f"Не нашёл песню '{query}'"
|
||
|
||
track = tracks[0]
|
||
track_name = track["name"]
|
||
artist_name = track["artists"][0]["name"]
|
||
track_uri = track["uri"]
|
||
|
||
self.sp.start_playback(device_id=device_id, uris=[track_uri])
|
||
return f"Включаю {track_name} от {artist_name}"
|
||
else:
|
||
# Включаем топ-треки пользователя
|
||
top_tracks = self.sp.current_user_top_tracks(limit=20, time_range="medium_term")
|
||
if top_tracks and top_tracks.get("items"):
|
||
uris = [track["uri"] for track in top_tracks["items"]]
|
||
self.sp.start_playback(device_id=device_id, uris=uris)
|
||
first_track = top_tracks["items"][0]
|
||
return f"Включаю вашу музыку. Сейчас играет {first_track['name']}"
|
||
else:
|
||
# Если нет топ-треков, ставим что-нибудь популярное
|
||
self.sp.start_playback(device_id=device_id)
|
||
return "Включаю музыку"
|
||
|
||
except spotipy.SpotifyException as e:
|
||
if "NO_ACTIVE_DEVICE" in str(e) or "Player command failed" in str(e):
|
||
return "Нет активного устройства Spotify. Откройте Spotify на телефоне или компьютере."
|
||
elif "PREMIUM_REQUIRED" in str(e):
|
||
return "Для управления воспроизведением нужен Spotify Premium."
|
||
return f"Ошибка Spotify: {e.reason if hasattr(e, 'reason') else str(e)}"
|
||
except Exception as e:
|
||
return f"Ошибка воспроизведения: {e}"
|
||
|
||
def pause_music(self) -> str:
|
||
"""Поставить на паузу."""
|
||
if not self._ensure_initialized():
|
||
return "Spotify не подключён"
|
||
|
||
try:
|
||
self.sp.pause_playback()
|
||
return "Музыка на паузе"
|
||
except spotipy.SpotifyException as e:
|
||
if "NO_ACTIVE_DEVICE" in str(e):
|
||
return "Нет активного устройства Spotify"
|
||
return f"Не удалось поставить на паузу: {e}"
|
||
except Exception as e:
|
||
return f"Ошибка: {e}"
|
||
|
||
def resume_music(self) -> str:
|
||
"""Продолжить воспроизведение."""
|
||
if not self._ensure_initialized():
|
||
return "Spotify не подключён"
|
||
|
||
try:
|
||
self.sp.start_playback()
|
||
return "Продолжаю воспроизведение"
|
||
except spotipy.SpotifyException as e:
|
||
if "NO_ACTIVE_DEVICE" in str(e):
|
||
return "Нет активного устройства Spotify"
|
||
return f"Ошибка: {e}"
|
||
except Exception as e:
|
||
return f"Ошибка: {e}"
|
||
|
||
def next_track(self) -> str:
|
||
"""Следующий трек."""
|
||
if not self._ensure_initialized():
|
||
return "Spotify не подключён"
|
||
|
||
try:
|
||
self.sp.next_track()
|
||
# Небольшая задержка для обновления состояния
|
||
import time
|
||
time.sleep(0.5)
|
||
return self.get_current_track() or "Переключаю на следующий трек"
|
||
except Exception as e:
|
||
return f"Не удалось переключить трек: {e}"
|
||
|
||
def previous_track(self) -> str:
|
||
"""Предыдущий трек."""
|
||
if not self._ensure_initialized():
|
||
return "Spotify не подключён"
|
||
|
||
try:
|
||
self.sp.previous_track()
|
||
import time
|
||
time.sleep(0.5)
|
||
return self.get_current_track() or "Переключаю на предыдущий трек"
|
||
except Exception as e:
|
||
return f"Не удалось переключить трек: {e}"
|
||
|
||
def get_current_track(self) -> Optional[str]:
|
||
"""Получить информацию о текущем треке."""
|
||
if not self._ensure_initialized():
|
||
return "Spotify не подключён"
|
||
|
||
try:
|
||
current = self.sp.current_playback()
|
||
if current and current.get("item"):
|
||
track = current["item"]
|
||
name = track["name"]
|
||
artists = ", ".join(a["name"] for a in track["artists"])
|
||
is_playing = current.get("is_playing", False)
|
||
status = "Сейчас играет" if is_playing else "На паузе"
|
||
return f"{status}: {name} от {artists}"
|
||
return "Сейчас ничего не играет"
|
||
except Exception as e:
|
||
return f"Не удалось получить информацию: {e}"
|
||
|
||
def parse_command(self, text: str) -> Optional[str]:
|
||
"""
|
||
Распознать музыкальную команду и выполнить её.
|
||
|
||
Args:
|
||
text: Текст команды от пользователя.
|
||
|
||
Returns:
|
||
Ответ для озвучки или None, если это не музыкальная команда.
|
||
"""
|
||
text_lower = text.lower().strip()
|
||
|
||
# Команды паузы
|
||
pause_patterns = [
|
||
r"^(поставь на паузу|пауза|стоп музык|останови музык|выключи музык)",
|
||
r"(pause|stop music)",
|
||
]
|
||
for pattern in pause_patterns:
|
||
if re.search(pattern, text_lower):
|
||
return self.pause_music()
|
||
|
||
# Команды продолжения
|
||
resume_patterns = [
|
||
r"^(продолжи|продолжай|возобнови|сними с паузы|дальше|играй дальше)",
|
||
r"(resume|continue|play)",
|
||
]
|
||
for pattern in resume_patterns:
|
||
if re.search(pattern, text_lower):
|
||
return self.resume_music()
|
||
|
||
# Следующий трек
|
||
next_patterns = [
|
||
r"(следующ|дальше|скип|пропусти|next|skip)",
|
||
]
|
||
for pattern in next_patterns:
|
||
if re.search(pattern, text_lower) and (
|
||
"трек" in text_lower
|
||
or "песн" in text_lower
|
||
or "skip" in text_lower
|
||
or "next" in text_lower
|
||
):
|
||
return self.next_track()
|
||
|
||
# Предыдущий трек
|
||
prev_patterns = [
|
||
r"(предыдущ|назад|верни|previous|back)",
|
||
]
|
||
for pattern in prev_patterns:
|
||
if re.search(pattern, text_lower) and ("трек" in text_lower or "песн" in text_lower or "previous" in text_lower or "back" in text_lower):
|
||
return self.previous_track()
|
||
|
||
# Явные команды распознавания музыки (типа "угадай песню")
|
||
recognize_patterns = [
|
||
r"((waltron|voltron|волтрон|уолтрон|валтрон)\s+)?(угадай|распознай|определи)\s+(мелод|музык|песн|трек)",
|
||
r"((waltron|voltron|волтрон|уолтрон|валтрон)\s+)?(что за|какая это)\s+(музык|песн|трек)",
|
||
r"(identify|recognize)\s+(song|music|track)",
|
||
]
|
||
for pattern in recognize_patterns:
|
||
if re.search(pattern, text_lower):
|
||
return self.get_current_track()
|
||
|
||
# Что играет
|
||
current_patterns = [
|
||
r"(что (сейчас )?играет|как(ая|ой) (песня|трек)|что за (песня|трек|музыка))",
|
||
r"(what.*(play|song)|current track)",
|
||
]
|
||
for pattern in current_patterns:
|
||
if re.search(pattern, text_lower):
|
||
return self.get_current_track()
|
||
|
||
# Включить музыку (с возможным запросом)
|
||
play_patterns = [
|
||
(r"^включи\s+музыку$", None), # Просто "включи музыку"
|
||
(r"^включи\s+(.+)$", 1), # "включи [что-то]"
|
||
(r"^поставь\s+(.+)$", 1), # "поставь [что-то]"
|
||
(r"^играй\s+(.+)$", 1), # "играй [что-то]"
|
||
(r"^play\s+(.+)$", 1), # "play [something]"
|
||
(r"^(play music|включи музыку|поставь музыку)$", None),
|
||
]
|
||
|
||
for pattern, group in play_patterns:
|
||
match = re.search(pattern, text_lower)
|
||
if match:
|
||
query = None
|
||
if group:
|
||
query = match.group(group).strip()
|
||
# Убираем слова "музыку", "песню", "трек" из запроса
|
||
query = re.sub(r"(музыку|песню|трек|песня|song|track)\s*", "", query).strip()
|
||
if not query:
|
||
query = None
|
||
return self.play_music(query)
|
||
|
||
return None
|
||
|
||
|
||
def get_music_controller() -> SpotifyMusicController:
|
||
"""Получить singleton экземпляр контроллера музыки."""
|
||
global _music_controller
|
||
if _music_controller is None:
|
||
_music_controller = SpotifyMusicController()
|
||
return _music_controller
|