修改代码以满足需求

import asyncio
import aiohttp
import json
import struct
import gzip
import uuid
import logging
import os
import subprocess
from typing import Optional, List, Dict, Any, Tuple, AsyncGenerator

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('run.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 常量定义
DEFAULT_SAMPLE_RATE = 16000
DEFAULT_OUTPUT_ENCODING = 'utf-8'

# 协议相关常量
ProtocolVersion = type('ProtocolVersion', (), {'V1': 0b0001})
MessageType = type('MessageType', (), {
    'CLIENT_FULL_REQUEST': 0b0001,
    'CLIENT_AUDIO_ONLY_REQUEST': 0b0010,
    'SERVER_FULL_RESPONSE': 0b1001,
    'SERVER_ERROR_RESPONSE': 0b1111
})
MessageTypeSpecificFlags = type('MessageTypeSpecificFlags', (), {
    'NO_SEQUENCE': 0b0000,
    'POS_SEQUENCE': 0b0001,
    'NEG_SEQUENCE': 0b0010,
    'NEG_WITH_SEQUENCE': 0b0011
})
SerializationType = type('SerializationType', (), {
    'NO_SERIALIZATION': 0b0000,
    'JSON': 0b0001
})
CompressionType = type('CompressionType', (), {'GZIP': 0b0001})

# 配置类
class Config:
    def __init__(self):
        self.auth = {
            ",
            "
        }

    @property
    def app_key(self) -> str:
        return self.auth["app_key"]

    @property
    def access_key(self) -> str:
        return self.auth["access_key"]

config = Config()

# 通用工具类
class CommonUtils:
    @staticmethod
    def gzip_compress(data: bytes) -> bytes:
        return gzip.compress(data)

    @staticmethod
    def gzip_decompress(data: bytes) -> bytes:
        return gzip.decompress(data)

    @staticmethod
    def convert_audio_to_pcm(audio_path: str, sample_rate: int = DEFAULT_SAMPLE_RATE) -> bytes:
        """
        直接将音频文件（MP3/WAV等）转换为原始PCM数据
        ✅ 安全版本：完全不操作源文件，仅读取转换
        """
        if not os.path.exists(audio_path):
            raise FileNotFoundError(f"音频文件不存在: {audio_path}")
            
        try:
            cmd = [
                "ffmpeg", "-v", "quiet", "-y", "-i", audio_path,
                "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(sample_rate),
                "-f", "s16le", "-"  # 输出到stdout，不生成任何文件
            ]
            # 执行ffmpeg转换，仅读取源文件，不修改/删除
            result = subprocess.run(
                cmd, 
                check=True, 
                stdout=subprocess.PIPE, 
                stderr=subprocess.PIPE
            )
            logger.info(f"音频转换完成，源文件 {audio_path} 保持完整")
            return result.stdout
            
        except subprocess.CalledProcessError as e:
            logger.error(f"FFmpeg转换失败: {e.stderr.decode()}")
            raise RuntimeError(f"音频转换失败: {e.stderr.decode()}")

# 请求头构建类
class AsrRequestHeader:
    def __init__(self):
        self.message_type = MessageType.CLIENT_FULL_REQUEST
        self.message_type_specific_flags = MessageTypeSpecificFlags.POS_SEQUENCE
        self.serialization_type = SerializationType.JSON
        self.compression_type = CompressionType.GZIP
        self.reserved_data = bytes([0x00])

    def to_bytes(self) -> bytes:
        header = bytearray()
        header.append((ProtocolVersion.V1 << 4) | 1)
        header.append((self.message_type << 4) | self.message_type_specific_flags)
        header.append((self.serialization_type << 4) | self.compression_type)
        header.extend(self.reserved_data)
        return bytes(header)

    @staticmethod
    def default_header() -> 'AsrRequestHeader':
        return AsrRequestHeader()

# 请求构建器
class RequestBuilder:
    @staticmethod
    def new_auth_headers() -> Dict[str, str]:
        return {
            "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
            "X-Api-Request-Id": str(uuid.uuid4()),
            "X-Api-Access-Key": config.access_key,
            "X-Api-App-Key": config.app_key
        }

    @staticmethod
    def new_full_client_request(seq: int) -> bytes:
        header = AsrRequestHeader.default_header()
        payload = {
            "user": {"uid": "demo_uid"},
            "audio": {"format": "pcm", "codec": "raw", "rate": 16000, "bits": 16, "channel": 1},
            "request": {
                "model_name": "bigmodel",
                "enable_itn": True,
                "enable_punc": True,
                "enable_ddc": True,
                "show_utterances": True,
                "enable_nonstream": False
            }
        }
        
        payload_bytes = json.dumps(payload).encode('utf-8')
        compressed_payload = CommonUtils.gzip_compress(payload_bytes)
        
        request = bytearray()
        request.extend(header.to_bytes())
        request.extend(struct.pack('>i', seq))
        request.extend(struct.pack('>I', len(compressed_payload)))
        request.extend(compressed_payload)
        
        return bytes(request)

    @staticmethod
    def new_audio_only_request(seq: int, segment: bytes, is_last: bool = False) -> bytes:
        header = AsrRequestHeader.default_header()
        header.message_type = MessageType.CLIENT_AUDIO_ONLY_REQUEST
        
        if is_last:
            header.message_type_specific_flags = MessageTypeSpecificFlags.NEG_WITH_SEQUENCE
            seq = -seq
        else:
            header.message_type_specific_flags = MessageTypeSpecificFlags.POS_SEQUENCE
        
        compressed_segment = CommonUtils.gzip_compress(segment)
        
        request = bytearray()
        request.extend(header.to_bytes())
        request.extend(struct.pack('>i', seq))
        request.extend(struct.pack('>I', len(compressed_segment)))
        request.extend(compressed_segment)
        
        return bytes(request)

# 响应处理类
class AsrResponse:
    def __init__(self):
        self.code = 0
        self.event = 0
        self.is_last_package = False
        self.payload_sequence = 0
        self.payload_size = 0
        self.payload_msg = None
        self.last_text = ""  # 记录上一次的完整文本，用于去重

    def to_dict(self) -> Dict[str, Any]:
        return {
            "code": self.code,
            "event": self.event,
            "is_last_package": self.is_last_package,
            "payload_sequence": self.payload_sequence,
            "payload_size": self.payload_size,
            "payload_msg": self.payload_msg
        }
    
    def get_incremental_text(self) -> str:
        """提取增量文本（解决重复问题）"""
        current_text = ""
        if self.payload_msg and isinstance(self.payload_msg, dict):
            if "result" in self.payload_msg:
                result = self.payload_msg["result"]
                if isinstance(result, dict):
                    # 优先获取完整文本
                    if "text" in result:
                        current_text = result["text"]
                    # 分句文本拼接
                    elif "utterances" in result and isinstance(result["utterances"], list):
                        utterances = []
                        for utt in result["utterances"]:
                            if isinstance(utt, dict) and "text" in utt:
                                utterances.append(utt["text"])
                        current_text = "".join(utterances)
        
        # 计算增量文本（只返回新增部分）
        incremental_text = current_text[len(self.last_text):]
        self.last_text = current_text  # 更新上一次的文本
        
        return incremental_text

class ResponseParser:
    @staticmethod
    def parse_response(msg: bytes) -> AsrResponse:
        response = AsrResponse()
        
        header_size = msg[0] & 0x0f
        message_type = msg[1] >> 4
        message_type_specific_flags = msg[1] & 0x0f
        serialization_method = msg[2] >> 4
        message_compression = msg[2] & 0x0f
        
        payload = msg[header_size*4:]
        
        # 解析标志位
        if message_type_specific_flags & 0x01:
            response.payload_sequence = struct.unpack('>i', payload[:4])[0]
            payload = payload[4:]
        if message_type_specific_flags & 0x02:
            response.is_last_package = True
        if message_type_specific_flags & 0x04:
            response.event = struct.unpack('>i', payload[:4])[0]
            payload = payload[4:]
            
        # 解析消息类型
        if message_type == MessageType.SERVER_FULL_RESPONSE:
            response.payload_size = struct.unpack('>I', payload[:4])[0]
            payload = payload[4:]
        elif message_type == MessageType.SERVER_ERROR_RESPONSE:
            response.code = struct.unpack('>i', payload[:4])[0]
            response.payload_size = struct.unpack('>I', payload[4:8])[0]
            payload = payload[8:]
            
        if not payload:
            return response
            
        # 解压缩
        if message_compression == CompressionType.GZIP:
            try:
                payload = CommonUtils.gzip_decompress(payload)
            except Exception as e:
                logger.error(f"解压缩失败: {e}")
                return response
                
        # 解析JSON
        try:
            if serialization_method == SerializationType.JSON:
                response.payload_msg = json.loads(payload.decode('utf-8'))
        except Exception as e:
            logger.error(f"解析JSON失败: {e}")
            
        return response

# 核心客户端类
class AsrWsClient:
    def __init__(self, url: str, segment_duration: int = 200, output_file: str = None):
        self.seq = 1
        self.url = url
        self.segment_duration = segment_duration
        self.conn = None
        self.session = None
        self.output_file = output_file
        self.full_recognized_text = ""  # 存储完整的识别文本
        self.response_parser = ResponseParser()

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc, tb):
        if self.conn and not self.conn.closed:
            await self.conn.close()
        if self.session and not self.session.closed:
            await self.session.close()
        
    async def read_audio_data(self, file_path: str) -> bytes:
        """读取并转换音频为PCM（安全版本）"""
        logger.info(f"正在转换音频文件 {file_path} 为PCM格式（不会修改源文件）...")
        pcm_data = CommonUtils.convert_audio_to_pcm(file_path, DEFAULT_SAMPLE_RATE)
        logger.info(f"PCM转换完成，大小: {len(pcm_data)} 字节")
        return pcm_data
            
    def get_segment_size(self) -> int:
        """计算PCM分段大小"""
        bytes_per_sample = 2  # 16bit
        samples_per_ms = DEFAULT_SAMPLE_RATE / 1000
        samples_per_segment = samples_per_ms * self.segment_duration
        segment_size = int(samples_per_segment * bytes_per_sample)
        
        # 确保是2的倍数
        if segment_size % 2 != 0:
            segment_size += 1
            
        logger.info(f"计算分段大小: {segment_size} 字节 (时长: {self.segment_duration}ms)")
        return segment_size
            
    async def create_connection(self) -> None:
        """创建WebSocket连接"""
        headers = RequestBuilder.new_auth_headers()
        self.conn = await self.session.ws_connect(self.url, headers=headers)
        logger.info(f"已连接到 {self.url}")
            
    async def send_full_client_request(self) -> None:
        """发送初始请求"""
        request = RequestBuilder.new_full_client_request(self.seq)
        self.seq += 1
        await self.conn.send_bytes(request)
        logger.info(f"发送初始请求，序列号: {self.seq-1}")
        
        # 接收初始响应
        msg = await self.conn.receive()
        if msg.type == aiohttp.WSMsgType.BINARY:
            response = self.response_parser.parse_response(msg.data)
            logger.info(f"收到初始响应: {response.to_dict()}")
            
    async def send_audio_segments(self, content: bytes) -> None:
        """发送音频分段"""
        segment_size = self.get_segment_size()
        audio_segments = []
        for i in range(0, len(content), segment_size):
            end = min(i + segment_size, len(content))
            audio_segments.append(content[i:end])
        
        total_segments = len(audio_segments)
        logger.info(f"音频分为 {total_segments} 个分段发送")
        
        for i, segment in enumerate(audio_segments):
            is_last = (i == total_segments - 1)
            request = RequestBuilder.new_audio_only_request(self.seq, segment, is_last)
            await self.conn.send_bytes(request)
            logger.debug(f"发送音频分段 {i+1}/{total_segments} (最后一段: {is_last})")
            
            if not is_last:
                self.seq += 1
                
            await asyncio.sleep(self.segment_duration / 1000)
            
    async def recv_messages(self) -> None:
        """接收并处理识别结果"""
        try:
            async for msg in self.conn:
                if msg.type == aiohttp.WSMsgType.BINARY:
                    response = self.response_parser.parse_response(msg.data)
                    
                    # 提取增量文本（避免重复）
                    incremental_text = response.get_incremental_text()
                    if incremental_text:
                        self.full_recognized_text += incremental_text
                        logger.debug(f"新增识别文本: {incremental_text[:50]}...")
                    
                    # 最后一包或出错时保存文件
                    if response.is_last_package or response.code != 0:
                        if self.output_file:
                            self.save_recognized_text()
                        break
                        
                elif msg.type in [aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED]:
                    logger.error(f"WebSocket连接异常: {msg.type}")
                    break
                    
        except Exception as e:
            logger.error(f"接收消息出错: {e}")
            raise
            
    def save_recognized_text(self):
        """保存识别结果到文件（仅写入输出文件，不操作源文件）"""
        try:
            # 创建输出目录（如果不存在）
            output_dir = os.path.dirname(self.output_file)
            if output_dir and not os.path.exists(output_dir):
                os.makedirs(output_dir, exist_ok=True)
            
            # 写入识别结果（仅写输出文件）
            with open(self.output_file, 'w', encoding=DEFAULT_OUTPUT_ENCODING) as f:
                f.write(self.full_recognized_text)
            
            logger.info(f"识别结果已保存到 {self.output_file}")
            logger.info(f"总识别文本长度: {len(self.full_recognized_text)} 字符")
            logger.info(f"源音频文件保持完整，未被修改或删除")
            
        except Exception as e:
            logger.error(f"保存文件失败: {e}")
            raise
        
    async def execute(self, file_path: str) -> None:
        """执行完整的ASR识别流程（安全版本）"""
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"音频文件不存在: {file_path}")
            
        # 重置状态
        self.seq = 1
        self.full_recognized_text = ""
        
        try:
            # 1. 转换音频（只读不写）
            content = await self.read_audio_data(file_path)
            
            # 2. 创建连接
            await self.create_connection()
            
            # 3. 发送初始请求
            await self.send_full_client_request()
            
            # 4. 并发发送音频和接收结果
            send_task = asyncio.create_task(self.send_audio_segments(content))
            recv_task = asyncio.create_task(self.recv_messages())
            
            await asyncio.gather(send_task, recv_task)
                
        finally:
            if self.conn:
                await self.conn.close()
            # 最终确认源文件存在
            if os.path.exists(file_path):
                logger.info(f"✅ 源文件 {file_path} 安全保留")
            else:
                logger.warning(f"⚠️ 源文件 {file_path} 未找到（可能之前已被错误删除）")

