AI 科普长视频（10-30分钟）配音 TTS 方案深度调研

git clone https://github.com/alibaba-damo-academy/CosyVoice
cd CosyVoice
# 下载模型（~2GB）
python -m pip install modelscope
# 推理
python inference.py --text "这是演示文本" --spk_id "中文女" --output_file "output.wav"

git clone https://github.com/index-tts/index-tts
cd index-tts
pip install -r requirements.txt
# ComfyUI集成也已支持

import re

def split_sentences(text, max_length=500):
    """
    按自然段切分，避免长文本崩坏
    - max_length: 每段最大字数（建议300-500）
    - 保留标点，便于后续处理
    """
    # 按句号、问号、感叹号切分
    sentences = re.split(r'(?<=[。？！])', text)
    
    segments = []
    current_segment = ""
    
    for sent in sentences:
        if len(current_segment) + len(sent) <= max_length:
            current_segment += sent
        else:
            if current_segment:
                segments.append(current_segment)
            current_segment = sent
    
    if current_segment:
        segments.append(current_segment)
    
    return segments

# 测试
text = "这是一个科普视频。它讲解了量子计算的原理。量子比特有独特的性质。..."
segments = split_sentences(text, max_length=300)
print(f"分成 {len(segments)} 段")

import asyncio
import httpx

async def batch_tts(segments, api_key, model="fish-audio-s2-pro"):
    """
    并发调用TTS API（并发数5-10）
    """
    async with httpx.AsyncClient() as client:
        tasks = []
        
        # 限制并发数为8
        semaphore = asyncio.Semaphore(8)
        
        async def call_tts(text, idx):
            async with semaphore:
                response = await client.post(
                    f"https://api.fish.audio/v1/tts",
                    json={
                        "text": text,
                        "model": model,
                        "voice_id": "default_cn"
                    },
                    headers={"Authorization": f"Bearer {api_key}"}
                )
                return idx, response.content
        
        tasks = [call_tts(seg, i) for i, seg in enumerate(segments)]
        results = await asyncio.gather(*tasks)
        
        # 按原始顺序排序
        results.sort(key=lambda x: x[0])
        return [r[1] for r in results]

# 使用
# audio_chunks = await batch_tts(segments, api_key="sk_xxx")

def preprocess_text_for_tts(text):
    """
    清理文本，便于TTS朗读
    """
    # 1. 数字转汉字
    text = convert_numbers_to_chinese(text)
    # 例：2024年 → 二〇二四年 或 两千零二十四年
    
    # 2. 英文首字母缩写展开
    text = re.sub(r'AI(?![A-Za-z])', 'A I', text)  # AI → A I
    text = re.sub(r'TTS(?![A-Za-z])', 'T T S', text)  # TTS → T T S
    text = re.sub(r'DNA(?![A-Za-z])', 'D N A', text)
    
    # 3. URL/邮箱移除
    text = re.sub(r'https?://\S+', '', text)
    text = re.sub(r'\S+@\S+', '', text)
    
    # 4. 去除特殊符号（保留基础标点）
    text = re.sub(r'[^\\u4e00-\\u9fff\\u3040-\\u309fa-zA-Z0-9。，、；：？！""''（）·]', '', text)
    
    return text

def convert_numbers_to_chinese(text):
    """
    数字转汉字示例（简化版）
    """
    # 实际生产中用更复杂的库：python-pinyin、num2words
    text = text.replace('2024', '二〇二四')
    text = text.replace('99.9%', '百分之九十九点九')
    return text

<!-- 火山引擎豆包TTS SSML示例 -->
<speak>
  <s>这是第一句话。<break time="500ms"/></s>
  <s>这是第二句话。<break time="800ms"/></s>
</speak>

<!-- OpenAI / Azure TTS SSML -->
<speak>
  段落一内容。<break time="500ms"/>
  段落二内容。<break time="1000ms"/>
</speak>

