lastfm integration

Author: cwilvx

app/api/plugins/__init__.py

@@ -2,7 +2,10 @@
 from flask_openapi3 import APIBlueprint
 from pydantic import BaseModel, Field
 from app.api.auth import admin_required
+from app.config import UserConfig
 from app.db.userdata import PluginTable
+from app.plugins.lastfm import LastFmPlugin
+from app.utils.auth import get_current_userid
 
 bp_tag = Tag(name="Plugins", description="Manage plugins")
 api = APIBlueprint("plugins", __name__, url_prefix="/plugins", abp_tags=[bp_tag])
@@ -61,3 +64,40 @@ def update_plugin_settings(body: PluginSettingsBody):
     plugin = PluginTable.get_by_name(plugin)
 
     return {"status": "success", "settings": plugin.settings}
+
+
+class LastFmSessionBody(BaseModel):
+    token: str = Field(description="The token to use to create the session")
+
+
+@api.post("/lastfm/session/create")
+def create_lastfm_session(body: LastFmSessionBody):
+    """
+    Create a Last.fm session
+    """
+    if not body.token:
+        return {"error": "Missing token"}, 400
+
+    lastfm = LastFmPlugin()
+    session_key = lastfm.get_session_key(body.token)
+
+    if session_key:
+        config = UserConfig()
+        current_user = get_current_userid()
+        config.lastfmSessionKeys[str(current_user)] = session_key
+        config.lastfmSessionKeys = config.lastfmSessionKeys
+
+    return {"status": "success", "session_key": session_key}
+
+
+@api.post("/lastfm/session/delete")
+def delete_lastfm_session():
+    """
+    Delete the Last.fm session
+    """
+    config = UserConfig()
+    current_user = get_current_userid()
+    config.lastfmSessionKeys[str(current_user)] = ""
+    config.lastfmSessionKeys = config.lastfmSessionKeys
+
+    return {"status": "success"}

app/api/scrobble/__init__.py

@@ -13,6 +13,7 @@
 from app.models.album import Album
 from app.models.stats import StatItem
 from app.models.track import Track
+from app.plugins.lastfm import LastFmPlugin
 from app.serializers.artist import serialize_for_card
 from app.serializers.album import serialize_for_card as serialize_for_album_card
 from app.serializers.track import serialize_track, serialize_tracks
@@ -97,6 +98,11 @@ def log_track(body: LogTrackBody):
     if track:
         track.increment_playcount(duration, timestamp)
 
+    lastfm = LastFmPlugin()
+
+    if lastfm.enabled:
+        lastfm.scrobble(trackentry.tracks[0], timestamp)
+
     return {"msg": "recorded"}, 201
 
 

app/config.py

@@ -48,6 +48,9 @@ class UserConfig:
 
     # plugins
     enablePlugins: bool = True
+    lastfmApiKey: str = "5e5306fbf3e8e3bc92f039b6c6c4bd4e"
+    lastfmApiSecret: str = "0553005e93f9a4b4819d835182181806"
+    lastfmSessionKeys: dict[str, str] = field(default_factory=dict)
 
     def __post_init__(self):
         """

app/plugins/lastfm.py

@@ -0,0 +1,76 @@
+import requests
+from typing import Any
+from hashlib import md5
+from urllib.parse import quote_plus
+
+from app.config import UserConfig
+from app.models.track import Track
+from app.utils.auth import get_current_userid
+from app.utils.threading import background
+from app.plugins import Plugin, plugin_method
+
+
+class LastFmPlugin(Plugin):
+    def __init__(self):
+        self.config = UserConfig()
+        super().__init__("lastfm", "Last.fm scrobbler")
+        self.set_active(
+            bool(
+                self.config.lastfmApiKey
+                and self.config.lastfmApiSecret
+                and self.config.lastfmSessionKeys.get(str(get_current_userid()))
+            )
+        )
+
+    def get_api_signature(self, data: dict[str, Any]) -> str:
+        params = {k: v for k, v in data.items()}
+
+        signature = "".join(f"{k}{v}" for k, v in sorted(params.items()))
+        signature += self.config.lastfmApiSecret
+
+        return md5(signature.encode("utf-8")).hexdigest()
+
+    def post(self, data: dict[str, Any], useSessionKey: bool = True):
+        url = "http://ws.audioscrobbler.com/2.0/?format=json"
+        data["api_key"] = self.config.lastfmApiKey
+        if useSessionKey:
+            data["sk"] = self.config.lastfmSessionKeys.get(str(get_current_userid()))
+
+        data["api_sig"] = self.get_api_signature(data)
+
+        final_url = (
+            url + "&" + "&".join(f"{k}={quote_plus(str(v))}" for k, v in data.items())
+        )
+
+        return requests.post(final_url)
+
+    def get_session_key(self, token: str):
+        data = {
+            "method": "auth.getSession",
+            "token": token,
+        }
+
+        try:
+            res = self.post(data, useSessionKey=False)
+            return res.json()["session"]["key"]
+        except Exception as e:
+            print("get_session_key error", e)
+            return None
+
+    @plugin_method
+    @background
+    def scrobble(self, track: Track, timestamp: int):
+        print("Last.fm: logging track: ", track.title, "-", track.artists[0]["name"])
+        data = {
+            "method": "track.scrobble",
+            "artist": track.artists[0]["name"],
+            "track": track.title,
+            "timestamp": timestamp,
+            "album": track.album,
+            "albumArtist": track.albumartists[0]["name"],
+        }
+
+        try:
+            self.post(data)
+        except Exception as e:
+            print("scrobble error", e)