225 lines
8.1 KiB
Python
Executable File
225 lines
8.1 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Spotify OAuth Helper for OpenClaw
|
|
Generates a permanent refresh_token for OpenClaw's Spotify MCP Server.
|
|
|
|
Usage:
|
|
python3 spotify_auth.py --client-id <ID> --client-secret <SECRET>
|
|
Or simply:
|
|
python3 spotify_auth.py
|
|
"""
|
|
|
|
import argparse
|
|
import base64
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import webbrowser
|
|
from http.server import BaseHTTPRequestHandler, HTTPServer
|
|
|
|
DEFAULT_PORT = 8888
|
|
DEFAULT_REDIRECT_URI = f"http://127.0.0.1:{DEFAULT_PORT}/callback"
|
|
|
|
SPOTIFY_AUTH_URL = "https://accounts.spotify.com/authorize"
|
|
SPOTIFY_TOKEN_URL = "https://accounts.spotify.com/api/token"
|
|
SPOTIFY_ME_URL = "https://api.spotify.com/v1/me"
|
|
|
|
SCOPES = [
|
|
# User profile & account
|
|
"user-read-private",
|
|
"user-read-email",
|
|
# Playback & Devices
|
|
"user-read-playback-state",
|
|
"user-modify-playback-state",
|
|
"user-read-currently-playing",
|
|
"user-read-recently-played",
|
|
# Library
|
|
"user-library-read",
|
|
"user-library-modify",
|
|
# Top stats & history
|
|
"user-top-read",
|
|
"user-read-playback-position",
|
|
# Playlists
|
|
"playlist-read-private",
|
|
"playlist-read-collaborative",
|
|
"playlist-modify-public",
|
|
"playlist-modify-private",
|
|
# Follow
|
|
"user-follow-read",
|
|
"user-follow-modify",
|
|
]
|
|
|
|
|
|
class OAuthCallbackHandler(BaseHTTPRequestHandler):
|
|
auth_code = None
|
|
error = None
|
|
|
|
def do_GET(self):
|
|
parsed_url = urllib.parse.urlparse(self.path)
|
|
if parsed_url.path == "/callback":
|
|
params = urllib.parse.parse_qs(parsed_url.query)
|
|
if "code" in params:
|
|
OAuthCallbackHandler.auth_code = params["code"][0]
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.end_headers()
|
|
html = """
|
|
<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<title>Spotify Auth Success</title>
|
|
<style>
|
|
body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif; display: flex; justify-content: center; align-items: center; height: 100vh; margin: 0; background: #121212; color: #fff; text-align: center; }
|
|
.card { background: #181818; padding: 40px; border-radius: 12px; box-shadow: 0 8px 24px rgba(0,0,0,0.5); border: 1px solid #282828; }
|
|
h1 { color: #1DB954; margin-bottom: 12px; }
|
|
p { color: #b3b3b3; font-size: 16px; }
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<div class="card">
|
|
<h1>✅ Authentication Successful!</h1>
|
|
<p>OpenClaw has successfully connected to your Spotify account.</p>
|
|
<p>You can now close this browser window and check your terminal.</p>
|
|
</div>
|
|
</body>
|
|
</html>
|
|
"""
|
|
self.wfile.write(html.encode("utf-8"))
|
|
else:
|
|
err_msg = params.get("error", ["Unknown error"])[0]
|
|
OAuthCallbackHandler.error = err_msg
|
|
self.send_response(400)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.end_headers()
|
|
self.wfile.write(f"<h1>Authentication Failed</h1><p>{err_msg}</p>".encode("utf-8"))
|
|
else:
|
|
self.send_response(404)
|
|
self.end_headers()
|
|
|
|
def log_message(self, format, *args):
|
|
# Silence standard HTTP access logging
|
|
pass
|
|
|
|
|
|
def exchange_code_for_tokens(client_id: str, client_secret: str, code: str, redirect_uri: str) -> dict:
|
|
auth_header = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
|
|
data = urllib.parse.urlencode({
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": redirect_uri,
|
|
}).encode("utf-8")
|
|
|
|
req = urllib.request.Request(
|
|
SPOTIFY_TOKEN_URL,
|
|
data=data,
|
|
headers={
|
|
"Authorization": f"Basic {auth_header}",
|
|
"Content-Type": "application/x-www-form-urlencoded",
|
|
},
|
|
method="POST"
|
|
)
|
|
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
|
|
def fetch_user_profile(access_token: str) -> dict:
|
|
req = urllib.request.Request(
|
|
SPOTIFY_ME_URL,
|
|
headers={"Authorization": f"Bearer {access_token}"}
|
|
)
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode("utf-8"))
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Spotify OAuth Helper for OpenClaw")
|
|
parser.add_argument("--client-id", help="Spotify Developer App Client ID")
|
|
parser.add_argument("--client-secret", help="Spotify Developer App Client Secret")
|
|
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help=f"Local callback port (default: {DEFAULT_PORT})")
|
|
args = parser.parse_args()
|
|
|
|
print("\n🎵 ===========================================")
|
|
print(" OpenClaw Spotify OAuth Token Generator")
|
|
print("===========================================\n")
|
|
|
|
client_id = args.client_id or os.environ.get("SPOTIFY_CLIENT_ID")
|
|
client_secret = args.client_secret or os.environ.get("SPOTIFY_CLIENT_SECRET")
|
|
|
|
if not client_id:
|
|
client_id = input("Enter Spotify Client ID: ").strip()
|
|
if not client_secret:
|
|
client_secret = input("Enter Spotify Client Secret: ").strip()
|
|
|
|
if not client_id or not client_secret:
|
|
print("❌ Error: Client ID and Client Secret are required.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
redirect_uri = f"http://127.0.0.1:{args.port}/callback"
|
|
|
|
# Start local HTTP server for OAuth callback
|
|
try:
|
|
server = HTTPServer(("127.0.0.1", args.port), OAuthCallbackHandler)
|
|
except OSError as e:
|
|
print(f"❌ Error: Port {args.port} is already in use. Try specifying another port with --port.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# Build auth URL
|
|
auth_params = {
|
|
"client_id": client_id,
|
|
"response_type": "code",
|
|
"redirect_uri": redirect_uri,
|
|
"scope": " ".join(SCOPES),
|
|
"show_dialog": "true",
|
|
}
|
|
auth_url = f"{SPOTIFY_AUTH_URL}?{urllib.parse.urlencode(auth_params)}"
|
|
|
|
print(f"\n🌐 Opening browser to authorize Spotify account...")
|
|
print(f"If the browser doesn't open automatically, visit this URL:\n\n{auth_url}\n")
|
|
webbrowser.open(auth_url)
|
|
|
|
print(f"Waiting for authorization on {redirect_uri} ...")
|
|
while OAuthCallbackHandler.auth_code is None and OAuthCallbackHandler.error is None:
|
|
server.handle_request()
|
|
|
|
if OAuthCallbackHandler.error:
|
|
print(f"\n❌ Authorization rejected or failed: {OAuthCallbackHandler.error}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
code = OAuthCallbackHandler.auth_code
|
|
print("🔄 Exchanging authorization code for permanent tokens...")
|
|
|
|
try:
|
|
token_data = exchange_code_for_tokens(client_id, client_secret, code, redirect_uri)
|
|
except urllib.error.HTTPError as e:
|
|
print(f"\n❌ Failed to exchange code for tokens: {e.read().decode('utf-8')}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
access_token = token_data.get("access_token")
|
|
refresh_token = token_data.get("refresh_token")
|
|
|
|
# Fetch user profile to display confirmation
|
|
user_name = "Spotify User"
|
|
try:
|
|
profile = fetch_user_profile(access_token)
|
|
user_name = f"{profile.get('display_name', 'User')} ({profile.get('id', '')})"
|
|
except Exception:
|
|
pass
|
|
|
|
print("\n🎉 ===================================================")
|
|
print(f" Successfully Authenticated as: {user_name}")
|
|
print("===================================================\n")
|
|
print("Add these lines to your k8s/openclaw/openclaw.secret.yaml:")
|
|
print("---------------------------------------------------")
|
|
print(f" spotify-client-id: \"{client_id}\"")
|
|
print(f" spotify-client-secret: \"{client_secret}\"")
|
|
print(f" spotify-refresh-token: \"{refresh_token}\"")
|
|
print("---------------------------------------------------\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|