def add_pauses_between_segments(segments, pause_duration_ms=800):
    """
    在分段文本间插入SSML停顿标记
    """
    ssml_segments = []
    
    for i, seg in enumerate(segments):
        # 移除末尾的句号，避免重复
        seg = seg.rstrip('。')
        
        if i < len(segments) - 1:
            # 非最后一段，添加停顿
            ssml = f'<s>{seg}<break time="{pause_duration_ms}ms"/></s>'
        else:
            ssml = f'<s>{seg}</s>'
        
        ssml_segments.append(ssml)
    
    # 生成最终SSML
    final_ssml = f'<speak>{"".join(ssml_segments)}</speak>'
    return final_ssml

# 测试
segments = ["这是介绍。", "现在讲解原理。", "最后是总结。"]
ssml = add_pauses_between_segments(segments, pause_duration_ms=1000)
print(ssml)
# 输出：
# <speak>
#   <s>这是介绍<break time="1000ms"/></s>
#   <s>现在讲解原理<break time="1000ms"/></s>
#   <s>最后是总结</s>
# </speak>

# 第一遍：测量音频
ffmpeg -i input.wav -af loudnorm=I=-16:TP=-1.5:LRA=11:print_format=json -f null - 2>&1 | grep -A 10 "Loudness"

# 第二遍：使用测量结果归一化
ffmpeg -i input.wav \
  -af "loudnorm=I=-16:TP=-1.5:LRA=11:measured_I=<measured_value>:measured_TP=<measured_tp>:measured_LRA=<measured_lra>" \
  output.wav

# 快速版（单遍）
ffmpeg -i input.wav -af "loudnorm=I=-16:TP=-1.5:LRA=11" output_normalized.wav

from pydub import AudioSegment
import os

def concat_audio_chunks(chunk_dir, output_file, target_lufs=-16):
    """
    拼接多段TTS音频并归一化
    """
    combined = AudioSegment.empty()
    
    # 按文件名顺序读取
    chunk_files = sorted([f for f in os.listdir(chunk_dir) if f.endswith('.wav')])
    
    for chunk_file in chunk_files:
        audio = AudioSegment.from_wav(os.path.join(chunk_dir, chunk_file))
        
        # 添加句间停顿（800ms）
        silence = AudioSegment.silent(duration=800)
        combined += audio + silence
    
    # 导出到MP3（压缩存储）
    combined.export(output_file, format="mp3", bitrate="192k")
    
    print(f"音频已拼接，保存到 {output_file}")
    print(f"总时长：{len(combined) / 1000:.1f}秒")

# 使用ffmpeg-python规范化
import subprocess

def normalize_audio_ffmpeg(input_file, output_file, target_lufs=-16):
    """
    使用FFmpeg对单个文件归一化
    """
    cmd = [
        'ffmpeg',
        '-i', input_file,
        '-af', f'loudnorm=I={target_lufs}:TP=-1.5:LRA=11',
        '-y',  # 覆盖输出文件
        output_file
    ]
    subprocess.run(cmd, check=True)
    print(f"✓ 已规范化到 {target_lufs} LUFS：{output_file}")

# 测试
# normalize_audio_ffmpeg("chunk_1.wav", "chunk_1_normalized.wav", target_lufs=-16)

<!-- 中文多音字SSML处理 -->
<speak>
  今天<phoneme py="jīn">今</phoneme>天气很好。
  <!-- 正确读音：jīn tiān -->
  
  这个<phoneme py="de">的</phoneme>字很难读。
  <!-- 正确读音：de（不是dī） -->
  
  科学<phoneme py="xuéxué">学</phoneme>原理。
  <!-- 多音字：xuéxué表示该字读xué -->
</speak>

from pypinyin import pinyin, lazy_pinyin, Style

def fix_polyphone_reading(text):
    """
    检测多音字并生成SSML标注
    """
    # 使用pypinyin库
    result = pinyin(text, style=Style.NORMAL, heteronym=True)
    
    ssml_text = "<speak>"
    
    for char, pinyins in zip(text, result):
        if len(pinyins) > 1:
            # 这是多音字，标注默认读音
            ssml_text += f'<phoneme py="{pinyins[0]}">{char}</phoneme>'
        else:
            ssml_text += char
    
    ssml_text += "</speak>"
    return ssml_text

# 测试
text = "这是量子计算的理论基础。"
ssml = fix_polyphone_reading(text)
print(ssml)

