diff --git a/sort_ym/lyrics.py b/sort_ym/lyrics.py new file mode 100644 index 0000000..f440d7d --- /dev/null +++ b/sort_ym/lyrics.py @@ -0,0 +1,87 @@ +from __future__ import annotations + +import json +import time +from pathlib import Path + +from yandex_music import Client +from yandex_music.exceptions import NotFoundError + +from . import language +from .ymclient import with_retries + +LYRICS_TEXT_CACHE_FILE = "lyrics_text.json" + + +def ru_lyric_track_ids(tracks_cache: dict[str, dict], lang_cache: dict[str, str | None]) -> list[str]: + """Числовые id RU-треков, у которых вообще есть текст. + + Гейт двойной: lyrics_available (иначе track_supplement заведомо вернёт запись без lyrics) + и RU - анализ INT-лирики вне скоупа фичи. Дедупликация нужна потому, что один и тот же + трек может быть лайкнут с разных альбомов: ключи tracks_cache при этом разные, а числовой + id (и, соответственно, ключ кэша текстов) - один. + """ + ids: list[str] = [] + seen: set[str] = set() + for t in tracks_cache.values(): + if not t.get("lyrics_available"): + continue + tid = str(t["id"]) + if tid in seen: + continue + lang = language.detect_language(t["title"], t["artists"], t["genre_raw"], lang_cache.get(tid)) + if lang == "RU": + seen.add(tid) + ids.append(tid) + return ids + + +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_lyrics_cache(cache_dir: Path) -> dict[str, str | None]: + cache_file = cache_dir / LYRICS_TEXT_CACHE_FILE + if cache_file.exists(): + return json.loads(cache_file.read_text(encoding="utf-8")) + return {} + + +def _fetch_one_lyrics(client: Client, track_id: str) -> str | None: + try: + supplement = with_retries(lambda: client.track_supplement(track_id)) + except NotFoundError: + return None + if supplement is None or supplement.lyrics is None: + return None + return supplement.lyrics.full_lyrics + + +def fetch_lyrics_text( + 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_TEXT_CACHE_FILE + cache = load_lyrics_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_lyrics(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) + without_text = sum(1 for tid in track_ids if cache.get(tid) is None) + print(f" готово: {len(track_ids)} треков, без текста {without_text}") + return cache diff --git a/tests/test_lyrics.py b/tests/test_lyrics.py new file mode 100644 index 0000000..48e67c9 --- /dev/null +++ b/tests/test_lyrics.py @@ -0,0 +1,119 @@ +import json +from pathlib import Path + +import pytest +from yandex_music.exceptions import NotFoundError + +from sort_ym import lyrics + + +def track(id_, title="Песня", artists=("Артист",), genre_raw="rock", lyrics_available=True): + return { + "id": id_, + "album_id": 10, + "title": title, + "artists": list(artists), + "artist_ids": [7], + "genre_raw": genre_raw, + "lyrics_available": lyrics_available, + } + + +def test_ru_lyric_track_ids_keeps_ru_and_drops_int(): + tracks_cache = { + "100:10": track(100, title="Тоска", artists=["Гражданская Оборона"]), + "200:10": track(200, title="Come as You Are", artists=["Nirvana"]), + } + + ids = lyrics.ru_lyric_track_ids(tracks_cache, lang_cache={}) + + assert ids == ["100"] + + +def test_ru_lyric_track_ids_trusts_api_language_over_alphabet(): + # Кириллическое название, но API говорит "en" - латиницы в тексте нет, а язык известен точно. + tracks_cache = {"100:10": track(100, title="Тоска", artists=["Артист"])} + + ids = lyrics.ru_lyric_track_ids(tracks_cache, lang_cache={"100": "en"}) + + assert ids == [] + + +def test_ru_lyric_track_ids_skips_tracks_without_lyrics(): + tracks_cache = {"100:10": track(100, title="Тоска", lyrics_available=False)} + + assert lyrics.ru_lyric_track_ids(tracks_cache, lang_cache={}) == [] + + +def test_ru_lyric_track_ids_deduplicates_same_numeric_id(): + # Один и тот же трек лайкнут с двух альбомов - числовой id общий, ключи кэша разные. + tracks_cache = { + "100:10": track(100, title="Тоска"), + "100:20": track(100, title="Тоска"), + } + + assert lyrics.ru_lyric_track_ids(tracks_cache, lang_cache={}) == ["100"] + + +class FakeLyrics: + def __init__(self, full_lyrics): + self.full_lyrics = full_lyrics + + +class FakeSupplement: + def __init__(self, full_lyrics): + self.lyrics = FakeLyrics(full_lyrics) if full_lyrics is not None else None + + +class FakeClient: + def __init__(self, texts): + self._texts = texts + self.requested: list[str] = [] + + def track_supplement(self, track_id): + self.requested.append(track_id) + if track_id not in self._texts: + raise NotFoundError("no such track") + return FakeSupplement(self._texts[track_id]) + + +def test_fetch_lyrics_text_caches_and_persists(tmp_path: Path): + client = FakeClient({"100": "Первая строка\nВторая строка"}) + + cache = lyrics.fetch_lyrics_text(client, ["100"], tmp_path, delay=0) + + assert cache["100"] == "Первая строка\nВторая строка" + assert lyrics.load_lyrics_cache(tmp_path) == cache + + +def test_fetch_lyrics_text_skips_already_cached(tmp_path: Path): + (tmp_path / lyrics.LYRICS_TEXT_CACHE_FILE).write_text( + json.dumps({"100": "уже загружено"}, ensure_ascii=False), encoding="utf-8" + ) + client = FakeClient({"100": "новый текст"}) + + cache = lyrics.fetch_lyrics_text(client, ["100"], tmp_path, delay=0) + + assert client.requested == [], "уже закэшированный трек не должен запрашиваться снова" + assert cache["100"] == "уже загружено" + + +def test_fetch_lyrics_text_stores_none_for_missing_track_and_does_not_retry(tmp_path: Path): + # NotFoundError - окончательный ответ, а не сбой; None пишется в кэш как "текста нет", + # иначе такой трек перезапрашивался бы при каждом запуске. + client = FakeClient({}) + + lyrics.fetch_lyrics_text(client, ["100"], tmp_path, delay=0) + second_client = FakeClient({}) + cache = lyrics.fetch_lyrics_text(second_client, ["100"], tmp_path, delay=0) + + assert cache["100"] is None + assert second_client.requested == [] + + +def test_fetch_lyrics_text_stores_none_when_supplement_has_no_lyrics(tmp_path: Path): + client = FakeClient({"100": None}) + + cache = lyrics.fetch_lyrics_text(client, ["100"], tmp_path, delay=0) + + assert cache["100"] is None