app.py

import gradio as gr
import openai
from openai import OpenAI
import numpy as np
import threading
import queue
import time
import json
import websocket
import base64
import pyaudio
import wave
import io
from typing import Generator, Tuple
import asyncio
import edge_tts

# OpenAI API 키 설정
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
client = OpenAI(api_key=OPENAI_API_KEY)

class RealtimeTranslator:
    def __init__(self):
        self.is_recording = False
        self.audio_queue = queue.Queue()
        self.text_queue = queue.Queue()
        self.translation_queue = queue.Queue()
        self.current_text = ""
        self.detected_language = None
        
    def detect_language(self, text: str) -> str:
        """텍스트의 언어를 감지합니다."""
        korean_chars = sum(1 for char in text if ord('가') <= ord(char) <= ord('힣'))
        total_chars = len(text.replace(" ", ""))
        
        if total_chars > 0:
            korean_ratio = korean_chars / total_chars
            if korean_ratio > 0.3:
                return "ko"
        return "en"
    
    def process_audio_chunk(self, audio_chunk):
        """오디오 청크를 처리하여 텍스트로 변환"""
        try:
            # 오디오 청크를 임시 파일로 저장
            import tempfile
            with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp_file:
                # WAV 파일로 저장
                import wave
                with wave.open(tmp_file.name, 'wb') as wav_file:
                    wav_file.setnchannels(1)
                    wav_file.setsampwidth(2)
                    wav_file.setframerate(16000)
                    wav_file.writeframes(audio_chunk)
                
                # Whisper API 호출
                with open(tmp_file.name, "rb") as audio_file:
                    transcript = client.audio.transcriptions.create(
                        model="whisper-1",
                        file=audio_file,
                        language=None,
                        prompt="실시간 대화를 번역합니다."
                    )
                
                return transcript.text
                
        except Exception as e:
            print(f"음성 인식 오류: {e}")
            return ""
    
    def translate_stream(self, text: str, source_lang: str) -> str:
        """텍스트를 실시간으로 번역"""
        try:
            if not text or text.strip() == "":
                return ""
            
            # 번역 프롬프트
            if source_lang == "ko":
                messages = [
                    {"role": "system", "content": "실시간 통역사입니다. 한국어를 영어로 즉시 번역합니다."},
                    {"role": "user", "content": text}
                ]
            else:
                messages = [
                    {"role": "system", "content": "실시간 통역사입니다. 영어를 한국어로 즉시 번역합니다."},
                    {"role": "user", "content": text}
                ]
            
            # 스트리밍 응답
            stream = client.chat.completions.create(
                model="gpt-4o-mini",
                messages=messages,
                stream=True,
                temperature=0.3,
                max_tokens=150
            )
            
            translated = ""
            for chunk in stream:
                if chunk.choices[0].delta.content:
                    translated += chunk.choices[0].delta.content
            
            return translated
            
        except Exception as e:
            print(f"번역 오류: {e}")
            return ""

translator = RealtimeTranslator()

def process_stream(audio_stream):
    """오디오 스트림을 실시간으로 처리"""
    if audio_stream is None:
        yield "🔴 마이크를 켜고 말씀해주세요", "", ""
        return
    
    sample_rate, audio_data = audio_stream
    
    # 오디오 데이터가 너무 짧으면 무시
    if len(audio_data) < sample_rate * 0.5:  # 0.5초 미만
        yield "🎤 듣고 있습니다...", "", ""
        return
    
    # 오디오 청크 처리
    audio_bytes = audio_data.tobytes()
    
    # 음성을 텍스트로 변환
    text = translator.process_audio_chunk(audio_bytes)
    
    if text:
        # 언어 감지
        detected_lang = translator.detect_language(text)
        
        # 실시간 번역
        translated = translator.translate_stream(text, detected_lang)
        
        # 결과 반환
        if detected_lang == "ko":
            yield f"✅ 한국어 감지", text, translated
        else:
            yield f"✅ English detected", translated, text

