path: root/bot.py

from collections import Counter
import os
import sqlite3
import sys
import warnings

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,
        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()

# ==============================================================================
# 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


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, 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:
        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"