Compare commits

..

No commits in common. "613df10217bc939cf26c141c717b05a5271df32c" and "4f0df8626c7ceae166ad5362d9e8263f0a1bb79e" have entirely different histories.

10 changed files with 341 additions and 2105 deletions

View file

@ -1,17 +0,0 @@
__pycache__/
*.py[cod]
*.pyo
*.pyd
*.db
*.sqlite3
.env
.git
.gitignore
.mypy_cache
.pytest_cache
dist
build
*.egg-info
downloads
*.log
node_modules

View file

@ -1,18 +0,0 @@
YMD_TOKEN=put_your_yandex_music_token_here
YMD_DOWNLOAD_DIR=/app/downloads
YMD_PATH_PATTERN=#album-artist/#album/#number-#title
YMD_QUALITY=2
YMD_LYRICS_FORMAT=none
YMD_COVER_RESOLUTION=400
YMD_COMPATIBILITY_LEVEL=1
YMD_TIMEOUT=20
YMD_TRIES=20
YMD_RETRY_DELAY=5
YMD_DELAY=0
YMD_SKIP_EXISTING=false
YMD_EMBED_COVER=false
YMD_STICK_TO_ARTIST=false
YMD_ONLY_MUSIC=false
YMD_UNSAFE_PATH=false
YMD_DISCO_TRIES=10
YMD_DISCO_RETRY_DELAY=5

172
.gitignore vendored
View file

@ -1,42 +1,160 @@
# Local environment # Byte-compiled / optimized / DLL files
.env
.env.*
!.env.example
# Python
__pycache__/ __pycache__/
*.py[cod] *.py[cod]
*.pyo *$py.class
*.pyd
.python-version
.venv/
venv/
env/
# Build artifacts # C extensions
*.so
# Distribution / packaging
.Python
build/ build/
develop-eggs/
dist/ dist/
downloads/
eggs/
.eggs/ .eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/ *.egg-info/
pip-wheel-metadata/ .installed.cfg
*.egg
MANIFEST
# Tooling caches # PyInstaller
.mypy_cache/ # Usually these files are written by a python script from a template
.pytest_cache/ # before PyInstaller builds the exe, so as to inject date/other infos into it.
.ruff_cache/ *.manifest
.pyre/ *.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/ .tox/
.nox/ .nox/
.coverage .coverage
.coverage.* .coverage.*
htmlcov/ .cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# IDEs and OS files # Translations
.idea/ *.mo
.vscode/ *.pot
.DS_Store
Thumbs.db
# Project runtime files # Django stuff:
downloads/
*.log *.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/

View file

