o2o-castad-backend/poc/timestamp_lyric/lyric_timestamp_mapper.py

42 lines
1.6 KiB
Python

from openai import OpenAI
from difflib import SequenceMatcher
from dataclasses import dataclass
from typing import List, Tuple
import aiohttp, json
@dataclass
class TimestampedLyric:
text: str
start: float
end: float
SUNO_BASE_URL="https://api.sunoapi.org"
SUNO_TIMESTAMP_ROUTE = "/api/v1/generate/get-timestamped-lyrics"
SUNO_DETAIL_ROUTE = "/api/v1/generate/record-info"
class LyricTimestampMapper:
suno_api_key : str
def __init__(self, suno_api_key):
self.suno_api_key = suno_api_key
async def get_suno_audio_id_from_task_id(self, suno_task_id): # expire if db save audio id
url = f"{SUNO_BASE_URL}{SUNO_DETAIL_ROUTE}"
headers = {"Authorization": f"Bearer {self.suno_api_key}"}
async with aiohttp.ClientSession() as session:
async with session.get(url, headers=headers, params={"taskId" : suno_task_id}) as response:
detail = await response.json()
result = detail['data']['response']['sunoData'][0]['id']
return result
async def get_suno_timestamp(self, suno_task_id, suno_audio_id): # expire if db save audio id
url = f"{SUNO_BASE_URL}{SUNO_TIMESTAMP_ROUTE}"
headers = {"Authorization": f"Bearer {self.suno_api_key}"}
payload = {
"task_id" : suno_task_id,
"audio_id" : suno_audio_id
}
async with aiohttp.ClientSession() as session:
async with session.post(url, headers=headers, data=json.dumps(payload)) as response:
result = await response.json()
return result