""" 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"((александр|александра|алесандр|alexander)\s+)?(угадай|распознай|определи)\s+(мелод|музык|песн|трек)", r"((александр|александра|алесандр|alexander)\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