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__/ __pycache__/
*.py[cod] *.py[cod]
*$py.class *.pyo
*.pyd
.python-version
.venv/
venv/
env/
# C extensions # Build artifacts
*.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/
.installed.cfg pip-wheel-metadata/
*.egg
MANIFEST
# PyInstaller # Tooling caches
# Usually these files are written by a python script from a template .mypy_cache/
# before PyInstaller builds the exe, so as to inject date/other infos into it. .pytest_cache/
*.manifest .ruff_cache/
*.spec .pyre/
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/ .tox/
.nox/ .nox/
.coverage .coverage
.coverage.* .coverage.*
.cache htmlcov/
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations # IDEs and OS files
*.mo .idea/
*.pot .vscode/
.DS_Store
Thumbs.db
# Django stuff: # Project runtime files
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/

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 # yandex-music-downloader
> Внимание! В версии v3 был изменен способ авторизации и некоторые Загрузчик музыки из Яндекс Музыки с двумя интерфейсами:
> аргументы. Смотрите [MIGRATION.md](MIGRATION.md) для получения информации
> об изменениях
## Содержание - `CLI` для точечных и пакетных загрузок
1. [О программе](#О-программе) - `Web UI` на FastAPI для загрузки, экспорта дискографии и работы с лайками
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`) - альбомы артистов из лайков
- Загрузка треков в lossless качестве - дискографию артиста в 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 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
``` ```
## Использование 2. Укажите токен и при необходимости остальные параметры.
```
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 <Токен>
Загрузчик музыки с сервиса Яндекс.Музыка 3. Запустите контейнер:
options: ```bash
-h, --help show this help message and exit docker compose up --build -d
Общие параметры:
--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 для способов получения
``` ```
## Уровни совместимости 4. Откройте веб-панель:
Уровень совместимости позволяет отойти от стандарта тегов, которого
придерживается библиотека mutagen. Сделано это для поддержки большего
количества музыкальных плееров. Ниже подробно описаны все уровни.
### 0 ```text
Стандартные теги mutagen. http://localhost:8228
```
### 1 Загрузка по умолчанию сохраняется в локальную папку `./downloads`.
Затрагиваемые форматы: `m4a`
- Теги с несколькими значениями (`\xa9ART` и `aART`) устанвливаются с ## Переменные окружения для Web UI / Docker
разделителем `;`. Пример: `Artist1; Artist2; Artist3`
Полный набор есть в [.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) при загрузке сборников
## Дисклеймер ## Дисклеймер
Данный проект является независимой разработкой и никак не связан с компанией Яндекс.
Проект не связан с Яндексом и является независимой разработкой.

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] [project]
name = "yandex-music-downloader" name = "yandex-music-downloader"
version = "3.4.7" version = "3.5.4"
description = "Загрузчик музыки с сервиса Яндекс.Музыка" description = "Загрузчик музыки с сервиса Яндекс.Музыка"
requires-python = ">=3.9" requires-python = ">=3.9"
readme = "README.md" readme = "README.md"
@ -15,8 +15,12 @@ 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 Generator, Iterable from collections.abc import Callable, 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,35 +34,25 @@ def show_default(text: Optional[str] = None) -> str:
return f"{text} ({default})" return f"{text} ({default})"
def quality_arg(astr: str) -> int: def checked_int_arg(
aint = int(astr) min_value: int, max_value: Optional[int] = None
if 0 <= aint <= 2: ) -> Callable[[str], int]:
return aint def func(astr: str) -> int:
raise ArgumentTypeError("Значение должно быть в промежутке от 0 до 2") 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)
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 int(astr) return checked_int_arg(100)(astr)
def lyrics_format_arg(astr: str) -> core.LyricsFormat: def lyrics_format_arg(astr: str) -> core.LyricsFormat:
@ -83,7 +73,7 @@ def main():
"--quality", "--quality",
metavar="<Качество>", metavar="<Качество>",
default=0, default=0,
type=quality_arg, type=checked_int_arg(0, 2),
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(
@ -115,7 +105,7 @@ def main():
"--delay", "--delay",
default=DEFAULT_DELAY, default=DEFAULT_DELAY,
metavar="<Задержка>", metavar="<Задержка>",
type=int, type=checked_int_arg(0),
help=show_default("Задержка между запросами, в секундах"), help=show_default("Задержка между запросами, в секундах"),
) )
common_group.add_argument( common_group.add_argument(
@ -132,20 +122,40 @@ def main():
"--compatibility-level", "--compatibility-level",
metavar="<Уровень совместимости>", metavar="<Уровень совместимости>",
default=1, default=1,
type=compatibility_level_arg, type=checked_int_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=natural_int_arg, type=checked_int_arg(1),
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")
@ -179,7 +189,7 @@ def main():
type=Path, type=Path,
help=show_default( help=show_default(
"Поддерживает следующие заполнители:" "Поддерживает следующие заполнители:"
" #number, #artist, #album-artist, #title," " #number, #track-artist, #album-artist, #title,"
" #album, #year, #artist-id, #album-id, #track-id, #number-padded" " #album, #year, #artist-id, #album-id, #track-id, #number-padded"
), ),
) )
@ -221,8 +231,13 @@ def main():
print("Параметер url указан в неверном формате") print("Параметер url указан в неверном формате")
return 1 return 1
client = core.init_client(args.token, args.timeout) client = core.init_client(
result_tracks: Iterable[Track] = [] 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]: def album_tracks_gen(album_ids: Iterable[Union[int, str]]) -> Generator[Track]:
for album_id in album_ids: for album_id in album_ids:
@ -230,6 +245,7 @@ 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:
@ -244,31 +260,46 @@ def main():
return True return True
return False return False
def albums_gen() -> Generator[Album]: def albums_id_gen() -> Generator[int]:
has_next = True has_next = True
page = 0 page = 0
while has_next: while has_next:
if albums := client.artists_direct_albums(args.artist_id, page): if albums_info := client.artists_direct_albums(args.artist_id, page):
yield from albums.albums 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: else:
break break
if pager := albums.pager: if pager := albums_info.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( result_tracks = album_tracks_gen(albums_id_gen())
a.id for a in albums_gen() if filter_album(a) and a.id is not None artist = client.artists(args.artist_id)[0]
) 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()
@ -278,17 +309,29 @@ 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"Трек {track.title} не доступен для скачивания") print(f"{progress_status}Трек {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(
@ -306,7 +349,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"{format_info} Загружается {downloadable.path}") print(f"{progress_status}{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,
@ -314,6 +357,11 @@ 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,5 +1,7 @@
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
@ -12,6 +14,7 @@ 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,
@ -31,6 +34,7 @@ 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 (
@ -44,7 +48,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
@ -52,6 +56,9 @@ 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):
@ -86,58 +93,180 @@ class AlbumCover:
mime_type: MimeType 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 = 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:
result = obj["title"] """Return title with optional version; be tolerant to missing keys."""
if result is None: title = getattr(obj, "title", None)
if title is None:
try:
title = obj["title"]
except Exception: # noqa: BLE001
title = None
if not title:
return "" return ""
if version := obj["version"]:
result += f" ({version})" version = getattr(obj, "version", None)
return result 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( 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:
path_str = str(path_pattern) path_str = str(path_pattern)
album = None album = None
artist = None album_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:
artist = artists[0] album_artist = artists[0]
if artist is None and (artists := track.artists): # позиция трека корректнее брать из самого трека
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]] = { repl_dict: dict[str, Union[str, int, None]] = {
"#number-padded": str(track_position.index).zfill(len(str(album.track_count))) "#number-padded": number_padded,
if track_position and album "#album-artist": album_artist_name,
else None, "#track-artist": track_artist_name,
"#album-artist": artist.name if artist else None, "#artist-id": track_artist.id if track_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": track_position.index if track_position else None, "#number": number_value,
"#artist": artist.name if artist else None, "#title": track_title,
"#title": full_title(track), "#album": album_title,
"#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():
replacement = str(replacement) if replacement is None:
if not unsafe_path: replacement = ""
clear_re = SAFE_PATH_CLEAR_RE replacement = sanitize_path_component(str(replacement), unsafe_path)
else: if not replacement:
clear_re = UNSAFE_PATH_CLEAR_RE replacement = sanitize_path_component("untitled", unsafe_path)
replacement = clear_re.sub("_", replacement)
path_str = path_str.replace(placeholder, replacement) 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( def set_tags(
@ -147,22 +276,57 @@ 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_title = full_title(album) album = track.albums[0] if track.albums else Album()
track_title = full_title(track) 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 track_number = None
disc_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 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:
@ -187,6 +351,8 @@ 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)
@ -221,6 +387,8 @@ 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
@ -246,6 +414,8 @@ 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
@ -262,6 +432,122 @@ 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,
@ -269,6 +555,11 @@ 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")
@ -321,19 +612,47 @@ 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)
write_via_temporary_file( source_tags: dict[str, Optional[Union[str, int]]] = {}
track_data,
target_path, def tag_hook(tmp_path: Path) -> None:
temporary_file_hook=lambda tmp_path: set_tags( nonlocal source_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
@ -371,8 +690,9 @@ 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_path.name) TEMPORARY_FILE_NAME_TEMPLATE.format(target_name)
) )
try: try:
temporary_file.write_bytes(data) temporary_file.write_bytes(data)

1340
ymd/webui.py Normal file

File diff suppressed because it is too large Load diff