Can't resolve my unauthorized issue

[ajayarora1235]

I've been working with the API code to have it run locally within an app of mine, but I'm running into the {detail: Unauthorized} bug that I can't get around when I call generate(). Here's my full code -- no issues before the generate call.

Is there a reason this code can't work locally or on its own in a single py file? would appreciate if you could pass in your session and cookie info and see if this works for you.

from datetime import datetime
from typing import Any, List, Optional, Union

from pydantic import BaseModel, Field
import asyncio

class schemas:
    class Response(BaseModel):
        code: Optional[int] = 0
        msg: Optional[str] = "success"
        data: Optional[Any] = None

    class CustomModeGenerateParam(BaseModel):
        """Generate with Custom Mode"""
        prompt: str = Field(..., description="lyrics")
        mv: str = Field(..., description="model version, default: chirp-v3-0", examples=["chirp-v3-0"])
        title: str = Field(..., description="song title")
        tags: str = Field(..., description="style of music")
        continue_at: Optional[int] = Field(default=None, description="continue a new clip from a previous song, format number", examples=[120])
        continue_clip_id: Optional[str] = None

    class DescriptionModeGenerateParam(BaseModel):
        """Generate with Song Description"""
        gpt_description_prompt: str
        make_instrumental: bool = False
        mv: str = Field(default='chirp-v3-0', description="model version, default: chirp-v3-0", examples=["chirp-v3-0"])
        prompt: str = Field(default="", description="Placeholder, keep it as an empty string, do not modify it")

## UTILS

import json
import os
import time

import aiohttp
# from dotenv import load_dotenv

# load_dotenv()

# BASE_URL = os.getenv("BASE_URL")

BASE_URL = "https://studio-api.suno.ai"

COMMON_HEADERS = {
    "Content-Type": "text/plain;charset=UTF-8",
    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
    "Referer": "https://suno.com",
    "Origin": "https://suno.com",
}


async def fetch(url, headers=None, data=None, method="POST"):
    if headers is None:
        headers = {}
    headers.update(COMMON_HEADERS)
    if data is not None:
        data = json.dumps(data)

    print(data, method, headers, url)

    async with aiohttp.ClientSession() as session:
        try:
            async with session.request(
                method=method, url=url, data=data, headers=headers
            ) as resp:
                return await resp.json()
        except Exception as e:
            return f"An error occurred: {e}"


async def get_feed(ids, token):
    headers = {"Authorization": f"Bearer {token}"}
    api_url = f"{BASE_URL}/api/feed/?ids={ids}"
    response = await fetch(api_url, headers, method="GET")
    return response


async def generate_music(data, token):
    headers = {"Authorization": f"Bearer {token}"}
    api_url = f"{BASE_URL}/api/generate/v2/"
    print(api_url, headers)
    response = await fetch(api_url, headers, data)
    return response


async def generate_lyrics(prompt, token):
    headers = {"Authorization": f"Bearer {token}"}
    api_url = f"{BASE_URL}/api/generate/lyrics/"
    data = {"prompt": prompt}
    return await fetch(api_url, headers, data)


async def get_lyrics(lid, token):
    headers = {"Authorization": f"Bearer {token}"}
    api_url = f"{BASE_URL}/api/generate/lyrics/{lid}"
    return await fetch(api_url, headers, method="GET")


## cookie

import os
import time
from http.cookies import SimpleCookie
from threading import Thread

import requests

# from utils import COMMON_HEADERS


class SunoCookie:
    def __init__(self):
        self.cookie = SimpleCookie()
        self.session_id = None
        self.token = None

    def load_cookie(self, cookie_str):
        self.cookie.load(cookie_str)

    def get_cookie(self):
        return ";".join([f"{i}={self.cookie.get(i).value}" for i in self.cookie.keys()])

    def set_session_id(self, session_id):
        self.session_id = session_id

    def get_session_id(self):
        return self.session_id

    def get_token(self):
        return self.token

    def set_token(self, token: str):
        self.token = token





def update_token(suno_cookie: SunoCookie):
    headers = {"cookie": suno_cookie.get_cookie()}
    headers.update(COMMON_HEADERS)
    session_id = suno_cookie.get_session_id()

    resp = requests.post(
        url=f"https://clerk.suno.com/v1/client/sessions/{session_id}/tokens?_clerk_js_version=4.72.0-snapshot.vc141245",
        headers=headers,
    )

    resp_headers = dict(resp.headers)
    set_cookie = resp_headers.get("Set-Cookie")
    suno_cookie.load_cookie(set_cookie)
    token = resp.json().get("jwt")
    suno_cookie.set_token(token)
    # print(set_cookie)
    # print(f"*** token -> {token} ***")


def keep_alive(suno_cookie: SunoCookie):
    while True:
        try:
            update_token(suno_cookie)
        except Exception as e:
            print(e)
        finally:
            time.sleep(5)


def start_keep_alive(suno_cookie: SunoCookie):
    t = Thread(target=keep_alive, args=(suno_cookie,))
    t.start()


import json
from fastapi import Depends, HTTPException, status

async def get_root():
    return schemas.Response()

async def generate(data: schemas.CustomModeGenerateParam, token: str = Depends(get_token)):
    try:
        resp = await generate_music(data.dict(), token)
        return resp
    except Exception as e:
        raise HTTPException(
            detail=str(e),
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

async def generate_with_song_description(data: schemas.DescriptionModeGenerateParam, token: str = Depends(get_token)):
    try:
        resp = await generate_music(data.dict(), token)
        return resp
    except Exception as e:
        raise HTTPException(
            detail=str(e),
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

async def fetch_feed(aid: str, token: str = Depends(get_token)):
    try:
        resp = await get_feed(aid, token)
        return resp
    except Exception as e:
        raise HTTPException(
            detail=str(e),
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

async def generate_lyrics_post(prompt: str, token: str = Depends(get_token)):
    if prompt is None:
        raise HTTPException(
            detail="prompt is required",
            status_code=status.HTTP_400_BAD_REQUEST
        )
    try:
        resp = await generate_lyrics(prompt, token)
        return resp
    except Exception as e:
        raise HTTPException(
            detail=str(e),
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

async def fetch_lyrics(lid: str, token: str = Depends(get_token)):
    try:
        resp = await get_lyrics(lid, token)
        return resp
    except Exception as e:
        raise HTTPException(
            detail=str(e),
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
        )

cookie_new = "replace this"
session = "replace this"

suno_auth = SunoCookie()
suno_auth.set_session_id(session)
suno_auth.load_cookie(cookie_new)

start_keep_alive(suno_auth)

def get_token():
  token = suno_auth.get_token()
  try:
      yield token
  finally:
      pass
  
tags = "fast, chill, happy"
prompt = """\[Verse 1\] Walking down the street The city's shining bright Neon signs are glowing Colors catching my sight (ooh-yeah) People passing by In a hurry Out of time Caught up in the chaos of this city's daily grind \[Verse 2\] Taxis honking horns Sirens fill the air Skyscrapers reaching high A bustling affair (ooh-ooh) Nightclubs pumping beats Dance floors packed tight Everyone's searching for a taste of the nightlife \[Chorus\] Lost in the city lights I feel the energy ignite Moving to the rhythm Partying into the night Living for the moment Taking in the sights Lost in the city lights We're catching dreams in flight"""

data_test = schemas.CustomModeGenerateParam(
    title="City Lights",
    tags=tags,
    prompt=prompt,
    mv="chirp-v3-0"
)

# Run the coroutine and get the result
async def main():
    result = await generate(data=data_test)
    print(result)

# Create an event loop and run the main function
loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()```