app.py

import requests
import gradio as gr
import os
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 {}

# URL 치환 함수
def replace_speak_url(url):
    if not url:
        return None
    return url.replace("https://create-test.icepeak.ai", PRIVATE_SPACE_URL)

# 음성 생성 함수
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}",
        "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.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

        # 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()
            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__":
    interface.launch()