import os import sys import logging import asyncio from pathlib import Path from typing import Optional, Tuple logger = logging.getLogger("AGYVoiceTranscriber") _whisper_model = None def get_whisper_model(): """Lazy loader for singleton WhisperModel instance.""" global _whisper_model if _whisper_model is None: try: from faster_whisper import WhisperModel logger.info("Initializing faster-whisper model (base, int8)...") _whisper_model = WhisperModel("base", device="cpu", compute_type="int8") logger.info("faster-whisper model initialized successfully.") except Exception as e: logger.error(f"Failed to load faster-whisper model: {e}", exc_info=True) _whisper_model = None return _whisper_model def _transcribe_sync(audio_path: str, language: Optional[str] = None) -> Tuple[bool, str, str]: model = get_whisper_model() if model is None: return False, "", "unknown" try: # Default to Persian (fa) unless explicitly specified otherwise lang = language if (language and language not in ("auto", "none")) else "fa" segments, info = model.transcribe( audio_path, beam_size=5, language=lang, vad_filter=True, initial_prompt="گفتگوی محاوره‌ای و رسمی به زبان فارسی", ) texts = [seg.text.strip() for seg in segments if seg.text.strip()] full_text = " ".join(texts).strip() detected_lang = getattr(info, "language", "unknown") return True, full_text, detected_lang except Exception as e: logger.error(f"Error transcribing audio {audio_path}: {e}", exc_info=True) return False, str(e), "unknown" async def transcribe_audio_file(audio_path: str, language: Optional[str] = None) -> Tuple[bool, str, str]: """ Asynchronously transcribes audio file (OGG, MP3, WAV, M4A, etc.) to text. Returns: (success, transcribed_text, detected_language) """ return await asyncio.to_thread(_transcribe_sync, audio_path, language)