# 建立领域词典，将生僻术语映射到正确发音
TECH_GLOSSARY = {
    "量子比特": "liàngzǐ bǐtè",  # 或用pinyin标注
    "叠加态": "diéjiātài",
    "EPR悖论": "E P R bèilùn",    # 英文字母分开读
    "λ": "lambda",  # 希腊字母需特殊处理
    "Schrödinger": "薛定谔",      # 外国人名需中文化
}

def apply_glossary(text, glossary=TECH_GLOSSARY):
    """
    将专业术语替换为可读文本
    """
    for term, reading in glossary.items():
        text = text.replace(term, f'<phoneme>{reading}</phoneme>')
    return text

def add_emotion_tags(text):
    """
    在关键段落添加情感标记
    """
    emotional_text = """
    <speak>
      <emotion name="friendly" intensity="0.8">
        你好，我来给大家讲解量子计算。
      </emotion>
      
      <emotion name="serious" intensity="0.6">
        这里需要注意的是，量子比特的叠加态是非常特殊的。
      </emotion>
      
      <emotion name="encouraging" intensity="0.7">
        别担心，下面我们会详细讲解这一概念。
      </emotion>
    </speak>
    """
    return emotional_text

# 火山引擎API调用
# POST /v1/tts
# {
#   "text": "<speak>...</speak>",
#   "voice_id": "zh_CN_female_1",
#   "format": "wav"
# }

<speak>
  <prosody rate="0.95">这一句话要读得慢一点，让听众仔细理解。</prosody>
  
  <prosody rate="1.0">这是正常语速。</prosody>
  
  <prosody pitch="+5%">这句话要读得稍微高一点，显得兴奋。</prosody>
  
  重要信息 <break time="1.5s"/> 停顿后重申。
</speak>

# 1. 生成基础配音
audio = client.tts(text, model="fish-audio-s2-pro")

# 2. 用Audacity/Adobe Audition人工调整：
#    - 加重关键词
#    - 调整标题/转场语速
#    - 插入背景音乐（-20dB混音）

# 3. 混音脚本化（ffmpeg）
import subprocess

def mix_with_bgm(tts_audio, bgm_audio, output):
    """
    混合TTS和背景音乐
    """
    cmd = [
        'ffmpeg',
        '-i', tts_audio,
        '-i', bgm_audio,
        '-filter_complex', 
        '[0]volume=1.0[a];[1]volume=0.2[b];[a][b]amix=inputs=2:duration=first[out]',
        '-map', '[out]',
        output
    ]
    subprocess.run(cmd, check=True)

最优质素材标准：
- 时长：30-60秒（越多越好，up to 5分钟）
- 环境：安静室内，避免背景噪音
- 设备：USB麦克风或录音笔（>44.1kHz采样率）
- 内容：朗读多种场景的文本（温和、激动、平静各种情感）
- 格式：WAV或MP3

采集脚本示例：
"你好，我是李明。今天为大家讲解量子计算的基本原理。
这是一个非常复杂但重要的话题。
我们先从经典计算开始。"

import requests

def clone_voice_with_fish_audio(audio_file_path, api_key, voice_name="my_voice"):
    """
    上传参考音频克隆声音（Fish Audio API）
    """
    with open(audio_file_path, 'rb') as f:
        files = {'file': f}
        response = requests.post(
            'https://api.fish.audio/v1/voices/clone',
            files=files,
            headers={'Authorization': f'Bearer {api_key}'},
            data={'name': voice_name}
        )
    
    result = response.json()
    voice_id = result['voice_id']  # 保存此ID用于后续合成
    print(f"✓ 声音克隆成功，voice_id: {voice_id}")
    return voice_id

# 使用克隆的声音生成配音
def tts_with_cloned_voice(text, voice_id, api_key):
    response = requests.post(
        'https://api.fish.audio/v1/tts',
        json={'text': text, 'voice_id': voice_id},
        headers={'Authorization': f'Bearer {api_key}'}
    )
    return response.content  # 音频二进制

# 1. 准备参考音频文件夹
mkdir -p reference_audio
# 放入5-10个.wav文件（共1-2分钟）

