Compare commits

...

10 commits

10 changed files with 2105 additions and 341 deletions

17
.dockerignore Normal file
View file

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

18
.env.example Normal file
View file

@ -0,0 +1,18 @@
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,160 +1,42 @@
# Byte-compiled / optimized / DLL files
# Local environment
.env
.env.*
!.env.example
# Python
__pycache__/
*.py[cod]
*$py.class
*.pyo
*.pyd
.python-version
.venv/
venv/
env/
# C extensions
*.so
# Distribution / packaging
.Python
# Build artifacts
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
pip-wheel-metadata/
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
# Tooling caches
.mypy_cache/
.pytest_cache/
.ruff_cache/
.pyre/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
htmlcov/
# Translations
*.mo
*.pot
# IDEs and OS files
.idea/
.vscode/
.DS_Store
Thumbs.db
# Django stuff:
# Project runtime files
downloads/
*.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/

24
Dockerfile Normal file
View file

@ -0,0 +1,24 @@
# 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,145 +1,244 @@
# yandex-music-downloader
> Внимание! В версии v3 был изменен способ авторизации и некоторые
> аргументы. Смотрите [MIGRATION.md](MIGRATION.md) для получения информации
> об изменениях
Загрузчик музыки из Яндекс Музыки с двумя интерфейсами:
## Содержание
1. [О программе](#О-программе)
2. [Установка](#Установка)
3. [Получение данных для авторизации](#Получение-данных-для-авторизации)
4. [Примеры использования](#Примеры-использования)
5. [Спасибо](#Спасибо)
6. [Использование](#Использование)
7. [Уровни совместимости](#Уровни-совместимости)
8. [Дисклеймер](#Дисклеймер)
- `CLI` для точечных и пакетных загрузок
- `Web UI` на FastAPI для загрузки, экспорта дискографии и работы с лайками
## О программе
Загрузчик, созданный вследствие наличия *фатального недостатка* в проекте [yandex-music-download](https://github.com/kaimi-io/yandex-music-download).
Проект поддерживает загрузку в AAC и FLAC, сохранение тегов, обложек и текстов, а также запуск через Docker.
### Возможности
- Возможность загрузки:
- Всех треков исполнителя
- Всех треков из альбома
- Всех треков из плейлиста
- Отдельного трека
- Загрузка всех метаданных трека/альбома:
- Номер трека
- Номер диска
- Название трека
- Исполнитель
- Дополнительные исполнители
- Дата выпуска альбома
- Обложка альбома
- Название альбома
- Текст песни (при использовании флага `--add-lyrics`)
- Загрузка треков в lossless качестве
- Поддержка паттерна для пути сохранения музыки
> В `v3` изменился способ авторизации и часть аргументов. Если обновляетесь со старой версии, смотрите [MIGRATION.md](MIGRATION.md).
## Возможности
- Скачать все треки артиста, альбом, плейлист или отдельный трек
- Загружать в качестве `AAC 64`, `AAC 192` или `FLAC`
- Встраивать обложку в файл
- Сохранять тексты песен в формате `text` или `lrc`
- Настраивать путь сохранения через шаблон
- Пропускать уже скачанные треки
- Отдельно настраивать совместимость тегов
- Работать через веб-панель без ручного ввода токена
- Смотреть лайки в веб-панели
- Скачивать из веб-панели:
- все понравившиеся треки
- альбомы артистов из лайков
- дискографию артиста в JSON
## Требования
- 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-web --help
```
## Получение токена
Инструкция по получению токена:
https://yandex-music.readthedocs.io/en/main/token.html
## Быстрый старт CLI
Скачать все треки артиста в FLAC:
```bash
yandex-music-downloader \
--token "<TOKEN>" \
--quality 2 \
--url "https://music.yandex.ru/artist/208167"
```
Скачать альбом и сохранить тексты в `LRC`:
```bash
yandex-music-downloader \
--token "<TOKEN>" \
--quality 1 \
--lyrics-format lrc \
--url "https://music.yandex.ru/album/294912"
```
Скачать плейлист в свою папку:
```bash
yandex-music-downloader \
--token "<TOKEN>" \
--playlist-id "user/playlists/123" \
--dir "./downloads" \
--skip-existing
```
Скачать один трек:
```bash
yandex-music-downloader \
--token "<TOKEN>" \
--url "https://music.yandex.ru/album/11644078/track/6705392"
```
## Основные аргументы CLI
- `--url` или один из `--artist-id`, `--album-id`, `--track-id`, `--playlist-id`
- `--token` для авторизации
- `--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`
Полный список:
```bash
yandex-music-downloader --help
```
## Получение данных для авторизации
https://yandex-music.readthedocs.io/en/main/token.html
## Веб-панель
## Примеры использования
Во всех примерах замените `<Токен>` на ваш токен.
Веб-панель использует токен только из переменной окружения `YMD_TOKEN`. В интерфейсе поле токена отсутствует.
### Скачать все треки [Arctic Monkeys](https://music.yandex.ru/artist/208167) в наилучшем качестве
```
yandex-music-downloader --token "<Токен>" --quality 2 --url "https://music.yandex.ru/artist/208167"
Локальный запуск:
```bash
export YMD_TOKEN="<TOKEN>"
yandex-music-downloader-web --host 127.0.0.1 --port 8000
```
### Скачать альбом [Nevermind](https://music.yandex.ru/album/294912) в высоком качестве, загружая тексты песен в формате LRC (с временными метками)
```
yandex-music-downloader --token "<Токен>" --quality 1 --lyrics-format lrc --url "https://music.yandex.ru/album/294912"
После запуска откройте:
```text
http://127.0.0.1:8000
```
### Скачать трек [Seven Nation Army](https://music.yandex.ru/album/11644078/track/6705392)
```
yandex-music-downloader --token "<Токен>" --url "https://music.yandex.ru/album/11644078/track/6705392"
Что умеет веб-панель:
- загрузка по ссылке, `artist_id`, `album_id`, `track_id`, `playlist_id`
- экспорт дискографии артиста в JSON
- просмотр лайкнутых треков
- отдельный список артистов из лайков
- кнопка `Скачать всё понравившееся`
- кнопка `Скачать альбомы артистов`
- отображение активных задач и прогресса
## Запуск через Docker
1. Создайте `.env` из шаблона:
```bash
cp .env.example .env
```
## Использование
```
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 <Токен>
2. Укажите токен и при необходимости остальные параметры.
Загрузчик музыки с сервиса Яндекс.Музыка
3. Запустите контейнер:
options:
-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 для способов получения
```bash
docker compose up --build -d
```
## Уровни совместимости
Уровень совместимости позволяет отойти от стандарта тегов, которого
придерживается библиотека mutagen. Сделано это для поддержки большего
количества музыкальных плееров. Ниже подробно описаны все уровни.
4. Откройте веб-панель:
### 0
Стандартные теги mutagen.
```text
http://localhost:8228
```
### 1
Затрагиваемые форматы: `m4a`
Загрузка по умолчанию сохраняется в локальную папку `./downloads`.
- Теги с несколькими значениями (`\xa9ART` и `aART`) устанвливаются с
разделителем `;`. Пример: `Artist1; Artist2; Artist3`
## Переменные окружения для 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)
- @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)
- @leowerd за [идею корректных имен исполнителей](https://github.com/llistochek/yandex-music-downloader/issues/93#issuecomment-2960210879) при загрузке сборников
## Дисклеймер
Данный проект является независимой разработкой и никак не связан с компанией Яндекс.
Проект не связан с Яндексом и является независимой разработкой.

12
docker-compose.yml Normal file
View file

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

View file

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

View file

@ -1,5 +1,7 @@
import datetime as dt
import hashlib
import re
import time
import typing
from collections.abc import Callable
from dataclasses import dataclass
@ -12,6 +14,7 @@ from mutagen.flac import FLAC, Picture
from mutagen.id3._frames import (
APIC,
TALB,
TCON,
TDRC,
TIT2,
TPE1,
@ -31,6 +34,7 @@ from yandex_music import (
Track,
YandexMusicModel,
)
from yandex_music.exceptions import NetworkError
from ymd import api
from ymd.api import (
@ -44,7 +48,7 @@ from ymd.mime_utils import MimeType, guess_mime_type
UNSAFE_PATH_CLEAR_RE = re.compile(r"[/\\]+")
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
MIN_COMPATIBILITY_LEVEL = 0
@ -52,6 +56,9 @@ MAX_COMPATIBILITY_LEVEL = 1
AUDIO_FILE_SUFFIXES = {".mp3", ".flac", ".m4a"}
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):
@ -86,58 +93,180 @@ class AlbumCover:
mime_type: MimeType
def init_client(token: str, timeout: int) -> Client:
def init_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.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()
def full_title(obj: YandexMusicModel) -> str:
result = obj["title"]
if result is None:
"""Return title with optional version; be tolerant to missing keys."""
title = getattr(obj, "title", None)
if title is None:
try:
title = obj["title"]
except Exception: # noqa: BLE001
title = None
if not title:
return ""
if version := obj["version"]:
result += f" ({version})"
return result
version = getattr(obj, "version", None)
if version is None:
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(
path_pattern: Path, track: Track, unsafe_path: bool = False
path_pattern: Path,
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_str = str(path_pattern)
album = None
artist = None
album_artist = None
track_artist = None
track_position = None
if albums := track.albums:
album = albums[0]
track_position = album.track_position
if artists := album.artists:
artist = artists[0]
if artist is None and (artists := track.artists):
artist = artists[0]
album_artist = artists[0]
# позиция трека корректнее брать из самого трека
track_position = getattr(track, "track_position", None) or track_position
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]] = {
"#number-padded": str(track_position.index).zfill(len(str(album.track_count)))
if track_position and album
else None,
"#album-artist": artist.name if artist else None,
"#artist-id": artist.id if artist else None,
"#number-padded": number_padded,
"#album-artist": album_artist_name,
"#track-artist": track_artist_name,
"#artist-id": track_artist.id if track_artist else None,
"#album-id": album.id if album else None,
"#track-id": track.id,
"#number": track_position.index if track_position else None,
"#artist": artist.name if artist else None,
"#title": full_title(track),
"#album": full_title(album) if album else None,
"#number": number_value,
"#title": track_title,
"#album": album_title,
"#year": album.year if album else None,
}
for placeholder, replacement in repl_dict.items():
replacement = str(replacement)
if not unsafe_path:
clear_re = SAFE_PATH_CLEAR_RE
else:
clear_re = UNSAFE_PATH_CLEAR_RE
replacement = clear_re.sub("_", replacement)
if replacement is None:
replacement = ""
replacement = sanitize_path_component(str(replacement), unsafe_path)
if not replacement:
replacement = sanitize_path_component("untitled", unsafe_path)
path_str = path_str.replace(placeholder, replacement)
return Path(path_str)
path = 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(
@ -147,22 +276,57 @@ def set_tags(
lyrics: Optional[str],
album_cover: Optional[AlbumCover],
compatibility_level: int,
source_tags: Optional[dict[str, Optional[Union[str, int]]]] = 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)
if file_type is None:
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)
album_title = full_title(album)
track_title = full_title(track)
album = track.albums[0] if track.albums else Album()
album_title = choose_first(
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
disc_number = None
if position := album.track_position:
position = getattr(track, "track_position", None) or album.track_position
if position:
track_number = position.index
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
release_year: Optional[str] = None
if album.release_date is not None:
@ -187,6 +351,8 @@ def set_tags(
tag["TRCK"] = TRCK(encoding=3, text=str(track_number))
if disc_number:
tag["TPOS"] = TPOS(encoding=3, text=str(disc_number))
if genre:
tag["TCON"] = TCON(encoding=3, text=genre)
if lyrics:
tag["USLT"] = USLT(encoding=3, text=lyrics)
@ -221,6 +387,8 @@ def set_tags(
tag["trkn"] = [(track_number, 0)]
if disc_number:
tag["disk"] = [(disc_number, 0)]
if genre:
tag["\xa9gen"] = genre
if lyrics:
tag["\xa9lyr"] = lyrics
@ -246,6 +414,8 @@ def set_tags(
tag["tracknumber"] = str(track_number)
if disc_number:
tag["discnumber"] = str(disc_number)
if genre:
tag["genre"] = genre
if lyrics:
tag["lyrics"] = lyrics
@ -262,6 +432,122 @@ def set_tags(
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(
track_info: DownloadableTrack,
cover_resolution: int = DEFAULT_COVER_RESOLUTION,
@ -269,6 +555,11 @@ def download_track(
embed_cover: bool = False,
covers_cache: Optional[dict[int, AlbumCover]] = None,
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:
raise RuntimeError("covers_cache isn't provided")
@ -321,19 +612,47 @@ def download_track(
download_info = track_info.download_info
track_data = api.download_track(client, download_info)
write_via_temporary_file(
track_data,
target_path,
temporary_file_hook=lambda tmp_path: set_tags(
source_tags: dict[str, Optional[Union[str, int]]] = {}
def tag_hook(tmp_path: Path) -> None:
nonlocal source_tags
try:
source_tags = extract_basic_tags(tmp_path, download_info.file_format.container)
except Exception:
source_tags = {}
set_tags(
tmp_path,
track,
download_info.file_format.container,
text_lyrics,
cover,
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(
track: Track, quality: CoreTrackQuality, base_path: Path
@ -371,8 +690,9 @@ def write_via_temporary_file(
target_path: Path,
temporary_file_hook: Optional[Callable[[Path], None]] = None,
) -> Path:
target_name = hashlib.sha256(target_path.name.encode()).hexdigest()
temporary_file = target_path.parent / (
TEMPORARY_FILE_NAME_TEMPLATE.format(target_path.name)
TEMPORARY_FILE_NAME_TEMPLATE.format(target_name)
)
try:
temporary_file.write_bytes(data)

1340
ymd/webui.py Normal file

File diff suppressed because it is too large Load diff