async def main():
    import argparse
    
    parser = argparse.ArgumentParser(description="ASR WebSocket客户端（安全版，支持MP3直接处理）")
    parser.add_argument("--file", type=str, required=True, help="音频文件路径（MP3/WAV）")
    parser.add_argument("--output", type=str, help="输出文本文件路径")
    parser.add_argument("--url", type=str, default="wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream", 
                       help="WebSocket URL")
    parser.add_argument("--seg-duration", type=int, default=200, 
                       help="每包音频时长(ms)，默认200")
    
    args = parser.parse_args()
    
    # 验证源文件存在
    if not os.path.exists(args.file):
        logger.error(f"错误：指定的音频文件不存在 - {args.file}")
        return
    
    # 自动生成输出文件名
    if not args.output:
        base_name = os.path.splitext(os.path.basename(args.file))[0]
        args.output = f"{base_name}_asr_result.txt"
    
    # 执行识别（安全模式）
    async with AsrWsClient(args.url, args.seg_duration, args.output) as client:
        try:
            await client.execute(args.file)
            logger.info("✅ ASR识别完成，所有源文件均未被修改或删除")
        except Exception as e:
            logger.error(f"ASR处理失败: {e}")
            # 即使出错也检查源文件状态
            if os.path.exists(args.file):
                logger.info(f"✅ 源文件 {args.file} 仍存在")
            raise

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

