diff --git a/README.md b/README.md index 07219c2..feb06e3 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # MTS Music CLI -Минимальный клиент для **своего** аккаунта MTS Music. +Минимальный клиент для **своего** аккаунта MTS Music: вход, статус аккаунта +и проверка подписки. Внутри используется тот же путь, что и в веб-клиенте: MTS SSO выдаёт OAuth-токен Яндекс.Музыки, дальше запросы идут на `api.music.yandex.net`. @@ -17,3 +18,20 @@ PLAYWRIGHT_BROWSERS_PATH="$PWD/.playwright" .venv/bin/playwright install chromiu - `.venv/` — зависимости - `.playwright/` — браузер Chromium - `.mts-music/` — профиль браузера и `credentials.json` + +## Вход + +```bash +.venv/bin/python mtsmusic.py login +``` + +Откроется браузер. Нужно вручную войти в MTS и пройти капчу. Токен и cookie +сохранятся в `.mts-music/credentials.json` с правами `600`. +Повторный вход нужен только когда токен перестанет работать. + +## Команды + +```bash +# кто я и есть ли подписка +.venv/bin/python mtsmusic.py status +``` diff --git a/mtsmusic.py b/mtsmusic.py new file mode 100644 index 0000000..2d99107 --- /dev/null +++ b/mtsmusic.py @@ -0,0 +1,378 @@ +#!/usr/bin/env python3 +"""MTS Music CLI — минимальный клиент для личного аккаунта. + +Авторизация: MTS SSO через браузер (Playwright), затем запросы к +Яндекс.Музыке с полученным OAuth-токеном. +""" + +from __future__ import annotations + +import argparse +import base64 +import hashlib +import hmac +import json +import os +import random +import shutil +import socket +import string +import subprocess +import sys +import time +import urllib.parse +from datetime import datetime, timezone +from pathlib import Path +from typing import NoReturn + +import requests + +PROJECT_DIR = Path(__file__).resolve().parent + +# Всё локальное храним рядом с проектом, а не в домашней директории. +os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", str(PROJECT_DIR / ".playwright")) + +CONFIG_DIR = Path(os.environ.get("MTSMUSIC_CONFIG_DIR", PROJECT_DIR / ".mts-music")) +CREDS_PATH = CONFIG_DIR / "credentials.json" +BROWSER_DIR = CONFIG_DIR / "browser" + +AUTH_BASE = "https://api.music.mts.ru/auth" +BFF_BASE = "https://api.music.mts.ru" +YANDEX_BASE = "https://api.music.yandex.net" + +MUSIC_CLIENT = "MTSMusicWebPremium/2.25.0-release-20046039" +# Константа из веб-клиента: подпись URL потока. +STREAM_SIGN_SECRET = "I5gMw9n41e7Ug2JYZV5Td6" +# Константа из веб-клиента: подпись запроса текста песни. +LYRICS_SIGN_KEY = "DBTX5BSsFvbhdajN0ahmza" + +WAVE_STATION = "user:onyourwave" + +USER_AGENT = "MTSMusicCLI/0.1 (+personal use)" + +_orig_getaddrinfo = socket.getaddrinfo + + +def force_ipv4() -> None: + """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) + + socket.getaddrinfo = getaddrinfo + + +def die(message: str, code: int = 1) -> NoReturn: + print(f"error: {message}", file=sys.stderr) + raise SystemExit(code) + + +# --------------------------------------------------------------------------- +# credentials +# --------------------------------------------------------------------------- + + +def load_credentials() -> dict: + if not CREDS_PATH.exists(): + return {} + try: + return json.loads(CREDS_PATH.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + die(f"не удалось прочитать {CREDS_PATH}: {exc}") + + +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") + try: + CREDS_PATH.chmod(0o600) + except OSError: + pass + + +def get_token() -> str | None: + return os.environ.get("MTS_YA_TOKEN") or load_credentials().get("ya_token") + + +# --------------------------------------------------------------------------- +# auth +# --------------------------------------------------------------------------- + + +def make_state(length: int = 24) -> str: + alphabet = string.ascii_letters + string.digits + rnd = random.SystemRandom() + return "".join(rnd.choice(alphabet) for _ in range(length)) + + +def get_auth_settings(reference_uri: str = "https://music.mts.ru/") -> dict: + force_ipv4() + response = requests.get( + f"{AUTH_BASE}/v1/web/auth_settings", + params={"reference_uri": reference_uri, "state": make_state()}, + headers={ + "x-yandex-music-client": MUSIC_CLIENT, + "Origin": "https://music.mts.ru", + "Referer": "https://music.mts.ru/", + "User-Agent": USER_AGENT, + }, + timeout=30, + ) + if response.status_code != 200: + die(f"auth_settings вернул HTTP {response.status_code}: {response.text[:300]}") + return response.json() + + +def cmd_login(args: argparse.Namespace) -> None: + try: + from playwright.sync_api import sync_playwright + except ImportError: + die("не установлен playwright. Выполни: pip install -r requirements.txt") + + settings = get_auth_settings() + authorize_uri = settings["web_sso_authorize_uri"] + print("Открываю браузер. Войди в аккаунт MTS (телефон + SMS/пароль).") + print("Капчу нужно пройти вручную. Окно закроется само после входа.\n") + + CONFIG_DIR.mkdir(parents=True, exist_ok=True) + with sync_playwright() as playwright: + context = playwright.chromium.launch_persistent_context( + str(BROWSER_DIR), + headless=args.headless, + args=["--disable-blink-features=AutomationControlled"], + viewport={"width": 1280, "height": 900}, + ) + try: + page = context.pages[0] if context.pages else context.new_page() + page.goto(authorize_uri, wait_until="domcontentloaded") + + deadline = time.time() + args.timeout + token = None + while time.time() < deadline: + try: + token = page.evaluate("localStorage.getItem('ya_token')") + except Exception: + token = None + if token: + break + # Токен может прийти в URL до того, как SPA успеет его сохранить. + if "yatoken=" in page.url: + 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("не дождался токена. Вход не завершён или истекло время ожидания.") + + ya_uid = page.evaluate("localStorage.getItem('ya_uid')") + user_id = page.evaluate("localStorage.getItem('userId')") + cookies = context.cookies() + + data = { + "ya_token": token, + "ya_uid": ya_uid, + "user_id": user_id, + "cookies": cookies, + "saved_at": datetime.now(timezone.utc).isoformat(), + } + save_credentials(data) + print(f"\nТокен сохранён: {CREDS_PATH} (chmod 600)") + if ya_uid: + print(f"Yandex UID: {ya_uid}") + finally: + context.close() + + +# --------------------------------------------------------------------------- +# API +# --------------------------------------------------------------------------- + + +class Api: + def __init__(self, token: str | None = None) -> None: + self.token = token or get_token() + self.session = requests.Session() + self.session.headers.update( + { + "X-Yandex-Music-Client": MUSIC_CLIENT, + "User-Agent": USER_AGENT, + "Accept": "application/json", + } + ) + + def _headers(self) -> dict: + if self.token: + return {"Authorization": f"OAuth {self.token}"} + return {} + + def get(self, path: str, **params) -> dict: + return self._request("GET", path, params=params) + + 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: + force_ipv4() + last_error: Exception | None = None + for attempt in range(3): + try: + response = self.session.request( + method, + f"{YANDEX_BASE}{path}", + params=params or None, + headers=self._headers(), + timeout=30, + ) + break + except requests.RequestException as exc: + last_error = exc + if attempt < 2: + time.sleep(1.5 * (attempt + 1)) + else: + die(f"сетевая ошибка на {path}: {last_error}") + raise AssertionError # для type checker + + if response.status_code == 401: + 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]}") + 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]}") + return response.json() + + +def require_token(api: Api, what: str) -> None: + if not api.token: + die(f"для «{what}» нужен токен. Сначала: python mtsmusic.py login") + + +def bff_get(path: str) -> dict | None: + """Запрос к BFF MTS с сохранёнными cookie. None, если cookie нет/не подошли.""" + cookies = load_credentials().get("cookies") or [] + if not cookies: + return None + + session = requests.Session() + for cookie in cookies: + session.cookies.set( + cookie["name"], + cookie["value"], + domain=cookie.get("domain"), + path=cookie.get("path", "/"), + ) + force_ipv4() + try: + response = session.get( + f"{BFF_BASE}{path}", + headers={ + "X-Music-Client": "web", + "X-Yandex-Music-Client": MUSIC_CLIENT, + "Origin": "https://music.mts.ru", + "Referer": "https://music.mts.ru/", + "User-Agent": USER_AGENT, + }, + timeout=30, + ) + except requests.RequestException: + return None + if response.status_code != 200: + return None + return response.json() + + +# --------------------------------------------------------------------------- +# catalog +# --------------------------------------------------------------------------- + + +def fmt_duration(ms: int | None) -> str: + total = int((ms or 0) / 1000) + return f"{total // 60}:{total % 60:02d}" + + +def fmt_track(track: dict) -> str: + artists = ", ".join(a.get("name", "?") for a in track.get("artists") or []) + title = track.get("title") or "?" + available = "" if track.get("available", True) else " [недоступен]" + return ( + f"{str(track.get('id', '?')):>12} " + f"{artists} — {title} " + f"[{fmt_duration(track.get('durationMs'))}]{available}" + ) + + +def cmd_status(args: argparse.Namespace) -> None: + api = Api() + require_token(api, "статус") + payload = api.get("/account/mts/status") + result = payload.get("result") or {} + account = result.get("account") 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 [] + + 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 'нет'}") + if bff_subscriptions: + for sub in bff_subscriptions: + name = sub.get("name") or sub.get("id") or "?" + until = sub.get("nextTarifficationDate") + 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") + 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"): + creds["ya_uid"] = str(account.get("uid") or "") + save_credentials(creds) + + +# --------------------------------------------------------------------------- +# cli +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="mtsmusic", + description="Минимальный клиент MTS Music для личного аккаунта.", + ) + 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.set_defaults(func=cmd_login) + + status = sub.add_parser("status", help="показать аккаунт и подписку") + status.set_defaults(func=cmd_status) + + return parser + + +def main() -> None: + parser = build_parser() + args = parser.parse_args() + try: + args.func(args) + except requests.RequestException as exc: + die(f"сетевая ошибка: {exc}") + + +if __name__ == "__main__": + main()