import asyncio
import hashlib
import os
import platform
import re
import time
from pathlib import Path
from typing import Optional

import edge_tts
import pyttsx3
from loguru import logger


class TTSEngine:
    """支持多平台和智能语音选择的多引擎TTS工具类"""

    def __init__(self):
        """
        初始化TTS引擎实例
        属性：
        - cache_dir: 音频缓存目录路径（软件运行目录下 cache/audio文件夹）
        - engine_priority: 引擎优先级列表
        - voice_mapping: 跨平台语音映射配置表
        """
        self.cache_dir = os.path.join(os.getcwd(), "cache", "audio")
        self._ensure_cache_dir()
        self.engine_priority = ['edge', 'pyttsx3']

        # 跨平台语音映射表
        self.voice_mapping = {
            'edge': {
                'zh-CN': 'zh-CN-YunxiNeural',
                'en-US': 'en-US-AriaNeural'
            },
            'pyttsx3': self._get_platform_voices()
        }

    @staticmethod
    def _get_platform_voices():
        """
        获取当前平台的默认语音配置

        返回：
        - dict: 包含中英文语音ID的字典，结构为{'zh-CN': voice_id, 'en-US': voice_id}

        平台支持：
        - Windows: 使用注册表路径标识语音
        - macOS: 使用Apple语音标识符
        - Linux: 使用espeak语音名称
        """
        current_os = platform.system()

        # Windows默认配置
        if current_os == 'Windows':
            return {
                'zh-CN': 'HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Speech\\Voices\\Tokens\\TTS_MS_ZH-CN_HUIHUI_11.0',
                'en-US': 'HKEY_LOCAL_MACHINE\\SOFTWARE\\Microsoft\\Speech\\Voices\\Tokens\\TTS_MS_EN-US_DAVID_11.0'
            }
        # macOS默认配置
        elif current_os == 'Darwin':
            return {
                'zh-CN': 'com.apple.speech.synthesis.voice.ting-ting.premium',
                'en-US': 'com.apple.speech.synthesis.voice.Alex'
            }
        # Linux默认配置 (espeak)
        else:
            return {
                'zh-CN': 'chinese',
                'en-US': 'english-us'
            }

    def _ensure_cache_dir(self):
        Path(self.cache_dir).mkdir(parents=True, exist_ok=True)

    @staticmethod
    def _generate_filename(text: str, engine: str) -> str:
        timestamp = str(int(time.time()))
        hash_str = hashlib.md5(text.encode()).hexdigest()[:8]
        return f"{engine}_{hash_str}_{timestamp}.mp3"

    @staticmethod
    async def _edge_tts(text: str, voice: str, file_path: str) -> str:
        communicate = edge_tts.Communicate(text, voice)
        await communicate.save(file_path)
        return file_path

    async def _pyttsx3_tts(self, text: str, voice: str, file_path: str) -> str:
        loop = asyncio.get_running_loop()
        return await loop.run_in_executor(
            None,
            lambda: self._sync_pyttsx3(text, voice, file_path)
        )

    @staticmethod
    def _sync_pyttsx3(text: str, voice: str, file_path: str):
        engine = None
        try:
            engine = pyttsx3.init()
            engine.connect('started-utterance', lambda name: None)
            engine.connect('finished-utterance', lambda name, completed: engine.stop())

            # 应用语音设置
            if voice:
                voices = engine.getProperty('voices')
                found_voice = next((v for v in voices if v.id == voice), None)
                if not found_voice:
                    raise ValueError(f"无效语音ID：{voice}")
                engine.setProperty('voice', found_voice.id)

            engine.save_to_file(text, file_path)
            start_time = time.time()
            engine.startLoop(False)
            while engine.isBusy():
                if time.time() - start_time > 10:
                    raise TimeoutError("pyttsx3生成超时")
                time.sleep(0.1)
                engine.iterate()
            engine.endLoop()
        finally:
            if engine:
                engine.stop()

    @staticmethod
    def _detect_language(text: str) -> str:
        """改进的语言检测方法"""
        if re.search(u'[\u4e00-\u9fff]', text):
            return 'zh-CN'
        return 'en-US'

    @staticmethod
    def _validate_pyttsx3_voice(voice_id: str, lang: str) -> str:
        """验证语音有效性，自动回退"""
        try:
            engine = pyttsx3.init()
            voices = engine.getProperty('voices')

            if any(v.id == voice_id for v in voices):
                return voice_id

            lang_voices = [v for v in voices if lang in str(v.languages)]
            if lang_voices:
                return lang_voices[0].id

            return engine.getProperty('voice')
        except Exception as e:
            logger.error(f"语音验证失败: {str(e)}")
            return ''

    async def _execute_engine(
            self,
            engine: str,
            text: str,
            voice: str,
            file_path: str,
            timeout: float
    ) -> str:
        """
        生成语音文件的核心异步方法

        参数：
        text (str): 要转换的文本内容（支持中英文自动检测）
        engine (str): 首选TTS引擎（默认edge）
        voice (str): 指定语音ID（可选），不指定则根据语言自动选择
        auto_fallback (bool): 引擎失败时是否自动回退（默认False）
        timeout (float): 单引擎超时时间（秒，默认10）
        filename (str): 自定义文件名（可选），不指定则自动生成

        返回：
        str: 生成的音频文件绝对路径

        异常：
        RuntimeError: 所有尝试的引擎均失败时抛出
        """
        try:
            if engine == "edge":
                task = self._edge_tts(text, voice, file_path)
            elif engine == "pyttsx3":
                task = self._pyttsx3_tts(text, voice, file_path)
            else:
                raise ValueError(f"不支持的引擎：{engine}")

            return await asyncio.wait_for(task, timeout=timeout)
        except asyncio.TimeoutError:
            raise RuntimeError(f"{engine}引擎执行超时")
        except Exception as e:
            raise RuntimeError(f"{engine}引擎错误：{str(e)}")

    async def generate_speech(
            self,
            text: str,
            engine: str = "edge",
            voice: Optional[str] = None,
            auto_fallback: bool = False,
            timeout: float = 10.0,
            filename: Optional[str] = None
    ) -> str:
        """核心生成方法"""

        # 自动语音选择逻辑
        lang = self._detect_language(text)
        if not voice:
            if engine == 'pyttsx3':
                voice = self.voice_mapping[engine].get(lang)
                voice = self._validate_pyttsx3_voice(voice, lang)
            else:
                voice = self.voice_mapping[engine][lang]

        filename = filename or self._generate_filename(text, engine)
        file_path = os.path.join(self.cache_dir, filename)

        errors = []
        attempted_engines = set()
        engines_to_try = [engine]
        if auto_fallback:
            for e in self.engine_priority:
                if e != engine and e not in engines_to_try:
                    engines_to_try.append(e)

        for current_engine in engines_to_try:
            if current_engine in attempted_engines:
                continue
            if current_engine not in self.engine_priority:
                continue

            attempted_engines.add(current_engine)

            try:
                await self._execute_engine(
                    engine=current_engine,
                    text=text,
                    voice=voice,
                    file_path=file_path,
                    timeout=timeout
                )

                actual_filename = self._generate_filename(text, current_engine)
                actual_path = os.path.join(self.cache_dir, actual_filename)
                os.rename(file_path, actual_path)

                if not os.path.exists(actual_path):
                    raise RuntimeError(f"语音文件生成失败: {actual_path}")

                logger.info(f"成功生成语音 | 引擎: {current_engine} | 路径: {actual_path}")
                return actual_path

            except Exception as e:
                errors.append(f"{current_engine}: {str(e)}")
                continue

        raise RuntimeError(
            f"所有引擎尝试失败\n" +
            "\n".join(errors)
        )

    def cleanup(self, max_age: int = 86400):
        now = time.time()
        for f in Path(self.cache_dir).glob("*.*"):
            if f.is_file() and (now - f.stat().st_mtime) > max_age:
                f.unlink()

    @staticmethod
    def delete_audio_file(file_path: str, retries: int = 3, delay: float = 0.5):
        """
        安全删除音频文件
        参数:
            retries: 重试次数
            delay: 重试间隔(秒)
        """
        for attempt in range(retries):
            try:
                if os.path.exists(file_path):
                    os.remove(file_path)
                    logger.info(f"成功删除音频文件: {file_path}")
                    return True
            except Exception as e:
                if attempt < retries - 1:
                    logger.warning(f"删除失败，正在重试 ({attempt + 1}/{retries}): {str(e)}")
                    time.sleep(delay)
                else:
                    logger.error(f"最终删除失败: {file_path} | 错误: {str(e)}")
        return False


def generate_speech_sync(
        text: str,
        engine: str = "edge",
        voice: Optional[str] = None,
        auto_fallback: bool = False,
        timeout: float = 10.0,
        filename: Optional[str] = None
) -> str:
    """同步生成方法"""
    tts = TTSEngine()
    return asyncio.run(tts.generate_speech(
        text=text,
        engine=engine,
        voice=voice,
        auto_fallback=auto_fallback,
        timeout=timeout,
        filename=filename
    ))


def list_pyttsx3_voices():
    """跨平台语音列表显示"""
    engine = pyttsx3.init()
    voices = engine.getProperty('voices')
    current_os = platform.system()

    for idx, voice in enumerate(voices):
        logger.info(f"\n[{current_os} 平台Pyttsx3可用语音包]"
                    f"\n{idx + 1}. ID: {voice.id}"
                    f"\n   名称: {voice.name}"
                    f"\n   语言: {voice.languages[0] if voice.languages else '未知'}"
                    f"\n   性别: {voice.gender}"
                    f"\n" + "-" * 60)