# AIfeng/2025-07-11 13:36:00
"""
豆包ASR语音识别服务使用示例
演示各种使用场景和最佳实践
"""

import asyncio
import os
import logging
from pathlib import Path
from typing import Dict, Any, Optional

# 导入ASR服务 - 支持相对导入和绝对导入
try:
    # 尝试相对导入（作为包运行时）
    from . import (
        recognize_file,
        recognize_audio_data,
        create_asr_service,
        run_recognition,
        ConfigManager
    )
except ImportError:
    # 回退到绝对导入（独立运行时）
    try:
        from asr_client import (
            recognize_file,
            recognize_audio_data,
            create_asr_service,
            run_recognition
        )
        from config_manager import ConfigManager
    except ImportError:
        # 最后尝试直接导入
        import sys
        from pathlib import Path
        
        # 添加当前目录到路径
        current_dir = Path(__file__).parent
        sys.path.insert(0, str(current_dir))
        
        from asr_client import (
            recognize_file,
            recognize_audio_data,
            create_asr_service,
            run_recognition
        )
        from config_manager import ConfigManager


# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)


class ASRExamples:
    """
    ASR使用示例集合
    """
    
    def __init__(self, app_key: str, access_key: str):
        self.app_key = app_key
        self.access_key = access_key
    
    """
        示例1: 简单文件识别
    """
    async def example_1_simple_file_recognition(self, audio_path: str):
        
        logger.info("=== 示例1: 简单文件识别 ===")
        
        try:
            result = await recognize_file(
                audio_path=audio_path,
                app_key=self.app_key,
                access_key=self.access_key,
                streaming=True
            )
            
            logger.info(f"识别结果: {result}")
            return result
            
        except Exception as e:
            logger.error(f"识别失败: {e}")
            return None
    
    """
    示例2: 流式识别with实时回调 - 简化版流式输出演示
    """
    async def example_2_streaming_with_callback(self, audio_path: str):
       
        logger.info("=== 示例2: 流式识别with实时回调 - 简化版流式输出演示 ===")
        
        # 流式输出状态
        self.current_text = ""
        self.result_count = 0
        
        def clear_line():
            """清除当前行"""
            print('\r' + ' ' * 100 + '\r', end='', flush=True)
        
        def print_streaming_result(text: str, is_final: bool = False):
            """打印流式结果"""
            clear_line()
            status = "[最终]" if is_final else "[流式]"
            timestamp = f"[{self.result_count:02d}]"
            print(f"{timestamp}{status} {text}", end='', flush=True)
            if is_final:
                print()  # 最终结果换行
        
        # 定义结果回调函数 - 演示实时文本更新
        def on_result(result: Dict[str, Any]):
            self.result_count += 1
            
            if result.get('payload_msg'):
                payload = result['payload_msg']
                
                # 检查是否有识别结果
                if 'result' in payload and 'text' in payload['result']:
                    new_text = payload['result']['text']
                    
                    # 显示累积文本更新
                    print_streaming_result(new_text, False)
                    self.current_text = new_text
            
            # 检查是否为最终结果
            if result.get('is_last_package', False):
                print_streaming_result(self.current_text, True)
                logger.info(f"识别完成，共收到{self.result_count}次流式结果")
        
        print("\n观察流式文本累积更新效果:")
        print()
        
        # 创建服务实例
        service = create_asr_service(
            app_key=self.app_key,
            access_key=self.access_key,
            streaming=True,
            debug=False  # 关闭调试日志以便观察输出
        )
        
        try:
            result = await service.recognize_file(
                audio_path,
                result_callback=on_result
            )
            
            print()
            logger.info(f"=== 识别结果摘要 ===")
            logger.info(f"最终文本: {self.current_text}")
            logger.info(f"流式更新次数: {self.result_count}")
            
            return result
            
        except Exception as e:
            logger.error(f"识别失败: {e}")
            return None
        finally:
            await service.close()
    
    
    """
    示例3: 非流式识别
    """
    async def example_3_non_streaming_recognition(self, audio_path: str):

        logger.info("=== 示例3: 非流式识别 ===")
        
        try:
            result = await recognize_file(
                audio_path=audio_path,
                app_key=self.app_key,
                access_key=self.access_key,
                streaming=False  # 非流式
            )
            
            logger.info(f"识别结果: {result}")
            return result
            
        except Exception as e:
            logger.error(f"识别失败: {e}")
            return None
    
    """
        示例4: 音频数据识别
    """
    async def example_4_audio_data_recognition(self, audio_data: bytes, audio_format: str = "wav"):

        logger.info("=== 示例4: 音频数据识别 ===")
        
        try:
            result = await recognize_audio_data(
                audio_data=audio_data,
                audio_format=audio_format,
                app_key=self.app_key,
                access_key=self.access_key,
                streaming=True
            )
            
            logger.info(f"识别结果: {result}")
            return result
            
        except Exception as e:
            logger.error(f"识别失败: {e}")
            return None
    
    """
        示例5: 基于配置文件的识别
    """
    async def example_5_config_based_recognition(self, audio_path: str, config_path: str):
        """
        示例5: 基于配置文件的识别
        """
        logger.info("=== 示例5: 基于配置文件的识别 ===")
        
        try:
            result = await recognize_file(
                audio_path=audio_path,
                config_path=config_path
            )
            
            logger.info(f"识别结果: {result}")
            return result
            
        except Exception as e:
            logger.error(f"识别失败: {e}")
            return None
    
    """
    示例6: 批量识别
    """
    async def example_6_batch_recognition(self, audio_files: list):
        
        logger.info("=== 示例6: 批量识别 ===")
        
        results = []
        
        # 创建服务实例（复用连接）
        service = create_asr_service(
            app_key=self.app_key,
            access_key=self.access_key,
            streaming=True
        )
        
        try:
            for i, audio_file in enumerate(audio_files):
                logger.info(f"处理文件 {i+1}/{len(audio_files)}: {audio_file}")
                
                try:
                    result = await service.recognize_file(audio_file)
                    results.append({
                        'file': audio_file,
                        'result': result,
                        'status': 'success'
                    })
                    
                except Exception as e:
                    logger.error(f"文件 {audio_file} 识别失败: {e}")
                    results.append({
                        'file': audio_file,
                        'result': None,
                        'status': 'failed',
                        'error': str(e)
                    })
            
            logger.info(f"批量识别完成，成功: {sum(1 for r in results if r['status'] == 'success')}/{len(results)}")
            return results
            
        finally:
            await service.close()
    
    """
        示例7: 同步识别（简单场景）
    """
    def example_7_sync_recognition(self, audio_path: str):
       
        logger.info("=== 示例7: 同步识别 ===")
        
        try:
            result = run_recognition(
                audio_path=audio_path,
                app_key=self.app_key,
                access_key=self.access_key,
                streaming=True
            )
            
            logger.info(f"识别结果: {result}")
            return result
            
        except Exception as e:
            logger.error(f"识别失败: {e}")
            return None
    
    """
        示例8: 自定义配置识别
    """
    async def example_8_custom_config_recognition(self, audio_path: str):
        
        logger.info("=== 示例8: 自定义配置识别 ===")
        
        # 自定义配置
        custom_config = {
            'asr_config': {
                'enable_punc': True,
                'seg_duration': 300,  # 自定义分段时长
                'streaming_mode': True
            },
            'audio_config': {
                'default_rate': 16000,
                'default_bits': 16,
                'default_channel': 1
            },
            'connection_config': {
                'timeout': 60,  # 自定义超时时间
                'retry_times': 5
            },
            'logging_config': {
                'enable_debug': True
            }
        }
        
        service = create_asr_service(
            app_key=self.app_key,
            access_key=self.access_key,
            custom_config=custom_config
        )
        
        try:
            result = await service.recognize_file(audio_path)
            logger.info(f"识别结果: {result}")
            return result
            
        except Exception as e:
            logger.error(f"识别失败: {e}")
            return None
        finally:
            await service.close()


