report: --order/--extra flags, added_at and artist stats in track cache

CSV can now keep source playlist order instead of forced playlist grouping,
and optionally include added_at/duration_ms/release_date/artist rating from
the API that were previously fetched but discarded.
This commit is contained in:
Andrew Pimenov 2026-08-11 01:25:51 +03:00
parent 188fcc8b20
commit 5d4cab5d68
11 changed files with 321 additions and 46 deletions

View file

@ -170,13 +170,19 @@ report/apply/digest --source <url> → source.parse_playlist_url() → (user_id,
- `.token` — the account's OAuth token (not committed).
- `cache/tracks.json` — liked track metadata, including `artist_ids` (the track's artist ids),
`album_title` and `year` (the album's title and release year, `Album.title`/`Album.year`; `year`
can be `null`).
- `cache/artist_genres.json` — `{artist_id: {name, genres: [slug, ...]}}`, one or more genres per
artist.
can be `null`), plus `added_at` (when the track was liked, `TrackShort.timestamp`),
`duration_ms`, `track_version`/`album_version`, `release_date`, `album_likes_count` - raw API
fields not used for classification but available via `report --extra` (see below).
- `cache/artist_genres.json` — `{artist_id: {name, genres: [slug, ...], counts, ratings}}`, one or
more genres per artist; `counts`/`ratings` are `Artist.counts`/`Artist.ratings`, same
`report --extra`-only purpose as `added_at` above.
- `cache/genre_catalog.json` — a flat snapshot of Yandex's genre tree (id -> title, root_id).
- `cache/lyrics_lang.json` — lyrics language per track (id -> language code | null).
- `out/report.csv` — a preview of the playlist breakdown before `apply` (including the
`fine_bucket` column — the sub-genre before any collapsing into a top-level bucket).
`report --order playlist` keeps source track order instead of sorting by target playlist;
`report --extra` adds the columns listed above (`added_at`, `duration_ms`, `release_date`,
`artist_rating_month/week/day`, etc. - full list in `report.EXTRA_FIELDNAMES`).
- `out/digest.md` — a compact library summary for pasting into an LLM chat (top artists, top
albums, genre/language/decade breakdowns).
- `out/report_source.csv`, `out/digest_source.md` — the same reports for a `--source` run; a

View file

@ -169,13 +169,19 @@ report/apply/digest --source <url> → source.parse_playlist_url() → (user_id,
- `.token` — OAuth-токен аккаунта (не коммитится).
- `cache/tracks.json` — метаданные лайкнутых треков, включая `artist_ids` (id артистов трека),
`album_title` и `year` (название и год релиза альбома, `Album.title`/`Album.year`; `year` может
быть `null`).
- `cache/artist_genres.json` — `{artist_id: {name, genres: [slug, ...]}}`, один и более жанров
на артиста.
быть `null`), а также `added_at` (дата добавления трека в лайки, `TrackShort.timestamp`),
`duration_ms`, `track_version`/`album_version`, `release_date`, `album_likes_count` - сырые поля
API, не используемые в классификации, но доступные в `report --extra` (см. ниже).
- `cache/artist_genres.json` — `{artist_id: {name, genres: [slug, ...], counts, ratings}}`, один и
более жанров на артиста; `counts`/`ratings` - те же поля `Artist.counts`/`Artist.ratings`, что
и `added_at` выше, только для `report --extra`.
- `cache/genre_catalog.json` — плоский снимок дерева жанров Яндекса (id -> title, root_id).
- `cache/lyrics_lang.json` — язык текста песни по треку (id -> language code | null).
- `out/report.csv` — предпросмотр распределения по плейлистам перед `apply` (включая колонку
`fine_bucket` — под-жанр до возможного схлопывания в крупную корзину).
`fine_bucket` — под-жанр до возможного схлопывания в крупную корзину). `report --order playlist`
сохраняет исходный порядок треков вместо сортировки по целевому плейлисту; `report --extra`
добавляет колонки из списка выше (`added_at`, `duration_ms`, `release_date`,
`artist_rating_month/week/day` и т.д. - полный список в `report.EXTRA_FIELDNAMES`).
- `out/digest.md` — сжатая сводка библиотеки для вставки в чат с LLM (топ-исполнители, топ-альбомы,
распределение по жанрам, языку и десятилетиям).
- `out/report_source.csv`, `out/digest_source.md` — те же отчёты, но для запуска с `--source`;

