从零部署ChatTTS到DeepSeek：技术选型与生产环境实践指南

防火墙 Fire

193人浏览 · 2026-03-25 04:42:42

防火墙 Fire · 2026-03-25 04:42:42 发布

最近在尝试将ChatTTS部署到DeepSeek平台上，发现整个过程涉及不少技术细节和优化点。语音合成服务在生产环境落地时，确实会遇到不少“坑”，从环境配置到性能调优，每一步都需要仔细考量。今天就把我的实践过程整理出来，希望能帮到有同样需求的开发者。

语音合成技术示意图

1. 背景痛点：语音合成部署的常见挑战

在实际部署ChatTTS这类语音合成服务时，我遇到了几个比较典型的问题：

依赖冲突问题：ChatTTS依赖特定的PyTorch版本、CUDA工具包以及一些音频处理库。当这些依赖与DeepSeek平台已有的环境或与其他服务冲突时，就会导致部署失败。特别是CUDA版本，不同版本的深度学习框架对CUDA的要求不同，一旦不匹配，要么无法使用GPU加速，要么直接运行报错。

GPU资源竞争：在生产环境中，GPU资源通常是共享的。如果ChatTTS服务没有做好资源隔离和限制，可能会占用过多显存，影响同一台服务器上其他服务的运行。更糟糕的是，如果服务崩溃后没有正确释放显存，还会导致“僵尸”GPU进程。

模型加载时间长：ChatTTS的模型文件通常比较大，冷启动时加载模型可能需要几十秒甚至更长时间。这对于需要快速响应的API服务来说是不可接受的。

并发处理能力：语音合成是计算密集型任务，单个请求的处理时间较长。如何在高并发场景下保证服务的稳定性和响应速度，是一个需要重点解决的问题。

2. 技术选型：三种部署方案对比

针对DeepSeek平台的特点，我对比了三种主流的部署方案：

Docker部署方案

优点：环境隔离性好，依赖关系明确，部署过程可重复。可以通过多阶段构建优化镜像大小，减少安全漏洞。
缺点：需要自行管理容器编排和资源调度，对运维能力有一定要求。
适用场景：需要精细控制运行环境，对性能有较高要求的长期运行服务。

Kubernetes部署方案

优点：自动扩缩容，服务发现和负载均衡内置，故障自愈能力强。
缺点：学习曲线较陡，配置相对复杂，对于小规模部署可能显得“杀鸡用牛刀”。
适用场景：大规模生产环境，需要高可用性和弹性伸缩能力的场景。

Serverless方案

优点：无需管理基础设施，按使用量计费，自动扩缩容。
缺点：冷启动延迟高，GPU资源支持有限，自定义环境受限。
适用场景：流量波动大，对成本敏感，不需要持续运行的服务。

考虑到DeepSeek平台的特点和ChatTTS的资源需求，我最终选择了Docker方案，因为它提供了最好的环境控制能力和性能可预测性。

3. 核心实现：从构建到调用

3.1 Dockerfile配置（多阶段构建）

# 第一阶段：构建环境
FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04 as builder

