最佳实践 #

┌─────────────────────────────────────────────────────────────┐
│                    最佳实践概览                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │  性能优化    │  │  成本控制    │  │  错误处理    │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
│                                                             │
│  ┌─────────────┐  ┌─────────────┐  ┌─────────────┐         │
│  │  安全建议    │  │  文本处理    │  │  语音选择    │         │
│  └─────────────┘  └─────────────┘  └─────────────┘         │
│                                                             │
└─────────────────────────────────────────────────────────────┘

┌─────────────────────────────────────────────────────────────┐
│                    模型选择策略                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  高质量场景：                                                │
│  ├── 有声书、播客                                           │
│  ├── 专业配音                                               │
│  └── 使用 eleven_multilingual_v2                            │
│                                                             │
│  低延迟场景：                                                │
│  ├── 实时对话                                               │
│  ├── 交互式应用                                             │
│  └── 使用 eleven_turbo_v2_5                                 │
│                                                             │
│  英语专用：                                                  │
│  ├── 英语内容                                               │
│  ├── 最高质量要求                                           │
│  └── 使用 eleven_monolingual_v1                             │
│                                                             │
└─────────────────────────────────────────────────────────────┘

# 推荐：使用流式处理长文本
def generate_long_text(client, text, voice_id, output_path):
    audio_stream = client.text_to_speech.convert_as_stream(
        text=text,
        voice_id=voice_id,
        model_id="eleven_multilingual_v2"
    )
    
    with open(output_path, "wb") as f:
        for chunk in audio_stream:
            f.write(chunk)

# 不推荐：等待完整响应
def generate_blocking(client, text, voice_id, output_path):
    audio = client.text_to_speech.convert(
        text=text,
        voice_id=voice_id,
        model_id="eleven_multilingual_v2"
    )
    
    with open(output_path, "wb") as f:
        for chunk in audio:
            f.write(chunk)

import asyncio
from elevenlabs import AsyncElevenLabs

class AudioGenerator:
    def __init__(self, api_key, max_concurrent=5):
        self.client = AsyncElevenLabs(api_key=api_key)
        self.semaphore = asyncio.Semaphore(max_concurrent)
    
    async def generate(self, text, voice_id, output_path):
        async with self.semaphore:
            audio = await self.client.text_to_speech.convert(
                text=text,
                voice_id=voice_id,
                model_id="eleven_multilingual_v2"
            )
            
            chunks = []
            async for chunk in audio:
                chunks.append(chunk)
            
            return b"".join(chunks)
    
    async def batch_generate(self, items):
        tasks = [
            self.generate(item["text"], item["voice_id"], item["output"])
            for item in items
        ]
        return await asyncio.gather(*tasks)

import hashlib
import os
import json

class AudioCache:
    def __init__(self, cache_dir="cache"):
        self.cache_dir = cache_dir
        os.makedirs(cache_dir, exist_ok=True)
    
    def _get_cache_key(self, text, voice_id, settings):
        content = f"{text}:{voice_id}:{json.dumps(settings, sort_keys=True)}"
        return hashlib.md5(content.encode()).hexdigest()
    
    def get(self, text, voice_id, settings=None):
        cache_key = self._get_cache_key(text, voice_id, settings or {})
        cache_path = os.path.join(self.cache_dir, f"{cache_key}.mp3")
        
        if os.path.exists(cache_path):
            with open(cache_path, "rb") as f:
                return f.read()
        
        return None
    
    def set(self, text, voice_id, audio_data, settings=None):
        cache_key = self._get_cache_key(text, voice_id, settings or {})
        cache_path = os.path.join(self.cache_dir, f"{cache_key}.mp3")
        
        with open(cache_path, "wb") as f:
            f.write(audio_data)

# 使用示例
cache = AudioCache()

def generate_with_cache(client, text, voice_id):
    cached = cache.get(text, voice_id)
    if cached:
        return cached
    
    audio = client.text_to_speech.convert(
        text=text,
        voice_id=voice_id,
        model_id="eleven_multilingual_v2"
    )
    
    audio_data = b"".join(audio)
    cache.set(text, voice_id, audio_data)
    
    return audio_data

