Compare commits

..

10 Commits

Author SHA1 Message Date
jads
5b202984fa docs: add MIT license and project description 2026-10-23 23:09:42 +03:00
jads
637ac23b73 feat: add settings (volume, seek step, start queue) with TUI/Qt pages
- settings.py persists to .mts-music/settings.json; CLI 'settings' command
- player gains volume support (ffplay -volume / mpv --volume)
- TUI settings screen and Qt settings page (volume, output device, seek step,
  autoplay, start queue, wave-from-history)
- start queue + autoplay applied at startup; arrows use configured seek step
- shared fetch_liked_tracks helper; cache TUI cover renders and Qt scaled cover
2026-10-23 22:44:55 +03:00
jads
b64c2dcd12 feat: polish Qt player and keep TUI cover in color 2026-10-23 22:21:47 +03:00
jads
0eb6909724 style: center TUI cover and switch player to monochrome 2026-10-23 22:04:51 +03:00
jads
26a58068cb feat: add phone-style PySide6 desktop player 2026-10-23 21:54:21 +03:00
jads
f77f81dead feat: redesign TUI as a compact phone-style player 2026-10-23 21:44:03 +03:00
jads
ff7edd9da7 fix: harden TUI startup and add diagnostics 2026-10-23 21:26:58 +03:00
jads
773ba69f0b feat: add Textual TUI player 2026-10-23 21:26:58 +03:00
jads
ce2e324dbe feat: add My Wave, favorites and lyrics support 2026-10-23 21:26:58 +03:00
jads
a41ec450fe feat: add search, playback and MP3 downloads 2026-10-23 21:26:58 +03:00
10 changed files with 3953 additions and 3 deletions

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2026 jads
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

174
README.md
View File