import asyncio
import aiohttp
import json
import struct
import gzip
import uuid
import logging
import os
import subprocess
from typing import Optional, List, Dict, Any, Tuple, AsyncGenerator
from datetime import datetime

# 配置日志 - 调整为显示DEBUG级别，确保进度信息可见
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('run.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 常量定义
DEFAULT_SAMPLE_RATE = 16000
DEFAULT_OUTPUT_ENCODING = 'utf-8'

# 协议相关常量
ProtocolVersion = type('ProtocolVersion', (), {'V1': 0b0001})
MessageType = type('MessageType', (), {
    'CLIENT_FULL_REQUEST': 0b0001,
    'CLIENT_AUDIO_ONLY_REQUEST': 0b0010,
    'SERVER_FULL_RESPONSE': 0b1001,
    'SERVER_ERROR_RESPONSE': 0b1111
})
MessageTypeSpecificFlags = type('MessageTypeSpecificFlags', (), {
    'NO_SEQUENCE': 0b0000,
    'POS_SEQUENCE': 0b0001,
    'NEG_SEQUENCE': 0b0010,
    'NEG_WITH_SEQUENCE': 0b0011
})
SerializationType = type('SerializationType', (), {
    'NO_SERIALIZATION': 0b0000,
    'JSON': 0b0001
})
CompressionType = type('CompressionType', (), {'GZIP': 0b0001})

# 配置类
class Config:
    def __init__(self):
        self.auth = {
            ",
            "
        }

    @property
    def app_key(self) -> str:
        return self.auth["app_key"]

    @property
    def access_key(self) -> str:
        return self.auth["access_key"]

config = Config()

# 通用工具类
class CommonUtils:
    @staticmethod
    def gzip_compress(data: bytes) -> bytes:
        return gzip.compress(data)

    @staticmethod
    def gzip_decompress(data: bytes) -> bytes:
        return gzip.decompress(data)

    @staticmethod
    def convert_audio_to_pcm(audio_path: str, sample_rate: int = DEFAULT_SAMPLE_RATE) -> bytes:
        """
        直接将音频文件（MP3/WAV等）转换为原始PCM数据
        ✅ 安全版本：完全不操作源文件，仅读取转换
        """
        if not os.path.exists(audio_path):
            raise FileNotFoundError(f"音频文件不存在: {audio_path}")
            
        try:
            cmd = [
                "ffmpeg", "-v", "quiet", "-y", "-i", audio_path,
                "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(sample_rate),
                "-f", "s16le", "-"  # 输出到stdout，不生成任何文件
            ]
            # 执行ffmpeg转换，仅读取源文件，不修改/删除
            result = subprocess.run(
                cmd, 
                check=True, 
                stdout=subprocess.PIPE, 
                stderr=subprocess.PIPE
            )
            logger.info(f"音频转换完成，源文件 {audio_path} 保持完整")
            return result.stdout
            
        except subprocess.CalledProcessError as e:
            logger.error(f"FFmpeg转换失败: {e.stderr.decode()}")
            raise RuntimeError(f"音频转换失败: {e.stderr.decode()}")

# 请求头构建类
class AsrRequestHeader:
    def __init__(self):
        self.message_type = MessageType.CLIENT_FULL_REQUEST
        self.message_type_specific_flags = MessageTypeSpecificFlags.POS_SEQUENCE
        self.serialization_type = SerializationType.JSON
        self.compression_type = CompressionType.GZIP
        self.reserved_data = bytes([0x00])

    def to_bytes(self) -> bytes:
        header = bytearray()
        header.append((ProtocolVersion.V1 << 4) | 1)
        header.append((self.message_type << 4) | self.message_type_specific_flags)
        header.append((self.serialization_type << 4) | self.compression_type)
        header.extend(self.reserved_data)
        return bytes(header)

    @staticmethod
    def default_header() -> 'AsrRequestHeader':
        return AsrRequestHeader()

# 请求构建器
class RequestBuilder:
    @staticmethod
    def new_auth_headers() -> Dict[str, str]:
        return {
            "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
            "X-Api-Request-Id": str(uuid.uuid4()),
            "X-Api-Access-Key": config.access_key,
            "X-Api-App-Key": config.app_key
        }

    @staticmethod
    def new_full_client_request(seq: int) -> bytes:
        header = AsrRequestHeader.default_header()
        payload = {
            "user": {"uid": "demo_uid"},
            "audio": {"format": "pcm", "codec": "raw", "rate": 16000, "bits": 16, "channel": 1},
            "request": {
                "model_name": "bigmodel",
                "enable_itn": True,
                "enable_punc": True,
                "enable_ddc": True,
                "show_utterances": True,
                "enable_nonstream": False
            }
        }
        
        payload_bytes = json.dumps(payload).encode('utf-8')
        compressed_payload = CommonUtils.gzip_compress(payload_bytes)
        
        request = bytearray()
        request.extend(header.to_bytes())
        request.extend(struct.pack('>i', seq))
        request.extend(struct.pack('>I', len(compressed_payload)))
        request.extend(compressed_payload)
        
        return bytes(request)

    @staticmethod
    def new_audio_only_request(seq: int, segment: bytes, is_last: bool = False) -> bytes:
        header = AsrRequestHeader.default_header()
        header.message_type = MessageType.CLIENT_AUDIO_ONLY_REQUEST
        
        if is_last:
            header.message_type_specific_flags = MessageTypeSpecificFlags.NEG_WITH_SEQUENCE
            seq = -seq
        else:
            header.message_type_specific_flags = MessageTypeSpecificFlags.POS_SEQUENCE
        
        compressed_segment = CommonUtils.gzip_compress(segment)
        
        request = bytearray()
        request.extend(header.to_bytes())
        request.extend(struct.pack('>i', seq))
        request.extend(struct.pack('>I', len(compressed_segment)))
        request.extend(compressed_segment)
        
        return bytes(request)

# 响应处理类
class AsrResponse:
    def __init__(self):
        self.code = 0
        self.event = 0
        self.is_last_package = False
        self.payload_sequence = 0
        self.payload_size = 0
        self.payload_msg = None
        self.last_text = ""  # 记录上一次的完整文本，用于去重

    def to_dict(self) -> Dict[str, Any]:
        return {
            "code": self.code,
            "event": self.event,
            "is_last_package": self.is_last_package,
            "payload_sequence": self.payload_sequence,
            "payload_size": self.payload_size,
            "payload_msg": self.payload_msg
        }
    
    def get_incremental_text(self) -> str:
        """提取增量文本（解决重复问题）"""
        current_text = ""
        if self.payload_msg and isinstance(self.payload_msg, dict):
            if "result" in self.payload_msg:
                result = self.payload_msg["result"]
                if isinstance(result, dict):
                    # 优先获取完整文本
                    if "text" in result:
                        current_text = result["text"]
                    # 分句文本拼接
                    elif "utterances" in result and isinstance(result["utterances"], list):
                        utterances = []
                        for utt in result["utterances"]:
                            if isinstance(utt, dict) and "text" in utt:
                                utterances.append(utt["text"])
                        current_text = "".join(utterances)
        
        # 计算增量文本（只返回新增部分）
        incremental_text = current_text[len(self.last_text):]
        self.last_text = current_text  # 更新上一次的文本
        
        return incremental_text

class ResponseParser:
    @staticmethod
    def parse_response(msg: bytes) -> AsrResponse:
        response = AsrResponse()
        
        header_size = msg[0] & 0x0f
        message_type = msg[1] >> 4
        message_type_specific_flags = msg[1] & 0x0f
        serialization_method = msg[2] >> 4
        message_compression = msg[2] & 0x0f
        
        payload = msg[header_size*4:]
        
        # 解析标志位
        if message_type_specific_flags & 0x01:
            response.payload_sequence = struct.unpack('>i', payload[:4])[0]
            payload = payload[4:]
        if message_type_specific_flags & 0x02:
            response.is_last_package = True
        if message_type_specific_flags & 0x04:
            response.event = struct.unpack('>i', payload[:4])[0]
            payload = payload[4:]
            
        # 解析消息类型
        if message_type == MessageType.SERVER_FULL_RESPONSE:
            response.payload_size = struct.unpack('>I', payload[:4])[0]
            payload = payload[4:]
        elif message_type == MessageType.SERVER_ERROR_RESPONSE:
            response.code = struct.unpack('>i', payload[:4])[0]
            response.payload_size = struct.unpack('>I', payload[4:8])[0]
            payload = payload[8:]
            
        if not payload:
            return response
            
        # 解压缩
        if message_compression == CompressionType.GZIP:
            try:
                payload = CommonUtils.gzip_decompress(payload)
            except Exception as e:
                logger.error(f"解压缩失败: {e}")
                return response
                
        # 解析JSON
        try:
            if serialization_method == SerializationType.JSON:
                response.payload_msg = json.loads(payload.decode('utf-8'))
        except Exception as e:
            logger.error(f"解析JSON失败: {e}")
            
        return response

# 核心客户端类
class AsrWsClient:
    def __init__(self, url: str, segment_duration: int = 200, output_file: str = None):
        self.seq = 1
        self.url = url
        self.segment_duration = segment_duration
        self.conn = None
        self.session = None
        self.output_file = output_file
        self.full_recognized_text = ""  # 存储完整的识别文本
        self.response_parser = ResponseParser()
        self.start_time = None  # 记录开始时间
        self.processed_segments = 0  # 已处理的分段数
        self.total_segments = 0  # 总分段数

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc, tb):
        if self.conn and not self.conn.closed:
            await self.conn.close()
        if self.session and not self.session.closed:
            await self.session.close()
        
    async def read_audio_data(self, file_path: str) -> bytes:
        """读取并转换音频为PCM（安全版本）"""
        logger.info(f"正在转换音频文件 {file_path} 为PCM格式（不会修改源文件）...")
        pcm_data = CommonUtils.convert_audio_to_pcm(file_path, DEFAULT_SAMPLE_RATE)
        logger.info(f"PCM转换完成，大小: {len(pcm_data)} 字节")
        
        # 计算音频时长
        audio_duration = len(pcm_data) / (2 * DEFAULT_SAMPLE_RATE)  # 2字节/采样
        logger.info(f"音频总时长: {audio_duration:.2f} 秒 ({audio_duration/60:.2f} 分钟)")
        
        return pcm_data
            
    def get_segment_size(self) -> int:
        """计算PCM分段大小"""
        bytes_per_sample = 2  # 16bit
        samples_per_ms = DEFAULT_SAMPLE_RATE / 1000
        samples_per_segment = samples_per_ms * self.segment_duration
        segment_size = int(samples_per_segment * bytes_per_sample)
        
        # 确保是2的倍数
        if segment_size % 2 != 0:
            segment_size += 1
            
        logger.info(f"计算分段大小: {segment_size} 字节 (时长: {self.segment_duration}ms)")
        return segment_size
            
    async def create_connection(self) -> None:
        """创建WebSocket连接"""
        headers = RequestBuilder.new_auth_headers()
        self.conn = await self.session.ws_connect(self.url, headers=headers)
        logger.info(f"已连接到 {self.url}")
        self.start_time = datetime.now()  # 记录开始时间
            
    async def send_full_client_request(self) -> None:
        """发送初始请求"""
        request = RequestBuilder.new_full_client_request(self.seq)
        self.seq += 1
        await self.conn.send_bytes(request)
        logger.info(f"发送初始请求，序列号: {self.seq-1}")
        
        # 接收初始响应
        msg = await self.conn.receive()
        if msg.type == aiohttp.WSMsgType.BINARY:
            response = self.response_parser.parse_response(msg.data)
            logger.info(f"收到初始响应: {response.to_dict()}")
            
    async def send_audio_segments(self, content: bytes) -> None:
        """发送音频分段（带详细进度显示）"""
        segment_size = self.get_segment_size()
        audio_segments = []
        for i in range(0, len(content), segment_size):
            end = min(i + segment_size, len(content))
            audio_segments.append(content[i:end])
        
        self.total_segments = len(audio_segments)
        logger.info(f"音频分为 {self.total_segments} 个分段发送 (预计耗时: {self.total_segments * self.segment_duration / 1000:.2f} 秒)")
        
        # 每10个分段显示一次进度（避免日志刷屏）
        progress_interval = max(1, self.total_segments // 50)  # 最多显示50次进度
        
        for i, segment in enumerate(audio_segments):
            self.processed_segments = i + 1
            is_last = (i == self.total_segments - 1)
            
            # 发送音频分段
            request = RequestBuilder.new_audio_only_request(self.seq, segment, is_last)
            await self.conn.send_bytes(request)
            
            # 显示进度（按间隔显示，避免刷屏）
            if i % progress_interval == 0 or is_last:
                progress = (self.processed_segments / self.total_segments) * 100
                elapsed_time = (datetime.now() - self.start_time).total_seconds()
                avg_speed = self.processed_segments / elapsed_time if elapsed_time > 0 else 0
                remaining_time = (self.total_segments - self.processed_segments) / avg_speed if avg_speed > 0 else 0
                
                logger.info(
                    f"发送进度: {self.processed_segments}/{self.total_segments} "
                    f"({progress:.1f}%) | "
                    f"已用时: {elapsed_time:.1f}s | "
                    f"剩余: {remaining_time:.1f}s | "
                    f"速度: {avg_speed:.1f} 段/秒"
                )
            
            if not is_last:
                self.seq += 1
                
            await asyncio.sleep(self.segment_duration / 1000)
        
        logger.info(f"✅ 所有音频分段发送完成！总计发送 {self.processed_segments} 个分段")
            
    async def recv_messages(self) -> None:
        """接收并处理识别结果（带实时文本预览）"""
        recognized_chars = 0
        last_progress_update = 0
        
        try:
            async for msg in self.conn:
                if msg.type == aiohttp.WSMsgType.BINARY:
                    response = self.response_parser.parse_response(msg.data)
                    
                    # 提取增量文本（避免重复）
                    incremental_text = response.get_incremental_text()
                    if incremental_text:
                        self.full_recognized_text += incremental_text
                        recognized_chars += len(incremental_text)
                        
                        # 每识别100个字符显示一次预览
                        if recognized_chars - last_progress_update >= 100 or response.is_last_package:
                            last_progress_update = recognized_chars
                            # 显示最新识别的文本（最后50个字符）
                            preview_text = self.full_recognized_text[-50:] if len(self.full_recognized_text) > 50 else self.full_recognized_text
                            logger.info(f"📝 识别进度: {recognized_chars} 字符 | 最新文本: {preview_text}")
                    
                    # 最后一包或出错时保存文件
                    if response.is_last_package or response.code != 0:
                        if self.output_file:
                            self.save_recognized_text()
                        break
                        
                elif msg.type in [aiohttp.WSMsgType.ERROR, aiohttp.WSMsgType.CLOSED]:
                    logger.error(f"WebSocket连接异常: {msg.type}")
                    break
                    
        except Exception as e:
            logger.error(f"接收消息出错: {e}")
            raise
            
    def save_recognized_text(self):
        """保存识别结果到文件（仅写入输出文件，不操作源文件）"""
        try:
            # 创建输出目录（如果不存在）
            output_dir = os.path.dirname(self.output_file)
            if output_dir and not os.path.exists(output_dir):
                os.makedirs(output_dir, exist_ok=True)
            
            # 写入识别结果（仅写输出文件）
            with open(self.output_file, 'w', encoding=DEFAULT_OUTPUT_ENCODING) as f:
                f.write(self.full_recognized_text)
            
            # 计算总耗时
            total_time = (datetime.now() - self.start_time).total_seconds() if self.start_time else 0
            
            logger.info("=" * 80)
            logger.info(f"🎉 识别完成！结果已保存到 {self.output_file}")
            logger.info(f"📊 统计信息:")
            logger.info(f"   - 总识别文本长度: {len(self.full_recognized_text)} 字符")
            logger.info(f"   - 总处理时间: {total_time:.2f} 秒")
            logger.info(f"   - 平均速度: {len(self.full_recognized_text)/total_time:.1f} 字符/秒" if total_time > 0 else "   - 平均速度: N/A")
            logger.info(f"✅ 源音频文件保持完整，未被修改或删除")
            logger.info("=" * 80)
            
        except Exception as e:
            logger.error(f"保存文件失败: {e}")
            raise
        
    async def execute(self, file_path: str) -> None:
        """执行完整的ASR识别流程（安全版本）"""
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"音频文件不存在: {file_path}")
            
        # 重置状态
        self.seq = 1
        self.full_recognized_text = ""
        self.processed_segments = 0
        self.total_segments = 0
        
        try:
            # 1. 转换音频（只读不写）
            content = await self.read_audio_data(file_path)
            
            # 2. 创建连接
            await self.create_connection()
            
            # 3. 发送初始请求
            await self.send_full_client_request()
            
            # 4. 并发发送音频和接收结果
            send_task = asyncio.create_task(self.send_audio_segments(content))
            recv_task = asyncio.create_task(self.recv_messages())
            
            await asyncio.gather(send_task, recv_task)
                
        finally:
            if self.conn:
                await self.conn.close()
            # 最终确认源文件存在
            if os.path.exists(file_path):
                logger.info(f"✅ 源文件 {file_path} 安全保留")
            else:
                logger.warning(f"⚠️ 源文件 {file_path} 未找到（可能之前已被错误删除）")

