cang-ying/engines/tts-engine.py

#!/usr/bin/env python3
"""
TTS Engine · Edge-TTS 配音引擎
==================================
为视频AI系统提供文本转语音能力，支持多角色音色配置。

依赖：
  pip install edge-tts

用法：
  # 基本用法（默认中文女声）
  python tts-engine.py --text "你好世界" --output output.mp3

  # 指定角色（从配置文件读取音色）
  python tts-engine.py --text "未来的天下第一宗！" --character "苏白" --output su-bai.mp3

  # 指定语音（Edge-TTS 语音名）
  python tts-engine.py --text "Hello World" --voice "en-US-JennyNeural" --output hello.mp3

  # 调整语速/音调/音量
  python tts-engine.py --text "你好" --rate "+20%" --pitch "+5Hz" --volume "+10%" --output output.mp3

  # 批量生成（从SRT字幕文件）
  python tts-engine.py --srt input.srt --output-dir ./audio/ --character "苏白"

  # 作为模块导入
  from tts_engine import generate_speech
  generate_speech("你好世界", "output.mp3", voice="zh-CN-XiaoxiaoNeural")

角色音色配置：
  video-ai-system/config/voices.json

路径：
  video-ai-system/engines/tts-engine.py
"""

import argparse
import asyncio
import json
import os
import sys
from pathlib import Path

try:
    import edge_tts
except ImportError:
    print("[ERROR] 缺少依赖：edge-tts")
    print("请先安装：pip install edge-tts")
    sys.exit(1)


# 默认角色音色配置
DEFAULT_VOICES = {
    "苏白": {
        "voice": "zh-CN-XiaoxiaoNeural",  # 阳光少年音
        "rate": "+5%",
        "pitch": "+0Hz"
    },
    "诸葛风": {
        "voice": "zh-CN-YunxiNeural",  # 沉稳男声
        "rate": "+0%",
        "pitch": "-5Hz"
    },
    "萧灵汐": {
        "voice": "zh-CN-XiaoyiNeural",  # 清冷女声
        "rate": "+0%",
        "pitch": "+0Hz"
    },
    "王执事": {
        "voice": "zh-CN-YunyangNeural",  # 中年男声
        "rate": "+0%",
        "pitch": "-10Hz"
    }
}


def load_voice_config(config_path: str = None) -> dict:
    """加载角色音色配置文件"""
    if config_path and os.path.isfile(config_path):
        try:
            with open(config_path, "r", encoding="utf-8") as f:
                return json.load(f)
        except Exception as e:
            print(f"[WARN] 无法读取配置文件 {config_path}：{e}")
            print("[INFO] 使用默认音色配置")
    return DEFAULT_VOICES


async def generate_speech_async(
    text: str,
    output_path: str,
    voice: str = "zh-CN-XiaoxiaoNeural",
    rate: str = "+0%",
    pitch: str = "+0Hz",
    volume: str = "+0%"
) -> bool:
    """
    异步生成语音（Edge-TTS）

    :param text: 要合成的文本
    :param output_path: 输出音频文件路径
    :param voice: Edge-TTS 语音名
    :param rate: 语速（如 "+20%"、"-10%"）
    :param pitch: 音调（如 "+5Hz"、"-10Hz"）
    :param volume: 音量（如 "+10%"、"-5%"）
    :return: 是否成功
    """
    try:
        # 确保输出目录存在
        os.makedirs(os.path.dirname(os.path.abspath(output_path)), exist_ok=True)

        communicate = edge_tts.Communicate(
            text, voice,
            rate=rate, pitch=pitch, volume=volume
        )
        await communicate.save(output_path)

        # 验证输出文件
        if os.path.isfile(output_path):
            file_size = os.path.getsize(output_path)
            print(f"[OK] 生成语音：{output_path} ({file_size // 1024} KB)")
            print(f"     语音：{voice} | 语速：{rate} | 音调：{pitch}")
            return True
        else:
            print(f"[ERROR] 输出文件未生成：{output_path}")
            return False

    except Exception as e:
        print(f"[ERROR] 生成语音失败：{e}")
        return False


def generate_speech(
    text: str,
    output_path: str,
    voice: str = "zh-CN-XiaoxiaoNeural",
    rate: str = "+0%",
    pitch: str = "+0Hz",
    volume: str = "+0%"
) -> bool:
    """
    同步包装器（供外部调用）

    :param text: 要合成的文本
    :param output_path: 输出音频文件路径
    :param voice: Edge-TTS 语音名
    :param rate: 语速
    :param pitch: 音调
    :param volume: 音量
    :return: 是否成功
    """
    return asyncio.run(generate_speech_async(text, output_path, voice, rate, pitch, volume))


