o2o-site-AEO/solution/backend/services/external/suno.py

"""Suno API — 가사를 받아 40초짜리 곡 한 편을 만든다.

    API 문서   https://docs.sunoapi.org
    가사       services/external/gemini_text.generate_song  (여기는 작곡만 한다)
    쓰는 곳    services/song_service                        (잡 흐름·저장)

★ **콜백을 쓰지 않고 폴링한다.**
  Suno 는 완료 시 `callBackUrl` 로 POST 를 보내 주는데, 그러려면 Suno 쪽에서 우리 백엔드에
  닿아야 한다. 이 서버는 로컬(:9800)이거나 사내망(킹서버)이라 그런 주소가 없다 —
  콜백을 믿게 만들어 두면 "요청은 성공했는데 결과가 영영 안 옴" 이 되고, 그건 화면상
  아무 일도 안 일어나는 종류의 실패다. 그래서 `generate` 로 taskId 를 받고 `record-info` 를
  직접 물어본다. 잡 워커에서 도는 코드라 몇 분 기다리는 것이 문제가 되지 않는다.
  (`callBackUrl` 은 API 가 필수로 요구해서 값만 채워 보낸다. 우리는 그 주소를 듣지 않는다.)

★ **한 요청에 곡이 두 편 온다.** Suno 는 같은 가사로 변주 두 개를 만들어 준다(sunoData 배열).
  우리는 **첫 번째 한 곡만** 쓴다 — 사장님에게 고르라고 묻는 화면이 없고, 두 곡을 다 실으면
  손님이 무엇을 듣는지 우리도 모른다.

★ **오디오 주소는 만료된다.** 여기서 돌려주는 `audio_url` 을 그대로 사이트에 싣지 않는다.
  받는 쪽(song_service)이 파일을 내려받아 우리 쪽에 보관한다.
"""
import asyncio
from typing import Any, Optional

import httpx

from common.logger import LOG
from config.server_configs import external_api_config

BASE_URL = "https://api.sunoapi.org/api/v1"

# 실측(참고 프로젝트 o2o-castad-backend): 스트림 주소는 30~40초, 내려받을 수 있는 주소는 2~3분.
# 우리는 파일을 받아야 하므로 뒤쪽 기준으로 기다린다.
POLL_INTERVAL_SEC = 10
POLL_TIMEOUT_SEC = 300

REQUEST_TIMEOUT = httpx.Timeout(60.0, connect=10.0)

# 40초짜리를 만든다. 헤더의 작은 플레이어에서 듣는 곡이라 길 이유가 없고,
# 길수록 생성 시간과 요금이 같이 는다.
SONG_SECONDS = 40
MODEL = "V5"


class SunoNotConfigured(RuntimeError):
    """SUNO_API_KEY 가 없다 — 노래만 건너뛰고 발행은 계속한다."""


class SunoError(RuntimeError):
    """호출 실패·거절. 잡의 last_error 로 남는다."""


def is_configured() -> bool:
    return bool((external_api_config.suno_api_key or "").strip())


def _headers() -> dict:
    return {
        "Authorization": f"Bearer {external_api_config.suno_api_key}",
        "Content-Type": "application/json",
    }


