stt_google.py

"""
Google Cloud Speech-to-Text Implementation
"""
import os
import asyncio
from typing import AsyncIterator, Optional, List, Any
from datetime import datetime
import sys
import queue
import threading
import time
import traceback
from logger import log_info, log_error, log_debug, log_warning

# Import Google Cloud Speech only if available
try:
    from google.cloud import speech
    from google.api_core import exceptions
    GOOGLE_SPEECH_AVAILABLE = True
except ImportError:
    GOOGLE_SPEECH_AVAILABLE = False
    log_info("⚠️ Google Cloud Speech library not installed")

from stt_interface import STTInterface, STTConfig, TranscriptionResult

class GoogleCloudSTT(STTInterface):
    """Google Cloud Speech-to-Text implementation"""
    
    def __init__(self, credentials_path: str):
        if not GOOGLE_SPEECH_AVAILABLE:
            raise ImportError("google-cloud-speech library not installed. Run: pip install google-cloud-speech")
            
        if credentials_path and os.path.exists(credentials_path):
            os.environ['GOOGLE_APPLICATION_CREDENTIALS'] = credentials_path
            log_info(f"✅ Google credentials set from: {credentials_path}")
            
            # Test credential'ları
            try:
                # Client'ı burada oluşturma, her seferinde yeni instance oluştur
                test_client = speech.SpeechClient()
                log_info("🔐 Testing Google credentials...")
                log_info("✅ Google credentials valid")
            except Exception as e:
                log_error(f"❌ Google credentials error", error=str(e))
                raise
        else:
            log_error(f"❌ Google credentials path not found: {credentials_path}")
            raise FileNotFoundError(f"Credentials file not found: {credentials_path}")
            
        # Client'ı burada oluşturma, start_streaming'de oluştur
        self.client = None
        self.streaming_config = None
        self.is_streaming = False
        self.audio_queue = queue.Queue()
        self.responses_queue = queue.Queue()
        self.stream_thread = None
        self.stop_event = threading.Event()
        self.credentials_path = credentials_path  # Sakla
        
    async def start_streaming(self, config: dict) -> None:
        """Initialize streaming session"""
        try:
            log_info(f"🎤 Starting Google STT streaming with config: {config}")
            
            # Her start_streaming'de yeni client oluştur
            self.client = speech.SpeechClient()
            log_info("✅ Created new Google Speech client")
            
            # Convert dict to STTConfig if needed
            if isinstance(config, dict):
                stt_config = STTConfig(
                    language=config.get("language", "tr-TR"),
                    sample_rate=config.get("sample_rate", 16000),
                    encoding=config.get("encoding", "WEBM_OPUS"),
                    enable_punctuation=config.get("enable_punctuation", True),
                    interim_results=config.get("interim_results", True),
                    single_utterance=config.get("single_utterance", False)
                )
            else:
                stt_config = config
            
            recognition_config = speech.RecognitionConfig(
                encoding=self._get_encoding(stt_config.encoding),
                sample_rate_hertz=stt_config.sample_rate,
                language_code=stt_config.language,
                enable_automatic_punctuation=stt_config.enable_punctuation,
                model="latest_long",
                use_enhanced=True
            )
            
            self.streaming_config = speech.StreamingRecognitionConfig(
                config=recognition_config,
                interim_results=stt_config.interim_results,
                single_utterance=stt_config.single_utterance
            )
            
            # Queue'ları temizle
            while not self.audio_queue.empty():
                try:
                    self.audio_queue.get_nowait()
                except:
                    pass
                    
            while not self.responses_queue.empty():
                try:
                    self.responses_queue.get_nowait()
                except:
                    pass
            
            self.is_streaming = True
            self.stop_event.clear()
            
            # Start streaming thread
            self.stream_thread = threading.Thread(target=self._run_stream)
            self.stream_thread.start()
            
            log_info("✅ Google STT streaming started successfully")
            
        except Exception as e:
            log_error(f"❌ Failed to start Google STT streaming", error=str(e))
            self.is_streaming = False
            raise

    def _put_result(self, result: TranscriptionResult):
        """Helper to put result in queue"""
        try:
            self.responses_queue.put(result)
            # Debug log'u kaldırdık
        except Exception as e:
            log_error(f"❌ Error queuing result: {e}")
    
    def _run_stream(self):
        """Run the streaming recognition in a separate thread"""
        try:
            log_info("🎤 Google STT stream thread started")
                
            def request_generator():
                """Generate streaming requests"""
                chunk_count = 0
                total_bytes = 0  # Toplam byte sayısı
                
                while not self.stop_event.is_set():
                    try:
                        chunk = self.audio_queue.get(timeout=0.1)
                        if chunk is None:
                            log_info("📛 Poison pill received, stopping request generator")
                            break
                        
                        chunk_count += 1
                        total_bytes += len(chunk)
                        
                        # İlk chunk'ta audio format kontrolü
                        if chunk_count == 1:
                            log_info(f"📤 First chunk - size: {len(chunk)} bytes")
                            # Audio header kontrolü (WEBM magic bytes)
                            if chunk[:4] == b'\x1a\x45\xdf\xa3':
                                log_info("✅ Valid WEBM header detected")
                            else:
                                log_warning(f"⚠️ Unknown audio format, first 4 bytes: {chunk[:4].hex()}")
                        
                        # Her 100 chunk'ta durum raporu
                        if chunk_count % 100 == 0:
                            avg_chunk_size = total_bytes / chunk_count
                            log_info(f"📤 Progress: {chunk_count} chunks, {total_bytes/1024:.1f}KB total, avg {avg_chunk_size:.0f} bytes/chunk")
                        
                        yield speech.StreamingRecognizeRequest(audio_content=chunk)
                        
                    except queue.Empty:
                        continue
                    except Exception as e:
                        log_error(f"❌ Error in request generator: {e}")
                        break
                
            # Create streaming client
            requests = request_generator()
            
            log_info("🎤 Creating Google STT streaming client...")
            
            try:
                responses = self.client.streaming_recognize(
                    self.streaming_config, 
                    requests, 
                    timeout=300 # 5 dakika timeout
                ) 
                
                log_info("✅ Google STT streaming client created")

                # Response timeout kontrolü
                last_response_time = time.time()
                RESPONSE_TIMEOUT = 30  # 30 saniye içinde response gelmezse
                
                # Process responses
                response_count = 0
                empty_response_count = 0
                
                for response in responses:
                    last_response_time = time.time()
                    response_count += 1

                    # Response type'ı logla
                    if response_count == 1:
                        log_info(f"📨 First response received from Google STT")
                    
                    if self.stop_event.is_set():
                        log_info("🛑 Stop event detected, breaking response loop")
                        break
                    
                    # Response içeriğini kontrol et
                    if not response.results:
                        empty_response_count += 1
                        if empty_response_count == 1:
                            log_debug("📭 Received empty response (no results)")
                        continue

                    for i, result in enumerate(response.results):
                        log_debug(f"📋 Result {i}: is_final={result.is_final}, alternatives={len(result.alternatives)}")
                        
                        if not result.alternatives:
                            log_debug(f"📋 Result {i} has no alternatives")
                            continue
                            
                        # İlk alternatifi al
                        alternative = result.alternatives[0]
                        
                        # Sadece anlamlı text'leri işle
                        if alternative.transcript.strip():
                            # Create transcription result
                            transcription = TranscriptionResult(
                                text=alternative.transcript,
                                is_final=result.is_final,
                                confidence=alternative.confidence if hasattr(alternative, 'confidence') and alternative.confidence else 0.0,
                                timestamp=datetime.now().timestamp()
                            )
                            
                            # Put result in queue
                            self._put_result(transcription)
                            
                            # SADECE final result'ları logla
                            if result.is_final:
                                log_info(f"🎯 GOOGLE STT FINAL: '{alternative.transcript}'")
                        else:
                            log_debug(f"📋 Result {i} has empty transcript")
                            continue    

                if time.time() - last_response_time > RESPONSE_TIMEOUT:
                    log_error(f"❌ No response from Google STT for {RESPONSE_TIMEOUT} seconds")

                log_info(f"📊 Google STT stream ended. Total responses: {response_count}, Empty: {empty_response_count}")
                        
            except Exception as e:
                error_msg = str(e)
                
                # Detaylı hata mesajları
                if "Exceeded maximum allowed stream duration" in error_msg:
                    log_warning("⚠️ Stream duration limit exceeded (5 minutes). This is expected for long sessions.")
                elif "Bad language code" in error_msg:
                    log_error(f"❌ Invalid language code in STT config. Check locale settings.")
                elif "invalid_argument" in error_msg:
                    log_error(f"❌ Invalid STT configuration. Check encoding and sample rate.")
                elif "Deadline Exceeded" in error_msg:
                    log_error(f"❌ Google STT response timeout - possibly network issue or slow connection")
                elif "503" in error_msg or "Service Unavailable" in error_msg:
                    log_error(f"❌ Google STT service temporarily unavailable. Will retry...")
                else:
                    log_error(f"❌ Google STT stream error: {error_msg}")
                    
        except Exception as e:
            log_error(f"❌ Fatal error in STT stream thread", error=str(e), traceback=traceback.format_exc())
        finally:
            log_info("🎤 Google STT stream thread ended")
            # Thread bittiğinde streaming flag'ini kapat
            self.is_streaming = False
            
    async def stream_audio(self, audio_chunk: bytes) -> AsyncIterator[TranscriptionResult]:
        """Stream audio chunk and get transcription results"""
        if not self.is_streaming:
            # Daha detaylı hata mesajı
            log_error(f"❌ STT not streaming - is_streaming: {self.is_streaming}, thread alive: {self.stream_thread and self.stream_thread.is_alive() if hasattr(self, 'stream_thread') else 'No thread'}")
            raise RuntimeError("Streaming not started. Call start_streaming() first.")
        
        try:
            # Put audio in queue for streaming thread
            self.audio_queue.put(audio_chunk)
            
            # Check for any results in queue
            while True:
                try:
                    # Non-blocking get from normal queue
                    result = self.responses_queue.get_nowait()
                    yield result
                except queue.Empty:
                    # No more results in queue
                    break
                    
        except Exception as e:
            log_error(f"❌ Google STT streaming error", error=str(e))
            # Stream'i tekrar başlatmayı tetikle
            self.is_streaming = False
            raise
    
    async def stop_streaming(self) -> Optional[TranscriptionResult]:
        """Stop streaming and get final result"""
        if not self.is_streaming:
            return None
            
        try:
            log_info("🛑 Stopping Google STT streaming...")
            
            self.is_streaming = False
            self.stop_event.set()
            
            # Send poison pill to queue
            self.audio_queue.put(None)
            
            # Wait for thread to finish with longer timeout
            if self.stream_thread and self.stream_thread.is_alive():
                self.stream_thread.join(timeout=10.0)
                if self.stream_thread.is_alive():
                    log_warning("⚠️ STT thread did not stop gracefully")
            
            # Clear queues
            while not self.audio_queue.empty():
                try:
                    self.audio_queue.get_nowait()
                except:
                    pass
                    
            final_result = None
            while not self.responses_queue.empty():
                try:
                    result = self.responses_queue.get_nowait()
                    if result.is_final:
                        final_result = result
                except:
                    pass
            
            # Reset thread reference
            self.stream_thread = None
            
            # Client'ı kapat
            if self.client:
                try:
                    self.client.transport.close()
                except:
                    pass
                self.client = None
            
            log_info("✅ Google STT streaming stopped")
            return final_result
            
        except Exception as e:
            log_error(f"❌ Failed to stop Google STT streaming", error=str(e))
            self.stream_thread = None
            self.client = None
            return None
    
    def supports_realtime(self) -> bool:
        """Google Cloud STT supports real-time streaming"""
        return True
    
    def get_supported_languages(self) -> List[str]:
        """Get list of supported language codes"""
        return [
            "tr-TR",  # Turkish
            "en-US",  # English (US)
            "en-GB",  # English (UK)
            "de-DE",  # German
            "fr-FR",  # French
            "es-ES",  # Spanish
            "it-IT",  # Italian
            "pt-BR",  # Portuguese (Brazil)
            "ru-RU",  # Russian
            "ja-JP",  # Japanese
            "ko-KR",  # Korean
            "zh-CN",  # Chinese (Simplified)
            "ar-SA",  # Arabic
        ]
    
    def get_provider_name(self) -> str:
        """Get provider name"""
        return "google"
    
    def _get_encoding(self, encoding_str: str):
        """Convert encoding string to Google Speech enum"""
        if not GOOGLE_SPEECH_AVAILABLE:
            return None
            
        encoding_map = {
            "WEBM_OPUS": speech.RecognitionConfig.AudioEncoding.WEBM_OPUS,
            "LINEAR16": speech.RecognitionConfig.AudioEncoding.LINEAR16,
            "FLAC": speech.RecognitionConfig.AudioEncoding.FLAC,
            "MP3": speech.RecognitionConfig.AudioEncoding.MP3,
            "OGG_OPUS": speech.RecognitionConfig.AudioEncoding.OGG_OPUS,
        }
        return encoding_map.get(encoding_str, speech.RecognitionConfig.AudioEncoding.WEBM_OPUS)