def count_characters(text):
    return len(text)

def estimate_cost(characters, model="multilingual"):
    # 大致估算（实际价格请参考官方定价）
    if model == "multilingual":
        cost_per_1k = 0.30  # 示例价格
    else:
        cost_per_1k = 0.20
    
    return (characters / 1000) * cost_per_1k

# 使用示例
text = "这是一段需要转换的文本。"
chars = count_characters(text)
cost = estimate_cost(chars)
print(f"Characters: {chars}, Estimated cost: ${cost:.4f}")

class BudgetManager:
    def __init__(self, monthly_budget, warning_threshold=0.8):
        self.monthly_budget = monthly_budget
        self.warning_threshold = warning_threshold
        self.current_usage = 0
    
    def can_generate(self, characters):
        estimated_cost = estimate_cost(characters)
        return (self.current_usage + estimated_cost) <= self.monthly_budget
    
    def record_usage(self, characters):
        cost = estimate_cost(characters)
        self.current_usage += cost
        
        if self.current_usage >= self.monthly_budget * self.warning_threshold:
            self._send_warning()
    
    def _send_warning(self):
        print(f"Warning: Usage at {self.current_usage/self.monthly_budget*100:.1f}%")

# 使用示例
budget = BudgetManager(monthly_budget=100)

def generate_with_budget(client, text, voice_id):
    chars = count_characters(text)
    
    if not budget.can_generate(chars):
        raise Exception("Budget exceeded")
    
    audio = client.text_to_speech.convert(
        text=text,
        voice_id=voice_id,
        model_id="eleven_multilingual_v2"
    )
    
    budget.record_usage(chars)
    return audio

def optimize_text(text):
    optimizations = []
    
    # 移除多余空白
    original_len = len(text)
    text = " ".join(text.split())
    if len(text) < original_len:
        optimizations.append("Removed extra whitespace")
    
    # 移除重复标点
    import re
    text = re.sub(r'([.!?])\1+', r'\1', text)
    
    return text, optimizations

# 使用示例
original = "这是   一段  文本。。。"
optimized, changes = optimize_text(original)
print(f"Original: {len(original)} chars")
print(f"Optimized: {len(optimized)} chars")
print(f"Changes: {changes}")

import time
from functools import wraps

def retry(max_attempts=3, delay=1, backoff=2):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            attempts = 0
            current_delay = delay
            
            while attempts < max_attempts:
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    attempts += 1
                    if attempts >= max_attempts:
                        raise e
                    
                    print(f"Attempt {attempts} failed: {e}")
                    time.sleep(current_delay)
                    current_delay *= backoff
            
        return wrapper
    return decorator

# 使用示例
@retry(max_attempts=3, delay=1)
def generate_audio(client, text, voice_id):
    return client.text_to_speech.convert(
        text=text,
        voice_id=voice_id,
        model_id="eleven_multilingual_v2"
    )

from elevenlabs import APIError, RateLimitError, AuthenticationError