View file

@ -37,6 +37,8 @@ For development (tests):
.venv\Scripts\python -m sort_ym auth # log in to a Yandex account via device-flow and save the token to .token
.venv\Scripts\python -m sort_ym fetch # download liked tracks (cache/tracks.json) and their artists' genres (cache/artist_genres.json); does not change the account
.venv\Scripts\python -m sort_ym report # compute genre and language per track, save out/report.csv (does not change the account)
.venv\Scripts\python -m sort_ym report --order playlist # skip playlist-based sorting - keep source track order (as in likes/the --source playlist)
.venv\Scripts\python -m sort_ym report --extra # + added_at/duration_ms/release_date/artist rating columns etc.
.venv\Scripts\python -m sort_ym apply --limit 20 --yes # dry run: create playlists and add the first 20 tracks
.venv\Scripts\python -m sort_ym apply --yes # create playlists and add all tracks (changes the account)
.venv\Scripts\python -m sort_ym digest # library summary (top artists, genres) to out/digest.md - to paste into an LLM chat; no network used
@ -193,6 +195,8 @@ python -m sort_ym apply --source https://music.yandex.ru/users/vasya/playlists/1
- `report --source`/`digest --source` write to separate files —
`out/report_source.csv`/`out/digest_source.md` — and never overwrite your main
`out/report.csv`/`out/digest.md`.
- `report --source <url> --order playlist` — tracks in the CSV keep the same order as in the
playlist itself, instead of being resorted by target playlist.
## Handy things

View file

@ -37,6 +37,8 @@ python -m venv .venv
.venv\Scripts\python -m sort_ym auth # войти в аккаунт Яндекса через device-flow и сохранить токен в .token
.venv\Scripts\python -m sort_ym fetch # скачать лайкнутые треки (cache/tracks.json) и жанры их артистов (cache/artist_genres.json); аккаунт не меняет
.venv\Scripts\python -m sort_ym report # посчитать жанр и язык каждого трека, сохранить out/report.csv (аккаунт не меняет)
.venv\Scripts\python -m sort_ym report --order playlist # без сортировки по плейлистам - треки в исходном порядке (как в лайках/--source плейлисте)
.venv\Scripts\python -m sort_ym report --extra # + колонки added_at/duration_ms/release_date/рейтинг артиста и т.п.
.venv\Scripts\python -m sort_ym apply --limit 20 --yes # пробный запуск: создать плейлисты и добавить в них первые 20 треков
.venv\Scripts\python -m sort_ym apply --yes # создать плейлисты и добавить в них все треки (меняет аккаунт)
.venv\Scripts\python -m sort_ym digest # сводка библиотеки (топ-исполнители, жанры) в out/digest.md - чтобы вставить в чат с LLM; сеть не используется
@ -194,6 +196,8 @@ python -m sort_ym apply --source https://music.yandex.ru/users/vasya/playlists/1
- Результаты `report --source`/`digest --source` сохраняются в отдельные файлы —
`out/report_source.csv`/`out/digest_source.md` — и не затирают ваши основные `out/report.csv`/
`out/digest.md`.
- `report --source <ссылка> --order playlist` — треки в CSV пойдут в том же порядке, что и в самом
плейлисте, без пересортировки по целевому плейлисту.
## Handy things

View file