def create_realtime_interface():
    with gr.Blocks(title="실시간 음성 번역기", theme=gr.themes.Soft()) as demo:
        gr.Markdown(
            """
            # 🎤 실시간 음성 번역기 (Real-time Voice Translator)
            
            ### 말하는 동안 실시간으로 번역됩니다!
            
            🔴 **시작** 버튼을 누르고 말하면, 실시간으로 번역이 표시됩니다.
            
            ---
            """
        )
        
        with gr.Row():
            with gr.Column(scale=2):
                audio_input = gr.Audio(
                    source="microphone",
                    type="numpy",
                    streaming=True,  # 스트리밍 모드 활성화
                    label="🎤 실시간 마이크 입력",
                    elem_id="audio-stream"
                )
            
            with gr.Column(scale=1):
                status_text = gr.Textbox(
                    label="📊 상태",
                    value="🔴 마이크를 켜고 말씀해주세요",
                    interactive=False
                )
        
        with gr.Row():
            with gr.Column():
                korean_output = gr.Textbox(
                    label="🇰🇷 한국어",
                    placeholder="한국어가 실시간으로 표시됩니다",
                    lines=8,
                    interactive=False,
                    elem_id="korean-text"
                )
            
            with gr.Column():
                english_output = gr.Textbox(
                    label="🇺🇸 English",
                    placeholder="English translation appears here in real-time",
                    lines=8,
                    interactive=False,
                    elem_id="english-text"
                )
        
        # 스트리밍 이벤트 설정
        audio_input.stream(
            fn=process_stream,
            inputs=[audio_input],
            outputs=[status_text, korean_output, english_output],
            show_progress=False
        )
        
        gr.Markdown(
            """
            ---
            
            ### 💡 사용 팁:
            - 명확하게 말할수록 인식률이 높아집니다
            - 문장이 끝날 때까지 잠시 멈추면 더 정확한 번역이 됩니다
            - 한국어와 영어를 자동으로 감지합니다
            
            ### ⚙️ 기술 사양:
            - **음성 인식**: OpenAI Whisper (실시간 스트리밍)
            - **번역**: GPT-4 (스트리밍 모드)
            - **지연 시간**: ~1-2초
            """
        )
        
        # CSS 스타일 추가
        demo.css = """
        #audio-stream {
            height: 150px !important;
        }
        #korean-text, #english-text {
            font-size: 18px !important;
            line-height: 1.5 !important;
        }
        .gradio-container {
            max-width: 1200px !important;
        }
        """
    
    return demo

# 대안: WebSocket 기반 실시간 번역 (더 낮은 지연시간)
class WebSocketTranslator:
    def __init__(self):
        self.ws_url = "wss://api.openai.com/v1/realtime"  # 예시 URL
        self.ws = None
        self.is_connected = False
        
    def connect(self):
        """WebSocket 연결"""
        headers = {
            "Authorization": f"Bearer {OPENAI_API_KEY}",
            "OpenAI-Beta": "realtime=v1"
        }
        
        try:
            self.ws = websocket.WebSocketApp(
                self.ws_url,
                header=headers,
                on_open=self.on_open,
                on_message=self.on_message,
                on_error=self.on_error,
                on_close=self.on_close
            )
            
            # 별도 스레드에서 실행
            wst = threading.Thread(target=self.ws.run_forever)
            wst.daemon = True
            wst.start()
            
        except Exception as e:
            print(f"WebSocket 연결 오류: {e}")
    
    def on_open(self, ws):
        self.is_connected = True
        print("WebSocket 연결됨")
        
    def on_message(self, ws, message):
        """메시지 수신 처리"""
        try:
            data = json.loads(message)
            if data.get("type") == "transcription":
                # 실시간 텍스트 처리
                text = data.get("text", "")
                self.process_realtime_text(text)
        except Exception as e:
            print(f"메시지 처리 오류: {e}")
    
    def on_error(self, ws, error):
        print(f"WebSocket 오류: {error}")
        
    def on_close(self, ws, close_status_code, close_msg):
        self.is_connected = False
        print("WebSocket 연결 종료")
    
    def send_audio(self, audio_data):
        """오디오 데이터 전송"""
        if self.is_connected and self.ws:
            # 오디오를 base64로 인코딩
            audio_base64 = base64.b64encode(audio_data).decode('utf-8')
            
            message = {
                "type": "audio",
                "audio": audio_base64
            }
            
            self.ws.send(json.dumps(message))
    
    def process_realtime_text(self, text):
        """실시간 텍스트 처리 및 번역"""
        # 언어 감지 및 번역 로직
        pass

# 메인 실행
if __name__ == "__main__":
    import os
    
    # API 키 확인
    if OPENAI_API_KEY == "your-api-key-here":
        api_key = os.getenv("OPENAI_API_KEY")
        if api_key:
            OPENAI_API_KEY = api_key
            client = OpenAI(api_key=OPENAI_API_KEY)
        else:
            print("⚠️  경고: OpenAI API 키를 설정해주세요!")
            print("환경 변수 OPENAI_API_KEY를 설정하거나 코드에 직접 입력하세요.")
    
    # Gradio 앱 실행
    demo = create_realtime_interface()
    demo.queue()  # 큐 활성화 (스트리밍에 필요)
    demo.launch(
        share=False,
        server_name="0.0.0.0",
        server_port=7860,
        debug=True
    )