def handle_api_error(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except AuthenticationError as e:
            print(f"Authentication failed: {e}")
            raise
        except RateLimitError as e:
            print(f"Rate limit exceeded: {e}")
            time.sleep(60)
            return wrapper(*args, **kwargs)
        except APIError as e:
            print(f"API error: {e}")
            if "content_violation" in str(e):
                print("Content policy violation detected")
            raise
        except Exception as e:
            print(f"Unexpected error: {e}")
            raise
    
    return wrapper

┌─────────────────────────────────────────────────────────────┐
│                    API Key 安全                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  ✅ 推荐：                                                   │
│  ├── 使用环境变量存储 API Key                               │
│  ├── 使用密钥管理服务                                       │
│  ├── 定期轮换 API Key                                       │
│  ├── 使用最小权限原则                                       │
│  └── 监控 API Key 使用情况                                  │
│                                                             │
│  ❌ 避免：                                                   │
│  ├── 硬编码 API Key                                         │
│  ├── 提交到版本控制                                         │
│  ├── 在日志中打印 API Key                                   │
│  ├── 在前端暴露 API Key                                     │
│  └── 共享 API Key                                           │
│                                                             │
└─────────────────────────────────────────────────────────────┘

import os
from dotenv import load_dotenv

load_dotenv()

API_KEY = os.getenv("ELEVENLABS_API_KEY")
if not API_KEY:
    raise ValueError("ELEVENLABS_API_KEY not set")

# 后端代理示例 (Flask)
from flask import Flask, request, jsonify
from elevenlabs import ElevenLabs

app = Flask(__name__)
client = ElevenLabs(api_key=os.getenv("ELEVENLABS_API_KEY"))

@app.route("/api/generate", methods=["POST"])
def generate():
    data = request.json
    text = data.get("text")
    voice_id = data.get("voice_id")
    
    if not text or not voice_id:
        return jsonify({"error": "Missing parameters"}), 400
    
    try:
        audio = client.text_to_speech.convert(
            text=text,
            voice_id=voice_id,
            model_id="eleven_multilingual_v2"
        )
        
        audio_data = b"".join(audio)
        return audio_data, 200, {"Content-Type": "audio/mpeg"}
        
    except Exception as e:
        return jsonify({"error": str(e)}), 500

def preprocess_text(text):
    # 标准化引号
    text = text.replace('"', '"').replace('"', '"')
    text = text.replace(''', "'").replace(''', "'")
    
    # 处理省略号
    text = text.replace('...', '…')
    
    # 处理数字
    import re
    text = re.sub(r'\b(\d+)\b', lambda m: number_to_words(int(m.group(1))), text)
    
    return text

def number_to_words(n):
    # 简单的数字转文字
    ones = ['zero', 'one', 'two', 'three', 'four', 'five', 
            'six', 'seven', 'eight', 'nine']
    if n < 10:
        return ones[n]
    return str(n)

def split_text(text, max_length=5000):
    sentences = text.replace('。', '。\n').replace('!', '!\n').replace('?', '?\n').split('\n')
    
    chunks = []
    current_chunk = ""
    
    for sentence in sentences:
        if len(current_chunk) + len(sentence) > max_length:
            if current_chunk:
                chunks.append(current_chunk.strip())
            current_chunk = sentence
        else:
            current_chunk += sentence
    
    if current_chunk:
        chunks.append(current_chunk.strip())
    
    return chunks

def select_voice(content_type, language="en"):
    voice_mapping = {
        ("audiobook", "en"): "JBFqnCBsd6RMkjVDRZzb",  # Rachel
        ("audiobook", "zh"): "chinese_voice_id",
        ("commercial", "en"): "AZnzlk1XvdvUeBn1ldMn",  # Domi
        ("tutorial", "en"): "ErXwLH5i43ZdrnRw6Rgd",  # Antoni
        ("game", "en"): "MF3mGyEYCl7XYWbV9V6O",  # Elli
    }
    
    return voice_mapping.get((content_type, language), "JBFqnCBsd6RMkjVDRZzb")

def test_voice_quality(client, voice_id, test_cases):
    results = []
    
    for case in test_cases:
        audio = client.text_to_speech.convert(
            text=case["text"],
            voice_id=voice_id,
            model_id="eleven_multilingual_v2"
        )
        
        audio_data = b"".join(audio)
        results.append({
            "name": case["name"],
            "text": case["text"],
            "size": len(audio_data)
        })
    
    return results

import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("elevenlabs")

class UsageMonitor:
    def __init__(self):
        self.requests = []
    
    def log_request(self, text, voice_id, duration, success):
        self.requests.append({
            "timestamp": datetime.now().isoformat(),
            "text_length": len(text),
            "voice_id": voice_id,
            "duration": duration,
            "success": success
        })
        
        logger.info(f"Request: {len(text)} chars, {voice_id}, {duration:.2f}s")
    
    def get_stats(self):
        if not self.requests:
            return {}
        
        total = len(self.requests)
        successful = sum(1 for r in self.requests if r["success"])
        avg_duration = sum(r["duration"] for r in self.requests) / total
        
        return {
            "total_requests": total,
            "success_rate": successful / total,
            "avg_duration": avg_duration
        }