async def main():
    import argparse
    
    parser = argparse.ArgumentParser(description="ASR WebSocket客户端（安全版，支持MP3直接处理）")
    parser.add_argument("--file", type=str, required=True, help="音频文件路径（MP3/WAV）")
    parser.add_argument("--output", type=str, help="输出文本文件路径")
    parser.add_argument("--url", type=str, default="wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream", 
                       help="WebSocket URL")
    parser.add_argument("--seg-duration", type=int, default=200, 
                       help="每包音频时长(ms)，默认200")
    
    args = parser.parse_args()
    
    # 验证源文件存在
    if not os.path.exists(args.file):
        logger.error(f"错误：指定的音频文件不存在 - {args.file}")
        return
    
    # 自动生成输出文件名
    if not args.output:
        base_name = os.path.splitext(os.path.basename(args.file))[0]
        args.output = f"{base_name}_asr_result.txt"
    
    # 执行识别（安全模式）
    async with AsrWsClient(args.url, args.seg_duration, args.output) as client:
        try:
            await client.execute(args.file)
            logger.info("\n🎯 ASR识别流程全部完成！")
        except Exception as e:
            logger.error(f"\n❌ ASR处理失败: {e}")
            # 即使出错也检查源文件状态
            if os.path.exists(args.file):
                logger.info(f"✅ 源文件 {args.file} 仍存在")
            raise

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

