Update spotizerr-auth to match spotizerr 3.0

This commit is contained in:
Xoconoch
2025-08-07 09:55:42 -06:00
parent f440e30308
commit 9da17b766f
3 changed files with 340 additions and 124 deletions

1
.gitignore vendored
View File

@@ -1 +1,2 @@
/.venv
credentials.json

View File

@@ -1,2 +1,3 @@
librespot-spotizerr
librespot-spotizerr==0.2.9
requests
rich

View File

@@ -4,206 +4,420 @@ import pathlib
import json
import requests
import sys
import os
import threading
class Colors:
BLUE = '\033[94m'
CYAN = '\033[96m'
GREEN = '\033[92m'
YELLOW = '\033[93m'
RED = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
# Rich library for beautiful terminal interfaces
from rich.console import Console
from rich.panel import Panel
from rich.prompt import Prompt, Confirm
from rich.rule import Rule
from rich.status import Status
from rich.text import Text
# Initialize the Rich console
console = Console()
# Define icons for consistency
class Icons:
SPOTIFY = "🎵"
CHECK = ""
CROSS = ""
WARNING = "⚠️"
INFO = ""
ARROW = ""
LOCK = "🔒"
KEY = "🔑"
USER = "👤"
GEAR = "⚙️"
DOWNLOAD = "⬇️"
SUCCESS = "🎉"
NETWORK = "🌐"
CLEAN = "🧹"
REGISTER = "👤➕"
ADMIN = "👑"
try:
from librespot.zeroconf import ZeroconfServer
except ImportError:
logging.error("librespot-spotizerr is not installed. Please install it with pip.")
logging.error("e.g. 'pip install -r requirements.txt' or 'pip install librespot-spotizerr'")
console.print("[bold red]Error: librespot-spotizerr is not installed. Please install it with pip.[/]")
console.print("[bold red]e.g. 'pip install -r requirements.txt' or 'pip install librespot-spotizerr'[/]")
sys.exit(1)
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
def print_header():
"""Print a modern header for the application using Rich Panel."""
header = Text.from_markup(f"""
{Icons.SPOTIFY} [bold cyan]SPOTIZERR AUTHENTICATION UTILITY[/]
[dim]Configure Spotify credentials for your Spotizerr instance[/]
""", justify="center")
console.print(Panel(header, border_style="cyan", expand=False))
def get_spotify_session_and_wait_for_credentials():
"""
Starts Zeroconf server and waits for librespot to store credentials.
Starts the patched Zeroconf server. Because it's patched to be a daemon
and not to log to the console, we can call it directly and cleanly.
"""
credential_file = pathlib.Path("credentials.json")
if credential_file.exists():
logging.info(f"Removing existing '{credential_file}'")
console.print(f"⚠️ [yellow]Removing existing '{credential_file}'[/]")
try:
credential_file.unlink()
except OSError as e:
logging.error(f"Could not remove existing 'credentials.json': {e}")
console.print(f"❌ [bold red]Could not remove existing 'credentials.json':[/] {e}")
sys.exit(1)
zs = ZeroconfServer.Builder().create()
device_name = "librespot-spotizerr"
# This is a bit of a hack to get the device name, but it's useful for the user.
if hasattr(zs, '_ZeroconfServer__server') and hasattr(zs._ZeroconfServer__server, 'name'):
device_name = zs._ZeroconfServer__server.name
# Start the patched ZeroconfServer. It will run silently in the background.
# No need to manage threads here anymore.
zeroconf_server = ZeroconfServer.Builder().create()
logging.info(f"Spotify Connect device '{Colors.CYAN}{device_name}{Colors.ENDC}' is now available on your network.")
logging.info(f"Please open Spotify on another device, and {Colors.BOLD}transfer playback to it{Colors.ENDC}.")
logging.info("This will capture your session and save it as 'credentials.json'.")
device_name = "Spotizerr Auth Tool" # A clear, static name for the user
instructions = f"""
[bold yellow]1.[/] Open Spotify on another device.
[bold yellow]2.[/] Look for '[bold white]{device_name}[/]' in the Connect menu.
[bold yellow]3.[/] [bold]Transfer playback[/] to capture the session.
"""
console.print(Panel(
instructions,
title=f"[cyan bold]{Icons.SPOTIFY} Connection Instructions[/]",
subtitle=f"[dim]{Icons.NETWORK} Now available on your network[/]",
border_style="cyan",
expand=False
))
try:
while True:
with Status("Waiting for Spotify connection...", spinner="dots", console=console):
while not (credential_file.exists() and credential_file.stat().st_size > 0):
time.sleep(1)
if credential_file.is_file() and credential_file.stat().st_size > 0:
logging.info(f"'{Colors.GREEN}credentials.json{Colors.ENDC}' has been created.")
if hasattr(zs, '_ZeroconfServer__session') and zs._ZeroconfServer__session:
try:
username = zs._ZeroconfServer__session.username()
logging.info(f"Session captured for user: {Colors.GREEN}{username}{Colors.ENDC}")
except Exception:
pass # It's ok if we can't get username
break
finally:
logging.info("Shutting down Spotify Connect server...")
zs.close()
def check_and_configure_api_creds(base_url):
"""
Checks if Spotizerr has Spotify API credentials and prompts user to add them if missing.
"""
api_config_url = f"{base_url.rstrip('/')}/api/credentials/spotify_api_config"
logging.info("Checking Spotizerr server for Spotify API configuration...")
console.print(f"✅ [green]Connection successful! Credential file has been created.[/]")
# We no longer need to manually close anything, the daemon will be handled
# automatically on script exit.
# --- ALL OTHER FUNCTIONS (check_auth_status, authenticate_user, etc.) REMAIN UNCHANGED ---
def check_auth_status(base_url):
"""
Check the authentication status of the Spotizerr instance using Rich for output.
"""
console.print(Rule(f"[bold blue]{Icons.LOCK} Authentication Status Check[/]", style="blue"))
auth_status_url = f"{base_url.rstrip('/')}/api/auth/status"
with console.status(f"Checking [underline]{base_url}[/]...", spinner="dots"):
try:
response = requests.get(api_config_url, timeout=10)
if response.status_code >= 400:
response = requests.get(auth_status_url, timeout=10)
response.raise_for_status()
auth_data = response.json()
except requests.exceptions.RequestException as e:
console.print(f"❌ [bold red]Failed to check auth status:[/] {e}")
return None
auth_enabled = auth_data.get('auth_enabled', False)
auth_icon = Icons.LOCK if auth_enabled else Icons.KEY
status_style = "red" if auth_enabled else "green"
console.print(f"{Icons.INFO} Authentication: [{status_style}]{auth_enabled}[/] {auth_icon}")
if auth_enabled:
authenticated = auth_data.get('authenticated', False)
auth_status_icon = Icons.CHECK if authenticated else Icons.CROSS
auth_status_style = "green" if authenticated else "red"
console.print(f"{Icons.INFO} Currently authenticated: [{auth_status_style}]{authenticated}[/] {auth_status_icon}")
console.print(f"{Icons.INFO} Registration enabled: [cyan]{auth_data.get('registration_enabled', False)}[/]")
if auth_data.get('sso_enabled', False):
providers = auth_data.get('sso_providers', [])
console.print(f"{Icons.INFO} SSO providers: [cyan]{', '.join(providers)}[/]")
else:
console.print(f"✅ [green]Authentication disabled - admin privileges are automatic.[/]")
return auth_data
def authenticate_user(base_url, auth_status):
"""
Handle user authentication using Rich Prompt for choices.
"""
if not auth_status.get('auth_enabled', False):
console.print("✅ [green]Authentication disabled, proceeding...[/]")
return None
if auth_status.get('authenticated', False):
console.print("✅ [green]Already authenticated.[/]")
return "existing_token"
console.print(Rule(f"[bold blue]{Icons.USER} User Authentication[/]", style="blue"))
choices = ["Login to existing account"]
if auth_status.get('registration_enabled', False):
choices.append("Register new account")
choice = Prompt.ask(
"[bold white]Select authentication method[/]",
choices=["Login", "Register"] if len(choices) > 1 else ["Login"],
default="Login"
)
if choice == "Login":
return login_user(base_url)
elif choice == "Register":
return register_user(base_url)
else:
console.print("❌ [bold red]Invalid choice.[/]")
return None
def login_user(base_url):
"""
Handle user login using Rich Prompt.
"""
console.print(Rule(f"[bold blue]{Icons.KEY} User Login[/]", style="blue"))
login_url = f"{base_url.rstrip('/')}/api/auth/login"
username = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Username")
password = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Password", password=True)
if not username or not password:
console.print("❌ [bold red]Username and password are required.[/]")
return None
payload = {"username": username, "password": password}
headers = {"Content-Type": "application/json"}
with console.status("Authenticating user...", spinner="dots"):
try:
response = requests.post(login_url, headers=headers, json=payload, timeout=10)
response.raise_for_status()
data = response.json()
client_id = data.get("client_id")
client_secret = data.get("client_secret")
token = data.get("access_token")
user_info = data.get("user", {})
if client_id and client_secret:
logging.info(f"{Colors.GREEN}Spotizerr API credentials are already configured.{Colors.ENDC}")
console.print(f"✅ [green]Welcome back, [bold]{user_info.get('username', 'unknown')}[/]![/]")
role = user_info.get('role', 'unknown')
role_icon = Icons.ADMIN if role == "admin" else Icons.USER
console.print(f"{Icons.INFO} Role: [cyan]{role}[/] {role_icon}")
return token
except requests.exceptions.RequestException as e:
console.print(f"❌ [bold red]Login failed:[/] {e}")
if hasattr(e, 'response') and e.response is not None:
try:
error_data = e.response.json()
console.print(f" [red]Details: {error_data.get('error', 'Unknown error')}[/]")
except json.JSONDecodeError:
console.print(f" [red]Response: {e.response.text}[/]")
return None
def register_user(base_url):
"""
Handle user registration using Rich Prompt.
"""
console.print(Rule(f"[bold blue]{Icons.REGISTER} User Registration[/]", style="blue"))
register_url = f"{base_url.rstrip('/')}/api/auth/register"
username = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Choose username")
email = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Email address")
password = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Choose password", password=True)
confirm_password = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Confirm password", password=True)
if not all([username, email, password]):
console.print("❌ [bold red]Username, email, and password are required.[/]")
return None
if password != confirm_password:
console.print("❌ [bold red]Passwords do not match.[/]")
return None
payload = {"username": username, "email": email, "password": password}
with console.status("Creating account...", spinner="dots"):
try:
response = requests.post(register_url, headers={"Content-Type": "application/json"}, json=payload, timeout=10)
response.raise_for_status()
console.print(f"{Icons.SUCCESS} [green]Account created for '[bold]{username}[/]'[/]")
console.print(f"{Icons.INFO} Please log in with your new credentials.")
return login_user(base_url)
except requests.exceptions.RequestException as e:
console.print(f"❌ [bold red]Registration failed:[/] {e}")
if hasattr(e, 'response') and e.response is not None:
try:
error_data = e.response.json()
console.print(f" [red]Details: {error_data.get('error', 'Unknown error')}[/]")
except json.JSONDecodeError:
console.print(f" [red]Response: {e.response.text}[/]")
return None
def get_auth_headers(token):
"""Get headers with authentication token if available."""
headers = {"Content-Type": "application/json"}
if token and token != "existing_token":
headers["Authorization"] = f"Bearer {token}"
return headers
def check_and_configure_api_creds(base_url, auth_token=None):
"""
Checks and configures Spotizerr API credentials using Rich UI components.
"""
console.print(Rule(f"[bold blue]{Icons.GEAR} Spotify API Configuration[/]", style="blue"))
api_config_url = f"{base_url.rstrip('/')}/api/credentials/spotify_api_config"
headers = get_auth_headers(auth_token)
try:
with console.status("Checking API credentials...", spinner="dots"):
response = requests.get(api_config_url, headers=headers, timeout=10)
response.raise_for_status()
data = response.json()
if data.get("client_id") and data.get("client_secret"):
console.print("✅ [green]Spotizerr API credentials are already configured.[/]")
return True
logging.warning(f"{Colors.YELLOW}Spotizerr server is missing Spotify API credentials (client_id/client_secret).{Colors.ENDC}")
logging.warning("You can get these from the Spotify Developer Dashboard: https://developer.spotify.com/dashboard")
configure_now = input(f"Do you want to configure them now? ({Colors.GREEN}y{Colors.ENDC}/{Colors.BOLD}N{Colors.ENDC}): ").lower()
console.print("⚠️ [yellow]Spotizerr server is missing Spotify API credentials (client_id/client_secret).[/]")
console.print(f"{Icons.INFO} Get these from: [underline]https://developer.spotify.com/dashboard[/]")
if configure_now != 'y':
logging.info("Please configure the API credentials on your Spotizerr server before proceeding.")
if not Confirm.ask("[bold]Do you want to configure them now?[/]", default=True):
console.print(f"{Icons.INFO} Please configure API credentials on your server before proceeding.")
return False
new_client_id = input(f"Enter your Spotify {Colors.CYAN}client_id{Colors.ENDC}: ")
new_client_secret = input(f"Enter your Spotify {Colors.CYAN}client_secret{Colors.ENDC}: ")
new_client_id = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Enter your Spotify [cyan]client_id[/]")
new_client_secret = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Enter your Spotify [cyan]client_secret[/]")
if not new_client_id or not new_client_secret:
logging.error(f"{Colors.RED}Both client_id and client_secret must be provided.{Colors.ENDC}")
console.print("❌ [bold red]Both client_id and client_secret must be provided.[/]")
return False
payload = {"client_id": new_client_id, "client_secret": new_client_secret}
headers = {"Content-Type": "application/json"}
with console.status("Updating API credentials...", spinner="dots"):
put_response = requests.put(api_config_url, headers=headers, json=payload, timeout=10)
put_response.raise_for_status()
logging.info("Successfully configured Spotizerr API credentials.")
console.print(f"{Icons.SUCCESS} [green]Successfully configured Spotizerr API credentials![/]")
return True
except requests.exceptions.RequestException as e:
logging.error(f"Failed to communicate with Spotizerr API at {api_config_url}: {e}")
if e.response is not None:
logging.error(f"Response status: {e.response.status_code}")
console.print(f"❌ [bold red]Failed to communicate with Spotizerr API at {api_config_url}:[/] {e}")
if hasattr(e, 'response') and e.response is not None:
console.print(f" [red]Response status: {e.response.status_code}[/]")
try:
logging.error(f"Response body: {e.response.json()}")
error_data = e.response.json()
console.print(f" [red]Response body: {error_data.get('error', e.response.text)}[/]")
except json.JSONDecodeError:
logging.error(f"Response body: {e.response.text}")
logging.error("Please ensure your Spotizerr instance is running and accessible at the specified URL.")
console.print(f" [red]Response body: {e.response.text}[/]")
return False
def main():
"""
Main function for the Spotizerr auth utility.
"""
try:
base_url = input("Enter the base URL of your Spotizerr instance [default: http://localhost:7171]: ")
if not base_url:
base_url = "http://localhost:7171"
logging.info(f"Using default base URL: {base_url}")
def main():
"""Main function for the Spotizerr auth utility."""
print_header()
try:
base_url = Prompt.ask(
f"[magenta]{Icons.ARROW}[/] Enter the base URL of your Spotizerr instance",
default="http://localhost:7171"
)
if not base_url.startswith(('http://', 'https://')):
base_url = 'http://' + base_url
if not check_and_configure_api_creds(base_url):
auth_status = check_auth_status(base_url)
if auth_status is None:
sys.exit(1)
account_name = input("Enter a name for this Spotify account: ")
auth_token = authenticate_user(base_url, auth_status)
if auth_status.get('auth_enabled', False) and auth_token is None:
console.print("❌ [bold red]Authentication was required but failed. Exiting.[/]")
sys.exit(1)
if not check_and_configure_api_creds(base_url, auth_token):
sys.exit(1)
console.print(Rule(f"[bold blue]{Icons.USER} Account Configuration[/]", style="blue"))
account_name = Prompt.ask(f"[magenta]{Icons.ARROW}[/] Enter a name for this Spotify account")
if not account_name:
logging.error("Account name cannot be empty.")
console.print("❌ [bold red]Account name cannot be empty.[/]")
sys.exit(1)
region = input("Enter your Spotify region (e.g., US, DE, MX). This is the 2-letter country code: ").upper()
region = Prompt.ask(
f"[magenta]{Icons.ARROW}[/] Enter your Spotify region (e.g., US, DE, MX). This is the 2-letter country code"
).upper()
if not region:
logging.error("Region cannot be empty.")
console.print("❌ [bold red]Region cannot be empty.[/]")
sys.exit(1)
console.print(Rule(f"[bold blue]{Icons.SPOTIFY} Spotify Session Capture[/]", style="blue"))
cred_file = pathlib.Path("credentials.json")
if cred_file.exists():
overwrite = input(f"'{cred_file}' already exists. Overwrite it by connecting to Spotify? (y/N): ").lower()
if overwrite == 'y':
console.print(f"⚠️ [yellow]'{cred_file}' already exists.[/]")
if Confirm.ask("Overwrite it by connecting to Spotify?", default=True):
get_spotify_session_and_wait_for_credentials()
else:
logging.info("Using existing 'credentials.json'.")
console.print(f"{Icons.INFO} Using existing 'credentials.json'.")
else:
get_spotify_session_and_wait_for_credentials()
if not cred_file.exists():
logging.error("Failed to obtain 'credentials.json'. Exiting.")
console.print("❌ [bold red]Failed to obtain 'credentials.json'. Exiting.[/]")
sys.exit(1)
console.print(Rule(f"[bold blue]{Icons.DOWNLOAD} Uploading Credentials[/]", style="blue"))
try:
with open(cred_file, "r") as f:
credentials_data = json.load(f)
except (FileNotFoundError, json.JSONDecodeError) as e:
logging.error(f"Could not read or parse 'credentials.json': {e}")
console.print(f"❌ [bold red]Could not read or parse 'credentials.json':[/] {e}")
sys.exit(1)
payload = {
"region": region,
"blob_content": credentials_data
}
payload = {"region": region, "blob_content": credentials_data}
api_url = f"{base_url.rstrip('/')}/api/credentials/spotify/{account_name}"
headers = {"Content-Type": "application/json"}
logging.info(f"Registering account '{account_name}' to Spotizerr at '{api_url}'")
headers = get_auth_headers(auth_token)
with console.status(f"Registering account '[bold]{account_name}[/]' to Spotizerr...", spinner="dots"):
try:
response = requests.post(api_url, headers=headers, json=payload)
response = requests.post(api_url, headers=headers, json=payload, timeout=10)
response.raise_for_status()
logging.info("Successfully registered/updated Spotify account in Spotizerr!")
if response.text:
logging.info(f"Response from server: {response.text}")
console.print(f"{Icons.SUCCESS} [green]Successfully registered/updated Spotify account in Spotizerr![/]")
if response.text and response.headers.get("Content-Type") == "application/json":
console.print(f"{Icons.INFO} Server response: {response.json()}")
except requests.exceptions.RequestException as e:
logging.error(f"Failed to call Spotizerr API: {e}")
if e.response is not None:
logging.error(f"Response status: {e.response.status_code}")
logging.error(f"Response body: {e.response.text}")
console.print(f"❌ [bold red]Failed to call Spotizerr API:[/] {e}")
if hasattr(e, 'response') and e.response is not None:
console.print(f" [red]Status: {e.response.status_code}[/]")
try:
error_data = e.response.json()
console.print(f" [red]Details: {error_data.get('error', e.response.text)}[/]")
except json.JSONDecodeError:
console.print(f" [red]Response body: {e.response.text}[/]")
sys.exit(1)
finally:
cleanup = input("Do you want to delete 'credentials.json' now? (y/N): ").lower()
if cleanup == 'y':
console.print(Rule(f"[bold blue]{Icons.CLEAN} Cleanup[/]", style="blue"))
if Confirm.ask("Do you want to delete 'credentials.json' now?", default=True):
try:
if cred_file.exists():
cred_file.unlink()
logging.info("'credentials.json' deleted.")
console.print("✅ [green]'credentials.json' deleted.[/]")
except OSError as e:
logging.error(f"Error deleting 'credentials.json': {e}")
console.print(f"❌ [bold red]Error deleting 'credentials.json':[/] {e}")
else:
logging.info("'credentials.json' not deleted.")
console.print(f"{Icons.INFO} 'credentials.json' was kept for future use.[/]")
sys.exit(0)
console.print(f"\n[bold green]{Icons.SUCCESS} Process completed successfully![/]")
console.print(f"[dim]Your Spotify account '{account_name}' is now configured in Spotizerr.[/]\n")
except KeyboardInterrupt:
logging.info("\nOperation cancelled by user. Exiting.")
console.print(f"\n[yellow]{Icons.WARNING} Operation cancelled by user.[/]")
sys.exit(0)
except Exception as e:
console.print(f"\n[bold red]An unexpected error occurred:[/] {e}")
console.print_exception(show_locals=True)
sys.exit(1)
if __name__ == "__main__":