Compare commits

..

2 Commits

Author SHA1 Message Date
jads
c8043d8b80 Merge remote-tracking branch 'origin/main'
# Conflicts:
#	README.md
#	mtsmusic.py
#	qt_app.py
2026-10-25 23:45:33 +03:00
jads
de9bf80d11 feat: download tracks with full ID3 metadata
- Add downloads.py: "Artist - Title.mp3" naming, no-overwrite dedupe,
  configurable download_dir, progress and per-item callbacks.
- Save maximum metadata via tags.py (ID3v2.3, mutagen): artists, album
  and album artist, track/disc, year/date, raw genre, label, duration,
  explicit flag, cover 600x600, plain and synced lyrics (MTS then
  LRCLIB), track URLs and Yandex/MTS ids.
- TUI/Qt: download current track and queue from the menu, "d" key in
  lists, folder picker in settings; CLI download gains -d/--dir and
  --no-tags.
- Add mutagen to requirements and document download/metadata in README.
- Fix MPRIS update_metadata crashing without dbus-next.
2026-10-25 20:06:59 +03:00
8 changed files with 754 additions and 95 deletions

View File

@@ -1,8 +1,9 @@
# MTS Music — another client
Неофициальный клиент для **своего** аккаунта MTS Music: «Моя волна», поиск,
избранное, тексты песен, воспроизведение и скачивание в MP3. Есть компактный
TUI-плеер и настольное Qt-приложение в стиле телефона.
избранное, тексты песен, воспроизведение и скачивание в MP3 с полными
ID3-тегами. Есть компактный TUI-плеер и настольное Qt-приложение в стиле
телефона.
> Только для личного использования с собственной платной подпиской MTS.
> Проект не связан с МТС и Яндексом; контент не перераспространяется.
@@ -50,7 +51,7 @@ PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromiu
большая кнопка play/pause и по бокам предыдущий/следующий треки, ниже
«Текст» и «Избранное». Кнопка «☰» в левом верхнем углу открывает боковое
меню: «Моя волна», избранное, история прослушивания, поиск, текст песни,
«Настройки» и выход.
«Скачать трек», «Скачать список», «Настройки» и выход.
Мышь работает: кнопки, меню, строки списков и полоса прогресса кликабельны
(клик по прогрессу перематывает трек).
@@ -67,6 +68,7 @@ PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromiu
| `f` | избранное |
| `h` | история |
| `s` | поиск |
| `d` | скачать выделенный трек в списке |
| `q` | выход |
История хранится в `.mts-music/history.json` (последние 100 треков).
@@ -89,7 +91,8 @@ PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromiu
Отдельное окно в стиле телефона: крупная цветная обложка, кнопка play/pause,
переключение треков, перемотка, текст песни и избранное. Кнопка «☰» открывает
боковую панель с «Моей волной», избранным, историей, поиском и настройками.
боковую панель с «Моей волной», избранным, историей, поиском, скачиванием
и настройками.
Воспроизведение — через `QMediaPlayer`/FFmpeg из PySide6, поэтому работает
перемотка и системный звук. По полосе прогресса можно кликнуть в любое место —
@@ -103,6 +106,35 @@ PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromiu
«Моя волна» продолжается с последних прослушанных треков (берётся история), а не
начинается каждый раз с одного и того же места.
## Скачивание и метаданные
В меню обоих клиентов пункт «Скачать» открывает выбор: текущий трек, всё
избранное, 20 треков «Моей волны» или результаты поиска; папка выбирается
диалогом в Qt и вводится вручную в TUI, прогресс и число ошибок видны на
экране. Быстрые пункты «Скачать трек»/«Скачать список» (и клавиша `d` на
экране списка в TUI) сразу кладут в папку `download_dir` без диалога. В CLI
это команда `download` с теми же источниками (`--likes`, `--wave`, `--search`).
Файлы кладутся в папку `download_dir` (`~/Music/MTS Music` по умолчанию),
имя — `Исполнитель - Название.mp3`; существующие файлы не перезаписываются,
при совпадении добавляется ` (2)`, ` (3)` и т.д. (`--force` перезаписывает).
В скачанный файл пишется максимум метаданных ID3v2.3 (через `mutagen`, он в
`requirements.txt`):
- название, все исполнители (`TPE1`), альбом (`TALB`) и альбомный исполнитель
(`TPE2`);
- номер трека и диска (`TRCK`/`TPOS`), год и дата релиза (`TYER`/`TDAT`),
жанр как в API (`TCON`), лейбл (`TPUB`), длительность (`TLEN`);
- обложка 600×600 (`APIC`);
- текст песни (`USLT`) и синхронный текст (`SYLT`), если он есть: сначала
пробуем MTS, затем [LRCLIB](https://lrclib.net);
- ссылка на трек (`WOAR`/`COMM`), пометка `explicit` (`ITUNESADVISORY`) и
служебные поля с id трека, альбома и `realId` (`TXXX`).
Без `mutagen` или при `--no-tags` теги не пишутся, скачивание при этом
работает как обычно.
## Управление из системной панели (MPRIS)
При запуске TUI и Qt-приложение регистрируются в сессионном D-Bus как источник
@@ -121,13 +153,6 @@ PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromiu
Для этого нужен `dbus-next` (он уже в `requirements.txt`). Без сессионного
D-Bus или без `dbus-next` приложение просто работает как обычно, без интеграции.
## Скачивание из интерфейса
В меню обоих клиентов есть пункт «Скачать»: текущий трек, всё избранное,
20 треков «Моей волны» или последние результаты поиска. Папка выбирается
диалогом в Qt и вводится вручную в TUI, прогресс и число ошибок видны на
экране. Скачивание идёт тем же кодом, что и в CLI.
## Настройки
Настройки открываются из меню (пункт «Настройки» в TUI и Qt) или из командной
@@ -148,10 +173,12 @@ D-Bus или без `dbus-next` приложение просто работае
| `autoplay` | `false` | начинать воспроизведение сразу при запуске |
| `start_queue` | `wave` | что открыть первым: `wave`, `favorites` или `history` |
| `wave_from_history` | `true` | продолжать «Мою волну» с последних треков |
| `download_dir` | `~/Music/MTS Music` | куда сохранять скачанные MP3 |
В Qt-приложении на странице настроек дополнительно выбирается устройство вывода
звука. Громкость меняется на лету, остальные настройки применяются к новым
трекам и при следующем запуске.
звука и папка загрузок (кнопка «Выбрать…»). В TUI путь задаётся в настройках
пунктом «Папка загрузок». Громкость меняется на лету, остальные настройки
применяются к новым трекам и при следующем запуске.
## Команды
@@ -189,11 +216,13 @@ D-Bus или без `dbus-next` приложение просто работае
# воспроизведение (нужен mpv или ffplay)
.venv/bin/python mtsmusic.py play 12345678
# скачивание одного или нескольких треков в свою папку
.venv/bin/python mtsmusic.py download 12345678 -d ~/Music/MTS
# скачивание: в папку из download_dir, имя «Исполнитель - Название.mp3»
.venv/bin/python mtsmusic.py download 12345678
# несколько треков или своя папка
.venv/bin/python mtsmusic.py download 12345678 87654321 -d ~/Music/MTS
# как в мобильном приложении: скачать избранное, волну или результаты поиска
# как в мобильном приложении: избранное, волна или результаты поиска
.venv/bin/python mtsmusic.py download --likes -d ~/Music/MTS
.venv/bin/python mtsmusic.py download --wave -n 20 -d ~/Music/MTS
.venv/bin/python mtsmusic.py download --search "Кино" -n 10 -d ~/Music/MTS
@@ -204,6 +233,9 @@ D-Bus или без `dbus-next` приложение просто работае
# то же самое есть в меню TUI и Qt — пункт «Скачать»
# без ID3-тегов (быстрее, без обложки и текста)
.venv/bin/python mtsmusic.py download 12345678 --no-tags
# настройки
.venv/bin/python mtsmusic.py settings
.venv/bin/python mtsmusic.py settings -s volume=60 seek_step=10

View File

@@ -317,6 +317,8 @@ class MprisService:
duration_ms: int = 0,
art_url: str = "",
):
if not HAVE_DBUS:
return
metadata: dict[str, Any] = {
"mpris:trackid": Variant("o", track_object_path(track_id)),
"mpris:length": Variant("x", int(duration_ms) * 1000),

View File

@@ -31,9 +31,11 @@ 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"
@@ -55,10 +57,11 @@ _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
@@ -84,7 +87,8 @@ 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:
@@ -120,7 +124,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()
@@ -128,7 +134,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"]
@@ -156,16 +164,19 @@ 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')")
@@ -214,7 +225,12 @@ 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):
@@ -236,18 +252,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()
@@ -257,7 +279,7 @@ 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
@@ -345,7 +367,8 @@ 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 {}
@@ -353,9 +376,19 @@ def cmd_status(args: argparse.Namespace) -> None:
bff_subscriptions = bff_profile.get("subscriptions") or []
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 'нет'}")
display_name = (
profile.get("alias")
or account.get("fullName")
or account.get("displayName")
or account.get("login")
or "?"
)
print(f"Имя: {display_name}")
print(
f"Телефон: {profile.get('phone') or account.get('phone') or '?'}")
subscription_state = "активна" if (
subscriptions or bff_subscriptions) else "нет"
print(f"Подписка: {subscription_state}")
if bff_subscriptions:
for sub in bff_subscriptions:
name = sub.get("name") or sub.get("id") or "?"
@@ -363,13 +396,18 @@ 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)
@@ -387,7 +425,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)
@@ -414,20 +454,27 @@ 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
@@ -492,7 +539,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:
@@ -536,9 +584,10 @@ def cmd_likes(args: argparse.Namespace) -> None:
# /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:
@@ -562,12 +611,15 @@ 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()
hmac.new(LYRICS_SIGN_KEY.encode(),
message.encode(), hashlib.sha256).digest()
).decode()
force_ipv4()
@@ -624,7 +676,8 @@ 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:
@@ -644,13 +697,16 @@ 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
@@ -688,7 +744,8 @@ 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)
@@ -710,7 +767,8 @@ def _best_download_item(api: Api, track_id: str) -> dict:
# Лучший вариант: не preview и максимальный битрейт.
items.sort(
key=lambda item: (not item.get("preview", False), item.get("bitrateInKbps", 0)),
key=lambda item: (not item.get("preview", False),
item.get("bitrateInKbps", 0)),
reverse=True,
)
return items[0]
@@ -727,7 +785,8 @@ def _stream_url(api: Api, item: dict) -> str:
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()
return f"https://{host}/get-mp3/{digest}/{timestamp}{path}"
@@ -780,7 +839,7 @@ def fetch_track_file(
track_id: str,
progress: "Callable[[int, int], None] | None" = None,
) -> tuple[Path, dict]:
"""Скачивает трек в локальный кэш и возвращает (путь, download_info_item)."""
"""Скачивает трек в кэш и возвращает (путь, download_info_item)."""
item = _best_download_item(api, track_id)
preview = bool(item.get("preview"))
CACHE_DIR.mkdir(parents=True, exist_ok=True)
@@ -814,7 +873,9 @@ def fetch_track_file(
if progress is not None:
progress(received, expected)
if expected and received != expected:
die(f"трек {track_id} скачался не полностью ({received}/{expected} байт)")
die(
f"трек {track_id} скачался не полностью "
f"({received}/{expected} байт)")
os.replace(part, target)
except requests.RequestException as exc:
try:
@@ -837,7 +898,8 @@ 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)
@@ -871,14 +933,21 @@ 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 <id>")
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)
print(f"\n▶ {index}/{len(tracks)} {fmt_track(track)}",
file=sys.stderr)
try:
path, item = fetch_track_file(api, track_id)
except SystemExit:
@@ -898,11 +967,14 @@ def cmd_play(args: argparse.Namespace) -> None:
print_track_info(api, args.track_id)
path, item = fetch_track_file(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 <id>")
die(
"не найден mpv/ffplay. Установи один из них "
"или используй: mtsmusic.py url <id>")
try:
play_url(player, str(path))
@@ -977,7 +1049,8 @@ def collect_download_tracks(
if wave:
limit = number if number and number > 0 else 10
try:
tracks.extend(item.get("track") or {} for item in collect_wave(api, limit))
tracks.extend(item.get("track") or {}
for item in collect_wave(api, limit))
except SystemExit as exc:
report(f"не удалось получить волну: {exc}")
@@ -1008,6 +1081,7 @@ def download_tracks(
force: bool = False,
output: Path | None = None,
pretty_names: bool = True,
tags: bool = True,
on_event: "Callable[[dict], None] | None" = None,
) -> tuple[int, int, int]:
"""Скачивает треки в папку. Возвращает (сохранено, пропущено, ошибок).
@@ -1035,7 +1109,8 @@ def download_tracks(
target = destination / f"{track_id}.mp3"
if target.exists() and target.stat().st_size > 0 and not force:
emit("skipped", index=index, total=len(tracks), track=track, path=target)
emit("skipped", index=index, total=len(
tracks), track=track, path=target)
used.add(target)
skipped += 1
continue
@@ -1055,7 +1130,8 @@ def download_tracks(
failed += 1
continue
if item.get("preview"):
emit("preview", index=index, total=len(tracks), track=track, path=target)
emit("preview", index=index, total=len(
tracks), track=track, path=target)
target.parent.mkdir(parents=True, exist_ok=True)
part = target.with_name(f".{target.name}.{os.getpid()}.part")
@@ -1078,11 +1154,25 @@ def download_tracks(
failed += 1
continue
saved += 1
if tags:
_write_track_tags(api, track, target)
emit("saved", index=index, total=len(tracks), track=track, path=target)
return saved, skipped, failed
def _write_track_tags(api: Api, track: dict, path: Path) -> None:
"""Пишет максимум ID3-метаданных; сбой тегов не срывает скачивание."""
try:
from tags import tag_track
except ImportError:
return
try:
tag_track(api, track, path)
except Exception as exc: # noqa: BLE001 - теги не критичны для файла
print(f" теги не записаны: {exc}", file=sys.stderr)
def _collect_download_tracks(api: Api, args: argparse.Namespace) -> list[dict]:
return collect_download_tracks(
api,
@@ -1099,7 +1189,8 @@ def _print_download_event(event: dict) -> None:
status = event["status"]
path = event.get("path")
if status == "start":
artists = ", ".join(a.get("name", "?") for a in track.get("artists") or [])
artists = ", ".join(a.get("name", "?")
for a in track.get("artists") or [])
title = f"{artists} — {track.get('title') or '?'}"
print(f"[{event['index']}/{event['total']}] {title}", file=sys.stderr)
elif status == "skipped":
@@ -1113,6 +1204,8 @@ def _print_download_event(event: dict) -> None:
def cmd_download(args: argparse.Namespace) -> None:
from settings import get_setting
api = Api()
require_token(api, "скачивание")
@@ -1129,7 +1222,7 @@ def cmd_download(args: argparse.Namespace) -> None:
elif args.dir:
destination = Path(args.dir).expanduser()
else:
destination = Path.cwd()
destination = Path(get_setting("download_dir")).expanduser()
saved, skipped, failed = download_tracks(
api,
@@ -1137,7 +1230,8 @@ def cmd_download(args: argparse.Namespace) -> None:
destination,
force=args.force,
output=Path(args.output).expanduser() if args.output else None,
pretty_names=bool(args.dir),
pretty_names=True,
tags=not args.no_tags,
on_event=_print_download_event,
)
@@ -1158,15 +1252,19 @@ def cmd_tui(args: argparse.Namespace) -> None:
if not sys.stdin.isatty() or not sys.stdout.isatty():
die(
"TUI нужно запускать в обычном терминале (stdin/stdout не TTY).\n"
"Запусти прямо в окне терминала: .venv/bin/python mtsmusic.py tui\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 не должен молча исчезать
@@ -1205,7 +1303,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}. "
f"Доступны: {', '.join(DEFAULTS)}")
updates[key] = value
current = save_settings(updates)
saved = True
@@ -1223,7 +1323,9 @@ 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()} "
f"({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()}")
@@ -1244,7 +1346,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 'нет'} "
f"({CREDS_PATH})")
if size.columns < 90 or size.lines < 30:
print("\nСовет: окно маленькое, TUI лучше смотреть при >= 90x30.")
@@ -1256,9 +1360,12 @@ 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="показать аккаунт и подписку")
@@ -1266,12 +1373,15 @@ 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)
@@ -1282,7 +1392,8 @@ 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="показать избранные треки текстом")
@@ -1300,9 +1411,11 @@ def build_parser() -> argparse.ArgumentParser:
)
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",
@@ -1311,21 +1424,28 @@ 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",
@@ -1347,9 +1467,11 @@ def build_parser() -> argparse.ArgumentParser:
download = sub.add_parser(
"download", help="скачать трек или подборку в MP3"
)
download.add_argument("track_ids", nargs="*", help="ID треков (можно несколько)")
download.add_argument(
"-d", "--dir", help="папка для сохранения (по умолчанию текущая)"
"track_ids", nargs="*", help="ID треков (можно несколько)"
)
download.add_argument(
"-d", "--dir", help="папка для сохранения (по умолчанию download_dir)"
)
download.add_argument(
"-o", "--output", help="имя файла (только для одного трека)"
@@ -1373,6 +1495,11 @@ def build_parser() -> argparse.ArgumentParser:
download.add_argument(
"--force", action="store_true", help="перезаписать существующие файлы"
)
download.add_argument(
"--no-tags",
action="store_true",
help="не записывать ID3-теги (текст, обложка, метаданные)",
)
download.set_defaults(func=cmd_download)
return parser

View File

@@ -1055,6 +1055,22 @@ class QtMusicApp(QMainWindow):
self.wave_check.toggled.connect(self.on_wave_from_history_toggled)
layout.addWidget(self.wave_check)
dir_row = QHBoxLayout()
dir_row.setSpacing(10)
dir_label = QLabel("Папка загрузок")
dir_label.setObjectName("settingLabel")
self.download_dir_edit = QLineEdit(
str(self.settings.get("download_dir") or ""))
self.download_dir_edit.setObjectName("settingCombo")
self.download_dir_edit.setReadOnly(True)
dir_button = QPushButton("Выбрать…")
dir_button.setCursor(Qt.CursorShape.PointingHandCursor)
dir_button.clicked.connect(self.choose_download_dir)
dir_row.addWidget(dir_label)
dir_row.addWidget(self.download_dir_edit, 1)
dir_row.addWidget(dir_button)
layout.addLayout(dir_row)
hint = QLabel("Настройки сохраняются в .mts-music/settings.json")
hint.setObjectName("pageStatus")
hint.setWordWrap(True)
@@ -1075,7 +1091,10 @@ class QtMusicApp(QMainWindow):
dir_row = QHBoxLayout()
dir_row.setSpacing(8)
self.download_dir = QLineEdit(str(Path.home() / "Music" / "MTS"))
self.download_dir = QLineEdit(
str(self.settings.get("download_dir")
or Path.home() / "Music" / "MTS")
)
self.download_dir.setObjectName("downloadDir")
dir_row.addWidget(self.download_dir, 1)
browse = QPushButton("Обзор…")
@@ -1140,6 +1159,9 @@ class QtMusicApp(QMainWindow):
self.toast("Укажи папку для сохранения")
return
destination = Path(raw).expanduser()
self.update_setting("download_dir", str(destination))
if hasattr(self, "download_dir_edit"):
self.download_dir_edit.setText(str(destination))
kind = self.download_source.currentData()
if kind == "current" and not self.current:
self.toast("Сначала выбери трек")
@@ -1185,6 +1207,7 @@ class QtMusicApp(QMainWindow):
self.download_btn.setEnabled(True)
self.download_progress.setRange(0, 1)
self.download_progress.setValue(1)
self.set_top_status(self.queue_name)
message = (
f"Готово: {saved} сохранено, {skipped} уже было, "
f"{failed} ошибок"
@@ -1229,6 +1252,58 @@ class QtMusicApp(QMainWindow):
def on_wave_from_history_toggled(self, checked: bool) -> None:
self.update_setting("wave_from_history", checked)
def choose_download_dir(self) -> None:
current = str(self.settings.get("download_dir") or "")
chosen = QFileDialog.getExistingDirectory(
self, "Папка загрузок", current)
if not chosen:
return
self.update_setting("download_dir", chosen)
self.download_dir_edit.setText(chosen)
self.toast(f"Папка загрузок: {chosen}")
# -- быстрое скачивание из меню ---------------------------------------
def download_current(self) -> None:
self.close_drawer()
if not self.current:
self.toast("Сначала выбери трек")
return
self._quick_download([self.current])
def download_queue(self) -> None:
self.close_drawer()
tracks = self.queue or ([self.current] if self.current else [])
if not tracks:
self.toast("Список пуст")
return
self._quick_download(list(tracks))
def _quick_download(self, tracks: list[dict]) -> None:
dest = str(self.settings.get("download_dir") or "")
if not dest:
self.toast("Укажи папку загрузок в настройках")
return
destination = Path(dest).expanduser()
self.download_btn.setEnabled(False)
self.download_progress.setRange(0, 0)
self.download_status.setText("Готовлю список…")
self.set_top_status("Скачиваю…")
self.run_progress_task(
lambda progress: self._download_tracks_job(
tracks, destination, progress),
self.download_done,
self.download_failed,
)
def _download_tracks_job(
self, tracks: list[dict], destination: Path, progress
) -> tuple:
saved, skipped, failed = download_tracks(
self.api, tracks, destination, on_event=progress
)
return saved, skipped, failed, destination
def show_settings(self) -> None:
self._show(self.settings_page)
@@ -1258,6 +1333,8 @@ class QtMusicApp(QMainWindow):
("Поиск", self.show_search),
("Скачать", self.show_download),
("Текст песни", self.show_lyrics),
("Скачать трек", self.download_current),
("Скачать список", self.download_queue),
("Настройки", self.show_settings),
):
button = QPushButton(text)

View File

@@ -3,3 +3,4 @@ requests>=2.31
textual>=0.80
pillow>=10
dbus-next>=0.2.3
mutagen>=1.47

View File

@@ -10,12 +10,15 @@ from mtsmusic import CONFIG_DIR
SETTINGS_PATH = CONFIG_DIR / "settings.json"
DEFAULT_DOWNLOAD_DIR = str(Path.home() / "Music" / "MTS Music")
DEFAULTS: dict[str, Any] = {
"volume": 80, # громкость 0..100
"seek_step": 5, # шаг перемотки стрелками, секунды
"autoplay": False, # запускать воспроизведение сразу
"start_queue": "wave", # wave | favorites | history
"wave_from_history": True, # продолжать волну с последних треков
"download_dir": DEFAULT_DOWNLOAD_DIR, # куда сохранять скачанные MP3
}
START_QUEUES = ("wave", "favorites", "history")
@@ -47,6 +50,11 @@ def _coerce(key: str, value: Any) -> Any:
if key == "start_queue":
text = str(value)
return text if text in START_QUEUES else DEFAULTS[key]
if key == "download_dir":
text = str(value).strip()
if not text:
return DEFAULTS[key]
return str(Path(text).expanduser())
return value

239
tags.py Normal file
View File

@@ -0,0 +1,239 @@
"""ID3-теги для скачанных треков: максимум метаданных.
Пишем ID3v2.3: название, исполнителей, альбом и альбомного исполнителя,
позицию трека и диска, год и дату, жанр, лейбл, длительность, ссылку,
служебные поля (id в Яндекс.Музыке и MTS, explicit), обложку 600×600 и
текст песни (обычный USLT и синхронный SYLT, если он есть).
Модуль необязательный: без `mutagen` теги просто не пишутся.
"""
from __future__ import annotations
import logging
import re
from pathlib import Path
import requests
from media import track_cover_url
from mtsmusic import (
USER_AGENT,
force_ipv4,
lrclib_lyrics,
lyrics_response,
lyrics_text,
)
try:
from mutagen.id3 import (
APIC,
COMM,
SYLT,
TALB,
TCON,
TDAT,
TIT2,
TLEN,
TPE1,
TPE2,
TPOS,
TPUB,
TRCK,
TXXX,
TYER,
USLT,
WOAR,
ID3,
)
HAVE_MUTAGEN = True
except Exception: # pragma: no cover - зависит от окружения
HAVE_MUTAGEN = False
log = logging.getLogger("tags")
COVER_SIZE = 600
_LRC_LINE = re.compile(r"\[(\d{1,2}):(\d{2})(?:[.:](\d{1,3}))?\]")
def _names(items) -> str:
return ", ".join(str(item.get("name") or "") for item in items or [])
def fetch_cover(track: dict) -> bytes | None:
"""Скачивает обложку 600×600 для встраивания в APIC."""
url = track_cover_url(track, COVER_SIZE)
if not url:
return None
try:
force_ipv4()
response = requests.get(
url, headers={"User-Agent": USER_AGENT}, timeout=30)
response.raise_for_status()
return response.content
except requests.RequestException:
return None
def _mts_lyrics(api, track_id: str, fmt: str) -> str:
try:
response = lyrics_response(api, track_id, fmt)
except Exception:
return ""
if response.status_code != 200:
return ""
try:
text = lyrics_text(api, response)
except Exception:
return ""
if "Подключите подписку" in text:
return ""
return text.strip()
def fetch_lyrics(api, track: dict) -> tuple[str, str]:
"""(обычный текст, синхронный LRC). Пробуем MTS, затем LRCLIB."""
track_id = str(track.get("id"))
plain = _mts_lyrics(api, track_id, "TEXT")
synced = _mts_lyrics(api, track_id, "LRC")
if not plain:
plain = (lrclib_lyrics(track, want_sync=False) or "").strip()
if not synced:
synced = (lrclib_lyrics(track, want_sync=True) or "").strip()
if synced and not _LRC_LINE.search(synced):
synced = ""
return plain, synced
def parse_lrc(text: str) -> list[tuple[str, int]]:
"""LRC → [(строка, миллисекунды)] для синхронного SYLT."""
entries: list[tuple[str, int]] = []
for line in text.splitlines():
stamps = list(_LRC_LINE.finditer(line))
if not stamps:
continue
content = _LRC_LINE.sub("", line).strip()
for stamp in stamps:
minutes = int(stamp.group(1))
seconds = int(stamp.group(2))
fraction = (stamp.group(3) or "0").ljust(3, "0")[:3]
millis = (minutes * 60 + seconds) * 1000 + int(fraction)
entries.append((content, millis))
entries.sort(key=lambda item: item[1])
return entries
def _cover_mime(data: bytes) -> str:
return "image/png" if data[:8].startswith(b"\x89PNG") else "image/jpeg"
def write_tags(
track: dict,
path: Path,
plain_lyrics: str = "",
synced_lyrics: str = "",
cover: bytes | None = None,
) -> None:
"""Проставляет ID3v2.3-теги в уже скачанном файле."""
if not HAVE_MUTAGEN:
return
album = (track.get("albums") or [{}])[0]
artist = _names(track.get("artists"))
album_artist = _names(album.get("artists")) or artist
track_id = str(track.get("id") or "")
try:
tags = ID3(path)
except Exception:
tags = ID3()
def add(frame) -> None:
tags.add(frame)
if track.get("title"):
add(TIT2(encoding=3, text=str(track["title"])))
if artist:
add(TPE1(encoding=3, text=artist))
if album_artist:
add(TPE2(encoding=3, text=album_artist))
if album.get("title"):
add(TALB(encoding=3, text=str(album["title"])))
position = album.get("trackPosition") or {}
index = position.get("index")
if index:
count = album.get("trackCount")
add(TRCK(encoding=3, text=f"{index}/{count}" if count else str(index)))
volume = position.get("volume")
if volume:
add(TPOS(encoding=3, text=str(volume)))
if album.get("year"):
add(TYER(encoding=3, text=str(album["year"])))
release = str(album.get("releaseDate") or "")
if release[:10].count("-") == 2:
_year, month, day = release[:10].split("-")
add(TDAT(encoding=3, text=f"{day}{month}")) # формат DDMM
if album.get("genre"):
add(TCON(encoding=3, text=str(album["genre"])))
labels = _names(album.get("labels"))
if labels:
add(TPUB(encoding=3, text=labels))
if track.get("durationMs"):
add(TLEN(encoding=3, text=str(int(track["durationMs"]))))
if track_id:
add(COMM(
encoding=3,
lang="XXX",
desc="",
text=f"https://music.yandex.ru/track/{track_id}",
))
add(WOAR(url=f"https://music.yandex.ru/track/{track_id}"))
def add_text(desc: str, value) -> None:
if value not in (None, ""):
add(TXXX(encoding=3, desc=desc, text=str(value)))
add_text("YANDEX_MUSIC_TRACK_ID", track.get("id"))
add_text("YANDEX_MUSIC_REAL_ID", track.get("realId"))
add_text("YANDEX_MUSIC_ALBUM_ID", album.get("id"))
add_text("MTS_MUSIC_SOURCE", track.get("trackSource"))
add_text("YANDEX_MUSIC_TYPE", track.get("type"))
if track.get("contentWarning") == "explicit":
add_text("ITUNESADVISORY", 1)
if cover:
add(APIC(
encoding=3,
mime=_cover_mime(cover),
type=3,
desc="Cover",
data=cover,
))
if plain_lyrics:
add(USLT(encoding=3, lang="XXX", desc="", text=plain_lyrics))
if synced_lyrics:
entries = parse_lrc(synced_lyrics)
if entries:
add(SYLT(
encoding=3,
lang="XXX",
format=2,
type=1,
desc="",
text=entries,
))
tags.save(path, v2_version=3)
def tag_track(api, track: dict, path: Path) -> None:
"""Собирает метаданные (текст, обложка) и пишет их в файл."""
if not HAVE_MUTAGEN:
return
plain, synced = fetch_lyrics(api, track)
cover = fetch_cover(track)
write_tags(track, Path(path), plain, synced, cover)

181
tui.py
View File

@@ -118,7 +118,10 @@ def track_row(track: dict) -> tuple[str, str, str, str]:
class ListScreen(Screen):
BINDINGS = [Binding("escape", "back", "Назад")]
BINDINGS = [
Binding("escape", "back", "Назад"),
Binding("d", "download", "Скачать"),
]
def __init__(self, title: str, tracks: list[dict]) -> None:
super().__init__()
@@ -144,6 +147,11 @@ class ListScreen(Screen):
def action_back(self) -> None:
self.app.pop_screen()
def action_download(self) -> None:
index = self.query_one("#list-table", DataTable).cursor_row
if 0 <= index < len(self.tracks):
self.app.download_tracks([self.tracks[index]], "трек")
@on(Button.Pressed, "#back")
def back_pressed(self) -> None:
self.action_back()
@@ -161,7 +169,10 @@ class ListScreen(Screen):
class FavoritesScreen(Screen):
BINDINGS = [Binding("escape", "back", "Назад")]
BINDINGS = [
Binding("escape", "back", "Назад"),
Binding("d", "download", "Скачать"),
]
def compose(self) -> ComposeResult:
yield make_topbar("Избранное")
@@ -176,6 +187,12 @@ class FavoritesScreen(Screen):
def action_back(self) -> None:
self.app.pop_screen()
def action_download(self) -> None:
index = self.query_one("#list-table", DataTable).cursor_row
tracks = self.app.favorites
if 0 <= index < len(tracks):
self.app.download_tracks([tracks[index]], "трек")
@on(Button.Pressed, "#back")
def back_pressed(self) -> None:
self.action_back()
@@ -225,7 +242,10 @@ class HistoryScreen(ListScreen):
class SearchScreen(Screen):
BINDINGS = [Binding("escape", "back", "Назад")]
BINDINGS = [
Binding("escape", "back", "Назад"),
Binding("d", "download", "Скачать"),
]
def compose(self) -> ComposeResult:
yield make_topbar("Поиск")
@@ -244,6 +264,11 @@ class SearchScreen(Screen):
def action_back(self) -> None:
self.app.pop_screen()
def action_download(self) -> None:
index = self.query_one("#list-table", DataTable).cursor_row
if 0 <= index < len(self.results):
self.app.download_tracks([self.results[index]], "трек")
@on(Button.Pressed, "#back")
def back_pressed(self) -> None:
self.action_back()
@@ -389,6 +414,10 @@ class SettingsScreen(Screen):
with Horizontal(classes="srow"):
yield Static("Волна с последних треков", classes="slabel")
yield Button("", id="wave-toggle", classes="stoggle")
with Horizontal(classes="srow"):
yield Static("Папка загрузок", classes="slabel")
yield Button("Изменить", id="dir-edit", classes="stoggle")
yield Static("", id="dir-value")
yield Static("", id="settings-path")
def on_mount(self) -> None:
@@ -413,10 +442,17 @@ class SettingsScreen(Screen):
self.query_one("#wave-toggle", Button).label = (
"вкл" if s["wave_from_history"] else "выкл"
)
self.query_one("#dir-value", Static).update(
str(s.get("download_dir") or ""))
def action_back(self) -> None:
self.app.pop_screen()
def dir_chosen(self, value: str | None) -> None:
if value:
self.app.update_setting("download_dir", value)
self.refresh_values()
@on(Button.Pressed)
def pressed(self, event: Button.Pressed) -> None:
bid = event.button.id or ""
@@ -443,6 +479,14 @@ class SettingsScreen(Screen):
self.app.update_setting(
"wave_from_history",
not s["wave_from_history"])
elif bid == "dir-edit":
self.app.push_screen(
PathScreen(
str(s.get("download_dir") or ""),
"Папка загрузок",
),
self.dir_chosen)
return
self.refresh_values()
@@ -454,7 +498,13 @@ class DownloadScreen(Screen):
def compose(self) -> ComposeResult:
yield make_topbar("Скачать")
yield Static("Папка для сохранения:", id="screen-status")
yield Input(value=str(Path.home() / "Music" / "MTS"), id="download-dir")
yield Input(
value=str(
self.app.settings.get("download_dir")
or Path.home() / "Music" / "MTS"
),
id="download-dir",
)
with Horizontal(id="download-actions"):
yield Button("Текущий трек", id="dl-current")
yield Button("Избранное", id="dl-likes")
@@ -584,6 +634,8 @@ class MenuScreen(ModalScreen[str | None]):
yield Button("Поиск", id="m-search")
yield Button("Скачать", id="m-download")
yield Button("Текст песни", id="m-lyrics")
yield Button("Скачать трек", id="m-download-current")
yield Button("Скачать список", id="m-download-list")
yield Button("Настройки", id="m-settings")
yield Static("", id="menu-spacer")
yield Button("Выход", id="m-quit", variant="error")
@@ -596,6 +648,45 @@ class MenuScreen(ModalScreen[str | None]):
self.dismiss(event.button.id)
class PathScreen(ModalScreen[str | None]):
"""Диалог ввода пути (например, папки загрузок)."""
BINDINGS = [Binding("escape", "cancel", "Отмена")]
def __init__(self, value: str, title: str = "Путь") -> None:
super().__init__()
self.value = value
self.dialog_title = title
def compose(self) -> ComposeResult:
with Vertical(id="path-panel"):
yield Static(self.dialog_title, id="path-title")
yield Input(value=self.value, id="path-input")
with Horizontal(id="path-actions"):
yield Button("Сохранить", id="path-save", variant="primary")
yield Button("Отмена", id="path-cancel")
def on_mount(self) -> None:
field = self.query_one("#path-input", Input)
field.focus()
field.cursor_position = len(field.value)
def action_cancel(self) -> None:
self.dismiss(None)
@on(Input.Submitted)
def submitted(self, event: Input.Submitted) -> None:
self.dismiss(event.value.strip() or None)
@on(Button.Pressed)
def pressed(self, event: Button.Pressed) -> None:
if event.button.id == "path-save":
value = self.query_one("#path-input", Input).value.strip()
self.dismiss(value or None)
else:
self.dismiss(None)
# ---------------------------------------------------------------------------
# главный экран
# ---------------------------------------------------------------------------
@@ -737,12 +828,26 @@ class MusicTui(App):
#menu-panel Button { width: 100%; margin: 1 0; }
#menu-spacer { height: 1fr; }
PathScreen { align: center middle; background: #000000 70%; }
#path-panel {
width: 64;
height: auto;
background: #0d0d0d;
border: thick #ffffff;
padding: 1 2;
}
#path-title { text-style: bold; padding: 0 0 1 0; color: #ffffff; }
#path-input { width: 100%; }
#path-actions { height: 3; align: right middle; }
#path-actions Button { width: 14; margin-left: 1; }
#settings-scroll { height: 1fr; padding: 1 2; }
.srow { height: 3; align: left middle; }
.slabel { width: 1fr; content-align: left middle; color: #d0d0d0; }
.svalue { width: 8; content-align: center middle; color: #ffffff; }
.sstep { width: 5; min-width: 5; }
.stoggle { width: 16; min-width: 12; }
#dir-value { padding: 0 0 1 0; color: #8a8a8a; }
#settings-path { padding: 1 0; color: #6a6a6a; }
#download-dir { margin: 0 1; }
@@ -1335,6 +1440,10 @@ class MusicTui(App):
self.push_screen(DownloadScreen())
elif action == "m-lyrics":
self.action_show_lyrics()
elif action == "m-download-current":
self.action_download_current()
elif action == "m-download-list":
self.action_download_list()
elif action == "m-settings":
self.push_screen(SettingsScreen())
elif action == "m-quit":
@@ -1346,6 +1455,70 @@ class MusicTui(App):
self.settings = update_settings(**{key: value})
self.apply_settings()
# -- скачивание -------------------------------------------------------
def action_download_current(self) -> None:
if not self.current:
self.notify("Нет текущего трека", severity="warning")
return
self.download_tracks([self.current], "трек")
def action_download_list(self) -> None:
tracks = self.queue or ([self.current] if self.current else [])
if not tracks:
self.notify("Список пуст", severity="warning")
return
self.download_tracks(list(tracks), self.queue_name)
def download_tracks(self, tracks: list[dict], label: str) -> None:
if not self.api:
return
self.set_topstatus(f"Скачиваю {label}…")
self.download_worker(tracks, label)
@work(thread=True, exclusive=False, group="download")
def download_worker(self, tracks: list[dict], label: str) -> None:
dest = str(self.settings.get("download_dir") or "")
if not dest:
self.call_from_thread(
self.notify,
"Не задана папка загрузок",
severity="warning",
)
return
def on_event(event: dict) -> None:
if event.get("status") != "start":
return
title = str(event["track"].get("title") or "")
self.call_from_thread(
self.set_topstatus,
f"Скачиваю {event['index']}/{event['total']}: {title}",
)
saved, _skipped, _failed = download_tracks(
self.api, tracks, Path(dest).expanduser(), on_event=on_event)
self.call_from_thread(
self.download_finished,
label,
saved,
len(tracks),
dest,
)
def download_finished(
self,
label: str,
saved: int,
total: int,
dest: str,
) -> None:
self.set_topstatus(self.queue_name)
if saved:
self.notify(f"Скачано {saved}/{total} → {dest}", timeout=6)
else:
self.notify("Не удалось скачать", severity="error")
def apply_settings(self) -> None:
volume = int(self.settings.get("volume", 80))
if self.player.volume != volume: