ajayarora1235's picture
fix suno bugs, link bugs
4162802
raw
history blame
10.7 kB
import requests
import time
import os
import asyncio
import httpx
base_url = "https://sss-api-blond.vercel.app"
api_endpoint_submit = f"{base_url}/generate/"
api_endpoint_concat = f"{base_url}/generate/concat"
api_endpoint_info = f"{base_url}/feed/"
api_key = "xZ1PhKexmwTR3dDskMvIGRlx137K40Il"
headers = {"api-key": api_key}
# tags = "lofi, chill, happy"
# prompt = "I'm a fish swimming in the ocean\nI'm a bird flying in the sky\nI'm a flower blooming in the garden\nI'm a tree standing tall and high"
def make_song(snippet_lyrics, inst_tags, continue_from_clip=None, continue_at=None):
"""
Generates a song based on provided lyrics and instrumental tags.
Args:
snippet_lyrics (str): The lyrics for the song snippet.
inst_tags (str): The instrumental tags for the song.
continue_from_clip (str, optional): The clip ID to continue from, if any. Defaults to None.
continue_at (int, optional): The position to continue at in the clip. Defaults to None.
Returns:
str: The link to the generated song.
"""
os.makedirs("audio_clips", exist_ok=True)
song_name = f"SG_{int(time.time())}"
suno_song_path = f"./audio_clips/suno_{song_name}.wav"
print("Passing to generate_song:", inst_tags, snippet_lyrics, suno_song_path)
song_link = generate_song(inst_tags, snippet_lyrics, suno_song_path, continue_from_clip, continue_at) \
if continue_from_clip else generate_song(inst_tags, snippet_lyrics, suno_song_path)
return song_link
# Takes about 30 seconds
def generate_song(tags, prompt, save_path, clip_id=None, continue_at=30):
# print("Generating song with tags", tags, "and prompt", prompt)
data = {
"title": "Songchat " + str(int(time.time())),
"tags": tags,
"prompt": prompt,
"mv": "chirp-v3-5"
}
if clip_id is not None:
data["continue_clip_id"] = clip_id
if continue_at is not None:
data["continue_at"] = continue_at
else:
data["continue_at"] = 30
feed_url = api_endpoint_info + clip_id
response = requests.get(feed_url, headers=headers)
response_data = response.json()
while True:
if response.status_code != 200:
print("No data in response, retrying", response_data)
time.sleep(2)
continue
# Check if response_data is a list or a dictionary
elif isinstance(response_data, list):
if len(response_data) == 0 or "status" not in response_data[0]:
print("Invalid response data, update later")
time.sleep(2)
continue
status = response_data[0]["status"]
elif isinstance(response_data, dict):
if "status" not in response_data:
print("Invalid response data, update later")
time.sleep(2)
continue
status = response_data["status"]
else:
print("Unexpected response format, update later")
time.sleep(2)
continue
if status == 'streaming':
return "Snippet to extend is still streaming, please wait to request later."
if status == 'complete':
break
else:
time.sleep(8)
continue
response = requests.post(api_endpoint_submit, json=data) #,headers=headers)
response_data = response.json()
# print(response_data)
if response.status_code != 200:
print("Failure while submitting song req, retrying", response_data)
time.sleep(5)
song_link = generate_song(tags, prompt, save_path, clip_id)
return song_link
print(response_data)
if "clips" in response_data:
song_ids = [d["id"] for d in response_data["clips"]]
else:
print('something went wrong, retrying')
time.sleep(5)
song_link = generate_song(tags, prompt, save_path, clip_id)
return song_link
print("got song ids", song_ids)
if song_ids == []:
print("No song ids returned, retrying with shorter prompt")
print("Response data was", response_data)
time.sleep(5)
# take off 30 words from the prompt
new_prompt = " ".join(prompt.split(" ")[:-30])
generate_song(tags, new_prompt, save_path)
return
song_id = song_ids[0] # Generally two song ids are returned
startTime = time.time()
while True:
response = requests.get(api_endpoint_info + song_id, headers=headers)
response_data = response.json()
status = ""
if response.status_code != 200:
print("No data in response, retrying", response_data)
time.sleep(2)
continue
# print("Got response", response_data)
# Check if response_data is a list or a dictionary
if isinstance(response_data, list):
if len(response_data) == 0 or "status" not in response_data[0]:
print("Invalid response data, update later")
time.sleep(2)
continue
status = response_data[0]["status"]
audio_url = response_data[0].get("audio_url", "")
elif isinstance(response_data, dict):
if "status" not in response_data:
print("Invalid response data, update later")
time.sleep(2)
continue
status = response_data["status"]
audio_url = response_data.get("audio_url", "")
else:
print("Unexpected response format, update later")
time.sleep(2)
continue
if status == 'streaming':
break
else:
time.sleep(2)
continue
# if time.time() - startTime > 300:
# raise Exception("Timeout while waiting for song completion")
print("Got song", audio_url)
url = audio_url
return url
def concat_snippets(clip_id):
concat_url = f"{api_endpoint_concat}?clip_id={clip_id}"
feed_url = api_endpoint_info + clip_id
while True:
status = ""
response = requests.get(feed_url, headers=headers)
response_data = response.json()
if response.status_code != 200:
print("No data in response, retrying", response_data)
time.sleep(2)
continue
if isinstance(response_data, list):
if len(response_data) == 0 or "status" not in response_data[0]:
print("Invalid list response data, update later")
continue
status = response_data[0]["status"]
elif isinstance(response_data, dict):
if "status" not in response_data:
print("Invalid dictionary response data, update later")
continue
status = response_data["status"]
else:
print("Unexpected response format, update later")
continue
if status == 'streaming':
return "Song is still streaming, please wait to request later.", None, None, []
if status == 'complete':
break
else:
print("Streaming status couldn't be retrieved, response_data was ", response_data)
time.sleep(8)
continue
response = requests.post(concat_url)
response_data = response.json()
print(response_data)
if response.status_code != 200:
print("Failure while submitting merge, retrying", response_data)
time.sleep(5)
url, lyrics, concatenated_clips = concat_snippets(clip_id)
return url, lyrics, concatenated_clips
lyrics = response_data["metadata"]["prompt"]
tags = response_data["metadata"]["tags"]
concatenated_clips = [x["id"] for x in response_data["metadata"]["concat_history"]]
song_id = response_data["id"]
startTime = time.time()
while True:
response = requests.get(api_endpoint_info + song_id, headers=headers)
response_data = response.json()
print("feed response for concatenated song", response_data)
if response.status_code != 200:
print("No data in response, retrying", response_data)
time.sleep(2)
continue
print("Got concat snippet response", response_data)
# Check if response_data is a list or a dictionary
if isinstance(response_data, list):
if len(response_data) == 0 or "status" not in response_data[0]:
print("Invalid response data, update later")
time.sleep(2)
continue
status = response_data[0]["status"]
audio_url = response_data[0].get("audio_url", "")
elif isinstance(response_data, dict):
if "status" not in response_data:
print("Invalid response data, update later")
time.sleep(2)
continue
status = response_data["status"]
audio_url = response_data.get("audio_url", "")
else:
print("Unexpected response format, update later")
time.sleep(2)
continue
if status == 'streaming' or audio_url != "" or status == 'complete':
break
else:
time.sleep(2)
continue
# if time.time() - startTime > 300:
# raise Exception("Timeout while waiting for song completion")
print("Got song", audio_url)
url = audio_url
return url, lyrics, tags, concatenated_clips
def update_song_links(generated_audios):
updated_generated_audios = generated_audios.copy()
for i, song_info in enumerate(generated_audios):
clip_path, lyrics, instrumental, title, status = song_info
if "audiopipe.suno.ai" in clip_path or status == "streaming":
clip_id = clip_path.split("?item_id=")[-1]
feed_url = api_endpoint_info + clip_id
response = requests.get(feed_url, headers=headers)
response_data = response.json()
if response.status_code != 200:
print("No data in response, retrying", response_data)
continue
# Check if response_data is a list or a dictionary
status = ""
audio_url = ""
if isinstance(response_data, list):
if len(response_data) == 0 or "status" not in response_data[0]:
print("Invalid list response data, update later")
continue
status = response_data[0]["status"]
audio_url = response_data[0].get("audio_url", "")
elif isinstance(response_data, dict):
if "status" not in response_data:
print("Invalid dictionary response data, update later")
continue
status = response_data["status"]
audio_url = response_data.get("audio_url", "")
else:
print("Unexpected response format, update later")
continue
if status == 'streaming':
print("still streaming, update later")
continue
if status == 'complete' and audio_url != "":
updated_clip_path = audio_url
print(updated_clip_path)
updated_generated_audios[i] = (updated_clip_path, lyrics, instrumental, title, "complete")
return updated_generated_audios