test_analyze_voice.py

"""
Gradio Test Page for analyze_voice_logic Function
Tests the core voice analysis logic independently
"""

import gradio as gr
import asyncio
import uuid
from datetime import datetime, timedelta
from pathlib import Path
import sys
import logging
from contextlib import AsyncExitStack

# Import from backend
from backend import analyze_voice_logic, voicekit_session, session_stack
import backend

# Setup logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# MCP initialization flag
mcp_initialized = False


async def initialize_mcp():
    """Initialize VoiceKit MCP connection if not already initialized"""
    global mcp_initialized
    
    if mcp_initialized and backend.voicekit_session is not None:
        logger.info("MCP already initialized")
        return
    
    logger.info("Initializing VoiceKit MCP connection...")
    backend.session_stack = AsyncExitStack()
    
    try:
        from mcp.client.sse import sse_client
        from mcp.client.session import ClientSession
        
        voicekit_url = "https://mcp-1st-birthday-voicekit.hf.space/gradio_api/mcp/sse"
        read, write = await backend.session_stack.enter_async_context(sse_client(voicekit_url))
        backend.voicekit_session = await backend.session_stack.enter_async_context(
            ClientSession(read, write)
        )
        await backend.voicekit_session.initialize()
        
        # List available tools
        tools_result = await backend.voicekit_session.list_tools()
        logger.info(f"✓ VoiceKit MCP connected. Tools: {[t.name for t in tools_result.tools]}")
        mcp_initialized = True
        
    except Exception as e:
        logger.error(f"Failed to initialize VoiceKit MCP: {e}")
        backend.voicekit_session = None
        raise


def format_result_display(result: dict) -> str:
    """Format the analysis result for display"""
    if result.get("status") == "error":
        return f"❌ **에러 발생**\n\n{result.get('message', 'Unknown error')}"
    
    # Success case
    output = f"""
# ✅ 분석 완료!

## 📊 점수 결과
- **전체 점수**: {result.get('overall', 0)}/100
- **음높이 (Pitch)**: {result.get('pitch', 0)}/100
- **리듬 (Rhythm)**: {result.get('rhythm', 0)}/100
- **에너지 (Energy)**: {result.get('energy', 0)}/100
- **발음 (Pronunciation)**: {result.get('pronunciation', 0)}/100
- **대사 정확도 (Transcript)**: {result.get('transcript', 0)}/100

## 🎯 결과
- **카테고리**: {result.get('category', 'N/A')}
- **정답 여부**: {'✅ 정답!' if result.get('is_correct') else '❌ 오답'}

## 💡 조언
{result.get('advice', '조언이 없습니다.')}

## 🎨 힌트
{result.get('hints', '힌트가 없습니다.')}
"""
    return output


def get_available_dates():
    """Get list of available puzzle dates"""
    # Return last 30 days as options
    dates = []
    today = datetime.now()
    for i in range(30):
        date = today - timedelta(days=i)
        dates.append(date.strftime("%Y-%m-%d"))
    return dates


async def test_analyze_voice(audio_file, date: str, session_id: str, generate_new_session: bool):
    """
    Test the analyze_voice_logic function
    
    Args:
        audio_file: Uploaded audio file (gr.Audio component returns file path)
        date: Date string in YYYY-MM-DD format
        session_id: Session ID (UUID)
        generate_new_session: If True, generate a new session ID
    """
    try:
        # Initialize MCP if not already done
        await initialize_mcp()
        
        # Generate new session ID if requested
        if generate_new_session or not session_id:
            session_id = str(uuid.uuid4())
        
        # Read audio file
        if audio_file is None:
            return "❌ 오디오 파일을 업로드해주세요.", session_id
        
        # Read audio bytes
        audio_path = Path(audio_file)
        if not audio_path.exists():
            return f"❌ 오디오 파일을 찾을 수 없습니다: {audio_file}", session_id
        
        with open(audio_path, "rb") as f:
            audio_bytes = f.read()
        
        # print(f"📁 Audio file size: {len(audio_bytes)} bytes")
        # print(f"📅 Date: {date}")
        # print(f"🆔 Session ID: {session_id}")
        
        # Call the function
        result = await analyze_voice_logic(audio_bytes, date, session_id)
        
        # Format and return result
        formatted_result = format_result_display(result)
        
        return formatted_result, session_id
        
    except Exception as e:
        import traceback
        error_msg = f"❌ **예외 발생**\n\n```\n{str(e)}\n\n{traceback.format_exc()}\n```"
        return error_msg, session_id


