feat(app): Add yandex music status

utils/ym.py

@@ -0,0 +1,145 @@
+import os
+import threading
+import time
+
+from flask import Blueprint, jsonify
+from yandex_music import Client
+from yandex_music.exceptions import YandexMusicError
+
+ym_bp = Blueprint("ym", __name__)
+
+YM_TOKEN = os.environ.get("YM_TOKEN", "").strip()
+HISTORY_TTL = 30
+HISTORY_LIMIT = 30
+
+_history_lock = threading.Lock()
+_history_cache: dict = {"at": 0.0, "payload": None, "error": None}
+_ym_client = None
+_ym_client_lock = threading.Lock()
+
+
+def _get_ym_client():
+    global _ym_client
+    if _ym_client is not None:
+        return _ym_client
+    with _ym_client_lock:
+        if _ym_client is None:
+            _ym_client = Client(YM_TOKEN).init()
+    return _ym_client
+
+
+def _track_url(track_id, album_id):
+    if album_id:
+        return f"https://music.yandex.ru/album/{album_id}/track/{track_id}"
+    return f"https://music.yandex.ru/track/{track_id}"
+
+
+def _serialize_track(track, day, context_title, track_id=None, album_id=None):
+    cover = None
+    try:
+        cover = track.get_cover_url("400x400")
+    except (YandexMusicError, AttributeError, TypeError):
+        cover = None
+
+    return {
+        "id": str(track_id or track.id),
+        "album_id": str(album_id) if album_id else None,
+        "title": track.title or "Без названия",
+        "artists": track.artists_name() if hasattr(track, "artists_name") else [],
+        "cover": cover,
+        "duration_ms": track.duration_ms,
+        "day": day,
+        "context": context_title,
+        "url": _track_url(track_id or track.id, album_id),
+    }
+
+
+def _context_title(ctx):
+    if ctx is None:
+        return None
+    for attr in ("album", "playlist", "artist", "wave"):
+        obj = getattr(ctx, attr, None)
+        title = getattr(obj, "title", None) or getattr(obj, "name", None)
+        if title:
+            return title
+    return None
+
+
+def _fetch_history():
+    client = _get_ym_client()
+    history = client.music_history(full_models_count=HISTORY_LIMIT)
+
+    entries = []
+    pending_ids = []
+
+    for tab in ((history.history_tabs if history else None) or []):
+        day = tab.date
+        for group in tab.items or []:
+            ctx = group.context.data.full_model if group.context and group.context.data else None
+            context_title = _context_title(ctx)
+
+            for item in group.tracks or []:
+                if item.type != "track" or item.data is None:
+                    continue
+                item_id = item.data.item_id
+                track_id = item_id.track_id if item_id else None
+                album_id = item_id.album_id if item_id else None
+                track = item.data.full_model
+
+                if track is not None:
+                    entries.append(_serialize_track(track, day, context_title, track_id, album_id))
+                elif track_id:
+                    pending_ids.append((str(track_id), str(album_id) if album_id else None, day, context_title))
+
+    if pending_ids:
+        ids = [f"{t}:{a}" if a else t for t, a, _, _ in pending_ids[:HISTORY_LIMIT]]
+        try:
+            fetched = client.tracks(ids) or []
+        except YandexMusicError:
+            fetched = []
+        by_id = {}
+        for tr in fetched:
+            if tr is None:
+                continue
+            if isinstance(tr, list):
+                tr = next((x for x in tr if x is not None), None)
+                if tr is None:
+                    continue
+            by_id[str(tr.id)] = tr
+        for track_id, album_id, day, context_title in pending_ids:
+            tr = by_id.get(track_id)
+            if tr is not None:
+                entries.append(_serialize_track(tr, day, context_title, track_id, album_id))
+
+    entries = entries[:HISTORY_LIMIT]
+    return {
+        "ok": True,
+        "updated_at": int(time.time()),
+        "count": len(entries),
+        "tracks": entries,
+    }
+
+
+@ym_bp.route("/api/ym/history")
+def ym_history():
+    if not YM_TOKEN:
+        return jsonify(ok=False, error="YM_TOKEN is not set"), 503
+
+    now = time.monotonic()
+    with _history_lock:
+        if _history_cache["payload"] is not None and now - _history_cache["at"] < HISTORY_TTL:
+            return jsonify(_history_cache["payload"])
+
+        try:
+            payload = _fetch_history()
+        except (YandexMusicError, OSError, ValueError, KeyError, AttributeError) as exc:
+            if _history_cache["payload"] is not None:
+                stale = dict(_history_cache["payload"])
+                stale["stale"] = True
+                return jsonify(stale)
+            _history_cache["error"] = str(exc)
+            return jsonify(ok=False, error=str(exc)), 502
+
+        _history_cache["at"] = now
+        _history_cache["payload"] = payload
+        return jsonify(payload)