def create_sample_config(config_path: str, app_key: str, access_key: str):
    """
    创建示例配置文件
    """
    config_manager = ConfigManager()
    
    # 创建配置
    config = config_manager.create_default_config(config_path)
    config['auth_config']['app_key'] = app_key
    config['auth_config']['access_key'] = access_key
    config['logging_config']['enable_debug'] = True
    
    # 更新配置管理器的配置并保存
    config_manager.update_config(config)
    config_manager.save_config(config_path)
    logger.info(f"示例配置文件已创建: {config_path}")


async def run_all_examples():
    """
    运行所有示例
    """
    # 从环境变量获取密钥
    app_key = os.getenv('DOUBAO_APP_KEY', '1549099156')
    access_key = os.getenv('DOUBAO_ACCESS_KEY', '0GcKVco6j09bThrIgQWTWa3g1nA91_9C')
    
    if app_key == 'your_app_key_here' or access_key == 'your_access_key_here':
        logger.warning("请设置环境变量 DOUBAO_APP_KEY 和 DOUBAO_ACCESS_KEY")
        logger.info("或者直接修改代码中的密钥")
        return
    
    # 示例音频文件路径（请替换为实际路径）
    audio_path = "E:\\fengyang\\eman_one\\speech.wav"
    
    if not Path(audio_path).exists():
        logger.warning(f"音频文件不存在: {audio_path}")
        logger.info("请替换为实际的音频文件路径")
        return
    
    # 创建示例实例
    examples = ASRExamples(app_key, access_key)
    
    # 创建示例配置文件
    config_path = "example_config.json"
    create_sample_config(config_path, app_key, access_key)
    
    try:
        # 运行示例
        # await examples.example_1_simple_file_recognition(audio_path)
        await examples.example_2_streaming_with_callback(audio_path)
        # await examples.example_3_non_streaming_recognition(audio_path)
        
        # 音频数据示例（需要实际音频数据）
        # with open(audio_path, 'rb') as f:
        #     audio_data = f.read()
        # await examples.example_4_audio_data_recognition(audio_data)
        
        # await examples.example_5_config_based_recognition(audio_path, config_path)
        
        # 批量识别示例
        # audio_files = [audio_path]  # 添加更多文件
        # await examples.example_6_batch_recognition(audio_files)
        
        # 同步识别示例
        # examples.example_7_sync_recognition(audio_path)
        
        # await examples.example_8_custom_config_recognition(audio_path)
        
    except Exception as e:
        logger.error(f"示例运行失败: {e}")
    
    finally:
        # 清理示例配置文件
        if Path(config_path).exists():
            os.remove(config_path)
            logger.info(f"已清理示例配置文件: {config_path}")


if __name__ == "__main__":
    # 运行所有示例
    asyncio.run(run_all_examples())
    
    # 或者运行单个示例
    # app_key = "your_app_key"
    # access_key = "your_access_key"
    # audio_path = "path/to/audio.wav"
    # 
    # examples = ASRExamples(app_key, access_key)
    # asyncio.run(examples.example_1_simple_file_recognition(audio_path))