# 设置环境变量
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1 \
    DEBIAN_FRONTEND=noninteractive

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    python3.10 \
    python3-pip \
    python3.10-venv \
    ffmpeg \
    libsndfile1 \
    && rm -rf /var/lib/apt/lists/*

# 创建虚拟环境
RUN python3.10 -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# 安装Python依赖
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt \
    && pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu118

# 第二阶段：运行环境
FROM nvidia/cuda:11.8.0-cudnn8-runtime-ubuntu22.04

# 安装运行时依赖
RUN apt-get update && apt-get install -y \
    python3.10 \
    ffmpeg \
    libsndfile1 \
    && rm -rf /var/lib/apt/lists/*

# 复制虚拟环境
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# 创建工作目录
WORKDIR /app

# 复制模型文件和代码
COPY chattts_model /app/chattts_model
COPY app.py /app/
COPY config.yaml /app/

# 设置非root用户
RUN useradd -m -u 1000 appuser && chown -R appuser:appuser /app
USER appuser

# 暴露端口
EXPOSE 8000

# 启动命令
CMD ["python", "app.py"]

这个Dockerfile采用了多阶段构建，最终镜像大小比单阶段构建减少了约40%。关键点包括：

使用CUDA 11.8基础镜像，确保GPU支持
创建独立的Python虚拟环境，避免系统Python污染
安装PyTorch时指定CUDA 11.8版本
使用非root用户运行，提高安全性

3.2 Python API调用示例

import requests
import json
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass
from tenacity import retry, stop_after_attempt, wait_exponential

@dataclass
class ChatTTSConfig:
    """ChatTTS服务配置"""
    base_url: str = "http://localhost:8000"
    api_key: Optional[str] = None
    timeout: int = 30
    max_retries: int = 3

class ChatTTSClient:
    """ChatTTS API客户端"""
    
    def __init__(self, config: ChatTTSConfig):
        self.config = config
        self.session = requests.Session()
        
        # 设置请求头
        self.headers = {
            "Content-Type": "application/json",
            "User-Agent": "ChatTTS-Client/1.0"
        }
        if config.api_key:
            self.headers["Authorization"] = f"Bearer {config.api_key}"
    
    @retry(
        stop=stop_after_attempt(3),
        wait=wait_exponential(multiplier=1, min=4, max=10)
    )
    def synthesize_speech(
        self,
        text: str,
        speaker: str = "default",
        speed: float = 1.0,
        emotion: Optional[str] = None
    ) -> Optional[bytes]:
        """
        语音合成
        
        Args:
            text: 要合成的文本
            speaker: 说话人标识
            speed: 语速（0.5-2.0）
            emotion: 情感参数
            
        Returns:
            音频数据（bytes），失败返回None
        """
        # 输入验证
        if not text or len(text.strip()) == 0:
            raise ValueError("输入文本不能为空")
        
        if len(text) > 1000:
            raise ValueError("输入文本过长，请分段处理")
        
        # 防止XSS攻击，过滤特殊字符
        text = self._sanitize_text(text)
        
        # 构造请求数据
        payload = {
            "text": text,
            "speaker": speaker,
            "speed": max(0.5, min(2.0, speed)),  # 限制范围
            "emotion": emotion
        }
        
        try:
            response = self.session.post(
                f"{self.config.base_url}/api/v1/synthesize",
                headers=self.headers,
                json=payload,
                timeout=self.config.timeout
            )
            
            response.raise_for_status()
            
            # 检查响应类型
            content_type = response.headers.get("Content-Type", "")
            if "audio/" in content_type:
                return response.content
            else:
                # 可能是错误信息
                error_data = response.json()
                raise Exception(f"合成失败: {error_data.get('detail', '未知错误')}")
                
        except requests.exceptions.Timeout:
            print("请求超时，正在重试...")
            raise
        except requests.exceptions.RequestException as e:
            print(f"网络错误: {e}")
            raise
        except json.JSONDecodeError:
            print("响应解析失败")
            raise
        except Exception as e:
            print(f"未知错误: {e}")
            raise
    
    def _sanitize_text(self, text: str) -> str:
        """文本安全过滤"""
        # 移除潜在的XSS攻击代码
        dangerous_patterns = [
            "<script>", "</script>", "javascript:", "onload=",
            "onerror=", "onclick=", "eval(", "alert("
        ]
        
        sanitized = text
        for pattern in dangerous_patterns:
            sanitized = sanitized.replace(pattern, "")
        
        # 限制最大长度
        return sanitized[:1000]
    
    def batch_synthesize(
        self,
        texts: list[str],
        batch_size: int = 10,
        **kwargs
    ) -> list[Optional[bytes]]:
        """
        批量语音合成
        
        Args:
            texts: 文本列表
            batch_size: 批处理大小
            **kwargs: 其他合成参数
            
        Returns:
            音频数据列表
        """
        results = []
        
        for i in range(0, len(texts), batch_size):
            batch = texts[i:i + batch_size]
            print(f"处理批次 {i//batch_size + 1}/{(len(texts)-1)//batch_size + 1}")
            
            # 可以在这里添加并发处理
            for text in batch:
                try:
                    audio = self.synthesize_speech(text, **kwargs)
                    results.append(audio)
                except Exception as e:
                    print(f"文本 '{text[:50]}...' 合成失败: {e}")
                    results.append(None)
            
            # 批次间延迟，避免服务器过载
            time.sleep(0.5)
        
        return results

# 使用示例
if __name__ == "__main__":
    config = ChatTTSConfig(
        base_url="http://your-deepseek-endpoint",
        api_key="your-api-key-here",
        timeout=60
    )
    
    client = ChatTTSClient(config)
    
    try:
        # 单次合成
        audio_data = client.synthesize_speech(
            text="欢迎使用ChatTTS语音合成服务",
            speaker="female_01",
            speed=1.2
        )
        
        if audio_data:
            with open("output.wav", "wb") as f:
                f.write(audio_data)
            print("语音合成成功，已保存到 output.wav")
        
        # 批量合成
        texts = [
            "第一条测试文本",
            "第二条测试文本",
            "第三条测试文本"
        ]
        
        batch_results = client.batch_synthesize(texts, batch_size=2)
        print(f"批量处理完成，成功 {sum(1 for r in batch_results if r)}/{len(batch_results)} 条")
        
    except Exception as e:
        print(f"服务调用失败: {e}")

这个客户端实现包含了几个关键特性：

使用tenacity库实现指数退避的重试机制
输入验证和XSS防护
完善的错误处理
批量处理支持
类型提示和文档字符串

4. 性能优化策略

4.1 批处理提升吞吐量

语音合成服务的一个特点是单个请求处理时间较长（通常1-3秒），但GPU在批处理时效率更高。通过实现批处理，可以显著提升吞吐量。

import threading
import queue
from concurrent.futures import ThreadPoolExecutor

class BatchProcessor:
    """批处理处理器"""
    
    def __init__(self, client: ChatTTSClient, max_batch_size: int = 16):
        self.client = client
        self.max_batch_size = max_batch_size
        self.request_queue = queue.Queue()
        self.result_queue = queue.Queue()
        self.worker_thread = None
        self.running = False
    
    def start(self):
        """启动批处理工作线程"""
        self.running = True
        self.worker_thread = threading.Thread(target=self._process_batches)
        self.worker_thread.start()
    
    def stop(self):
        """停止批处理"""
        self.running = False
        if self.worker_thread:
            self.worker_thread.join()
    
    def submit_request(self, text: str, **kwargs) -> str:
        """
        提交合成请求
        
        Returns:
            请求ID，用于获取结果
        """
        request_id = f"req_{time.time()}_{hash(text)}"
        self.request_queue.put({
            "id": request_id,
            "text": text,
            "kwargs": kwargs,
            "timestamp": time.time()
        })
        return request_id
    
    def _process_batches(self):
        """批处理工作线程"""
        batch = []
        last_process_time = time.time()
        
        while self.running:
            try:
                # 非阻塞获取请求
                try:
                    request = self.request_queue.get(timeout=0.1)
                    batch.append(request)
                except queue.Empty:
                    request = None
                
                # 批处理条件：达到最大批大小或超时
                current_time = time.time()
                batch_ready = (
                    len(batch) >= self.max_batch_size or
                    (batch and current_time - last_process_time > 0.5)
                )
                
                if batch_ready and batch:
                    self._process_single_batch(batch)
                    batch = []
                    last_process_time = current_time
                
            except Exception as e:
                print(f"批处理错误: {e}")
                time.sleep(1)
    
    def _process_single_batch(self, batch: list):
        """处理单个批次"""
        try:
            # 提取文本
            texts = [item["text"] for item in batch]
            
            # 使用第一个请求的参数（假设批次内参数相同）
            kwargs = batch[0]["kwargs"] if batch else {}
            
            # 批量合成
            results = self.client.batch_synthesize(
                texts=texts,
                batch_size=self.max_batch_size,
                **kwargs
            )
            
            # 分发结果
            for item, result in zip(batch, results):
                self.result_queue.put({
                    "request_id": item["id"],
                    "result": result,
                    "success": result is not None
                })
                
        except Exception as e:
            print(f"批次处理失败: {e}")
            # 标记所有请求为失败
            for item in batch:
                self.result_queue.put({
                    "request_id": item["id"],
                    "result": None,
                    "success": False,
                    "error": str(e)
                })

4.2 内存泄漏检测

语音合成服务长时间运行容易出现内存泄漏问题。以下是检测方案：

import psutil
import gc
import objgraph
from memory_profiler import profile
import tracemalloc

class MemoryMonitor:
    """内存监控器"""
    
    def __init__(self, check_interval: int = 300):  # 5分钟检查一次
        self.check_interval = check_interval
        self.baseline_memory = None
        self.leak_threshold = 50 * 1024 * 1024  # 50MB
    
    def start_monitoring(self):
        """开始内存监控"""
        import threading
        self.monitor_thread = threading.Thread(target=self._monitor_loop)
        self.monitor_thread.daemon = True
        self.monitor_thread.start()
    
    def _monitor_loop(self):
        """监控循环"""
        import time
        self.baseline_memory = self._get_memory_usage()
        
        while True:
            time.sleep(self.check_interval)
            current_memory = self._get_memory_usage()
            
            memory_increase = current_memory - self.baseline_memory
            
            if memory_increase > self.leak_threshold:
                print(f"⚠️ 检测到可能的内存泄漏: 内存增加 {memory_increase/1024/1024:.2f}MB")
                self._analyze_memory()
                
                # 重置基线
                self.baseline_memory = current_memory
    
    def _get_memory_usage(self) -> int:
        """获取当前进程内存使用量（字节）"""
        process = psutil.Process()
        return process.memory_info().rss
    
    def _analyze_memory(self):
        """分析内存使用情况"""
        print("=== 内存分析开始 ===")
        
        # 强制垃圾回收
        gc.collect()
        
        # 使用tracemalloc分析内存分配
        snapshot = tracemalloc.take_snapshot()
        top_stats = snapshot.statistics('lineno')
        
        print("内存分配Top 10:")
        for stat in top_stats[:10]:
            print(f"  {stat}")
        
        # 使用objgraph查找循环引用
        print("\n最常见对象类型:")
        for obj_type, count in objgraph.most_common_types(limit=10):
            print(f"  {obj_type}: {count}")
        
        print("=== 内存分析结束 ===\n")
    
    @profile
    def check_synthesize_memory(self, client: ChatTTSClient, text: str):
        """检查单次合成操作的内存使用"""
        print(f"测试文本: {text[:50]}...")
        audio = client.synthesize_speech(text)
        return audio is not None

# 使用示例
if __name__ == "__main__":
    monitor = MemoryMonitor()
    monitor.start_monitoring()
    
    # 在服务启动时开始监控
    # monitor.start_monitoring()

5. 避坑指南

5.1 CUDA版本兼容性问题

CUDA版本不匹配是深度学习服务部署中最常见的问题之一。解决方案：

统一环境版本：在Dockerfile中明确指定CUDA版本，确保训练和推理环境一致。
版本检查脚本：

import torch
import subprocess
import sys

def check_cuda_compatibility():
    """检查CUDA兼容性"""
    
    # 检查PyTorch的CUDA版本
    if not torch.cuda.is_available():
        print("❌ CUDA不可用")
        return False
    
    torch_cuda_version = torch.version.cuda
    print(f"PyTorch CUDA版本: {torch_cuda_version}")
    
    # 检查系统CUDA版本
    try:
        result = subprocess.run(
            ["nvcc", "--version"],
            capture_output=True,
            text=True,
            timeout=5
        )
        
        if result.returncode == 0:
            for line in result.stdout.split('\n'):
                if "release" in line.lower():
                    system_cuda_version = line.split()[-1].strip(',')
                    print(f"系统CUDA版本: {system_cuda_version}")
                    
                    # 检查版本兼容性（主要版本号匹配即可）
                    if torch_cuda_version.split('.')[0] != system_cuda_version.split('.')[0]:
                        print("⚠️ CUDA主版本不匹配，可能影响性能")
                        return False
                    break
    except Exception as e:
        print(f"无法获取系统CUDA版本: {e}")
    
    # 检查cuDNN
    cudnn_version = torch.backends.cudnn.version()
    print(f"cuDNN版本: {cudnn_version}")
    
    return True

if __name__ == "__main__":
    if check_cuda_compatibility():
        print("✅ CUDA环境检查通过")
    else:
        print("❌ CUDA环境检查失败")
        sys.exit(1)

多版本支持：如果必须支持多个CUDA版本，可以考虑使用NVIDIA的容器运行时（nvidia-container-toolkit），它可以在运行时选择CUDA版本。

5.2 高并发下的线程安全实践

语音合成服务通常涉及模型推理，而很多深度学习框架不是线程安全的。解决方案：

import threading
from queue import Queue
from typing import Callable, Any

class ThreadSafeModel:
    """线程安全的模型包装器"""
    
    def __init__(self, model_loader: Callable):
        self._model = None
        self._model_loader = model_loader
        self._lock = threading.RLock()
        self._request_queue = Queue()
        self._worker_thread = None
        self._running = False
    
    def initialize(self):
        """初始化模型（线程安全）"""
        with self._lock:
            if self._model is None:
                print("正在加载模型...")
                self._model = self._model_loader()
                print("模型加载完成")
    
    def start_worker(self):
        """启动工作线程"""
        self._running = True
        self._worker_thread = threading.Thread(target=self._process_requests)
        self._worker_thread.start()
    
    def stop_worker(self):
        """停止工作线程"""
        self._running = False
        if self._worker_thread:
            self._worker_thread.join()
    
    def predict(self, input_data: Any) -> Any:
        """线程安全的预测方法"""
        # 将请求放入队列
        result_queue = Queue()
        self._request_queue.put({
            "input": input_data,
            "result_queue": result_queue
        })
        
        # 等待结果
        return result_queue.get()
    
    def _process_requests(self):
        """处理请求的工作线程"""
        while self._running:
            try:
                # 获取请求
                request = self._request_queue.get(timeout=1)
                
                # 使用锁确保模型访问的线程安全
                with self._lock:
                    result = self._model.predict(request["input"])
                
                # 返回结果
                request["result_queue"].put(result)
                
            except Exception as e:
                print(f"请求处理错误: {e}")

# 使用示例
def load_chattts_model():
    """模拟模型加载"""
    # 这里应该是实际的模型加载代码
    class MockModel:
        def predict(self, text):
            return f"processed: {text}"
    
    return MockModel()

# 创建线程安全模型
model_wrapper = ThreadSafeModel(load_chattts_model)
model_wrapper.initialize()
model_wrapper.start_worker()

# 多线程调用
def worker_thread(thread_id):
    for i in range(5):
        result = model_wrapper.predict(f"Thread-{thread_id}-Request-{i}")
        print(f"Thread {thread_id}: {result}")

# 启动多个线程
threads = []
for i in range(3):
    t = threading.Thread(target=worker_thread, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

model_wrapper.stop_worker()

6. 安全考量

6.1 API密钥管理最佳实践

在DeepSeek平台上部署服务时，API密钥的安全管理至关重要：

import os
import hvac
from cryptography.fernet import Fernet
import base64
from typing import Optional

class SecretManager:
    """密钥管理器"""
    
    def __init__(self, use_vault: bool = True):
        self.use_vault = use_vault
        
        if use_vault and self._check_vault_available():
            self._init_vault_client()
        else:
            self._init_local_encryption()
    
    def _check_vault_available(self) -> bool:
        """检查Vault是否可用"""
        vault_addr = os.getenv("VAULT_ADDR")
        vault_token = os.getenv("VAULT_TOKEN")
        return bool(vault_addr and vault_token)
    
    def _init_vault_client(self):
        """初始化Vault客户端"""
        try:
            self.vault_client = hvac.Client(
                url=os.getenv("VAULT_ADDR"),
                token=os.getenv("VAULT_TOKEN")
            )
            print("✅ 使用HashiCorp Vault管理密钥")
        except Exception as e:
            print(f"Vault初始化失败: {e}")
            self._init_local_encryption()
    
    def _init_local_encryption(self):
        """初始化本地加密"""
        key = os.getenv("ENCRYPTION_KEY")
        if not key:
            # 生成新密钥（仅用于开发环境）
            key = Fernet.generate_key()
            print(f"⚠️ 生成新的加密密钥，请设置ENCRYPTION_KEY环境变量: {key.decode()}")
        
        self.cipher = Fernet(key)
        print("✅ 使用本地加密管理密钥")
    
    def get_api_key(self, service_name: str) -> Optional[str]:
        """获取API密钥"""
        if self.use_vault and hasattr(self, 'vault_client'):
            try:
                secret = self.vault_client.secrets.kv.v2.read_secret_version(
                    path=f"api-keys/{service_name}"
                )
                return secret['data']['data']['key']
            except Exception as e:
                print(f"从Vault获取密钥失败: {e}")
                return None
        else:
            # 从环境变量获取（加密存储）
            env_key = os.getenv(f"{service_name.upper()}_API_KEY")
            if env_key:
                try:
                    return self.cipher.decrypt(env_key.encode()).decode()
                except:
                    # 可能是未加密的密钥
                    return env_key
            return None
    
    def store_api_key(self, service_name: str, api_key: str):
        """存储API密钥"""
        if self.use_vault and hasattr(self, 'vault_client'):
            try:
                self.vault_client.secrets.kv.v2.create_or_update_secret(
                    path=f"api-keys/{service_name}",
                    secret={'key': api_key}
                )
                print(f"✅ 密钥已存储到Vault: {service_name}")
            except Exception as e:
                print(f"存储到Vault失败: {e}")
        else:
            # 加密后存储到环境变量（仅用于演示，生产环境不推荐）
            encrypted_key = self.cipher.encrypt(api_key.encode()).decode()
            print(f"加密后的密钥（设置环境变量 {service_name.upper()}_API_KEY）:")
            print(encrypted_key)

# 使用示例
if __name__ == "__main__":
    # 初始化密钥管理器
    secret_manager = SecretManager(use_vault=False)
    
    # 存储密钥（首次运行）
    # secret_manager.store_api_key("chattts", "your-actual-api-key-here")
    
    # 获取密钥
    api_key = secret_manager.get_api_key("chattts")
    if api_key:
        print(f"获取到API密钥: {api_key[:10]}...")
    else:
        print("未找到API密钥")

6.2 输入文本的XSS防护

语音合成服务接收用户输入的文本，必须做好安全防护：

import html
import re
from urllib.parse import quote

class TextSanitizer:
    """文本安全处理器"""
    
    def __init__(self):
        # 定义危险模式
        self.dangerous_patterns = [
            # XSS攻击
            (r'<script.*?>.*?</script>', '', re.IGNORECASE | re.DOTALL),
            (r'javascript:', '', re.IGNORECASE),
            (r'on\w+\s*=', '', re.IGNORECASE),
            (r'expression\s*\(', '', re.IGNORECASE),
            
            # SQL注入（如果文本会存储到数据库）
            (r'(\'|"|;|--|/\*|\*/|@@|char|nchar|varchar|nvarchar|alter|begin|cast|create|cursor|declare|delete|drop|end|exec|execute|fetch|insert|kill|open|select|sys|sysobjects|syscolumns|table|update)', 
             r'\\\1', re.IGNORECASE),
            
            # 命令注入
            (r'(\||&|;|`|\$\(|\n|\r)', '', re.IGNORECASE),
            
            # 路径遍历
            (r'\.\./', '', re.IGNORECASE),
        ]
        
        # 允许的HTML标签（如果需要富文本）
        self.allowed_tags = {
            'b', 'strong', 'i', 'em', 'u', 'br', 'p', 'span'
        }
    
    def sanitize(self, text: str, allow_html: bool = False) -> str:
        """
        清理文本
        
        Args:
            text: 输入文本
            allow_html: 是否允许HTML标签
            
        Returns:
            清理后的文本
        """
        if not text:
            return ""
        
        # 1. 移除不可见字符（除了空格、换行、制表符）
        cleaned = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', text)
        
        # 2. 处理危险模式
        for pattern, replacement, flags in self.dangerous_patterns:
            cleaned = re.sub(pattern, replacement, cleaned, flags=flags)
        
        # 3. HTML转义（如果不允许HTML）
        if not allow_html:
            cleaned = html.escape(cleaned)
        else:
            # 白名单过滤HTML标签
            cleaned = self._filter_html(cleaned)
        
        # 4. URL编码特殊字符
        cleaned = self._encode_special_chars(cleaned)
        
        # 5. 长度限制
        max_length = 5000  # 根据实际需求调整
        if len(cleaned) > max_length:
            cleaned = cleaned[:max_length] + "...[已截断]"
        
        return cleaned
    
    def _filter_html(self, html_text: str) -> str:
        """过滤HTML标签，只允许白名单内的标签"""
        from html.parser import HTMLParser
        
        class HTMLFilter(HTMLParser):
            def __init__(self, allowed_tags):
                super().__init__()
                self.allowed_tags = allowed_tags
                self.result = []
                self.current_tag = None
            
            def handle_starttag(self, tag, attrs):
                if tag in self.allowed_tags:
                    self.current_tag = tag
                    attrs_str = ' '.join([f'{k}="{v}"' for k, v in attrs])
                    self.result.append(f'<{tag} {attrs_str}>' if attrs_str else f'<{tag}>')
                else:
                    self.current_tag = None
            
            def handle_endtag(self, tag):
                if tag in self.allowed_tags and self.current_tag == tag:
                    self.result.append(f'</{tag}>')
                    self.current_tag = None
            
            def handle_data(self, data):
                self.result.append(data)
        
        parser = HTMLFilter(self.allowed_tags)
        parser.feed(html_text)
        return ''.join(parser.result)
    
    def _encode_special_chars(self, text: str) -> str:
        """编码特殊字符"""
        # 保留常见的标点符号
        safe_chars = set(".,!?;:'\"()[]{}<>-–—~@#$%^&*+=|/\\ \t\n\r")
        
        result = []
        for char in text:
            if char.isalnum() or char in safe_chars or ord(char) > 127:
                result.append(char)
            else:
                result.append(quote(char))
        
        return ''.join(result)
    
    def validate(self, text: str) -> tuple[bool, str]:
        """
        验证文本安全性
        
        Returns:
            (是否安全, 错误信息)
        """
        # 检查长度
        if len(text) > 10000:
            return False, "文本过长"
        
        # 检查危险内容
        test_sanitized = self.sanitize(text, allow_html=False)
        if len(test_sanitized) < len(text) * 0.5:
            # 如果清理后长度减少超过50%，可能包含大量危险内容
            return False, "文本包含不安全内容"
        
        # 检查编码
        try:
            text.encode('utf-8')
        except UnicodeEncodeError:
            return False, "文本编码错误"
        
        return True, ""

# 使用示例
if __name__ == "__main__":
    sanitizer = TextSanitizer()
    
    # 测试用例
    test_cases = [
        "正常文本",
        "<script>alert('xss')</script>恶意代码",
        "正常文本中有' OR '1'='1这样的SQL注入",
        "rm -rf / 命令注入",
        "../../etc/passwd 路径遍历",
        "正常文本<script>alert(1)</script>混合内容",
    ]
    
    for test in test_cases:
        cleaned = sanitizer.sanitize(test)
        is_valid, message = sanitizer.validate(test)
        
        print(f"原始: {test[:50]}...")
        print(f"清理后: {cleaned[:50]}...")
        print(f"验证: {'✅' if is_valid else '❌'} {message}")
        print("-" * 50)

部署架构示意图