Update generate_speech demo

app.py

@@ -1,70 +1,189 @@
-import gradio as gr
 import requests
+import gradio as gr
 import os
-import json
+import time
+
+# 환경 변수 초기화 및 검증
+def initialize_environment():
+    private_space_url = os.getenv("PRIVATE_SPACE_URL")
+    hf_token = os.getenv("HF_TOKEN")
+
+    if not private_space_url:
+        raise EnvironmentError("PRIVATE_SPACE_URL 환경 변수가 설정되지 않았습니다.")
+    if not hf_token:
+        raise EnvironmentError("HF_TOKEN 환경 변수가 설정되지 않았습니다.")
+
+    return private_space_url, hf_token
+
+PRIVATE_SPACE_URL, HF_TOKEN = initialize_environment()
+
+# Base URL로 GET 요청 보내는 함수
+def test_base_url():
+    headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+    print(f"[DEBUG] Sending GET request to base URL: {PRIVATE_SPACE_URL}")
+    print(f"[DEBUG] HF Token: {HF_TOKEN}")
+    try:
+        response = requests.get(PRIVATE_SPACE_URL, headers=headers)
+        print(f"[DEBUG] Base URL Response Status Code: {response.status_code}")
+    except requests.exceptions.RequestException as e:
+        print(f"[ERROR] Base URL 요청 중 오류 발생: {e}")
+
+# Actor ID 조회 함수
+def fetch_actor_ids():
+    url = f"{PRIVATE_SPACE_URL}/api/actor"
+    headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+    try:
+        response = requests.get(url, headers=headers)
+        response.raise_for_status()
+        data = response.json()
+
+        # Actor 리스트에서 이름과 ID 추출
+        return {
+            actor["name"]["en"]: actor["actor_id"]
+            for actor in data.get("result", [])
+        }
+    except requests.exceptions.RequestException as e:
+        print(f"[ERROR] Actor ID 조회 중 오류 발생: {e}")
+        return {}
 
-PRIVATE_SPACE_URL = os.getenv("PRIVATE_SPACE_URL")
-HF_TOKEN = os.getenv("HF_TOKEN")
+# URL 치환 함수
+def replace_speak_url(url):
+    if not url:
+        return None
+    return url.replace("https://create-test.icepeak.ai", PRIVATE_SPACE_URL)
 
-def fetch_actor_data():
-    if not PRIVATE_SPACE_URL:
-        return "Error: PRIVATE_SPACE_URL not found. Check Secrets."
-    if not HF_TOKEN:
-        return "Error: HF_TOKEN not found. Check Secrets."
+# 음성 생성 함수
+def generate_speech(text, actor_name, lang="en", speed_x=1.0, volume=100):
+    actor_ids = fetch_actor_ids()
+    if not actor_ids:
+        return "Actor 목록을 가져오는 데 실패했습니다.", None
 
+    actor_id = actor_ids.get(actor_name)
+    if not actor_id:
+        return "선택한 Actor를 찾을 수 없습니다.", None
+
+    url = f"{PRIVATE_SPACE_URL}/api/speak"
     headers = {
-        "Authorization": f"Bearer {HF_TOKEN}"
+        "Authorization": f"Bearer {HF_TOKEN}",
+        "Content-Type": "application/json"
+    }
+    payload = {
+        "actor_id": actor_id,
+        "lang": lang,
+        "text": text,
+        "speed_x": speed_x,
+        "volume": volume,
+        "tts_mode": "actor"
     }
 
+    print(f"[DEBUG] Sending speech generation request to: {url}")
+    print(f"[DEBUG] Payload: {payload}")
     try:
-        response = requests.get(f"{PRIVATE_SPACE_URL}/api/actor", headers=headers)
+        response = requests.post(url, json=payload, headers=headers)
+        print(f"[DEBUG] Response Status Code: {response.status_code}")
+        print(f"[DEBUG] Response Content: {response.text}")
+        response.raise_for_status()
+        data = response.json()
+        print(f"[DEBUG] Response Data: {data}")
+
+        speak_url = replace_speak_url(data.get("result", {}).get("speak_url"))
+        if not speak_url:
+            return "오류: 유효한 speak_url을 반환하지 않았습니다.", None
 
-        if response.status_code == 200:
-            # 응답 JSON 파싱
+        # Polling for audio generation completion
+        audio_url = poll_audio_url(speak_url)
+        if not audio_url:
+            return "오류: 오디오 다운로드 URL을 가져올 수 없습니다.", None
+
+        # Download the audio file
+        audio_content = download_audio(audio_url)
+        if not audio_content:
+            return "오류: 음성 파일을 다운로드할 수 없습니다.", None
+
+        # Save the audio content to a temporary file
+        audio_file_path = "temp_audio.wav"
+        with open(audio_file_path, "wb") as audio_file:
+            audio_file.write(audio_content)
+
+        # Print debug information
+        file_size = os.path.getsize(audio_file_path)
+        print(f"[DEBUG] Saved audio file: {audio_file_path}, Size: {file_size} bytes")
+
+        return "음성이 성공적으로 생성되었습니다.", audio_file_path
+    except requests.exceptions.RequestException as e:
+        print(f"[ERROR] 음성 생성 중 오류 발생: {e}")
+        return f"오류: {str(e)}", None
+
+# Polling 함수
+def poll_audio_url(speak_url, timeout=30, interval=2):
+    headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+    start_time = time.time()
+
+    while time.time() - start_time < timeout:
+        try:
+            print(f"[DEBUG] Polling speak URL: {speak_url}")
+            response = requests.get(speak_url, headers=headers)
+            response.raise_for_status()
             data = response.json()