# 2. 克隆
python GPT-SoVITS/gpt_sovits.py \
  --ref_audio reference_audio \
  --voice_name "my_voice" \
  --model "gpt-sovits-v4"

# 3. 生成配音
python inference.py \
  --text "要合成的文本" \
  --voice_id "my_voice" \
  --output "output.wav"

# 方案A：克隆多个声音
voices = {
    'host': clone_voice_with_fish_audio('host_sample.wav'),
    'expert': clone_voice_with_fish_audio('expert_sample.wav'),
    'narrator': clone_voice_with_fish_audio('narrator_sample.wav'),
}

# 方案B：混合使用预设音色 + 克隆音色
voices = {
    'host': 'cloned_voice_id_xxx',
    'expert': 'default_male_2',  # 预设音色
    'narrator': 'cloned_voice_id_yyy',
}

def generate_multi_speaker_audio(dialogue_script, voices, output_dir):
    """
    对话脚本格式：
    [host]: "欢迎嘉宾..."
    [expert]: "感谢主持..."
    """
    segments = []
    
    for line in dialogue_script.split('\n'):
        if line.strip():
            speaker, text = line.split(']: ')
            speaker = speaker.strip('[')
            text = text.strip().strip('"')
            
            audio = tts_with_cloned_voice(
                text, 
                voices[speaker],
                api_key='sk_xxx'
            )
            segments.append(audio)
    
    # 拼接和后处理...
    return concat_audio_chunks(segments)

import requests
import json

def get_tts_with_timestamps(text, api_key):
    """
    Fish Audio TTS + word-level timestamp
    """
    response = requests.post(
        'https://api.fish.audio/v1/tts',
        json={
            'text': text,
            'model': 'fish-audio-s2-pro',
            'with_timing': True  # 启用时间戳
        },
        headers={'Authorization': f'Bearer {api_key}'}
    )
    
    result = response.json()
    return {
        'audio': result['audio_base64'],
        'timestamps': result['word_timings']  # [{word, start_ms, end_ms}, ...]
    }

# 返回的timestamps格式
# [
#   {'word': '这', 'start': 0, 'end': 120},
#   {'word': '是', 'start': 120, 'end': 240},
#   {'word': '一', 'start': 240, 'end': 340},
#   ...
# ]

def generate_srt_from_timestamps(timestamps, video_duration_ms):
    """
    将word-level时间戳转换为SRT字幕
    """
    srt_content = ""
    idx = 1
    
    # 每行显示3-5个词
    words_per_line = 4
    i = 0
    
    while i < len(timestamps):
        start_time = timestamps[i]['start'] / 1000
        end_time = timestamps[min(i + words_per_line - 1, len(timestamps) - 1)]['end'] / 1000
        
        words = [t['word'] for t in timestamps[i:i+words_per_line]]
        text = ''.join(words)
        
        srt_content += f"{idx}\n"
        srt_content += f"{format_time(start_time)} --> {format_time(end_time)}\n"
        srt_content += f"{text}\n\n"
        
        idx += 1
        i += words_per_line
    
    return srt_content

