diff --git a/services/music-render/providers/sync_ops.py b/services/music-render/providers/sync_ops.py new file mode 100644 index 0000000..f7c86ec --- /dev/null +++ b/services/music-render/providers/sync_ops.py @@ -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