commit 353be65e3af8b5ff48304b9cf20231d4631ac867 Author: Andrew Pimenov Date: Mon Jul 6 09:49:36 2026 +0300 init: сортировка лайков Яндекс.Музыки по жанру и языку CLI на yandex-music (device-flow auth, fetch/report/apply), жанр через реальное дерево client.genres(), язык гибридом (track_supplement + алфавит). diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0267313 --- /dev/null +++ b/.gitignore @@ -0,0 +1,8 @@ +.venv/ +__pycache__/ +*.pyc +.token +cache/ +out/ +.claude/ +plans/ diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md new file mode 100644 index 0000000..4d6fbc3 --- /dev/null +++ b/ARCHITECTURE.md @@ -0,0 +1,75 @@ +# Архитектура + +Консольная утилита на Python, которая читает лайкнутые треки Яндекс.Музыки и раскладывает их +по плейлистам вида `Жанр — RU` / `Жанр — INT`, используя неофициальную библиотеку `yandex-music`. + +## Структура + +``` +sort_ym/ + config.py — чтение config.toml (throttle, пути к кэшу/токену) + auth.py — OAuth device-flow, хранение токена в .token + ymclient.py — создание Client, батч-нарезка списков + fetch.py — лайки -> полные Track -> cache/tracks.json (докачивается инкрементально) + genres.py — реальное дерево жанров Яндекса -> 11 крупных корзин (ROOT_BUCKET) + language.py — гибридная детекция языка (API текста / жанр-подсказка / алфавит) + classify.py — (корзина, язык) -> имя плейлиста + report.py — сборка строк отчёта, запись out/report.csv, сводки + apply.py — создание плейлистов + идемпотентное добавление треков + cli.py — команды auth / fetch / report / apply + +tests/ + test_genres.py, test_language.py — тесты чистых функций (маппинг, детекция языка) + +cache/ — кэш API (в .gitignore) +out/ — отчёты (в .gitignore) +``` + +## Поток данных + +``` +auth → .token +fetch → users_likes_tracks() → client.tracks(batch) → cache/tracks.json +report → genres.load_or_fetch_catalog() → cache/genre_catalog.json + → language.fetch_api_languages() → cache/lyrics_lang.json + → build_rows() → out/report.csv (dry-run, аккаунт не меняется) +apply --yes → users_playlists_create / users_playlists_insert_track +``` + +## Ключевые решения + +- **Жанр трека** доступен только через альбом (`Track.albums[0].genre`), у самого трека поля + жанра нет. Раскладка на 11 крупных групп сделана через реальное дерево `client.genres()` + (родитель → под-жанры), а не через угаданный список слагов — так таблица `ROOT_BUCKET` в + `genres.py` опирается на подтверждённые id, а не на предположения. Жанры без совпадения падают + в корзину `other` и печатаются в выводе `report` для ручного дополнения таблицы. + +- **Язык** определяется гибридно, в порядке приоритета: + 1. `client.track_supplement(track_id).lyrics.text_language` — единственное поле в библиотеке + с реальным языком текста песни. Оно есть только в устаревшем (deprecated) методе + `track_supplement`, а не в новом `tracks_lyrics()` — тот текст возвращает, но язык не отдаёт. + Используется сознательно, несмотря на пометку deprecated в самой библиотеке. + 2. Подсказка по слагу жанра (`rusrap`, `shanson`, `bard`, ...). + 3. Эвристика по алфавиту названия/исполнителя (кириллица vs латиница, `unicodedata`). + 4. Иначе — `UNKNOWN` → отдельный плейлист «Не определено». + +- **Идентификаторы треков**: у трека есть composite `track_id` (`"id:album_id"`, используется как + ключ кэша и для сверки с уже добавленными треками в плейлисте) и отдельные `id`/`album_id` + (нужны как раздельные параметры для `track_supplement`, `tracks_lyrics` и + `users_playlists_insert_track`). + +- **Идемпотентность `apply`**: перед добавлением треков в плейлист читается его текущий состав + (`users_playlists(kind)`), уже присутствующие треки пропускаются. Повторный запуск `apply` + безопасен и не создаёт дублей. Лайки не трогаются, ничего не удаляется. + +- **Троттлинг и резюмируемость**: все сетевые циклы (загрузка треков, языков) делают паузу между + запросами (`config.toml`, `[throttle]`) и пишут кэш атомарно (temp-файл + rename) после каждого + батча — прерванный прогон можно просто перезапустить, уже полученное не запрашивается повторно. + +## Хранилище + +- `.token` — OAuth-токен аккаунта (не коммитится). +- `cache/tracks.json` — метаданные лайкнутых треков. +- `cache/genre_catalog.json` — плоский снимок дерева жанров Яндекса (id -> title, root_id). +- `cache/lyrics_lang.json` — язык текста песни по треку (id -> language code | null). +- `out/report.csv` — предпросмотр распределения по плейлистам перед `apply`. diff --git a/README.md b/README.md new file mode 100644 index 0000000..185db1e --- /dev/null +++ b/README.md @@ -0,0 +1,85 @@ +# Sort YM + +Раскладывает лайкнутые треки Яндекс.Музыки по плейлистам вида `Жанр — RU` / `Жанр — INT`. + +``` +$ python -m sort_ym report + +Распределение по плейлистам: + Рок — INT: 143 + Рок — RU: 287 + Рэп — RU: 512 + ... + +Отчёт сохранён: out/report.csv +``` + +## Установка + +Требования: Python 3.11+ (используется `tomllib` из стандартной библиотеки), Windows/Linux/macOS, доступ в интернет. + +``` +python -m venv .venv +.venv\Scripts\pip install -r requirements.txt +``` + +Для разработки (тесты): + +``` +.venv\Scripts\pip install -r requirements-dev.txt +``` + +## Что делает + +1. **`auth`** — вход в аккаунт Яндекса через OAuth device-flow: покажет код, вы подтверждаете + вход на странице Яндекса, токен сохраняется в `.token` (локально, в `.gitignore`). +2. **`fetch`** — скачивает список лайкнутых треков и их метаданные (жанр альбома, доступность + текста) в `cache/tracks.json`. Повторный запуск докачивает только новые треки. +3. **`report`** — определяет жанр (крупная группа) и язык (RU/иностранный) каждого трека, + считает распределение по будущим плейлистам и сохраняет `out/report.csv` для проверки. + **Ничего не меняет в аккаунте.** +4. **`apply --yes`** — создаёт недостающие плейлисты и добавляет в них треки. Уже добавленные + треки не дублируются при повторном запуске. Лайки не трогает, ничего не удаляет. + +## Как определяется жанр и язык + +- **Жанр**: у трека берётся жанр альбома, затем он сопоставляется с одной из ~11 крупных + групп (Поп, Рок, Метал, Рэп, Электроника, Джаз и блюз, Классика, Фолк и world, Инди и + альтернатива, Саундтреки, Разное) через реальное дерево жанров Яндекса + (`sort_ym/genres.py`, таблица `ROOT_BUCKET` — легко дополняется). +- **Язык** — гибрид: + 1. Если у трека есть текст песни — реальный язык из API (`ru` → RU, иначе → INT). + 2. Иначе — подсказка по жанру (`rusrap`, `shanson`, `bard`, ...). + 3. Иначе — по алфавиту названия/исполнителя (кириллица/латиница). + 4. Если ничего не сработало — трек попадает в `... — Не определено`. + +После `report` в консоли также печатается список нераспознанных жанров (попавших в «Разное»), +чтобы можно было дополнить таблицу в `sort_ym/genres.py`. + +## Handy things + +- Первый прогон `apply` делайте с `--limit 20 --yes` и проверьте результат в приложении, прежде + чем запускать на все ~2000 треков. +- Все промежуточные данные лежат в `cache/` — их можно удалить, чтобы пересчитать с нуля. +- Настройки задержек между запросами — в `config.toml` (`[throttle]`). + +## Риски (неофициальный API) + +Официального API для личной библиотеки Яндекс.Музыки нет — инструмент использует +неофициальную библиотеку `yandex-music`. OAuth-токен в `.token` даёт полный доступ к аккаунту +Яндекса — храните его локально и не публикуйте. Инструмент выполняет только безопасные действия +(чтение лайков, создание плейлистов, добавление треков) и ничего не удаляет. + +## For developers + +Python 3.11+, `yandex-music==3.0.0`, стандартная библиотека (`tomllib`, `csv`, `unicodedata`). + +``` +.venv\Scripts\python -m pytest tests/ -v +.venv\Scripts\python -m sort_ym auth +.venv\Scripts\python -m sort_ym fetch +.venv\Scripts\python -m sort_ym report +.venv\Scripts\python -m sort_ym apply --limit 20 --yes +``` + +Структура кода и поток данных — в [ARCHITECTURE.md](ARCHITECTURE.md). diff --git a/config.toml b/config.toml new file mode 100644 index 0000000..050819b --- /dev/null +++ b/config.toml @@ -0,0 +1,14 @@ +[throttle] +# Пауза между запросами к API, секунды +fetch_batch_delay = 0.4 +lyrics_request_delay = 0.25 +apply_request_delay = 0.3 + +[fetch] +# Сколько id треков в одном батч-запросе client.tracks(...) +batch_size = 100 + +[paths] +token_file = ".token" +cache_dir = "cache" +out_dir = "out" diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..152b054 --- /dev/null +++ b/requirements-dev.txt @@ -0,0 +1,2 @@ +-r requirements.txt +pytest==8.3.4 diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..631f100 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +yandex-music==3.0.0 diff --git a/sort_ym/__init__.py b/sort_ym/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/sort_ym/__main__.py b/sort_ym/__main__.py new file mode 100644 index 0000000..9ae637f --- /dev/null +++ b/sort_ym/__main__.py @@ -0,0 +1,4 @@ +from .cli import main + +if __name__ == "__main__": + main() diff --git a/sort_ym/apply.py b/sort_ym/apply.py new file mode 100644 index 0000000..c207d6a --- /dev/null +++ b/sort_ym/apply.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +import time + +from yandex_music import Client, Playlist + + +def _existing_playlists_by_title(client: Client) -> dict[str, Playlist]: + playlists = client.users_playlists_list() + return {p.title: p for p in playlists if p.title} + + +def _existing_track_ids(client: Client, playlist: Playlist) -> set[str]: + full = client.users_playlists(playlist.kind) + if full is None: + return set() + return {t.track_id for t in full.tracks} + + +def _get_or_create_playlist(client: Client, title: str, existing: dict[str, Playlist]) -> Playlist: + if title in existing: + return existing[title] + playlist = client.users_playlists_create(title, visibility="private") + existing[title] = playlist + return playlist + + +def apply_classification( + client: Client, + rows: list[dict], + delay: float, + limit: int | None = None, +) -> None: + if limit is not None: + rows = rows[:limit] + + existing_playlists = _existing_playlists_by_title(client) + + by_playlist: dict[str, list[dict]] = {} + for row in rows: + by_playlist.setdefault(row["target_playlist"], []).append(row) + + for title, playlist_rows in by_playlist.items(): + playlist = _get_or_create_playlist(client, title, existing_playlists) + already = _existing_track_ids(client, playlist) + revision = playlist.revision or 1 + + added = 0 + for row in playlist_rows: + track_key = f"{row['id']}:{row['album_id']}" if row["album_id"] else str(row["id"]) + if track_key in already: + continue + + updated = client.users_playlists_insert_track( + playlist.kind, + row["id"], + row["album_id"], + revision=revision, + ) + if updated is not None and updated.revision is not None: + revision = updated.revision + already.add(track_key) + added += 1 + time.sleep(delay) + + skipped = len(playlist_rows) - added + print(f"{title}: добавлено {added}, уже было {skipped}") diff --git a/sort_ym/auth.py b/sort_ym/auth.py new file mode 100644 index 0000000..0bb314f --- /dev/null +++ b/sort_ym/auth.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from pathlib import Path + +from yandex_music import Client, DeviceCode +from yandex_music.exceptions import DeviceAuthError + + +def load_token(token_file: Path) -> str | None: + if token_file.exists(): + token = token_file.read_text(encoding="utf-8").strip() + return token or None + return None + + +def save_token(token_file: Path, token: str) -> None: + token_file.parent.mkdir(parents=True, exist_ok=True) + token_file.write_text(token, encoding="utf-8") + + +def _on_code(code: DeviceCode) -> None: + print(f"Откройте {code.verification_url} и введите код: {code.user_code}") + print("Ожидание подтверждения...") + + +def device_login(token_file: Path) -> str: + client = Client() + try: + oauth_token = client.device_auth(_on_code) + except DeviceAuthError as e: + raise SystemExit(f"Не удалось авторизоваться: {e}") from e + + save_token(token_file, oauth_token.access_token) + print("Токен сохранён.") + return oauth_token.access_token + + +def get_token(token_file: Path) -> str: + token = load_token(token_file) + if token: + return token + return device_login(token_file) diff --git a/sort_ym/classify.py b/sort_ym/classify.py new file mode 100644 index 0000000..f74a02e --- /dev/null +++ b/sort_ym/classify.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +BUCKET_LABELS: dict[str, str] = { + "pop": "Поп", + "rock": "Рок", + "metal": "Метал", + "rap": "Рэп", + "electronic": "Электроника", + "jazz-blues": "Джаз и блюз", + "classical": "Классика", + "folk-world": "Фолк и world", + "indie-alt": "Инди и альтернатива", + "soundtrack": "Саундтреки", + "other": "Разное", +} + +LANG_LABELS: dict[str, str] = { + "RU": "RU", + "INT": "INT", + "UNKNOWN": "Не определено", +} + + +def playlist_name(bucket: str, lang: str) -> str: + bucket_label = BUCKET_LABELS.get(bucket, BUCKET_LABELS["other"]) + lang_label = LANG_LABELS.get(lang, LANG_LABELS["UNKNOWN"]) + return f"{bucket_label} — {lang_label}" diff --git a/sort_ym/cli.py b/sort_ym/cli.py new file mode 100644 index 0000000..576035f --- /dev/null +++ b/sort_ym/cli.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import argparse + +from . import apply as apply_mod +from . import auth, fetch, genres, language, report +from .config import load_config +from .ymclient import make_client + + +def cmd_auth(args: argparse.Namespace) -> None: + cfg = load_config() + auth.device_login(cfg.token_file) + + +def cmd_fetch(args: argparse.Namespace) -> None: + cfg = load_config() + client = make_client(auth.get_token(cfg.token_file)) + fetch.fetch_liked_tracks(client, cfg.cache_dir, cfg.batch_size, cfg.fetch_batch_delay) + + +def cmd_report(args: argparse.Namespace) -> None: + cfg = load_config() + client = make_client(auth.get_token(cfg.token_file)) + + tracks_cache = fetch.load_tracks_cache(cfg.cache_dir) + if not tracks_cache: + raise SystemExit("Кэш треков пуст. Сначала запустите: python -m sort_ym fetch") + + catalog = genres.load_or_fetch_catalog(client, cfg.cache_dir) + + ids_needing_lang = [str(t["id"]) for t in tracks_cache.values() if t["lyrics_available"]] + lang_cache = language.fetch_api_languages(client, ids_needing_lang, cfg.cache_dir, cfg.lyrics_request_delay) + + rows = report.build_rows(tracks_cache, lang_cache, catalog) + out_file = report.write_report(rows, cfg.out_dir) + + print("\nРаспределение по плейлистам:") + for name, n in sorted(report.summarize(rows).items()): + print(f" {name}: {n}") + + other = report.other_breakdown(rows) + if other: + print("\nНе распознанные жанры (попали в 'Разное'), можно дополнить sort_ym/genres.py:") + for genre_raw, n in sorted(other.items(), key=lambda kv: -kv[1]): + print(f" {genre_raw}: {n}") + + print(f"\nОтчёт сохранён: {out_file}") + + +def cmd_apply(args: argparse.Namespace) -> None: + if not args.yes: + raise SystemExit("Это изменит ваш аккаунт. Добавьте флаг --yes для подтверждения.") + + cfg = load_config() + client = make_client(auth.get_token(cfg.token_file)) + + tracks_cache = fetch.load_tracks_cache(cfg.cache_dir) + if not tracks_cache: + raise SystemExit("Кэш треков пуст. Сначала запустите: python -m sort_ym fetch") + + catalog = genres.load_or_fetch_catalog(client, cfg.cache_dir) + lang_cache = language.load_lang_cache(cfg.cache_dir) + rows = report.build_rows(tracks_cache, lang_cache, catalog) + + apply_mod.apply_classification(client, rows, cfg.apply_request_delay, limit=args.limit) + + +def main(argv: list[str] | None = None) -> None: + parser = argparse.ArgumentParser(prog="sort_ym", description="Сортировка лайкнутых треков Яндекс.Музыки по жанру и языку") + sub = parser.add_subparsers(dest="command", required=True) + + sub.add_parser("auth", help="войти через device-flow и сохранить токен") + sub.add_parser("fetch", help="загрузить лайкнутые треки в локальный кэш") + sub.add_parser("report", help="посчитать распределение по плейлистам и сохранить out/report.csv (ничего не меняет в аккаунте)") + + p_apply = sub.add_parser("apply", help="создать плейлисты и добавить треки") + p_apply.add_argument("--yes", action="store_true", help="подтвердить внесение изменений в аккаунт") + p_apply.add_argument("--limit", type=int, default=None, help="ограничить количество треков (для пробного запуска)") + + args = parser.parse_args(argv) + + commands = { + "auth": cmd_auth, + "fetch": cmd_fetch, + "report": cmd_report, + "apply": cmd_apply, + } + commands[args.command](args) + + +if __name__ == "__main__": + main() diff --git a/sort_ym/config.py b/sort_ym/config.py new file mode 100644 index 0000000..d046d94 --- /dev/null +++ b/sort_ym/config.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import tomllib +from dataclasses import dataclass +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent + + +@dataclass(frozen=True) +class Config: + fetch_batch_delay: float + lyrics_request_delay: float + apply_request_delay: float + batch_size: int + token_file: Path + cache_dir: Path + out_dir: Path + + +def load_config(path: Path | None = None) -> Config: + path = path or ROOT / "config.toml" + with path.open("rb") as f: + data = tomllib.load(f) + + return Config( + fetch_batch_delay=data["throttle"]["fetch_batch_delay"], + lyrics_request_delay=data["throttle"]["lyrics_request_delay"], + apply_request_delay=data["throttle"]["apply_request_delay"], + batch_size=data["fetch"]["batch_size"], + token_file=ROOT / data["paths"]["token_file"], + cache_dir=ROOT / data["paths"]["cache_dir"], + out_dir=ROOT / data["paths"]["out_dir"], + ) diff --git a/sort_ym/fetch.py b/sort_ym/fetch.py new file mode 100644 index 0000000..a2e8208 --- /dev/null +++ b/sort_ym/fetch.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +import json +import time +from pathlib import Path + +from yandex_music import Client, Track + +from .ymclient import chunked + +TRACKS_CACHE_FILE = "tracks.json" + + +def _track_genre(track: Track) -> str | None: + for album in track.albums: + if album.genre: + return album.genre + return None + + +def _serialize(track: Track) -> dict: + album_id = track.albums[0].id if track.albums else None + return { + "id": track.id, + "album_id": album_id, + "title": track.title or "", + "artists": [a.name for a in track.artists if a.name], + "genre_raw": _track_genre(track), + "lyrics_available": bool(track.lyrics_available), + } + + +def _atomic_write_json(path: Path, data: dict) -> None: + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + tmp.replace(path) + + +def load_tracks_cache(cache_dir: Path) -> dict[str, dict]: + cache_file = cache_dir / TRACKS_CACHE_FILE + if cache_file.exists(): + return json.loads(cache_file.read_text(encoding="utf-8")) + return {} + + +def fetch_liked_tracks( + client: Client, + cache_dir: Path, + batch_size: int, + batch_delay: float, +) -> dict[str, dict]: + cache_dir.mkdir(parents=True, exist_ok=True) + cache_file = cache_dir / TRACKS_CACHE_FILE + cache = load_tracks_cache(cache_dir) + + likes = client.users_likes_tracks() + if likes is None: + print("Не удалось получить список лайкнутых треков.") + return cache + + all_ids = likes.tracks_ids + missing_ids = [tid for tid in all_ids if tid not in cache] + + print(f"Лайкнутых треков: {len(all_ids)}, новых для загрузки: {len(missing_ids)}") + + for batch in chunked(missing_ids, batch_size): + tracks = client.tracks(batch) + for track in tracks: + cache[track.track_id] = _serialize(track) + _atomic_write_json(cache_file, cache) + print(f" загружено {len(cache)}/{len(all_ids)}") + time.sleep(batch_delay) + + stale = set(cache) - set(all_ids) + for tid in stale: + del cache[tid] + if stale: + _atomic_write_json(cache_file, cache) + print(f" убрано из кэша (больше не лайкнуто): {len(stale)}") + + return cache diff --git a/sort_ym/genres.py b/sort_ym/genres.py new file mode 100644 index 0000000..af40db4 --- /dev/null +++ b/sort_ym/genres.py @@ -0,0 +1,97 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from yandex_music import Client, Genre + +GENRE_CATALOG_FILE = "genre_catalog.json" + +# Корневые жанры каталога Яндекс.Музыки -> наши крупные корзины. +# Список получен из реального дерева client.genres() и может дополняться: +# если жанр трека не находится ни в одном известном корне, он попадает в "other" — +# смотрите разбивку "other" в выводе команды report, чтобы дополнить таблицу. +ROOT_BUCKET: dict[str, str] = { + "pop": "pop", + "ruspop": "pop", + "estrada": "pop", + "rnb": "pop", + "disco": "pop", + "rock": "rock", + "rusrock": "rock", + "hardrock": "rock", + "prog": "rock", + "alternative": "indie-alt", + "indie": "indie-alt", + "punk": "indie-alt", + "postrock": "indie-alt", + "metal": "metal", + "extreme": "metal", + "industrial": "metal", + "rap": "rap", + "rusrap": "rap", + "hip": "rap", + "hiphop": "rap", + "trap": "rap", + "electronics": "electronic", + "dance": "electronic", + "house": "electronic", + "techno": "electronic", + "trance": "electronic", + "dnb": "electronic", + "dubstep": "electronic", + "jazz": "jazz-blues", + "blues": "jazz-blues", + "soul": "jazz-blues", + "funk": "jazz-blues", + "classicalmusic": "classical", + "classical": "classical", + "opera": "classical", + "folk": "folk-world", + "bard": "folk-world", + "author": "folk-world", + "avtorskaya": "folk-world", + "shanson": "folk-world", + "world": "folk-world", + "reggae": "folk-world", + "country": "folk-world", + "films": "soundtrack", + "soundtrack": "soundtrack", + "musical": "soundtrack", + "cartoon": "soundtrack", +} + + +def flatten_catalog(roots: list[Genre]) -> dict[str, dict]: + """id жанра/поджанра -> {title, root_id}.""" + flat: dict[str, dict] = {} + + def walk(genre: Genre, root_id: str) -> None: + flat[genre.id] = {"title": genre.title, "root_id": root_id} + for sub in genre.sub_genres: + walk(sub, root_id) + + for root in roots: + walk(root, root.id) + + return flat + + +def load_or_fetch_catalog(client: Client, cache_dir: Path) -> dict[str, dict]: + cache_dir.mkdir(parents=True, exist_ok=True) + cache_file = cache_dir / GENRE_CATALOG_FILE + if cache_file.exists(): + return json.loads(cache_file.read_text(encoding="utf-8")) + + roots = client.genres() + flat = flatten_catalog(roots) + cache_file.write_text(json.dumps(flat, ensure_ascii=False, indent=2), encoding="utf-8") + return flat + + +def bucket_for(genre_raw: str | None, catalog: dict[str, dict]) -> str: + if not genre_raw: + return "other" + entry = catalog.get(genre_raw) + root_id = entry["root_id"] if entry else genre_raw + return ROOT_BUCKET.get(root_id, "other") diff --git a/sort_ym/language.py b/sort_ym/language.py new file mode 100644 index 0000000..2bb5a98 --- /dev/null +++ b/sort_ym/language.py @@ -0,0 +1,133 @@ +from __future__ import annotations + +import json +import time +import unicodedata +from pathlib import Path + +from yandex_music import Client +from yandex_music.exceptions import YandexMusicError + +LYRICS_LANG_CACHE_FILE = "lyrics_lang.json" + +RU_GENRE_HINTS = { + "rusrap", + "ruspop", + "rusrock", + "shanson", + "bard", + "author", + "avtorskaya", + "estrada", +} + + +def _script_counts(text: str) -> tuple[int, int]: + cyrillic = 0 + latin = 0 + for ch in text: + if not ch.isalpha(): + continue + try: + name = unicodedata.name(ch) + except ValueError: + continue + if "CYRILLIC" in name: + cyrillic += 1 + elif "LATIN" in name: + latin += 1 + return cyrillic, latin + + +def alphabet_heuristic(title: str, artists: list[str]) -> str | None: + text = " ".join([title, *artists]) + cyrillic, latin = _script_counts(text) + if cyrillic == 0 and latin == 0: + return None + if cyrillic > latin: + return "RU" + if latin > cyrillic: + return "INT" + return None + + +def genre_hint(genre_raw: str | None) -> str | None: + if genre_raw and genre_raw.lower() in RU_GENRE_HINTS: + return "RU" + return None + + +def normalize_api_language(text_language: str | None) -> str | None: + if not text_language: + return None + return "RU" if text_language.strip().lower() == "ru" else "INT" + + +def detect_language( + title: str, + artists: list[str], + genre_raw: str | None, + api_language: str | None, +) -> str: + lang = normalize_api_language(api_language) + if lang: + return lang + + lang = genre_hint(genre_raw) + if lang: + return lang + + lang = alphabet_heuristic(title, artists) + if lang: + return lang + + return "UNKNOWN" + + +def _atomic_write_json(path: Path, data: dict) -> None: + tmp = path.with_suffix(".tmp") + tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") + tmp.replace(path) + + +def load_lang_cache(cache_dir: Path) -> dict[str, str | None]: + cache_file = cache_dir / LYRICS_LANG_CACHE_FILE + if cache_file.exists(): + return json.loads(cache_file.read_text(encoding="utf-8")) + return {} + + +def _fetch_one_language(client: Client, track_id: str) -> str | None: + try: + supplement = client.track_supplement(track_id) + except YandexMusicError: + return None + if supplement is None or supplement.lyrics is None: + return None + return supplement.lyrics.text_language + + +def fetch_api_languages( + client: Client, + track_ids: list[str], + cache_dir: Path, + delay: float, +) -> dict[str, str | None]: + cache_dir.mkdir(parents=True, exist_ok=True) + cache_file = cache_dir / LYRICS_LANG_CACHE_FILE + cache = load_lang_cache(cache_dir) + + missing = [tid for tid in track_ids if tid not in cache] + if not missing: + return cache + + print(f"Запрос языка текста песни для {len(missing)} треков...") + for i, tid in enumerate(missing, 1): + cache[tid] = _fetch_one_language(client, tid) + if i % 20 == 0: + _atomic_write_json(cache_file, cache) + print(f" обработано {i}/{len(missing)}") + time.sleep(delay) + + _atomic_write_json(cache_file, cache) + return cache diff --git a/sort_ym/report.py b/sort_ym/report.py new file mode 100644 index 0000000..bf0fd43 --- /dev/null +++ b/sort_ym/report.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +import csv +from pathlib import Path + +from . import classify, genres, language + +FIELDNAMES = ["title", "artists", "genre_raw", "bucket", "lang", "target_playlist", "id", "album_id"] + + +def build_rows( + tracks_cache: dict[str, dict], + lang_cache: dict[str, str | None], + genre_catalog: dict[str, dict], +) -> list[dict]: + rows = [] + for t in tracks_cache.values(): + bucket = genres.bucket_for(t["genre_raw"], genre_catalog) + api_lang = lang_cache.get(str(t["id"])) + lang = language.detect_language(t["title"], t["artists"], t["genre_raw"], api_lang) + playlist = classify.playlist_name(bucket, lang) + rows.append( + { + "title": t["title"], + "artists": ", ".join(t["artists"]), + "genre_raw": t["genre_raw"] or "", + "bucket": bucket, + "lang": lang, + "target_playlist": playlist, + "id": t["id"], + "album_id": t["album_id"], + } + ) + rows.sort(key=lambda r: (r["target_playlist"], r["artists"], r["title"])) + return rows + + +def write_report(rows: list[dict], out_dir: Path) -> Path: + out_dir.mkdir(parents=True, exist_ok=True) + out_file = out_dir / "report.csv" + with out_file.open("w", encoding="utf-8-sig", newline="") as f: + writer = csv.DictWriter(f, fieldnames=FIELDNAMES) + writer.writeheader() + writer.writerows(rows) + return out_file + + +def summarize(rows: list[dict]) -> dict[str, int]: + counts: dict[str, int] = {} + for r in rows: + counts[r["target_playlist"]] = counts.get(r["target_playlist"], 0) + 1 + return counts + + +def other_breakdown(rows: list[dict]) -> dict[str, int]: + counts: dict[str, int] = {} + for r in rows: + if r["bucket"] == "other" and r["genre_raw"]: + counts[r["genre_raw"]] = counts.get(r["genre_raw"], 0) + 1 + return counts diff --git a/sort_ym/ymclient.py b/sort_ym/ymclient.py new file mode 100644 index 0000000..17b376b --- /dev/null +++ b/sort_ym/ymclient.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from typing import Iterator, List, TypeVar + +from yandex_music import Client + +T = TypeVar("T") + + +def make_client(token: str) -> Client: + client = Client(token=token) + client.init() + return client + + +def chunked(items: List[T], size: int) -> Iterator[List[T]]: + for i in range(0, len(items), size): + yield items[i : i + size] diff --git a/tests/test_genres.py b/tests/test_genres.py new file mode 100644 index 0000000..d5dfc07 --- /dev/null +++ b/tests/test_genres.py @@ -0,0 +1,46 @@ +import pytest + +from sort_ym import genres + + +@pytest.fixture +def catalog(): + return { + "rock": {"title": "Рок", "root_id": "rock"}, + "rusrock": {"title": "Русский рок", "root_id": "rock"}, + "postrock": {"title": "Пост-рок", "root_id": "postrock"}, + "rap": {"title": "Рэп", "root_id": "rap"}, + "trap": {"title": "Трэп", "root_id": "rap"}, + } + + +@pytest.mark.parametrize( + "genre_raw,expected", + [ + ("rock", "rock"), + ("rusrock", "rock"), + ("postrock", "indie-alt"), + ("rap", "rap"), + ("trap", "rap"), + (None, "other"), + ("", "other"), + ("totally-unknown-slug", "other"), + ], +) +def test_bucket_for(catalog, genre_raw, expected): + assert genres.bucket_for(genre_raw, catalog) == expected + + +def test_flatten_catalog_walks_sub_genres(): + class FakeGenre: + def __init__(self, id_, title, sub_genres=None): + self.id = id_ + self.title = title + self.sub_genres = sub_genres or [] + + root = FakeGenre("rock", "Рок", [FakeGenre("rusrock", "Русский рок")]) + + flat = genres.flatten_catalog([root]) + + assert flat["rock"] == {"title": "Рок", "root_id": "rock"} + assert flat["rusrock"] == {"title": "Русский рок", "root_id": "rock"} diff --git a/tests/test_language.py b/tests/test_language.py new file mode 100644 index 0000000..aee357a --- /dev/null +++ b/tests/test_language.py @@ -0,0 +1,85 @@ +import pytest + +from sort_ym import language + + +@pytest.mark.parametrize( + "text_language,expected", + [ + ("ru", "RU"), + ("RU", "RU"), + ("en", "INT"), + ("de", "INT"), + (None, None), + ("", None), + ], +) +def test_normalize_api_language(text_language, expected): + assert language.normalize_api_language(text_language) == expected + + +@pytest.mark.parametrize( + "genre_raw,expected", + [ + ("rusrap", "RU"), + ("shanson", "RU"), + ("rock", None), + (None, None), + ], +) +def test_genre_hint(genre_raw, expected): + assert language.genre_hint(genre_raw) == expected + + +@pytest.mark.parametrize( + "title,artists,expected", + [ + ("Кино не для всех", ["Земфира"], "RU"), + ("Bohemian Rhapsody", ["Queen"], "INT"), + ("AB", ["АБ"], None), # ровно поровну — не решаем + ("123", [], None), + ("", [], None), + ], +) +def test_alphabet_heuristic(title, artists, expected): + assert language.alphabet_heuristic(title, artists) == expected + + +def test_detect_language_prefers_api_over_everything(): + result = language.detect_language( + title="Bohemian Rhapsody", + artists=["Queen"], + genre_raw="rusrap", + api_language="ru", + ) + assert result == "RU" + + +def test_detect_language_falls_back_to_genre_hint(): + result = language.detect_language( + title="???", + artists=[], + genre_raw="shanson", + api_language=None, + ) + assert result == "RU" + + +def test_detect_language_falls_back_to_alphabet(): + result = language.detect_language( + title="Отпусти меня", + artists=["ДДТ"], + genre_raw="rock", + api_language=None, + ) + assert result == "RU" + + +def test_detect_language_unknown_when_nothing_matches(): + result = language.detect_language( + title="123", + artists=[], + genre_raw=None, + api_language=None, + ) + assert result == "UNKNOWN"