def format_time(seconds):
    """SRT时间格式：00:00:00,000"""
    hours = int(seconds // 3600)
    minutes = int((seconds % 3600) // 60)
    secs = int(seconds % 60)
    millis = int((seconds % 1) * 1000)
    return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}"

# 使用示例
# timestamps = get_tts_with_timestamps("这是一个科普视频的介绍", api_key)['timestamps']
# srt = generate_srt_from_timestamps(timestamps, video_duration_ms=30000)
# with open('subtitles.srt', 'w', encoding='utf-8') as f:
#     f.write(srt)

def volcengine_tts_with_timestamps(text, api_key, voice_id):
    """
    火山引擎TTS支持segment-level和word-level时间戳
    """
    import json
    import websocket
    
    ws = websocket.create_connection('wss://tts.volcengineapi.com/api/v1/tts/ws')
    
    req = {
        "app_id": "your_app_id",
        "uid": "user_123",
        "text": text,
        "voice_id": voice_id,
        "req_id": "req_001",
        "return_timing": True,  # 返回时间戳
        "format": "pcm"
    }
    
    ws.send(json.dumps(req))
    
    audio_data = b""
    timing_data = []
    
    while True:
        data = ws.recv()
        resp = json.loads(data)
        
        if 'audio' in resp:
            # base64解码音频
            import base64
            audio_data += base64.b64decode(resp['audio'])
        
        if 'timing' in resp:
            # 提取词级时间戳
            timing_data.extend(resp['timing'])
        
        if resp.get('finish'):
            break
    
    ws.close()
    return audio_data, timing_data

# 安装WhisperX
pip install git+https://github.com/m-bain/whisperX.git

# 或用Conda
conda install -c conda-forge -c m-bain whisperx

import whisperx
import torch

def align_tts_with_whisperx(audio_file, text_file):
    """
    使用WhisperX对齐TTS生成的音频与原文本
    精度：±50ms
    """
    device = "cuda" if torch.cuda.is_available() else "cpu"
    
    # 加载WhisperX模型
    model = whisperx.load_model("large-v3", device=device)
    
    # 1. 转录音频
    audio = whisperx.load_audio(audio_file)
    result = model.transcribe(audio, batch_size=16, language="zh")
    
    # 2. 加载预训练对齐模型
    model_a, metadata = whisperx.load_align_model(language_code="zh", device=device)
    
    # 3. 强制对齐
    result = whisperx.align(result["segments"], model_a, metadata, audio, device)
    
    # 4. 提取word-level时间戳
    timestamps = []
    for seg in result["segments"]:
        for word_info in seg.get("words", []):
            timestamps.append({
                'word': word_info['word'],
                'start': word_info['start'],
                'end': word_info['end']
            })
    
    return timestamps

# 使用
# timestamps = align_tts_with_whisperx("audio.wav", "script.txt")
# srt = generate_srt_from_timestamps(timestamps, duration)

# 安装
conda install -c conda-forge montreal-forced-aligner

# 对齐
mfa align audio.wav transcript.txt mandarin_mfa output_directory

# 输出TextGrid格式，包含phoneme-level精度

from aeneas.executetask import ExecuteTask
from aeneas.task import Task

def align_with_aeneas(audio_file, text_file, output_srt):
    """
    使用Aeneas进行强制对齐
    算法：DTW（Dynamic Time Warping）
    """
    # 创建task
    task = Task()
    task.audio_file_path_absolute = audio_file
    task.text_file_path_absolute = text_file
    task.output_file_path_absolute = output_srt
    task.task_type = Task.TASK_SYNC_MAP_ON_WORDS
    task.language = "zho"  # 中文
    
    # 执行对齐
    executor = ExecuteTask(task)
    executor.execute()
    
    print(f"✓ 对齐完成，字幕已保存到 {output_srt}")

# 使用
# align_with_aeneas("audio.wav", "script.txt", "output.srt")

TTS主服务: Azure TTS
  - 500万字符/月免费额度
  - 超出后 $16/百万字符（或0.10元/千字）
  - 中文自然度8.5/10
  - SSML完整支持

本地克隆: GPT-SoVITS
  - 本地部署，无成本
  - RTX 3050 8GB显存即可
  - 中英混合，表现力强

字幕对齐: TTS返回时间戳
  - Azure WebAPI包含segment-level时间戳
  - 拼接Aeneas处理（无GPU需求）

音频处理: FFmpeg + Python
  - 开源免费
  - ffmpeg loudnorm -16LUFS标准化

总成本: $0-5/月（仅在超出免费额度时付费）

# 1. 文本分句
python preprocess.py --input script.txt --max_length 400

# 2. 调用Azure TTS（免费额度范围内）
python azure_tts_batch.py --segments segments.json --output audio_chunks/

# 3. 本地克隆自己声音（可选）
# 使用预设Azure音色，或在云端克隆后用OpenAI TTS推断成本

# 4. 拼接和归一化
ffmpeg -i input_combined.wav -af "loudnorm=I=-16" output_normalized.mp3

# 5. 生成字幕（Aeneas）
python aeneas_align.py --audio output_normalized.mp3 --text script.txt

TTS主服务: Fish Audio S2 Pro
  - 中文盲测最强（65.7%胜率 vs ElevenLabs）
  - $15/百万字符
  - word-level时间戳原生支持
  - 零样本声音克隆（15秒）

补充方案: 火山引擎豆包TTS 2.0
  - 备选，国内节点快
  - 9.2/10自然度评分
  - 指令式情感标记
  - 超出Fish成本时平衡使用

克隆方案: Fish Audio原生克隆
  - 15秒样本即可
  - 相似度95%+

字幕对齐: Fish Audio时间戳 + WhisperX微调
  - 原生时间戳±80ms
  - 需要时用WhisperX精细到±50ms

音频处理: FFmpeg + Audacity人工调整
  - 关键字加重
  - 背景音乐混音
  - 情感强调

总成本: $100-200/月（取决于产出量）

假设月产出200分钟视频：
- 平均脚本400万字/月
- Fish Audio: 400万 × $15/百万 = $60
- 预留克隆/实验: $20
- 字幕/后期工具: $20
- 总计: $100/月

Plus版本（高产）：
- 年产1000+分钟：$150-200/月

# Fish Audio批量API
import asyncio
import httpx

async def batch_tts_fish_audio(segments, api_key):
    """
    Fish Audio高效批处理
    并发8个请求，避免流控
    """
    async with httpx.AsyncClient() as client:
        semaphore = asyncio.Semaphore(8)
        
        tasks = [
            call_fish_tts(client, seg, api_key, semaphore)
            for seg in segments
        ]
        results = await asyncio.gather(*tasks)
        return results

async def call_fish_tts(client, text, api_key, semaphore):
    async with semaphore:
        response = await client.post(
            'https://api.fish.audio/v1/tts',
            json={
                'text': text,
                'model': 'fish-audio-s2-pro',
                'voice_id': 'default_cn',  # 克隆声音ID
                'with_timing': True,
                'mp3_bitrate': '192k'
            },
            headers={'Authorization': f'Bearer {api_key}'}
        )
        return response.json()

# 运行
# asyncio.run(batch_tts_fish_audio(segments, api_key))

GPU: RTX 4070 Super          ¥5,800
Nvme SSD 4TB                  ¥2,500
CPU: Ryzen 7 7700X           ¥1,800
内存: 64GB DDR5               ¥1,200
电源/散热/机箱等              ¥2,000
────────────────────────────
总硬件投入                    ¥13,300 ≈ $1,900

摊销到12个月: $158/月
摊销到24个月: $79/月

TTS引擎: IndexTTS2 (B站开源，最新)
  - 8GB显存要求 → RTX 4070满足
  - WER最优（开源方案中）
  - 推理速度RTF 0.12-0.18（超快）
  - 同/跨语种克隆内置
  - ComfyUI集成可用

备选: CosyVoice2 (阿里，完全开源)
  - 中文自然度最高（官方评测第一）
  - 支持方言
  - 模型包含，无需额外下载

克隆方案: GPT-SoVITS本地微调
  - 用IndexTTS2的克隆功能
  - 或用GPT-SoVITS在RTX 4070上微调（5分钟）

字幕对齐: WhisperX本地部署
  - 精度±50ms
  - GPU加速快（15分钟音频<30秒）

音频处理: FFmpeg + Librosa
  - loudnorm归一化
  - 全本地处理，无网络

总成本: $0/月运营（仅电费+维保）
        摊销后 $79-158/月

# Dockerfile for local TTS stack
FROM nvidia/cuda:12.4-devel-ubuntu22.04

# IndexTTS2
RUN git clone https://github.com/index-tts/index-tts /app/index-tts
WORKDIR /app/index-tts
RUN pip install -r requirements.txt
# 预加载模型（避免每次启动下载）
RUN python -c "from index_tts import load_model; load_model('index_tts2')"

# GPT-SoVITS补充
RUN git clone https://github.com/RVC-Boss/GPT-SoVITS /app/gpt-sovits
RUN cd /app/gpt-sovits && pip install -r requirements.txt

# WhisperX
RUN pip install git+https://github.com/m-bain/whisperX.git

# API Server
COPY tts_server.py /app/
EXPOSE 8000

CMD ["uvicorn", "tts_server:app", "--host", "0.0.0.0", "--port", "8000"]

# tts_server.py
from fastapi import FastAPI, File, UploadFile
from index_tts import TTSModel
import asyncio

app = FastAPI()
tts_model = TTSModel()

@app.post("/tts")
async def synthesize(text: str, voice_id: str = "default"):
    """本地TTS API"""
    audio = await asyncio.to_thread(
        tts_model.synthesize,
        text,
        voice_id
    )
    return {"audio_base64": audio}

@app.post("/align")
async def align_subtitles(file: UploadFile, text: str):
    """本地字幕对齐"""
    # 用WhisperX对齐
    ...

# 启动
# uvicorn tts_server:app --reload

方案1（云端每月$50）：
- 24个月 × $50 = $1,200

方案3（硬件投入摊销）：
- 硬件投入 + 电费 ≈ $1,100
- 机制：一样成本，但获得完全自主可控

#!/usr/bin/env python3
"""
长视频TTS配音 + 字幕生成完整流程
环境：Python 3.9+，FFmpeg，Fish Audio API
"""

import asyncio
import json
import os
import re
from pathlib import Path
from typing import List, Dict, Tuple

import httpx
from pydub import AudioSegment
import subprocess


class TTS_Config:
    """配置"""
    FISH_API_KEY = os.getenv("FISH_API_KEY", "sk_xxx")
    FISH_API_URL = "https://api.fish.audio/v1"
    OUTPUT_DIR = "output"
    MAX_SEGMENT_LENGTH = 400  # 中文字符
    TTS_CONCURRENT = 8


class TextProcessor:
    """文本预处理"""
    
    @staticmethod
    def split_sentences(text: str, max_length: int = 400) -> List[str]:
        """按自然段分句"""
        sentences = re.split(r'(?<=[。？！])', text.strip())
        
        segments = []
        current = ""
        
        for sent in sentences:
            if len(current) + len(sent) <= max_length:
                current += sent
            else:
                if current:
                    segments.append(current)
                current = sent
        
        if current:
            segments.append(current)
        
        return segments
    
    @staticmethod
    def preprocess_for_tts(text: str) -> str:
        """清理文本（去URL、特殊符号等）"""
        # 去URL
        text = re.sub(r'https?://\S+', '', text)
        # AI → A I（英文缩写分开）
        text = re.sub(r'(?<![A-Z])[A-Z]{2,}(?![a-z])', lambda m: ' '.join(m.group()), text)
        return text.strip()


class FishAudioTTS:
    """Fish Audio TTS调用"""
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.fish.audio/v1"
    
    async def synthesize_batch(self, segments: List[str]) -> List[Tuple[bytes, List[Dict]]]:
        """批量合成并返回音频 + 时间戳"""
        async with httpx.AsyncClient() as client:
            semaphore = asyncio.Semaphore(TTS_Config.TTS_CONCURRENT)
            
            tasks = [
                self._call_tts(client, seg, idx, semaphore)
                for idx, seg in enumerate(segments)
            ]
            
            results = await asyncio.gather(*tasks)
            return sorted(results, key=lambda x: x[2])  # 按原始顺序排序
    
    async def _call_tts(self, client, text: str, idx: int, semaphore):
        """调用单个TTS请求"""
        async with semaphore:
            try:
                response = await client.post(
                    f"{self.base_url}/tts",
                    json={
                        "text": text,
                        "model": "fish-audio-s2-pro",
                        "voice_id": "default_cn",
                        "format": "wav",
                        "with_timing": True,
                    },
                    headers={"Authorization": f"Bearer {self.api_key}"},
                    timeout=30.0
                )
                
                data = response.json()
                audio_b64 = data.get("audio")
                timestamps = data.get("timestamps", [])
                
                # 解码音频
                import base64
                audio_bytes = base64.b64decode(audio_b64)
                
                return audio_bytes, timestamps, idx
            
            except Exception as e:
                print(f"✗ 第 {idx} 段合成失败: {e}")
                return b"", [], idx


class AudioProcessor:
    """音频处理"""
    
    @staticmethod
    def concat_chunks(chunk_list: List[bytes], pause_ms: int = 800) -> bytes:
        """拼接音频块"""
        combined = AudioSegment.empty()
        silence = AudioSegment.silent(duration=pause_ms)
        
        for chunk in chunk_list:
            audio = AudioSegment.from_wav(io.BytesIO(chunk))
            combined += audio + silence
        
        return combined.export(format="wav").read()
    
    @staticmethod
    def normalize_loudness(audio_file: str, output_file: str, target_lufs: float = -16):
        """FFmpeg响度归一化"""
        cmd = [
            'ffmpeg', '-i', audio_file,
            '-af', f'loudnorm=I={target_lufs}:TP=-1.5:LRA=11',
            '-y', output_file
        ]
        subprocess.run(cmd, check=True, capture_output=True)
        print(f"✓ 已归一化到 {target_lufs} LUFS")


class SubtitleGenerator:
    """字幕生成"""
    
    @staticmethod
    def timestamps_to_srt(all_timestamps: List[List[Dict]], output_file: str):
        """将时间戳序列转换为SRT"""
        srt = ""
        idx = 1
        
        for segment_ts in all_timestamps:
            for ts in segment_ts:
                if 'start' in ts and 'end' in ts:
                    start_sec = ts['start'] / 1000
                    end_sec = ts['end'] / 1000
                    word = ts.get('word', '')
                    
                    srt += f"{idx}\n"
                    srt += f"{SubtitleGenerator._format_time(start_sec)} --> "
                    srt += f"{SubtitleGenerator._format_time(end_sec)}\n"
                    srt += f"{word}\n\n"
                    
                    idx += 1
        
        with open(output_file, 'w', encoding='utf-8') as f:
            f.write(srt)
        
        print(f"✓ 字幕已生成: {output_file}")
    
    @staticmethod
    def _format_time(seconds: float) -> str:
        """SRT时间格式"""
        h = int(seconds // 3600)
        m = int((seconds % 3600) // 60)
        s = int(seconds % 60)
        ms = int((seconds % 1) * 1000)
        return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"


async def main():
    """主流程"""
    
    # 1. 读取脚本
    script_file = "script.txt"
    with open(script_file, 'r', encoding='utf-8') as f:
        script = f.read()
    
    print(f"📖 脚本长度: {len(script)} 字")
    
    # 2. 文本预处理和分句
    script = TextProcessor.preprocess_for_tts(script)
    segments = TextProcessor.split_sentences(script, TTS_Config.MAX_SEGMENT_LENGTH)
    print(f"✓ 分成 {len(segments)} 段")
    
    # 3. TTS合成
    print("🎤 正在合成配音...")
    tts = FishAudioTTS(TTS_Config.FISH_API_KEY)
    audio_chunks, timestamp_list, _ = zip(*await tts.synthesize_batch(segments))
    
    print(f"✓ 合成完毕，共 {len(audio_chunks)} 段音频")
    
    # 4. 拼接音频
    print("🔗 正在拼接音频...")
    combined_audio = AudioProcessor.concat_chunks(list(audio_chunks))
    
    temp_wav = os.path.join(TTS_Config.OUTPUT_DIR, "temp_combined.wav")
    os.makedirs(TTS_Config.OUTPUT_DIR, exist_ok=True)
    
    with open(temp_wav, 'wb') as f:
        f.write(combined_audio)
    
    # 5. 响度归一化
    print("📊 正在归一化音量...")
    output_wav = os.path.join(TTS_Config.OUTPUT_DIR, "final_audio.wav")
    AudioProcessor.normalize_loudness(temp_wav, output_wav)
    
    # 6. 生成字幕
    print("📝 正在生成字幕...")
    srt_file = os.path.join(TTS_Config.OUTPUT_DIR, "subtitles.srt")
    SubtitleGenerator.timestamps_to_srt(timestamp_list, srt_file)
    
    print("\n✨ 完成！")
    print(f"  音频: {output_wav}")
    print(f"  字幕: {srt_file}")


if __name__ == "__main__":
    asyncio.run(main())

# 1. 设置API密钥
export FISH_API_KEY="sk_xxx"

# 2. 准备script.txt（UTF-8编码）

# 3. 运行
python tts_workflow.py

# 输出文件在output/目录下