def sync_test_analyze_voice(audio_file, date: str, session_id: str, generate_new_session: bool):
    """Synchronous wrapper for async function"""
    return asyncio.run(test_analyze_voice(audio_file, date, session_id, generate_new_session))


# Gradio Interface
with gr.Blocks(title="Analyze Voice Logic Test") as demo:
    gr.Markdown("# 🎤 Analyze Voice Logic Test")
    gr.Markdown("Test the `analyze_voice_logic` function from `backend.py`")
    
    with gr.Row():
        with gr.Column(scale=1):
            gr.Markdown("## 📥 입력")
            
            # Audio input
            audio_input = gr.Audio(
                label="오디오 파일",
                type="filepath",
                sources=["upload", "microphone"]
            )
            
            # Date selection
            date_input = gr.Dropdown(
                choices=get_available_dates(),
                label="날짜 선택",
                value=datetime.now().strftime("%Y-%m-%d"),
                allow_custom_value=True
            )
            
            # Session ID
            with gr.Row():
                session_id_input = gr.Textbox(
                    label="세션 ID (UUID)",
                    value=str(uuid.uuid4()),
                    interactive=True
                )
                generate_session_btn = gr.Checkbox(
                    label="새 세션 ID 생성",
                    value=False
                )
            
            # Submit button
            submit_btn = gr.Button("🚀 분석 시작", variant="primary", size="lg")
            
        with gr.Column(scale=2):
            gr.Markdown("## 📊 결과")
            result_output = gr.Markdown(
                "결과가 여기에 표시됩니다.",
                label="분석 결과"
            )
    
    # Info section
    with gr.Accordion("ℹ️ 사용 방법", open=False):
        gr.Markdown("""
        1. **오디오 파일**: 테스트할 음성 파일을 업로드하거나 마이크로 녹음하세요
        2. **날짜**: 퍼즐이 있는 날짜를 선택하세요 (DB에 해당 날짜의 퍼즐이 있어야 합니다)
        3. **세션 ID**: 자동 생성되거나 직접 입력할 수 있습니다
        4. **분석 시작**: 버튼을 클릭하여 분석을 시작합니다
        
        ### 주의사항
        - 이 페이지는 독립적으로 실행됩니다 (Backend 서버 불필요)
        - MCP 연결은 자동으로 초기화됩니다
        - 데이터베이스에 해당 날짜의 퍼즐이 있어야 합니다
        - VoiceKit MCP (외부 서버)와 Gemini API가 정상 작동해야 합니다
        """)
    
    with gr.Accordion("🔍 디버그 정보", open=False):
        gr.Markdown("""
        ### 상태 확인
        - Database: PostgreSQL 연결 확인 (`docker ps | grep voice-komentle-db`)
        - MCP: VoiceKit MCP 서버 (외부) - 자동 연결
        - Gemini: GOOGLE_API_KEY 환경변수 확인
        
        ### 필수 환경 변수
        ```bash
        DATABASE_URL=postgresql://user:pass@host:port/dbname
        GOOGLE_API_KEY=your_api_key_here
        ```
        """)
    
    # Event handler
    submit_btn.click(
        fn=sync_test_analyze_voice,
        inputs=[audio_input, date_input, session_id_input, generate_session_btn],
        outputs=[result_output, session_id_input]
    )
    
    # Auto-generate new session ID when checkbox is checked
    def update_session_id(generate_new):
        if generate_new:
            return str(uuid.uuid4())
        return gr.skip()
    
    generate_session_btn.change(
        fn=update_session_id,
        inputs=[generate_session_btn],
        outputs=[session_id_input]
    )


if __name__ == "__main__":
    # print("=" * 60)
    # print("🎤 Analyze Voice Logic Test Page")
    # print("=" * 60)
    # print("\n✅ 사전 준비:")
    # print("   1. PostgreSQL 실행: docker-compose up -d postgres")
    # print("   2. 환경변수 설정: GOOGLE_API_KEY, DATABASE_URL")
    # print("   3. VoiceKit MCP 서버 (외부): 자동 연결")
    # print("\n💡 이 페이지는 독립적으로 실행됩니다 (Backend 서버 불필요)")
    # print("\n" + "=" * 60 + "\n")
    
    demo.launch(
        server_name="127.0.0.1",
        server_port=7861,  # Different port from main gradio_ui.py
        share=False
    )