import asyncio
import aiohttp
import json
import struct
import gzip
import uuid
import logging
import os
import subprocess
from datetime import datetime

# 基础配置 - 简洁实用
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[logging.FileHandler('run.log'), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

# 核心常量
DEFAULT_SAMPLE_RATE = 16000
CONFIG = {
    ",
    ",
    "ws_url": "wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream"
}

# 协议常量
MSG_TYPE = {
    'CLIENT_FULL': 0b0001,
    'CLIENT_AUDIO': 0b0010,
    'SERVER_FULL': 0b1001,
    'SERVER_ERROR': 0b1111
}
FLAGS = {
    'POS_SEQ': 0b0001,
    'NEG_SEQ': 0b0011,
    'LAST_PACKAGE': 0b0010
}

def convert_audio_to_pcm(audio_path: str) -> bytes:
    """仅转换音频为PCM，不碰源文件"""
    if not os.path.exists(audio_path):
        raise FileNotFoundError(f"文件不存在: {audio_path}")
    
    cmd = [
        "ffmpeg", "-v", "quiet", "-i", audio_path,
        "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(DEFAULT_SAMPLE_RATE),
        "-f", "s16le", "-"
    ]
    result = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    logger.info(f"✅ 音频转换完成，源文件 {audio_path} 完好")
    return result.stdout

def build_request_header(msg_type: int, is_last: bool = False) -> bytes:
    """构建基础请求头"""
    header = bytearray([0x11, 0x00, 0x11, 0x00])
    header[1] = (msg_type << 4) | (FLAGS['POS_SEQ'] if not is_last else FLAGS['NEG_SEQ'])
    return bytes(header)

def build_init_request(seq: int) -> bytes:
    """构建初始请求"""
    payload = json.dumps({
        "user": {"uid": "demo_uid"},
        "audio": {"format": "pcm", "codec": "raw", "rate": 16000, "bits": 16, "channel": 1},
        "request": {
            "model_name": "bigmodel",
            "enable_itn": True,
            "enable_punc": True,
            "enable_ddc": True,
            "enable_nonstream": False
        }
    }).encode()
    
    compressed = gzip.compress(payload)
    request = bytearray()
    request.extend(build_request_header(MSG_TYPE['CLIENT_FULL']))
    request.extend(struct.pack('>i', seq))
    request.extend(struct.pack('>I', len(compressed)))
    request.extend(compressed)
    return bytes(request)

def build_audio_request(seq: int, audio_data: bytes, is_last: bool = False) -> bytes:
    """构建音频数据请求"""
    compressed = gzip.compress(audio_data)
    request = bytearray()
    request.extend(build_request_header(MSG_TYPE['CLIENT_AUDIO'], is_last))
    request.extend(struct.pack('>i', seq if not is_last else -seq))
    request.extend(struct.pack('>I', len(compressed)))
    request.extend(compressed)
    return bytes(request)

def parse_response(data: bytes) -> tuple[str, bool]:
    """修复版响应解析 - 核心修复点"""
    try:
        # 正确解析头部标志位
        is_last = bool(data[1] & FLAGS['LAST_PACKAGE'])
        
        # 正确的payload截取逻辑（修复核心错误）
        header_size = (data[0] & 0x0F) * 4  # 头部长度 = 低4位 * 4
        payload = data[header_size:]
        
        # 跳过sequence字段（4字节）
        if len(payload) > 4:
            payload = payload[4:]
            
            # 解压缩（仅当数据是gzip格式时）
            try:
                # 检查是否是gzip数据（避免解压错误）
                if payload[:2] == b'\x1f\x8b':
                    payload = gzip.decompress(payload)
                resp = json.loads(payload.decode('utf-8'))
                final_text = resp.get('result', {}).get('text', '')
                return final_text, is_last
            except gzip.BadGzipFile:
                # 非gzip数据直接解析
                try:
                    resp = json.loads(payload.decode('utf-8'))
                    final_text = resp.get('result', {}).get('text', '')
                    return final_text, is_last
                except:
                    pass
    except Exception as e:
        # 只记录关键错误，避免刷屏
        if "Not a gzipped file" not in str(e):
            logger.error(f"解析响应出错: {e}")
    
    return "", is_last

async def main(file_path: str, output_file: str):
    # 1. 基础检查
    if not os.path.exists(file_path):
        logger.error(f"❌ 源文件不存在: {file_path}")
        return
    
    # 2. 转换音频
    logger.info(f"📥 处理音频文件: {file_path}")
    pcm_data = convert_audio_to_pcm(file_path)
    audio_len = len(pcm_data)
    logger.info(f"🎵 音频大小: {audio_len} 字节, 时长: {audio_len/(2*16000):.2f}秒")
    
    # 3. 准备分段
    segment_size = 6400  # 200ms per segment
    segments = [pcm_data[i:i+segment_size] for i in range(0, audio_len, segment_size)]
    total_segs = len(segments)
    logger.info(f"📦 分为 {total_segs} 个音频段发送")
    
    # 4. 创建WS连接
    headers = {
        "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
        "X-Api-Request-Id": str(uuid.uuid4()),
        "X-Api-Access-Key": CONFIG['access_key'],
        "X-Api-App-Key": CONFIG['app_key']
    }
    
    async with aiohttp.ClientSession() as session:
        async with session.ws_connect(CONFIG['ws_url'], headers=headers) as ws:
            logger.info(f"🔌 已连接到ASR服务")
            
            # 5. 发送初始请求
            await ws.send_bytes(build_init_request(1))
            await ws.receive()  # 忽略初始响应
            
            # 6. 发送音频段（带进度）
            final_text = ""
            seq = 2
            start_time = datetime.now()
            
            for i, seg in enumerate(segments):
                is_last = (i == total_segs - 1)
                await ws.send_bytes(build_audio_request(seq, seg, is_last))
                
                # 显示进度（每5%显示一次，避免刷屏）
                progress_percent = (i+1)/total_segs*100
                if i % max(1, total_segs//20) == 0 or is_last:
                    elapsed = (datetime.now()-start_time).total_seconds()
                    logger.info(f"📤 发送进度: {i+1}/{total_segs} ({progress_percent:.1f}%) | 已用时: {elapsed:.1f}s")
                
                # 接收响应（只保留最后一次完整文本）
                try:
                    msg = await asyncio.wait_for(ws.receive(), timeout=5.0)
                    if msg.type == aiohttp.WSMsgType.BINARY:
                        text, is_last_resp = parse_response(msg.data)
                        if text:  # 只更新有内容的文本
                            final_text = text
                        if is_last_resp:
                            break
                except asyncio.TimeoutError:
                    logger.warning(f"⚠️ 接收响应超时，继续发送下一段")
                
                seq += 1
                await asyncio.sleep(0.001)
            
            # 7. 写入最终结果
            if final_text:
                os.makedirs(os.path.dirname(output_file) or ".", exist_ok=True)
                with open(output_file, 'w', encoding='utf-8') as f:
                    f.write(final_text)
                
                logger.info("="*60)
                logger.info(f"✅ 识别完成！结果已写入: {output_file}")
                logger.info(f"📝 文本长度: {len(final_text)} 字符")
                logger.info(f"🔍 文本预览: {final_text[:100]}...")
                logger.info("="*60)
            else:
                logger.error("❌ 未获取到识别文本")
            
            logger.info(f"✅ 源文件 {file_path} 完整保留，未被修改")

if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="ASR客户端 - 修复版")
    parser.add_argument("--file", required=True, help="音频文件路径(MP3/WAV)")
    parser.add_argument("--output", help="输出文件路径", default="asr_result.txt")
    args = parser.parse_args()
    
    if args.output == "asr_result.txt" and args.file:
        base = os.path.splitext(os.path.basename(args.file))[0]
        args.output = f"{base}_asr_result.txt"
    
    try:
        asyncio.run(main(args.file, args.output))
    except Exception as e:
        logger.error(f"程序异常: {e}")
        if os.path.exists(args.file):
            logger.info(f"✅ 源文件 {args.file} 依然存在")

import asyncio
import aiohttp
import json
import struct
import gzip
import uuid
import logging
import os
import subprocess
from typing import Optional, List, Dict, Any, Tuple, AsyncGenerator

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('run.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

# 常量定义
DEFAULT_SAMPLE_RATE = 16000
DEFAULT_OUTPUT_ENCODING = 'utf-8'

class ProtocolVersion:
    V1 = 0b0001

class MessageType:
    CLIENT_FULL_REQUEST = 0b0001
    CLIENT_AUDIO_ONLY_REQUEST = 0b0010
    SERVER_FULL_RESPONSE = 0b1001
    SERVER_ERROR_RESPONSE = 0b1111

class MessageTypeSpecificFlags:
    NO_SEQUENCE = 0b0000
    POS_SEQUENCE = 0b0001
    NEG_SEQUENCE = 0b0010
    NEG_WITH_SEQUENCE = 0b0011

class SerializationType:
    NO_SERIALIZATION = 0b0000
    JSON = 0b0001

class CompressionType:
    GZIP = 0b0001


class Config:
    def __init__(self):
        # 填入控制台获取的app id和access token
        self.auth = {
            ",
            "
        }

    @property
    def app_key(self) -> str:
        return self.auth["app_key"]

    @property
    def access_key(self) -> str:
        return self.auth["access_key"]

config = Config()

class CommonUtils:
    @staticmethod
    def gzip_compress(data: bytes) -> bytes:
        return gzip.compress(data)

    @staticmethod
    def gzip_decompress(data: bytes) -> bytes:
        return gzip.decompress(data)

    @staticmethod
    def judge_wav(data: bytes) -> bool:
        if len(data) < 44:
            return False
        return data[:4] == b'RIFF' and data[8:12] == b'WAVE'

    @staticmethod
    def convert_audio_to_pcm(audio_path: str, sample_rate: int = DEFAULT_SAMPLE_RATE) -> bytes:
        """
        直接将音频文件（支持MP3/WAV等）转换为原始PCM数据
        避免生成中间WAV文件，提高效率
        """
        try:
            cmd = [
                "ffmpeg", "-v", "quiet", "-y", "-i", audio_path,
                "-acodec", "pcm_s16le", "-ac", "1", "-ar", str(sample_rate),
                "-f", "s16le", "-"  # 直接输出原始PCM，无WAV头
            ]
            result = subprocess.run(cmd, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
            
            # 尝试删除原始文件（可选）
            try:
                os.remove(audio_path)
            except OSError as e:
                logger.warning(f"Failed to remove original file: {e}")
                
            return result.stdout
        except subprocess.CalledProcessError as e:
            logger.error(f"FFmpeg conversion failed: {e.stderr.decode()}")
            raise RuntimeError(f"Audio conversion failed: {e.stderr.decode()}")

    @staticmethod
    def read_wav_info(data: bytes) -> Tuple[int, int, int, int, bytes]:
        if len(data) < 44:
            raise ValueError("Invalid WAV file: too short")
            
        # 解析WAV头
        chunk_id = data[:4]
        if chunk_id != b'RIFF':
            raise ValueError("Invalid WAV file: not RIFF format")
            
        format_ = data[8:12]
        if format_ != b'WAVE':
            raise ValueError("Invalid WAV file: not WAVE format")
            
        # 解析fmt子块
        audio_format = struct.unpack('<H', data[20:22])[0]
        num_channels = struct.unpack('<H', data[22:24])[0]
        sample_rate = struct.unpack('<I', data[24:28])[0]
        bits_per_sample = struct.unpack('<H', data[34:36])[0]
        
        # 查找data子块
        pos = 36
        while pos < len(data) - 8:
            subchunk_id = data[pos:pos+4]
            subchunk_size = struct.unpack('<I', data[pos+4:pos+8])[0]
            if subchunk_id == b'data':
                wave_data = data[pos+8:pos+8+subchunk_size]
                return (
                    num_channels,
                    bits_per_sample // 8,
                    sample_rate,
                    subchunk_size // (num_channels * (bits_per_sample // 8)),
                    wave_data
                )
            pos += 8 + subchunk_size
            
        raise ValueError("Invalid WAV file: no data subchunk found")

class AsrRequestHeader:
    def __init__(self):
        self.message_type = MessageType.CLIENT_FULL_REQUEST
        self.message_type_specific_flags = MessageTypeSpecificFlags.POS_SEQUENCE
        self.serialization_type = SerializationType.JSON
        self.compression_type = CompressionType.GZIP
        self.reserved_data = bytes([0x00])

    def with_message_type(self, message_type: int) -> 'AsrRequestHeader':
        self.message_type = message_type
        return self

    def with_message_type_specific_flags(self, flags: int) -> 'AsrRequestHeader':
        self.message_type_specific_flags = flags
        return self

    def with_serialization_type(self, serialization_type: int) -> 'AsrRequestHeader':
        self.serialization_type = serialization_type
        return self

    def with_compression_type(self, compression_type: int) -> 'AsrRequestHeader':
        self.compression_type = compression_type
        return self

    def with_reserved_data(self, reserved_data: bytes) -> 'AsrRequestHeader':
        self.reserved_data = reserved_data
        return self

    def to_bytes(self) -> bytes:
        header = bytearray()
        header.append((ProtocolVersion.V1 << 4) | 1)
        header.append((self.message_type << 4) | self.message_type_specific_flags)
        header.append((self.serialization_type << 4) | self.compression_type)
        header.extend(self.reserved_data)
        return bytes(header)

    @staticmethod
    def default_header() -> 'AsrRequestHeader':
        return AsrRequestHeader()

class RequestBuilder:
    @staticmethod
    def new_auth_headers() -> Dict[str, str]:
        reqid = str(uuid.uuid4())
        return {
            "X-Api-Resource-Id": "volc.bigasr.sauc.duration",
            "X-Api-Request-Id": reqid,
            "X-Api-Access-Key": config.access_key,
            "X-Api-App-Key": config.app_key
        }

    @staticmethod
    def new_full_client_request(seq: int) -> bytes:
        header = AsrRequestHeader.default_header() \
            .with_message_type_specific_flags(MessageTypeSpecificFlags.POS_SEQUENCE)
        
        payload = {
            "user": {
                "uid": "demo_uid"
            },
            "audio": {
                "format": "pcm",  # 修改为pcm格式
                "codec": "raw",
                "rate": 16000,
                "bits": 16,
                "channel": 1
            },
            "request": {
                "model_name": "bigmodel",
                "enable_itn": True,
                "enable_punc": True,
                "enable_ddc": True,
                "show_utterances": True,
                "enable_nonstream": False
            }
        }
        
        payload_bytes = json.dumps(payload).encode('utf-8')
        compressed_payload = CommonUtils.gzip_compress(payload_bytes)
        payload_size = len(compressed_payload)
        
        request = bytearray()
        request.extend(header.to_bytes())
        request.extend(struct.pack('>i', seq))
        request.extend(struct.pack('>I', payload_size))
        request.extend(compressed_payload)
        
        return bytes(request)

    @staticmethod
    def new_audio_only_request(seq: int, segment: bytes, is_last: bool = False) -> bytes:
        header = AsrRequestHeader.default_header()
        if is_last:
            header.with_message_type_specific_flags(MessageTypeSpecificFlags.NEG_WITH_SEQUENCE)
            seq = -seq
        else:
            header.with_message_type_specific_flags(MessageTypeSpecificFlags.POS_SEQUENCE)
        header.with_message_type(MessageType.CLIENT_AUDIO_ONLY_REQUEST)
        
        request = bytearray()
        request.extend(header.to_bytes())
        request.extend(struct.pack('>i', seq))
        
        compressed_segment = CommonUtils.gzip_compress(segment)
        request.extend(struct.pack('>I', len(compressed_segment)))
        request.extend(compressed_segment)
        
        return bytes(request)

class AsrResponse:
    def __init__(self):
        self.code = 0
        self.event = 0
        self.is_last_package = False
        self.payload_sequence = 0
        self.payload_size = 0
        self.payload_msg = None

    def to_dict(self) -> Dict[str, Any]:
        return {
            "code": self.code,
            "event": self.event,
            "is_last_package": self.is_last_package,
            "payload_sequence": self.payload_sequence,
            "payload_size": self.payload_size,
            "payload_msg": self.payload_msg
        }
    
    def get_recognized_text(self) -> str:
        """提取识别出的文本内容"""
        text = ""
        if self.payload_msg and isinstance(self.payload_msg, dict):
            # 提取不同层级的文本内容
            if "result" in self.payload_msg:
                result = self.payload_msg["result"]
                if isinstance(result, dict):
                    # 提取完整文本
                    if "text" in result:
                        text = result["text"]
                    # 提取分句文本
                    elif "utterances" in result and isinstance(result["utterances"], list):
                        utterances = []
                        for utt in result["utterances"]:
                            if isinstance(utt, dict) and "text" in utt:
                                utterances.append(utt["text"])
                        text = "".join(utterances)
        return text

class ResponseParser:
    @staticmethod
    def parse_response(msg: bytes) -> AsrResponse:
        response = AsrResponse()
        
        header_size = msg[0] & 0x0f
        message_type = msg[1] >> 4
        message_type_specific_flags = msg[1] & 0x0f
        serialization_method = msg[2] >> 4
        message_compression = msg[2] & 0x0f
        
        payload = msg[header_size*4:]
        
        # 解析message_type_specific_flags
        if message_type_specific_flags & 0x01:
            response.payload_sequence = struct.unpack('>i', payload[:4])[0]
            payload = payload[4:]
        if message_type_specific_flags & 0x02:
            response.is_last_package = True
        if message_type_specific_flags & 0x04:
            response.event = struct.unpack('>i', payload[:4])[0]
            payload = payload[4:]
            
        # 解析message_type
        if message_type == MessageType.SERVER_FULL_RESPONSE:
            response.payload_size = struct.unpack('>I', payload[:4])[0]
            payload = payload[4:]
        elif message_type == MessageType.SERVER_ERROR_RESPONSE:
            response.code = struct.unpack('>i', payload[:4])[0]
            response.payload_size = struct.unpack('>I', payload[4:8])[0]
            payload = payload[8:]
            
        if not payload:
            return response
            
        # 解压缩
        if message_compression == CompressionType.GZIP:
            try:
                payload = CommonUtils.gzip_decompress(payload)
            except Exception as e:
                logger.error(f"Failed to decompress payload: {e}")
                return response
                
        # 解析payload
        try:
            if serialization_method == SerializationType.JSON:
                response.payload_msg = json.loads(payload.decode('utf-8'))
        except Exception as e:
            logger.error(f"Failed to parse payload: {e}")
            
        return response

class AsrWsClient:
    def __init__(self, url: str, segment_duration: int = 200, output_file: str = None):
        self.seq = 1
        self.url = url
        self.segment_duration = segment_duration
        self.conn = None
        self.session = None
        self.output_file = output_file  # 识别结果输出文件路径
        self.recognized_text = []  # 存储识别出的文本

    async def __aenter__(self):
        self.session = aiohttp.ClientSession()
        return self
    
    async def __aexit__(self, exc_type, exc, tb):
        if self.conn and not self.conn.closed:
            await self.conn.close()
        if self.session and not self.session.closed:
            await self.session.close()
        
    async def read_audio_data(self, file_path: str) -> bytes:
        """
        读取音频文件并转换为PCM格式
        支持MP3、WAV等格式直接转换
        """
        try:
            # 直接转换为PCM，无需先判断格式
            logger.info(f"Converting audio file {file_path} to PCM format...")
            pcm_data = CommonUtils.convert_audio_to_pcm(file_path, DEFAULT_SAMPLE_RATE)
            logger.info(f"Successfully converted to PCM, size: {len(pcm_data)} bytes")
            return pcm_data
        except Exception as e:
            logger.error(f"Failed to read/convert audio data: {e}")
            raise
            
    def get_segment_size(self, content: bytes) -> int:
        """
        计算PCM数据的分段大小
        基于segment_duration（毫秒）计算
        """
        try:
            # PCM参数：16bit(2字节)、单声道、16000采样率
            bytes_per_sample = 2  # 16bit
            samples_per_ms = DEFAULT_SAMPLE_RATE / 1000
            samples_per_segment = samples_per_ms * self.segment_duration
            segment_size = int(samples_per_segment * bytes_per_sample)
            
            # 确保分段大小是2的倍数（PCM 16bit）
            if segment_size % 2 != 0:
                segment_size += 1
                
            logger.info(f"Calculated segment size: {segment_size} bytes (duration: {self.segment_duration}ms)")
            return segment_size
        except Exception as e:
            logger.error(f"Failed to calculate segment size: {e}")
            raise
            
    async def create_connection(self) -> None:
        headers = RequestBuilder.new_auth_headers()
        try:
            self.conn = await self.session.ws_connect(
                self.url,
                headers=headers
            )
            logger.info(f"Connected to {self.url}")
        except Exception as e:
            logger.error(f"Failed to connect to WebSocket: {e}")
            raise
            
    async def send_full_client_request(self) -> None:
        request = RequestBuilder.new_full_client_request(self.seq)
        self.seq += 1
        try:
            await self.conn.send_bytes(request)
            logger.info(f"Sent full client request with seq: {self.seq-1}")
            
            msg = await self.conn.receive()
            if msg.type == aiohttp.WSMsgType.BINARY:
                response = ResponseParser.parse_response(msg.data)
                logger.info(f"Received initial response: {response.to_dict()}")
            else:
                logger.error(f"Unexpected message type: {msg.type}")
        except Exception as e:
            logger.error(f"Failed to send full client request: {e}")
            raise
            
    async def send_messages(self, segment_size: int, content: bytes) -> AsyncGenerator[None, None]:
        audio_segments = self.split_audio(content, segment_size)
        total_segments = len(audio_segments)
        
        for i, segment in enumerate(audio_segments):
            is_last = (i == total_segments - 1)
            request = RequestBuilder.new_audio_only_request(
                self.seq, 
                segment,
                is_last=is_last
            )
            await self.conn.send_bytes(request)
            logger.info(f"Sent audio segment {i+1}/{total_segments} with seq: {self.seq} (last: {is_last})")
            
            if not is_last:
                self.seq += 1
                
            await asyncio.sleep(self.segment_duration / 1000)
            yield
            
    async def recv_messages(self) -> AsyncGenerator[AsrResponse, None]:
        try:
            async for msg in self.conn:
                if msg.type == aiohttp.WSMsgType.BINARY:
                    response = ResponseParser.parse_response(msg.data)
                    
                    # 提取并保存识别文本
                    text = response.get_recognized_text()
                    if text:
                        self.recognized_text.append(text)
                        logger.info(f"Recognized text: {text}")
                    
                    yield response
                    
                    if response.is_last_package or response.code != 0:
                        # 保存识别结果到文件
                        if self.output_file and self.recognized_text:
                            self.save_recognized_text()
                        break
                elif msg.type == aiohttp.WSMsgType.ERROR:
                    logger.error(f"WebSocket error: {msg.data}")
                    break
                elif msg.type == aiohttp.WSMsgType.CLOSED:
                    logger.info("WebSocket connection closed")
                    break
        except Exception as e:
            logger.error(f"Error receiving messages: {e}")
            raise
            
    def save_recognized_text(self):
        """将识别出的文本保存到文件"""
        try:
            full_text = "".join(self.recognized_text)
            
            # 创建输出目录（如果不存在）
            output_dir = os.path.dirname(self.output_file)
            if output_dir and not os.path.exists(output_dir):
                os.makedirs(output_dir, exist_ok=True)
            
            # 写入文件
            with open(self.output_file, 'w', encoding=DEFAULT_OUTPUT_ENCODING) as f:
                f.write(full_text)
            
            logger.info(f"Successfully saved recognized text to {self.output_file}")
            logger.info(f"Total recognized text length: {len(full_text)} characters")
            
        except Exception as e:
            logger.error(f"Failed to save recognized text: {e}")
            raise
            
    async def start_audio_stream(self, segment_size: int, content: bytes) -> AsyncGenerator[AsrResponse, None]:
        async def sender():
            async for _ in self.send_messages(segment_size, content):
                pass
                
        sender_task = asyncio.create_task(sender())
        
        try:
            async for response in self.recv_messages():
                yield response
        finally:
            sender_task.cancel()
            try:
                await sender_task
            except asyncio.CancelledError:
                pass
                
    @staticmethod
    def split_audio(data: bytes, segment_size: int) -> List[bytes]:
        if segment_size <= 0:
            return []
            
        segments = []
        for i in range(0, len(data), segment_size):
            end = i + segment_size
            if end > len(data):
                end = len(data)
            segments.append(data[i:end])
        return segments
        
    async def execute(self, file_path: str) -> AsyncGenerator[AsrResponse, None]:
        if not file_path:
            raise ValueError("File path is empty")
            
        if not self.url:
            raise ValueError("URL is empty")
            
        self.seq = 1
        self.recognized_text = []  # 重置识别文本
        
        try:
            # 1. 读取并转换音频文件为PCM
            content = await self.read_audio_data(file_path)
            
            # 2. 计算分段大小
            segment_size = self.get_segment_size(content)
            
            # 3. 创建WebSocket连接
            await self.create_connection()
            
            # 4. 发送完整客户端请求
            await self.send_full_client_request()
            
            # 5. 启动音频流处理
            async for response in self.start_audio_stream(segment_size, content):
                yield response
                
        except Exception as e:
            logger.error(f"Error in ASR execution: {e}")
            raise
        finally:
            if self.conn:
                await self.conn.close()

async def main():
    import argparse
    
    parser = argparse.ArgumentParser(description="ASR WebSocket Client (Support MP3 directly)")
    parser.add_argument("--file", type=str, required=True, help="Audio file path (MP3/WAV supported)")
    parser.add_argument("--output", type=str, default="asr_result.txt", help="Output text file path")
    
    # WebSocket URL选项
    parser.add_argument("--url", type=str, default="wss://openspeech.bytedance.com/api/v3/sauc/bigmodel_nostream", 
                       help="WebSocket URL")
    parser.add_argument("--seg-duration", type=int, default=200, 
                       help="Audio duration(ms) per packet, default:200")
    
    args = parser.parse_args()
    
    # 自动生成输出文件名（如果未指定）
    if args.output == "asr_result.txt" and args.file:
        base_name = os.path.splitext(os.path.basename(args.file))[0]
        args.output = f"{base_name}_asr_result.txt"
    
    async with AsrWsClient(args.url, args.seg_duration, args.output) as client:
        try:
            async for response in client.execute(args.file):
                logger.info(f"Response received: {json.dumps(response.to_dict(), indent=2, ensure_ascii=False)}")
                
                # 如果有错误，输出错误信息
                if response.code != 0:
                    logger.error(f"ASR Error: code={response.code}, message={response.payload_msg}")
                    
        except Exception as e:
            logger.error(f"ASR processing failed: {e}")

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

# 用法：
# python3 sauc_websocket_demo.py --file audio.mp3 --output result.txt
# python3 sauc_websocket_demo.py --file audio.mp3  # 自动生成输出文件
# python3 sauc_websocket_demo.py --file audio.wav --seg-duration 300