@ -82,8 +82,11 @@ def cmd_report(args: argparse.Namespace) -> None:
if args.source:
ctx = _source_context(cfg, args.source)
rows = report.build_rows(ctx.tracks, ctx.lang_cache, ctx.catalog, ctx.artist_genres, cfg.small_group_min)
out_file = report.write_report(rows, cfg.out_dir, report.REPORT_SOURCE_FILE)
rows = report.build_rows(
ctx.tracks, ctx.lang_cache, ctx.catalog, ctx.artist_genres, cfg.small_group_min,
order=args.order, extra_fields=args.extra,
)
out_file = report.write_report(rows, cfg.out_dir, report.REPORT_SOURCE_FILE, extra_fields=args.extra)
_print_report_summary(rows, out_file)
return
@ -99,8 +102,11 @@ def cmd_report(args: argparse.Namespace) -> None:
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, artist_genres, cfg.small_group_min)
out_file = report.write_report(rows, cfg.out_dir)
rows = report.build_rows(
tracks_cache, lang_cache, catalog, artist_genres, cfg.small_group_min,
order=args.order, extra_fields=args.extra,
)
out_file = report.write_report(rows, cfg.out_dir, extra_fields=args.extra)
_print_report_summary(rows, out_file)
@ -195,6 +201,16 @@ def main(argv: list[str] | None = None) -> None:
p_report = sub.add_parser("report", help="посчитать распределение по плейлистам и сохранить out/report.csv (ничего не меняет в аккаунте)")
p_report.add_argument("--source", metavar="URL", default=None, help=SOURCE_HELP)
p_report.add_argument(
"--order", choices=["grouped", "playlist"], default="grouped",
help="grouped (по умолчанию) - сортировка по целевому плейлисту; "
"playlist - как треки идут в источнике (в лайках или в --source плейлисте), без сортировки",
)
p_report.add_argument(
"--extra", action="store_true",
help="добавить в CSV колонки с доп. данными API: дата добавления, длительность, дата релиза "
"альбома, рейтинг/счётчики первого артиста трека",
)
p_apply = sub.add_parser("apply", help="создать плейлисты и добавить треки")
p_apply.add_argument("--yes", action="store_true", help="подтвердить внесение изменений в аккаунт")

View file

