Upload 47 files

app.py

@@ -31,6 +31,8 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'backend'))
 # 导入后端模块
 from modules.gateway import GradioAPIGateway, GatewayConfig
 from modules.groq_client import GroqConfig
+from modules.siliconflow_client import SiliconFlowConfig
+from modules.stream_processor import StreamProcessor, StreamConfig
 from modules.logging_config import setup_logging, get_component_logger, Component
 
 # 配置日志
@@ -39,13 +41,14 @@ logger = get_component_logger(Component.SYSTEM)
 
 from contextlib import asynccontextmanager
 
-# 全局网关实例
+# 全局实例
 gateway: Optional[GradioAPIGateway] = None
+stream_processor: Optional[StreamProcessor] = None
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """应用生命周期管理"""
-    global gateway
+    global gateway, stream_processor
     
     # 启动时初始化
     logger.info("初始化 Universal Fast Dubbing 后端...")
@@ -58,6 +61,13 @@ async def lifespan(app: FastAPI):
         llm_model=os.getenv("LLM_MODEL", "llama3-8b-8192")
     ) if groq_api_key else None
     
+    # 配置 SiliconFlow
+    siliconflow_api_key = os.getenv("SILICONFLOW_API_KEY")
+    siliconflow_config = SiliconFlowConfig(
+        api_key=siliconflow_api_key,
+        tts_model=os.getenv("SILICONFLOW_TTS_MODEL", "fishaudio/fish-speech-1.5")
+    ) if siliconflow_api_key else None
+    
     # 配置网关
     gateway_config = GatewayConfig(
         temp_dir=os.getenv("TEMP_DIR", "temp/gateway"),
@@ -69,14 +79,28 @@ async def lifespan(app: FastAPI):
     gateway = GradioAPIGateway(config=gateway_config, groq_config=groq_config)
     await gateway.initialize()
     
+    # 初始化流式处理器
+    stream_config = StreamConfig(
+        temp_dir=os.getenv("TEMP_DIR", "temp/stream"),
+        max_segment_duration=float(os.getenv("MAX_SEGMENT_DURATION", "120")),
+        first_segment_target=float(os.getenv("FIRST_SEGMENT_TARGET", "30"))
+    )
+    stream_processor = StreamProcessor(
+        config=stream_config,
+        groq_config=groq_config,
+        siliconflow_config=siliconflow_config
+    )
+    await stream_processor.initialize()
+    
     logger.info("后端初始化完成")
     
     yield  # 应用运行期间
     
     # 关闭时清理
     if gateway:
-        # 这里可以添加清理逻辑
         pass
+    if stream_processor:
+        stream_processor.cleanup()
 
 # 创建 FastAPI 应用
 app = FastAPI(
@@ -452,10 +476,10 @@ async def quick_start_dubbing(
     """
     
     async def event_generator() -> AsyncGenerator[str, None]:
-        """SSE 事件生成器"""
+        """SSE 事件生成器 - 使用新的流式处理器"""
         try:
-            if not gateway or not gateway.is_initialized:
-                yield f"data: {json.dumps({'type': 'error', 'message': '后端未初始化'})}\n\n"
+            if not stream_processor or not stream_processor.is_initialized:
+                yield f"data: {json.dumps({'type': 'error', 'message': '流式处理器未初始化'})}\n\n"
                 return
             
             # 解析客户端配置
@@ -468,55 +492,125 @@ async def quick_start_dubbing(
             
             logger.info(f"[快速启动] 收到音频文件: {audio.filename}")
             
-            # 读取音频数据
+            # 读取音频数据并保存到临时文件
             audio_content = await audio.read()
+            temp_audio_path = f"temp/stream/upload_{os.urandom(4).hex()}.wav"
+            os.makedirs(os.path.dirname(temp_audio_path), exist_ok=True)
+            
+            with open(temp_audio_path, 'wb') as f:
+                f.write(audio_content)
             
             # 发送初始进度
-            yield f"data: {json.dumps({'type': 'progress', 'progress': 10, 'message': '音频上传完成，开始处理...'})}\n\n"
+            yield f"data: {json.dumps({'type': 'progress', 'progress': 5, 'message': '音频上传完成，开始处理...'})}\n\n"
             
-            # 准备处理数据
-            process_data = {
-                "audio_data": audio_content,
-                "client_config": ext_config,
-                "segment_duration": 60  # 快速模式使用较短分段
-            }
+            # 使用流式处理器处理
+            async for update in stream_processor.process_stream(temp_audio_path, ext_config):
+                yield f"data: {json.dumps(update)}\n\n"
             
-            # 使用分段处理模式
-            segment_index = 0
-            total_duration = 0
+            # 清理临时文件
+            try:
+                if os.path.exists(temp_audio_path):
+                    os.remove(temp_audio_path)
+            except:
+                pass
             
-            async for update in gateway.process_request_segmented("upload", process_data):
-                update_type = update.get("type", "progress")
-                
-                if update_type == "progress":
-                    yield f"data: {json.dumps({'type': 'progress', 'progress': update.get('progress', 0), 'message': update.get('message', '')})}\n\n"
-                    
-                elif update_type == "segment_ready":
-                    segment_data = update.get("segment", {})
-                    audio_data = segment_data.get("audio_data")
-                    
-                    # 二进制数据转base64
-                    if isinstance(audio_data, bytes):
-                        audio_data = base64.b64encode(audio_data).decode('utf-8')
-                    
-                    yield f"data: {json.dumps({'type': 'segment_ready', 'index': segment_index, 'startTime': segment_data.get('start_time', total_duration), 'duration': segment_data.get('duration', 60), 'audioData': audio_data})}\n\n"
-                    
-                    total_duration += segment_data.get("duration", 60)
-                    segment_index += 1
-                    
-                elif update_type == "complete":
-                    yield f"data: {json.dumps({'type': 'complete', 'totalSegments': segment_index, 'totalDuration': total_duration, 'processingTime': update.get('processing_time', 0)})}\n\n"
-                    break
-                    
-                elif update_type == "error":
-                    yield f"data: {json.dumps({'type': 'error', 'message': update.get('message', '处理失败')})}\n\n"
-                    break
+        except Exception as e:
+            logger.error(f"[快速启动] 错误: {e}")
+            yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+    
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        }
+    )
+
+
+@app.post("/api/upload_and_process")
+async def upload_and_process(
+    audio: UploadFile = File(...),
+    client_config: str = Form(default="{}")
+):
+    """
+    上传音频并流式处理 API
+    
+    完整流程：
+    扩展上传音频 → 语音识别 (Whisper V3 带时间戳) → 带时间戳分段异步处理
+        → 翻译 + 角色识别 (Llama 3) → 语音合成 (Edge-TTS / SiliconFlow)
+        → 音频同步对齐 (时间戳匹配) → 配音音频分段输出
+    
+    参数:
+        audio: 音频文件（来自扩展端下载或录制）
+        client_config: 客户端配置JSON，包含:
+            - asrProvider: ASR提供商 (groq/siliconflow)
+            - ttsProvider: TTS提供商 (edge-tts/siliconflow)
+            - voiceMale: 男声语音模型
+            - voiceFemale: 女声语音模型
+            - syncOffset: 同步偏移量（毫秒）
+    
+    返回:
+        SSE 事件流
+    """
+    
+    async def event_generator() -> AsyncGenerator[str, None]:
+        """SSE 事件生成器"""
+        temp_audio_path = None
+        
+        try:
+            if not stream_processor or not stream_processor.is_initialized:
+                yield f"data: {json.dumps({'type': 'error', 'message': '流式处理器未初始化'})}\n\n"
+                return
+            
+            # 解析客户端配置
+            ext_config = {}
+            if client_config and client_config != "{}":
+                try:
+                    ext_config = json.loads(client_config)
+                    logger.info(f"[上传处理] 客户端配置: {list(ext_config.keys())}")
+                except json.JSONDecodeError as e:
+                    logger.warning(f"[上传处理] 配置解析失败: {e}")
+            
+            # 获取文件信息
+            filename = audio.filename or "unknown"
+            content_type = audio.content_type or "audio/mpeg"
             
-            logger.info(f"[快速启动] 处理完成: {segment_index}段, 总时长: {total_duration}秒")
+            logger.info(f"[上传处理] 收到文件: {filename}, 类型: {content_type}")
+            
+            # 保存上传的音频到临时文件
+            file_ext = os.path.splitext(filename)[1] or ".mp3"
+            temp_audio_path = f"temp/stream/upload_{os.urandom(4).hex()}{file_ext}"
+            os.makedirs(os.path.dirname(temp_audio_path), exist_ok=True)
+            
+            audio_content = await audio.read()
+            file_size_mb = len(audio_content) / (1024 * 1024)
+            
+            with open(temp_audio_path, 'wb') as f:
+                f.write(audio_content)
+            
+            logger.info(f"[上传处理] 音频已保存: {temp_audio_path}, 大小: {file_size_mb:.2f}MB")
+            
+            # 发送初始进度
+            yield f"data: {json.dumps({'type': 'progress', 'progress': 5, 'message': f'音频上传完成 ({file_size_mb:.1f}MB)，开始处理...'})}\n\n"
+            
+            # 使用流式处理器处理
+            async for update in stream_processor.process_stream(temp_audio_path, ext_config):
+                yield f"data: {json.dumps(update)}\n\n"
             
         except Exception as e:
-            logger.error(f"[快速启动] 错误: {e}")
+            logger.error(f"[上传处理] 错误: {e}", exc_info=True)
             yield f"data: {json.dumps({'type': 'error', 'message': str(e)})}\n\n"
+            
+        finally:
+            # 清理临时文件
+            if temp_audio_path:
+                try:
+                    if os.path.exists(temp_audio_path):
+                        os.remove(temp_audio_path)
+                except Exception as e:
+                    logger.warning(f"[上传处理] 清理临时文件失败: {e}")
     
     return StreamingResponse(
         event_generator(),

backend/.env.example

@@ -25,7 +25,8 @@ PORT=7860
 # ============================================================================
 # 处理配置
 # ============================================================================
-MAX_SEGMENT_DURATION=480
+MAX_SEGMENT_DURATION=120
+FIRST_SEGMENT_TARGET=30
 MAX_CONCURRENT_WORKERS=3
 CACHE_DURATION=3600
 

backend/modules/__init__.py

@@ -112,6 +112,23 @@ from .performance_monitor import (
     track_performance,
 )
 
+from .stream_processor import (
+    StreamProcessor,
+    StreamConfig,
+    ProcessingStage,
+    SegmentData,
+)
+
+from .siliconflow_client import (
+    SiliconFlowClient,
+    SiliconFlowConfig,
+    SiliconFlowError,
+    SiliconFlowRateLimitError,
+    SiliconFlowTimeoutError,
+    SiliconFlowAuthError,
+    SiliconFlowConnectionError,
+)
+
 __all__ = [
     # 统一错误响应
     "ErrorCode",
@@ -148,6 +165,14 @@ __all__ = [
     "GroqAuthError",
     "GroqConnectionError",
     "RetryStats",
+    # SiliconFlow 客户端
+    "SiliconFlowClient",
+    "SiliconFlowConfig",
+    "SiliconFlowError",
+    "SiliconFlowRateLimitError",
+    "SiliconFlowTimeoutError",
+    "SiliconFlowAuthError",
+    "SiliconFlowConnectionError",
     # TTS 生成器
     "TTSGenerator",
     "TTSConfig",
@@ -192,4 +217,9 @@ __all__ = [
     "OperationTracker",
     "get_performance_monitor",
     "track_performance",
+    # 流式处理器
+    "StreamProcessor",
+    "StreamConfig",
+    "ProcessingStage",
+    "SegmentData",
 ]

backend/modules/router.py

@@ -93,7 +93,7 @@ class RequestRouter:
         路由处理请求
         
         参数:
-            mode: 处理模式 ("record", "upload", "url")
+            mode: 处理模式 ("record", "upload")
             data: 请求数据
             session_id: 会话ID（可选）
         
@@ -108,20 +108,8 @@ class RequestRouter:
         
         logger.info(f"[{session_id}] 路由请求: mode={mode}")
         
-        # 检查是否有音频数据（上传模式）
-        if data.get('audio_data') or data.get('audio_path'):
-            return await self._handle_audio_upload(data, session_id)
-        
-        # URL模式：返回错误，提示需要音频数据
-        if mode == "url":
-            raise RouterError(
-                "URL模式需要扩展端先获取音频后上传。"
-                "请确保扩展端音频拦截功能正常工作，或使用录制模式。"
-            )
-        
-        # 其他情况
-        raise RouterError("需要提供 audio_data 或 audio_path")
-    
+        # 所有模式都走录制/上传处理
+        return await self._handle_audio_upload(data, session_id)
     
     async def _handle_audio_upload(
         self,

backend/modules/stream_processor.py

@@ -0,0 +1,576 @@
+"""
+流式异步处理模块
+
+实现完整的流式配音处理流程：
+扩展上传音频 → 语音识别 (Whisper V3 带时间戳) → 带时间戳分段异步处理
+    → 翻译 + 角色识别 (Llama 3) → 语音合成 (Edge-TTS / SiliconFlow)
+    → 音频同步对齐 (时间戳匹配) → 配音音频分段输出
+
+支持：
+- 流式 SSE 输出
+- 分段并行处理
+- 第一段完成即可开始播放
+"""
+
+import os
+import asyncio
+import logging
+import time
+import uuid
+import base64
+from typing import Dict, Any, Optional, List, AsyncGenerator
+from dataclasses import dataclass, field
+from enum import Enum
+
+from .groq_client import GroqClient, GroqConfig, GroqError
+from .siliconflow_client import SiliconFlowClient, SiliconFlowConfig
+from .tts_generator import TTSGenerator, TTSConfig
+from .audio_sync import AudioSyncEngine, SyncConfig
+
+logger = logging.getLogger(__name__)
+
+
+class ProcessingStage(Enum):
+    """处理阶段枚举"""
+    INIT = "init"
+    ASR = "asr"
+    TRANSLATE = "translate"
+    TTS = "tts"
+    SYNC = "sync"
+    COMPLETE = "complete"
+    ERROR = "error"
+
+
+@dataclass
+class StreamConfig:
+    """
+    流式处理配置
+    
+    属性:
+        temp_dir: 临时文件目录
+        max_segment_duration: 每段最大时长（秒）
+        first_segment_target: 首段目标时长（秒），用于快速开始播放
+        parallel_tts: 是否并行生成TTS
+        max_tts_workers: TTS最大并发数
+    """
+    temp_dir: str = "temp/stream"
+    max_segment_duration: float = 120.0  # 2分钟
+    first_segment_target: float = 30.0   # 首段30秒，快速开始
+    parallel_tts: bool = True
+    max_tts_workers: int = 3
+
+
+@dataclass
+class SegmentData:
+    """
+    分段数据
+    
+    属性:
+        index: 分段索引
+        start_time: 开始时间（秒）
+        end_time: 结束时间（秒）
+        original_text: 原文
+        translated_text: 译文
+        role: 角色标签
+        audio_path: 生成的音频路径
+        audio_data: 音频二进制数据
+    """
+    index: int
+    start_time: float
+    end_time: float
+    original_text: str = ""
+    translated_text: str = ""
+    role: str = "MALE"
+    audio_path: Optional[str] = None
+    audio_data: Optional[bytes] = None
+
+
+class StreamProcessor:
+    """
+    流式异步处理器
+    
+    实现完整的配音处理流程，支持流式输出。
+    
+    使用示例:
+        processor = StreamProcessor()
+        await processor.initialize()
+        
+        async for update in processor.process_stream(audio_path, config):
+            if update['type'] == 'segment_ready':
+                # 播放分段音频
+                play_audio(update['audio_data'])
+    """
+    
+    def __init__(
+        self,
+        config: Optional[StreamConfig] = None,
+        groq_config: Optional[GroqConfig] = None,
+        siliconflow_config: Optional[SiliconFlowConfig] = None
+    ):
+        """
+        初始化流式处理器
+        
+        参数:
+            config: 流式处理配置
+            groq_config: Groq 客户端配置
+            siliconflow_config: SiliconFlow 客户端配置
+        """
+        self.config = config or StreamConfig()
+        self._groq_config = groq_config
+        self._siliconflow_config = siliconflow_config
+        
+        # 子模块
+        self.groq_client: Optional[GroqClient] = None
+        self.siliconflow_client: Optional[SiliconFlowClient] = None
+        self.tts_generator: Optional[TTSGenerator] = None
+        self.audio_sync: Optional[AudioSyncEngine] = None
+        
+        self._initialized = False
+        self._temp_files: List[str] = []
+        
+        # 确保临时目录存在
+        os.makedirs(self.config.temp_dir, exist_ok=True)
+        
+        logger.info("流式处理器配置完成")
+    
+    async def initialize(self) -> None:
+        """初始化所有子模块"""
+        if self._initialized:
+            return
+        
+        logger.info("初始化流式处理器...")
+        
+        # 初始化 Groq 客户端（ASR + LLM）
+        if self._groq_config:
+            self.groq_client = GroqClient(self._groq_config)
+            await self.groq_client.initialize()
+        
+        # 初始化 SiliconFlow 客户端（可选的 ASR + TTS）
+        if self._siliconflow_config:
+            self.siliconflow_client = SiliconFlowClient(self._siliconflow_config)
+            await self.siliconflow_client.initialize()
+        
+        # 初始化 TTS 生成器
+        tts_config = TTSConfig(temp_dir=os.path.join(self.config.temp_dir, "tts"))
+        self.tts_generator = TTSGenerator(tts_config)
+        
+        # 初始化音频同步引擎
+        sync_config = SyncConfig(temp_dir=os.path.join(self.config.temp_dir, "sync"))
+        self.audio_sync = AudioSyncEngine(sync_config)
+        
+        self._initialized = True
+        logger.info("流式处理器初始化完成")
+    
+    def _ensure_initialized(self) -> None:
+        """确保处理器已初始化"""
+        if not self._initialized:
+            raise RuntimeError("流式处理器未初始化，请先调用 initialize()")
+    
+    async def process_stream(
+        self,
+        audio_path: str,
+        client_config: Optional[Dict[str, Any]] = None
+    ) -> AsyncGenerator[Dict[str, Any], None]:
+        """
+        流式处理音频文件
+        
+        完整流程：
+        1. ASR 语音识别（带时间戳）
+        2. 按时间戳分段
+        3. 并行处理每段：翻译 → TTS → 同步
+        4. 流式输出每段结果
+        
+        参数:
+            audio_path: 音频文件路径
+            client_config: 客户端配置
+        
+        生成:
+            处理更新字典，包含:
+            - type: 消息类型 (progress/segment_ready/complete/error)
+            - 其他字段根据类型不同
+        """
+        self._ensure_initialized()
+        
+        session_id = str(uuid.uuid4())[:8]
+        start_time = time.time()
+        
+        logger.info(f"[{session_id}] 开始流式处理: {audio_path}")
+        
+        try:
+            # 1. 发送初始进度
+            yield {
+                'type': 'progress',
+                'stage': ProcessingStage.INIT.value,
+                'progress': 5,
+                'message': '正在分析音频...'
+            }
+            
+            # 2. ASR 语音识别
+            yield {
+                'type': 'progress',
+                'stage': ProcessingStage.ASR.value,
+                'progress': 10,
+                'message': '语音识别中...'
+            }
+            
+            asr_result = await self._do_asr(audio_path, client_config)
+            
+            if not asr_result.get('segments'):
+                yield {
+                    'type': 'error',
+                    'message': '语音识别结果为空'
+                }
+                return
+            
+            source_language = asr_result.get('language', 'unknown')
+            total_duration = asr_result.get('duration', 0)
+            
+            logger.info(
+                f"[{session_id}] ASR完成: 语言={source_language}, "
+                f"时长={total_duration:.1f}s, 片段数={len(asr_result['segments'])}"
+            )
+            
+            yield {
+                'type': 'progress',
+                'stage': ProcessingStage.ASR.value,
+                'progress': 25,
+                'message': f'识别完成: {source_language}, {len(asr_result["segments"])} 个片段'
+            }
+            
+            # 3. 智能分组（按时间戳分段，优化首段快速输出）
+            segment_groups = self._group_segments_for_streaming(
+                asr_result['segments'],
+                total_duration
+            )
+            
+            logger.info(f"[{session_id}] 分组完成: {len(segment_groups)} 组")
+            
+            # 4. 流式处理每个分组
+            total_groups = len(segment_groups)
+            processed_groups = 0
+            
+            for group_index, group in enumerate(segment_groups):
+                group_start_time = group[0].get('start', 0) if group else 0
+                group_end_time = group[-1].get('end', 0) if group else 0
+                
+                yield {
+                    'type': 'progress',
+                    'stage': ProcessingStage.TRANSLATE.value,
+                    'progress': 25 + (group_index / total_groups) * 60,
+                    'message': f'处理第 {group_index + 1}/{total_groups} 段...'
+                }
+                
+                # 4.1 翻译当前分组
+                translated_segments = await self._translate_segments(
+                    group,
+                    source_language,
+                    client_config
+                )
+                
+                # 4.2 TTS 生成
+                yield {
+                    'type': 'progress',
+                    'stage': ProcessingStage.TTS.value,
+                    'progress': 25 + (group_index / total_groups) * 60 + 20,
+                    'message': f'生成配音 {group_index + 1}/{total_groups}...'
+                }
+                
+                tts_results = await self._generate_tts(
+                    translated_segments,
+                    client_config
+                )
+                
+                # 4.3 音频同步
+                yield {
+                    'type': 'progress',
+                    'stage': ProcessingStage.SYNC.value,
+                    'progress': 25 + (group_index / total_groups) * 60 + 40,
+                    'message': f'同步音频 {group_index + 1}/{total_groups}...'
+                }
+                
+                synced_audio = await self._sync_audio(
+                    tts_results,
+                    translated_segments,
+                    group_end_time - group_start_time,
+                    client_config
+                )
+                
+                # 4.4 读取音频数据
+                audio_data = None
+                if synced_audio and os.path.exists(synced_audio):
+                    with open(synced_audio, 'rb') as f:
+                        audio_data = f.read()
+                
+                # 4.5 输出分段结果
+                processed_groups += 1
+                
+                yield {
+                    'type': 'segment_ready',
+                    'index': group_index,
+                    'start_time': group_start_time,
+                    'end_time': group_end_time,
+                    'duration': group_end_time - group_start_time,
+                    'audio_data': base64.b64encode(audio_data).decode('utf-8') if audio_data else None,
+                    'segments': [
+                        {
+                            'original': seg.get('text', ''),
+                            'translated': seg.get('cn', ''),
+                            'role': seg.get('role', 'MALE'),
+                            'start': seg.get('start', 0),
+                            'end': seg.get('end', 0)
+                        }
+                        for seg in translated_segments
+                    ]
+                }
+                
+                logger.info(
+                    f"[{session_id}] 分组 {group_index + 1} 完成: "
+                    f"{group_start_time:.1f}s - {group_end_time:.1f}s"
+                )
+            
+            # 5. 处理完成
+            processing_time = time.time() - start_time
+            
+            yield {
+                'type': 'complete',
+                'total_segments': processed_groups,
+                'total_duration': total_duration,
+                'processing_time': processing_time,
+                'source_language': source_language
+            }
+            
+            logger.info(
+                f"[{session_id}] 流式处理完成: "
+                f"{processed_groups} 段, 耗时 {processing_time:.1f}s"
+            )
+            
+        except Exception as e:
+            logger.error(f"[{session_id}] 流式处理失败: {e}")
+            yield {
+                'type': 'error',
+                'message': str(e)
+            }
+    
+    async def _do_asr(
+        self,
+        audio_path: str,
+        client_config: Optional[Dict[str, Any]] = None
+    ) -> Dict[str, Any]:
+        """
+        执行语音识别
+        
+        根据配置选择 Groq Whisper 或 SiliconFlow SenseVoice
+        """
+        # 获取 ASR 提供商配置
+        asr_provider = 'groq'  # 默认使用 Groq
+        if client_config:
+            asr_provider = client_config.get('asrProvider', 'groq')
+        
+        if asr_provider == 'siliconflow' and self.siliconflow_client:
+            logger.info("使用 SiliconFlow SenseVoice 进行语音识别")
+            return await self.siliconflow_client.transcribe(audio_path)
+        elif self.groq_client:
+            logger.info("使用 Groq Whisper V3 进行语音识别")
+            return await self.groq_client.transcribe(audio_path)
+        else:
+            raise RuntimeError("没有可用的 ASR 服务")
+    
+    def _group_segments_for_streaming(
+        self,
+        segments: List[Dict[str, Any]],
+        total_duration: float
+    ) -> List[List[Dict[str, Any]]]:
+        """
+        将 ASR 片段分组用于流式处理
+        
+        策略：
+        1. 首段尽量短（30秒内），快速开始播放
+        2. 后续段按最大时长分组
+        3. 尽量在句子边界分割
+        """
+        if not segments:
+            return []
+        
+        groups = []
+        current_group = []
+        current_duration = 0
+        is_first_group = True
+        
+        # 首段目标时长
+        target_duration = self.config.first_segment_target if is_first_group else self.config.max_segment_duration
+        
+        for seg in segments:
+            seg_start = seg.get('start', 0)
+            seg_end = seg.get('end', 0)
+            seg_duration = seg_end - seg_start
+            
+            # 检查是否需要开始新组
+            if current_group and current_duration + seg_duration > target_duration:
+                groups.append(current_group)
+                current_group = []
+                current_duration = 0
+                is_first_group = False
+                target_duration = self.config.max_segment_duration
+            
+            current_group.append(seg)
+            current_duration += seg_duration
+        
+        # 添加最后一组
+        if current_group:
+            groups.append(current_group)
+        
+        return groups
+    
+    async def _translate_segments(
+        self,
+        segments: List[Dict[str, Any]],
+        source_language: str,
+        client_config: Optional[Dict[str, Any]] = None
+    ) -> List[Dict[str, Any]]:
+        """
+        翻译片段并识别角色
+        """
+        if not self.groq_client:
+            # 如果没有 Groq 客户端，返回原始片段
+            return [
+                {
+                    **seg,
+                    'cn': seg.get('text', ''),
+                    'role': 'MALE'
+                }
+                for seg in segments
+            ]
+        
+        # 合并文本进行翻译
+        text = ' '.join(seg.get('text', '') for seg in segments)
+        
+        translation_result = await self.groq_client.translate(
+            text,
+            source_language,
+            segments
+        )
+        
+        # 合并翻译结果到原始片段
+        translated_segments = translation_result.get('segments', [])
+        
+        result = []
+        for i, seg in enumerate(segments):
+            translated = translated_segments[i] if i < len(translated_segments) else {}
+            result.append({
+                **seg,
+                'cn': translated.get('cn', seg.get('text', '')),
+                'role': translated.get('role', 'MALE')
+            })
+        
+        return result
+    
+    async def _generate_tts(
+        self,
+        segments: List[Dict[str, Any]],
+        client_config: Optional[Dict[str, Any]] = None
+    ) -> List[Optional[str]]:
+        """
+        生成 TTS 音频
+        
+        根据配置选择 Edge-TTS 或 SiliconFlow TTS
+        """
+        # 获取 TTS 提供商配置
+        tts_provider = 'edge-tts'  # 默认使用 Edge-TTS
+        if client_config:
+            tts_provider = client_config.get('ttsProvider', 'edge-tts')
+        
+        # 准备 TTS 输入
+        tts_segments = [
+            {
+                'cn': seg.get('cn', ''),
+                'role': seg.get('role', 'MALE')
+            }
+            for seg in segments
+        ]
+        
+        if tts_provider == 'siliconflow' and self.siliconflow_client:
+            logger.info("使用 SiliconFlow TTS 生成配音")
+            # 使用 SiliconFlow TTS
+            results = []
+            for seg in tts_segments:
+                if seg['cn']:
+                    output_path = os.path.join(
+                        self.config.temp_dir,
+                        f"tts_{uuid.uuid4().hex[:8]}.mp3"
+                    )
+                    await self.siliconflow_client.synthesize(
+                        seg['cn'],
+                        seg['role'],
+                        output_path
+                    )
+                    results.append(output_path)
+                    self._temp_files.append(output_path)
+                else:
+                    results.append(None)
+            return results
+        else:
+            logger.info("使用 Edge-TTS 生成配音")
+            # 使用 Edge-TTS
+            return await self.tts_generator.generate(tts_segments, client_config)
+    
+    async def _sync_audio(
+        self,
+        tts_paths: List[Optional[str]],
+        segments: List[Dict[str, Any]],
+        target_duration: float,
+        client_config: Optional[Dict[str, Any]] = None
+    ) -> Optional[str]:
+        """
+        同步音频到时间戳
+        """
+        # 过滤有效的 TTS 路径
+        valid_paths = []
+        valid_segments = []
+        
+        for i, (path, seg) in enumerate(zip(tts_paths, segments)):
+            if path and os.path.exists(path):
+                valid_paths.append(path)
+                valid_segments.append({
+                    'start': seg.get('start', 0),
+                    'end': seg.get('end', 0)
+                })
+        
+        if not valid_paths:
+            logger.warning("没有有效的 TTS 音频")
+            return None
+        
+        # 调用音频同步引擎
+        return await self.audio_sync.align(
+            valid_paths,
+            valid_segments,
+            target_duration,
+            client_config
+        )
+    
+    def cleanup(self) -> int:
+        """清理临时文件"""
+        cleaned = 0
+        
+        for path in self._temp_files:
+            try:
+                if os.path.exists(path):
+                    os.remove(path)
+                    cleaned += 1
+            except Exception as e:
+                logger.warning(f"清理临时文件失败 {path}: {e}")
+        
+        self._temp_files.clear()
+        
+        if self.tts_generator:
+            cleaned += self.tts_generator.cleanup()
+        
+        if self.audio_sync:
+            cleaned += self.audio_sync.cleanup()
+        
+        logger.info(f"流式处理器清理完成: {cleaned} 个文件")
+        return cleaned
+    
+    @property
+    def is_initialized(self) -> bool:
+        """检查处理器是否已初始化"""
+        return self._initialized