Blame for app.py

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}")

@app.route('/')
def index():
    return render_template('index.html')

@app.route('/login')
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)

@app.route('/guest_login')
def guest_login():
    session.clear()
    session['guest'] = True
    return redirect(url_for('dashboard'))

@app.route('/callback')
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

@app.route('/dashboard')
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)

@app.route('/generate', methods=['POST'])
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

@app.route('/load_more')
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

@app.route('/playlist')
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))

@app.route('/download')
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]

@app.route('/create_playlist')
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

@app.errorhandler(404)
@app.errorhandler(500)
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)