""" 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 utils.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") # Test client'ı kapat if hasattr(test_client, 'transport') and hasattr(test_client.transport, 'close'): test_client.transport.close() 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 = None # Queue'ları None olarak başlat self.responses_queue = None self.stream_thread = None self.stop_event = threading.Event() self.credentials_path = credentials_path # Session tracking self.session_id = 0 self.total_audio_bytes = 0 self.total_chunks = 0 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) async def stream_audio(self, audio_chunk: bytes) -> AsyncIterator[TranscriptionResult]: """Stream audio chunk and get transcription results""" if not self.is_streaming: log_error(f"❌ STT not streaming - is_streaming: {self.is_streaming}") 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 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)) self.is_streaming = False raise async def stop_streaming(self) -> Optional[TranscriptionResult]: """Stop streaming and clean up all resources""" if not self.is_streaming and not self.stream_thread: log_debug("Already stopped, nothing to do") return None try: log_info(f"🛑 Stopping Google STT streaming session #{self.session_id}") # Flag'i hemen kapat self.is_streaming = False self.stop_event.set() # Send poison pill to stop request generator if self.audio_queue: try: self.audio_queue.put(None) except: pass # Thread'i durdur if self.stream_thread and self.stream_thread.is_alive(): log_info("⏳ Waiting for stream thread to finish...") self.stream_thread.join(timeout=5.0) if self.stream_thread.is_alive(): log_warning("⚠️ STT thread did not stop gracefully after 5s") else: log_info("✅ Stream thread finished") # Final result'ı al - ✅ BURADA DÜZELTME final_result = None if self.responses_queue: while not self.responses_queue.empty(): try: result = self.responses_queue.get_nowait() # ✅ await değil, get_nowait() if result.is_final: final_result = result except queue.Empty: # ✅ queue.Empty kullan break # Client'ı kapat if self.client: try: if hasattr(self.client, 'transport') and hasattr(self.client.transport, 'close'): self.client.transport.close() log_debug("✅ Client transport closed") if hasattr(self.client, '_transport') and hasattr(self.client._transport, '_grpc_channel'): self.client._transport._grpc_channel.close() log_debug("✅ gRPC channel closed") except Exception as e: log_warning(f"⚠️ Error closing Google client: {e}") finally: self.client = None # Queue'ları None yap self.audio_queue = None self.responses_queue = None # Diğer değişkenleri resetle self.stream_thread = None self.streaming_config = None self.stop_event.clear() log_info(f"✅ Google STT streaming session #{self.session_id} stopped and cleaned") return final_result except Exception as e: log_error(f"❌ Error during stop_streaming", error=str(e)) # Force cleanup on error self.is_streaming = False self.stream_thread = None self.client = None self.streaming_config = None self.stop_event.clear() self.audio_queue = None self.responses_queue = 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 _reset_session_data(self): """Reset all session-specific data""" # Queue'ları temizle if self.audio_queue: while not self.audio_queue.empty(): try: self.audio_queue.get_nowait() except: pass if self.responses_queue: while not self.responses_queue.empty(): try: self.responses_queue.get_nowait() except: pass # Counters'ı sıfırla self.total_audio_bytes = 0 self.total_chunks = 0 # Yeni session ID self.session_id += 1 log_info(f"🔄 Google STT session data reset. New session ID: {self.session_id}") def _create_fresh_queues(self): """Create fresh queue instances""" # Eski queue'ları temizle if self.audio_queue: while not self.audio_queue.empty(): try: self.audio_queue.get_nowait() except: pass if self.responses_queue: while not self.responses_queue.empty(): try: self.responses_queue.get_nowait() except: pass # Yeni queue'lar oluştur self.audio_queue = queue.Queue(maxsize=1000) # Max size ekle self.responses_queue = queue.Queue(maxsize=100) log_debug("✅ Created fresh queues") async def start_streaming(self, config: dict) -> None: """Initialize streaming session with clean state""" try: # Önce mevcut stream'i temizle if self.is_streaming or self.stream_thread: log_warning("⚠️ Previous stream still active, stopping it first") await self.stop_streaming() # Temizlik için bekle await asyncio.sleep(0.5) # Session verilerini resetle ve ID'yi artır self._reset_session_data() log_info(f"🎤 Starting Google STT streaming session #{self.session_id} with config: {config}") # Fresh queue'lar oluştur self._create_fresh_queues() # Stop event'i temizle self.stop_event.clear() # Yeni client oluştur (TEK SEFER) 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", False), single_utterance=config.get("single_utterance", True) ) 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, # Bu parametreleri kaldırıyoruz - v1 API'de yok # enable_voice_activity_events=True, # audio_channel_count=1 ) self.streaming_config = speech.StreamingRecognitionConfig( config=recognition_config, interim_results=stt_config.interim_results, single_utterance=stt_config.single_utterance # enable_voice_activity_events kaldırıldı ) self.is_streaming = True # Start streaming thread with unique name self.stream_thread = threading.Thread( target=self._run_stream, name=f"GoogleSTT-Session-{self.session_id}" ) self.stream_thread.daemon = True # Daemon thread olarak işaretle self.stream_thread.start() log_info(f"✅ Google STT streaming session #{self.session_id} started successfully") except Exception as e: log_error(f"❌ Failed to start Google STT streaming", error=str(e)) self.is_streaming = False self.client = None self._create_fresh_queues() # Hata durumunda da queue'ları temizle 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(f"🎤 Google STT stream thread started - Single utterance mode: {self.streaming_config.single_utterance}") def request_generator(): """Generate streaming requests""" chunk_count = 0 total_bytes = 0 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") if len(chunk) >= 4 and chunk[:4] == b'\x1a\x45\xdf\xa3': log_info("✅ Valid WEBM header detected") # Her 50 chunk'ta durum raporu if chunk_count % 50 == 0: log_info(f"📤 Progress: {chunk_count} chunks, {total_bytes/1024:.1f}KB total") yield speech.StreamingRecognizeRequest(audio_content=chunk) except queue.Empty: continue except Exception as e: log_error(f"❌ Error in request generator: {e}") break log_info(f"📊 Request generator finished. Total chunks: {chunk_count}, Total bytes: {total_bytes}") # Create streaming client requests = request_generator() log_info("🎤 Creating Google STT streaming client...") try: # Start streaming responses = self.client.streaming_recognize( self.streaming_config, requests, timeout=300 ) log_info("✅ Google STT streaming client created, waiting for responses...") # Process responses response_count = 0 result_count = 0 last_log_time = time.time() # Response iterator'ı başlat try: for response in responses: response_count += 1 # İlk response'u logla if response_count == 1: log_info(f"📨 First response received from Google STT") # Her 5 saniyede bir durum logu if time.time() - last_log_time > 5: log_info(f"📊 Still listening... Responses: {response_count}, Results: {result_count}") last_log_time = time.time() if self.stop_event.is_set(): log_info("🛑 Stop event detected") break # Process results if not response.results: log_debug(f"📭 Response #{response_count} has no results") continue for result in response.results: result_count += 1 if not result.alternatives: continue alternative = result.alternatives[0] # Log all transcripts, even empty ones log_debug(f"📝 Transcript: '{alternative.transcript}' (is_final: {result.is_final})") if alternative.transcript.strip(): # Create transcription result transcription = TranscriptionResult( text=alternative.transcript, is_final=result.is_final, confidence=getattr(alternative, 'confidence', 0.0), timestamp=datetime.now().timestamp() ) # Put result in queue self._put_result(transcription) if result.is_final: log_info(f"🎯 FINAL TRANSCRIPT: '{alternative.transcript}'") # Single utterance modunda Google STT otomatik kapanır if self.streaming_config.single_utterance: log_info("✅ Single utterance completed - Stream ending") return else: log_debug(f"📝 Interim: '{alternative.transcript}'") except StopIteration: log_info("✅ Google STT stream ended (StopIteration)") except Exception as e: log_error(f"❌ Error processing responses: {e}") log_info(f"📊 Google STT stream ended. Responses: {response_count}, Results: {result_count}") except Exception as e: error_msg = str(e) # Beklenen hatalar if "iterating requests" in error_msg: log_info("✅ Stream ended normally") elif "Exceeded maximum allowed stream duration" in error_msg: log_warning("⚠️ Stream duration limit (5 min)") elif "InvalidArgument" in error_msg: log_error(f"❌ Invalid STT configuration: {error_msg}") else: log_error(f"❌ Google STT error: {error_msg}") except Exception as e: log_error(f"❌ Fatal error in STT stream", error=str(e), traceback=traceback.format_exc()) finally: log_info("🎤 Google STT stream thread ended") self.is_streaming = False