# audio_generator.py - V273.0 (Google Cloud TTS - High Quality Hebrew) import asyncio import base64 import os import tempfile import logging # Configure Logging logger = logging.getLogger(__name__) # ═══════════════════════════════════════════════════════════════ # 🎙️ Google Cloud TTS Configuration # ═══════════════════════════════════════════════════════════════ # # קולות עבריים זמינים: # - he-IL-Wavenet-A (נקבה, איכות גבוהה) ⭐ מומלץ # - he-IL-Wavenet-B (זכר, איכות גבוהה) # - he-IL-Standard-A (נקבה, איכות רגילה) # - he-IL-Standard-B (זכר, איכות רגילה) # # Free Tier: 1 מיליון תווים/חודש (WaveNet: 1M, Standard: 4M) # ═══════════════════════════════════════════════════════════════ GOOGLE_VOICE_NAME = "he-IL-Wavenet-A" # Female, high quality GOOGLE_LANGUAGE_CODE = "he-IL" SPEAKING_RATE = 0.95 # מעט יותר איטי לבהירות PITCH = 1.0 # גובה קול רגיל # Fallback to edge-tts if Google Cloud not configured USE_EDGE_TTS_FALLBACK = True EDGE_TTS_VOICE = "he-IL-HilaNeural" from firebase_manager import firebase_manager # V261.17 def _is_google_cloud_configured() -> bool: """בדיקה אם Google Cloud מוגדר""" # Option 1: Environment variable if os.environ.get("GOOGLE_APPLICATION_CREDENTIALS"): return True # Option 2: Check for credentials file in common locations common_paths = [ "/app/google-credentials.json", "./google-credentials.json", os.path.expanduser("~/.config/gcloud/application_default_credentials.json") ] for path in common_paths: if os.path.exists(path): os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = path return True return False async def _generate_with_google_cloud(text: str, output_path: str) -> bool: """ יצירת אודיו עם Google Cloud TTS מחזיר True אם הצליח, False אם נכשל """ try: from google.cloud import texttospeech # Create client client = texttospeech.TextToSpeechClient() # Build the voice request voice = texttospeech.VoiceSelectionParams( language_code=GOOGLE_LANGUAGE_CODE, name=GOOGLE_VOICE_NAME, ) # Select the audio format audio_config = texttospeech.AudioConfig( audio_encoding=texttospeech.AudioEncoding.MP3, speaking_rate=SPEAKING_RATE, pitch=PITCH, ) # Build the synthesis input synthesis_input = texttospeech.SynthesisInput(text=text) # Perform the text-to-speech request logger.info(f"🎙️ Google Cloud TTS: Generating audio for {len(text)} chars...") # Run in thread pool to not block async loop = asyncio.get_running_loop() response = await loop.run_in_executor( None, lambda: client.synthesize_speech( input=synthesis_input, voice=voice, audio_config=audio_config ) ) # Write the audio content to file with open(output_path, "wb") as out: out.write(response.audio_content) logger.info(f"✅ Google Cloud TTS: Audio saved to {output_path}") return True except ImportError: logger.warning("⚠️ google-cloud-texttospeech not installed. Run: pip install google-cloud-texttospeech") return False except Exception as e: logger.error(f"❌ Google Cloud TTS failed: {e}") return False async def _generate_with_edge_tts(text: str, output_path: str) -> bool: """ יצירת אודיו עם edge-tts (Fallback) """ try: import edge_tts logger.info(f"🎙️ Edge TTS (Fallback): Generating audio...") communicate = edge_tts.Communicate(text, EDGE_TTS_VOICE) await communicate.save(output_path) logger.info(f"✅ Edge TTS: Audio saved to {output_path}") return True except Exception as e: logger.error(f"❌ Edge TTS failed: {e}") return False async def generate_teacher_audio(text: str, output_path: str = None) -> str: """ V273.0: יצירת אודיו עם Google Cloud TTS (איכות גבוהה) מנסה קודם Google Cloud TTS, אם לא מוגדר/נכשל → edge-tts fallback Returns: - Public URL (if Firebase upload success) - Base64 string (fallback) - None (if all failed) """ try: if not text: return None # Clean text for TTS (remove emojis and special chars that cause issues) clean_text = _clean_text_for_tts(text) if not clean_text: return None logger.info(f"🎙️ TTS Request: {clean_text[:50]}...") # Determine output path if output_path: os.makedirs(os.path.dirname(output_path), exist_ok=True) final_path = output_path else: timestamp = int(asyncio.get_event_loop().time() * 1000) final_path = os.path.join(tempfile.gettempdir(), f"audio_{timestamp}.mp3") # Try Google Cloud TTS first success = False if _is_google_cloud_configured(): success = await _generate_with_google_cloud(clean_text, final_path) else: logger.info("ℹ️ Google Cloud not configured, using Edge TTS") # Fallback to edge-tts if not success and USE_EDGE_TTS_FALLBACK: success = await _generate_with_edge_tts(clean_text, final_path) if not success: logger.error("❌ All TTS methods failed") return None # Try Firebase Upload try: blob_name = f"audio/{os.path.basename(final_path)}" loop = asyncio.get_running_loop() public_url = await loop.run_in_executor( None, lambda: firebase_manager.upload_file(final_path, blob_name) ) if public_url: logger.info(f"☁️ Firebase URL: {public_url}") # Clean up local file if not output_path: os.remove(final_path) return public_url except Exception as fb_err: logger.warning(f"⚠️ Firebase upload failed ({fb_err}). Using Base64.") # Fallback: Return Base64 with open(final_path, "rb") as audio_file: audio_bytes = audio_file.read() audio_base64 = base64.b64encode(audio_bytes).decode('utf-8') # Clean up temp file if not output_path: os.remove(final_path) return audio_base64 except Exception as e: logger.error(f"❌ TTS Generation Failed: {e}") return None def _clean_text_for_tts(text: str) -> str: """ ניקוי טקסט לפני TTS - הסרת אימוג'ים וסימנים בעייתיים """ import re if not text: return "" # Remove emojis emoji_pattern = re.compile("[" u"\U0001F600-\U0001F64F" # emoticons u"\U0001F300-\U0001F5FF" # symbols & pictographs u"\U0001F680-\U0001F6FF" # transport & map symbols u"\U0001F1E0-\U0001F1FF" # flags u"\U00002702-\U000027B0" u"\U000024C2-\U0001F251" "]+", flags=re.UNICODE) clean = emoji_pattern.sub('', text) # Remove multiple spaces clean = re.sub(r'\s+', ' ', clean) # Remove LaTeX remnants that might have slipped through clean = clean.replace('$', '').replace('\\', '') return clean.strip() # ═══════════════════════════════════════════════════════════════ # 🧪 Testing # ═══════════════════════════════════════════════════════════════ if __name__ == "__main__": async def main(): text = """ איזה יופי של תרגיל! היינו צריכים למצוא את נקודות הקיצון של הפונקציה. השתמשנו בנגזרת ראשונה כדי למצוא איפה השיפוע מתאפס. הטריק לזכור - נגזרת אפס תמיד מסמנת נקודת קיצון אפשרית. כל הכבוד על ההתמדה! """ print(f"🎙️ Testing TTS...") print(f"📝 Text length: {len(text)} chars") print(f"☁️ Google Cloud configured: {_is_google_cloud_configured()}") result = await generate_teacher_audio(text) if result: if result.startswith("http"): print(f"✅ Got URL: {result}") else: print(f"✅ Got Base64: {len(result)} chars") else: print("❌ TTS failed") asyncio.run(main())