meme/docs/TTS_ARCHITECTURE.md

TTS 模块架构设计文档

📐 整体架构

┌─────────────────────────────────────────────────────────────┐
│                        应用层 (API)                         │
│  ┌──────────────────┐  ┌──────────────────┐  ┌───────────┐ │
│  │  FastAPI 路由    │  │  定时任务        │  │  服务层   │ │
│  │ (tts_routes.py)  │  │ (scheduler)      │  │ (services)│ │
│  └──────────────────┘  └──────────────────┘  └───────────┘ │
└──────────────────────────────────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────────┐
│                   高级服务层 (Service)                       │
│          ┌─────────────────────────────────────┐            │
│          │        TTSService                   │            │
│          │  - synthesize()                     │            │
│          │  - get_supported_voices()           │            │
│          │  - get_engine_info()                │            │
│          │  - 自动使用配置文件设置             │            │
│          └─────────────────────────────────────┘            │
└──────────────────────────────────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────────┐
│                  工厂层 (Factory)                            │
│          ┌─────────────────────────────────────┐            │
│          │   TTSEngineFactory                  │            │
│          │  - create()      (创建引擎)         │            │
│          │  - register_engine() (注册引擎)     │            │
│          │  - 单例模式缓存引擎实例             │            │
│          └─────────────────────────────────────┘            │
└──────────────────────────────────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────────┐
│                 抽象层 (Abstract Base)                       │
│          ┌─────────────────────────────────────┐            │
│          │        TTSEngine                    │            │
│          │  - synthesize()                     │            │
│          │  - get_supported_voices()           │            │
│          │  - get_engine_name()                │            │
│          │  - get_engine_version()             │            │
│          └─────────────────────────────────────┘            │
└──────────────────────────────────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────────┐
│              具体实现层 (Implementations)                    │
│  ┌──────────────────┐  ┌──────────────────┐  ┌───────────┐ │
│  │  EdgeTTSEngine   │  │  GoogleTTSEngine │  │  ...      │ │
│  │  (现有)          │  │  (待实现)        │  │  (扩展)   │ │
│  └──────────────────┘  └──────────────────┘  └───────────┘ │
└──────────────────────────────────────────────────────────────┘
                          ▼
┌──────────────────────────────────────────────────────────────┐
│                 第三方 TTS 服务                             │
│  ┌──────────────────┐  ┌──────────────────┐  ┌───────────┐ │
│  │   Edge TTS       │  │   Google TTS     │  │   ...     │ │
│  │   (Microsoft)    │  │   (Google Cloud) │  │   (其他)  │ │
│  └──────────────────┘  └──────────────────┘  └───────────┘ │
└──────────────────────────────────────────────────────────────┘

🏗️ 类关系图

                    ┌──────────────────┐
                    │   TTSEngine      │
                    │   (ABC)          │
                    └──────────────────┘
                            ▲
                            │ inherits
                            │
        ┌───────────────────┼───────────────────┐
        │                   │                   │
  ┌─────────────────┐ ┌──────────────┐ ┌──────────────┐
  │ EdgeTTSEngine   │ │GoogleTTSEngine│ │...           │
  │ (已实现)        │ │(待实现)       │ │(扩展)        │
  └─────────────────┘ └──────────────┘ └──────────────┘

    ┌─────────────────────────────────┐
    │   TTSEngineFactory              │
    │  manages instances of ▲          │
    │                      │           │
    │  - create()          │           │
    │  - register_engine() │           │
    │  - _instances cache  │           │
    └─────────────────────────────────┘

    ┌─────────────────────────────────┐
    │   TTSService                    │
    │  uses ▼                         │
    │                                 │
    │  - TTSEngineFactory             │
    │  - Settings (config)            │
    │  - Logger                       │
    └─────────────────────────────────┘

🔄 执行流程

1. 初始化流程

应用启动
    │
    ├─→ 加载配置 (settings.py)
    │      │
    │      └─→ TTS_ENGINE="edge-tts"
    │          TTS_LANGUAGE="zh-CN"
    │          ...
    │
    └─→ 应用就绪
           │
           └─→ 第一次 TTS 请求到达时初始化引擎

