diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml new file mode 100644 index 0000000..7f8d307 --- /dev/null +++ b/.github/workflows/build.yml @@ -0,0 +1,46 @@ +name: Build + +on: + workflow_call: + workflow_dispatch: + push: + branches: [main, master] + paths-ignore: ["README.md", "LICENSE", ".gitignore"] + pull_request: + branches: [main, master] + +permissions: + contents: read + +jobs: + build: + strategy: + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + fail-fast: false + + runs-on: ${{ matrix.os }} + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: "3.12" + + - name: Install dependencies + run: | + python -m pip install --upgrade pip + pip install -r requirements.txt + pip install pyinstaller + + - name: Build with PyInstaller + run: python bundle/pyinstaller.py + + - name: Upload artifact + uses: actions/upload-artifact@v4 + with: + name: SpotFetch-${{ runner.os }} + path: dist/SpotFetch* + if-no-files-found: error diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml new file mode 100644 index 0000000..4379b21 --- /dev/null +++ b/.github/workflows/release.yml @@ -0,0 +1,35 @@ +name: Release + +on: + push: + tags: ["v*"] + +permissions: + contents: write + +jobs: + build: + uses: ./.github/workflows/build.yml + + release: + needs: build + runs-on: ubuntu-latest + permissions: + contents: write + + steps: + - uses: actions/checkout@v4 + + - name: Download all artifacts + uses: actions/download-artifact@v4 + with: + path: artifacts + + - name: List artifacts + run: ls -R artifacts/ + + - name: Create Release + uses: softprops/action-gh-release@v2 + with: + files: artifacts/**/* + generate_release_notes: true diff --git a/.gitignore b/.gitignore index c795a7f..de6aeee 100644 --- a/.gitignore +++ b/.gitignore @@ -4,3 +4,7 @@ __pycache__/ .vscode/ *.pyc cli_to_code.py +dist/ +build/ +*.spec +*.egg-info/ diff --git a/README.md b/README.md index db810aa..3be21f7 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ A simple python program to download Music from various platfroms using yt-dlp ( ## What it can do : -- Download Spotify playlists after exporting the playlist as a csv file from [Exportify](https://exportify.app) +- Download Spotify playlists - batch download music from a .txt file with URLs one by line, or using a custom CSV file with headers *name,artist* - Direct download from a Youtube url, can be a video or playlist. - Search then download a song using its name and artist name. @@ -19,13 +19,25 @@ A simple python program to download Music from various platfroms using yt-dlp ( # Installation : -### Requirements : +### Option 1: Standalone executable (recommended) + +Download the latest `SpotFetch.exe` (Windows), `SpotFetch_macos` (macOS), or `SpotFetch_linux` (Linux) from the [Releases page](https://github.com/MrElyazid/SpotFetch/releases/latest). + +#### Requirements : - First make sure you have ffmpeg installed on your machine [Download here](https://ffmpeg.org/download.html). -- Make sure you have [Python](https://www.python.org/downloads/) installed too, and is in your system path. + +That's it — just run the downloaded file. + +### Option 2: Run from source with Python + +#### Requirements : + +- Make sure you have [Python](https://www.python.org/downloads/) installed and in your system path. +- Make sure you have ffmpeg installed on your machine [Download here](https://ffmpeg.org/download.html). - If you dont have Git to clone the repo thats fine, you can download it as a zip file and uncompress it, see [here](.github/if_no_git.png). -### setup : +#### setup : - Clone this repository ( or just download it as a zip file and uncompress it ): @@ -58,6 +70,16 @@ It is recommended to use a virtual environement since installing requirements gl before running `pip install -r requirements.txt` make sure you [create](https://docs.python.org/3/library/venv.html#creating-virtual-environments) and then [activate](https://docs.python.org/3/library/venv.html#how-venvs-work) a venv, and if on Windows and running powershell you might need to run `Set-ExecutionPolicy -ExecutionPolicy RemoteSigned -Scope CurrentUser` before activating the venv. +### Spotify API Setup (for direct Spotify downloads) : + +SpotFetch can download playlists and liked songs directly from Spotify. +A built-in Client ID is provided, so no account setup is needed. + +On first use, SpotFetch will open a browser window for you to log in to Spotify and authorize the app. +Your session is saved for future use (no need to log in again). + +To use your own Spotify app instead (optional), set the `SPOTIFY_CLIENT_ID` environment variable. + ## Some details : diff --git a/bundle/pyinstaller.py b/bundle/pyinstaller.py new file mode 100644 index 0000000..b2b3dca --- /dev/null +++ b/bundle/pyinstaller.py @@ -0,0 +1,52 @@ +#!/usr/bin/env python3 +import os +import sys +import platform + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from PyInstaller.__main__ import run as run_pyinstaller + +NAME = "SpotFetch" +ENTRY_POINT = "menu.py" + +OS_NAME = sys.platform +MACHINE = platform.machine().lower() + +if OS_NAME == "win32": + SUFFIX = "" + EXT = ".exe" +elif OS_NAME == "darwin": + SUFFIX = "_macos" + EXT = "" +else: + SUFFIX = "_linux" + EXT = "" + +out_name = f"{NAME}{SUFFIX}{EXT}" +dist_path = os.path.join("dist", out_name) + + +def main(): + opts = [ + f"--name={NAME}", + "--noconfirm", + "--onefile", + "--console", + "--add-data=README.md:.", + *sys.argv[1:], + ENTRY_POINT, + ] + + print(f"Building {out_name} for {sys.platform} {MACHINE}") + print(f"Destination: {dist_path}") + print(f"PyInstaller args: {opts}\n") + + run_pyinstaller(opts) + + final = dist_path if os.path.exists(dist_path) else f"dist/{NAME}{EXT}" + print(f"\nDone: {final}") + + +if __name__ == "__main__": + main() diff --git a/bundle/requirements/pyinstaller.txt b/bundle/requirements/pyinstaller.txt new file mode 100644 index 0000000..c45a45b --- /dev/null +++ b/bundle/requirements/pyinstaller.txt @@ -0,0 +1 @@ +pyinstaller>=6.0 diff --git a/menu.py b/menu.py index fed0948..1a788da 100644 --- a/menu.py +++ b/menu.py @@ -3,6 +3,8 @@ import functions import os import sys +import spotify_auth +import spotify_api from rich.console import Console from rich.panel import Panel from rich.text import Text @@ -18,7 +20,8 @@ "output_path": ".", "cookie_file": None, "platform": "ytmusic", - "tolerance": 2 + "tolerance": 2, + "spotify_client_id": os.environ.get("SPOTIFY_CLIENT_ID"), } art = r""" @@ -67,6 +70,12 @@ def show_current_settings(): ("\n"), ("Duration Tolerance: ", "white"), (f"{settings['tolerance']}min", "cyan"), + ("\n"), + ("Spotify API: ", "white"), + ( + f"{'Configured' if settings['spotify_client_id'] else 'Not configured (set in Settings)'}", + "cyan", + ), ) panel = Panel( @@ -96,8 +105,13 @@ def configure_settings(): "Set duration tolerance", f"Currently tolerance={settings['tolerance']}min", ), - ("6", "Reset to Defaults", "Reset all settings"), - ("7", "Back to Main Menu", "Return to main menu"), + ( + "6", + "Configure Spotify API", + f"Client ID: {settings['spotify_client_id'] or 'Not set'}", + ), + ("7", "Reset to Defaults", "Reset all settings"), + ("8", "Back to Main Menu", "Return to main menu"), ] table = Table(title="Settings Menu", box=box.ROUNDED, title_style="bold cyan") @@ -113,8 +127,8 @@ def configure_settings(): choice = Prompt.ask( "Select setting to configure", - choices=["1", "2", "3", "4", "5", "6", "7"], - default="7", + choices=["1", "2", "3", "4", "5", "6", "7", "8"], + default="8", ) if choice == "1": @@ -128,11 +142,13 @@ def configure_settings(): elif choice == "5": set_duration_tolerance() elif choice == "6": - reset_settings() + configure_spotify_settings() elif choice == "7": + reset_settings() + elif choice == "8": return - if choice != "7": + if choice != "8": console.print() show_current_settings() if Confirm.ask("\nConfigure another setting?", default=False): @@ -141,11 +157,11 @@ def configure_settings(): def set_duration_tolerance(): - """Set the duration tolerance when downloading using exportify""" - console.print(Panel("Set Download Platform", style="bold yellow")) + """Set the duration tolerance for song matching""" + console.print(Panel("Set Duration Tolerance", style="bold yellow")) console.print( - "This duration is used to detect wrong song matches when searching youtube for the song in the exportify csv\n" + "This duration is used to detect wrong song matches when searching youtube for a song\n" "Set it to something like 1 (tight checking) or 3 (good to ignore long unwanted songs) in minutes" ) @@ -496,6 +512,278 @@ def download_spotify_songs_from_list(songs, platform): ) +def get_spotify_access_token(): + """Get a valid Spotify access token. Returns None if not configured.""" + client_id = settings.get("spotify_client_id") or os.environ.get("SPOTIFY_CLIENT_ID") or spotify_auth.DEFAULT_CLIENT_ID + if not client_id: + console.print("[yellow]Spotify Client ID is not configured.[/yellow]") + console.print("Set the SPOTIFY_CLIENT_ID environment variable, or configure it in Settings.") + return None + + auth = spotify_auth.SpotifyAuth(client_id) + if auth.is_authenticated(): + token = auth.get_access_token() + if token: + return token + console.print("[yellow]Spotify session expired. Re-authenticating...[/yellow]") + + try: + token = auth.authenticate() + return token + except spotify_auth.SpotifyAuthError as e: + console.print(f"[red]{e}[/red]") + return None + + +def process_spotify_playlist(): + """Download tracks from a Spotify playlist URL.""" + console.clear() + show_banner() + show_current_settings() + + console.print(Panel("Download Spotify Playlist", style="bold blue")) + console.print( + "Paste a Spotify playlist URL, e.g.:\n" + " https://open.spotify.com/playlist/37i9dQZF1DXcBWIGoYBM5M", + style="italic white", + ) + + url = Prompt.ask("Enter playlist URL").strip() + if not url: + return + + try: + playlist_id = spotify_api.extract_playlist_id(url) + except ValueError as e: + console.print(f"[red]{e}[/red]") + Prompt.ask("\nPress Enter to continue...") + return + + token = get_spotify_access_token() + if not token: + Prompt.ask("\nPress Enter to continue...") + return + + console.print("Fetching playlist info...", style="yellow") + try: + playlist_name = spotify_api.fetch_playlist_name(playlist_id, token) + console.print(f"Playlist: [cyan]{playlist_name}[/cyan]") + except Exception: + playlist_name = "Playlist" + + console.print("Fetching tracks...", style="yellow") + try: + raw_tracks = spotify_api.fetch_playlist_tracks(playlist_id, token) + except Exception as e: + console.print(f"[red]Failed to fetch tracks: {e}[/red]") + Prompt.ask("\nPress Enter to continue...") + return + + if not raw_tracks: + console.print("[yellow]No tracks found in this playlist.[/yellow]") + Prompt.ask("\nPress Enter to continue...") + return + + songs = [spotify_api.track_to_metadata(t) for t in raw_tracks] + console.print(f"Found [cyan]{len(songs)}[/cyan] tracks. Starting download...\n") + + download_spotify_songs_from_list(songs, settings["platform"]) + Prompt.ask("\nPress Enter to continue...") + + +def process_spotify_liked_songs(): + """Download the user's Liked Songs from Spotify.""" + console.clear() + show_banner() + show_current_settings() + + console.print(Panel("Download Liked Songs", style="bold blue")) + console.print("This will download all your saved/liked songs from Spotify.", style="italic white") + + if not Confirm.ask("Continue?"): + return + + token = get_spotify_access_token() + if not token: + Prompt.ask("\nPress Enter to continue...") + return + + console.print("Fetching your Liked Songs...", style="yellow") + try: + raw_tracks = spotify_api.fetch_liked_songs(token) + except Exception as e: + console.print(f"[red]Failed to fetch liked songs: {e}[/red]") + Prompt.ask("\nPress Enter to continue...") + return + + if not raw_tracks: + console.print("[yellow]No liked songs found.[/yellow]") + Prompt.ask("\nPress Enter to continue...") + return + + songs = [spotify_api.track_to_metadata(t) for t in raw_tracks] + console.print(f"Found [cyan]{len(songs)}[/cyan] liked songs. Starting download...\n") + + download_spotify_songs_from_list(songs, settings["platform"]) + Prompt.ask("\nPress Enter to continue...") + + +def process_spotify_list_playlists(): + """List user's playlists and let them pick one to download.""" + token = get_spotify_access_token() + if not token: + Prompt.ask("\nPress Enter to continue...") + return + + console.print("Fetching your playlists...", style="yellow") + try: + playlists = spotify_api.fetch_user_playlists(token) + except Exception as e: + console.print(f"[red]Failed to fetch playlists: {e}[/red]") + Prompt.ask("\nPress Enter to continue...") + return + + if not playlists: + console.print("[yellow]No playlists found on your account.[/yellow]") + Prompt.ask("\nPress Enter to continue...") + return + + while True: + console.clear() + show_banner() + console.print(Panel("Your Spotify Playlists", style="bold green")) + + table = Table(box=box.ROUNDED) + table.add_column("#", style="cyan", justify="right", width=4) + table.add_column("Playlist", style="yellow") + table.add_column("Tracks", style="white", justify="right") + table.add_column("Visibility", style="white") + + for i, pl in enumerate(playlists, 1): + vis = "Public" if pl['public'] else "Private" + table.add_row(str(i), pl['name'], str(pl['tracks_total']), vis) + + console.print(table) + console.print() + + choice = Prompt.ask( + "Enter number to download, or 0 to go back", + default="0", + ) + + try: + idx = int(choice) + except ValueError: + continue + + if idx == 0: + return + + if 1 <= idx <= len(playlists): + selected = playlists[idx - 1] + console.print(f"\nSelected: [cyan]{selected['name']}[/cyan]") + console.print("Fetching tracks...", style="yellow") + + try: + raw_tracks = spotify_api.fetch_playlist_tracks(selected['id'], token) + except Exception as e: + console.print(f"[red]Failed to fetch tracks: {e}[/red]") + Prompt.ask("\nPress Enter to continue...") + return + + if not raw_tracks: + console.print("[yellow]No tracks found.[/yellow]") + Prompt.ask("\nPress Enter to continue...") + return + + songs = [spotify_api.track_to_metadata(t) for t in raw_tracks] + console.print(f"Found [cyan]{len(songs)}[/cyan] tracks. Starting download...\n") + download_spotify_songs_from_list(songs, settings["platform"]) + Prompt.ask("\nPress Enter to continue...") + return + + +def spotify_submenu(): + """Spotify API operations submenu.""" + while True: + console.clear() + show_banner() + show_current_settings() + + console.print(Panel("Spotify Download Options", style="bold green")) + + table = Table(title="Spotify Menu", box=box.ROUNDED, title_style="bold cyan") + table.add_column("Option", style="cyan", justify="center", width=8) + table.add_column("Action", style="yellow", width=25) + table.add_column("Description", style="white") + + options = [ + ("1", "List My Playlists", "Choose from your saved playlists"), + ("2", "Download Liked Songs", "Download all your saved/liked songs"), + ("3", "Download a Playlist", "Paste a Spotify playlist URL"), + ("4", "Log Out of Spotify", "Clear stored authentication tokens"), + ("5", "Back to Main Menu", "Return to the main menu"), + ] + + for opt, action, desc in options: + table.add_row(opt, action, desc) + + console.print(table) + console.print() + + choice = Prompt.ask( + "Select an option", choices=["1", "2", "3", "4", "5"], default="1" + ) + + if choice == "1": + process_spotify_list_playlists() + elif choice == "2": + process_spotify_liked_songs() + elif choice == "3": + process_spotify_playlist() + elif choice == "4": + client_id = settings.get("spotify_client_id") or os.environ.get("SPOTIFY_CLIENT_ID") + if client_id: + auth = spotify_auth.SpotifyAuth(client_id) + auth.clear_tokens() + console.print("[green]Logged out of Spotify.[/green]") + else: + console.print("[yellow]No Spotify session to log out from.[/yellow]") + Prompt.ask("\nPress Enter to continue...") + elif choice == "5": + return + + +def configure_spotify_settings(): + """Configure Spotify API settings.""" + console.clear() + show_banner() + console.print(Panel("Configure Spotify API", style="bold yellow")) + + console.print( + "1. Go to https://developer.spotify.com/dashboard/\n" + "2. Create an App with redirect URI http://127.0.0.1:3000/\n" + "3. Copy the Client ID and paste it below\n\n" + "You can also set the SPOTIFY_CLIENT_ID environment variable.\n", + style="white", + ) + + current = settings.get("spotify_client_id") or "" + console.print(f"Current Client ID: [cyan]{current or 'Not set'}[/cyan]") + + new_id = Prompt.ask("Enter Spotify Client ID (or leave empty to keep current)", default="") + + if new_id.strip(): + settings["spotify_client_id"] = new_id.strip() + console.print("[green]Spotify Client ID updated![/green]") + elif not current: + console.print("[yellow]No Client ID set. Spotify features will not work until configured.[/yellow]") + else: + console.print("[yellow]Client ID unchanged.[/yellow]") + + Prompt.ask("\nPress Enter to continue...") + + def main_menu(): """Main application menu""" while True: @@ -507,35 +795,40 @@ def main_menu(): menu_options = [ ( "1", + "Download from Spotify (via API)", + "Download playlists or liked songs directly from Spotify", + ), + ( + "2", "Download using Exportify CSV", "Export your playlist csv here : https://exportify.app/", ), ( - "2", + "3", "Download using TuneMyMusic CSV", "Export your playlist csv here : https://www.tunemymusic.com/transfer (make sure you export to a file!)", ), ( - "3", + "4", "Download from URLs File", "Batch download from text file with YouTube URLs one by line.", ), ( - "4", + "5", "Download from Custom CSV", "Download from CSV with name,artist as headers", ), ( - "5", + "6", "Download from Single URL", "Download audio from a direct URL ( can be a YT video url or playlist )", ), ( - "6", + "7", "Settings", - "Configure format (MP3/FLAC/M4A), output directory, and cookies", + "Configure format (MP3/FLAC/M4A), output directory, and Spotify API", ), - ("7", "Exit", "Exit the application"), + ("8", "Exit", "Exit the application"), ] table = Table( @@ -552,22 +845,24 @@ def main_menu(): console.print() choice = Prompt.ask( - "Select an option", choices=[str(i) for i in range(1, 8)], default="1" + "Select an option", choices=[str(i) for i in range(1, 9)], default="1" ) if choice == "1": - process_exportify_csv() + spotify_submenu() elif choice == "2": - process_tunemymusic_csv() + process_exportify_csv() elif choice == "3": - download_from_urls_file() + process_tunemymusic_csv() elif choice == "4": - download_from_custom_csv() + download_from_urls_file() elif choice == "5": - download_single_url() + download_from_custom_csv() elif choice == "6": - configure_settings() + download_single_url() elif choice == "7": + configure_settings() + elif choice == "8": console.print("\nThank you for using SpotFetch!", style="bold cyan") console.print("Bye Bye!!", style="bold yellow") sys.exit(0) diff --git a/spotify_api.py b/spotify_api.py new file mode 100644 index 0000000..9eebf78 --- /dev/null +++ b/spotify_api.py @@ -0,0 +1,130 @@ +import re +import requests +from functions import sanitize_string + + +class SpotifyAPIError(Exception): + pass + + +PLAYLIST_URL_PATTERN = re.compile(r'playlist/([a-zA-Z0-9]+)') + + +def extract_playlist_id(url): + match = PLAYLIST_URL_PATTERN.search(url) + if match: + return match.group(1) + raise ValueError(f"Could not extract playlist ID from URL: {url}") + + +def _check_response(resp): + if resp.status_code == 429: + raise SpotifyAPIError("Spotify API rate limit exceeded. Try again later.") + if not resp.ok: + detail = resp.text + raise SpotifyAPIError( + f"{resp.status_code} {resp.reason}: {detail}" + ) + + +def fetch_playlist_name(playlist_id, access_token): + url = f"https://api.spotify.com/v1/playlists/{playlist_id}" + headers = {'Authorization': f'Bearer {access_token}'} + resp = requests.get(url, headers=headers, timeout=30) + _check_response(resp) + data = resp.json() + return data.get('name', 'Unknown Playlist') + + +def fetch_playlist_tracks(playlist_id, access_token): + tracks = [] + url = f"https://api.spotify.com/v1/playlists/{playlist_id}/tracks" + headers = {'Authorization': f'Bearer {access_token}'} + + while url: + resp = requests.get(url, headers=headers, timeout=30) + _check_response(resp) + data = resp.json() + + for item in data.get('items', []): + track = item.get('track') + if track and track.get('id'): + tracks.append(track) + + url = data.get('next') + + return tracks + + +def fetch_user_id(access_token): + url = "https://api.spotify.com/v1/me" + headers = {'Authorization': f'Bearer {access_token}'} + resp = requests.get(url, headers=headers, timeout=30) + _check_response(resp) + return resp.json()['id'] + + +def fetch_user_playlists(access_token): + playlists = [] + url = "https://api.spotify.com/v1/me/playlists?limit=50" + headers = {'Authorization': f'Bearer {access_token}'} + + while url: + resp = requests.get(url, headers=headers, timeout=30) + _check_response(resp) + data = resp.json() + + for item in data.get('items', []): + playlists.append({ + 'id': item['id'], + 'name': item['name'], + 'tracks_total': item['tracks']['total'], + 'public': item['public'], + 'owner': item['owner']['display_name'], + }) + + url = data.get('next') + + return playlists + + +def fetch_liked_songs(access_token): + tracks = [] + url = "https://api.spotify.com/v1/me/tracks?limit=50" + headers = {'Authorization': f'Bearer {access_token}'} + + while url: + resp = requests.get(url, headers=headers, timeout=30) + _check_response(resp) + data = resp.json() + + for item in data.get('items', []): + track = item.get('track') + if track and track.get('id'): + tracks.append(track) + + url = data.get('next') + + return tracks + + +def track_to_metadata(spotify_track): + return { + 'track_name': sanitize_string(spotify_track.get('name', 'Unknown Track')), + 'artist_names': [ + sanitize_string(a.get('name', 'Unknown Artist')) + for a in spotify_track.get('artists', []) + ], + 'album_name': sanitize_string( + spotify_track.get('album', {}).get('name', 'Unknown Album') + ), + 'album_artist_names': [ + sanitize_string(a.get('name', 'Unknown Artist')) + for a in spotify_track.get('album', {}).get('artists', []) + ], + 'track_duration_ms': spotify_track.get('duration_ms', 0), + 'album_release_date': sanitize_string( + spotify_track.get('album', {}).get('release_date', '') + ), + 'genres': [], + } diff --git a/spotify_auth.py b/spotify_auth.py new file mode 100644 index 0000000..b820ac6 --- /dev/null +++ b/spotify_auth.py @@ -0,0 +1,261 @@ +import json +import os +import hashlib +import base64 +import secrets +import urllib.parse +import webbrowser +from http.server import HTTPServer, BaseHTTPRequestHandler +import threading +import requests + +CONFIG_DIR = os.path.expanduser("~/.config/spotfetch") +TOKEN_FILE = os.path.join(CONFIG_DIR, "tokens.json") + +DEFAULT_CLIENT_ID = "d9739ff737b348928da15234f91cf697" + +SCOPES = "playlist-read-private playlist-read-collaborative user-library-read" +REDIRECT_PORT = 3000 +REDIRECT_URI = f"http://127.0.0.1:{REDIRECT_PORT}/" + + +class SpotifyAuthError(Exception): + pass + + +class _AuthHandler(BaseHTTPRequestHandler): + auth_code = None + code_verifier = None + auth_event = None + + def do_GET(self): + parsed = urllib.parse.urlparse(self.path) + params = urllib.parse.parse_qs(parsed.query) + + if 'code' in params: + self._handle_callback(params) + elif 'error' in params: + self._handle_callback(params) + elif parsed.path == '/': + self._serve_login_page() + else: + self.send_response(404) + self.end_headers() + + def _serve_login_page(self): + self.send_response(200) + self.send_header('Content-type', 'text/html') + self.end_headers() + html = f""" + +SpotFetch - Spotify Login + +
+

SpotFetch

+

Connect your Spotify account to download playlists

+ + Log in with Spotify + +
+ +""" + self.wfile.write(html.encode()) + + def _get_auth_url(self): + verifier = secrets.token_urlsafe(64) + _AuthHandler.code_verifier = verifier + challenge = base64.urlsafe_b64encode( + hashlib.sha256(verifier.encode()).digest() + ).rstrip(b'=').decode() + + params = { + 'response_type': 'code', + 'client_id': self.server.client_id, + 'scope': SCOPES, + 'redirect_uri': REDIRECT_URI, + 'code_challenge_method': 'S256', + 'code_challenge': challenge, + } + return f"https://accounts.spotify.com/authorize?{urllib.parse.urlencode(params)}" + + def _handle_callback(self, params): + code = params.get('code', [None])[0] + error = params.get('error', [None])[0] + + if error: + self._serve_error_page(f"Authorization denied: {error}") + if _AuthHandler.auth_event: + _AuthHandler.auth_event.set() + return + + if code: + _AuthHandler.auth_code = code + if _AuthHandler.auth_event: + _AuthHandler.auth_event.set() + self._serve_success_page() + else: + self._serve_error_page("No authorization code received") + + def _serve_success_page(self): + self.send_response(200) + self.send_header('Content-type', 'text/html') + self.end_headers() + html = """ + +Authenticated + +
+
✓
+

Authenticated!

+

You can close this window and return to SpotFetch.

+
+ +""" + self.wfile.write(html.encode()) + + def _serve_error_page(self, message): + self.send_response(400) + self.send_header('Content-type', 'text/html') + self.end_headers() + html = f""" + +Authentication Failed + +
+
✗
+

Authentication Failed

+

{message}

+
+ +""" + self.wfile.write(html.encode()) + + def log_message(self, format, *args): + pass + + +class _AuthServer(HTTPServer): + def __init__(self, client_id, port=REDIRECT_PORT): + self.client_id = client_id + super().__init__(('127.0.0.1', port), _AuthHandler) + + +class SpotifyAuth: + def __init__(self, client_id=None): + self.client_id = client_id or os.environ.get("SPOTIFY_CLIENT_ID") or DEFAULT_CLIENT_ID + self._token_info = None + self._load_tokens() + + def _load_tokens(self): + if os.path.exists(TOKEN_FILE): + try: + with open(TOKEN_FILE) as f: + self._token_info = json.load(f) + except (json.JSONDecodeError, IOError): + self._token_info = None + + def _save_tokens(self, token_info): + os.makedirs(CONFIG_DIR, exist_ok=True) + with open(TOKEN_FILE, 'w') as f: + json.dump(token_info, f) + self._token_info = token_info + + def clear_tokens(self): + if os.path.exists(TOKEN_FILE): + os.remove(TOKEN_FILE) + self._token_info = None + + def is_authenticated(self): + return self._token_info is not None and 'refresh_token' in self._token_info + + def get_access_token(self): + if not self.client_id: + return None + if self._token_info and 'refresh_token' in self._token_info: + return self._refresh_access_token() + return None + + def _refresh_access_token(self): + resp = requests.post( + 'https://accounts.spotify.com/api/token', + data={ + 'grant_type': 'refresh_token', + 'refresh_token': self._token_info['refresh_token'], + 'client_id': self.client_id, + }, + headers={'Content-Type': 'application/x-www-form-urlencoded'}, + timeout=30, + ) + if resp.ok: + data = resp.json() + self._token_info['access_token'] = data['access_token'] + if 'refresh_token' in data: + self._token_info['refresh_token'] = data['refresh_token'] + self._save_tokens(self._token_info) + return data['access_token'] + self._token_info = None + return None + + def authenticate(self): + if not self.client_id: + raise SpotifyAuthError( + "Spotify Client ID not configured. " + "Set it via environment variable SPOTIFY_CLIENT_ID " + "or in SpotFetch settings." + ) + + os.makedirs(CONFIG_DIR, exist_ok=True) + + _AuthHandler.auth_code = None + _AuthHandler.code_verifier = None + auth_event = threading.Event() + _AuthHandler.auth_event = auth_event + + try: + server = _AuthServer(self.client_id) + except OSError as e: + raise SpotifyAuthError( + f"Could not start authentication server on port {REDIRECT_PORT}: {e}\n" + f"Make sure port {REDIRECT_PORT} is available." + ) + + thread = threading.Thread(target=server.serve_forever) + thread.daemon = True + thread.start() + + webbrowser.open(f'http://127.0.0.1:{REDIRECT_PORT}/') + print(f"\nOpening browser for Spotify login...") + print(f"If the browser doesn't open, visit: http://127.0.0.1:{REDIRECT_PORT}/\n") + + auth_event.wait(timeout=120) + + server.shutdown() + thread.join(timeout=2) + + if _AuthHandler.auth_code is None: + raise SpotifyAuthError( + "Authentication timed out or was denied. Please try again." + ) + + resp = requests.post( + 'https://accounts.spotify.com/api/token', + data={ + 'grant_type': 'authorization_code', + 'code': _AuthHandler.auth_code, + 'redirect_uri': REDIRECT_URI, + 'client_id': self.client_id, + 'code_verifier': _AuthHandler.code_verifier, + }, + headers={'Content-Type': 'application/x-www-form-urlencoded'}, + timeout=30, + ) + + if not resp.ok: + raise SpotifyAuthError(f"Failed to exchange code for token: {resp.text}") + + token_info = resp.json() + self._save_tokens(token_info) + print("Successfully authenticated with Spotify!") + return token_info['access_token']