async def generate(lyrics: str, *, title: str, style: str, client: httpx.AsyncClient) -> str:
    """작곡 요청. taskId 를 돌려준다.

    ★ `customMode=True` 다 — prompt 를 '주제' 가 아니라 **가사 그대로** 쓰라는 뜻이다.
      false 로 두면 Suno 가 가사를 자기가 새로 쓴다. 우리는 이 숙소의 사실로 쓴 가사를
      넘기는 것이므로, 그걸 버리면 이 기능의 의미가 없다.
    """
    if not is_configured():
        raise SunoNotConfigured("SUNO_API_KEY 미설정")

    body = {
        "model": MODEL,
        "customMode": True,
        "instrumental": False,
        # 길이는 API 파라미터가 아니라 프롬프트로 지시한다(참고 프로젝트와 같은 방식).
        "prompt": f"[Song Duration: Around {SONG_SECONDS} seconds]\n{lyrics}",
        "title": title[:80],
        "style": style,
        # 듣지 않는 주소다(머리주석). 비워서 보내면 거절당한다.
        "callBackUrl": external_api_config.suno_callback_url or "https://example.com/api/suno/callback",
    }

    try:
        res = await client.post(f"{BASE_URL}/generate", headers=_headers(), json=body, timeout=REQUEST_TIMEOUT)
    except httpx.HTTPError as ex:
        raise SunoError(f"generate 호출 실패: {type(ex).__name__}: {ex}") from ex

    if res.status_code != 200:
        raise SunoError(f"generate HTTP {res.status_code}: {res.text[:300]}")

    data = res.json() or {}
    if data.get("code") != 200:
        raise SunoError(f"generate 거절: {data.get('msg')}")

    task_id = ((data.get("data") or {}).get("taskId"))
    if not task_id:
        raise SunoError(f"generate 응답에 taskId 가 없다: {str(data)[:300]}")
    return task_id


def _first_clip(payload: dict) -> Optional[dict]:
    """완성된 클립 하나. 아직이면 None.

    ★ 상태 문자열을 믿기 전에 **주소가 실제로 있는지** 본다. SUCCESS 인데 audioUrl 이
      아직 비어 오는 응답을 참고 프로젝트가 겪었다(스트림만 먼저 나오는 구간).
    """
    data = (payload or {}).get("data") or {}
    status = (data.get("status") or "").upper()
    if status in {"CREATE_TASK_FAILED", "GENERATE_AUDIO_FAILED", "CALLBACK_EXCEPTION", "SENSITIVE_WORD_ERROR"}:
        raise SunoError(f"작곡 실패: {status} {data.get('errorMessage') or ''}".strip())

    clips = ((data.get("response") or {}).get("sunoData")) or []
    for clip in clips:
        if clip.get("audioUrl") or clip.get("sourceAudioUrl"):
            return clip
    return None


async def wait_for_clip(task_id: str, *, client: httpx.AsyncClient) -> dict[str, Any]:
    """완성될 때까지 물어본다. 돌려주는 것은 첫 클립 하나.

    ★ 상한(POLL_TIMEOUT_SEC)을 둔다. Suno 가 영영 안 끝내는 경우 잡이 그대로 매달리면
      워커 한 자리를 계속 차지한다 — 노래 하나 때문에 다른 사업장의 수집이 멈춘다.
    """
    waited = 0
    while waited < POLL_TIMEOUT_SEC:
        await asyncio.sleep(POLL_INTERVAL_SEC)
        waited += POLL_INTERVAL_SEC

        try:
            res = await client.get(
                f"{BASE_URL}/generate/record-info",
                headers=_headers(),
                params={"taskId": task_id},
                timeout=REQUEST_TIMEOUT,
            )
            res.raise_for_status()
        except httpx.HTTPError as ex:
            # 폴링 한 번 실패는 실패가 아니다 — 다음 차례에 다시 묻는다.
            LOG.w(f"[suno] 상태 조회 실패(계속 기다린다) task={task_id}: {type(ex).__name__}")
            continue

        clip = _first_clip(res.json() or {})
        if clip:
            LOG.i(f"[suno] 작곡 완료 task={task_id} ({waited}초)")
            return clip

    raise SunoError(f"{POLL_TIMEOUT_SEC}초 안에 완성되지 않았다 task={task_id}")


async def download(url: str, *, client: httpx.AsyncClient) -> bytes:
    """오디오 파일을 받아 온다. 보관은 부르는 쪽이 한다."""
    res = await client.get(url, timeout=httpx.Timeout(180.0, connect=10.0), follow_redirects=True)
    res.raise_for_status()
    return res.content