2. 合成语音流程

用户请求 (HTTP POST)
    │
    ├─→ TTSService.synthesize()
    │      │
    │      ├─→ 检查引擎是否已初始化
    │      │      │
    │      │      └─→ 否: TTSEngineFactory.create()
    │      │          │
    │      │          └─→ 创建 EdgeTTSEngine 实例
    │      │              └─→ 缓存实例
    │      │
    │      ├─→ 从配置获取参数
    │      │      (language, voice, rate, pitch)
    │      │
    │      ├─→ engine.synthesize()
    │      │      │
    │      │      └─→ Edge TTS API 调用
    │      │          │
    │      │          └─→ 返回 BytesIO 对象
    │      │
    │      └─→ 返回音频数据
    │
    └─→ API 响应成功

3. 获取声音列表流程

用户请求 (HTTP GET)
    │
    ├─→ TTSService.get_supported_voices()
    │      │
    │      ├─→ 获取或创建引擎实例
    │      │
    │      ├─→ engine.get_supported_voices()
    │      │      │
    │      │      └─→ Edge TTS API 调用
    │      │          │
    │      │          └─→ 获取完整声音列表
    │      │              │
    │      │              └─→ 按语言筛选
    │      │
    │      └─→ 返回声音列表
    │
    └─→ API 响应成功

🎯 设计模式

1. 工厂模式 (Factory Pattern)

目的：统一创建和管理引擎实例

实现：TTSEngineFactory

# 使用工厂创建引擎
engine = TTSEngineFactory.create("edge-tts")

# 自动缓存，再次创建返回同一实例
engine2 = TTSEngineFactory.create("edge-tts")
assert engine is engine2  # True (单例)

优点：

• 统一的创建入口
• 易于添加新引擎
• 自动实例缓存和生命周期管理

2. 抽象基类模式 (ABC Pattern)

目的：定义引擎的统一接口

实现：TTSEngine 基类

from abc import ABC, abstractmethod

class TTSEngine(ABC):
    @abstractmethod
    async def synthesize(self, ...):
        pass

优点：

• 强制所有引擎实现相同接口
• IDE 自动补全和类型检查
• 清晰的契约定义

3. 服务外观模式 (Facade Pattern)

目的：为客户端提供简化的高级接口

实现：TTSService

# 简化的调用
audio = await TTSService.synthesize("文本")

# vs 复杂的调用
engine = TTSEngineFactory.create(settings.TTS_ENGINE)
audio = await engine.synthesize(
    text="文本",
    language=settings.TTS_LANGUAGE,
    voice=settings.TTS_VOICE or None,
    rate=settings.TTS_RATE,
    pitch=settings.TTS_PITCH,
)

优点：

• 隐藏复杂性
• 统一配置管理
• 易于使用

4. 单例模式 (Singleton Pattern)

目的：确保同一引擎类型只有一个实例

实现：TTSEngineFactory._instances

_instances: dict[TTSEngineType, TTSEngine] = {}

# 第一次创建
engine1 = TTSEngineFactory.create("edge-tts")  # 创建新实例

# 第二次创建
engine2 = TTSEngineFactory.create("edge-tts")  # 返回缓存实例

assert engine1 is engine2  # True

优点：

• 节省资源
• 避免重复初始化
• 状态管理简化

📦 模块职责

模块	职责	关键类/函数
base.py	定义引擎接口	TTSEngine
edge_tts_engine.py	实现 Edge-TTS	EdgeTTSEngine
factory.py	管理引擎创建	TTSEngineFactory, TTSEngineType
service.py	提供高级接口	TTSService
tts_routes.py	REST API 端点	4 个 API 端点

🔌 扩展点

新增引擎支持

1. 创建新引擎类 (步骤 1)

   # tts/google_tts_engine.py
   class GoogleTTSEngine(TTSEngine):
       async def synthesize(self, ...):
           # 实现 Google TTS 调用
           pass
   

