Update app.py

app.py

@@ -1,21 +1,26 @@
 import gradio as gr
-import requests
+import httpx
 import edge_tts
 import os
 import re
+import uuid
+import torch
+import numpy as np
 import scipy.io.wavfile
+import asyncio
 from transformers import pipeline
 
+# 환경 변수 로드 및 검증
 GENAI_KEY = os.getenv("GEMINI_KEY")
 
-# 음악 모델 로드
+# 음악 생성 모델 로드
 try:
     music_synthesiser = pipeline("text-to-audio", "facebook/musicgen-small")
 except Exception as e:
     print(f"Music Model Load Error: {e}")
     music_synthesiser = None
 
-# 8인 멤버 데이터 (절대 축소 없음)
+# 8인 멤버 데이터 (원본 보존)
 MEMBERS = {
     "서윤 (Korea)": { "voice": "ko-KR-SunHiNeural", "prompt": "보컬 서윤. 시니컬하지만 진심임." },
     "Chloe (USA)": { "voice": "en-US-AriaNeural", "prompt": "Guitarist Chloe. Cool rockstar vibe." },
@@ -28,7 +33,15 @@ MEMBERS = {
 }
 
 async def band_consulting(user_input, member_name, lang_code):
+    # 리퀘스트별 고유 ID 생성 (파일 충돌 방지)
+    req_id = str(uuid.uuid4())[:8]
+    voice_path = f"/tmp/v_{req_id}.mp3"
+    music_path = f"/tmp/m_{req_id}.wav"
+    
     try:
+        if not GENAI_KEY:
+            raise ValueError("API Key is missing in Environment Variables")
+
         member_data = MEMBERS.get(member_name, MEMBERS["서윤 (Korea)"])
         lang_map = {"ko":"Korean","en":"English","ja":"Japanese","pt":"Portuguese","es":"Spanish","ar":"Arabic","zh":"Chinese","fr":"French"}
         target_lang = lang_map.get(lang_code, "Korean")
@@ -40,57 +53,69 @@ async def band_consulting(user_input, member_name, lang_code):
         3. 마지막에 [MUSIC: 영어 프롬프트]를 포함하세요.
         """
         
+        # 비동기 HTTP 클라이언트 사용 (Blocking 방지)
         url = f"https://generativelanguage.googleapis.com/v1beta/models/gemini-1.5-flash:generateContent?key={GENAI_KEY}"
         payload = {"contents": [{"parts": [{"text": f"{system_instruction}\n질문: {user_input}"}]}]}
         
-        response = requests.post(url, json=payload, timeout=20)
-        res_data = response.json()
+        async with httpx.AsyncClient() as client:
+            response = await client.post(url, json=payload, timeout=25.0)
+            res_data = response.json()
         
         if 'candidates' not in res_data:
-            clean_text = "영감이 꼬였어. 다시 시도해줘!"
-            tab_display = "데이터 오류"
-            music_p = "rock music riff"
+            ai_text_raw = "영감이 잠시 차단됐어. 다시 물어봐줄래? [TAB] API 응답 오류 [MUSIC: rock guitar riff]"
         else:
             ai_text_raw = res_data['candidates'][0]['content']['parts'][0]['text']
-            tab_match = re.search(r'\[TAB\](.*?)(\[|$)', ai_text_raw, re.DOTALL)
-            music_match = re.search(r'\[MUSIC:(.*?)\]', ai_text_raw, re.IGNORECASE)
-            tab_display = tab_match.group(1).strip() if tab_match else "상세 악보 없음"
-            clean_text = re.sub(r'\[TAB\].*?(\[|$)', '', ai_text_raw, flags=re.DOTALL)
-            clean_text = re.sub(r'\[MUSIC:.*?\]', '', clean_text).strip()
-            music_p = music_match.group(1).strip() if music_match else "rock guitar riff"
-
-        # 음성 생성
-        tts_final = "\n".join(clean_text.split('\n')[:5])
-        voice_path = f"/tmp/v_{member_name.split()[0]}.mp3"
-        communicate = edge_tts.Communicate(re.sub(r'[\*\#\-\_\~\|]', '', tts_final), member_data['voice'])
+        
+        # 정규표현식 파싱 (대소문자 무시 및 유연한 매칭)
+        tab_match = re.search(r'\[TAB\](.*?)(\[|$)', ai_text_raw, re.DOTALL | re.IGNORECASE)
+        music_match = re.search(r'\[MUSIC:(.*?)\]', ai_text_raw, re.IGNORECASE)
+        
+        tab_display = tab_match.group(1).strip() if tab_match else "상세 정보 없음"
+        clean_text = re.sub(r'\[TAB\].*?(\[|$)', '', ai_text_raw, flags=re.DOTALL | re.IGNORECASE)
+        clean_text = re.sub(r'\[MUSIC:.*?\]', '', clean_text, flags=re.IGNORECASE).strip()
+        
+        # TTS 음성 생성
+        tts_text = "\n".join(clean_text.split('\n')[:5])
+        communicate = edge_tts.Communicate(re.sub(r'[\*\#\-\_\~\|]', '', tts_text), member_data['voice'])
         await communicate.save(voice_path)
         
-        # 음악 생성
-        music_path = None
+        # 음악 생성 (Blocking 방지 위해 별도 스레드 고려 가능하나 Gradio Queue가 있으므로 직접 실행)
         if music_synthesiser:
-            try:
-                # 필터 단어 우회 및 길이 조절 (12초)
-                safe_p = music_p.lower()
-                if any(x in safe_p for x in ['laugh', 'vocal', 'voice', 'human']):
-                    safe_p = "heavy rock electric guitar riff"
-                
-                music_output = music_synthesiser(safe_p, forward_params={"max_new_tokens": 512})
-                if music_output and "audio" in music_output:
-                    music_path = f"/tmp/m_{member_name.split()[0]}.wav"
-                    scipy.io.wavfile.write(music_path, music_output["sampling_rate"], music_output["audio"][0].T)
-            except:
-                music_path = None
+            music_p = music_match.group(1).strip() if music_match else "rock music"
+            # 가이드라인 위반 단어 우회
+            if any(x in music_p.lower() for x in ['laugh', 'vocal', 'voice', 'human']):
+                music_p = "energetic rock electric guitar solo"
+            
+            # [MUSICGEN 실행] - 512 토큰 (약 12초)
+            music_output = music_synthesiser(music_p, forward_params={"max_new_tokens": 512})
+            
+            # 오디오 데이터 처리 (Float32 -> Int16)
+            audio_tensor = music_output["audio"]
+            if torch.is_tensor(audio_tensor):
+                audio_data = audio_tensor.cpu().numpy()
+            else:
+                audio_data = np.array(audio_tensor)
+            
+            audio_data = np.squeeze(audio_data) # 차원 압축
+            audio_int16 = (audio_data * 32767).astype(np.int16) # 정규화
+            scipy.io.wavfile.write(music_path, music_output["sampling_rate"], audio_int16)
+        else:
+            music_path = None
 
-        return tts_final, voice_path, music_path, tab_display
+        return tts_text, voice_path, music_path, tab_display
 
     except Exception as e:
-        return f"System Overload: {str(e)}", None, None, "Error"
+        print(f"Critical Error: {e}")
+        return f"System Overload: 다시 시도해주세요. ({str(e)})", None, None, "Error"
 
+# Gradio Interface
 with gr.Blocks() as demo:
-    # API 전용 최소 UI (모든 로직은 predict로 연결)
-    i1 = gr.Textbox(); i2 = gr.Textbox(); i3 = gr.Textbox()
-    o1 = gr.Textbox(); o2 = gr.Audio(); o3 = gr.Audio(); o4 = gr.Textbox()
-    btn = gr.Button("GO"); btn.click(band_consulting, [i1, i2, i3], [o1, o2, o3, o4], api_name="predict")
+    # 비동기 처리를 위한 내부 통로
+    i1 = gr.Textbox(visible=False); i2 = gr.Textbox(visible=False); i3 = gr.Textbox(visible=False)
+    o1 = gr.Textbox(visible=False); o2 = gr.Audio(visible=False); o3 = gr.Audio(visible=False); o4 = gr.Textbox(visible=False)
+    
+    btn = gr.Button("GO", visible=False) # JS에서 호출용
+    btn.click(band_consulting, [i1, i2, i3], [o1, o2, o3, o4], api_name="predict")
 
-# 큐 시스템을 적용하되 ssr_mode는 False로 설정하여 asyncio 에러 방지
+# 큐 활성화 및 실행
 demo.queue().launch(ssr_mode=False)