init, todo: fix spotify profile picture fetching

1 files changed, 436 insertions, 0 deletions

diff --git a/bot.py b/bot.py
new file mode 100644
index 0000000..17bed8a
--- /dev/null
+++ b/bot.py
@@ -0,0 +1,436 @@
+from collections import Counter
+import os
+import sqlite3
+import sys
+
+import aiohttp
+import discord
+from discord.ext import commands
+from dotenv import load_dotenv
+import spotipy
+from spotipy.oauth2 import SpotifyOAuth
+from urllib.parse import parse_qs, urlparse
+
+load_dotenv()
+
+REQUIRED_ENV = [
+    "SPOTIFY_CLIENT_ID",
+    "SPOTIFY_CLIENT_SECRET",
+    "SPOTIFY_REDIRECT_URI",
+    "DISCORD_BOT_TOKEN",
+    "DISCORD_APP_ID",
+]
+
+for var in REQUIRED_ENV:
+    if not os.getenv(var):
+        print(f"❌ Error: Missing required environment variable '{var}' in .env file.")
+        sys.exit(1)
+
+SPOTIFY_CLIENT_ID = os.getenv("SPOTIFY_CLIENT_ID")
+SPOTIFY_CLIENT_SECRET = os.getenv("SPOTIFY_CLIENT_SECRET")
+SPOTIFY_REDIRECT_URI = os.getenv("SPOTIFY_REDIRECT_URI", "http://localhost:8888/callback")
+DISCORD_BOT_TOKEN = os.getenv("DISCORD_BOT_TOKEN")
+DISCORD_APP_ID = os.getenv("DISCORD_APP_ID")
+DISCORD_IDENTITY_ID = os.getenv("DISCORD_IDENTITY_ID", "0")
+
+SCOPE = "user-read-private user-top-read user-read-currently-playing user-read-playback-state user-read-recently-played"
+# ==============================================================================
+# 2. DATABASE SETUP
+# ==============================================================================
+conn = sqlite3.connect("tokens.db", check_same_thread=False)
+cursor = conn.cursor()
+cursor.execute(
+    """
+    CREATE TABLE IF NOT EXISTS user_tokens (
+        discord_id TEXT PRIMARY KEY,
+        refresh_token TEXT NOT NULL
+    )
+"""
+)
+conn.commit()
+
+# ==============================================================================
+# 3. SPOTIFY API & AUTHENTICATION HELPERS
+# ==============================================================================
+def get_oauth() -> SpotifyOAuth:
+    return SpotifyOAuth(
+        client_id=SPOTIFY_CLIENT_ID,
+        client_secret=SPOTIFY_CLIENT_SECRET,
+        redirect_uri=SPOTIFY_REDIRECT_URI,
+        scope=SCOPE,
+    )
+
+
+def get_spotify_client(discord_id: str) -> spotipy.Spotify | None:
+    """Retrieves user's refresh token from SQLite and returns an authenticated Spotify client."""
+    cursor.execute(
+        "SELECT refresh_token FROM user_tokens WHERE discord_id = ?",
+        (str(discord_id),),
+    )
+    row = cursor.fetchone()
+
+    if not row:
+        return None
+
+    try:
+        sp_oauth = get_oauth()
+        token_info = sp_oauth.refresh_access_token(row[0])
+        return spotipy.Spotify(auth=token_info["access_token"])
+    except Exception as e:
+        print(f"Error refreshing token for user {discord_id}: {e}")
+        return None
+
+
+class SpotifyAPI:
+
+    @staticmethod
+    def get_profile_picture(sp: spotipy.Spotify) -> str | None:
+        user = sp.current_user()
+        images = user.get("images", [])
+        return images[0]["url"] if images else None
+
+    @staticmethod
+    def get_top_track(sp: spotipy.Spotify, time_range="short_term") -> dict | None:
+        res = sp.current_user_top_tracks(limit=1, time_range=time_range)
+        items = res.get("items", [])
+        if not items:
+            return None
+
+        track = items[0]
+        return {
+            "title": track["name"],
+            "artist": track["artists"][0]["name"],
+            "album_art": track["album"]["images"][0]["url"]
+            if track["album"]["images"]
+            else None,
+            "url": track["external_urls"]["spotify"],
+        }
+
+    @staticmethod
+    def get_top_artist(sp: spotipy.Spotify, time_range="short_term") -> dict | None:
+        res = sp.current_user_top_artists(limit=1, time_range=time_range)
+        items = res.get("items", [])
+        if not items:
+            return None
+
+        artist = items[0]
+        return {
+            "name": artist["name"],
+            "image_url": artist["images"][0]["url"] if artist["images"] else None,
+            "url": artist["external_urls"]["spotify"],
+        }
+
+    @staticmethod
+    def get_top_album(sp: spotipy.Spotify, time_range="short_term") -> dict | None:
+        res = sp.current_user_top_tracks(limit=50, time_range=time_range)
+        tracks = res.get("items", [])
+        if not tracks:
+            return None
+
+        album_counts = Counter(t["album"]["id"] for t in tracks)
+        top_album_id = album_counts.most_common(1)[0][0]
+
+        for t in tracks:
+            if t["album"]["id"] == top_album_id:
+                album = t["album"]
+                return {
+                    "title": album["name"],
+                    "artist": album["artists"][0]["name"],
+                    "cover_art": album["images"][0]["url"]
+                    if album["images"]
+                    else None,
+                    "url": album["external_urls"]["spotify"],
+                }
+        return None
+
+    @staticmethod
+    def get_currently_playing(sp: spotipy.Spotify) -> dict:
+        playback = sp.current_playback()
+        if playback and playback.get("is_playing") and playback.get("item"):
+            item = playback["item"]
+            return {
+                "status": "playing",
+                "is_playing": True,
+                "title": item["name"],
+                "artist": item["artists"][0]["name"],
+                "album_art": item["album"]["images"][0]["url"]
+                if item["album"]["images"]
+                else None,
+                "url": item["external_urls"]["spotify"],
+            }
+
+        recently_played = sp.current_user_recently_played(limit=1)
+        items = recently_played.get("items", [])
+
+        if items:
+            track = items[0]["track"]
+            return {
+                "status": "recently_played",
+                "is_playing": False,
+                "title": track["name"],
+                "artist": track["artists"][0]["name"],
+                "album_art": track["album"]["images"][0]["url"]
+                if track["album"]["images"]
+                else None,
+                "url": track["external_urls"]["spotify"],
+            }
+
+        return {
+            "status": "offline",
+            "is_playing": False,
+            "title": "No listening history",
+            "artist": "N/A",
+            "album_art": None,
+            "url": None,
+        }
+
+
+# ==============================================================================
+# 4. DISCORD PAYLOAD BUILDER & DISCORD IDENTITY API UPDATER
+# ==============================================================================
+def build_discord_profile_payload(
+    spotify_username: str,
+    spotify_profile_pic: str | None,
+    top_song: dict | None,
+    top_album: dict | None,
+    top_artist: dict | None,
+    current_track: dict,
+    time_range_label: str = "4 weeks",
+) -> dict:
+    DEFAULT_IMAGE = "https://cdn.discordapp.com/embed/avatars/0.png"
+    status_label = (
+        "Currently Playing" if current_track.get("is_playing") else "Last Played"
+    )
+
+    return {
+        "data": {
+            "dynamic": [
+                {
+                    "type": 3,
+                    "name": "spotify_profile_picture",
+                    "value": {"url": spotify_profile_pic or DEFAULT_IMAGE},
+                },
+                {
+                    "type": 1,
+                    "name": "spotfy_user_name",
+                    "value": spotify_username or "Unknown User",
+                },
+                {
+                    "type": 3,
+                    "name": "most_listened_song_picture",
+                    "value": {
+                        "url": top_song.get("album_art") if top_song else DEFAULT_IMAGE
+                    },
+                },
+                {
+                    "type": 1,
+                    "name": "most_listened_song_name",
+                    "value": top_song.get("title") if top_song else "None",
+                },
+                {
+                    "type": 3,
+                    "name": "most_listened_album_picture",
+                    "value": {
+                        "url": top_album.get("cover_art") if top_album else DEFAULT_IMAGE
+                    },
+                },
+                {
+                    "type": 1,
+                    "name": "most_listened_album_name",
+                    "value": top_album.get("title") if top_album else "None",
+                },
+                {
+                    "type": 3,
+                    "name": "most_listened_artist_picture",
+                    "value": {
+                        "url": top_artist.get("image_url")
+                        if top_artist
+                        else DEFAULT_IMAGE
+                    },
+                },
+                {
+                    "type": 1,
+                    "name": "most_listened_artist_name",
+                    "value": top_artist.get("name") if top_artist else "None",
+                },
+                {
+                    "type": 3,
+                    "name": "current_song_picture",
+                    "value": {
+                        "url": current_track.get("album_art") or DEFAULT_IMAGE
+                    },
+                },
+                {
+                    "type": 1,
+                    "name": "current_song_name",
+                    "value": current_track.get("title", "None"),
+                },
+                {"type": 1, "name": "status", "value": status_label},
+                {"type": 1, "name": "range", "value": time_range_label},
+            ]
+        }
+    }
+
+
+async def update_discord_widget(discord_user_id: str, payload: dict) -> bool:
+    url = f"https://discord.com/api/v9/applications/{DISCORD_APP_ID}/users/{discord_user_id}/identities/0/profile"
+    headers = {
+        "Content-Type": "application/json",
+        "Authorization": f"Bot {DISCORD_BOT_TOKEN}",
+        "User-Agent": "DiscordBot (https://github.com/discord/discord-api-docs, 1.0.0)",
+    }
+
+    async with aiohttp.ClientSession() as session:
+        async with session.patch(url, headers=headers, json=payload) as resp:
+            if resp.status == 200:
+                print(f"Successfully updated widget for user: {discord_user_id}")
+                return True
+            else:
+                response_text = await resp.text()
+                print(f"Failed to update ({resp.status}): {response_text}")
+                return False
+
+
+async def process_user_widget_update(discord_user_id: str) -> bool:
+    """Helper function to orchestrate fetching Spotify stats and pushing to Discord."""
+    sp = get_spotify_client(discord_user_id)
+    if not sp:
+        return False
+
+    user_info = sp.current_user()
+    spotify_username = user_info.get("display_name") or user_info.get("id")
+    profile_pic = SpotifyAPI.get_profile_picture(sp)
+    top_song = SpotifyAPI.get_top_track(sp, time_range="short_term")
+    top_album = SpotifyAPI.get_top_album(sp, time_range="short_term")
+    top_artist = SpotifyAPI.get_top_artist(sp, time_range="short_term")
+    current_track = SpotifyAPI.get_currently_playing(sp)
+
+    payload = build_discord_profile_payload(
+        spotify_username=spotify_username,
+        spotify_profile_pic=profile_pic,
+        top_song=top_song,
+        top_album=top_album,
+        top_artist=top_artist,
+        current_track=current_track,
+        time_range_label="4 weeks",
+    )
+
+    return await update_discord_widget(discord_user_id, payload)
+
+
+# ==============================================================================
+# 5. DISCORD BOT COMMANDS SETUP
+# ==============================================================================
+bot = commands.Bot(command_prefix="!", intents=discord.Intents.default())
+
+
[email protected]
+async def on_ready():
+    await bot.tree.sync()
+    print(f"Logged in as {bot.user} (ID: {bot.user.id})")
+
+
[email protected](
+    name="spotify_login",
+    description="Get link to connect your Spotify account",
+)
+async def spotify_login(interaction: discord.Interaction):
+    sp_oauth = get_oauth()
+    auth_url = sp_oauth.get_authorize_url()
+
+    msg = (
+        f"1️⃣ [Click here to log into Spotify]({auth_url})\n"
+        f"2️⃣ After logging in, you'll reach a page that fails to load (e.g. `localhost`).\n"
+        f"3️⃣ Copy the long **`code`** parameter from the URL bar (e.g. `?code=ABC123...`).\n"
+        f"4️⃣ Run `/submit_code code:<YOUR_CODE>` to link your account!"
+    )
+    await interaction.response.send_message(msg, ephemeral=True)
+
+
+def extract_code(input_text: str) -> str:
+    """Extracts the 'code' parameter from a full Spotify callback URL, query string, or raw input."""
+    input_text = input_text.strip()
+
+    if "code=" in input_text:
+        parsed_url = urlparse(input_text)
+        query_params = parse_qs(parsed_url.query or parsed_url.path)
+
+        if "code" in query_params:
+            return query_params["code"][0]
+
+        try:
+            return input_text.split("code=")[1].split("&")[0]
+        except IndexingError:
+            pass
+
+    return input_text
+
+
[email protected](
+    name="submit_code",
+    description="Submit your Spotify auth code or full callback URL",
+)
+async def submit_code(interaction: discord.Interaction, link_or_code: str):
+    await interaction.response.defer(ephemeral=True)
+
+    # Automatically extract the actual code string
+    clean_code = extract_code(link_or_code)
+
+    sp_oauth = get_oauth()
+    try:
+        token_info = sp_oauth.get_access_token(clean_code, check_cache=False)
+        refresh_token = token_info["refresh_token"]
+
+        discord_id = str(interaction.user.id)
+        cursor.execute(
+            """
+            INSERT INTO user_tokens (discord_id, refresh_token) VALUES (?, ?)
+            ON CONFLICT(discord_id) DO UPDATE SET refresh_token=excluded.refresh_token
+        """,
+            (discord_id, refresh_token),
+        )
+        conn.commit()
+
+        # Update widget immediately upon linking
+        success = await process_user_widget_update(discord_id)
+
+        if success:
+            await interaction.followup.send(
+                "✅ Spotify account linked and widget updated successfully!",
+                ephemeral=True,
+            )
+        else:
+            await interaction.followup.send(
+                "✅ Account linked, but failed to push initial widget update to Discord.",
+                ephemeral=True,
+            )
+
+    except Exception as e:
+        await interaction.followup.send(
+            f"❌ Invalid link/code or error linking account: {e}", ephemeral=True
+        )
+
+
[email protected](name="update_widget", description="Force update your Spotify widget payload")
+async def update_widget(interaction: discord.Interaction):
+    await interaction.response.defer(ephemeral=True)
+    discord_id = str(interaction.user.id)
+
+    sp = get_spotify_client(discord_id)
+    if not sp:
+        await interaction.followup.send(
+            "❌ You haven't linked your Spotify account yet! Run `/spotify_login` first.",
+            ephemeral=True,
+        )
+        return
+
+    success = await process_user_widget_update(discord_id)
+    if success:
+        await interaction.followup.send("✅ Widget successfully updated!", ephemeral=True)
+    else:
+        await interaction.followup.send(
+            "❌ Failed to update widget identity via Discord API.", ephemeral=True
+        )
+
+if __name__ == "__main__":
+    bot.run(DISCORD_BOT_TOKEN)
\ No newline at end of file