src/speech/speech.py

import re
from typing import Optional

import edge_tts
from langdetect import detect

from src.utils.logger import setup_logger

logger = setup_logger(__name__)


class TextToSpeechService:
    """
    Service responsible for converting generated summaries
    or notes into speech using Microsoft Edge TTS.

    Features:
    1. Text cleaning
    2. Automatic language detection
    3. Voice selection based on language
    4. Audio generation
    """

    VOICE_MAP = {
        "ar": "ar-EG-SalmaNeural",
        "en": "en-US-AriaNeural",
        "fr": "fr-FR-DeniseNeural",
        "ja": "ja-JP-NanamiNeural",
        "de": "de-DE-KatjaNeural",
        "es": "es-ES-ElviraNeural",
        "it": "it-IT-ElsaNeural",
        "pt": "pt-BR-FranciscaNeural",
        "ru": "ru-RU-SvetlanaNeural",
        "zh-cn": "zh-CN-XiaoxiaoNeural",
        "zh-tw": "zh-TW-HsiaoChenNeural",
        "ko": "ko-KR-SunHiNeural",
        "tr": "tr-TR-EmelNeural",
        "pl": "pl-PL-ZofiaNeural",
        "sv": "sv-SE-SofieNeural",
        "hi": "hi-IN-SwaraNeural",
        "id": "id-ID-GadisNeural",
    }

    DEFAULT_VOICE = "en-US-AriaNeural"

    # ──────────────────────────────────────────────
    # Text Cleaning
    # ──────────────────────────────────────────────
    def _clean_text(self, text: str) -> str:
        """
        Remove URLs, markdown symbols, YouTube metadata,
        and unnecessary characters before TTS generation.
        """
        # 1. Remove YouTube header block first (before anything else)
        text = re.sub(
            r'#\s*YouTube Video.*?---',
            '',
            text,
            flags=re.DOTALL
        )

        text = re.sub(
            r'^.*?(المصدر|المدة|Source|Duration|YouTube|youtu\.be|youtube\.com).*$',
            '',
            text,
            flags=re.MULTILINE | re.IGNORECASE
        )
        # 2. Remove URLs
        text = re.sub(
            r'https?://\S+|www\.\S+',
            ' ',
            text
        )
        # 3. Remove standalone numbers on their own line
        text = re.sub(
            r'^\s*\d+\s*$',
            '',
            text,
            flags=re.MULTILINE
        )
        # 4. Remove Markdown headers
        text = re.sub(r'#+', ' ', text)
        # 5. Remove Markdown symbols
        text = re.sub(r'[*_~`>|]', ' ', text)
        # 6. Remove dashes
        text = re.sub(r'-+', ' ', text)
        # 7. Keep only useful characters
        text = re.sub(
            r'[^\w\s\u0600-\u06FF.,!?؟؛:\n]',
            ' ',
            text
        )
        # 8. Collapse extra spaces
        text = re.sub(r'\s+', ' ', text)

        return text.strip()

    # ──────────────────────────────────────────────
    # Language Detection
    # ──────────────────────────────────────────────
    def _detect_voice(self, text: str) -> str:
        """
        Detect language and return matching voice.
        """

        try:
            language = detect(text)

            logger.info(
                f"Detected language: {language}"
            )

            return self.VOICE_MAP.get(
                language,
                self.DEFAULT_VOICE
            )

        except Exception as e:
            logger.warning(
                f"Language detection failed: {e}"
            )

            return self.DEFAULT_VOICE

    # ──────────────────────────────────────────────
    # Generate Speech
    # ──────────────────────────────────────────────
    async def generate_audio(
        self,
        text: str,
        output_file: str = "summary.mp3",
        rate: str = "-10%"
    ) -> str:
        """
        Convert text into speech and save it as MP3.
        Returns generated audio path.
        """

        cleaned_text = self._clean_text(text)

        if not cleaned_text:
            raise ValueError(
                "No valid text available for speech generation."
            )

        voice = self._detect_voice(cleaned_text)

        logger.info(
            f"Generating speech using voice: {voice}"
        )

        communicate = edge_tts.Communicate(
            text=cleaned_text,
            voice=voice,
            rate=rate
        )

        await communicate.save(output_file)

        logger.info(
            f"Audio saved successfully: {output_file}"
        )

        return output_file