fix duplicate track/playlist inserts on network timeout

users_playlists_insert_track и users_playlists_create - операции записи:
таймаут не значит, что запрос не дошёл до сервера. Прошлый фикс слепо
повторял такой вызов при NetworkError, и если сервер уже применил
вставку (а ответ просто потерялся), трек добавлялся в плейлист второй
раз. Теперь при сетевой ошибке сначала перепроверяется фактическое
состояние плейлиста/списка плейлистов, и повтор происходит только если
трек/плейлист там правда ещё отсутствует.

sort_ym/apply.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import time
 
 from yandex_music import Client, Playlist
+from yandex_music.exceptions import NetworkError
 
 from .ymclient import with_retries
 
@@ -22,11 +23,60 @@ def _existing_track_ids(client: Client, playlist: Playlist) -> set[str]:
 def _get_or_create_playlist(client: Client, title: str, existing: dict[str, Playlist]) -> Playlist:
     if title in existing:
         return existing[title]
-    playlist = with_retries(lambda: client.users_playlists_create(title, visibility="private"))
+    try:
+        playlist = with_retries(lambda: client.users_playlists_create(title, visibility="private"), max_attempts=1)
+    except NetworkError:
+        # Таймаут не значит, что запрос не дошёл до сервера - плейлист мог всё же
+        # создаться. Перепроверяем по имени вместо слепого повтора (иначе - дубль плейлиста).
+        fresh = _existing_playlists_by_title(client)
+        if title in fresh:
+            existing[title] = fresh[title]
+            return fresh[title]
+        raise
     existing[title] = playlist
     return playlist
 
 
+def _insert_track_verified(
+    client: Client,
+    playlist: Playlist,
+    row: dict,
+    track_key: str,
+    revision: int,
+    max_attempts: int = 4,
+) -> int:
+    """Добавляет трек в плейлист, возвращает новую revision.
+
+    users_playlists_insert_track - операция записи, а не чтения: при таймауте
+    непонятно, применилась ли вставка на сервере. Слепой повтор такого вызова
+    рискует добавить трек второй раз. Поэтому при сетевой ошибке перепроверяем
+    реальное содержимое плейлиста и повторяем вставку, только если трека там
+    действительно ещё нет.
+    """
+    for attempt in range(max_attempts):
+        try:
+            updated = with_retries(
+                lambda: client.users_playlists_insert_track(
+                    playlist.kind, row["id"], row["album_id"], revision=revision
+                ),
+                max_attempts=1,
+            )
+            return updated.revision if updated is not None and updated.revision is not None else revision
+        except NetworkError as e:
+            last_attempt = attempt == max_attempts - 1
+            print(f"  сетевая ошибка при вставке трека ({e}), проверяю фактическое состояние плейлиста...")
+            fresh = with_retries(lambda: client.users_playlists(playlist.kind))
+            if fresh is not None:
+                revision = fresh.revision or revision
+                if track_key in {t.track_id for t in fresh.tracks}:
+                    return revision
+            if last_attempt:
+                raise
+            time.sleep(2.0 * (2**attempt))
+
+    raise RuntimeError("unreachable")
+
+
 def apply_classification(
     client: Client,
     rows: list[dict],
@@ -53,16 +103,7 @@ def apply_classification(
             if track_key in already:
                 continue
 
-            updated = with_retries(
+            revision = _insert_track_verified(client, playlist, row, track_key, revision)
-                lambda: 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)

tests/test_apply.py

@@ -0,0 +1,94 @@
+from yandex_music.exceptions import TimedOutError
+
+from sort_ym import apply as apply_mod
+
+
+class FakePlaylist:
+    def __init__(self, kind, title, revision=1, tracks=None):
+        self.kind = kind
+        self.title = title
+        self.revision = revision
+        self.tracks = tracks or []
+
+
+class FakeTrackShort:
+    def __init__(self, track_id):
+        self.track_id = track_id
+
+
+class FakeClient:
+    def __init__(self):
+        self.playlists = {100: FakePlaylist(100, "Рок — RU", revision=5, tracks=[FakeTrackShort("1:10")])}
+        self.created = []
+        self.inserted = []
+        self.next_kind = 200
+
+    def users_playlists_list(self):
+        return list(self.playlists.values())
+
+    def users_playlists(self, kind):
+        return self.playlists.get(kind)
+
+    def users_playlists_create(self, title, visibility="private"):
+        kind = self.next_kind
+        self.next_kind += 1
+        p = FakePlaylist(kind, title)
+        self.playlists[kind] = p
+        self.created.append(title)
+        return p
+
+    def users_playlists_insert_track(self, kind, track_id, album_id, revision=1):
+        self.inserted.append((kind, track_id, album_id, revision))
+        p = self.playlists[kind]
+        p.revision = revision + 1
+        return p
+
+
+class FlakyClient:
+    """insert_track кидает таймаут на первой попытке, но трек всё же
+    успевает примениться на сервере (ответ теряется, а не сам запрос)."""
+
+    def __init__(self):
+        self.playlist = FakePlaylist(100, "Рок — RU", revision=5, tracks=[])
+        self.insert_calls = 0
+        self.server_tracks: list[tuple[int, int]] = []
+
+    def users_playlists_list(self):
+        return [self.playlist]
+
+    def users_playlists(self, kind):
+        self.playlist.tracks = [FakeTrackShort(f"{i}:{a}") for (i, a) in self.server_tracks]
+        return self.playlist
+
+    def users_playlists_insert_track(self, kind, track_id, album_id, revision=1):
+        self.insert_calls += 1
+        self.server_tracks.append((track_id, album_id))
+        if self.insert_calls == 1:
+            raise TimedOutError()
+        self.playlist.revision += 1
+        self.playlist.tracks = [FakeTrackShort(f"{i}:{a}") for (i, a) in self.server_tracks]
+        return self.playlist
+
+
+def test_apply_skips_existing_and_creates_only_missing_playlist():
+    client = FakeClient()
+    rows = [
+        {"target_playlist": "Рок — RU", "id": 1, "album_id": 10},  # уже в плейлисте
+        {"target_playlist": "Рок — RU", "id": 5, "album_id": 50},  # новый трек в существующий плейлист
+        {"target_playlist": "Поп — INT", "id": 9, "album_id": 90},  # совсем новый плейлист
+    ]
+
+    apply_mod.apply_classification(client, rows, delay=0)
+
+    assert client.created == ["Поп — INT"]
+    assert client.inserted == [(100, 5, 50, 5), (200, 9, 90, 1)]
+
+
+def test_apply_does_not_duplicate_track_on_timeout_after_server_applied_write():
+    client = FlakyClient()
+    rows = [{"target_playlist": "Рок — RU", "id": 1, "album_id": 10}]
+
+    apply_mod.apply_classification(client, rows, delay=0)
+
+    assert client.insert_calls == 1, "insert_track не должен вызываться повторно вслепую при таймауте"
+    assert client.server_tracks == [(1, 10)], "трек должен оказаться в плейлисте ровно один раз"