@ -19,7 +19,7 @@ def _track_genre(track: Track) -> str | None:
return None
def serialize_track(track: Track) -> dict:
def serialize_track(track: Track, added_at: str | None = None) -> dict:
# album_id, album_title и year берутся из одного и того же albums[0] - иначе название
# альбома может разъехаться с его id (у трека может быть несколько альбомов).
album = track.albums[0] if track.albums else None
@ -33,6 +33,14 @@ def serialize_track(track: Track) -> dict:
"artist_ids": [a.id for a in track.artists if a.name and a.id is not None],
"genre_raw": _track_genre(track),
"lyrics_available": bool(track.lyrics_available),
# added_at - из TrackShort.timestamp (список лайков/плейлиста), не из самого Track -
# полный Track этой даты не содержит вообще, её знает только позиция в списке.
"added_at": added_at,
"duration_ms": track.duration_ms,
"track_version": track.version,
"album_version": album.version if album else None,
"release_date": album.release_date if album else None,
"album_likes_count": album.likes_count if album else None,
}
@ -65,6 +73,7 @@ def fetch_liked_tracks(
return cache
all_ids = likes.tracks_ids
added_at_by_id = {str(t.id): t.timestamp for t in likes.tracks}
# "not in cache" - трек ещё не загружен; "artist_ids"/"year" not in cache[tid] - трек загружен
# старой версией кэша (до добавления этих полей) и должен быть докачан повторно. Проверяем
# наличие ключа, а не правдивость значения - year легитимно может быть None у части альбомов,
@ -72,7 +81,10 @@ def fetch_liked_tracks(
missing_ids = [
tid
for tid in all_ids
if tid not in cache or "artist_ids" not in cache[tid] or "year" not in cache[tid]
if tid not in cache
or "artist_ids" not in cache[tid]
or "year" not in cache[tid]
or "duration_ms" not in cache[tid]
]
print(f"Лайкнутых треков: {len(all_ids)}, новых для загрузки: {len(missing_ids)}")
@ -89,7 +101,7 @@ def fetch_liked_tracks(
# альбомах, "основной" альбом в ответе может отличаться от того, под которым трек лайкнут.
# Иначе такой трек тут же попадает под stale-очистку ниже как "больше не лайкнутый".
for original_id, track in zip(batch, tracks):
cache[original_id] = serialize_track(track)
cache[original_id] = serialize_track(track, added_at=added_at_by_id.get(original_id.split(":")[0]))
_atomic_write_json(cache_file, cache)
print(f" загружено {len(cache)}/{len(all_ids)}")
time.sleep(batch_delay)
@ -137,7 +149,7 @@ def fetch_artist_genres(
tracks_cache = load_tracks_cache(cache_dir)
all_artist_ids = unique_artist_ids(tracks_cache)
missing_ids = [aid for aid in all_artist_ids if aid not in cache]
missing_ids = [aid for aid in all_artist_ids if aid not in cache or "counts" not in cache[aid]]
if not missing_ids:
return cache
@ -150,6 +162,8 @@ def fetch_artist_genres(
cache[str(artist.id)] = {
"name": artist.name or "",
"genres": list(artist.genres or []),
"counts": artist.counts.to_dict() if artist.counts else None,
"ratings": artist.ratings.to_dict() if artist.ratings else None,
}
_atomic_write_json(cache_file, cache)
print(f" загружено {len(cache)}/{len(all_artist_ids)}")

View file

@ -7,16 +7,61 @@ from . import classify, genres, language
FIELDNAMES = ["title", "artists", "genre_raw", "fine_bucket", "bucket", "lang", "target_playlist", "id", "album_id"]
# Только вместе с --extra: сырые поля API, которых нет в базовом отчёте (см. обсуждение в чате -
# added_at из TrackShort.timestamp, остальное - из Track/Album/Artist, но не заполнено стабильно
# у part данных вроде explicit/lyrics_info, поэтому в набор не включено).
EXTRA_FIELDNAMES = [
"added_at",
"duration_ms",
"track_version",
"album_version",
"release_date",
"album_likes_count",
"artist_track_count",
"artist_direct_albums",
"artist_also_albums",
"artist_also_tracks",
"artist_rating_month",
"artist_rating_week",
"artist_rating_day",
]
REPORT_FILE = "report.csv"
REPORT_SOURCE_FILE = "report_source.csv"
def _extra_row_fields(t: dict, artist_genres: dict[str, dict]) -> dict:
# Счётчики/рейтинг берём у первого артиста трека - у треков с несколькими артистами
# остальные не имеют выделенной колонки (см. обсуждение в чате).
artist_ids = t.get("artist_ids", [])
artist = artist_genres.get(str(artist_ids[0])) if artist_ids else None
counts = (artist or {}).get("counts") or {}
ratings = (artist or {}).get("ratings") or {}
return {
"added_at": t.get("added_at"),
"duration_ms": t.get("duration_ms"),
"track_version": t.get("track_version") or "",
"album_version": t.get("album_version") or "",
"release_date": t.get("release_date"),
"album_likes_count": t.get("album_likes_count"),
"artist_track_count": counts.get("tracks"),
"artist_direct_albums": counts.get("direct_albums"),
"artist_also_albums": counts.get("also_albums"),
"artist_also_tracks": counts.get("also_tracks"),
"artist_rating_month": ratings.get("month"),
"artist_rating_week": ratings.get("week"),
"artist_rating_day": ratings.get("day"),
}
def build_rows(
tracks_cache: dict[str, dict],
lang_cache: dict[str, str | None],
genre_catalog: dict[str, dict],
artist_genres: dict[str, dict],
small_group_min: int,
order: str = "grouped",
extra_fields: bool = False,
) -> list[dict]:
# Проход 1: под-жанр (fine) и язык на каждый трек, размеры групп (fine, lang) фиксируем сразу.
prelim = []
@ -45,28 +90,36 @@ def build_rows(
else:
bucket = fine
playlist = classify.playlist_name(bucket, lang)
rows.append(
{
"title": t["title"],
"artists": ", ".join(t["artists"]),
"genre_raw": t["genre_raw"] or "",
"fine_bucket": fine,
"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"]))
row = {
"title": t["title"],
"artists": ", ".join(t["artists"]),
"genre_raw": t["genre_raw"] or "",
"fine_bucket": fine,
"bucket": bucket,
"lang": lang,
"target_playlist": playlist,
"id": t["id"],
"album_id": t["album_id"],
}
if extra_fields:
row.update(_extra_row_fields(t, artist_genres))
rows.append(row)
# "playlist" - порядок как в источнике (tracks_cache сохраняет порядок плейлиста, см. source.py);
# "grouped" (по умолчанию) - принудительная сортировка по целевому плейлисту, как раньше.
if order == "grouped":
rows.sort(key=lambda r: (r["target_playlist"], r["artists"], r["title"]))
elif order != "playlist":
raise ValueError(f"неизвестный order: {order!r}, ожидается 'grouped' или 'playlist'")
return rows
def write_report(rows: list[dict], out_dir: Path, filename: str = REPORT_FILE) -> Path:
def write_report(rows: list[dict], out_dir: Path, filename: str = REPORT_FILE, extra_fields: bool = False) -> Path:
out_dir.mkdir(parents=True, exist_ok=True)
out_file = out_dir / filename
fieldnames = FIELDNAMES + EXTRA_FIELDNAMES if extra_fields else FIELDNAMES
with out_file.open("w", encoding="utf-8-sig", newline="") as f:
writer = csv.DictWriter(f, fieldnames=FIELDNAMES)
writer = csv.DictWriter(f, fieldnames=fieldnames)
writer.writeheader()
writer.writerows(rows)
return out_file

View file

@ -154,6 +154,7 @@ def fetch_playlist_tracks(
# Дедуп с сохранением порядка - чужие плейлисты чаще лайков содержат один трек дважды.
ids = list(dict.fromkeys(t.track_id for t in short))
added_at_by_id = {t.track_id: t.timestamp for t in short}
print(f"Треков в плейлисте «{info.title}»: {len(ids)}")
@ -170,7 +171,7 @@ def fetch_playlist_tracks(
if track is None:
missing += 1
continue
tracks[requested_id] = fetch.serialize_track(track)
tracks[requested_id] = fetch.serialize_track(track, added_at=added_at_by_id.get(requested_id))
print(f" загружено {len(tracks)}/{len(ids)}")
time.sleep(batch_delay)
@ -191,7 +192,12 @@ def fetch_artist_genres_map(
for batch in chunked(artist_ids, batch_size):
artists = with_retries(lambda: client.artists(batch))
for artist in artists:
result[str(artist.id)] = {"name": artist.name or "", "genres": list(artist.genres or [])}
result[str(artist.id)] = {
"name": artist.name or "",
"genres": list(artist.genres or []),
"counts": artist.counts.to_dict() if artist.counts else None,
"ratings": artist.ratings.to_dict() if artist.ratings else None,
}
time.sleep(batch_delay)
return result

View file

@ -5,32 +5,49 @@ from sort_ym import fetch
class FakeArtist:
def __init__(self, id_, name, genres=None):
def __init__(self, id_, name, genres=None, counts=None, ratings=None):
self.id = id_
self.name = name
self.genres = genres
self.counts = counts
self.ratings = ratings
class FakeAlbum:
def __init__(self, id_, genre=None, title="", year=None):
def __init__(self, id_, genre=None, title="", year=None, version=None, release_date=None, likes_count=None):
self.id = id_
self.genre = genre
self.title = title
self.year = year
self.version = version
self.release_date = release_date
self.likes_count = likes_count
class FakeTrack:
def __init__(self, id_, albums, title="Title", artists=None, lyrics_available=False):
def __init__(self, id_, albums, title="Title", artists=None, lyrics_available=False, duration_ms=None, version=None):
self.id = id_
self.albums = albums
self.title = title
self.artists = artists or [FakeArtist(1, "Artist")]
self.lyrics_available = lyrics_available
self.duration_ms = duration_ms
self.version = version
class FakeTrackShort:
def __init__(self, id_, timestamp=None):
self.id = id_
self.timestamp = timestamp
class FakeLikes:
def __init__(self, tracks_ids):
def __init__(self, tracks_ids, tracks=None):
self.tracks_ids = tracks_ids
# По умолчанию строим TrackShort-заглушки из tracks_ids ("100:10" -> id=100) -
# реальный API всегда возвращает .tracks вместе с .tracks_ids, тесты не завязанные
# на timestamp могут его не указывать явно.
self.tracks = tracks if tracks is not None else [FakeTrackShort(tid.split(":")[0]) for tid in tracks_ids]
class FakeClient:
@ -182,6 +199,34 @@ def test_fetch_re_downloads_old_cache_entries_missing_year(tmp_path: Path):
assert cache["100:10"]["album_title"] == "Iowa"
def test_fetch_re_downloads_old_cache_entries_missing_duration_ms(tmp_path: Path):
# Кэш до добавления added_at/duration_ms и т.п. (но уже с artist_ids/year) должен
# быть докачан повторно.
cache_dir = tmp_path
cache_dir.mkdir(exist_ok=True)
old_entry = {
"id": 100,
"album_id": 10,
"album_title": "Iowa",
"year": 2001,
"title": "Old",
"artists": ["Artist A"],
"artist_ids": [7],
"genre_raw": "pop",
"lyrics_available": False,
}
(cache_dir / fetch.TRACKS_CACHE_FILE).write_text(
json.dumps({"100:10": old_entry}, ensure_ascii=False), encoding="utf-8"
)
track = FakeTrack(id_=100, albums=[FakeAlbum(10, title="Iowa", year=2001)], artists=[FakeArtist(7, "Artist A")], duration_ms=250000)
client = FakeClient(likes_ids=["100:10"], tracks_by_batch={("100:10",): [track]})
cache = fetch.fetch_liked_tracks(client, cache_dir, batch_size=100, batch_delay=0)
assert cache["100:10"]["duration_ms"] == 250000
def test_fetch_does_not_redownload_entries_with_legitimately_unknown_year(tmp_path: Path):
# Регрессия: year=None легитимен (у альбома действительно нет года), и такая запись
# не должна перекачиваться при каждом fetch - проверяем наличие ключа, а не значение.
@ -197,6 +242,7 @@ def test_fetch_does_not_redownload_entries_with_legitimately_unknown_year(tmp_pa
"artist_ids": [7],
"genre_raw": "pop",
"lyrics_available": False,
"duration_ms": 200000,
}
(cache_dir / fetch.TRACKS_CACHE_FILE).write_text(
json.dumps({"100:10": entry}, ensure_ascii=False), encoding="utf-8"
@ -231,8 +277,8 @@ def test_fetch_artist_genres_batches_and_caches(tmp_path: Path):
result = fetch.fetch_artist_genres(client, cache_dir, batch_size=100, batch_delay=0)
assert result["7"] == {"name": "Artist A", "genres": ["indie"]}
assert result["9"] == {"name": "Artist B", "genres": ["rock", "punk"]}
assert result["7"] == {"name": "Artist A", "genres": ["indie"], "counts": None, "ratings": None}
assert result["9"] == {"name": "Artist B", "genres": ["rock", "punk"], "counts": None, "ratings": None}
persisted = fetch.load_artist_genres(cache_dir)
assert persisted == result
@ -247,7 +293,8 @@ def test_fetch_artist_genres_skips_already_cached(tmp_path: Path):
json.dumps(tracks_cache, ensure_ascii=False), encoding="utf-8"
)
(cache_dir / fetch.ARTIST_GENRES_CACHE_FILE).write_text(
json.dumps({"7": {"name": "Artist A", "genres": ["indie"]}}, ensure_ascii=False), encoding="utf-8"
json.dumps({"7": {"name": "Artist A", "genres": ["indie"], "counts": None, "ratings": None}}, ensure_ascii=False),
encoding="utf-8",
)
def boom(batch):
@ -258,4 +305,28 @@ def test_fetch_artist_genres_skips_already_cached(tmp_path: Path):
result = fetch.fetch_artist_genres(client, cache_dir, batch_size=100, batch_delay=0)
assert result == {"7": {"name": "Artist A", "genres": ["indie"]}}
assert result == {"7": {"name": "Artist A", "genres": ["indie"], "counts": None, "ratings": None}}
def test_fetch_artist_genres_re_downloads_old_cache_entries_missing_counts(tmp_path: Path):
# Кэш до добавления counts/ratings (но уже с name/genres) должен быть докачан повторно.
cache_dir = tmp_path
tracks_cache = {
"100:10": {"id": 100, "album_id": 10, "title": "T1", "artists": ["A"], "artist_ids": [7], "genre_raw": "pop", "lyrics_available": False},
}
(cache_dir / fetch.TRACKS_CACHE_FILE).write_text(
json.dumps(tracks_cache, ensure_ascii=False), encoding="utf-8"
)
(cache_dir / fetch.ARTIST_GENRES_CACHE_FILE).write_text(
json.dumps({"7": {"name": "Artist A", "genres": ["indie"]}}, ensure_ascii=False), encoding="utf-8"
)
client = FakeClient(
likes_ids=[],
tracks_by_batch={},
artists_by_batch={("7",): [FakeArtist(7, "Artist A", genres=["indie"])]},
)
result = fetch.fetch_artist_genres(client, cache_dir, batch_size=100, batch_delay=0)
assert result["7"] == {"name": "Artist A", "genres": ["indie"], "counts": None, "ratings": None}

View file

@ -65,6 +65,88 @@ def test_write_report_accepts_custom_filename_for_source_runs(tmp_path):
assert out_file.exists()
def test_order_playlist_preserves_source_order():
# tracks_cache - dict, порядок вставки == порядок в источнике (см. source.py).
# order="playlist" должен сохранить этот порядок вместо сортировки по target_playlist.
tracks_cache = {
"3:3": _track(3, "C", ["Z Artist"], "punk"),
"1:1": _track(1, "A", ["A Artist"], "indie"),
"2:2": _track(2, "B", ["M Artist"], "punk"),
}
rows = report.build_rows(
tracks_cache=tracks_cache,
lang_cache={},
genre_catalog=CATALOG,
artist_genres={},
small_group_min=12,
order="playlist",
)
assert [r["title"] for r in rows] == ["C", "A", "B"]
def test_order_grouped_sorts_by_target_playlist():
tracks_cache = {
"3:3": _track(3, "C", ["Z Artist"], "punk"),
"1:1": _track(1, "A", ["A Artist"], "indie"),
"2:2": _track(2, "B", ["M Artist"], "punk"),
}
rows = report.build_rows(
tracks_cache=tracks_cache,
lang_cache={},
genre_catalog=CATALOG,
artist_genres={},
small_group_min=12,
order="grouped",
)
assert rows == sorted(rows, key=lambda r: (r["target_playlist"], r["artists"], r["title"]))
def test_extra_fields_adds_columns_from_track_and_artist_cache():
tracks_cache = {
"1:1": {
**_track(1, "T", ["A"], "punk", artist_ids=[42]),
"added_at": "2026-01-01T00:00:00+00:00",
"duration_ms": 123000,
"track_version": "Remix",
"album_version": None,
"release_date": "2020-05-01T00:00:00+03:00",
"album_likes_count": 10,
},
}
artist_genres = {"42": {"name": "A", "genres": [], "counts": {"tracks": 5}, "ratings": {"month": 7}}}
rows = report.build_rows(
tracks_cache=tracks_cache,
lang_cache={},
genre_catalog=CATALOG,
artist_genres=artist_genres,
small_group_min=12,
extra_fields=True,
)
row = rows[0]
assert row["added_at"] == "2026-01-01T00:00:00+00:00"
assert row["duration_ms"] == 123000
assert row["track_version"] == "Remix"
assert row["release_date"] == "2020-05-01T00:00:00+03:00"
assert row["artist_track_count"] == 5
assert row["artist_rating_month"] == 7
def test_write_report_extra_fields_writes_extended_header(tmp_path):
rows = [{"title": "T", "artists": "A", "genre_raw": "punk", "fine_bucket": "punk", "bucket": "punk", "lang": "RU", "target_playlist": "Панк — RU", "id": 1, "album_id": 10, "added_at": "x", "duration_ms": 1, "track_version": "", "album_version": "", "release_date": "y", "album_likes_count": 1, "artist_track_count": 1, "artist_direct_albums": 1, "artist_also_albums": 1, "artist_also_tracks": 1, "artist_rating_month": 1, "artist_rating_week": 1, "artist_rating_day": 1}]
out_file = report.write_report(rows, tmp_path, extra_fields=True)
header = out_file.read_text(encoding="utf-8-sig").splitlines()[0]
assert "added_at" in header
assert "artist_rating_day" in header
def test_collapse_is_single_pass_and_terminates():
# Группа ровно на границе порога (== small_group_min) НЕ схлопывается (строгое <).
tracks_cache = {

View file

@ -5,33 +5,41 @@ from sort_ym import digest, report, source
class FakeArtist:
def __init__(self, id_, name, genres=None):
def __init__(self, id_, name, genres=None, counts=None, ratings=None):
self.id = id_
self.name = name
self.genres = genres
self.counts = counts
self.ratings = ratings
class FakeAlbum:
def __init__(self, id_, genre=None, title="", year=None):
def __init__(self, id_, genre=None, title="", year=None, version=None, release_date=None, likes_count=None):
self.id = id_
self.genre = genre
self.title = title
self.year = year
self.version = version
self.release_date = release_date
self.likes_count = likes_count
class FakeTrack:
def __init__(self, id_, albums, title="Title", artists=None, lyrics_available=False):
def __init__(self, id_, albums, title="Title", artists=None, lyrics_available=False, duration_ms=None, version=None):
self.id = id_
self.albums = albums
self.title = title
self.artists = artists or [FakeArtist(1, "Artist")]
self.lyrics_available = lyrics_available
self.duration_ms = duration_ms
self.version = version
class FakeTrackShort:
def __init__(self, id_, album_id=None):
def __init__(self, id_, album_id=None, timestamp=None):
self.id = id_
self.album_id = album_id
self.timestamp = timestamp
@property
def track_id(self):
@ -243,7 +251,10 @@ def test_fetch_artist_genres_map_batches_without_touching_disk(tmp_path):
result = source.fetch_artist_genres_map(client, ["7", "9"], batch_size=100, batch_delay=0)
assert result == {"7": {"name": "A", "genres": ["indie"]}, "9": {"name": "B", "genres": ["rock"]}}
assert result == {
"7": {"name": "A", "genres": ["indie"], "counts": None, "ratings": None},
"9": {"name": "B", "genres": ["rock"], "counts": None, "ratings": None},
}
assert list(tmp_path.iterdir()) == [] # ничего не записано на диск
@ -253,10 +264,12 @@ def test_source_track_shape_matches_fetch_cache_shape(tmp_path):
expected_keys = {
"id", "album_id", "album_title", "year", "title",
"artists", "artist_ids", "genre_raw", "lyrics_available",
"added_at", "duration_ms", "track_version", "album_version", "release_date", "album_likes_count",
}
class FakeLikes:
tracks_ids = ["100:10"]
tracks = [FakeTrackShort(100, timestamp="2026-01-01T00:00:00+00:00")]
class FakeLikesClient:
def users_likes_tracks(self):