stt/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 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 - ✅ TEK DEĞİŞİKLİK BURASI
            final_result = None
            if self.responses_queue:
                while not self.responses_queue.empty():
                    try:
                        result = self.responses_queue.get_nowait()  # ✅ await yerine get_nowait()
                        if result.is_final:
                            final_result = result
                    except queue.Empty:  # ✅ queue.Empty exception'ı 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