init: сортировка лайков Яндекс.Музыки по жанру и языку

CLI на yandex-music (device-flow auth, fetch/report/apply), жанр через
реальное дерево client.genres(), язык гибридом (track_supplement + алфавит).

.gitignore

@@ -0,0 +1,8 @@
+.venv/
+__pycache__/
+*.pyc
+.token
+cache/
+out/
+.claude/
+plans/

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`.

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).

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"

requirements-dev.txt

@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest==8.3.4

requirements.txt

@@ -0,0 +1 @@
+yandex-music==3.0.0

sort_ym/__main__.py

@@ -0,0 +1,4 @@
+from .cli import main
+
+if __name__ == "__main__":
+    main()

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}")

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)

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}"

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()

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"],
+    )

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

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")

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

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

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]

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"}

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"