From e71f5623a764940837cb0164d958ef41e00cb391 Mon Sep 17 00:00:00 2001 From: jads Date: Fri, 23 Oct 2026 23:52:14 +0300 Subject: [PATCH] style: format all code to PEP8 (flake8 clean) Reformat every module with autopep8 and hand-tune the remaining long lines so that `flake8` passes with default settings (max line length 79). No behavior changes: long f-strings, messages, QSS/Textual CSS rules and argument lists are only wrapped or split, an unused import and an unused local are dropped, and dbus-next D-Bus signature annotations carry targeted `# noqa` comments. --- media.py | 8 +- mpris.py | 76 ++++++++------ mtsmusic.py | 291 +++++++++++++++++++++++++++++++++++++++------------- player.py | 10 +- qt_app.py | 205 ++++++++++++++++++++++++++---------- settings.py | 3 +- tui.py | 178 ++++++++++++++++++++++++-------- 7 files changed, 558 insertions(+), 213 deletions(-) diff --git a/media.py b/media.py index 118cac0..f4319b8 100644 --- a/media.py +++ b/media.py @@ -3,7 +3,6 @@ from __future__ import annotations import json -from pathlib import Path from mtsmusic import CONFIG_DIR @@ -12,7 +11,8 @@ HISTORY_LIMIT = 100 def track_artist(track: dict) -> str: - return ", ".join(a.get("name", "?") for a in track.get("artists") or []) or "?" + return ", ".join(a.get("name", "?") + for a in track.get("artists") or []) or "?" def track_album(track: dict) -> str: @@ -58,7 +58,9 @@ def record_history(track: dict) -> list[dict]: track_id = str(track.get("id") or "") if not track_id: return load_history() - history = [item for item in load_history() if str(item.get("id")) != track_id] + history = [ + item for item in load_history() if str( + item.get("id")) != track_id] history.insert(0, track) history = history[:HISTORY_LIMIT] try: diff --git a/mpris.py b/mpris.py index 311d99e..5c646b7 100644 --- a/mpris.py +++ b/mpris.py @@ -19,7 +19,12 @@ try: # pragma: no cover - зависит от окружения from dbus_next import Variant from dbus_next.aio import MessageBus from dbus_next.constants import PropertyAccess - from dbus_next.service import ServiceInterface, dbus_property, method, signal + from dbus_next.service import ( + ServiceInterface, + dbus_property, + method, + signal, + ) HAVE_DBUS = True except Exception: # pragma: no cover @@ -31,7 +36,8 @@ OBJECT_PATH = "/org/mpris/MediaPlayer2" def track_object_path(track_id: Any) -> str: """D-Bus object path для mpris:trackid (только [A-Za-z0-9_]).""" - safe = "".join(ch if ch.isalnum() else "_" for ch in str(track_id)) or "unknown" + safe = "".join(ch if ch.isalnum() + else "_" for ch in str(track_id)) or "unknown" return f"/mtsmusic/track/{safe}" @@ -51,31 +57,31 @@ if HAVE_DBUS: self._svc.dispatch("quit") @dbus_property(access=PropertyAccess.READ) - def Identity(self) -> "s": + def Identity(self) -> "s": # noqa: F821 return self._svc.identity @dbus_property(access=PropertyAccess.READ) - def DesktopEntry(self) -> "s": + def DesktopEntry(self) -> "s": # noqa: F821 return self._svc.desktop_entry @dbus_property(access=PropertyAccess.READ) - def CanQuit(self) -> "b": + def CanQuit(self) -> "b": # noqa: F821 return True @dbus_property(access=PropertyAccess.READ) - def CanRaise(self) -> "b": + def CanRaise(self) -> "b": # noqa: F821 return True @dbus_property(access=PropertyAccess.READ) - def HasTrackList(self) -> "b": + def HasTrackList(self) -> "b": # noqa: F821 return False @dbus_property(access=PropertyAccess.READ) - def SupportedUriSchemes(self) -> "as": + def SupportedUriSchemes(self) -> "as": # noqa: F722 return [] @dbus_property(access=PropertyAccess.READ) - def SupportedMimeTypes(self) -> "as": + def SupportedMimeTypes(self) -> "as": # noqa: F722 return [] class _PlayerInterface(ServiceInterface): @@ -109,97 +115,97 @@ if HAVE_DBUS: self._svc.dispatch("previous") @method() - def Seek(self, offset: "x"): + def Seek(self, offset: "x"): # noqa: F821 self._svc.dispatch("seek", int(offset)) @method() - def SetPosition(self, track_id: "o", position: "x"): + def SetPosition(self, track_id: "o", position: "x"): # noqa: F821 self._svc.dispatch("set_position", str(track_id), int(position)) @method() - def OpenUri(self, uri: "s"): + def OpenUri(self, uri: "s"): # noqa: F821 self._svc.dispatch("open_uri", str(uri)) # -- сигналы ----------------------------------------------------- @signal() - def Seeked(self, position: "x") -> "x": + def Seeked(self, position: "x") -> "x": # noqa: F821 return position # -- свойства ---------------------------------------------------- @dbus_property(access=PropertyAccess.READ) - def PlaybackStatus(self) -> "s": + def PlaybackStatus(self) -> "s": # noqa: F821 return self._svc.playback_status @dbus_property(access=PropertyAccess.READ) - def Metadata(self) -> "a{sv}": + def Metadata(self) -> "a{sv}": # noqa: F722 return self._svc.metadata @dbus_property(access=PropertyAccess.READ) - def Position(self) -> "x": + def Position(self) -> "x": # noqa: F821 return int(self._svc.position) @dbus_property(access=PropertyAccess.READ) - def MinimumRate(self) -> "d": + def MinimumRate(self) -> "d": # noqa: F821 return 1.0 @dbus_property(access=PropertyAccess.READ) - def MaximumRate(self) -> "d": + def MaximumRate(self) -> "d": # noqa: F821 return 1.0 @dbus_property(access=PropertyAccess.READ) - def CanGoNext(self) -> "b": + def CanGoNext(self) -> "b": # noqa: F821 return self._svc.can_next @dbus_property(access=PropertyAccess.READ) - def CanGoPrevious(self) -> "b": + def CanGoPrevious(self) -> "b": # noqa: F821 return self._svc.can_previous @dbus_property(access=PropertyAccess.READ) - def CanPlay(self) -> "b": + def CanPlay(self) -> "b": # noqa: F821 return self._svc.can_play @dbus_property(access=PropertyAccess.READ) - def CanPause(self) -> "b": + def CanPause(self) -> "b": # noqa: F821 return self._svc.can_pause @dbus_property(access=PropertyAccess.READ) - def CanSeek(self) -> "b": + def CanSeek(self) -> "b": # noqa: F821 return self._svc.can_seek @dbus_property(access=PropertyAccess.READ) - def CanControl(self) -> "b": + def CanControl(self) -> "b": # noqa: F821 return True @dbus_property(access=PropertyAccess.READWRITE) - def Volume(self) -> "d": + def Volume(self) -> "d": # noqa: F821 return float(self._svc.volume) @Volume.setter - def Volume(self, value: "d"): + def Volume(self, value: "d"): # noqa: F821 self._svc.dispatch("volume", float(value)) @dbus_property(access=PropertyAccess.READWRITE) - def LoopStatus(self) -> "s": + def LoopStatus(self) -> "s": # noqa: F821 return self._svc.loop_status @LoopStatus.setter - def LoopStatus(self, value: "s"): + def LoopStatus(self, value: "s"): # noqa: F821 self._svc.loop_status = str(value) @dbus_property(access=PropertyAccess.READWRITE) - def Shuffle(self) -> "b": + def Shuffle(self) -> "b": # noqa: F821 return self._svc.shuffle @Shuffle.setter - def Shuffle(self, value: "b"): + def Shuffle(self, value: "b"): # noqa: F821 self._svc.shuffle = bool(value) @dbus_property(access=PropertyAccess.READWRITE) - def Rate(self) -> "d": + def Rate(self) -> "d": # noqa: F821 return float(self._svc.rate) @Rate.setter - def Rate(self, value: "d"): + def Rate(self, value: "d"): # noqa: F821 self._svc.rate = float(value) @@ -290,7 +296,8 @@ class MprisService: def _emit(self, changed: dict[str, Any]): if not (self.enabled and self._loop): return - self._loop.call_soon_threadsafe(self._player.emit_properties_changed, changed) + self._loop.call_soon_threadsafe( + self._player.emit_properties_changed, changed) # -- обновление состояния (из основного потока) ---------------------- @@ -334,7 +341,8 @@ class MprisService: def notify_seeked(self, position_us: int): self.update_position(position_us) if self.enabled and self._loop: - self._loop.call_soon_threadsafe(self._player.Seeked, int(position_us)) + self._loop.call_soon_threadsafe( + self._player.Seeked, int(position_us)) def update_volume(self, volume: float): volume = max(0.0, min(1.0, float(volume))) diff --git a/mtsmusic.py b/mtsmusic.py index 603b8ff..717a38c 100644 --- a/mtsmusic.py +++ b/mtsmusic.py @@ -30,9 +30,15 @@ import requests PROJECT_DIR = Path(__file__).resolve().parent # Всё локальное храним рядом с проектом, а не в домашней директории. -os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", str(PROJECT_DIR / ".playwright")) +os.environ.setdefault( + "PLAYWRIGHT_BROWSERS_PATH", str( + PROJECT_DIR / ".playwright")) -CONFIG_DIR = Path(os.environ.get("MTSMUSIC_CONFIG_DIR", PROJECT_DIR / ".mts-music")) +CONFIG_DIR = Path( + os.environ.get( + "MTSMUSIC_CONFIG_DIR", + PROJECT_DIR / + ".mts-music")) CREDS_PATH = CONFIG_DIR / "credentials.json" BROWSER_DIR = CONFIG_DIR / "browser" @@ -54,10 +60,12 @@ _orig_getaddrinfo = socket.getaddrinfo def force_ipv4() -> None: - """api.music.yandex.net иногда недоступен по IPv6 — ходим только по IPv4.""" + """api.music.yandex.net иногда недоступен по IPv6 — ходим только + по IPv4.""" def getaddrinfo(host, port, family=0, type=0, proto=0, flags=0): - return _orig_getaddrinfo(host, port, socket.AF_INET, type, proto, flags) + return _orig_getaddrinfo( + host, port, socket.AF_INET, type, proto, flags) socket.getaddrinfo = getaddrinfo @@ -83,7 +91,12 @@ def load_credentials() -> dict: def save_credentials(data: dict) -> None: CONFIG_DIR.mkdir(parents=True, exist_ok=True) - CREDS_PATH.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + CREDS_PATH.write_text( + json.dumps( + data, + ensure_ascii=False, + indent=2), + encoding="utf-8") try: CREDS_PATH.chmod(0o600) except OSError: @@ -119,7 +132,9 @@ def get_auth_settings(reference_uri: str = "https://music.mts.ru/") -> dict: timeout=30, ) if response.status_code != 200: - die(f"auth_settings вернул HTTP {response.status_code}: {response.text[:300]}") + die( + f"auth_settings вернул HTTP {response.status_code}: " + f"{response.text[:300]}") return response.json() @@ -127,7 +142,9 @@ def cmd_login(args: argparse.Namespace) -> None: try: from playwright.sync_api import sync_playwright except ImportError: - die("не установлен playwright. Выполни: pip install -r requirements.txt") + die( + "не установлен playwright. Выполни: " + "pip install -r requirements.txt") settings = get_auth_settings() authorize_uri = settings["web_sso_authorize_uri"] @@ -155,16 +172,20 @@ def cmd_login(args: argparse.Namespace) -> None: token = None if token: break - # Токен может прийти в URL до того, как SPA успеет его сохранить. + # Токен может прийти в URL до того, как SPA успеет его + # сохранить. if "yatoken=" in page.url: - query = urllib.parse.parse_qs(urllib.parse.urlparse(page.url).query) + query = urllib.parse.parse_qs( + urllib.parse.urlparse(page.url).query) token = (query.get("yatoken") or [None])[0] if token: break page.wait_for_timeout(1000) if not token: - die("не дождался токена. Вход не завершён или истекло время ожидания.") + die( + "не дождался токена. Вход не завершён " + "или истекло время ожидания.") ya_uid = page.evaluate("localStorage.getItem('ya_uid')") user_id = page.evaluate("localStorage.getItem('userId')") @@ -213,7 +234,11 @@ class Api: def post(self, path: str, **params) -> dict: return self._request("POST", path, params=params) - def _request(self, method: str, path: str, params: dict | None = None) -> dict: + def _request( + self, + method: str, + path: str, + params: dict | None = None) -> dict: force_ipv4() last_error: Exception | None = None for attempt in range(3): @@ -235,18 +260,24 @@ class Api: raise AssertionError # для type checker if response.status_code == 401: - die("токен не принят (401). Выполни заново: python mtsmusic.py login") + die( + "токен не принят (401). Выполни заново: " + "python mtsmusic.py login") if response.status_code == 404: die(f"не найдено (404) на {path}") if response.status_code != 200: - die(f"HTTP {response.status_code} на {path}: {response.text[:300]}") + die( + f"HTTP {response.status_code} на {path}: " + f"{response.text[:300]}") return response.json() def get_url(self, url: str) -> dict: force_ipv4() response = self.session.get(url, headers=self._headers(), timeout=30) if response.status_code != 200: - die(f"HTTP {response.status_code} на {url[:120]}: {response.text[:300]}") + die( + f"HTTP {response.status_code} на {url[:120]}: " + f"{response.text[:300]}") return response.json() @@ -256,7 +287,10 @@ def require_token(api: Api, what: str) -> None: def bff_get(path: str) -> dict | None: - """Запрос к BFF MTS с сохранёнными cookie. None, если cookie нет/не подошли.""" + """Запрос к BFF MTS с сохранёнными cookie. + + None, если cookie нет/не подошли. + """ cookies = load_credentials().get("cookies") or [] if not cookies: return None @@ -339,17 +373,27 @@ def cmd_status(args: argparse.Namespace) -> None: payload = api.get("/account/mts/status") result = payload.get("result") or {} account = result.get("account") or {} - subscriptions = ((result.get("masterhub") or {}).get("activeSubscriptions")) or [] + subscriptions = ((result.get("masterhub") or {}).get( + "activeSubscriptions")) or [] # BFF даёт человекочитаемое название тарифа. bff_profile = bff_get("/v1/user-profile") or {} profile = bff_profile.get("profile") or {} bff_subscriptions = bff_profile.get("subscriptions") or [] + alias = ( + profile.get("alias") + or account.get("fullName") + or account.get("displayName") + or account.get("login") + or "?" + ) + phone = profile.get("phone") or account.get("phone") or "?" + has_sub = subscriptions or bff_subscriptions print(f"UID: {account.get('uid', '?')}") - print(f"Имя: {profile.get('alias') or account.get('fullName') or account.get('displayName') or account.get('login') or '?'}") - print(f"Телефон: {profile.get('phone') or account.get('phone') or '?'}") - print(f"Подписка: {'активна' if subscriptions or bff_subscriptions else 'нет'}") + print(f"Имя: {alias}") + print(f"Телефон: {phone}") + print(f"Подписка: {'активна' if has_sub else 'нет'}") if bff_subscriptions: for sub in bff_subscriptions: name = sub.get("name") or sub.get("id") or "?" @@ -357,13 +401,15 @@ def cmd_status(args: argparse.Namespace) -> None: print(f" - {name}{f' (до {until})' if until else ''}") else: for sub in subscriptions: - name = sub.get("name") or sub.get("productName") or sub.get("subscriptionId") or sub.get("id") + name = sub.get("name") or sub.get("productName") or sub.get( + "subscriptionId") or sub.get("id") until = sub.get("expires") or sub.get("expirationDate") print(f" - {name}{f' (до {until})' if until else ''}") # Дописываем uid в креды, если при логине SPA не успела его сохранить. creds = load_credentials() - if creds and creds.get("ya_token") == api.token and not creds.get("ya_uid"): + if creds and creds.get( + "ya_token") == api.token and not creds.get("ya_uid"): creds["ya_uid"] = str(account.get("uid") or "") save_credentials(creds) @@ -381,7 +427,9 @@ def get_uid(api: Api) -> str: result = api.get("/account/mts/status").get("result") or {} uid = (result.get("account") or {}).get("uid") if not uid: - die("не удалось определить uid аккаунта. Выполни: python mtsmusic.py login") + die( + "не удалось определить uid аккаунта. Выполни: " + "python mtsmusic.py login") creds = load_credentials() or {} creds["ya_uid"] = str(uid) save_credentials(creds) @@ -408,21 +456,25 @@ def track_album_ref(track_id: str, album: str | None = None) -> str: def fetch_mix_batch(api: Api, queue: list[str] | None = None) -> dict: params = {"queue": ",".join(queue)} if queue else {} - return api.get(f"/rotor/station/{WAVE_STATION}/tracks", **params).get("result") or {} + return api.get( + f"/rotor/station/{WAVE_STATION}/tracks", + **params).get("result") or {} def fetch_liked_tracks(api: Api, uid: str, limit: int = 300) -> list[dict]: """Избранные треки с деталями: /tracks принимает не больше 50 id за раз.""" payload = api.get(f"/users/{uid}/likes/tracks") - refs = ((payload.get("result") or {}).get("library") or {}).get("tracks") or [] + refs = ((payload.get("result") or {}).get( + "library") or {}).get("tracks") or [] ids = [str(ref.get("id")) for ref in refs[:limit]] tracks: list[dict] = [] for start in range(0, len(ids), 50): - chunk = ids[start : start + 50] + chunk = ids[start: start + 50] if chunk: tracks.extend( - api.get("/tracks", trackIds=",".join(chunk)).get("result") or [] - ) + api.get( + "/tracks", + trackIds=",".join(chunk)).get("result") or []) return tracks @@ -479,7 +531,8 @@ def cmd_mix(args: argparse.Namespace) -> None: print_mix_item(index, item) if args.play: - play_items(api, [item.get("track") or {} for item in collected], args.player) + play_items(api, [item.get("track") or {} + for item in collected], args.player) def cmd_like(args: argparse.Namespace) -> None: @@ -517,15 +570,16 @@ def cmd_likes(args: argparse.Namespace) -> None: limit = max(1, args.number) liked = liked[:limit] - refs = ",".join(str(item.get("id")) for item in liked) details: dict[str, dict] = {} # /tracks отдаёт не больше сотни id за раз — бьём на порции. ids = [str(item.get("id")) for item in liked] for start in range(0, len(ids), 50): - chunk = ids[start : start + 50] + chunk = ids[start: start + 50] try: - tracks = api.get("/tracks", trackIds=",".join(chunk)).get("result") or [] + tracks = api.get( + "/tracks", + trackIds=",".join(chunk)).get("result") or [] except SystemExit: tracks = [] for track in tracks: @@ -541,13 +595,18 @@ def cmd_likes(args: argparse.Namespace) -> None: print(f"{index:>3}. {track_id:>12} (альбом {album})") -def lyrics_response(api: Api, track_id: str, fmt: str = "TEXT") -> requests.Response: +def lyrics_response( + api: Api, + track_id: str, + fmt: str = "TEXT") -> requests.Response: """Запрос текста с подписью из веб-клиента MTS.""" timestamp = int(time.time()) message = f"{track_id}{timestamp}" signature = base64.b64encode( - hmac.new(LYRICS_SIGN_KEY.encode(), message.encode(), hashlib.sha256).digest() - ).decode() + hmac.new( + LYRICS_SIGN_KEY.encode(), + message.encode(), + hashlib.sha256).digest()).decode() force_ipv4() return api.session.get( @@ -603,7 +662,11 @@ def lrclib_lyrics(track: dict, want_sync: bool) -> str | None: candidates: list[dict] = [] for params in queries: try: - response = requests.get(LRCLIB_URL, params=params, headers=headers, timeout=20) + response = requests.get( + LRCLIB_URL, + params=params, + headers=headers, + timeout=20) except requests.RequestException: continue if response.status_code != 200: @@ -623,13 +686,18 @@ def lrclib_lyrics(track: dict, want_sync: bool) -> str | None: duration = (track.get("durationMs") or 0) / 1000 def score(item: dict) -> float: - diff = abs(float(item.get("duration") or 0) - duration) if duration else 0 - exact = 0 if (item.get("trackName") or "").casefold() == title.casefold() else 50 + diff = abs( + float( + item.get("duration") or 0) - + duration) if duration else 0 + exact = 0 if (item.get("trackName") or "").casefold( + ) == title.casefold() else 50 return diff + exact usable = [item for item in candidates if (item.get(field) or "").strip()] if not usable and want_sync: - usable = [item for item in candidates if (item.get("plainLyrics") or "").strip()] + usable = [item for item in candidates if ( + item.get("plainLyrics") or "").strip()] field = "plainLyrics" if not usable: return None @@ -667,7 +735,9 @@ def cmd_lyrics(args: argparse.Namespace) -> None: if args.output: output = Path(args.output) output.write_text(text, encoding="utf-8") - print(f"Сохранено: {output} ({len(text)} символов, источник: {source})") + print( + f"Сохранено: {output} ({ + len(text)} символов, источник: {source})") else: if source != "MTS": print(f"# источник текста: {source}", file=sys.stderr) @@ -690,9 +760,10 @@ def resolve_stream(api: Api, track_id: str) -> tuple[str, dict]: # Лучший вариант: не preview и максимальный битрейт. items.sort( - key=lambda item: (not item.get("preview", False), item.get("bitrateInKbps", 0)), - reverse=True, - ) + key=lambda item: ( + not item.get( + "preview", False), item.get( + "bitrateInKbps", 0)), reverse=True, ) item = items[0] download_url = item["downloadInfoUrl"] @@ -705,7 +776,8 @@ def resolve_stream(api: Api, track_id: str) -> tuple[str, dict]: host = meta["host"] timestamp = meta["ts"] - digest = hashlib.md5((STREAM_SIGN_SECRET + path + signature).encode()).hexdigest() + digest = hashlib.md5( + (STREAM_SIGN_SECRET + path + signature).encode()).hexdigest() stream_url = f"https://{host}/get-mp3/{digest}/{timestamp}{path}" return stream_url, item @@ -714,7 +786,9 @@ def cmd_url(args: argparse.Namespace) -> None: api = Api() url, item = resolve_stream(api, args.track_id) if item.get("preview"): - print("# preview: полный трек недоступен без активной подписки", file=sys.stderr) + print( + "# preview: полный трек недоступен без активной подписки", + file=sys.stderr) print(url) @@ -748,14 +822,20 @@ def play_url(player: str, url: str) -> None: raise -def play_items(api: Api, tracks: list[dict], explicit_player: str | None = None) -> None: +def play_items( + api: Api, + tracks: list[dict], + explicit_player: str | None = None) -> None: player = pick_player(explicit_player) if not player: - die("не найден mpv/ffplay. Установи один из них или используй: mtsmusic.py url ") + die( + "не найден mpv/ffplay. Установи один из них или используй: " + "mtsmusic.py url ") for index, track in enumerate(tracks, 1): track_id = str(track.get("id")) - print(f"\n▶ {index}/{len(tracks)} {fmt_track(track)}", file=sys.stderr) + print(f"\n▶ {index}/{len(tracks)} {fmt_track(track)}", + file=sys.stderr) try: url, item = resolve_stream(api, track_id) except SystemExit: @@ -775,11 +855,15 @@ def cmd_play(args: argparse.Namespace) -> None: print_track_info(api, args.track_id) url, item = resolve_stream(api, args.track_id) if item.get("preview"): - print("Внимание: это preview-фрагмент (нет активной подписки).", file=sys.stderr) + print( + "Внимание: это preview-фрагмент (нет активной подписки).", + file=sys.stderr) player = pick_player(args.player) if not player: - die("не найден mpv/ffplay. Установи один из них или используй: mtsmusic.py url ") + die( + "не найден mpv/ffplay. Установи один из них или используй: " + "mtsmusic.py url ") try: play_url(player, url) @@ -794,7 +878,12 @@ def cmd_download(args: argparse.Namespace) -> None: force_ipv4() output = Path(args.output or f"{args.track_id}.mp3") - with requests.get(url, stream=True, headers={"User-Agent": USER_AGENT}, timeout=60) as response: + with requests.get( + url, + stream=True, + headers={"User-Agent": USER_AGENT}, + timeout=60, + ) as response: response.raise_for_status() total = int(response.headers.get("Content-Length") or 0) done = 0 @@ -805,7 +894,12 @@ def cmd_download(args: argparse.Namespace) -> None: handle.write(chunk) done += len(chunk) if total: - print(f"\r{done * 100 // total:3d}% {done // 1024} KiB", end="", file=sys.stderr) + percent = done * 100 // total + kib = done // 1024 + print( + f"\r{percent:3d}% {kib} KiB", + end="", + file=sys.stderr) print(file=sys.stderr) print(f"Сохранено: {output} ({output.stat().st_size} байт)") @@ -821,16 +915,24 @@ def cmd_tui(args: argparse.Namespace) -> None: return if not sys.stdin.isatty() or not sys.stdout.isatty(): die( - "TUI нужно запускать в обычном терминале (stdin/stdout не TTY).\n" - "Запусти прямо в окне терминала: .venv/bin/python mtsmusic.py tui\n" + "TUI нужно запускать в обычном терминале " + "(stdin/stdout не TTY).\n" + "Запусти прямо в окне терминала: " + ".venv/bin/python mtsmusic.py tui\n" "Если это IDE/панель вывода — открой настоящий терминал." ) try: from tui import MusicTui except ImportError as exc: - die(f"для TUI нужен textual и pillow: {exc}. Установи: pip install -r requirements.txt") + die( + f"для TUI нужен textual и pillow: {exc}. Установи: " + "pip install -r requirements.txt" + ) try: - MusicTui(player=args.player).run(inline=args.inline, mouse=not args.no_mouse) + MusicTui(player=args.player).run( + inline=args.inline, + mouse=not args.no_mouse, + ) except KeyboardInterrupt: pass except Exception as exc: # TUI не должен молча исчезать @@ -869,7 +971,9 @@ def cmd_settings(args: argparse.Namespace) -> None: key, value = pair.split("=", 1) key = key.strip() if key not in DEFAULTS: - die(f"неизвестная настройка: {key}. Доступны: {', '.join(DEFAULTS)}") + die( + f"неизвестная настройка: {key}. Доступны: { + ', '.join(DEFAULTS)}") updates[key] = value current = save_settings(updates) saved = True @@ -887,7 +991,11 @@ def print_tui_check() -> None: import platform print(f"Python: {sys.executable}") - print(f"Версия: {platform.python_version()} ({platform.system()} {platform.release()})") + print( + f"Версия: { + platform.python_version()} ({ + platform.system()} { + platform.release()})") print(f"TERM: {os.environ.get('TERM', '?')}") print(f"COLORTERM: {os.environ.get('COLORTERM', '?')}") print(f"stdout TTY: {sys.stdout.isatty()}") @@ -908,7 +1016,9 @@ def print_tui_check() -> None: print("pillow: НЕ УСТАНОВЛЕН") for name in ("ffplay", "mpv"): print(f"{name}: {shutil.which(name) or 'нет'}") - print(f"Креды: {'есть' if CREDS_PATH.exists() else 'нет'} ({CREDS_PATH})") + print( + f"Креды: { + 'есть' if CREDS_PATH.exists() else 'нет'} ({CREDS_PATH})") if size.columns < 90 or size.lines < 30: print("\nСовет: окно маленькое, TUI лучше смотреть при >= 90x30.") @@ -920,9 +1030,14 @@ def build_parser() -> argparse.ArgumentParser: ) sub = parser.add_subparsers(dest="command", required=True) - login = sub.add_parser("login", help="войти через браузер и сохранить токен") - login.add_argument("--headless", action="store_true", help="без окна браузера (капчу не пройти)") - login.add_argument("--timeout", type=int, default=600, help="секунд на вход (по умолчанию 600)") + login = sub.add_parser( + "login", help="войти через браузер и сохранить токен") + login.add_argument( + "--headless", + action="store_true", + help="без окна браузера (капчу не пройти)") + login.add_argument("--timeout", type=int, default=600, + help="секунд на вход (по умолчанию 600)") login.set_defaults(func=cmd_login) status = sub.add_parser("status", help="показать аккаунт и подписку") @@ -930,12 +1045,21 @@ def build_parser() -> argparse.ArgumentParser: search = sub.add_parser("search", help="поиск треков") search.add_argument("query", help="поисковый запрос") - search.add_argument("-n", "--limit", type=int, default=10, help="сколько результатов (по умолчанию 10)") + search.add_argument( + "-n", + "--limit", + type=int, + default=10, + help="сколько результатов (по умолчанию 10)") search.set_defaults(func=cmd_search) mix = sub.add_parser("mix", help="«Моя волна»: подборка треков") - mix.add_argument("-n", "--number", type=int, default=10, help="сколько треков (по умолчанию 10)") - mix.add_argument("--play", action="store_true", help="сразу воспроизвести подборку") + mix.add_argument("-n", "--number", type=int, default=10, + help="сколько треков (по умолчанию 10)") + mix.add_argument( + "--play", + action="store_true", + help="сразу воспроизвести подборку") mix.add_argument("--player", help="путь к mpv/ffplay") mix.set_defaults(func=cmd_mix) @@ -946,16 +1070,27 @@ def build_parser() -> argparse.ArgumentParser: unlike = sub.add_parser("unlike", help="убрать трек из избранного") unlike.add_argument("track_id") - unlike.add_argument("--album", help="id альбома (если нужно переопределить)") + unlike.add_argument( + "--album", + help="id альбома (если нужно переопределить)") unlike.set_defaults(func=cmd_unlike) likes = sub.add_parser("likes", help="показать избранные треки") - likes.add_argument("-n", "--number", type=int, default=50, help="сколько показать (по умолчанию 50)") + likes.add_argument( + "-n", + "--number", + type=int, + default=50, + help="сколько показать (по умолчанию 50)") likes.set_defaults(func=cmd_likes) - lyrics = sub.add_parser("lyrics", help="показать или сохранить текст песни") + lyrics = sub.add_parser( + "lyrics", help="показать или сохранить текст песни") lyrics.add_argument("track_id") - lyrics.add_argument("--lrc", action="store_true", help="формат LRC с таймкодами") + lyrics.add_argument( + "--lrc", + action="store_true", + help="формат LRC с таймкодами") lyrics.add_argument("-o", "--output", help="сохранить текст в файл") lyrics.add_argument( "--no-fallback", @@ -964,21 +1099,31 @@ def build_parser() -> argparse.ArgumentParser: ) lyrics.set_defaults(func=cmd_lyrics) - tui = sub.add_parser("tui", help="полноценный TUI-плеер (мышь, обложка, избранное)") + tui = sub.add_parser( + "tui", help="полноценный TUI-плеер (мышь, обложка, избранное)") tui.add_argument("--player", help="путь к mpv/ffplay") - tui.add_argument("--check", action="store_true", help="диагностика окружения и выход") + tui.add_argument( + "--check", + action="store_true", + help="диагностика окружения и выход") tui.add_argument( "--inline", action="store_true", - help="без alternate screen: рисовать прямо в терминале (если экран пустой)", + help=( + "без alternate screen: рисовать прямо в терминале " + "(если экран пустой)"), ) - tui.add_argument("--no-mouse", action="store_true", help="отключить поддержку мыши") + tui.add_argument( + "--no-mouse", + action="store_true", + help="отключить поддержку мыши") tui.set_defaults(func=cmd_tui) qt = sub.add_parser("qt", help="настольное Qt-приложение (PySide6)") qt.set_defaults(func=cmd_qt) - settings = sub.add_parser("settings", help="показать или изменить настройки") + settings = sub.add_parser("settings", + help="показать или изменить настройки") settings.add_argument( "-s", "--set", diff --git a/player.py b/player.py index 18838e5..5476f2e 100644 --- a/player.py +++ b/player.py @@ -67,7 +67,8 @@ class MusicPlayer: return command + [url] async def set_volume(self, volume: int) -> None: - """Меняет громкость; ffplay/mpv не умеют на лету, поэтому перезапускаем.""" + """Меняет громкость; ffplay/mpv не умеют на лету, поэтому + перезапускаем.""" self.volume = max(0, min(100, int(volume))) if self.url and self.is_playing(): position = self.elapsed() @@ -145,7 +146,12 @@ class MusicPlayer: if not self.proc: return 0.0 now = self.paused_at if self.paused else time.monotonic() - return max(0.0, self.offset + now - self.started_at - self.paused_total) + return max( + 0.0, + self.offset + + now - + self.started_at - + self.paused_total) async def seek(self, delta: float, duration: float) -> None: """Seek by delta seconds. Restarts ffplay at the new position.""" diff --git a/qt_app.py b/qt_app.py index ccf7d7c..ae18375 100644 --- a/qt_app.py +++ b/qt_app.py @@ -111,7 +111,9 @@ QLabel { background: transparent; } #albumLabel { color: #6f6f82; font-size: 12px; } #timeLabel { color: #8b8b9e; font-size: 11px; } #queueLabel { color: #6f6f82; font-size: 11px; } -QSlider::groove:horizontal { height: 4px; background: #2a2a36; border-radius: 2px; } +QSlider::groove:horizontal { + height: 4px; background: #2a2a36; border-radius: 2px; +} QSlider::sub-page:horizontal { background: #ff3b5c; border-radius: 2px; } QSlider::handle:horizontal { width: 12px; margin: -4px 0; border-radius: 6px; background: #ffffff; @@ -119,24 +121,32 @@ QSlider::handle:horizontal { QSlider::handle:horizontal:hover { background: #ffd9df; } #playBtn { background: #ff3b5c; border: none; border-radius: 38px; } #playBtn:hover { background: #ff5574; } -#prevBtn, #nextBtn { background: rgba(255,255,255,0.05); border: none; border-radius: 22px; } +#prevBtn, #nextBtn { + background: rgba(255,255,255,0.05); border: none; border-radius: 22px; +} #prevBtn:hover, #nextBtn:hover { background: rgba(255,255,255,0.11); } #actionBtn { - background: rgba(255,255,255,0.05); border: 1px solid rgba(255,255,255,0.08); border-radius: 16px; + background: rgba(255,255,255,0.05); + border: 1px solid rgba(255,255,255,0.08); border-radius: 16px; padding: 8px 18px; font-size: 12px; } #actionBtn:hover { background: rgba(255,255,255,0.09); } -#actionBtn[liked="true"] { color: #ff8296; border-color: rgba(255,59,92,0.40); } +#actionBtn[liked="true"] { + color: #ff8296; border-color: rgba(255,59,92,0.40); +} #menuBtn:focus, #prevBtn:focus, #nextBtn:focus, #playBtn:focus, #actionBtn:focus, #backBtn:focus, #searchBtn:focus { border: 2px solid #ffffff; } QSlider:focus { border: 1px solid #ffffff; border-radius: 4px; } -#drawer { background: #16161f; border-right: 1px solid rgba(255,255,255,0.06); } +#drawer { + background: #16161f; border-right: 1px solid rgba(255,255,255,0.06); +} #drawerTitle { font-size: 18px; font-weight: 700; padding: 4px 2px 10px 2px; } #drawerBrand { color: #6f6f82; font-size: 11px; padding: 0 2px 12px 2px; } #drawer QPushButton { - text-align: left; background: transparent; border: none; border-radius: 10px; + text-align: left; background: transparent; + border: none; border-radius: 10px; padding: 11px 12px; font-size: 14px; color: #e8e8f0; } #drawer QPushButton:hover { background: rgba(255,255,255,0.06); } @@ -151,7 +161,8 @@ QSlider:focus { border: 1px solid #ffffff; border-radius: 4px; } QListWidget { background: transparent; border: none; outline: 0; } QListWidget::item { border: none; } QLineEdit { - background: #1b1b25; border: 1px solid rgba(255,255,255,0.07); border-radius: 13px; + background: #1b1b25; border: 1px solid rgba(255,255,255,0.07); + border-radius: 13px; padding: 9px 12px; font-size: 14px; selection-background-color: #ff3b5c; } QLineEdit:focus { border-color: rgba(255,59,92,0.53); } @@ -170,7 +181,8 @@ QComboBox QAbstractItemView { QCheckBox { color: #dcdce6; font-size: 13px; spacing: 8px; } QCheckBox::indicator { width: 16px; height: 16px; } QCheckBox::indicator:unchecked { - border: 1px solid rgba(255,255,255,0.35); border-radius: 4px; background: #1b1b25; + border: 1px solid rgba(255,255,255,0.35); + border-radius: 4px; background: #1b1b25; } #searchBtn { background: #ff3b5c; border: none; border-radius: 13px; @@ -188,14 +200,16 @@ QScrollBar::handle:vertical { QScrollBar::add-line, QScrollBar::sub-line { height: 0; } QScrollBar::add-page, QScrollBar::sub-page { background: transparent; } #toast { - background: #23232f; border: 1px solid rgba(255,255,255,0.09); border-radius: 14px; + background: #23232f; border: 1px solid rgba(255,255,255,0.09); + border-radius: 14px; padding: 10px 16px; color: #f2f2f7; font-size: 12px; } """ def media_icon(kind: str, size: int = 64) -> QIcon: - """Рисует белые иконки плеера, чтобы они выглядели одинаково в любой теме.""" + """Рисует белые иконки плеера, чтобы они выглядели одинаково + в любой теме.""" pixmap = QPixmap(size, size) pixmap.fill(Qt.GlobalColor.transparent) painter = QPainter(pixmap) @@ -211,10 +225,31 @@ def media_icon(kind: str, size: int = 64) -> QIcon: path.closeSubpath() painter.drawPath(path) elif kind == "pause": - painter.drawRoundedRect(QRectF(20 * s, 16 * s, 9 * s, 32 * s), 2 * s, 2 * s) - painter.drawRoundedRect(QRectF(35 * s, 16 * s, 9 * s, 32 * s), 2 * s, 2 * s) + painter.drawRoundedRect( + QRectF( + 20 * s, + 16 * s, + 9 * s, + 32 * s), + 2 * s, + 2 * s) + painter.drawRoundedRect( + QRectF( + 35 * s, + 16 * s, + 9 * s, + 32 * s), + 2 * s, + 2 * s) elif kind == "prev": - painter.drawRoundedRect(QRectF(15 * s, 14 * s, 7 * s, 36 * s), 2 * s, 2 * s) + painter.drawRoundedRect( + QRectF( + 15 * s, + 14 * s, + 7 * s, + 36 * s), + 2 * s, + 2 * s) path = QPainterPath() path.moveTo(52 * s, 14 * s) path.lineTo(52 * s, 50 * s) @@ -222,7 +257,14 @@ def media_icon(kind: str, size: int = 64) -> QIcon: path.closeSubpath() painter.drawPath(path) elif kind == "next": - painter.drawRoundedRect(QRectF(42 * s, 14 * s, 7 * s, 36 * s), 2 * s, 2 * s) + painter.drawRoundedRect( + QRectF( + 42 * s, + 14 * s, + 7 * s, + 36 * s), + 2 * s, + 2 * s) path = QPainterPath() path.moveTo(12 * s, 14 * s) path.lineTo(12 * s, 50 * s) @@ -320,14 +362,14 @@ class CoverView(QWidget): gradient.setColorAt(0.0, QColor("#252533")) gradient.setColorAt(1.0, QColor("#15151d")) painter.fillRect(rect, gradient) - painter.setPen(QColor(255,255,255,48)) + painter.setPen(QColor(255, 255, 255, 48)) font = QFont() font.setPointSize(64) painter.setFont(font) painter.drawText(rect, Qt.AlignmentFlag.AlignCenter, "♪") painter.setClipping(False) - painter.setPen(QPen(QColor(255,255,255,31), 1)) + painter.setPen(QPen(QColor(255, 255, 255, 31), 1)) painter.setBrush(Qt.BrushStyle.NoBrush) painter.drawRoundedRect(rect.adjusted(0.5, 0.5, -0.5, -0.5), 20, 20) painter.end() @@ -422,7 +464,12 @@ class TrackDelegate(QStyledItemDelegate): painter.setPen(QColor("#6f6f82")) painter.drawText( - QRectF(rect.right() - duration_width, rect.top(), duration_width, 22), + QRectF( + rect.right() - + duration_width, + rect.top(), + duration_width, + 22), Qt.AlignmentFlag.AlignRight | Qt.AlignmentFlag.AlignVCenter, duration, ) @@ -539,7 +586,8 @@ class QtMusicApp(QMainWindow): self._update_cover_size() self.play_btn.setFocus() - # MPRIS: система видит плеер как источник звука (панель GNOME, playerctl). + # MPRIS: система видит плеер как источник звука (панель GNOME, + # playerctl). self.mpris_bridge = MprisBridge(self) self.mpris_bridge.play.connect(self.mpris_play) self.mpris_bridge.pause.connect(self.mpris_pause) @@ -608,9 +656,12 @@ class QtMusicApp(QMainWindow): ) self.pages.addWidget(self.wave_page) - self.search_page, self.search_input, self.search_list, self.search_status = ( - self._build_search_page() - ) + ( + self.search_page, + self.search_input, + self.search_list, + self.search_status, + ) = self._build_search_page() self.search_list._queue_name = "Поиск" self.search_list._tracks = [] self.search_list.itemDoubleClicked.connect( @@ -618,9 +669,12 @@ class QtMusicApp(QMainWindow): ) self.pages.addWidget(self.search_page) - self.lyrics_page, self.lyrics_title, self.lyrics_status, self.lyrics_body = ( - self._build_lyrics_page() - ) + ( + self.lyrics_page, + self.lyrics_title, + self.lyrics_status, + self.lyrics_body, + ) = self._build_lyrics_page() self.pages.addWidget(self.lyrics_page) self.settings_page = self._build_settings_page() @@ -708,8 +762,10 @@ class QtMusicApp(QMainWindow): self.slider.setRange(0, 0) self.slider.seekRequested.connect(self.on_slider_seek) self.slider.actionTriggered.connect(self.on_slider_action) - self.slider.scrubStarted.connect(lambda: setattr(self, "_seeking", True)) - self.slider.scrubFinished.connect(lambda: setattr(self, "_seeking", False)) + self.slider.scrubStarted.connect( + lambda: setattr(self, "_seeking", True)) + self.slider.scrubFinished.connect( + lambda: setattr(self, "_seeking", False)) layout.addWidget(self.slider) layout.addSpacing(3) @@ -794,7 +850,8 @@ class QtMusicApp(QMainWindow): row.addStretch(1) return row - def _build_list_page(self, title: str) -> tuple[QWidget, QListWidget, QLabel]: + def _build_list_page( + self, title: str) -> tuple[QWidget, QListWidget, QLabel]: page = QWidget() layout = QVBoxLayout(page) layout.setContentsMargins(16, 10, 16, 12) @@ -813,7 +870,8 @@ class QtMusicApp(QMainWindow): layout.addWidget(widget, 1) return page, widget, status - def _build_search_page(self) -> tuple[QWidget, QLineEdit, QListWidget, QLabel]: + def _build_search_page( + self) -> tuple[QWidget, QLineEdit, QListWidget, QLabel]: page = QWidget() layout = QVBoxLayout(page) layout.setContentsMargins(16, 10, 16, 12) @@ -930,15 +988,21 @@ class QtMusicApp(QMainWindow): self.start_combo.setObjectName("settingCombo") for key in START_QUEUES: self.start_combo.addItem(START_QUEUE_LABELS[key], key) - start_index = self.start_combo.findData(self.settings.get("start_queue", "wave")) + start_index = self.start_combo.findData( + self.settings.get("start_queue", "wave")) self.start_combo.setCurrentIndex(max(0, start_index)) self.start_combo.currentIndexChanged.connect(self.on_start_changed) start_row.addWidget(start_label) start_row.addWidget(self.start_combo, 1) layout.addLayout(start_row) - self.wave_check = QCheckBox("Продолжать «Мою волну» с последних треков") - self.wave_check.setChecked(bool(self.settings.get("wave_from_history", True))) + self.wave_check = QCheckBox( + "Продолжать «Мою волну» с последних треков") + self.wave_check.setChecked( + bool( + self.settings.get( + "wave_from_history", + True))) self.wave_check.toggled.connect(self.on_wave_from_history_toggled) layout.addWidget(self.wave_check) @@ -1147,7 +1211,11 @@ class QtMusicApp(QMainWindow): self.set_top_status("Загрузка…") self.run_task(self._fetch_initial, self.apply_initial, self.on_error) - def _fetch_initial(self) -> tuple[str, list[dict], set[str], list[dict], str]: + def _fetch_initial(self) -> tuple[str, + list[dict], + set[str], + list[dict], + str]: api = self.api uid = self.uid or get_uid(api) # Продолжаем волну с последних прослушанных треков, а не с начала. @@ -1170,7 +1238,8 @@ class QtMusicApp(QMainWindow): liked = {str(track.get("id")) for track in start} else: payload = api.get(f"/users/{uid}/likes/tracks") - refs = ((payload.get("result") or {}).get("library") or {}).get("tracks") or [] + refs = ((payload.get("result") or {}).get( + "library") or {}).get("tracks") or [] liked = {str(ref.get("id")) for ref in refs} if want == "history": start, name = list(self.history), "История" @@ -1214,14 +1283,17 @@ class QtMusicApp(QMainWindow): def update_like_button(self, liked: bool | None = None) -> None: if liked is None: - liked = bool(self.current and str(self.current.get("id")) in self.liked_ids) + liked = bool( + self.current and str( + self.current.get("id")) in self.liked_ids) self.like_btn.setText("♥ В избранном" if liked else "♡ В избранное") self.like_btn.setProperty("liked", "true" if liked else "false") self.like_btn.style().unpolish(self.like_btn) self.like_btn.style().polish(self.like_btn) def update_play_button(self, *args) -> None: - playing = self.player.playbackState() == QMediaPlayer.PlaybackState.PlayingState + state = self.player.playbackState() + playing = state == QMediaPlayer.PlaybackState.PlayingState self.play_btn.setIcon(media_icon("pause" if playing else "play", 80)) self._update_mpris_status() @@ -1239,7 +1311,9 @@ class QtMusicApp(QMainWindow): return self.cover.set_pixmap(None) request = QNetworkRequest(QUrl(url)) - request.setHeader(QNetworkRequest.KnownHeaders.UserAgentHeader, USER_AGENT) + request.setHeader( + QNetworkRequest.KnownHeaders.UserAgentHeader, + USER_AGENT) reply = self.covers.get(request) def done() -> None: @@ -1260,7 +1334,10 @@ class QtMusicApp(QMainWindow): # -- воспроизведение -------------------------------------------------- - def play_from_list(self, widget: QListWidget, item: QListWidgetItem) -> None: + def play_from_list( + self, + widget: QListWidget, + item: QListWidgetItem) -> None: tracks = getattr(widget, "_tracks", []) index = widget.row(item) if not (0 <= index < len(tracks)): @@ -1362,7 +1439,8 @@ class QtMusicApp(QMainWindow): self.mpris.notify_seeked(value * 1000) def on_slider_action(self, action: int) -> None: - # Колесо и стрелки, когда слайдер в фокусе (клик обрабатывает ClickSlider). + # Колесо и стрелки, когда слайдер в фокусе (клик обрабатывает + # ClickSlider). if self.slider.maximum() > 0: value = self.slider.value() self.player.setPosition(value) @@ -1374,7 +1452,8 @@ class QtMusicApp(QMainWindow): if self.player.source().isEmpty(): return duration = self.slider.maximum() - position = max(0, min(self.player.position() + delta_ms, max(0, duration - 500))) + position = max(0, min(self.player.position() + + delta_ms, max(0, duration - 500))) self.player.setPosition(position) self.slider.setValue(position) self.elapsed_label.setText(fmt_millis(position)) @@ -1384,11 +1463,13 @@ class QtMusicApp(QMainWindow): # -- MPRIS ------------------------------------------------------------ def mpris_play(self) -> None: - if self.player.playbackState() != QMediaPlayer.PlaybackState.PlayingState: + state = self.player.playbackState() + if state != QMediaPlayer.PlaybackState.PlayingState: self.toggle_play() def mpris_pause(self) -> None: - if self.player.playbackState() == QMediaPlayer.PlaybackState.PlayingState: + state = self.player.playbackState() + if state == QMediaPlayer.PlaybackState.PlayingState: self.toggle_play() def mpris_stop(self) -> None: @@ -1401,9 +1482,8 @@ class QtMusicApp(QMainWindow): if self.player.source().isEmpty(): return duration = self.slider.maximum() - position = max( - 0, min(self.player.position() + offset_us // 1000, max(0, duration - 500)) - ) + position = max(0, min(self.player.position() + + offset_us // 1000, max(0, duration - 500))) self.player.setPosition(position) self.slider.setValue(position) self.elapsed_label.setText(fmt_millis(position)) @@ -1491,13 +1571,18 @@ class QtMusicApp(QMainWindow): if item.get("type") == "track" and item.get("track") ] - self.run_task(fetch, lambda tracks: self.apply_more_wave(tracks, then_play), self.on_error) + self.run_task( + fetch, lambda tracks: self.apply_more_wave( + tracks, then_play), self.on_error) def apply_more_wave(self, tracks: list[dict], then_play: int) -> None: known = {str(track.get("id")) for track in self.wave} - added = [track for track in tracks if str(track.get("id")) not in known] + added = [ + track for track in tracks if str( + track.get("id")) not in known] self.wave.extend(added) - self.set_top_status(f"Волна · {len(self.wave)} · ♥ {len(self.liked_ids)}") + self.set_top_status( + f"Волна · {len(self.wave)} · ♥ {len(self.liked_ids)}") if self.queue_name == "Моя волна": self.queue = list(self.wave) if then_play < len(self.queue) and added: @@ -1536,7 +1621,12 @@ class QtMusicApp(QMainWindow): ) return track_id - self.run_task(request, lambda _: self.apply_like(track, not liked), self.like_failed) + self.run_task( + request, + lambda _: self.apply_like( + track, + not liked), + self.like_failed) def apply_like(self, track: dict, liked: bool) -> None: track_id = str(track.get("id")) @@ -1548,7 +1638,8 @@ class QtMusicApp(QMainWindow): if self.current and str(self.current.get("id")) == track_id: self.update_like_button(liked) self.set_top_status(f"{self.queue_name} · ♥ {len(self.liked_ids)}") - self.toast("♥ Добавлено в избранное" if liked else "♡ Убрано из избранного") + self.toast( + "♥ Добавлено в избранное" if liked else "♡ Убрано из избранного") if self.pages.currentWidget() is self.favorites_page: self.load_favorites(force=True) @@ -1664,14 +1755,20 @@ class QtMusicApp(QMainWindow): if self._favorites_loading: return if self.favorites and not force: - self.fill_track_list(self.favorites_list, self.favorites, "Избранное") + self.fill_track_list( + self.favorites_list, + self.favorites, + "Избранное") self.favorites_status.setText(f"{len(self.favorites)} трек(ов)") return self._favorites_loading = True self.favorites_status.setText("Загружаю…") if not self.favorites: self.favorites_list.clear() - self.run_task(self._fetch_favorites, self.apply_favorites, self.favorites_failed) + self.run_task( + self._fetch_favorites, + self.apply_favorites, + self.favorites_failed) def _fetch_favorites(self) -> list[dict]: uid = self.uid or get_uid(self.api) @@ -1705,10 +1802,8 @@ class QtMusicApp(QMainWindow): **{"page-size": 25}, nocorrect="true", ) - return ( - ((payload.get("result") or {}).get("tracks") or {}).get("results") - or [] - ) + return (((payload.get("result") or {}).get( + "tracks") or {}).get("results") or []) self.run_task(search, self.apply_search, self.search_failed) diff --git a/settings.py b/settings.py index 2ac4d35..acb0256 100644 --- a/settings.py +++ b/settings.py @@ -42,7 +42,8 @@ def _coerce(key: str, value: Any) -> Any: if key in ("autoplay", "wave_from_history"): if isinstance(value, bool): return value - return str(value).strip().lower() in ("1", "true", "yes", "on", "да", "вкл") + return str(value).strip().lower() in ( + "1", "true", "yes", "on", "да", "вкл") if key == "start_queue": text = str(value) return text if text in START_QUEUES else DEFAULTS[key] diff --git a/tui.py b/tui.py index 59090f1..162a168 100644 --- a/tui.py +++ b/tui.py @@ -149,9 +149,9 @@ class ListScreen(Screen): def row_selected(self, event: DataTable.RowSelected) -> None: track_id = str(event.row_key.value) index = next( - (i for i, item in enumerate(self.tracks) if str(item.get("id")) == track_id), - -1, - ) + (i for i, item in enumerate( + self.tracks) if str( + item.get("id")) == track_id), -1, ) if index >= 0: self.app.play_from_list(self.tracks, index, self.list_title) self.app.pop_screen() @@ -209,9 +209,8 @@ class FavoritesScreen(Screen): track_id = str(event.row_key.value) tracks = self.app.favorites index = next( - (i for i, item in enumerate(tracks) if str(item.get("id")) == track_id), - -1, - ) + (i for i, item in enumerate(tracks) if str( + item.get("id")) == track_id), -1, ) if index >= 0: self.app.play_from_list(tracks, index, "Избранное") self.app.pop_screen() @@ -275,7 +274,8 @@ class SearchScreen(Screen): **{"page-size": 25}, nocorrect="true", ) - tracks = (((payload.get("result") or {}).get("tracks") or {}).get("results")) or [] + tracks = (((payload.get("result") or {}).get( + "tracks") or {}).get("results")) or [] self.app.call_from_thread(self.fill, tracks) except SystemExit as exc: self.app.call_from_thread(self.failed, str(exc)) @@ -297,9 +297,9 @@ class SearchScreen(Screen): def row_selected(self, event: DataTable.RowSelected) -> None: track_id = str(event.row_key.value) index = next( - (i for i, item in enumerate(self.results) if str(item.get("id")) == track_id), - -1, - ) + (i for i, item in enumerate( + self.results) if str( + item.get("id")) == track_id), -1, ) if index >= 0: self.app.play_from_list(self.results, index, "Поиск") self.app.pop_screen() @@ -391,14 +391,21 @@ class SettingsScreen(Screen): self.refresh_values() from settings import settings_path - self.query_one("#settings-path", Static).update(f"Файл: {settings_path()}") + self.query_one( + "#settings-path", + Static).update( + f"Файл: { + settings_path()}") def refresh_values(self) -> None: s = self.app.settings self.query_one("#vol-value", Static).update(f"{s['volume']} %") self.query_one("#seek-value", Static).update(f"{s['seek_step']} с") - self.query_one("#autoplay-toggle", Button).label = "вкл" if s["autoplay"] else "выкл" - self.query_one("#start-toggle", Button).label = START_QUEUE_LABELS[s["start_queue"]] + self.query_one( + "#autoplay-toggle", + Button).label = "вкл" if s["autoplay"] else "выкл" + self.query_one("#start-toggle", + Button).label = START_QUEUE_LABELS[s["start_queue"]] self.query_one("#wave-toggle", Button).label = ( "вкл" if s["wave_from_history"] else "выкл" ) @@ -426,9 +433,12 @@ class SettingsScreen(Screen): elif bid == "start-toggle": order = list(START_QUEUES) index = order.index(s["start_queue"]) - self.app.update_setting("start_queue", order[(index + 1) % len(order)]) + self.app.update_setting( + "start_queue", order[(index + 1) % len(order)]) elif bid == "wave-toggle": - self.app.update_setting("wave_from_history", not s["wave_from_history"]) + self.app.update_setting( + "wave_from_history", + not s["wave_from_history"]) self.refresh_values() @@ -532,7 +542,10 @@ class MusicTui(App): } #menu { width: 5; min-width: 5; margin: 0 1 0 0; } #brand { text-style: bold; padding: 0 1; color: #f2f2f2; } - #topstatus { width: 1fr; text-align: right; content-align: right middle; color: #8a8a8a; } + #topstatus { + width: 1fr; text-align: right; content-align: right middle; + color: #8a8a8a; + } #phone { height: 1fr; @@ -548,7 +561,10 @@ class MusicTui(App): color: #8a8a8a; margin-bottom: 1; } - #title { width: 100%; text-align: center; text-style: bold; color: #ffffff; } + #title { + width: 100%; text-align: center; text-style: bold; + color: #ffffff; + } #artist { width: 100%; text-align: center; color: #b0b0b0; } #album { width: 100%; text-align: center; color: #6a6a6a; } #time { width: 100%; text-align: center; color: #8a8a8a; margin-top: 1; } @@ -566,7 +582,10 @@ class MusicTui(App): background: #0a0a0a; } #back { width: 5; min-width: 5; margin-right: 1; } - #screen-title { text-style: bold; width: 1fr; content-align: left middle; color: #f2f2f2; } + #screen-title { + text-style: bold; width: 1fr; content-align: left middle; + color: #f2f2f2; + } #screen-status { height: 1; padding: 0 1; color: #8a8a8a; } #list-table { height: 1fr; background: #000000; } #search-row { height: 3; padding: 0 1; } @@ -618,7 +637,10 @@ class MusicTui(App): self.settings = load_settings() self.api: Api | None = None self.uid = "" - self.player = MusicPlayer(player, volume=int(self.settings.get("volume", 80))) + self.player = MusicPlayer( + player, volume=int( + self.settings.get( + "volume", 80))) self.wave: list[dict] = [] self.favorites: list[dict] = [] self.liked_ids: set[str] = set() @@ -648,7 +670,12 @@ class MusicTui(App): yield Static("", id="artist") yield Static("", id="album") yield Static("0:00 / 0:00", id="time") - yield ProgressBar(total=100, show_eta=False, show_percentage=False, id="progress") + yield ProgressBar( + total=100, + show_eta=False, + show_percentage=False, + id="progress", + ) with Horizontal(id="controls"): yield Button("⏮", id="prev") yield Button("▶", id="playpause") @@ -671,7 +698,10 @@ class MusicTui(App): self.exit() return if not self.api.token: - self.notify("Сначала войди: python mtsmusic.py login", severity="error", timeout=8) + self.notify( + "Сначала войди: python mtsmusic.py login", + severity="error", + timeout=8) self.set_timer(3, self.exit) return self.load_initial() @@ -690,7 +720,8 @@ class MusicTui(App): return value - (value % 2) def apply_cover_width(self) -> None: - self.cover_width = self.cover_width_for(self.size.width, self.size.height) + self.cover_width = self.cover_width_for( + self.size.width, self.size.height) self.apply_column_width(self.cover_width) def apply_column_width(self, width: int) -> None: @@ -751,15 +782,22 @@ class MusicTui(App): liked = {str(track.get("id")) for track in start} else: payload = api.get(f"/users/{self.uid}/likes/tracks") - refs = ((payload.get("result") or {}).get("library") or {}).get("tracks") or [] + refs = ((payload.get("result") or {}).get( + "library") or {}).get("tracks") or [] liked = {str(ref.get("id")) for ref in refs} if want == "history": start, name = list(self.history), "История" self.call_from_thread(self.apply_initial, wave, liked, start, name) except SystemExit as exc: - self.call_from_thread(self.notify, f"Ошибка загрузки: {exc}", severity="error") + self.call_from_thread( + self.notify, + f"Ошибка загрузки: {exc}", + severity="error") except requests.RequestException as exc: - self.call_from_thread(self.notify, f"Сеть недоступна: {exc}", severity="error") + self.call_from_thread( + self.notify, + f"Сеть недоступна: {exc}", + severity="error") def apply_initial( self, @@ -795,7 +833,10 @@ class MusicTui(App): track_id = str(track.get("id")) liked = track_id in self.liked_ids - self.query_one("#title", Static).update(Text(track.get("title") or "?")) + self.query_one( + "#title", Static).update( + Text( + track.get("title") or "?")) self.query_one("#artist", Static).update(Text(track_artist(track))) self.query_one("#album", Static).update(Text(track_album(track))) self.update_like_button(liked) @@ -806,19 +847,26 @@ class MusicTui(App): def update_like_button(self, liked: bool | None = None) -> None: if liked is None: - liked = bool(self.current and str(self.current.get("id")) in self.liked_ids) + liked = bool( + self.current and str( + self.current.get("id")) in self.liked_ids) button = self.query_one("#like", Button) heart = "♥" if liked else "♡" suffix = " Избранное" if self.cover_width >= 34 else "" button.label = heart + suffix def update_play_button(self) -> None: - self.query_one("#playpause", Button).label = "⏸" if self.player.paused else "▶" + self.query_one( + "#playpause", + Button).label = "⏸" if self.player.paused else "▶" self._update_mpris_status() def update_time(self) -> None: duration = (self.current or {}).get("durationMs") or 0 - elapsed = min(self.player.elapsed(), duration / 1000 if duration else 0) + elapsed = min( + self.player.elapsed(), + duration / + 1000 if duration else 0) bar = self.query_one("#progress", ProgressBar) bar.update(total=max(duration / 1000, 1), progress=elapsed) self.query_one("#time", Static).update( @@ -867,7 +915,9 @@ class MusicTui(App): def fetch_cover(self, token: int, url: str) -> None: try: force_ipv4() - response = requests.get(url, headers={"User-Agent": USER_AGENT}, timeout=20) + response = requests.get( + url, headers={ + "User-Agent": USER_AGENT}, timeout=20) response.raise_for_status() data = response.content except (requests.RequestException, OSError): @@ -884,7 +934,11 @@ class MusicTui(App): # -- воспроизведение -------------------------------------------------- - def play_from_list(self, tracks: list[dict], index: int, name: str) -> None: + def play_from_list( + self, + tracks: list[dict], + index: int, + name: str) -> None: if not (0 <= index < len(tracks)): return self.queue_name = name @@ -927,14 +981,20 @@ class MusicTui(App): url, item = resolve_stream(api, str(track.get("id"))) except SystemExit as exc: self.call_from_thread( - self.notify, f"Не удалось открыть трек: {exc}", severity="error" - ) + self.notify, + f"Не удалось открыть трек: {exc}", + severity="error") return self.call_from_thread( self.start_playback, token, track, url, bool(item.get("preview")) ) - def start_playback(self, token: int, track: dict, url: str, preview: bool) -> None: + def start_playback( + self, + token: int, + track: dict, + url: str, + preview: bool) -> None: if token != self.play_token: return self.run_worker(self._play_url(track, url, preview), group="playback") @@ -962,7 +1022,10 @@ class MusicTui(App): def action_play_current(self) -> None: if self.current: - self.play_track(self.current, self.queue or self.wave, self.current_index) + self.play_track( + self.current, + self.queue or self.wave, + self.current_index) def try_next(self, direction: int) -> None: queue = self.queue or self.wave @@ -995,7 +1058,12 @@ class MusicTui(App): if not self.current or not self.player.is_playing(): return duration = (self.current.get("durationMs") or 0) / 1000 - self.run_worker(self._seek(delta, duration), group="seek", exclusive=True) + self.run_worker( + self._seek( + delta, + duration), + group="seek", + exclusive=True) async def _seek(self, delta: float, duration: float) -> None: await self.player.seek(delta, duration) @@ -1035,19 +1103,29 @@ class MusicTui(App): ] self.call_from_thread(self.apply_more_wave, tracks, then_play) except SystemExit as exc: - self.call_from_thread(self.notify, f"Ошибка волны: {exc}", severity="error") + self.call_from_thread( + self.notify, + f"Ошибка волны: {exc}", + severity="error") - def apply_more_wave(self, tracks: list[dict], then_play: int | None) -> None: + def apply_more_wave( + self, + tracks: list[dict], + then_play: int | None) -> None: known = {str(track.get("id")) for track in self.wave} - added = [track for track in tracks if str(track.get("id")) not in known] + added = [ + track for track in tracks if str( + track.get("id")) not in known] self.wave.extend(added) - self.set_topstatus(f"Волна · {len(self.wave)} · ♥ {len(self.liked_ids)}") + self.set_topstatus( + f"Волна · {len(self.wave)} · ♥ {len(self.liked_ids)}") if then_play is not None and then_play < len(self.queue): self.play_track(self.queue[then_play], self.queue, then_play) elif then_play is not None and added: self.queue = self.wave self.current_index = len(self.wave) - len(added) - self.play_track(self.wave[self.current_index], self.wave, self.current_index) + self.play_track(self.wave[self.current_index], + self.wave, self.current_index) # -- избранное -------------------------------------------------------- @@ -1064,7 +1142,10 @@ class MusicTui(App): albums = track.get("albums") or [] album = str(albums[0].get("id")) if albums else "" if not album: - self.call_from_thread(self.notify, "У трека нет альбома", severity="error") + self.call_from_thread( + self.notify, + "У трека нет альбома", + severity="error") return liked = track_id in self.liked_ids try: @@ -1079,7 +1160,10 @@ class MusicTui(App): **{"track-id": f"{track_id}:{album}"}, ) except SystemExit as exc: - self.call_from_thread(self.notify, f"Ошибка избранного: {exc}", severity="error") + self.call_from_thread( + self.notify, + f"Ошибка избранного: {exc}", + severity="error") return self.call_from_thread(self.apply_like, track, not liked) @@ -1124,7 +1208,10 @@ class MusicTui(App): def apply_settings(self) -> None: volume = int(self.settings.get("volume", 80)) if self.player.volume != volume: - self.run_worker(self.player.set_volume(volume), group="volume", exclusive=True) + self.run_worker( + self.player.set_volume(volume), + group="volume", + exclusive=True) if self.mpris: self.mpris.update_volume(volume / 100) @@ -1194,7 +1281,8 @@ class MusicTui(App): self.seek(target - self.player.elapsed()) def mpris_set_volume(self, value: float) -> None: - self.update_setting("volume", int(round(max(0.0, min(1.0, value)) * 100))) + self.update_setting("volume", int( + round(max(0.0, min(1.0, value)) * 100))) def action_show_favorites(self) -> None: self.push_screen(FavoritesScreen())