-            
-            # 기대 구조: data["data"]["result"] => list of actor items
-            actor_ids = []
-            if (
-                "data" in data
-                and isinstance(data["data"], dict)
-                and "result" in data["data"]
-                and isinstance(data["data"]["result"], list)
-            ):
-                for item in data["data"]["result"]:
-                    if "actor_id" in item:
-                        actor_ids.append(item["actor_id"])
-                    if len(actor_ids) >= 10:
-                        break
-
-            if actor_ids:
-                return f"Found actor_ids: {actor_ids}"
-            else:
-                # actor_id가 없을 경우 구조 파악용 반환
-                top_level_keys = list(data["data"].keys()) if "data" in data and isinstance(data["data"], dict) else []
-                first_item = data["data"]["result"][0] if "data" in data and "result" in data["data"] and len(data["data"]["result"]) > 0 else {}
-                return (
-                    "No actor_id found.\n\n"
-                    f"Keys in data['data']: {top_level_keys}\n\n"
-                    "First item in data['data']['result']:\n" +
-                    json.dumps(first_item, indent=2, ensure_ascii=False)
-                )
-        else:
-            return f"Error {response.status_code}\nResponse Text:\n{response.text}"
-    except Exception as e:
-        return f"Request failed: {str(e)}"
-
-iface = gr.Interface(
-    fn=fetch_actor_data,
-    inputs=None,
-    outputs="text",
-    title="Fetch Actor Data",
-    description=(
-        "Fetches actor data from the private Hugging Face Space.\n"
-        "Assumes structure: {status_code:..., data:{count:..., result:[{actor_id:...}]}}\n"
-        "If no actor_id found, shows partial keys for debugging."
-    )
+            status = data.get("result", {}).get("status")
+            if status == "done":
+                audio_info = data.get("result", {}).get("audio", {})
+                audio_url = replace_speak_url(audio_info.get("url"))
+                print(f"[DEBUG] Audio URL: {audio_url}")
+                return audio_url
+        except requests.exceptions.RequestException as e:
+            print(f"[DEBUG] Polling attempt failed: {e}")
+        time.sleep(interval)
+
+    print("[ERROR] Polling timed out.")
+    return None
+
+# Audio 다운로드 함수
+def download_audio(audio_url):
+    headers = {"Authorization": f"Bearer {HF_TOKEN}"}
+    try:
+        print(f"[DEBUG] Downloading audio from: {audio_url}")
+        response = requests.get(audio_url, headers=headers, stream=True)
+        response.raise_for_status()
+        return response.content
+    except requests.exceptions.RequestException as e:
+        print(f"[ERROR] Audio 다운로드 중 오류 발생: {e}")
+        return None
+
+# Gradio 인터페이스 함수
+def interface_function(text, actor_name, lang, speed_x, volume):
+    result_message, audio_file_path = generate_speech(text, actor_name, lang, speed_x, volume)
+    if audio_file_path:
+        return result_message, audio_file_path
+    return result_message, None
+
+# Fetch actors to populate dropdown
+actors = fetch_actor_ids()
+actor_names = list(actors.keys())
+
+if not actor_names:
+    print("[WARNING] Actor 목록을 가져올 수 없어 기본 옵션이 설정됩니다.")
+
+# Base URL 테스트
+test_base_url()
+
+# Gradio 인터페이스 생성
+interface = gr.Interface(
+    fn=interface_function,
+    inputs=[
+        gr.Textbox(label="텍스트 입력", placeholder="여기에 음성을 합성할 텍스트를 입력하세요."),
+        gr.Dropdown(choices=actor_names, label="Actor 선택", interactive=True),
+        gr.Dropdown(choices=["en", "ko"], value="en", label="언어"),
+        gr.Slider(minimum=0.5, maximum=2.0, step=0.1, value=1.0, label="속도"),
+        gr.Slider(minimum=50, maximum=200, step=10, value=100, label="볼륨"),
+    ],
+    outputs=[
+        gr.Textbox(label="결과 메시지"),
+        gr.Audio(label="생성된 음성"),
+    ],
+    title="음성 생성 데모",
+    description="쿼리 파라미터가 적용된 Actor 조회 후 음성을 생성합니다.",
 )
 
 if __name__ == "__main__":
-    iface.launch()
+    interface.launch()