feat(app): Add yandex music status
This commit is contained in:
Black_Raven committed
1 parent
76032d9869
commit
f2e433d1e4
8 files changed
+632
-1
No files matched your search
Whitespace-only changes.
@@ -0,0 +1,123 @@
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
import urllib.parse
|
||||
import urllib.request
|
||||
|
||||
from flask import Blueprint, jsonify
|
||||
|
||||
listenbrainz_bp = Blueprint("listenbrainz", __name__)
|
||||
|
||||
LB_USERNAME = os.environ.get("LISTENBRAINZ_USERNAME", "").strip()
|
||||
LB_TOKEN = os.environ.get("LISTENBRAINZ_TOKEN", "").strip()
|
||||
HISTORY_TTL = 30
|
||||
HISTORY_LIMIT = 30
|
||||
API_BASE = "https://api.listenbrainz.org/1"
|
||||
|
||||
_history_lock = threading.Lock()
|
||||
_history_cache: dict = {"at": 0.0, "payload": None, "error": None}
|
||||
|
||||
_YT_ID_RE = re.compile(r"(?:youtu\.be/|v=|shorts/)([\w-]{6,})")
|
||||
|
||||
|
||||
def _get(url):
|
||||
req = urllib.request.Request(url, headers={"Accept": "application/json"})
|
||||
if LB_TOKEN:
|
||||
req.add_header("Authorization", f"Token {LB_TOKEN}")
|
||||
with urllib.request.urlopen(req, timeout=10) as resp:
|
||||
data = json.load(resp)
|
||||
if isinstance(data, dict) and data.get("error"):
|
||||
raise RuntimeError(data["error"])
|
||||
return data
|
||||
|
||||
|
||||
def _cover(meta, origin_url):
|
||||
mbid = (meta.get("mbid_mapping") or {}).get("release_mbid")
|
||||
if mbid:
|
||||
return f"https://coverartarchive.org/release/{mbid}/front-250"
|
||||
|
||||
if origin_url:
|
||||
match = _YT_ID_RE.search(origin_url)
|
||||
if match:
|
||||
return f"https://i.ytimg.com/vi/{match.group(1)}/hqdefault.jpg"
|
||||
return None
|
||||
|
||||
|
||||
def _serialize_listen(item):
|
||||
meta = item.get("track_metadata") or {}
|
||||
additional = meta.get("additional_info") or {}
|
||||
origin_url = additional.get("origin_url")
|
||||
|
||||
listened_at = item.get("listened_at")
|
||||
day = time.strftime("%Y-%m-%d", time.localtime(int(listened_at))) if listened_at else None
|
||||
|
||||
duration = additional.get("duration")
|
||||
|
||||
return {
|
||||
"id": item.get("recording_msid") or f"{meta.get('artist_name')}:{meta.get('track_name')}:{listened_at}",
|
||||
"album_id": None,
|
||||
"title": meta.get("track_name") or "Без названия",
|
||||
"artists": [meta["artist_name"]] if meta.get("artist_name") else [],
|
||||
"cover": _cover(meta, origin_url),
|
||||
"duration_ms": int(duration) * 1000 if duration else None,
|
||||
"day": day,
|
||||
"nowplaying": bool(item.get("playing_now")),
|
||||
"context": meta.get("release_name") or None,
|
||||
"url": origin_url,
|
||||
}
|
||||
|
||||
|
||||
def _fetch_recent():
|
||||
user = urllib.parse.quote(LB_USERNAME)
|
||||
listens = _get(f"{API_BASE}/user/{user}/listens?count={HISTORY_LIMIT}")
|
||||
now = _get(f"{API_BASE}/user/{user}/playing-now")
|
||||
|
||||
entries = []
|
||||
now_payload = now.get("payload") or {}
|
||||
playing = now_payload.get("playing_now")
|
||||
if isinstance(playing, dict):
|
||||
entries.append(_serialize_listen(playing))
|
||||
elif playing is True:
|
||||
for item in now_payload.get("listens") or []:
|
||||
if isinstance(item, dict) and item.get("playing_now"):
|
||||
entries.append(_serialize_listen(item))
|
||||
break
|
||||
|
||||
for item in ((listens.get("payload") or {}).get("listens") or []):
|
||||
if isinstance(item, dict) and not item.get("playing_now"):
|
||||
entries.append(_serialize_listen(item))
|
||||
|
||||
entries = entries[:HISTORY_LIMIT + 1]
|
||||
return {
|
||||
"ok": True,
|
||||
"updated_at": int(time.time()),
|
||||
"count": len(entries),
|
||||
"tracks": entries,
|
||||
}
|
||||
|
||||
|
||||
@listenbrainz_bp.route("/api/listenbrainz/recent")
|
||||
def listenbrainz_recent():
|
||||
if not LB_USERNAME:
|
||||
return jsonify(ok=False, error="LISTENBRAINZ_USERNAME 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_recent()
|
||||
except (RuntimeError, 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)
|
||||
+145
@@ -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)
|
||||
Reference in new issue
Block a user