feat(music-render): providers/sync_ops.py — sync Suno helpers (SP-5)
NAS sync 함수 4종 이식: generate_lyrics, get_credits, get_timestamped_lyrics, generate_style_boost. NAS main.py가 httpx로 forward하여 호출. Plan-B-Music Phase 2. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
131
services/music-render/providers/sync_ops.py
Normal file
131
services/music-render/providers/sync_ops.py
Normal file
@@ -0,0 +1,131 @@
|
||||
"""Sync Suno API helpers — main.py FastAPI sync endpoints에서 호출.
|
||||
|
||||
NAS music-lab/app/suno_provider.py의 sync 함수들 이식.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from typing import Optional
|
||||
|
||||
import requests
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
SUNO_BASE_URL = "https://api.sunoapi.org/api/v1"
|
||||
SUNO_API_KEY = os.getenv("SUNO_API_KEY", "")
|
||||
|
||||
|
||||
def _headers() -> dict:
|
||||
return {
|
||||
"Authorization": f"Bearer {SUNO_API_KEY}",
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
|
||||
def generate_lyrics(prompt: str) -> Optional[dict]:
|
||||
"""Suno 가사 생성 API — 폴링 결과 반환."""
|
||||
if not SUNO_API_KEY:
|
||||
return None
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{SUNO_BASE_URL}/lyrics",
|
||||
headers=_headers(),
|
||||
json={"prompt": prompt[:200]},
|
||||
timeout=30,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
return None
|
||||
body = resp.json()
|
||||
if body.get("code") != 200:
|
||||
return body
|
||||
task_id = body.get("data", {}).get("taskId", "")
|
||||
if not task_id:
|
||||
return body
|
||||
return _poll_lyrics(task_id)
|
||||
except Exception as e:
|
||||
logger.warning("Suno lyrics API error: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
def _poll_lyrics(lyrics_task_id: str) -> Optional[dict]:
|
||||
for _ in range(15):
|
||||
time.sleep(3)
|
||||
try:
|
||||
resp = requests.get(
|
||||
f"{SUNO_BASE_URL}/lyrics/record-info",
|
||||
headers=_headers(),
|
||||
params={"taskId": lyrics_task_id},
|
||||
timeout=15,
|
||||
)
|
||||
if resp.status_code != 200:
|
||||
continue
|
||||
body = resp.json()
|
||||
data = body.get("data", {})
|
||||
if data.get("status") == "complete":
|
||||
items = data.get("data") or data.get("sunoData") or []
|
||||
if items and isinstance(items, list):
|
||||
return {
|
||||
"id": lyrics_task_id,
|
||||
"status": "complete",
|
||||
"text": items[0].get("text", ""),
|
||||
"title": items[0].get("title", ""),
|
||||
}
|
||||
return {"id": lyrics_task_id, "status": "complete", "text": ""}
|
||||
except Exception:
|
||||
continue
|
||||
return None
|
||||
|
||||
|
||||
def get_credits() -> Optional[dict]:
|
||||
if not SUNO_API_KEY:
|
||||
return None
|
||||
for path in ["/generate/credit", "/get-credits"]:
|
||||
try:
|
||||
resp = requests.get(f"{SUNO_BASE_URL}{path}", headers=_headers(), timeout=15)
|
||||
if resp.status_code == 200:
|
||||
body = resp.json()
|
||||
data = body.get("data", body)
|
||||
if isinstance(data, (int, float)):
|
||||
return {"credits_left": int(data)}
|
||||
return data
|
||||
except Exception as e:
|
||||
logger.warning("Suno credits API error (%s): %s", path, e)
|
||||
return None
|
||||
|
||||
|
||||
def get_timestamped_lyrics(suno_task_id: str, suno_id: str) -> Optional[dict]:
|
||||
if not SUNO_API_KEY:
|
||||
return None
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{SUNO_BASE_URL}/generate/get-timestamped-lyrics",
|
||||
headers=_headers(),
|
||||
json={"taskId": suno_task_id, "audioId": suno_id},
|
||||
timeout=30,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
body = resp.json()
|
||||
return body.get("data", body)
|
||||
except Exception as e:
|
||||
logger.warning("Timestamped lyrics error: %s", e)
|
||||
return None
|
||||
|
||||
|
||||
def generate_style_boost(content: str) -> Optional[dict]:
|
||||
if not SUNO_API_KEY:
|
||||
return None
|
||||
try:
|
||||
resp = requests.post(
|
||||
f"{SUNO_BASE_URL}/style/generate",
|
||||
headers=_headers(),
|
||||
json={"content": content},
|
||||
timeout=30,
|
||||
)
|
||||
if resp.status_code == 200:
|
||||
body = resp.json()
|
||||
return body.get("data", body)
|
||||
except Exception as e:
|
||||
logger.warning("Style boost error: %s", e)
|
||||
return None
|
||||
Reference in New Issue
Block a user