@ -1,24 +0,0 @@
# syntax=docker/dockerfile:1
FROM python:3.11-slim
ENV PYTHONUNBUFFERED=1 \
PIP_DISABLE_PIP_VERSION_CHECK=on \
PIP_NO_CACHE_DIR=1 \
YMD_DOWNLOAD_DIR=/app/downloads \
YMD_PATH_PATTERN="#album-artist/#album/#number-#title"
WORKDIR /app
RUN apt-get update \
&& apt-get install -y --no-install-recommends build-essential \
&& rm -rf /var/lib/apt/lists/*
COPY . .
RUN pip install --upgrade pip \
&& pip install .
EXPOSE 8000
CMD ["uvicorn", "ymd.webui:app", "--host", "0.0.0.0", "--port", "8000"]

321
README.md
View file

@ -1,244 +1,145 @@
# yandex-music-downloader # yandex-music-downloader
Загрузчик музыки из Яндекс Музыки с двумя интерфейсами: > Внимание! В версии v3 был изменен способ авторизации и некоторые
> аргументы. Смотрите [MIGRATION.md](MIGRATION.md) для получения информации
> об изменениях
- `CLI` для точечных и пакетных загрузок ## Содержание
- `Web UI` на FastAPI для загрузки, экспорта дискографии и работы с лайками 1. [О программе](#О-программе)
2. [Установка](#Установка)
3. [Получение данных для авторизации](#Получение-данных-для-авторизации)
4. [Примеры использования](#Примеры-использования)
5. [Спасибо](#Спасибо)
6. [Использование](#Использование)
7. [Уровни совместимости](#Уровни-совместимости)
8. [Дисклеймер](#Дисклеймер)
Проект поддерживает загрузку в AAC и FLAC, сохранение тегов, обложек и текстов, а также запуск через Docker. ## О программе
Загрузчик, созданный вследствие наличия *фатального недостатка* в проекте [yandex-music-download](https://github.com/kaimi-io/yandex-music-download).
> В `v3` изменился способ авторизации и часть аргументов. Если обновляетесь со старой версии, смотрите [MIGRATION.md](MIGRATION.md). ### Возможности
- Возможность загрузки:
## Возможности - Всех треков исполнителя
- Всех треков из альбома
- Скачать все треки артиста, альбом, плейлист или отдельный трек - Всех треков из плейлиста
- Загружать в качестве `AAC 64`, `AAC 192` или `FLAC` - Отдельного трека
- Встраивать обложку в файл - Загрузка всех метаданных трека/альбома:
- Сохранять тексты песен в формате `text` или `lrc` - Номер трека
- Настраивать путь сохранения через шаблон - Номер диска
- Пропускать уже скачанные треки - Название трека
- Отдельно настраивать совместимость тегов - Исполнитель
- Работать через веб-панель без ручного ввода токена - Дополнительные исполнители
- Смотреть лайки в веб-панели - Дата выпуска альбома
- Скачивать из веб-панели: - Обложка альбома
- все понравившиеся треки - Название альбома
- альбомы артистов из лайков - Текст песни (при использовании флага `--add-lyrics`)
- дискографию артиста в JSON - Загрузка треков в lossless качестве
- Поддержка паттерна для пути сохранения музыки
## Требования
- Python `3.9+`
- Токен Яндекс Музыки
## Установка ## Установка
Для запуска скрипта требуется Python 3.9+
Локально:
```bash
pip install .
``` ```
pip install -U https://github.com/llistochek/yandex-music-downloader/archive/main.zip
После установки доступны две команды:
```bash
yandex-music-downloader --help yandex-music-downloader --help
yandex-music-downloader-web --help
``` ```
## Получение токена ## Получение данных для авторизации
Инструкция по получению токена:
https://yandex-music.readthedocs.io/en/main/token.html https://yandex-music.readthedocs.io/en/main/token.html
## Быстрый старт CLI ## Примеры использования
Во всех примерах замените `<Токен>` на ваш токен.
Скачать все треки артиста в FLAC: ### Скачать все треки [Arctic Monkeys](https://music.yandex.ru/artist/208167) в наилучшем качестве
```
```bash yandex-music-downloader --token "<Токен>" --quality 2 --url "https://music.yandex.ru/artist/208167"
yandex-music-downloader \
--token "<TOKEN>" \
--quality 2 \
--url "https://music.yandex.ru/artist/208167"
``` ```
Скачать альбом и сохранить тексты в `LRC`: ### Скачать альбом [Nevermind](https://music.yandex.ru/album/294912) в высоком качестве, загружая тексты песен в формате LRC (с временными метками)
```
```bash yandex-music-downloader --token "<Токен>" --quality 1 --lyrics-format lrc --url "https://music.yandex.ru/album/294912"
yandex-music-downloader \
--token "<TOKEN>" \
--quality 1 \
--lyrics-format lrc \
--url "https://music.yandex.ru/album/294912"
``` ```
Скачать плейлист в свою папку: ### Скачать трек [Seven Nation Army](https://music.yandex.ru/album/11644078/track/6705392)
```
```bash yandex-music-downloader --token "<Токен>" --url "https://music.yandex.ru/album/11644078/track/6705392"
yandex-music-downloader \
--token "<TOKEN>" \
--playlist-id "user/playlists/123" \
--dir "./downloads" \
--skip-existing
``` ```
Скачать один трек: ## Использование
```
usage: yandex-music-downloader [-h] [--quality <Качество>] [--skip-existing]
[--lyrics-format {none,text,lrc}]
[--embed-cover]
[--cover-resolution <Разрешение обложки>]
[--delay <Задержка>] [--stick-to-artist]
[--only-music]
[--compatibility-level <Уровень совместимости>]
[--timeout <Время ожидания>]
(--artist-id <ID исполнителя> | --album-id <ID альбома> | --track-id <ID трека> | --playlist-id <владелец плейлиста>/<тип плейлиста> | -u URL)
[--unsafe-path] [--dir <Папка>]
[--path-pattern <Паттерн>] --token <Токен>
```bash Загрузчик музыки с сервиса Яндекс.Музыка
yandex-music-downloader \
--token "<TOKEN>" \ options:
--url "https://music.yandex.ru/album/11644078/track/6705392" -h, --help show this help message and exit
Общие параметры:
--quality <Качество> Качество трека:
0 - Низкое (AAC 64kbps)
1 - Оптимальное (AAC 192kbps)
2 - Лучшее (FLAC)
(по умолчанию: 0)
--skip-existing Пропускать уже загруженные треки
--lyrics-format {none,text,lrc}
Формат текста песни (по умолчанию: none)
--embed-cover Встраивать обложку в аудиофайл
--cover-resolution <Разрешение обложки>
Разрешение обложки (в пикселях). Передайте "original" для загрузки в оригинальном (наилучшем) разрешении (по умолчанию: 400)
--delay <Задержка> Задержка между запросами, в секундах (по умолчанию: 0)
--stick-to-artist Загружать альбомы, созданные только данным исполнителем
--only-music Загружать только музыкальные альбомы (пропускать подкасты и аудиокниги)
--compatibility-level <Уровень совместимости>
Уровень совместимости, от 0 до 1. См. README для подробного описания (по умолчанию: 1)
--timeout <Время ожидания>
Время ожидания ответа от сервера, в секундах. Увеличьте если возникают сетевые ошибки (по умолчанию: 20)
ID:
--artist-id <ID исполнителя>
--album-id <ID альбома>
--track-id <ID трека>
--playlist-id <владелец плейлиста>/<тип плейлиста>
-u URL, --url URL URL исполнителя/альбома/трека/плейлиста
Указание пути:
--unsafe-path Не очищать путь от недопустимых символов
--dir <Папка> Папка для загрузки музыки (по умолчанию: .)
--path-pattern <Паттерн>
Поддерживает следующие заполнители: #number, #artist, #album-artist, #title, #album, #year, #artist-id, #album-id, #track-id, #number-padded (по умолчанию: #album-artist/#album/#number - #title)
Авторизация:
--token <Токен> Токен для авторизации. См. README для способов получения
``` ```
## Основные аргументы CLI ## Уровни совместимости
Уровень совместимости позволяет отойти от стандарта тегов, которого
придерживается библиотека mutagen. Сделано это для поддержки большего
количества музыкальных плееров. Ниже подробно описаны все уровни.
- `--url` или один из `--artist-id`, `--album-id`, `--track-id`, `--playlist-id` ### 0
- `--token` для авторизации Стандартные теги mutagen.
- `--quality 0|1|2`
- `--lyrics-format none|text|lrc`
- `--embed-cover`
- `--cover-resolution 400|800|original`
- `--skip-existing`
- `--dir`
- `--path-pattern`
- `--unsafe-path`
- `--stick-to-artist`
- `--only-music`
- `--timeout`
- `--tries`
- `--retry-delay`
Полный список: ### 1
Затрагиваемые форматы: `m4a`
```bash - Теги с несколькими значениями (`\xa9ART` и `aART`) устанвливаются с
yandex-music-downloader --help разделителем `;`. Пример: `Artist1; Artist2; Artist3`
```
## Веб-панель
Веб-панель использует токен только из переменной окружения `YMD_TOKEN`. В интерфейсе поле токена отсутствует.
Локальный запуск:
```bash
export YMD_TOKEN="<TOKEN>"
yandex-music-downloader-web --host 127.0.0.1 --port 8000
```
После запуска откройте:
```text
http://127.0.0.1:8000
```
Что умеет веб-панель:
- загрузка по ссылке, `artist_id`, `album_id`, `track_id`, `playlist_id`
- экспорт дискографии артиста в JSON
- просмотр лайкнутых треков
- отдельный список артистов из лайков
- кнопка `Скачать всё понравившееся`
- кнопка `Скачать альбомы артистов`
- отображение активных задач и прогресса
## Запуск через Docker
1. Создайте `.env` из шаблона:
```bash
cp .env.example .env
```
2. Укажите токен и при необходимости остальные параметры.
3. Запустите контейнер:
```bash
docker compose up --build -d
```
4. Откройте веб-панель:
```text
http://localhost:8228
```
Загрузка по умолчанию сохраняется в локальную папку `./downloads`.
## Переменные окружения для Web UI / Docker
Полный набор есть в [.env.example](.env.example). Основные переменные:
- `YMD_TOKEN` — токен Яндекс Музыки
- `YMD_DOWNLOAD_DIR` — корневая папка загрузки
- `YMD_PATH_PATTERN` — шаблон пути
- `YMD_QUALITY` — качество по умолчанию (`0`, `1`, `2`)
- `YMD_LYRICS_FORMAT``none`, `text`, `lrc`
- `YMD_COVER_RESOLUTION` — размер обложки
- `YMD_COMPATIBILITY_LEVEL` — уровень совместимости тегов
- `YMD_SKIP_EXISTING` — пропускать существующие файлы
- `YMD_EMBED_COVER` — встраивать обложку
- `YMD_STICK_TO_ARTIST` — брать только альбомы самого артиста
- `YMD_ONLY_MUSIC` — исключать подкасты и аудиокниги
- `YMD_TIMEOUT` — таймаут запросов
- `YMD_TRIES` — число повторов
- `YMD_RETRY_DELAY` — пауза между повторами
- `YMD_DELAY` — задержка между треками
- `YMD_UNSAFE_PATH` — не чистить путь от небезопасных символов
## Шаблон пути
По умолчанию используется:
```text
#album-artist/#album/#number-#title
```
Поддерживаемые заполнители:
- `#number`
- `#number-padded`
- `#track-artist`
- `#album-artist`
- `#title`
- `#album`
- `#year`
- `#artist-id`
- `#album-id`
- `#track-id`
Пример:
```text
#album-artist/#album/#number-padded-#title
```
## Уровни совместимости тегов
### `0`
Стандартное поведение `mutagen`.
### `1`
Для `m4a` многозначные поля (`©ART`, `aART`) записываются через `;`, что лучше работает в части плееров.
## Изменения в текущей версии
- Добавлен Web UI на FastAPI
- Добавлен Docker-образ и `docker compose`
- Токен для веб-панели переведен только в `YMD_TOKEN`
- Добавлен просмотр лайков
- Добавлены действия `Скачать всё понравившееся` и `Скачать альбомы артистов`
- Улучшена логика построения путей и переименования по тегам
## Спасибо ## Спасибо
- Разработчикам проекта [yandex-music-api](https://github.com/MarshalX/yandex-music-api) - Разработчикам проекта [yandex-music-api](https://github.com/MarshalX/yandex-music-api)
- @ArtemBay за [скрипт](https://github.com/MarshalX/yandex-music-api/issues/656#issuecomment-2306542725) получения ссылки на загрузку в lossless качестве - @ArtemBay за [скрипт](https://github.com/MarshalX/yandex-music-api/issues/656#issuecomment-2306542725) получения ссылки на загрузку в lossless качестве
- @keltecc за [метод дешифрования файлов](https://github.com/llistochek/yandex-music-downloader/issues/112#issuecomment-2812535100) - @keltecc за [метод дешифрования файлов](https://github.com/llistochek/yandex-music-downloader/issues/112#issuecomment-2812535100)
- @leowerd за [идею корректных имен исполнителей](https://github.com/llistochek/yandex-music-downloader/issues/93#issuecomment-2960210879) при загрузке сборников
## Дисклеймер ## Дисклеймер
Данный проект является независимой разработкой и никак не связан с компанией Яндекс.
Проект не связан с Яндексом и является независимой разработкой.

View file

@ -1,12 +0,0 @@
version: "3.9"
services:
ymd:
build: .
container_name: ymd
env_file: .env
ports:
- "8228:8000"
volumes:
- ./downloads:/app/downloads
restart: unless-stopped

View file

@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
[project] [project]
name = "yandex-music-downloader" name = "yandex-music-downloader"
version = "3.5.4" version = "3.4.7"
description = "Загрузчик музыки с сервиса Яндекс.Музыка" description = "Загрузчик музыки с сервиса Яндекс.Музыка"
requires-python = ">=3.9" requires-python = ">=3.9"
readme = "README.md" readme = "README.md"
@ -15,12 +15,8 @@ dependencies = [
"yandex-music @ https://github.com/llistochek/yandex-music-api/archive/9623fbca7704f47766614efe51d66c9fd496714c.zip", "yandex-music @ https://github.com/llistochek/yandex-music-api/archive/9623fbca7704f47766614efe51d66c9fd496714c.zip",
"mutagen>=1.47.0", "mutagen>=1.47.0",
"StrEnum", "StrEnum",
"pycryptodome", "pycryptodome"
"fastapi>=0.110.0",
"uvicorn[standard]>=0.23.0",
"pydantic>=2.7.0"
] ]
[project.scripts] [project.scripts]
yandex-music-downloader = "ymd.cli:main" yandex-music-downloader = "ymd.cli:main"
yandex-music-downloader-web = "ymd.webui:main"

View file

@ -6,7 +6,7 @@ import re
import time import time
import typing import typing
from argparse import ArgumentTypeError from argparse import ArgumentTypeError
from collections.abc import Callable, Generator, Iterable from collections.abc import Generator, Iterable
from pathlib import Path from pathlib import Path
from typing import Optional, Union from typing import Optional, Union
from urllib.parse import urlparse from urllib.parse import urlparse
@ -34,25 +34,35 @@ def show_default(text: Optional[str] = None) -> str:
return f"{text} ({default})" return f"{text} ({default})"
def checked_int_arg( def quality_arg(astr: str) -> int:
min_value: int, max_value: Optional[int] = None aint = int(astr)
) -> Callable[[str], int]: if 0 <= aint <= 2:
def func(astr: str) -> int: return aint
aint = int(astr) raise ArgumentTypeError("Значение должно быть в промежутке от 0 до 2")
if aint >= min_value and (max_value is None or aint <= max_value):
return aint
error_text = f"Значение должен быть >= {min_value}"
if max_value is not None:
error_text += f" и <= {max_value}"
raise ArgumentTypeError(error_text)
return func
def compatibility_level_arg(astr: str) -> int:
aint = int(astr)
min_val = core.MIN_COMPATIBILITY_LEVEL
max_val = core.MAX_COMPATIBILITY_LEVEL
if min_val <= aint <= max_val:
return aint
raise ArgumentTypeError(
f"Значение должен быть в промежутке от {min_val} до {max_val}"
)
def natural_int_arg(astr: str) -> int:
aint = int(astr)
if aint > 0:
return aint
raise ArgumentTypeError("Значение должен быть > 0")
def cover_resolution_arg(astr: str) -> int: def cover_resolution_arg(astr: str) -> int:
if astr == "original": if astr == "original":
return -1 return -1
return checked_int_arg(100)(astr) return int(astr)
def lyrics_format_arg(astr: str) -> core.LyricsFormat: def lyrics_format_arg(astr: str) -> core.LyricsFormat:
@ -73,7 +83,7 @@ def main():
"--quality", "--quality",
metavar="<Качество>", metavar="<Качество>",
default=0, default=0,
type=checked_int_arg(0, 2), type=quality_arg,
help="Качество трека:\n0 - Низкое (AAC 64kbps)\n1 - Оптимальное (AAC 192kbps)\n2 - Лучшее (FLAC)\n(по умолчанию: %(default)s)", help="Качество трека:\n0 - Низкое (AAC 64kbps)\n1 - Оптимальное (AAC 192kbps)\n2 - Лучшее (FLAC)\n(по умолчанию: %(default)s)",
) )
common_group.add_argument( common_group.add_argument(
@ -105,7 +115,7 @@ def main():
"--delay", "--delay",
default=DEFAULT_DELAY, default=DEFAULT_DELAY,
metavar="<Задержка>", metavar="<Задержка>",
type=checked_int_arg(0), type=int,
help=show_default("Задержка между запросами, в секундах"), help=show_default("Задержка между запросами, в секундах"),
) )
common_group.add_argument( common_group.add_argument(
@ -122,40 +132,20 @@ def main():
"--compatibility-level", "--compatibility-level",
metavar="<Уровень совместимости>", metavar="<Уровень совместимости>",
default=1, default=1,
type=checked_int_arg( type=compatibility_level_arg,
core.MIN_COMPATIBILITY_LEVEL, core.MAX_COMPATIBILITY_LEVEL
),
help=show_default( help=show_default(
f"Уровень совместимости, от {core.MIN_COMPATIBILITY_LEVEL} до {core.MAX_COMPATIBILITY_LEVEL}. См. README для подробного описания" f"Уровень совместимости, от {core.MIN_COMPATIBILITY_LEVEL} до {core.MAX_COMPATIBILITY_LEVEL}. См. README для подробного описания"
), ),
) )
common_group.add_argument(
network_group = parser.add_argument_group("Сетевые параметры")
network_group.add_argument(
"--timeout", "--timeout",
metavar="<Время ожидания>", metavar="<Время ожидания>",
default=20, default=20,
type=checked_int_arg(1), type=natural_int_arg,
help=show_default( help=show_default(
"Время ожидания ответа от сервера, в секундах. Увеличьте если возникают сетевые ошибки" "Время ожидания ответа от сервера, в секундах. Увеличьте если возникают сетевые ошибки"
), ),
) )
network_group.add_argument(
"--tries",
metavar="<Количество попыток>",
default=20,
type=checked_int_arg(0),
help=show_default(
"Количество попыток при возникновении сетевых ошибок. 0 - бесконечное количество попыток"
),
)
network_group.add_argument(
"--retry-delay",
metavar="<Задержка>",
default=5,
type=checked_int_arg(0),
help=show_default("Задержка между повторными запросами при сетевых ошибках"),
)
common_group.add_argument("--debug", action="store_true", help=argparse.SUPPRESS) common_group.add_argument("--debug", action="store_true", help=argparse.SUPPRESS)
id_group_meta = parser.add_argument_group("ID") id_group_meta = parser.add_argument_group("ID")
@ -189,7 +179,7 @@ def main():
type=Path, type=Path,
help=show_default( help=show_default(
"Поддерживает следующие заполнители:" "Поддерживает следующие заполнители:"
" #number, #track-artist, #album-artist, #title," " #number, #artist, #album-artist, #title,"
" #album, #year, #artist-id, #album-id, #track-id, #number-padded" " #album, #year, #artist-id, #album-id, #track-id, #number-padded"
), ),
) )
@ -231,13 +221,8 @@ def main():
print("Параметер url указан в неверном формате") print("Параметер url указан в неверном формате")
return 1 return 1
client = core.init_client( client = core.init_client(args.token, args.timeout)
token=args.token, result_tracks: Iterable[Track] = []
timeout=args.timeout,
max_try_count=args.tries,
retry_delay=args.retry_delay,
)
result_tracks: Iterable[Track]
def album_tracks_gen(album_ids: Iterable[Union[int, str]]) -> Generator[Track]: def album_tracks_gen(album_ids: Iterable[Union[int, str]]) -> Generator[Track]:
for album_id in album_ids: for album_id in album_ids:
@ -245,7 +230,6 @@ def main():
if volumes := full_album.volumes: if volumes := full_album.volumes:
yield from itertools.chain.from_iterable(volumes) yield from itertools.chain.from_iterable(volumes)
total_track_count = None
if args.artist_id is not None: if args.artist_id is not None:
def filter_album(album: Album) -> bool: def filter_album(album: Album) -> bool:
@ -260,46 +244,31 @@ def main():
return True return True
return False return False
def albums_id_gen() -> Generator[int]: def albums_gen() -> Generator[Album]:
has_next = True has_next = True
page = 0 page = 0
while has_next: while has_next:
if albums_info := client.artists_direct_albums(args.artist_id, page): if albums := client.artists_direct_albums(args.artist_id, page):
for album in albums_info.albums: yield from albums.albums
if filter_album(album):
assert album.id
yield album.id
else:
nonlocal total_track_count
if (
track_count := album.track_count
) and total_track_count is not None:
total_track_count -= track_count
else: else:
break break
if pager := albums_info.pager: if pager := albums.pager:
page = pager.page + 1 page = pager.page + 1
has_next = pager.per_page * page < pager.total has_next = pager.per_page * page < pager.total
else: else:
break break
result_tracks = album_tracks_gen(albums_id_gen()) result_tracks = album_tracks_gen(
artist = client.artists(args.artist_id)[0] a.id for a in albums_gen() if filter_album(a) and a.id is not None
if counts := artist.counts: )
total_track_count = counts.tracks
elif args.album_id is not None: elif args.album_id is not None:
result_tracks = album_tracks_gen((args.album_id,)) result_tracks = album_tracks_gen((args.album_id,))
if album := client.albums_with_tracks(args.album_id):
total_track_count = album.track_count
elif args.track_id is not None: elif args.track_id is not None:
track = client.tracks(args.track_id) track = client.tracks(args.track_id)
result_tracks = track result_tracks = track
total_track_count = 1
elif args.playlist_id is not None: elif args.playlist_id is not None:
user, kind = args.playlist_id.split("/") user, kind = args.playlist_id.split("/")
playlist = typing.cast(Playlist, client.users_playlists(kind, user)) playlist = typing.cast(Playlist, client.users_playlists(kind, user))
total_track_count = playlist.track_count
def playlist_tracks_gen() -> Generator[Track]: def playlist_tracks_gen() -> Generator[Track]:
tracks = playlist.fetch_tracks() tracks = playlist.fetch_tracks()
@ -309,29 +278,17 @@ def main():
) )
result_tracks = playlist_tracks_gen() result_tracks = playlist_tracks_gen()
else:
raise ValueError("Invalid ID argument")
track_counter = 0
progress_status = ""
covers_cache = {} covers_cache = {}
for track in result_tracks: for track in result_tracks:
track_counter += 1
if total_track_count:
progress_status = f"[{track_counter}/{total_track_count}] "
else:
progress_status = ""
if not track.available: if not track.available:
print(f"{progress_status}Трек {track.title} не доступен для скачивания") print(f"Трек {track.title} не доступен для скачивания")
continue continue
save_path = args.dir / core.prepare_base_path( save_path = args.dir / core.prepare_base_path(
args.path_pattern, args.path_pattern,
track, track,
args.unsafe_path, args.unsafe_path,
fallback_number=track_counter,
fallback_total=total_track_count,
) )
if args.skip_existing: if args.skip_existing:
if any( if any(
@ -349,7 +306,7 @@ def main():
if bitrate > 0: if bitrate > 0:
format_info += f" {bitrate}kbps" format_info += f" {bitrate}kbps"
format_info += "]" format_info += "]"
print(f"{progress_status}{format_info} Загружается {downloadable.path}") print(f"{format_info} Загружается {downloadable.path}")
core.download_track( core.download_track(
track_info=downloadable, track_info=downloadable,
lyrics_format=args.lyrics_format, lyrics_format=args.lyrics_format,
@ -357,11 +314,6 @@ def main():
cover_resolution=args.cover_resolution, cover_resolution=args.cover_resolution,
covers_cache=covers_cache, covers_cache=covers_cache,
compatibility_level=args.compatibility_level, compatibility_level=args.compatibility_level,
path_pattern=args.path_pattern,
unsafe_path=args.unsafe_path,
fallback_number=track_counter,
fallback_total=total_track_count,
rename_using_tags=True,
) )
if args.delay > 0: if args.delay > 0:
time.sleep(args.delay) time.sleep(args.delay)

View file

@ -1,7 +1,5 @@
import datetime as dt import datetime as dt
import hashlib
import re import re
import time
import typing import typing
from collections.abc import Callable from collections.abc import Callable
from dataclasses import dataclass from dataclasses import dataclass
@ -14,7 +12,6 @@ from mutagen.flac import FLAC, Picture
from mutagen.id3._frames import ( from mutagen.id3._frames import (
APIC, APIC,
TALB, TALB,
TCON,
TDRC, TDRC,
TIT2, TIT2,
TPE1, TPE1,
@ -34,7 +31,6 @@ from yandex_music import (
Track, Track,
YandexMusicModel, YandexMusicModel,
) )
from yandex_music.exceptions import NetworkError
from ymd import api from ymd import api
from ymd.api import ( from ymd.api import (
@ -48,7 +44,7 @@ from ymd.mime_utils import MimeType, guess_mime_type
UNSAFE_PATH_CLEAR_RE = re.compile(r"[/\\]+") UNSAFE_PATH_CLEAR_RE = re.compile(r"[/\\]+")
SAFE_PATH_CLEAR_RE = re.compile(r"([^\w\-\'() ]|^\s+|\s+$)") SAFE_PATH_CLEAR_RE = re.compile(r"([^\w\-\'() ]|^\s+|\s+$)")
DEFAULT_PATH_PATTERN = Path("#album-artist", "#album", "#number-#title") DEFAULT_PATH_PATTERN = Path("#album-artist", "#album", "#number - #title")
DEFAULT_COVER_RESOLUTION = 400 DEFAULT_COVER_RESOLUTION = 400
MIN_COMPATIBILITY_LEVEL = 0 MIN_COMPATIBILITY_LEVEL = 0
@ -56,9 +52,6 @@ MAX_COMPATIBILITY_LEVEL = 1
AUDIO_FILE_SUFFIXES = {".mp3", ".flac", ".m4a"} AUDIO_FILE_SUFFIXES = {".mp3", ".flac", ".m4a"}
TEMPORARY_FILE_NAME_TEMPLATE = ".yandex-music-downloader.{}.tmp" TEMPORARY_FILE_NAME_TEMPLATE = ".yandex-music-downloader.{}.tmp"
MAX_FILE_NAME_LENGTH_WITHOUT_SUFFIX = 255 - max(
len(suffix) for suffix in AUDIO_FILE_SUFFIXES
)
class CoreTrackQuality(IntEnum): class CoreTrackQuality(IntEnum):
@ -93,180 +86,58 @@ class AlbumCover:
mime_type: MimeType mime_type: MimeType
def init_client( def init_client(token: str, timeout: int) -> Client:
token: str, timeout: int, max_try_count: int, retry_delay: int
) -> Client:
assert timeout > 0
assert max_try_count >= 0
assert retry_delay >= 0
client = Client(token) client = Client(token)
client.request.set_timeout(timeout) client.request.set_timeout(timeout)
original_wrapper = client.request._request_wrapper
def retry_wrapper(*args, **kwargs):
try_count = 0
while True:
try:
return original_wrapper(*args, **kwargs)
except NetworkError as error:
if max_try_count == 0 or try_count < max_try_count:
try_count += 1
time.sleep(retry_delay)
continue
raise error
client.request._request_wrapper = retry_wrapper
return client.init() return client.init()
def full_title(obj: YandexMusicModel) -> str: def full_title(obj: YandexMusicModel) -> str:
"""Return title with optional version; be tolerant to missing keys.""" result = obj["title"]
title = getattr(obj, "title", None) if result is None:
if title is None:
try:
title = obj["title"]
except Exception: # noqa: BLE001
title = None
if not title:
return "" return ""
if version := obj["version"]:
version = getattr(obj, "version", None) result += f" ({version})"
if version is None: return result
try:
version = obj["version"]
except Exception: # noqa: BLE001
version = None
if version:
title = f"{title} ({version})"
return title
def sanitize_path_component(text: str, unsafe_path: bool) -> str:
"""Return path-safe component."""
if unsafe_path:
return UNSAFE_PATH_CLEAR_RE.sub("_", text)
text = text.strip()
text = SAFE_PATH_CLEAR_RE.sub("_", text)
text = re.sub(r"\s+", "_", text)
text = re.sub(r"_+", "_", text)
return text
def choose_first(*values: Optional[Union[str, int]]) -> str:
for candidate in values:
if candidate is None:
continue
text = str(candidate).strip()
if text:
return text
return ""
def prepare_base_path( def prepare_base_path(
path_pattern: Path, path_pattern: Path, track: Track, unsafe_path: bool = False
track: Track,
unsafe_path: bool = False,
fallback_number: Optional[int] = None,
fallback_total: Optional[int] = None,
override_title: Optional[str] = None,
override_album: Optional[str] = None,
override_album_artist: Optional[str] = None,
override_track_artist: Optional[str] = None,
override_track_number: Optional[int] = None,
) -> Path: ) -> Path:
path_str = str(path_pattern) path_str = str(path_pattern)
album = None album = None
album_artist = None artist = None
track_artist = None
track_position = None track_position = None
if albums := track.albums: if albums := track.albums:
album = albums[0] album = albums[0]
track_position = album.track_position
if artists := album.artists: if artists := album.artists:
album_artist = artists[0] artist = artists[0]
# позиция трека корректнее брать из самого трека if artist is None and (artists := track.artists):
track_position = getattr(track, "track_position", None) or track_position artist = artists[0]
if artists := track.artists:
track_artist = artists[0]
track_id = getattr(track, "id", None)
track_title = choose_first(
override_title,
full_title(track),
f"track_{track_id}" if track_id is not None else None,
f"track_{fallback_number}" if fallback_number is not None else None,
"track",
)
album_title = choose_first(
override_album,
full_title(album) if album else None,
str(album.id) if album and album.id is not None else None,
"album",
)
album_artist_name = choose_first(
override_album_artist,
album_artist.name if album_artist else None,
"artist",
)
track_artist_name = choose_first(
override_track_artist,
track_artist.name if track_artist else None,
album_artist_name,
"artist",
)
pad_width = None
if track_position and album and album.track_count:
pad_width = max(2, len(str(album.track_count)))
elif fallback_total:
pad_width = max(2, len(str(fallback_total)))
elif fallback_number:
pad_width = max(2, len(str(fallback_number)))
number = (
override_track_number
if override_track_number is not None
else (track_position.index if track_position else fallback_number)
)
number_padded = (
str(number).zfill(pad_width) if number is not None and pad_width is not None else number
)
number_value = number_padded if number_padded is not None else number
repl_dict: dict[str, Union[str, int, None]] = { repl_dict: dict[str, Union[str, int, None]] = {
"#number-padded": number_padded, "#number-padded": str(track_position.index).zfill(len(str(album.track_count)))
"#album-artist": album_artist_name, if track_position and album
"#track-artist": track_artist_name, else None,
"#artist-id": track_artist.id if track_artist else None, "#album-artist": artist.name if artist else None,
"#artist-id": artist.id if artist else None,
"#album-id": album.id if album else None, "#album-id": album.id if album else None,
"#track-id": track.id, "#track-id": track.id,
"#number": number_value, "#number": track_position.index if track_position else None,
"#title": track_title, "#artist": artist.name if artist else None,
"#album": album_title, "#title": full_title(track),
"#album": full_title(album) if album else None,
"#year": album.year if album else None, "#year": album.year if album else None,
} }
for placeholder, replacement in repl_dict.items(): for placeholder, replacement in repl_dict.items():
if replacement is None: replacement = str(replacement)
replacement = "" if not unsafe_path:
replacement = sanitize_path_component(str(replacement), unsafe_path) clear_re = SAFE_PATH_CLEAR_RE
if not replacement: else:
replacement = sanitize_path_component("untitled", unsafe_path) clear_re = UNSAFE_PATH_CLEAR_RE
replacement = clear_re.sub("_", replacement)
path_str = path_str.replace(placeholder, replacement) path_str = path_str.replace(placeholder, replacement)
path = Path(path_str) return Path(path_str)
if not unsafe_path:
path = Path(
*(
re.sub(r"_+", "_", part.replace(" ", "_"))
for part in path.parts
)
)
trimmed_parts = [
part
if len(part) <= MAX_FILE_NAME_LENGTH_WITHOUT_SUFFIX
else part[:MAX_FILE_NAME_LENGTH_WITHOUT_SUFFIX]
for part in path.parts
]
return Path(*trimmed_parts)
def set_tags( def set_tags(
@ -276,57 +147,22 @@ def set_tags(
lyrics: Optional[str], lyrics: Optional[str],
album_cover: Optional[AlbumCover], album_cover: Optional[AlbumCover],
compatibility_level: int, compatibility_level: int,
source_tags: Optional[dict[str, Optional[Union[str, int]]]] = None,
) -> None: ) -> None:
album = track.albums[0] if track.albums else Album()
track_artists = [a.name for a in track.artists if a.name]
album_artists = [a.name for a in album.artists if a.name]
file_type = CONTAINER_MUTAGEN_MAPPING.get(container) file_type = CONTAINER_MUTAGEN_MAPPING.get(container)
if file_type is None: if file_type is None:
raise ValueError(f"Unknown container: {container}") raise ValueError(f"Unknown container: {container}")
source_tags = source_tags or {}
def as_list(value: Optional[Union[str, list[str], tuple[str, ...]]]) -> list[str]:
if value is None:
return []
if isinstance(value, (list, tuple, set)):
return [str(v).strip() for v in value if str(v).strip()]
text = str(value).strip()
return [text] if text else []
tag = file_type(path) tag = file_type(path)
album = track.albums[0] if track.albums else Album() album_title = full_title(album)
album_title = choose_first( track_title = full_title(track)
source_tags.get("album"),
full_title(album),
str(album.id) if getattr(album, "id", None) is not None else None,
)
track_title = choose_first(
source_tags.get("title"),
full_title(track),
f"track_{track.id}" if getattr(track, "id", None) is not None else None,
"track",
)
track_artists = [a.name for a in track.artists if a.name]
if not track_artists:
track_artists = as_list(source_tags.get("track_artist"))
album_artists = [a.name for a in album.artists if a.name]
if not album_artists:
album_artists = as_list(source_tags.get("album_artist"))
if not track_artists:
track_artists = album_artists
genre = None
if album.genre:
genre = album.genre
track_number = None track_number = None
disc_number = None disc_number = None
position = getattr(track, "track_position", None) or album.track_position if position := album.track_position:
if position:
track_number = position.index track_number = position.index
disc_number = position.volume disc_number = position.volume
if track_number is None and (tag_track := source_tags.get("track_number")):
try:
track_number = int(tag_track)
except (TypeError, ValueError):
track_number = None
iso8601_release_date = None iso8601_release_date = None
release_year: Optional[str] = None release_year: Optional[str] = None
if album.release_date is not None: if album.release_date is not None:
@ -351,8 +187,6 @@ def set_tags(
tag["TRCK"] = TRCK(encoding=3, text=str(track_number)) tag["TRCK"] = TRCK(encoding=3, text=str(track_number))
if disc_number: if disc_number:
tag["TPOS"] = TPOS(encoding=3, text=str(disc_number)) tag["TPOS"] = TPOS(encoding=3, text=str(disc_number))
if genre:
tag["TCON"] = TCON(encoding=3, text=genre)
if lyrics: if lyrics:
tag["USLT"] = USLT(encoding=3, text=lyrics) tag["USLT"] = USLT(encoding=3, text=lyrics)
@ -387,8 +221,6 @@ def set_tags(
tag["trkn"] = [(track_number, 0)] tag["trkn"] = [(track_number, 0)]
if disc_number: if disc_number:
tag["disk"] = [(disc_number, 0)] tag["disk"] = [(disc_number, 0)]
if genre:
tag["\xa9gen"] = genre
if lyrics: if lyrics:
tag["\xa9lyr"] = lyrics tag["\xa9lyr"] = lyrics
@ -414,8 +246,6 @@ def set_tags(
tag["tracknumber"] = str(track_number) tag["tracknumber"] = str(track_number)
if disc_number: if disc_number:
tag["discnumber"] = str(disc_number) tag["discnumber"] = str(disc_number)
if genre:
tag["genre"] = genre
if lyrics: if lyrics:
tag["lyrics"] = lyrics tag["lyrics"] = lyrics
@ -432,122 +262,6 @@ def set_tags(
tag.save() tag.save()
def extract_basic_tags(path: Path, container: Container) -> dict[str, Optional[Union[str, int]]]:
"""Read minimal tags to be used for renaming."""
title = album = album_artist = track_artist = None
track_number: Optional[int] = None
tag = CONTAINER_MUTAGEN_MAPPING[container](path)
def first_text(value) -> Optional[str]:
if value is None:
return None
if isinstance(value, list):
if not value:
return None
value = value[0]
if hasattr(value, "text"):
value = value.text
if isinstance(value, list):
value = value[0] if value else None
if isinstance(value, bytes):
try:
value = value.decode("utf-8", "ignore")
except Exception:
return None
return str(value) if value else None
if isinstance(tag, MP3):
title = first_text(tag.get("TIT2"))
album = first_text(tag.get("TALB"))
track_artist = first_text(tag.get("TPE1"))
album_artist = first_text(tag.get("TPE2"))
if trck := first_text(tag.get("TRCK")):
try:
track_number = int(str(trck).split("/")[0])
except ValueError:
track_number = None
elif isinstance(tag, MP4):
title = first_text(tag.tags.get("\xa9nam") if tag.tags else None)
album = first_text(tag.tags.get("\xa9alb") if tag.tags else None)
track_artist = first_text(tag.tags.get("\xa9ART") if tag.tags else None)
album_artist = first_text(tag.tags.get("aART") if tag.tags else None)
if trkn := tag.tags.get("trkn") if tag.tags else None:
try:
track_number = int(trkn[0][0])
except Exception:
track_number = None
elif isinstance(tag, FLAC):
def get_flac(key: str) -> Optional[str]:
values = tag.get(key)
if not values:
return None
return values[0]
title = get_flac("title")
album = get_flac("album")
artists_val = get_flac("artist")
if artists_val:
track_artist = artists_val
album_artist = get_flac("albumartist")
if trck := get_flac("tracknumber"):
try:
track_number = int(str(trck).split("/")[0])
except ValueError:
track_number = None
return {
"title": title,
"album": album,
"album_artist": album_artist,
"track_artist": track_artist,
"track_number": track_number,
}
def retarget_with_tags(
path_pattern: Path,
track: Track,
container: Container,
current_path: Path,
unsafe_path: bool,
fallback_number: Optional[int],
fallback_total: Optional[int],
override_tags: Optional[dict[str, Optional[Union[str, int]]]] = None,
) -> Path:
tags: dict[str, Optional[Union[str, int]]] = {}
if override_tags:
tags.update(override_tags)
try:
extracted = extract_basic_tags(current_path, container)
except Exception:
extracted = {}
for key, value in extracted.items():
if key not in tags or not tags[key]:
tags[key] = value
new_base = prepare_base_path(
path_pattern,
track,
unsafe_path=unsafe_path,
fallback_number=fallback_number,
fallback_total=fallback_total,
override_title=tags.get("title"),
override_album=tags.get("album"),
override_album_artist=tags.get("album_artist"),
override_track_artist=tags.get("track_artist"),
override_track_number=tags.get("track_number"),
)
new_target = new_base.with_suffix(current_path.suffix)
try:
same = new_target.resolve() == current_path.resolve()
except FileNotFoundError:
same = False
if same:
return current_path
new_target.parent.mkdir(parents=True, exist_ok=True)
current_path.replace(new_target)
return new_target
def download_track( def download_track(
track_info: DownloadableTrack, track_info: DownloadableTrack,
cover_resolution: int = DEFAULT_COVER_RESOLUTION, cover_resolution: int = DEFAULT_COVER_RESOLUTION,
@ -555,11 +269,6 @@ def download_track(
embed_cover: bool = False, embed_cover: bool = False,
covers_cache: Optional[dict[int, AlbumCover]] = None, covers_cache: Optional[dict[int, AlbumCover]] = None,
compatibility_level: int = 1, compatibility_level: int = 1,
path_pattern: Optional[Path] = None,
unsafe_path: bool = False,
fallback_number: Optional[int] = None,
fallback_total: Optional[int] = None,
rename_using_tags: bool = False,
): ):
if embed_cover and covers_cache is None: if embed_cover and covers_cache is None:
raise RuntimeError("covers_cache isn't provided") raise RuntimeError("covers_cache isn't provided")
@ -612,47 +321,19 @@ def download_track(
download_info = track_info.download_info download_info = track_info.download_info
track_data = api.download_track(client, download_info) track_data = api.download_track(client, download_info)
source_tags: dict[str, Optional[Union[str, int]]] = {} write_via_temporary_file(
track_data,
def tag_hook(tmp_path: Path) -> None: target_path,
nonlocal source_tags temporary_file_hook=lambda tmp_path: set_tags(
try:
source_tags = extract_basic_tags(tmp_path, download_info.file_format.container)
except Exception:
source_tags = {}
set_tags(
tmp_path, tmp_path,
track, track,
download_info.file_format.container, download_info.file_format.container,
text_lyrics, text_lyrics,
cover, cover,
compatibility_level, compatibility_level,
source_tags=source_tags, ),
)
write_via_temporary_file(
track_data,
target_path,
temporary_file_hook=tag_hook,
) )
if rename_using_tags and path_pattern is not None:
try:
renamed_path = retarget_with_tags(
path_pattern=path_pattern,
track=track,
container=download_info.file_format.container,
current_path=target_path,
unsafe_path=unsafe_path,
fallback_number=fallback_number,
fallback_total=fallback_total,
override_tags=source_tags if source_tags else None,
)
target_path = renamed_path
except Exception:
# Renaming is non-critical; ignore failures to keep download intact
pass
def to_downloadable_track( def to_downloadable_track(
track: Track, quality: CoreTrackQuality, base_path: Path track: Track, quality: CoreTrackQuality, base_path: Path
@ -690,9 +371,8 @@ def write_via_temporary_file(
target_path: Path, target_path: Path,
temporary_file_hook: Optional[Callable[[Path], None]] = None, temporary_file_hook: Optional[Callable[[Path], None]] = None,
) -> Path: ) -> Path:
target_name = hashlib.sha256(target_path.name.encode()).hexdigest()
temporary_file = target_path.parent / ( temporary_file = target_path.parent / (
TEMPORARY_FILE_NAME_TEMPLATE.format(target_name) TEMPORARY_FILE_NAME_TEMPLATE.format(target_path.name)
) )
try: try:
temporary_file.write_bytes(data) temporary_file.write_bytes(data)

File diff suppressed because it is too large Load diff