def generate_by_character(
    text: str,
    output_path: str,
    character: str,
    config: dict = None
) -> bool:
    """
    按角色名生成语音（自动读取音色配置）

    :param text: 要合成的文本
    :param output_path: 输出音频文件路径
    :param character: 角色名（如 "苏白"）
    :param config: 音色配置字典（可选，默认加载 DEFAULT_VOICES）
    :return: 是否成功
    """
    if config is None:
        config = load_voice_config()

    if character not in config:
        print(f"[WARN] 角色 '{character}' 未配置音色，使用默认音色")
        return generate_speech(text, output_path)

    voice_config = config[character]
    return generate_speech(
        text, output_path,
        voice=voice_config.get("voice", "zh-CN-XiaoxiaoNeural"),
        rate=voice_config.get("rate", "+0%"),
        pitch=voice_config.get("pitch", "+0Hz")
    )


def process_srt(srt_path: str, output_dir: str, character: str = None, config: dict = None):
    """
    从SRT字幕文件批量生成语音

    :param srt_path: SRT 文件路径
    :param output_dir: 输出目录
    :param character: 角色名（所有台词用同一音色）
    :param config: 音色配置字典
    """
    if not os.path.isfile(srt_path):
        print(f"[ERROR] SRT 文件不存在：{srt_path}")
        return False

    os.makedirs(output_dir, exist_ok=True)

    # 简单SRT解析（按空行分割）
    with open(srt_path, "r", encoding="utf-8") as f:
        content = f.read()

    blocks = content.strip().split("\n\n")
    print(f"[INFO] 找到 {len(blocks)} 条字幕，开始生成语音...")

    success = 0
    for block in blocks:
        lines = block.strip().split("\n")
        if len(lines) < 3:
            continue

        idx = lines[0].strip()
        # timeine = lines[1].strip()  # 暂不使用时序
        text = " ".join(lines[2:]).strip()

        if not text:
            continue

        output_path = os.path.join(output_dir, f"{idx.zfill(4)}.mp3")

        if character:
            ok = generate_by_character(text, output_path, character, config)
        else:
            ok = generate_speech(text, output_path)

        if ok:
            success += 1

    print(f"\n[INFO] SRT批量生成完成：{success}/{len(blocks)} 成功")
    return success == len(blocks)


def list_voices():
    """列出所有可用的Edge-TTS语音"""
    print("[INFO] 正在获取可用语音列表...\n")
    asyncio.run(_list_voices_async())


async def _list_voices_async():
    """异步列出语音"""
    voices = await edge_tts.list_voices()
    print("中文语音：")
    for v in voices:
        if v["Locale"].startswith("zh-"):
            print(f"  {v['ShortName']:30s} - {v['FriendlyName']}")
    print(f"\n共 {len([v for v in voices if v['Locale'].startswith('zh-')])} 个中文语音")


def main():
    parser = argparse.ArgumentParser(
        description="TTS Engine · Edge-TTS 配音引擎",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
示例：
  python tts-engine.py --text "你好世界" --output output.mp3
  python tts-engine.py --text "未来的天下第一宗！" --character "苏白" --output su-bai.mp3
  python tts-engine.py --text "Hello" --voice "en-US-JennyNeural" --output hello.mp3
  python tts-engine.py --srt input.srt --output-dir ./audio/ --character "苏白"
  python tts-engine.py --list-voices
"""
    )
    parser.add_argument("--text", help="要合成的文本")
    parser.add_argument("--output", help="输出音频文件路径")
    parser.add_argument("--voice", default="zh-CN-XiaoxiaoNeural", help="Edge-TTS 语音名")
    parser.add_argument("--character", help="角色名（从配置文件读取音色）")
    parser.add_argument("--rate", default="+0%", help="语速（如 +20%%、-10%%）")
    parser.add_argument("--pitch", default="+0Hz", help="音调（如 +5Hz、-10Hz）")
    parser.add_argument("--volume", default="+0%", help="音量（如 +10%%、-5%%）")
    parser.add_argument("--srt", help="从SRT字幕文件批量生成")
    parser.add_argument("--output-dir", help="批量生成时的输出目录")
    parser.add_argument("--config", help="角色音色配置文件路径")
    parser.add_argument("--list-voices", action="store_true", help="列出所有可用语音")

    args = parser.parse_args()

    if args.list_voices:
        list_voices()
        sys.exit(0)

    if args.srt:
        # 批量模式
        if not args.output_dir:
            print("[ERROR] 批量模式需要指定 --output-dir")
            sys.exit(1)
        config = load_voice_config(args.config)
        ok = process_srt(args.srt, args.output_dir, args.character, config)
        sys.exit(0 if ok else 1)

    if not args.text or not args.output:
        print("[ERROR] 需要指定 --text 和 --output")
        parser.print_help()
        sys.exit(1)

    # 单文件模式
    if args.character:
        config = load_voice_config(args.config)
        ok = generate_by_character(args.text, args.output, args.character, config)
    else:
        ok = generate_speech(
            args.text, args.output,
            voice=args.voice, rate=args.rate,
            pitch=args.pitch, volume=args.volume
        )

    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()