From 900f45c2ff120fba209e6def1be5c0f05e9ad3b8 Mon Sep 17 00:00:00 2001
From: gahusb <bgg8988@gmail.com>
Date: Tue, 19 May 2026 04:58:23 +0900
Subject: [PATCH] =?UTF-8?q?feat(music-render):=20providers/sync=5Fops.py?=
 =?UTF-8?q?=20=E2=80=94=20sync=20Suno=20helpers=20(SP-5)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

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>
---
 services/music-render/providers/sync_ops.py | 131 ++++++++++++++++++++
 1 file changed, 131 insertions(+)
 create mode 100644 services/music-render/providers/sync_ops.py

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