@@ -1,7 +1,11 @@
# MTS Music CLI # MTS Music — another client
Минимальный клиент для **своего** аккаунта MTS Music: вход, статус аккаунта Неофициальный клиент для **своего** аккаунта MTS Music: «Моя волна», поиск,
и проверка подписки. избранное, тексты песен, воспроизведение и скачивание в MP3. Есть компактный
TUI-плеер и настольное Qt-приложение в стиле телефона.
> Только для личного использования с собственной платной подпиской MTS.
> Проект не связан с МТС и Яндексом; контент не перераспространяется.
Внутри используется тот же путь, что и в веб-клиенте: MTS SSO выдаёт Внутри используется тот же путь, что и в веб-клиенте: MTS SSO выдаёт
OAuth-токен Яндекс.Музыки, дальше запросы идут на `api.music.yandex.net`. OAuth-токен Яндекс.Музыки, дальше запросы идут на `api.music.yandex.net`.
@@ -14,6 +18,12 @@ python3 -m venv .venv
PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromium PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromium
``` ```
Для Qt-приложения дополнительно:
```bash
.venv/bin/pip install -r requirements-qt.txt
```
Всё локальное лежит рядом с проектом: Всё локальное лежит рядом с проектом:
- `.venv/` — зависимости - `.venv/` — зависимости
- `.playwright/` — браузер Chromium - `.playwright/` — браузер Chromium
@@ -29,9 +39,167 @@ PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromiu
сохранятся в `.mts-music/credentials.json` с правами `600`. сохранятся в `.mts-music/credentials.json` с правами `600`.
Повторный вход нужен только когда токен перестанет работать. Повторный вход нужен только когда токен перестанет работать.
## TUI-плеер
```bash
.venv/bin/python mtsmusic.py tui
```
Интерфейс в терминале сделан как экран телефона: чёрно-белые элементы и
цветная обложка. Обложка по центру, строго под ней по той же ширине —
большая кнопка play/pause и по бокам предыдущий/следующий треки, ниже
«Текст» и «Избранное». Кнопка «☰» в левом верхнем углу открывает боковое
меню: «Моя волна», избранное, история прослушивания, поиск, текст песни,
«Настройки» и выход.
Мышь работает: кнопки, меню, строки списков и полоса прогресса кликабельны
(клик по прогрессу перематывает трек).
| Клавиша | Действие |
|---------|----------|
| `Space` | пауза / продолжить |
| `←` / `→` | перемотка на −5 / +5 секунд |
| `n` / `p` | следующий / предыдущий трек |
| `t` | текст песни |
| `l` | добавить / убрать из избранного |
| `m` | открыть меню |
| `w` | список «Моей волны» |
| `f` | избранное |
| `h` | история |
| `s` | поиск |
| `q` | выход |
История хранится в `.mts-music/history.json` (последние 100 треков).
Плеер — `ffplay` (или `mpv`, если установлен). Пауза реализована через
`SIGSTOP`/`SIGCONT`, а перемотка перезапускает поток с нужной секунды,
поэтому отдельный демон не нужен.
## Qt-приложение
```bash
.venv/bin/python mtsmusic.py qt
```
Отдельное окно в стиле телефона: крупная цветная обложка, кнопка play/pause,
переключение треков, перемотка, текст песни и избранное. Кнопка «☰» открывает
боковую панель с «Моей волной», избранным, историей, поиском и настройками.
Воспроизведение — через `QMediaPlayer`/FFmpeg из PySide6, поэтому работает
перемотка и системный звук. По полосе прогресса можно кликнуть в любое место —
каретка сразу перейдёт туда (и потянется, если вести мышью). Окно растягивается
вместе с обложкой, списки прокручиваются, двойной клик по строке запускает трек.
`Tab` переключает фокус по элементам: меню, полоса, предыдущий, play, следующий,
«Текст», «Избранное». Горячие клавиши: `Space`, `←`/`→`, `N`/`P`, `L`, `T`, `M`,
`F`, `H`, `S`, `W`; `Esc` закрывает меню/экран, `Ctrl+Q` — выход.
«Моя волна» продолжается с последних прослушанных треков (берётся история), а не
начинается каждый раз с одного и того же места.
## Настройки
Настройки открываются из меню (пункт «Настройки» в TUI и Qt) или из командной
строки и хранятся в `.mts-music/settings.json`:
```bash
# показать текущие настройки
.venv/bin/python mtsmusic.py settings
# изменить: громкость 60%, шаг перемотки 10 секунд
.venv/bin/python mtsmusic.py settings -s volume=60 seek_step=10
```
| Настройка | По умолчанию | Описание |
|-----------|--------------|----------|
| `volume` | `80` | громкость плеера, 0–100 |
| `seek_step` | `5` | шаг перемотки стрелками, секунды |
| `autoplay` | `false` | начинать воспроизведение сразу при запуске |
| `start_queue` | `wave` | что открыть первым: `wave`, `favorites` или `history` |
| `wave_from_history` | `true` | продолжать «Мою волну» с последних треков |
В Qt-приложении на странице настроек дополнительно выбирается устройство вывода
звука. Громкость меняется на лету, остальные настройки применяются к новым
трекам и при следующем запуске.
## Команды ## Команды
```bash ```bash
# кто я и есть ли подписка # кто я и есть ли подписка
.venv/bin/python mtsmusic.py status .venv/bin/python mtsmusic.py status
# поиск
.venv/bin/python mtsmusic.py search "miyagi" -n 10
# «Моя волна»: 10 треков, ♥ — уже в избранном
.venv/bin/python mtsmusic.py mix -n 10
# сразу слушать подборку (Ctrl-C — остановить)
.venv/bin/python mtsmusic.py mix -n 5 --play
# избранное: добавить / убрать / показать
.venv/bin/python mtsmusic.py like 12345678
.venv/bin/python mtsmusic.py unlike 12345678
.venv/bin/python mtsmusic.py likes -n 20
# текст песни: показать, сохранить, или с таймкодами
.venv/bin/python mtsmusic.py lyrics 12345678
.venv/bin/python mtsmusic.py lyrics 12345678 -o track.txt
.venv/bin/python mtsmusic.py lyrics 12345678 --lrc -o track.lrc
# только права MTS, без внешнего источника
.venv/bin/python mtsmusic.py lyrics 12345678 --no-fallback
# прямая ссылка на MP3
.venv/bin/python mtsmusic.py url 12345678
# воспроизведение (нужен mpv или ffplay)
.venv/bin/python mtsmusic.py play 12345678
# скачивание
.venv/bin/python mtsmusic.py download 12345678 -o track.mp3
# настройки
.venv/bin/python mtsmusic.py settings
.venv/bin/python mtsmusic.py settings -s volume=60 seek_step=10
``` ```
Токен можно передать и через переменную окружения:
```bash
export MTS_YA_TOKEN='...'
```
## Если TUI не отображается
```bash
# диагностика окружения
.venv/bin/python mtsmusic.py tui --check
# без alternate screen (если экран пустой)
.venv/bin/python mtsmusic.py tui --inline
# отключить мышь, если терминал плохо её поддерживает
.venv/bin/python mtsmusic.py tui --inline --no-mouse
```
TUI нужно запускать в настоящем терминале (stdin/stdout — TTY), а не в панели
вывода IDE. При падении traceback пишется в `.mts-music/tui-error.log`.
## Ограничения
- Логин автоматизирован не полностью: QRATOR-капчу проходит человек.
- `reference_uri` у MTS должен быть на домене `music.mts.ru`, поэтому
локальный callback не используется — токен забирается из браузера.
- Без активной подписки `download-info` отдаёт `preview` (короткий фрагмент).
- Тексты песен — отдельная подписка Яндекс Плюс, которой у MTS-аккаунта нет
(`hasPlus: false`), поэтому MTS возвращает `404` даже для треков с текстом.
`lyrics` автоматически берёт текст из бесплатного [LRCLIB](https://lrclib.net)
и помечает источник в выводе. `--no-fallback` отключает это поведение.
- Использовать только для личного прослушивания, не перераздавать контент.
- Если `api.music.yandex.net` недоступен, проверь IPv6: клиент принудительно
ходит по IPv4.
## Лицензия
[MIT](LICENSE) © 2026 jads. Проект предоставляется «как есть», без гарантий;
используйте на свой страх и риск и только со своим аккаунтом.

72
media.py Normal file
View File

@@ -0,0 +1,72 @@
"""Shared track helpers and local history storage."""
from __future__ import annotations
import json
from pathlib import Path
from mtsmusic import CONFIG_DIR
HISTORY_PATH = CONFIG_DIR / "history.json"
HISTORY_LIMIT = 100
def track_artist(track: dict) -> str:
return ", ".join(a.get("name", "?") for a in track.get("artists") or []) or "?"
def track_album(track: dict) -> str:
albums = track.get("albums") or []
return (albums[0].get("title") or "—") if albums else "—"
def track_cover_url(track: dict, size: int = 360) -> str | None:
uri = track.get("coverUri")
if not uri:
albums = track.get("albums") or []
uri = albums[0].get("coverUri") if albums else None
if not uri:
return None
uri = uri.replace("%%", f"{size}x{size}")
if not uri.startswith("http"):
uri = "https://" + uri
return uri
def load_history() -> list[dict]:
try:
data = json.loads(HISTORY_PATH.read_text(encoding="utf-8"))
except (OSError, ValueError):
return []
if not isinstance(data, list):
return []
seen: set[str] = set()
history: list[dict] = []
for track in data:
if not isinstance(track, dict):
continue
track_id = str(track.get("id") or "")
if not track_id or track_id in seen:
continue
seen.add(track_id)
history.append(track)
return history[:HISTORY_LIMIT]
def record_history(track: dict) -> list[dict]:
"""Добавляет трек в начало истории и сохраняет последние 100."""
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.insert(0, track)
history = history[:HISTORY_LIMIT]
try:
HISTORY_PATH.parent.mkdir(parents=True, exist_ok=True)
HISTORY_PATH.write_text(
json.dumps(history, ensure_ascii=False, separators=(",", ":")),
encoding="utf-8",
)
except OSError:
pass
return history

View File

@@ -310,6 +310,29 @@ def fmt_track(track: dict) -> str:
) )
def cmd_search(args: argparse.Namespace) -> None:
api = Api()
payload = api.get(
"/search",
text=args.query,
type="all",
page=0,
**{"page-size": args.limit},
nocorrect="true",
)
result = payload.get("result") or {}
tracks = ((result.get("tracks") or {}).get("results")) or []
if not tracks:
best = (result.get("best") or {}).get("result")
if isinstance(best, dict) and best.get("id"):
print(fmt_track(best))
return
print("Ничего не найдено.")
return
for track in tracks[: args.limit]:
print(fmt_track(track))
def cmd_status(args: argparse.Namespace) -> None: def cmd_status(args: argparse.Namespace) -> None:
api = Api() api = Api()
require_token(api, "статус") require_token(api, "статус")
@@ -345,8 +368,551 @@ def cmd_status(args: argparse.Namespace) -> None:
save_credentials(creds) save_credentials(creds)
# ---------------------------------------------------------------------------
# моя волна, избранное, тексты
# ---------------------------------------------------------------------------
def get_uid(api: Api) -> str:
uid = (load_credentials() or {}).get("ya_uid")
if uid:
return str(uid)
# Резервный путь: uid есть в ответе MTS-статуса.
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")
creds = load_credentials() or {}
creds["ya_uid"] = str(uid)
save_credentials(creds)
return str(uid)
def get_track(api: Api, track_id: str) -> dict:
result = api.get(f"/tracks/{track_id}").get("result") or []
if not result:
die(f"трек {track_id} не найден")
return result[0]
def album_id(track: dict) -> str | None:
albums = track.get("albums") or []
if albums:
return str(albums[0].get("id"))
return None
def track_album_ref(track_id: str, album: str | None = None) -> str:
return f"{track_id}:{album}" if album else str(track_id)
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 {}
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 []
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]
if chunk:
tracks.extend(
api.get("/tracks", trackIds=",".join(chunk)).get("result") or []
)
return tracks
def print_mix_item(index: int, item: dict) -> None:
track = item.get("track") or {}
heart = "♥" if item.get("liked") else " "
print(f"{heart} {index:>3}. {fmt_track(track)}", flush=True)
def cmd_mix(args: argparse.Namespace) -> None:
api = Api()
require_token(api, "моя волна")
limit = max(1, args.number)
collected: list[dict] = []
seen: set[str] = set()
queue: list[str] | None = None
attempts = 0
max_attempts = max(10, limit)
print(f"Моя волна — {limit} трек(ов):", file=sys.stderr)
while len(collected) < limit and attempts < max_attempts:
attempts += 1
result = fetch_mix_batch(api, queue)
sequence = result.get("sequence") or []
if not sequence:
break
new_items = []
for item in sequence:
if item.get("type") != "track":
continue
track_id = str((item.get("track") or {}).get("id") or "")
if not track_id or track_id in seen:
continue
seen.add(track_id)
new_items.append(item)
if not new_items:
break
for item in new_items:
if len(collected) >= limit:
break
collected.append(item)
ids = [str((item.get("track") or {}).get("id")) for item in sequence]
queue = ids[-5:]
if not collected:
die("волна вернула пустой список")
for index, item in enumerate(collected, 1):
print_mix_item(index, item)
if args.play:
play_items(api, [item.get("track") or {} for item in collected], args.player)
def cmd_like(args: argparse.Namespace) -> None:
api = Api()
require_token(api, "избранное")
uid = get_uid(api)
track = get_track(api, args.track_id)
album = args.album or album_id(track)
ref = track_album_ref(args.track_id, album)
api.post(f"/users/{uid}/likes/tracks/add", **{"track-id": ref})
print(f"♥ добавлено в избранное: {fmt_track(track)}")
def cmd_unlike(args: argparse.Namespace) -> None:
api = Api()
require_token(api, "избранное")
uid = get_uid(api)
track = get_track(api, args.track_id)
album = args.album or album_id(track)
ref = track_album_ref(args.track_id, album)
api.post(f"/users/{uid}/likes/tracks/remove", **{"track-ids": ref})
print(f"♡ убрано из избранного: {fmt_track(track)}")
def cmd_likes(args: argparse.Namespace) -> None:
api = Api()
require_token(api, "избранное")
uid = get_uid(api)
payload = api.get(f"/users/{uid}/likes/tracks")
library = (payload.get("result") or {}).get("library") or {}
liked = library.get("tracks") or []
if not liked:
print("В избранном пока пусто.")
return
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]
try:
tracks = api.get("/tracks", trackIds=",".join(chunk)).get("result") or []
except SystemExit:
tracks = []
for track in tracks:
details[str(track.get("id"))] = track
for index, item in enumerate(liked, 1):
track_id = str(item.get("id"))
track = details.get(track_id)
if track:
print(f"{index:>3}. {fmt_track(track)}")
else:
album = item.get("albumId")
print(f"{index:>3}. {track_id:>12} (альбом {album})")
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()
force_ipv4()
return api.session.get(
f"{YANDEX_BASE}/tracks/{track_id}/lyrics",
params={"timeStamp": timestamp, "sign": signature, "format": fmt},
headers=api._headers(),
timeout=30,
)
def lyrics_text(api: Api, response: requests.Response) -> str:
"""Достаёт сам текст: API отдаёт ссылку на файл, а не строки."""
payload = response.json()
result = payload.get("result") if isinstance(payload, dict) else None
if isinstance(result, str):
return result
if not isinstance(result, dict):
return ""
for key in ("lyrics", "full_lyrics", "fullLyrics", "text"):
value = result.get(key)
if isinstance(value, str) and value.strip():
return value
download_url = result.get("download_url") or result.get("downloadUrl")
if download_url:
force_ipv4()
file_response = requests.get(
download_url, headers={"User-Agent": USER_AGENT}, timeout=60
)
file_response.raise_for_status()
return file_response.text
return ""
LRCLIB_URL = "https://lrclib.net/api/search"
def lrclib_lyrics(track: dict, want_sync: bool) -> str | None:
"""Резервный источник текста: LRCLIB (бесплатный API, без ключа)."""
title = (track.get("title") or "").strip()
artists = track.get("artists") or []
artist = (artists[0].get("name") or "").strip() if artists else ""
if not title:
return None
force_ipv4()
headers = {"User-Agent": "MTSMusicCLI/0.1 (personal use)"}
queries = [
{"track_name": title, "artist_name": artist},
{"q": f"{artist} {title}".strip()},
]
candidates: list[dict] = []
for params in queries:
try:
response = requests.get(LRCLIB_URL, params=params, headers=headers, timeout=20)
except requests.RequestException:
continue
if response.status_code != 200:
continue
try:
data = response.json()
except ValueError:
continue
if isinstance(data, list) and data:
candidates = data
break
if not candidates:
return None
# Выбираем версию, ближайшую по длительности, где есть нужный формат.
field = "syncedLyrics" if want_sync else "plainLyrics"
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
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()]
field = "plainLyrics"
if not usable:
return None
best = min(usable, key=score)
return (best.get(field) or "").strip() or None
def cmd_lyrics(args: argparse.Namespace) -> None:
api = Api()
require_token(api, "текст песни")
track = get_track(api, args.track_id)
fmt = "LRC" if args.lrc else "TEXT"
text = ""
source = "MTS"
response = lyrics_response(api, args.track_id, fmt)
if response.status_code == 200:
try:
text = lyrics_text(api, response)
except (ValueError, requests.RequestException):
text = ""
if "Подключите подписку" in text:
text = ""
elif response.status_code not in (403, 404):
die(f"HTTP {response.status_code}: {response.text[:300]}")
if not text.strip() and not args.no_fallback:
text = lrclib_lyrics(track, want_sync=args.lrc) or ""
source = "LRCLIB"
if not text.strip():
print(f"Текст не найден: {fmt_track(track)}")
return
if args.output:
output = Path(args.output)
output.write_text(text, encoding="utf-8")
print(f"Сохранено: {output} ({len(text)} символов, источник: {source})")
else:
if source != "MTS":
print(f"# источник текста: {source}", file=sys.stderr)
print(f"# {fmt_track(track)}")
print()
print(text)
# ---------------------------------------------------------------------------
# streaming
# ---------------------------------------------------------------------------
def resolve_stream(api: Api, track_id: str) -> tuple[str, dict]:
"""Возвращает (url_mp3, download_info_item)."""
info = api.get(f"/tracks/{track_id}/download-info")
items = info.get("result") or []
if not items:
die(f"для трека {track_id} нет download-info (возможно, недоступен)")
# Лучший вариант: не preview и максимальный битрейт.
items.sort(
key=lambda item: (not item.get("preview", False), item.get("bitrateInKbps", 0)),
reverse=True,
)
item = items[0]
download_url = item["downloadInfoUrl"]
parsed = urllib.parse.urlparse(download_url)
query = parsed.query + ("&" if parsed.query else "") + "format=json"
meta = api.get_url(f"{YANDEX_BASE}{parsed.path}?{query}")
path = meta["path"]
signature = meta["s"]
host = meta["host"]
timestamp = meta["ts"]
digest = hashlib.md5((STREAM_SIGN_SECRET + path + signature).encode()).hexdigest()
stream_url = f"https://{host}/get-mp3/{digest}/{timestamp}{path}"
return stream_url, item
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(url)
def pick_player(explicit: str | None) -> str | None:
if explicit:
return explicit
for candidate in ("mpv", "ffplay"):
path = shutil.which(candidate)
if path:
return path
return None
def print_track_info(api: Api, track_id: str) -> None:
try:
track = api.get(f"/tracks/{track_id}")["result"][0]
except (SystemExit, KeyError, IndexError):
return
print(fmt_track(track), file=sys.stderr)
def play_url(player: str, url: str) -> None:
name = Path(player).name
if name == "ffplay":
command = [player, "-nodisp", "-autoexit", "-loglevel", "warning", url]
else:
command = [player, "--no-video", "--really-quiet", url]
try:
subprocess.run(command, check=False)
except KeyboardInterrupt:
raise
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 <id>")
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)
try:
url, item = resolve_stream(api, track_id)
except SystemExit:
print(" пропускаю: не удалось получить поток", file=sys.stderr)
continue
if item.get("preview"):
print(" внимание: preview-фрагмент", file=sys.stderr)
try:
play_url(player, url)
except KeyboardInterrupt:
print("\nОстановлено.", file=sys.stderr)
return
def cmd_play(args: argparse.Namespace) -> None:
api = Api()
print_track_info(api, args.track_id)
url, item = resolve_stream(api, args.track_id)
if item.get("preview"):
print("Внимание: это preview-фрагмент (нет активной подписки).", file=sys.stderr)
player = pick_player(args.player)
if not player:
die("не найден mpv/ffplay. Установи один из них или используй: mtsmusic.py url <id>")
try:
play_url(player, url)
except KeyboardInterrupt:
pass
def cmd_download(args: argparse.Namespace) -> None:
api = Api()
print_track_info(api, args.track_id)
url, _ = resolve_stream(api, args.track_id)
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:
response.raise_for_status()
total = int(response.headers.get("Content-Length") or 0)
done = 0
with output.open("wb") as handle:
for chunk in response.iter_content(chunk_size=256 * 1024):
if not chunk:
continue
handle.write(chunk)
done += len(chunk)
if total:
print(f"\r{done * 100 // total:3d}% {done // 1024} KiB", end="", file=sys.stderr)
print(file=sys.stderr)
print(f"Сохранено: {output} ({output.stat().st_size} байт)")
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------
# cli # cli
# ---------------------------------------------------------------------------
def cmd_tui(args: argparse.Namespace) -> None:
if args.check:
print_tui_check()
return
if not sys.stdin.isatty() or not sys.stdout.isatty():
die(
"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")
try:
MusicTui(player=args.player).run(inline=args.inline, mouse=not args.no_mouse)
except KeyboardInterrupt:
pass
except Exception as exc: # TUI не должен молча исчезать
log = CONFIG_DIR / "tui-error.log"
try:
import traceback
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
log.write_text(traceback.format_exc(), encoding="utf-8")
except OSError:
log = Path("?")
die(f"TUI упал: {exc}. Подробности: {log}")
def cmd_qt(args: argparse.Namespace) -> None:
try:
from qt_app import main as qt_main
except ImportError as exc:
die(
"для Qt-приложения нужен PySide6: "
f"{exc}. Установи: pip install -r requirements-qt.txt"
)
raise SystemExit(qt_main())
def cmd_settings(args: argparse.Namespace) -> None:
# Импорт внутри функции: settings.py сам зависит от mtsmusic.CONFIG_DIR.
from settings import DEFAULTS, load_settings, settings_path, save_settings
saved = False
if args.set:
updates: dict = {}
for pair in args.set:
if "=" not in pair:
die(f"неверный формат: {pair} (нужно ключ=значение)")
key, value = pair.split("=", 1)
key = key.strip()
if key not in DEFAULTS:
die(f"неизвестная настройка: {key}. Доступны: {', '.join(DEFAULTS)}")
updates[key] = value
current = save_settings(updates)
saved = True
else:
current = load_settings()
for key, default in DEFAULTS.items():
print(f"{key:<18} = {current.get(key, default)}")
print(f"\nФайл: {settings_path()}")
if saved:
print("Сохранено.")
def print_tui_check() -> None:
import platform
print(f"Python: {sys.executable}")
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()}")
print(f"stdin TTY: {sys.stdin.isatty()}")
size = shutil.get_terminal_size()
print(f"Терминал: {size.columns}x{size.lines}")
try:
import textual
print(f"textual: {textual.__version__}")
except ImportError:
print("textual: НЕ УСТАНОВЛЕН")
try:
import PIL
print(f"pillow: {PIL.__version__}")
except ImportError:
print("pillow: НЕ УСТАНОВЛЕН")
for name in ("ffplay", "mpv"):
print(f"{name}: {shutil.which(name) or 'нет'}")
print(f"Креды: {'есть' if CREDS_PATH.exists() else 'нет'} ({CREDS_PATH})")
if size.columns < 90 or size.lines < 30:
print("\nСовет: окно маленькое, TUI лучше смотреть при >= 90x30.")
def build_parser() -> argparse.ArgumentParser: def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser( parser = argparse.ArgumentParser(
prog="mtsmusic", prog="mtsmusic",
@@ -362,6 +928,80 @@ def build_parser() -> argparse.ArgumentParser:
status = sub.add_parser("status", help="показать аккаунт и подписку") status = sub.add_parser("status", help="показать аккаунт и подписку")
status.set_defaults(func=cmd_status) status.set_defaults(func=cmd_status)
search = sub.add_parser("search", help="поиск треков")
search.add_argument("query", help="поисковый запрос")
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("--player", help="путь к mpv/ffplay")
mix.set_defaults(func=cmd_mix)
like = sub.add_parser("like", help="добавить трек в избранное")
like.add_argument("track_id")
like.add_argument("--album", help="id альбома (если нужно переопределить)")
like.set_defaults(func=cmd_like)
unlike = sub.add_parser("unlike", help="убрать трек из избранного")
unlike.add_argument("track_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.set_defaults(func=cmd_likes)
lyrics = sub.add_parser("lyrics", help="показать или сохранить текст песни")
lyrics.add_argument("track_id")
lyrics.add_argument("--lrc", action="store_true", help="формат LRC с таймкодами")
lyrics.add_argument("-o", "--output", help="сохранить текст в файл")
lyrics.add_argument(
"--no-fallback",
action="store_true",
help="не использовать LRCLIB, если MTS не отдал текст",
)
lyrics.set_defaults(func=cmd_lyrics)
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(
"--inline",
action="store_true",
help="без alternate screen: рисовать прямо в терминале (если экран пустой)",
)
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.add_argument(
"-s",
"--set",
nargs="+",
metavar="KEY=VALUE",
help="изменить настройки, например: -s volume=60 seek_step=10",
)
settings.set_defaults(func=cmd_settings)
url = sub.add_parser("url", help="получить прямую ссылку на MP3")
url.add_argument("track_id")
url.set_defaults(func=cmd_url)
play = sub.add_parser("play", help="воспроизвести трек через mpv/ffplay")
play.add_argument("track_id")
play.add_argument("--player", help="путь к mpv/ffplay")
play.set_defaults(func=cmd_play)
download = sub.add_parser("download", help="скачать трек в MP3")
download.add_argument("track_id")
download.add_argument("-o", "--output", help="имя файла")
download.set_defaults(func=cmd_download)
return parser return parser

161
player.py Normal file
View File

@@ -0,0 +1,161 @@
"""Playback engine for the MTS Music clients.
ffplay supports pause via SIGSTOP/SIGCONT. Seeking restarts the stream with
`-ss` at the requested offset, which keeps the implementation tiny and avoids
requiring mpv/libmpv.
"""
from __future__ import annotations
import asyncio
import os
import shutil
import signal
import time
from pathlib import Path
class MusicPlayer:
def __init__(self, explicit: str | None = None, volume: int = 80) -> None:
self.path = self._pick(explicit)
self.proc: asyncio.subprocess.Process | None = None
self.url: str | None = None
self.offset = 0.0
self.paused = False
self.started_at = 0.0
self.paused_total = 0.0
self.paused_at = 0.0
self.switching = False
self.volume = max(0, min(100, int(volume)))
@staticmethod
def _pick(explicit: str | None) -> str | None:
if explicit:
return explicit
env = os.environ.get("MTSMUSIC_PLAYER")
if env:
return env
for candidate in ("mpv", "ffplay"):
found = shutil.which(candidate)
if found:
return found
return None
def _command(self, url: str, start: float = 0.0) -> list[str]:
name = Path(self.path or "").name
if name == "ffplay":
command = [
self.path or "ffplay",
"-nodisp",
"-autoexit",
"-volume",
str(self.volume),
"-loglevel",
"error",
]
if start > 0:
command += ["-ss", f"{start:.3f}"]
return command + [url]
command = [
self.path or "mpv",
"--no-video",
"--really-quiet",
f"--volume={self.volume}",
]
if start > 0:
command.append(f"--start={start:.3f}")
return command + [url]
async def set_volume(self, volume: int) -> None:
"""Меняет громкость; ffplay/mpv не умеют на лету, поэтому перезапускаем."""
self.volume = max(0, min(100, int(volume)))
if self.url and self.is_playing():
position = self.elapsed()
was_paused = self.paused
await self.play(self.url, position)
if was_paused:
self.pause()
async def play(self, url: str, start: float = 0.0) -> None:
await self.stop()
if not self.path:
raise RuntimeError("не найден mpv/ffplay")
self.switching = True
self.url = url
self.offset = max(0.0, start)
self.proc = await asyncio.create_subprocess_exec(
*self._command(url, self.offset),
stdin=asyncio.subprocess.DEVNULL,
stdout=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.DEVNULL,
)
self.paused = False
self.started_at = time.monotonic()
self.paused_total = 0.0
self.switching = False
async def stop(self) -> None:
self.switching = True
proc = self.proc
if proc and proc.returncode is None:
if self.paused:
try:
proc.send_signal(signal.SIGCONT)
except ProcessLookupError:
pass
proc.terminate()
try:
await asyncio.wait_for(proc.wait(), timeout=2)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
self.proc = None
self.paused = False
self.switching = False
def pause(self) -> None:
if self.proc and self.proc.returncode is None and not self.paused:
try:
self.proc.send_signal(signal.SIGSTOP)
except ProcessLookupError:
return
self.paused = True
self.paused_at = time.monotonic()
def resume(self) -> None:
if self.proc and self.proc.returncode is None and self.paused:
try:
self.proc.send_signal(signal.SIGCONT)
except ProcessLookupError:
return
self.paused = False
self.paused_total += time.monotonic() - self.paused_at
def toggle_pause(self) -> bool:
if self.paused:
self.resume()
else:
self.pause()
return self.paused
def is_playing(self) -> bool:
return bool(self.proc and self.proc.returncode is None)
def elapsed(self) -> float:
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)
async def seek(self, delta: float, duration: float) -> None:
"""Seek by delta seconds. Restarts ffplay at the new position."""
if not self.url or not self.is_playing():
return
limit = max(1.0, duration - 1.0) if duration else None
target = max(0.0, self.elapsed() + delta)
if limit:
target = min(target, limit)
was_paused = self.paused
await self.play(self.url, target)
if was_paused:
self.pause()

1659
qt_app.py Normal file

File diff suppressed because it is too large Load Diff

2
requirements-qt.txt Normal file
View File

@@ -0,0 +1,2 @@
-r requirements.txt
PySide6>=6.7

View File

@@ -1,2 +1,4 @@
playwright>=1.40 playwright>=1.40
requests>=2.31 requests>=2.31
textual>=0.80
pillow>=10

104
settings.py Normal file
View File

@@ -0,0 +1,104 @@
"""Настройки MTS Music: хранятся в .mts-music/settings.json."""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from mtsmusic import CONFIG_DIR
SETTINGS_PATH = CONFIG_DIR / "settings.json"
DEFAULTS: dict[str, Any] = {
"volume": 80, # громкость 0..100
"seek_step": 5, # шаг перемотки стрелками, секунды
"autoplay": False, # запускать воспроизведение сразу
"start_queue": "wave", # wave | favorites | history
"wave_from_history": True, # продолжать волну с последних треков
}
START_QUEUES = ("wave", "favorites", "history")
START_QUEUE_LABELS = {
"wave": "Моя волна",
"favorites": "Избранное",
"history": "История",
}
_cache: dict[str, Any] | None = None
def _coerce(key: str, value: Any) -> Any:
if key == "volume":
try:
return max(0, min(100, int(value)))
except (TypeError, ValueError):
return DEFAULTS[key]
if key == "seek_step":
try:
return max(1, min(60, int(value)))
except (TypeError, ValueError):
return DEFAULTS[key]
if key in ("autoplay", "wave_from_history"):
if isinstance(value, bool):
return value
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]
return value
def load_settings() -> dict[str, Any]:
"""Возвращает настройки с дефолтами; файл читается один раз."""
global _cache
if _cache is not None:
return dict(_cache)
raw: dict[str, Any] = {}
try:
data = json.loads(SETTINGS_PATH.read_text(encoding="utf-8"))
if isinstance(data, dict):
raw = data
except (OSError, ValueError):
raw = {}
merged = dict(DEFAULTS)
for key, value in raw.items():
if key in DEFAULTS:
merged[key] = _coerce(key, value)
_cache = merged
return dict(merged)
def save_settings(values: dict[str, Any]) -> dict[str, Any]:
"""Сохраняет изменённые настройки и возвращает актуальный набор."""
global _cache
merged = load_settings()
for key, value in values.items():
if key in DEFAULTS:
merged[key] = _coerce(key, value)
_cache = merged
try:
SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True)
SETTINGS_PATH.write_text(
json.dumps(merged, ensure_ascii=False, indent=2),
encoding="utf-8",
)
except OSError:
pass
return dict(merged)
def get_setting(key: str, default: Any = None) -> Any:
return load_settings().get(key, DEFAULTS.get(key, default))
def set_setting(key: str, value: Any) -> Any:
return save_settings({key: value}).get(key)
def update_settings(**values: Any) -> dict[str, Any]:
return save_settings(dict(values))
def settings_path() -> Path:
return SETTINGS_PATH

1121
tui.py Normal file

File diff suppressed because it is too large Load Diff