Spaces:
Sleeping
Sleeping
| import os | |
| import re | |
| import time | |
| import uuid | |
| import tempfile | |
| import httpx | |
| import base64 | |
| import hashlib | |
| import secrets | |
| from flask import Flask, request, redirect, url_for, session, render_template, send_file | |
| from spotipy import Spotify | |
| from spotipy.exceptions import SpotifyException | |
| # Flask app setup | |
| app = Flask(__name__) | |
| app.secret_key = os.urandom(24) | |
| app.config['SESSION_COOKIE_NAME'] = 'playlist-generator-session' | |
| # Spotify API credentials | |
| SPOTIFY_CLIENT_ID = os.getenv("CLIENT_ID") | |
| SPOTIFY_REDIRECT_URI = os.getenv("REDIRECT_URL") | |
| # Shape API config | |
| SHAPE_API_KEY = os.getenv("OPENAI_API_KEY") | |
| SHAPE_API_BASE_URL = "https://api.shapes.inc/v1/chat/completions" | |
| SHAPE_API_MODEL = os.getenv("OPENAI_MODEL") | |
| # Temp file for recommendations | |
| temp_file = os.path.join(tempfile.gettempdir(), "playlist_recommendations.txt") | |
| def generate_pkce_pair(): | |
| code_verifier = base64.urlsafe_b64encode(secrets.token_bytes(64)).decode('utf-8').rstrip('=') | |
| code_challenge = base64.urlsafe_b64encode( | |
| hashlib.sha256(code_verifier.encode()).digest() | |
| ).decode('utf-8').rstrip('=') | |
| return code_verifier, code_challenge | |
| def get_spotify_token(): | |
| if session.get('guest'): | |
| return None | |
| token_info = session.get('token_info') | |
| if not token_info: | |
| return None | |
| if 'expires_at' in token_info and token_info['expires_at'] - int(time.time()) < 60: | |
| return None # PKCE flow does not support refresh token | |
| return token_info.get('access_token') | |
| def clean_response(response_text): | |
| cleaned_text = re.sub(r'^.*?</think>', '', response_text, flags=re.DOTALL) | |
| cleaned_text = '\n'.join(line.strip() for line in cleaned_text.splitlines() if line.strip()) | |
| return cleaned_text | |
| def call_shape_api(prompt, max_tokens=2000, temperature=0.7): | |
| headers = { | |
| "Authorization": f"Bearer {SHAPE_API_KEY}", | |
| "Content-Type": "application/json", | |
| "X-User-ID": str(uuid.uuid4()), | |
| "X-Channel-ID": str(uuid.uuid4()), | |
| } | |
| json_payload = { | |
| "model": SHAPE_API_MODEL, | |
| "messages": [{"role": "user", "content": prompt}], | |
| } | |
| try: | |
| response = httpx.post(SHAPE_API_BASE_URL, headers=headers, json=json_payload, timeout=None) | |
| response.raise_for_status() | |
| data = response.json() | |
| content = data.get("choices", [{}])[0].get("message", {}).get("content", "") | |
| return content | |
| except Exception as e: | |
| raise RuntimeError(f"Shape API request failed: {e}") | |
| def index(): | |
| return render_template('index.html') | |
| def login(): | |
| code_verifier, code_challenge = generate_pkce_pair() | |
| session['code_verifier'] = code_verifier | |
| params = { | |
| "client_id": SPOTIFY_CLIENT_ID, | |
| "response_type": "code", | |
| "redirect_uri": SPOTIFY_REDIRECT_URI, | |
| "scope": "playlist-modify-public playlist-modify-private user-read-private", | |
| "code_challenge_method": "S256", | |
| "code_challenge": code_challenge, | |
| } | |
| auth_url = f"https://accounts.spotify.com/authorize?{httpx.QueryParams(params)}" | |
| return redirect(auth_url) | |
| def guest_login(): | |
| session.clear() | |
| session['guest'] = True | |
| return redirect(url_for('dashboard')) | |
| def callback(): | |
| code = request.args.get('code') | |
| error = request.args.get('error') | |
| if error: | |
| return render_template("error.html", error_message=f"Spotify login failed: {error}", return_url=url_for('index')), 400 | |
| code_verifier = session.get('code_verifier') | |
| if not code or not code_verifier: | |
| return render_template("error.html", error_message="Invalid request (missing code or verifier).", return_url=url_for('index')), 400 | |
| token_data = { | |
| "client_id": SPOTIFY_CLIENT_ID, | |
| "grant_type": "authorization_code", | |
| "code": code, | |
| "redirect_uri": SPOTIFY_REDIRECT_URI, | |
| "code_verifier": code_verifier, | |
| } | |
| try: | |
| response = httpx.post("https://accounts.spotify.com/api/token", data=token_data, timeout=10) | |
| response.raise_for_status() | |
| token_info = response.json() | |
| token_info['expires_at'] = int(time.time()) + token_info['expires_in'] | |
| session['token_info'] = token_info | |
| session.pop('guest', None) | |
| return redirect(url_for('dashboard')) | |
| except Exception as e: | |
| return render_template("error.html", error_message=f"Failed to get token: {str(e)}", return_url=url_for('index')), 500 | |
| def dashboard(): | |
| guest = session.get('guest', False) | |
| token = get_spotify_token() | |
| user = None | |
| if token: | |
| try: | |
| sp = Spotify(auth=token) | |
| user = sp.current_user() | |
| except SpotifyException: | |
| user = None | |
| return render_template('dashboard.html', user=user, guest=guest) | |
| def generate(): | |
| prompt_input = request.form.get('prompt') | |
| playlist_name = request.form.get('playlist_name', 'My Playlist').strip() | |
| session['playlist_name'] = playlist_name | |
| prompt = f"!web make playlist provide it without any additional comment and numbering, in format Artist - Name, in topic: {prompt_input}" | |
| session['last_prompt'] = prompt | |
| try: | |
| raw_response = call_shape_api(prompt, max_tokens=2000) | |
| cleaned = clean_response(raw_response) | |
| with open(temp_file, "w", encoding="utf-8") as f: | |
| f.write(cleaned) | |
| return redirect(url_for('playlist')) | |
| except Exception as e: | |
| return render_template("error.html", error_message=str(e), return_url=url_for('dashboard')), 500 | |
| def load_more(): | |
| prompt = session.get('last_prompt', '') | |
| if not prompt: | |
| return render_template("error.html", error_message="No prompt found in session.", return_url=url_for('dashboard')), 400 | |
| try: | |
| raw_response = call_shape_api(prompt, max_tokens=500) | |
| cleaned = clean_response(raw_response) | |
| with open(temp_file, "a", encoding="utf-8") as f: | |
| f.write("\n" + cleaned) | |
| return redirect(url_for('playlist')) | |
| except Exception as e: | |
| return render_template("error.html", error_message=str(e), return_url=url_for('dashboard')), 500 | |
| def playlist(): | |
| try: | |
| with open(temp_file, "r", encoding="utf-8") as f: | |
| recommendations = f.read().splitlines() | |
| except FileNotFoundError: | |
| recommendations = [] | |
| return render_template('playlist.html', recommendations=recommendations, guest=session.get('guest', False)) | |
| def download(): | |
| return send_file(temp_file, as_attachment=True, download_name="playlist.txt") | |
| def chunkify(lst, n): | |
| for i in range(0, len(lst), n): | |
| yield lst[i:i + n] | |
| def create_playlist(): | |
| if session.get('guest'): | |
| return render_template("error.html", error_message="Spotify features are disabled in guest mode.", return_url=url_for('dashboard')), 403 | |
| token = get_spotify_token() | |
| if not token: | |
| return redirect(url_for('login')) | |
| sp = Spotify(auth=token) | |
| try: | |
| user_id = sp.current_user()['id'] | |
| except SpotifyException: | |
| return render_template("error.html", error_message="Spotify access denied. Make sure your account is registered in the Spotify Developer Dashboard.", return_url=url_for('dashboard')), 403 | |
| except Exception as e: | |
| return render_template("error.html", error_message=f"Unexpected error: {e}", return_url=url_for('dashboard')), 500 | |
| playlist_name = session.get('playlist_name', 'My Playlist') | |
| try: | |
| with open(temp_file, "r", encoding="utf-8") as f: | |
| recommendations = f.read().splitlines() | |
| except FileNotFoundError: | |
| return render_template("error.html", error_message="No songs to add.", return_url=url_for('playlist')), 400 | |
| if not recommendations: | |
| return render_template("error.html", error_message="No songs to add.", return_url=url_for('playlist')), 400 | |
| playlist = sp.user_playlist_create(user_id, playlist_name, public=False) | |
| track_uris = [] | |
| for song in recommendations: | |
| if song.strip(): | |
| results = sp.search(q=song, type="track", limit=1) | |
| tracks = results.get('tracks', {}).get('items', []) | |
| if tracks: | |
| track_uris.append(tracks[0]['uri']) | |
| if track_uris: | |
| for chunk in chunkify(track_uris, 10): | |
| sp.playlist_add_items(playlist['id'], chunk) | |
| time.sleep(0.5) | |
| return render_template("in_spotify.html", playlist_url=playlist['external_urls']['spotify'], playlist_name=playlist_name) | |
| else: | |
| return render_template("error.html", error_message="No valid tracks found.", return_url=url_for('playlist')), 400 | |
| def handle_errors(e): | |
| return render_template("error.html", error_message=str(e), return_url=url_for('index')), getattr(e, 'code', 500) | |
| if __name__ == '__main__': | |
| app.run(host="0.0.0.0", port=7860) |