jads/MTS-music-another-client
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

style: format all code to PEP8 (flake8 clean)

Browse Source 
Reformat every module with autopep8 and hand-tune the remaining long
lines so that `flake8` passes with default settings (max line length
79). No behavior changes: long f-strings, messages, QSS/Textual CSS
rules and argument lists are only wrapped or split, an unused import
and an unused local are dropped, and dbus-next D-Bus signature
annotations carry targeted `# noqa` comments.
This commit is contained in:
jads
2026-10-23 23:52:14 +03:00
parent 4460791a95
commit e71f5623a7
7 changed files with 558 additions and 213 deletions
Download Patch File Download Diff File Expand all files Collapse all files

291
mtsmusic.py
View File

@@ -30,9 +30,15 @@ import requests
PROJECT_DIR = Path(__file__).resolve().parent
# Всё локальное храним рядом с проектом, а не в домашней директории.
os.environ.setdefault("PLAYWRIGHT_BROWSERS_PATH", str(PROJECT_DIR / ".playwright"))
os.environ.setdefault(
"PLAYWRIGHT_BROWSERS_PATH", str(
PROJECT_DIR / ".playwright"))
CONFIG_DIR = Path(os.environ.get("MTSMUSIC_CONFIG_DIR", PROJECT_DIR / ".mts-music"))
CONFIG_DIR = Path(
os.environ.get(
"MTSMUSIC_CONFIG_DIR",
PROJECT_DIR /
".mts-music"))
CREDS_PATH = CONFIG_DIR / "credentials.json"
BROWSER_DIR = CONFIG_DIR / "browser"
@@ -54,10 +60,12 @@ _orig_getaddrinfo = socket.getaddrinfo
def force_ipv4() -> None:
"""api.music.yandex.net иногда недоступен по IPv6 — ходим только по IPv4."""
"""api.music.yandex.net иногда недоступен по IPv6 — ходим только
по IPv4."""
def getaddrinfo(host, port, family=0, type=0, proto=0, flags=0):
return _orig_getaddrinfo(host, port, socket.AF_INET, type, proto, flags)
return _orig_getaddrinfo(
host, port, socket.AF_INET, type, proto, flags)
socket.getaddrinfo = getaddrinfo
@@ -83,7 +91,12 @@ def load_credentials() -> dict:
def save_credentials(data: dict) -> None:
CONFIG_DIR.mkdir(parents=True, exist_ok=True)
CREDS_PATH.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
CREDS_PATH.write_text(
json.dumps(
data,
ensure_ascii=False,
indent=2),
encoding="utf-8")
try:
CREDS_PATH.chmod(0o600)
except OSError:
@@ -119,7 +132,9 @@ def get_auth_settings(reference_uri: str = "https://music.mts.ru/") -> dict:
timeout=30,
)
if response.status_code != 200:
die(f"auth_settings вернул HTTP {response.status_code}: {response.text[:300]}")
die(
f"auth_settings вернул HTTP {response.status_code}: "
f"{response.text[:300]}")
return response.json()
@@ -127,7 +142,9 @@ def cmd_login(args: argparse.Namespace) -> None:
try:
from playwright.sync_api import sync_playwright
except ImportError:
die("не установлен playwright. Выполни: pip install -r requirements.txt")
die(
"не установлен playwright. Выполни: "
"pip install -r requirements.txt")
settings = get_auth_settings()
authorize_uri = settings["web_sso_authorize_uri"]
@@ -155,16 +172,20 @@ def cmd_login(args: argparse.Namespace) -> None:
token = None
if token:
break
# Токен может прийти в URL до того, как SPA успеет его сохранить.
# Токен может прийти в URL до того, как SPA успеет его
# сохранить.
if "yatoken=" in page.url:
query = urllib.parse.parse_qs(urllib.parse.urlparse(page.url).query)
query = urllib.parse.parse_qs(
urllib.parse.urlparse(page.url).query)
token = (query.get("yatoken") or [None])[0]
if token:
break
page.wait_for_timeout(1000)
if not token:
die("не дождался токена. Вход не завершён или истекло время ожидания.")
die(
"не дождался токена. Вход не завершён "
"или истекло время ожидания.")
ya_uid = page.evaluate("localStorage.getItem('ya_uid')")
user_id = page.evaluate("localStorage.getItem('userId')")
@@ -213,7 +234,11 @@ class Api:
def post(self, path: str, **params) -> dict:
return self._request("POST", path, params=params)
def _request(self, method: str, path: str, params: dict | None = None) -> dict:
def _request(
self,
method: str,
path: str,
params: dict | None = None) -> dict:
force_ipv4()
last_error: Exception | None = None
for attempt in range(3):
@@ -235,18 +260,24 @@ class Api:
raise AssertionError # для type checker
if response.status_code == 401:
die("токен не принят (401). Выполни заново: python mtsmusic.py login")
die(
"токен не принят (401). Выполни заново: "
"python mtsmusic.py login")
if response.status_code == 404:
die(f"не найдено (404) на {path}")
if response.status_code != 200:
die(f"HTTP {response.status_code} на {path}: {response.text[:300]}")
die(
f"HTTP {response.status_code} на {path}: "
f"{response.text[:300]}")
return response.json()
def get_url(self, url: str) -> dict:
force_ipv4()
response = self.session.get(url, headers=self._headers(), timeout=30)
if response.status_code != 200:
die(f"HTTP {response.status_code} на {url[:120]}: {response.text[:300]}")
die(
f"HTTP {response.status_code} на {url[:120]}: "
f"{response.text[:300]}")
return response.json()
@@ -256,7 +287,10 @@ def require_token(api: Api, what: str) -> None:
def bff_get(path: str) -> dict | None:
"""Запрос к BFF MTS с сохранёнными cookie. None, если cookie нет/не подошли."""
"""Запрос к BFF MTS с сохранёнными cookie.
None, если cookie нет/не подошли.
"""
cookies = load_credentials().get("cookies") or []
if not cookies:
return None
@@ -339,17 +373,27 @@ def cmd_status(args: argparse.Namespace) -> None:
payload = api.get("/account/mts/status")
result = payload.get("result") or {}
account = result.get("account") or {}
subscriptions = ((result.get("masterhub") or {}).get("activeSubscriptions")) or []
subscriptions = ((result.get("masterhub") or {}).get(
"activeSubscriptions")) or []
# BFF даёт человекочитаемое название тарифа.
bff_profile = bff_get("/v1/user-profile") or {}
profile = bff_profile.get("profile") or {}
bff_subscriptions = bff_profile.get("subscriptions") or []
alias = (
profile.get("alias")
or account.get("fullName")
or account.get("displayName")
or account.get("login")
or "?"
)
phone = profile.get("phone") or account.get("phone") or "?"
has_sub = subscriptions or bff_subscriptions
print(f"UID: {account.get('uid', '?')}")
print(f"Имя: {profile.get('alias') or account.get('fullName') or account.get('displayName') or account.get('login') or '?'}")
print(f"Телефон: {profile.get('phone') or account.get('phone') or '?'}")
print(f"Подписка: {'активна' if subscriptions or bff_subscriptions else 'нет'}")
print(f"Имя: {alias}")
print(f"Телефон: {phone}")
print(f"Подписка: {'активна' if has_sub else 'нет'}")
if bff_subscriptions:
for sub in bff_subscriptions:
name = sub.get("name") or sub.get("id") or "?"
@@ -357,13 +401,15 @@ def cmd_status(args: argparse.Namespace) -> None:
print(f" - {name}{f' (до {until})' if until else ''}")
else:
for sub in subscriptions:
name = sub.get("name") or sub.get("productName") or sub.get("subscriptionId") or sub.get("id")
name = sub.get("name") or sub.get("productName") or sub.get(
"subscriptionId") or sub.get("id")
until = sub.get("expires") or sub.get("expirationDate")
print(f" - {name}{f' (до {until})' if until else ''}")
# Дописываем uid в креды, если при логине SPA не успела его сохранить.
creds = load_credentials()
if creds and creds.get("ya_token") == api.token and not creds.get("ya_uid"):
if creds and creds.get(
"ya_token") == api.token and not creds.get("ya_uid"):
creds["ya_uid"] = str(account.get("uid") or "")
save_credentials(creds)
@@ -381,7 +427,9 @@ def get_uid(api: Api) -> str:
result = api.get("/account/mts/status").get("result") or {}
uid = (result.get("account") or {}).get("uid")
if not uid:
die("не удалось определить uid аккаунта. Выполни: python mtsmusic.py login")
die(
"не удалось определить uid аккаунта. Выполни: "
"python mtsmusic.py login")
creds = load_credentials() or {}
creds["ya_uid"] = str(uid)
save_credentials(creds)
@@ -408,21 +456,25 @@ def track_album_ref(track_id: str, album: str | None = None) -> str:
def fetch_mix_batch(api: Api, queue: list[str] | None = None) -> dict:
params = {"queue": ",".join(queue)} if queue else {}
return api.get(f"/rotor/station/{WAVE_STATION}/tracks", **params).get("result") or {}
return api.get(
f"/rotor/station/{WAVE_STATION}/tracks",
**params).get("result") or {}
def fetch_liked_tracks(api: Api, uid: str, limit: int = 300) -> list[dict]:
"""Избранные треки с деталями: /tracks принимает не больше 50 id за раз."""
payload = api.get(f"/users/{uid}/likes/tracks")
refs = ((payload.get("result") or {}).get("library") or {}).get("tracks") or []
refs = ((payload.get("result") or {}).get(
"library") or {}).get("tracks") or []
ids = [str(ref.get("id")) for ref in refs[:limit]]
tracks: list[dict] = []
for start in range(0, len(ids), 50):
chunk = ids[start : start + 50]
chunk = ids[start: start + 50]
if chunk:
tracks.extend(
api.get("/tracks", trackIds=",".join(chunk)).get("result") or []
)
api.get(
"/tracks",
trackIds=",".join(chunk)).get("result") or [])
return tracks
@@ -479,7 +531,8 @@ def cmd_mix(args: argparse.Namespace) -> None:
print_mix_item(index, item)
if args.play:
play_items(api, [item.get("track") or {} for item in collected], args.player)
play_items(api, [item.get("track") or {}
for item in collected], args.player)
def cmd_like(args: argparse.Namespace) -> None:
@@ -517,15 +570,16 @@ def cmd_likes(args: argparse.Namespace) -> None:
limit = max(1, args.number)
liked = liked[:limit]
refs = ",".join(str(item.get("id")) for item in liked)
details: dict[str, dict] = {}
# /tracks отдаёт не больше сотни id за раз — бьём на порции.
ids = [str(item.get("id")) for item in liked]
for start in range(0, len(ids), 50):
chunk = ids[start : start + 50]
chunk = ids[start: start + 50]
try:
tracks = api.get("/tracks", trackIds=",".join(chunk)).get("result") or []
tracks = api.get(
"/tracks",
trackIds=",".join(chunk)).get("result") or []
except SystemExit:
tracks = []
for track in tracks:
@@ -541,13 +595,18 @@ def cmd_likes(args: argparse.Namespace) -> None:
print(f"{index:>3}. {track_id:>12} (альбом {album})")
def lyrics_response(api: Api, track_id: str, fmt: str = "TEXT") -> requests.Response:
def lyrics_response(
api: Api,
track_id: str,
fmt: str = "TEXT") -> requests.Response:
"""Запрос текста с подписью из веб-клиента MTS."""
timestamp = int(time.time())
message = f"{track_id}{timestamp}"
signature = base64.b64encode(
hmac.new(LYRICS_SIGN_KEY.encode(), message.encode(), hashlib.sha256).digest()
).decode()
hmac.new(
LYRICS_SIGN_KEY.encode(),
message.encode(),
hashlib.sha256).digest()).decode()
force_ipv4()
return api.session.get(
@@ -603,7 +662,11 @@ def lrclib_lyrics(track: dict, want_sync: bool) -> str | None:
candidates: list[dict] = []
for params in queries:
try:
response = requests.get(LRCLIB_URL, params=params, headers=headers, timeout=20)
response = requests.get(
LRCLIB_URL,
params=params,
headers=headers,
timeout=20)
except requests.RequestException:
continue
if response.status_code != 200:
@@ -623,13 +686,18 @@ def lrclib_lyrics(track: dict, want_sync: bool) -> str | None:
duration = (track.get("durationMs") or 0) / 1000
def score(item: dict) -> float:
diff = abs(float(item.get("duration") or 0) - duration) if duration else 0
exact = 0 if (item.get("trackName") or "").casefold() == title.casefold() else 50
diff = abs(
float(
item.get("duration") or 0) -
duration) if duration else 0
exact = 0 if (item.get("trackName") or "").casefold(
) == title.casefold() else 50
return diff + exact
usable = [item for item in candidates if (item.get(field) or "").strip()]
if not usable and want_sync:
usable = [item for item in candidates if (item.get("plainLyrics") or "").strip()]
usable = [item for item in candidates if (
item.get("plainLyrics") or "").strip()]
field = "plainLyrics"
if not usable:
return None
@@ -667,7 +735,9 @@ def cmd_lyrics(args: argparse.Namespace) -> None:
if args.output:
output = Path(args.output)
output.write_text(text, encoding="utf-8")
print(f"Сохранено: {output} ({len(text)} символов, источник: {source})")
print(
f"Сохранено: {output} ({
len(text)} символов, источник: {source})")
else:
if source != "MTS":
print(f"# источник текста: {source}", file=sys.stderr)
@@ -690,9 +760,10 @@ def resolve_stream(api: Api, track_id: str) -> tuple[str, dict]:
# Лучший вариант: не preview и максимальный битрейт.
items.sort(
key=lambda item: (not item.get("preview", False), item.get("bitrateInKbps", 0)),
reverse=True,
)
key=lambda item: (
not item.get(
"preview", False), item.get(
"bitrateInKbps", 0)), reverse=True, )
item = items[0]
download_url = item["downloadInfoUrl"]
@@ -705,7 +776,8 @@ def resolve_stream(api: Api, track_id: str) -> tuple[str, dict]:
host = meta["host"]
timestamp = meta["ts"]
digest = hashlib.md5((STREAM_SIGN_SECRET + path + signature).encode()).hexdigest()
digest = hashlib.md5(
(STREAM_SIGN_SECRET + path + signature).encode()).hexdigest()
stream_url = f"https://{host}/get-mp3/{digest}/{timestamp}{path}"
return stream_url, item
@@ -714,7 +786,9 @@ def cmd_url(args: argparse.Namespace) -> None:
api = Api()
url, item = resolve_stream(api, args.track_id)
if item.get("preview"):
print("# preview: полный трек недоступен без активной подписки", file=sys.stderr)
print(
"# preview: полный трек недоступен без активной подписки",
file=sys.stderr)
print(url)
@@ -748,14 +822,20 @@ def play_url(player: str, url: str) -> None:
raise
def play_items(api: Api, tracks: list[dict], explicit_player: str | None = None) -> None:
def play_items(
api: Api,
tracks: list[dict],
explicit_player: str | None = None) -> None:
player = pick_player(explicit_player)
if not player:
die("не найден mpv/ffplay. Установи один из них или используй: mtsmusic.py url <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:
url, item = resolve_stream(api, track_id)
except SystemExit:
@@ -775,11 +855,15 @@ def cmd_play(args: argparse.Namespace) -> None:
print_track_info(api, args.track_id)
url, item = resolve_stream(api, args.track_id)
if item.get("preview"):
print("Внимание: это preview-фрагмент (нет активной подписки).", file=sys.stderr)
print(
"Внимание: это preview-фрагмент (нет активной подписки).",
file=sys.stderr)
player = pick_player(args.player)
if not player:
die("не найден mpv/ffplay. Установи один из них или используй: mtsmusic.py url <id>")
die(
"не найден mpv/ffplay. Установи один из них или используй: "
"mtsmusic.py url <id>")
try:
play_url(player, url)
@@ -794,7 +878,12 @@ def cmd_download(args: argparse.Namespace) -> None:
force_ipv4()
output = Path(args.output or f"{args.track_id}.mp3")
with requests.get(url, stream=True, headers={"User-Agent": USER_AGENT}, timeout=60) as response:
with requests.get(
url,
stream=True,
headers={"User-Agent": USER_AGENT},
timeout=60,
) as response:
response.raise_for_status()
total = int(response.headers.get("Content-Length") or 0)
done = 0
@@ -805,7 +894,12 @@ def cmd_download(args: argparse.Namespace) -> None:
handle.write(chunk)
done += len(chunk)
if total:
print(f"\r{done * 100 // total:3d}% {done // 1024} KiB", end="", file=sys.stderr)
percent = done * 100 // total
kib = done // 1024
print(
f"\r{percent:3d}% {kib} KiB",
end="",
file=sys.stderr)
print(file=sys.stderr)
print(f"Сохранено: {output} ({output.stat().st_size} байт)")
@@ -821,16 +915,24 @@ def cmd_tui(args: argparse.Namespace) -> None:
return
if not sys.stdin.isatty() or not sys.stdout.isatty():
die(
"TUI нужно запускать в обычном терминале (stdin/stdout не TTY).\n"
"Запусти прямо в окне терминала: .venv/bin/python mtsmusic.py tui\n"
"TUI нужно запускать в обычном терминале "
"(stdin/stdout не TTY).\n"
"Запусти прямо в окне терминала: "
".venv/bin/python mtsmusic.py tui\n"
"Если это IDE/панель вывода — открой настоящий терминал."
)
try:
from tui import MusicTui
except ImportError as exc:
die(f"для TUI нужен textual и pillow: {exc}. Установи: pip install -r requirements.txt")
die(
f"для TUI нужен textual и pillow: {exc}. Установи: "
"pip install -r requirements.txt"
)
try:
MusicTui(player=args.player).run(inline=args.inline, mouse=not args.no_mouse)
MusicTui(player=args.player).run(
inline=args.inline,
mouse=not args.no_mouse,
)
except KeyboardInterrupt:
pass
except Exception as exc: # TUI не должен молча исчезать
@@ -869,7 +971,9 @@ def cmd_settings(args: argparse.Namespace) -> None:
key, value = pair.split("=", 1)
key = key.strip()
if key not in DEFAULTS:
die(f"неизвестная настройка: {key}. Доступны: {', '.join(DEFAULTS)}")
die(
f"неизвестная настройка: {key}. Доступны: {
', '.join(DEFAULTS)}")
updates[key] = value
current = save_settings(updates)
saved = True
@@ -887,7 +991,11 @@ def print_tui_check() -> None:
import platform
print(f"Python: {sys.executable}")
print(f"Версия: {platform.python_version()} ({platform.system()} {platform.release()})")
print(
f"Версия: {
platform.python_version()} ({
platform.system()} {
platform.release()})")
print(f"TERM: {os.environ.get('TERM', '?')}")
print(f"COLORTERM: {os.environ.get('COLORTERM', '?')}")
print(f"stdout TTY: {sys.stdout.isatty()}")
@@ -908,7 +1016,9 @@ def print_tui_check() -> None:
print("pillow: НЕ УСТАНОВЛЕН")
for name in ("ffplay", "mpv"):
print(f"{name}: {shutil.which(name) or 'нет'}")
print(f"Креды: {'есть' if CREDS_PATH.exists() else 'нет'} ({CREDS_PATH})")
print(
f"Креды: {
'есть' if CREDS_PATH.exists() else 'нет'} ({CREDS_PATH})")
if size.columns < 90 or size.lines < 30:
print("\nСовет: окно маленькое, TUI лучше смотреть при >= 90x30.")
@@ -920,9 +1030,14 @@ def build_parser() -> argparse.ArgumentParser:
)
sub = parser.add_subparsers(dest="command", required=True)
login = sub.add_parser("login", help="войти через браузер и сохранить токен")
login.add_argument("--headless", action="store_true", help="без окна браузера (капчу не пройти)")
login.add_argument("--timeout", type=int, default=600, help="секунд на вход (по умолчанию 600)")
login = sub.add_parser(
"login", help="войти через браузер и сохранить токен")
login.add_argument(
"--headless",
action="store_true",
help="без окна браузера (капчу не пройти)")
login.add_argument("--timeout", type=int, default=600,
help="секунд на вход (по умолчанию 600)")
login.set_defaults(func=cmd_login)
status = sub.add_parser("status", help="показать аккаунт и подписку")
@@ -930,12 +1045,21 @@ def build_parser() -> argparse.ArgumentParser:
search = sub.add_parser("search", help="поиск треков")
search.add_argument("query", help="поисковый запрос")
search.add_argument("-n", "--limit", type=int, default=10, help="сколько результатов (по умолчанию 10)")
search.add_argument(
"-n",
"--limit",
type=int,
default=10,
help="сколько результатов (по умолчанию 10)")
search.set_defaults(func=cmd_search)
mix = sub.add_parser("mix", help="«Моя волна»: подборка треков")
mix.add_argument("-n", "--number", type=int, default=10, help="сколько треков (по умолчанию 10)")
mix.add_argument("--play", action="store_true", help="сразу воспроизвести подборку")
mix.add_argument("-n", "--number", type=int, default=10,
help="сколько треков (по умолчанию 10)")
mix.add_argument(
"--play",
action="store_true",
help="сразу воспроизвести подборку")
mix.add_argument("--player", help="путь к mpv/ffplay")
mix.set_defaults(func=cmd_mix)
@@ -946,16 +1070,27 @@ def build_parser() -> argparse.ArgumentParser:
unlike = sub.add_parser("unlike", help="убрать трек из избранного")
unlike.add_argument("track_id")
unlike.add_argument("--album", help="id альбома (если нужно переопределить)")
unlike.add_argument(
"--album",
help="id альбома (если нужно переопределить)")
unlike.set_defaults(func=cmd_unlike)
likes = sub.add_parser("likes", help="показать избранные треки")
likes.add_argument("-n", "--number", type=int, default=50, help="сколько показать (по умолчанию 50)")
likes.add_argument(
"-n",
"--number",
type=int,
default=50,
help="сколько показать (по умолчанию 50)")
likes.set_defaults(func=cmd_likes)
lyrics = sub.add_parser("lyrics", help="показать или сохранить текст песни")
lyrics = sub.add_parser(
"lyrics", help="показать или сохранить текст песни")
lyrics.add_argument("track_id")
lyrics.add_argument("--lrc", action="store_true", help="формат LRC с таймкодами")
lyrics.add_argument(
"--lrc",
action="store_true",
help="формат LRC с таймкодами")
lyrics.add_argument("-o", "--output", help="сохранить текст в файл")
lyrics.add_argument(
"--no-fallback",
@@ -964,21 +1099,31 @@ def build_parser() -> argparse.ArgumentParser:
)
lyrics.set_defaults(func=cmd_lyrics)
tui = sub.add_parser("tui", help="полноценный TUI-плеер (мышь, обложка, избранное)")
tui = sub.add_parser(
"tui", help="полноценный TUI-плеер (мышь, обложка, избранное)")
tui.add_argument("--player", help="путь к mpv/ffplay")
tui.add_argument("--check", action="store_true", help="диагностика окружения и выход")
tui.add_argument(
"--check",
action="store_true",
help="диагностика окружения и выход")
tui.add_argument(
"--inline",
action="store_true",
help="без alternate screen: рисовать прямо в терминале (если экран пустой)",
help=(
"без alternate screen: рисовать прямо в терминале "
"(если экран пустой)"),
)
tui.add_argument("--no-mouse", action="store_true", help="отключить поддержку мыши")
tui.add_argument(
"--no-mouse",
action="store_true",
help="отключить поддержку мыши")
tui.set_defaults(func=cmd_tui)
qt = sub.add_parser("qt", help="настольное Qt-приложение (PySide6)")
qt.set_defaults(func=cmd_qt)
settings = sub.add_parser("settings", help="показать или изменить настройки")
settings = sub.add_parser("settings",
help="показать или изменить настройки")
settings.add_argument(
"-s",
"--set",