diff options
Diffstat (limited to 'bot.py')
| -rw-r--r-- | bot.py | 205 |
1 files changed, 190 insertions, 15 deletions
@@ -1,7 +1,8 @@ -from collections import Counter +from collections import Counter import os import sqlite3 import sys +import warnings import aiohttp import discord @@ -43,10 +44,16 @@ cursor.execute( """ CREATE TABLE IF NOT EXISTS user_tokens ( discord_id TEXT PRIMARY KEY, - refresh_token TEXT NOT NULL + refresh_token TEXT NOT NULL, + time_range TEXT NOT NULL DEFAULT '4w' ) """ ) +# Migrate existing tables that don't yet have the time_range column +try: + cursor.execute("ALTER TABLE user_tokens ADD COLUMN time_range TEXT NOT NULL DEFAULT '4w'") +except sqlite3.OperationalError: + pass # Column already exists conn.commit() # ============================================================================== @@ -81,13 +88,112 @@ def get_spotify_client(discord_id: str) -> spotipy.Spotify | None: return None +def get_user_time_range(discord_id: str) -> str: + """Returns the stored time range string for a user, defaulting to '4w'.""" + cursor.execute( + "SELECT time_range FROM user_tokens WHERE discord_id = ?", + (str(discord_id),), + ) + row = cursor.fetchone() + return row[0] if row else "4w" + + +def parse_time_range(range_str: str) -> tuple[str, str, str]: + """Parses a time range string (e.g. '4w', '30d', '6m', '1y') into + a (spotify_time_range, human_label, tier_note) tuple. + + Spotify only supports three fixed ranges: + short_term ≈ last 4 weeks + medium_term ≈ last 6 months + long_term ≈ several years (Spotify's maximum) + + Mapping: + d (days): 1–28 → short_term, 29–180 → medium_term, >180 → long_term + w (weeks): 1–4 → short_term, 5–26 → medium_term, >26 → long_term + m (months): 1 → short_term, 2–6 → medium_term, >6 → long_term + y (years): any → long_term + """ + range_str = range_str.strip().lower() + if not range_str: + raise ValueError("Empty range string.") + + unit = range_str[-1] + if unit not in ("d", "w", "m", "y"): + raise ValueError(f"Unknown unit '{unit}'. Use d (days), w (weeks), m (months), y (years).") + + try: + value = int(range_str[:-1]) + except ValueError: + raise ValueError(f"Invalid number in range '{range_str}'.") + + if value <= 0: + raise ValueError("Range value must be a positive integer.") + + # Human-readable label + unit_names = {"d": "day", "w": "week", "m": "month", "y": "year"} + unit_name = unit_names[unit] + label = f"{value} {unit_name}{'s' if value != 1 else ''}" + + # Map to Spotify API time range + if unit == "d": + days = value + elif unit == "w": + days = value * 7 + elif unit == "m": + days = value * 30 + else: # y + days = value * 365 + + if days <= 28: + spotify_range = "short_term" + tier_note = "Spotify tier: **~last 4 weeks** (Spotify's shortest window)" + elif days <= 180: + spotify_range = "medium_term" + tier_note = "Spotify tier: **~last 6 months**" + else: + spotify_range = "long_term" + tier_note = "Spotify tier: **~last 1–2 years** (Spotify's maximum — going beyond 6 months always uses this cap)" + + return spotify_range, label, tier_note + + 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 + def get_profile_picture(sp: spotipy.Spotify, user_info: dict = None) -> str | None: + try: + user = user_info or sp.current_user() + if not user or not isinstance(user, dict): + print("[ProfilePic] No user_info returned from Spotify.") + return None + + images = user.get("images") + print(f"[ProfilePic] Private profile images: {images}") + + if isinstance(images, list) and len(images) > 0: + first_image = images[0] + if isinstance(first_image, dict): + url = first_image.get("url") + print(f"[ProfilePic] Resolved from private profile: {url}") + return url + + # Fallback to public profile if images list is empty + if "id" in user: + print(f"[ProfilePic] Private images empty, trying public profile for id={user['id']}") + public_user = sp.user(user["id"]) + public_images = public_user.get("images") + print(f"[ProfilePic] Public profile images: {public_images}") + if isinstance(public_images, list) and len(public_images) > 0: + first_public_image = public_images[0] + if isinstance(first_public_image, dict): + url = first_public_image.get("url") + print(f"[ProfilePic] Resolved from public profile: {url}") + return url + + print("[ProfilePic] No profile picture found in either private or public profile.") + except Exception as e: + print(f"[ProfilePic] Error fetching Spotify profile picture: {e}") + return None @staticmethod def get_top_track(sp: spotipy.Spotify, time_range="short_term") -> dict | None: @@ -301,8 +407,9 @@ async def update_discord_widget(discord_user_id: str, payload: dict) -> bool: 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}") + # Discord returns 204 No Content on success for this endpoint + if resp.status in (200, 204): + print(f"Successfully updated widget for user: {discord_user_id} (HTTP {resp.status})") return True else: response_text = await resp.text() @@ -316,12 +423,18 @@ async def process_user_widget_update(discord_user_id: str) -> bool: if not sp: return False + raw_range = get_user_time_range(discord_user_id) + try: + spotify_range, range_label, _ = parse_time_range(raw_range) + except ValueError: + spotify_range, range_label = "short_term", "4 weeks" + 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") + spotify_username = user_info.get("display_name") or user_info.get("id") if user_info else None + profile_pic = SpotifyAPI.get_profile_picture(sp, user_info) + top_song = SpotifyAPI.get_top_track(sp, time_range=spotify_range) + top_album = SpotifyAPI.get_top_album(sp, time_range=spotify_range) + top_artist = SpotifyAPI.get_top_artist(sp, time_range=spotify_range) last_liked = SpotifyAPI.get_last_liked_track(sp) payload = build_discord_profile_payload( @@ -331,7 +444,7 @@ async def process_user_widget_update(discord_user_id: str) -> bool: top_album=top_album, top_artist=top_artist, current_track=last_liked, - time_range_label="4 weeks", + time_range_label=range_label, ) return await update_discord_widget(discord_user_id, payload) @@ -397,7 +510,9 @@ async def submit_code(interaction: discord.Interaction, link_or_code: str): sp_oauth = get_oauth() try: - token_info = sp_oauth.get_access_token(clean_code, check_cache=False) + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + token_info = sp_oauth.get_access_token(clean_code, check_cache=False) refresh_token = token_info["refresh_token"] discord_id = str(interaction.user.id) @@ -451,5 +566,65 @@ async def update_widget(interaction: discord.Interaction): "❌ Failed to update widget identity via Discord API.", ephemeral=True ) + + name="range", + description="Set the time range for your Spotify stats. Format: 7d, 4w, 6m, 1y", +) +async def set_range(interaction: discord.Interaction, range: str): + await interaction.response.defer(ephemeral=True) + discord_id = str(interaction.user.id) + + # Check the user is linked + cursor.execute("SELECT refresh_token FROM user_tokens WHERE discord_id = ?", (discord_id,)) + if not cursor.fetchone(): + await interaction.followup.send( + "❌ You haven't linked your Spotify account yet! Run `/spotify_login` first.", + ephemeral=True, + ) + return + + # Validate the range string + try: + _, label, tier_note = parse_time_range(range) + except ValueError as e: + await interaction.followup.send( + f"❌ Invalid range `{range}`: {e}\n" + "Use a positive number followed by a unit:\n" + "> `d` — days · `w` — weeks · `m` — months · `y` — years\n" + "**Examples:** `7d` · `4w` · `3m` · `1y`", + ephemeral=True, + ) + return + + # Save to DB + cursor.execute( + "UPDATE user_tokens SET time_range = ? WHERE discord_id = ?", + (range.strip().lower(), discord_id), + ) + conn.commit() + + # Immediately refresh the widget with the new range + success = await process_user_widget_update(discord_id) + + note = ( + "\n\nℹ️ **Note:** Spotify only supports 3 fixed time windows, so your input is mapped to the closest one:\n" + "> `≤4 weeks` → ~last 4 weeks\n" + "> `>4 weeks up to 6 months` → ~last 6 months\n" + "> `>6 months` → ~last 1–2 years *(Spotify's maximum — 1y and 2y will show identical results)*" + ) + + if success: + await interaction.followup.send( + f"✅ Time range set to **{label}** and widget updated!\n{tier_note}{note}", + ephemeral=True, + ) + else: + await interaction.followup.send( + f"✅ Time range set to **{label}**, but failed to push widget update to Discord.\n{tier_note}{note}", + ephemeral=True, + ) + + if __name__ == "__main__": bot.run(DISCORD_BOT_TOKEN)
\ No newline at end of file |