2. 注册到工厂 (步骤 2)

   # factory.py
   class TTSEngineType(Enum):
       EDGE_TTS = "edge-tts"
       GOOGLE_TTS = "google-tts"  # 新增
   
   class TTSEngineFactory:
       _engines = {
           TTSEngineType.EDGE_TTS: EdgeTTSEngine,
           TTSEngineType.GOOGLE_TTS: GoogleTTSEngine,  # 新增
       }
   

3. 更新配置 (步骤 3)

   # settings.py
   TTS_ENGINE: str = Field("edge-tts")  # 支持 google-tts
   GOOGLE_TTS_API_KEY: str  # 新增 Google TTS 配置
   

4. 在 .env 中配置 (步骤 4)

   TTS_ENGINE=google-tts
   GOOGLE_TTS_API_KEY=your_api_key
   

完成！系统会自动使用新引擎。

🧵 异步设计

所有 TTS 操作都是异步的：

# 高效的并发处理
async def synthesize_multiple(texts: list[str]):
    """并发合成多个文本"""
    tasks = [TTSService.synthesize(text) for text in texts]
    audios = await asyncio.gather(*tasks)
    return audios

好处：

• 支持高并发请求
• 不阻塞其他操作
• 与 FastAPI/APScheduler 无缝集成

💾 状态管理

应用级状态

class TTSService:
    _engine: Optional[TTSEngine] = None  # 共享引擎实例
    _instances: dict = {}                # 引擎实例缓存

特点：

• 懒加载：第一次使用时初始化
• 线程安全：使用装饰器 @classmethod
• 可重置：TTSService.reset_engine()

⚙️ 配置管理

┌─────────────────────────────────┐
│      .env 文件                  │
│  TTS_ENGINE=edge-tts            │
│  TTS_LANGUAGE=zh-CN             │
│  TTS_RATE=1.0                   │
└──────────────┬──────────────────┘
               │
               ▼
┌─────────────────────────────────┐
│      Settings 对象              │
│   (config/settings.py)          │
│  - 验证配置值                   │
│  - 提供类型提示                 │
│  - 支持环境变量覆盖             │
└──────────────┬──────────────────┘
               │
               ▼
┌─────────────────────────────────┐
│      TTSService                 │
│  - 自动使用配置                 │
│  - 可被参数覆盖                 │
└─────────────────────────────────┘

🚦 错误处理

错误处理流程

try:
    audio = await engine.synthesize(text)
except Exception as e:
    logger.error(f"TTS Error: {e}")
    │
    ├─→ 网络错误
    │      └─→ 重试或返回错误信息
    │
    ├─→ 不支持的语言
    │      └─→ 返回支持的语言列表
    │
    ├─→ API 限制
    │      └─→ 实现本地缓存或队列
    │
    └─→ 其他错误
           └─→ 记录详细日志并返回 500

日志级别

• DEBUG: 详细的执行过程
• INFO: 成功的操作
• ERROR: 错误和异常

📊 性能考虑

优化策略

1. 实例缓存

   # 引擎实例只创建一次
   engine = TTSEngineFactory.create("edge-tts")
   engine = TTSEngineFactory.create("edge-tts")  # 返回缓存
   

2. 异步操作

   # 不阻塞其他请求
   audio = await engine.synthesize(text)
   

3. 结果缓存（可选）

   # 缓存常见文本的合成结果
   @functools.lru_cache
   async def synthesize(text: str):
       ...
   

4. 批量处理（可选）

   # 合成多个文本时并发处理
   audios = await asyncio.gather(*tasks)
   

🔐 安全考虑

1. 输入验证

   • 验证文本长度
   • 检查不支持的字符

2. 速率限制

   • 限制每秒请求数
   • 实现请求队列

3. 日志安全

   • 不记录敏感信息
   • 限制日志输出大小

4. API 密钥管理

   • 存储在环境变量中
   • 不在代码中硬编码

📚 相关文件

• tts/README.md - 完整的 API 文档
• TTS_QUICK_START.md - 快速开始指南
• TTS_IMPLEMENTATION_SUMMARY.md - 实现总结
• TTS_INTEGRATION_CHECKLIST.md - 集成检查清单

---

版本: 1.0.0
最后更新: 2025-11-27
作者: AI Assistant