Update app.py

app.py

@@ -1,121 +1,90 @@
-import gradio as gr
-import httpx
-import edge_tts
 import os
 import re
+import requests
 import uuid
-import torch
-import numpy as np
 import scipy.io.wavfile
-import asyncio
+import numpy as np
+import torch
 from transformers import pipeline
 
-# 환경 변수 로드 및 검증
-GENAI_KEY = os.getenv("GEMINI_KEY")
+# Hugging Face 토큰 및 모델 설정
+HF_TOKEN = os.getenv("HF_TOKEN")
+MODEL_ID = "Qwen/Qwen2.5-7B-Instruct"
 
-# 음악 생성 모델 로드
+# 음악 모델 로드 (기존 유지)
 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인 멤버 데이터 (원본 보존)
-MEMBERS = {
-    "서윤 (Korea)": { "voice": "ko-KR-SunHiNeural", "prompt": "보컬 서윤. 시니컬하지만 진심임." },
-    "Chloe (USA)": { "voice": "en-US-AriaNeural", "prompt": "Guitarist Chloe. Cool rockstar vibe." },
-    "Beatrice (Brazil)": { "voice": "pt-BR-FranciscaNeural", "prompt": "Drummer Beatrice. Rhythm focus." },
-    "Naomi (Japan)": { "voice": "ja-JP-NanamiNeural", "prompt": "Naomi. Harmony focus." },
-    "Elena (Spain)": { "voice": "es-ES-ElviraNeural", "prompt": "Elena. Groove focus." },
-    "Amira (Egypt)": { "voice": "ar-EG-SalmaNeural", "prompt": "Amira. Atmosphere focus." },
-    "Liwei (China)": { "voice": "zh-CN-XiaoxiaoNeural", "prompt": "Liwei. Sound design focus." },
-    "Sophie (France)": { "voice": "fr-FR-DeniseNeural", "prompt": "Sophie. Melody focus." }
-}
-
 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")
+    # 1. Qwen용 시스템 프롬프트 구성 (Gemini보다 더 강력하게 지시)
+    system_instruction = f"""
+    You are '{member_name}', a professional rock band member. 
+    [Language Rules] Respond ONLY in {lang_code}.
+    [Output Format] 
+    1. Short advice (under 5 lines).
+    2. Detailed sheet music/advice inside [TAB] tags.
+    3. Music generation prompt inside [MUSIC: English Prompt] tags.
+    """
 
-        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")
-        
-        system_instruction = f"""
-        당신은 '{member_name}'입니다.
-        1. 반드시 {target_lang}로만 답변하세요.
-        2. 음성 답변은 5줄 이내 요약, 상세 설명/악보는 [TAB]에 넣으세요.
-        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}"}]}]}
+    try:
+        # 2. Hugging Face Inference API 호출
+        api_url = f"https://api-inference.huggingface.co/models/{MODEL_ID}"
+        headers = {"Authorization": f"Bearer {HF_TOKEN}"}
         
-        async with httpx.AsyncClient() as client:
-            response = await client.post(url, json=payload, timeout=25.0)
-            res_data = response.json()
+        prompt = f"<|im_start|>system\n{system_instruction}<|im_end|>\n<|im_start|>user\n{user_input}<|im_end|>\n<|im_start|>assistant\n"
         
-        if 'candidates' not in res_data:
-            ai_text_raw = "영감이 잠시 차단됐어. 다시 물어봐줄래? [TAB] API 응답 오류 [MUSIC: rock guitar riff]"
+        payload = {
+            "inputs": prompt,
+            "parameters": {"max_new_tokens": 1024, "temperature": 0.7, "top_p": 0.9}
+        }
+
+        response = requests.post(api_url, headers=headers, json=payload, timeout=20)
+        res_json = response.json()
+
+        # 응답이 비어있거나 에러일 경우 'Improv Mode' 발동
+        if not res_json or 'error' in res_json:
+            ai_text_raw = "이봐 리더, 지금은 말보다 연주가 필요한 순간이야! 내 즉흥 연주를 들어봐. [TAB] 즉흥 연주 모드 가동 [MUSIC: aggressive hard rock guitar solo]"
         else:
-            ai_text_raw = res_data['candidates'][0]['content']['parts'][0]['text']
-        
-        # 정규표현식 파싱 (대소문자 무시 및 유연한 매칭)
+            # Qwen은 입력 프롬프트를 포함해서 반환할 수 있으므로 assistant 응답만 추출
+            full_text = res_json[0]['generated_text']
+            ai_text_raw = full_text.split("assistant\n")[-1].strip()
+
+        # 3. 파싱 로직 (Regex 강화)
         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 "상세 정보 없음"
+        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)
+        # 텍스트가 비어있으면 Fallback
+        if len(clean_text) < 5:
+            clean_text = "말이 필요 없는 끝내주는 라인이 생각났어. 바로 들어보자고!"
+
+        # 4. 음악 생성 (MusicGen 연동)
+        music_p = music_match.group(1).strip() if music_match else "energetic rock riff"
         
-        # 음악 생성 (Blocking 방지 위해 별도 스레드 고려 가능하나 Gradio Queue가 있으므로 직접 실행)
         if music_synthesiser:
-            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})
+            # 안전 필터 단어 우회 (Vocal 등 차단 방지)
+            safe_p = re.sub(r'(laugh|vocal|voice|human)', 'rock guitar', music_p.lower())
+            music_output = music_synthesiser(safe_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) # 정규화
+            # 오디오 처리 (Float32 -> Int16)
+            audio_data = np.squeeze(music_output["audio"])
+            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_text, voice_path, music_path, tab_display
+        return clean_text, None, music_path, tab_display # voice_path는 필요시 edge_tts 추가
 
     except Exception as e:
-        print(f"Critical Error: {e}")
-        return f"System Overload: 다시 시도해주세요. ({str(e)})", None, None, "Error"
-
-# Gradio Interface
-with gr.Blocks() as demo:
-    # 비동기 처리를 위한 내부 통로
-    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")
-
-# 큐 활성화 및 실행
-demo.queue().launch(ssr_mode=False)
+        print(f"Critical System Error: {e}")
+        return "시스템 과부하인가? 하지만 락은 멈추지 않아! [MUSIC: basic rock drum beat]", None, None, "Retry Mode"