""" Backend 로직이 통합된 단일 앱 버전 (포트 1개만 사용) 음성 검증 앱 - 메인 애플리케이션 (통합 버전) Author: Kevin's Team Description: 음성 인식 기반 발음 검증 시스템 """ import os # .env 로드 (다른 모듈 import 전에 먼저 실행) from dotenv import load_dotenv load_dotenv(os.path.join(os.path.dirname(__file__), "..", ".env")) import sys import asyncio # 백엔드 모듈 import를 위해 상위 디렉토리 추가 sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import gradio as gr from datetime import datetime from utils.audio_validator import AudioValidator from utils.stt_handler import STTHandler from utils.state_manager import StateManager from config.settings import AppConfig from frontend.app_ui import AppUI from frontend.components.failure_modal import FailureModalComponent from frontend.components.custom_modal import Modal from frontend.components.history_display import HistoryDisplayComponent # 게임 상태 관리 (UUID 등) from utils.game_state import GameStateManager, create_default_game_state # 통합된 백엔드 서비스 (HTTP 호출 대신 직접 import) from services.analysis_service import analyze_voice from services.database import get_dashboard_stats from services.voice_analyzer import initialize_voicekit_mcp, cleanup_voicekit_mcp # Gradio 임시 파일 디렉토리 설정 (리눅스 권한 문제 해결) # 프로젝트 내부에 uploads 디렉토리 사용 UPLOAD_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "uploads") os.makedirs(UPLOAD_DIR, exist_ok=True) os.environ["GRADIO_TEMP_DIR"] = UPLOAD_DIR # 프로젝트 루트 디렉토리 (client/ 상위) PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # docs 디렉토리 경로 (tech-stack.html 등) DOCS_DIR = os.path.join(PROJECT_ROOT, "docs") IMAGES_DIR = os.path.join(PROJECT_ROOT, "images") REFERENCE_AUDIO_DIR = os.path.join(PROJECT_ROOT, "reference_audio") # 디버그: 경로 확인 # print(f"[PATH DEBUG] __file__: {__file__}") # print(f"[PATH DEBUG] PROJECT_ROOT: {PROJECT_ROOT}") # print(f"[PATH DEBUG] DOCS_DIR: {DOCS_DIR}") # print(f"[PATH DEBUG] DOCS_DIR exists: {os.path.exists(DOCS_DIR)}") if os.path.exists(DOCS_DIR): # print(f"[PATH DEBUG] DOCS_DIR contents: {os.listdir(DOCS_DIR)}") pass # 환경변수로 allowed_paths 설정 (Spaces 배포용) os.environ["GRADIO_ALLOWED_PATHS"] = f"{UPLOAD_DIR},{DOCS_DIR},{IMAGES_DIR},{REFERENCE_AUDIO_DIR}" # print(f"[PATH DEBUG] GRADIO_ALLOWED_PATHS: {os.environ['GRADIO_ALLOWED_PATHS']}") class AudioValidationApp: """메인 애플리케이션 클래스""" def __init__(self): """초기화""" self.config = AppConfig() self.stt_handler = STTHandler() self.validator = AudioValidator(self.config) self.state_manager = StateManager() # UI 빌더 self.ui_builder = AppUI(self.validator) self.components = None def _fetch_dashboard_stats_sync(self) -> dict: """ 대시보드 통계 데이터 가져오기 (직접 DB 호출) Returns: dict: 통계 데이터 (전체 데이터 포함) """ try: data = get_dashboard_stats() # print(f"[DASHBOARD] 통계 데이터: {data}") return data # 전체 데이터 반환 (answer_word, reference_audio_path 포함) except Exception as e: print(f"[DASHBOARD ERROR] {e}") return { "today_participants": 0, "today_success_rate": 0, "today_attempts": 0, "total_participants": 0, "total_success_rate": 0, "total_attempts": 0, "answer_word": None, "reference_audio_path": None } async def _fetch_dashboard_stats(self) -> dict: """ 대시보드 통계 데이터 가져오기 (비동기 버전 - 동일 로직) Returns: dict: 통계 데이터 (today_participants, today_success_rate 등) """ # DB 호출은 동기이므로 동일 함수 사용 return self._fetch_dashboard_stats_sync() async def validate_audio_handler(self, audio_path, current_history, browser_storage, game_state): """ 오디오 검증 메인 핸들러 (직접 서비스 호출 - HTTP 없음) Args: audio_path (str): 업로드된 오디오 파일 경로 current_history (list): 현재까지의 실패 기록 browser_storage (dict): localStorage에 저장된 데이터 game_state (dict): 게임 상태 (UUID, 세션 등) Returns: dict: UI 컴포넌트 업데이트 딕셔너리 """ # UUID 가져오기 session_id = GameStateManager.get_session_id(game_state) # 고정 난이도 (난이도 선택 기능 제거) current_difficulty = 1 # 1. 입력 검증 if audio_path is None: return ( gr.HTML(), # 0: history_html current_history, # 1: failure_history browser_storage, # 2: browser_history game_state, # 3: game_state Modal.hide(), # 4: failure_modal gr.Column(visible=True), # 5: main_screen gr.Column(visible=False), # 6: success_screen "" # 7: success_content ) # 시도 횟수 = 현재 실패 기록 개수 attempt_count = len(current_history) expected_text = self.validator.get_expected_text(current_difficulty) today = datetime.now().strftime("%Y-%m-%d") # print("=" * 60) print("[AUDIO VALIDATE] 음성 검증 요청") # print(f" - Session ID (UUID): {session_id}") # print(f" - Audio Path: {audio_path}") # print(f" - 시도 횟수: {attempt_count + 1}번째") # print(f" - 기대 텍스트: {expected_text}") # print("=" * 60) # ========== 직접 서비스 호출 (HTTP 없음) ========== try: # MCP 연결 확인 (첫 요청 시 초기화) await ensure_mcp_initialized_async() # 오디오 파일 읽기 with open(audio_path, 'rb') as f: audio_bytes = f.read() # 직접 분석 서비스 호출 (HTTP 대신 함수 호출) api_result = await analyze_voice(audio_bytes, today, session_id) # print(f"[ANALYSIS RESULT] {api_result}") # API 오류 처리 if api_result.get("status") == "error": # print(f"[API ERROR] {api_result.get('message')}") # 오류 시 기본 실패 처리 return ( HistoryDisplayComponent.render_failure_history(current_history), current_history, browser_storage, game_state, Modal.show( FailureModalComponent.create_modal_content( "분석 오류", 0, api_result.get('message', '오류가 발생했습니다'), audio_path=audio_path, metrics={"pronunciation": 0, "tone": 0, "pitch": 0, "rhythm": 0, "energy": 0} ) ), gr.Column(visible=True), gr.Column(visible=False), "" ) # API 응답에서 데이터 추출 is_correct = api_result.get("is_correct", False) score = api_result.get("overall", 0) advice = api_result.get("advice", "") category = api_result.get("category", "") answer_word = api_result.get("answer_word", "") reference_audio_path_rel = api_result.get("reference_audio_path", "") # 상대 경로 user_text = api_result.get("user_text", "") # STT 결과 # reference_audio_path를 절대 경로로 변환 (Gradio file serving용) if reference_audio_path_rel: reference_audio_path = os.path.join(PROJECT_ROOT, reference_audio_path_rel.lstrip("/")) # .wav가 없으면 .mp3 등 대체 확장자 찾기 if not os.path.exists(reference_audio_path): base_path = os.path.splitext(reference_audio_path)[0] for ext in ['.mp3', '.wav', '.m4a', '.ogg', '.flac']: alt_path = base_path + ext if os.path.exists(alt_path): reference_audio_path = alt_path break else: reference_audio_path = "" # 메트릭 데이터 매핑 (API: pitch, rhythm, energy, pronunciation, transcript) metrics = { "pronunciation": api_result.get("pronunciation", 0), "tone": api_result.get("transcript", 0), # transcript를 tone으로 매핑 "pitch": api_result.get("pitch", 0), "rhythm": api_result.get("rhythm", 0), "energy": api_result.get("energy", 0), "overall": score # overall 점수 추가 } # Use actual STT transcription from MCP, fallback to score if not available recognized_text = user_text or f"Score: {score:.1f}" except Exception as e: print(f"[API EXCEPTION] {e}") import traceback traceback.print_exc() # 예외 시 오류 모달 표시 return ( HistoryDisplayComponent.render_failure_history(current_history), current_history, browser_storage, game_state, Modal.show( FailureModalComponent.create_modal_content( "연결 오류", 0, f"서버 연결에 실패했습니다: {str(e)}", audio_path=audio_path, metrics={"pronunciation": 0, "tone": 0, "pitch": 0, "rhythm": 0, "energy": 0} ) ), gr.Column(visible=True), gr.Column(visible=False), "" ) # ========== 결과 처리 ========== if not is_correct: # ❌ 실패 처리 # 실패 기록 생성 (메트릭 + advice + user_text 포함) new_entry = self.state_manager.create_failure_entry( audio_path, recognized_text, metrics, advice, user_text # STT 결과 전달 ) updated_history = current_history + [new_entry] # 실패 기록 HTML 렌더링 history_html = HistoryDisplayComponent.render_failure_history(updated_history) print(f"[CONSOLE] 실패 처리 - Modal 표시") # localStorage 업데이트 (날짜와 실패 기록 저장) updated_storage = { "date": today, "failures": updated_history, "successes": browser_storage.get("successes", []) } # Modal 내용 생성 (오디오 + 그래프 + API advice 포함) modal_content = FailureModalComponent.create_modal_content( recognized_text, score, advice, # API에서 받은 advice 사용 audio_path=audio_path, metrics=metrics, user_text=user_text # STT 결과 전달 ) # game_state에 추측 기록 추가 updated_game_state = GameStateManager.add_guess( game_state, recognized_text, audio_path, { "overall_score": score, # Standardize field name "advice": advice, # Include AI-generated hints "category": category, # Puzzle category for chatbot "answerWord": answer_word, # Answer word for chatbot context "referenceAudioPath": reference_audio_path, # For TTS voice cloning "userText": user_text, # STT result for chatbot context **metrics } ) # print(f"[AUDIO VALIDATE] 실패 처리 - Modal 표시, 총 추측: {len(updated_game_state.get('guesses', []))}개") return ( history_html, # 0: history_html updated_history, # 1: failure_history updated_storage, # 2: browser_history updated_game_state, # 3: game_state Modal.show(modal_content), # 4: failure_modal gr.Column(visible=True), # 5: main_screen gr.Column(visible=False), # 6: success_screen "" # 7: success_content ) else: # ✅ 성공 처리 # 실패 기록 HTML (현재까지 쌓인 기록 유지) history_html = HistoryDisplayComponent.render_failure_history(current_history) # localStorage 업데이트 (성공 기록 추가) success_entry = { "timestamp": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), "difficulty": current_difficulty, "attempts": len(current_history) + 1 } successes = browser_storage.get("successes", []) successes.append(success_entry) updated_storage = { "date": today, "failures": current_history, "successes": successes } # game_state에 성공한 오디오 기록 추가 (User Audio 표시용) updated_game_state = GameStateManager.add_guess( game_state, recognized_text, audio_path, { "score": score, "userText": user_text, "answerWord": answer_word, "referenceAudioPath": reference_audio_path, "category": category, **metrics } ) # game_state 성공 처리 updated_game_state = GameStateManager.set_win_state(updated_game_state, win=True) print(f"[AUDIO VALIDATE] 성공 처리!") # print(f" - Session ID: {session_id}") # print(f" - 총 시도 횟수: {len(current_history) + 1}") # print(f" - 승리 연속: {updated_game_state.get('stats', {}).get('winStreak', 0)}") # 통계는 JavaScript에서 화면 표시 시 API 호출하여 업데이트 # print(f"[SUCCESS] 성공 화면으로 전환 - JS가 통계 API 호출 예정") # outputs 순서: history_html, failure_history, browser_history, game_state, # failure_modal, main_screen, success_screen, success_content return ( history_html, # 0: history_html current_history, # 1: failure_history updated_storage, # 2: browser_history updated_game_state, # 3: game_state Modal.hide(), # 4: failure_modal gr.Column(visible=False), # 5: main_screen gr.Column(visible=True), # 6: success_screen gr.skip() # 7: success_content (JS가 업데이트) ) async def give_up_handler(self, game_state): """ 포기하기 핸들러 (async) """ print("[DEBUG] Give Up Handler Triggered!") # 1. 게임 상태 업데이트 (포기) updated_game_state = GameStateManager.set_win_state(game_state, win=False) # 2. 정답 정보 가져오기 # 고정 난이도 1 expected_text = self.validator.get_expected_text(1) # 3. 마지막 사용자 오디오 가져오기 guesses = updated_game_state.get('guesses', []) last_audio = "" if guesses: last_audio = guesses[-1].get('audioFile', "") # 통계는 JavaScript에서 화면 표시 시 API 호출하여 업데이트 # print(f"[GIVEUP] 포기 화면으로 전환 - JS가 통계 API 호출 예정") # outputs 순서: main_screen, giveup_screen, giveup_content, game_state return ( gr.Column(visible=False), # 0: main_screen gr.Column(visible=True), # 1: giveup_screen gr.skip(), # 2: giveup_content (JS가 업데이트) updated_game_state # 3: game_state ) def load_and_filter_history(self, browser_storage, game_state): """ 페이지 로드 시 날짜 체크 및 히스토리 필터링 + UUID 체크/생성 Args: browser_storage (dict): localStorage에 저장된 데이터 game_state (dict): 게임 상태 (UUID, 세션 등) Returns: tuple: (history_html, filtered_storage, failure_list, updated_game_state) """ today = datetime.now().strftime("%Y-%m-%d") stored_date = browser_storage.get("date", "") # ============================================================ # UUID 체크/생성 (GameStateManager 사용) # ============================================================ updated_game_state = GameStateManager.get_or_create_session(game_state) session_id = GameStateManager.get_session_id(updated_game_state) # print("=" * 60) # print("[PAGE LOAD] 페이지 로드") # print(f" - 오늘 날짜: {today}") # print(f" - 저장된 날짜: {stored_date}") # print(f" - Session ID (UUID): {session_id}") # print(f" - 게임 상태: {updated_game_state.get('winState', -1)} (-1:진행중, 0:포기, 1:성공)") # print("=" * 60) if stored_date != today: # 날짜가 다르면 기록 초기화 print(f"[PAGE LOAD] 날짜 변경 감지 - 기록 초기화") new_storage = { "date": today, "failures": [], "successes": [] } return HistoryDisplayComponent.render_empty_history(), new_storage, [], updated_game_state else: # 같은 날이면 기존 기록 로드 failures = browser_storage.get("failures", []) print(f"[PAGE LOAD] 기존 기록 로드 - 실패 {len(failures)}건") if failures: history_html = HistoryDisplayComponent.render_failure_history(failures) else: history_html = HistoryDisplayComponent.render_empty_history() return history_html, browser_storage, failures, updated_game_state def build_ui(self): """ Gradio UI 구성 Returns: gr.Blocks: Gradio 앱 인스턴스 """ # 앱 시작 시점에 통계 데이터 가져오기 stats = self._fetch_dashboard_stats_sync() # print(f"[BUILD UI] 초기 통계 데이터: {stats}") # 이벤트 핸들러 설정 handlers = { 'on_load': self.load_and_filter_history, 'validate_audio': self.validate_audio_handler, # async 함수 직접 전달 'give_up': self.give_up_handler } # UI 빌드 (이벤트 바인딩 포함, 통계 데이터 전달) self.components = self.ui_builder.build(handlers, stats=stats) return self.components['demo'] def launch(self, **kwargs): """ 앱 실행 Args: **kwargs: Gradio launch 파라미터 """ from frontend.styles.custom_css import get_all_css, get_head_scripts demo = self.build_ui() # Gradio 6: css와 head를 launch()에서 전달 demo.launch(css=get_all_css(), head=get_head_scripts(), **kwargs) # VoiceKit MCP 초기화 상태 _mcp_initialized = False async def ensure_mcp_initialized_async(): """VoiceKit MCP 연결이 초기화되었는지 확인하고, 아니면 초기화 (비동기)""" global _mcp_initialized if not _mcp_initialized: _mcp_initialized = True await initialize_voicekit_mcp() print("[MCP] VoiceKit MCP initialized") # Gradio Spaces용 - 모듈 레벨에서 app과 demo 생성 from frontend.styles.custom_css import get_all_css, get_head_scripts from fastapi.responses import JSONResponse # 중요: Blocks 생성 전에 static paths 설정 (docs 파일 서빙용) gr.set_static_paths(paths=[DOCS_DIR]) audio_app = AudioValidationApp() demo = audio_app.build_ui() # API 엔드포인트를 Gradio Blocks 내부에 추가 with demo: # 숨겨진 버튼으로 API 호출 트리거 dashboard_api_btn = gr.Button(visible=False, elem_id="dashboard-api-btn") dashboard_api_output = gr.JSON(visible=False, elem_id="dashboard-api-output") def get_dashboard_api(): """Dashboard API 핸들러""" return get_dashboard_stats() dashboard_api_btn.click( fn=get_dashboard_api, inputs=[], outputs=[dashboard_api_output], api_name="dashboard" # /api/dashboard 엔드포인트 생성 ) # Hugging Face Spaces 및 로컬 실행 모두 지원 # Spaces에서는 demo.launch()가 자동 호출되지만, allowed_paths 설정을 위해 직접 호출 server_host = os.getenv("SERVER_HOST", "0.0.0.0") frontend_port = int(os.getenv("FRONTEND_PORT")) # Gradio 실행 (CSS 적용) demo.launch( css=get_all_css(), head=get_head_scripts(), share=False, server_name=server_host, server_port=frontend_port, show_error=True, allowed_paths=[UPLOAD_DIR, DOCS_DIR, IMAGES_DIR, REFERENCE_AUDIO_DIR], )