# -------------------------- 主函数（控制skip_uploading参数） --------------------------
def main():
    parser = argparse.ArgumentParser(
        description='上传MP4文件到S3并生成MP3',
        formatter_class=argparse.RawTextHelpFormatter
    )
    parser.add_argument(
        '--since', 
        help='只处理在此时间之后创建的文件，格式必须为:\n  YYYY-MM-DD HH:MM:SS\n例如:\n  --since "2023-10-01 08:30:00"'
    )
    # 新增：添加--check-md5参数，控制是否开启MD5校验
    parser.add_argument(
        '--check-md5', 
        action='store_true',
        help='开启本地与S3文件的MD5校验（默认关闭）'
    )
    args = parser.parse_args()

    reference_time = get_reference_time(args.since, parser)
    print(f"⏱️ 参考时间: {reference_time.strftime('%Y-%m-%d %H:%M:%S')}")
    # 新增：读取MD5校验开关
    s3_md5_check = args.check_md5
    print(f"🔍 MD5校验开关: {'开启' if s3_md5_check else '关闭'}")

    print("========================================")
    if DRY_RUN:
        print("🔴 当前为DRY RUN测试模式")
    else:
        print("🟢 当前为正式模式")
        # 清空并初始化链接文件（确保所有文件重新生成列表）
        safe_exec(f"> {LINK_FILE}")
        safe_exec(f"echo '<ol>' > {LINK_FILE}")
        safe_exec(f"> {MP3_LINK_FILE}")
        safe_exec(f"echo '<ol>' > {MP3_LINK_FILE}")

    # 1. 收集所有MP4文件（不筛选时间，全部加入列表）
    mp4_files = []
    for file in os.listdir(LOCAL_DIR):
        if file.lower().endswith(".mp4"):
            mp4_files.append(os.path.join(LOCAL_DIR, file))  # 存储绝对路径
    mp4_files.sort(key=lambda x: natural_sort_key(os.path.basename(x)))

    # 2. 处理所有文件：根据时间决定是否跳过上传
    for mp4_file in mp4_files:
        filename = os.path.basename(mp4_file)
        create_time = datetime.fromtimestamp(os.path.getctime(mp4_file))  # 用绝对路径取创建时间
        # 核心逻辑：创建时间早于参考时间则跳过上传（但仍生成列表）
        skip_uploading = create_time <= reference_time
        
        print(f"\n===== 处理文件：{filename}（{'跳过上传' if skip_uploading else '正常处理'}） =====")
        # 修改：把s3_md5_check传入process_mp4/process_mp3
        process_mp4(mp4_file, skip_uploading, s3_md5_check)  
        process_mp3(mp4_file, skip_uploading, s3_md5_check)  


# -------------------------- 处理函数（添加skip_uploading和s3_md5_check参数） --------------------------
def process_mp4(local_mp4, skip_uploading, s3_md5_check):  # 新增s3_md5_check参数
    """处理MP4文件：skip_uploading=True时仅生成列表，跳过上传"""
    filename = os.path.basename(local_mp4)
    s3_mp4_path = f"s3://{S3_BUCKET}/{S3_DIR}{filename}"
    s3_mp4_url = f"http://{CLOUDFRONT}/{S3_DIR}{filename}"

    print(f"\n=== 处理MP4文件：{filename} ===")
    
    # 仅当不跳过上传时，执行检查和上传逻辑
    if not skip_uploading:
        print(f"[步骤1] 检查S3是否存在: {s3_mp4_path}")
        s3_exists = s3_file_exists(s3_mp4_path)
        # 移除硬编码：从参数获取MD5校验开关（原代码是s3_md5_check = False）
        # s3_md5_check = False  # 已删除，改为参数传入

        if s3_exists and s3_md5_check:  # 现在由外部参数控制
            local_md5 = get_file_md5(local_mp4)
            s3_md5 = get_s3_md5(s3_mp4_path)
            print(f"[对比] 本地MD5: {local_md5} | S3 MD5: {s3_md5}")
            if local_md5 != s3_md5:
                print("[结论] MD5不一致，需要重新上传")
                s3_exists = False
            else:
                print("[结论] MD5一致，跳过上传")

        if not s3_exists:
            print("[步骤2] 开始上传到S3")
            safe_exec(f"s3cmd put --mime-type=\"video/mp4\" --acl-public --continue --check-md5 '{local_mp4}' '{s3_mp4_path}'")
    else:
        print("[跳过上传] 仅生成列表链接")

    # 无论是否跳过上传，均生成列表链接（核心需求）
    timestamp = get_file_timestamp(local_mp4)
    link_line = f"<li><a href=\"{s3_mp4_url}\">{filename} ({timestamp})</a></li>"
    safe_exec(f"echo '{link_line}' >> {LINK_FILE}")
    print("[步骤3] 已添加链接到列表")


def process_mp3(local_mp4, skip_uploading, s3_md5_check):  # 新增s3_md5_check参数
    """处理MP3文件：skip_uploading=True时仅生成列表，跳过转码和上传"""
    filename = os.path.basename(local_mp4)
    filename_noext = os.path.splitext(filename)[0]
    mp3_filename = f"{filename_noext}.mp3"
    local_mp3_dir = os.path.join(LOCAL_DIR, MP3_LOCAL_SUBDIR)
    local_mp3 = os.path.join(local_mp3_dir, mp3_filename)
    s3_mp3_path = f"s3://{S3_BUCKET}/{S3_DIR}{MP3_S3_SUBDIR}{mp3_filename}"
    s3_mp3_url = f"http://{CLOUDFRONT}/{S3_DIR}{MP3_S3_SUBDIR}{mp3_filename}"

    print(f"\n=== 处理MP3文件：{mp3_filename} ===")
    
    # 仅当不跳过上传时，执行转码和上传逻辑
    if not skip_uploading:
        safe_exec(f"mkdir -p {local_mp3_dir}")
        print(f"[步骤1] 检查本地MP3: {local_mp3}")
        if not os.path.exists(local_mp3) or DRY_RUN:
            if not DRY_RUN and not os.path.exists(local_mp3):
                print("[生成] 本地不存在，使用ffmpeg提取音频")
                safe_exec(f"ffmpeg -y -i '{local_mp4}' -vn -acodec libmp3lame '{local_mp3}'")
            else:
                print("[DRY RUN] 模拟生成MP3文件")

        print(f"[步骤2] 检查S3是否存在: {s3_mp3_path}")
        s3_exists = s3_file_exists(s3_mp3_path)
        # 移除硬编码：从参数获取MD5校验开关（原代码是s3_md5_check = False）
        # s3_md5_check = False  # 已删除，改为参数传入

        if s3_exists and s3_md5_check:  # 现在由外部参数控制
            local_md5 = get_file_md5(local_mp3)
            s3_md5 = get_s3_md5(s3_mp3_path)
            print(f"[对比] 本地MD5: {local_md5} | S3 MD5: {s3_md5}")
            if local_md5 != s3_md5:
                print("[结论] MD5不一致，需要重新上传")
                s3_exists = False
            else:
                print("[结论] MD5一致，跳过上传")

        if not s3_exists:
            print("[步骤3] 开始上传到S3")
            safe_exec(f"s3cmd put --mime-type=\"audio/mpeg\" --acl-public --continue --check-md5 '{local_mp3}' '{s3_mp3_path}'")
    else:
        print("[跳过上传] 仅生成列表链接")

    # 无论是否跳过上传，均生成列表链接（核心需求）
    timestamp = get_file_timestamp(local_mp4)
    link_line = f"<li><a href=\"{s3_mp3_url}\">{mp3_filename} ({timestamp})</a></li>"
    safe_exec(f"echo '{link_line}' >> {MP3_LINK_FILE}")
    print("[步骤4] 已添加链接到列表")

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)

        # ===================== 【关键修改开始】=====================
        # 1. 读取原始识别文本并去首尾空格
        text = self.recognized_text.strip()
        
        # 2. 导入正则模块（仅用于句号换行）
        import re
        
        # 3. 核心规则：仅在中文句号「。」后强制换行（唯一逻辑）
        text = re.sub(r'。', r'。\n', text)
        
        # 4. 仅压缩连续空行（避免原文本空行导致格式混乱）
        text = re.sub(r'\n+', '\n', text)
        
        # 5. 最终格式化文本（去首尾空行，确保格式干净）
        formatted_text = text.strip()
        # ===================== 【关键修改结束】=====================

        # ===================== 【核心替换点】=====================
        # ！！！重点：将原代码中写入的 self.recognized_text 替换为 formatted_text ！！！
        # 若此处未替换，所有格式化逻辑都无效！！！
        with open(self.output_file, 'w', encoding=DEFAULT_OUTPUT_ENCODING) as f:
            f.write(formatted_text)  # 【替换点】原代码为 f.write(self.recognized_text)
        # ===================== 【核心替换点结束】=====================

        logger.info(f"Successfully saved final recognized text to {self.output_file}")
        logger.info(f"Final recognized text length: {len(self.recognized_text)} characters")

    except Exception as e:
        logger.error(f"Failed to save recognized text: {e}")
        raise

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

# 配置日志 - 增加debug开关控制
def setup_logging(debug: bool = False):
    """配置日志，debug模式输出详细信息，默认只输出INFO及以上"""
    log_level = logging.DEBUG if debug else logging.INFO
    logging.basicConfig(
        level=log_level,
        format='%(asctime)s - %(levelname)s - %(message)s',
        handlers=[
            logging.FileHandler('run.log'),
            logging.StreamHandler()
        ]
    )
    # 过滤掉非debug模式下的DEBUG日志
    if not debug:
        # 获取根logger并设置级别
        root_logger = logging.getLogger()
        root_logger.setLevel(logging.INFO)
        # 为本模块logger设置级别
        logger = logging.getLogger(__name__)
        logger.setLevel(logging.INFO)
    return 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)
            
            # 不强制删除原始文件
            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")
            
        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")
            
        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]
        
        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",
                "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:]
        
        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"Failed to decompress payload: {e}")
                return response
                
        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, debug: bool = False):
        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 = ""  # 直接存储完整文本，覆盖更新
        self.debug = debug  # 保存debug模式状态

    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:
        try:
            logger.info(f"Converting audio file {file_path} to PCM format...")
            pcm_data = CommonUtils.convert_audio_to_pcm(file_path, DEFAULT_SAMPLE_RATE)
            logger.debug(f"Successfully converted to PCM, size: {len(pcm_data)} bytes")  # debug级别
            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:
        try:
            bytes_per_sample = 2
            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)
            
            if segment_size % 2 != 0:
                segment_size += 1
                
            logger.debug(f"Calculated segment size: {segment_size} bytes (duration: {self.segment_duration}ms)")  # debug级别
            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.debug(f"Sent full client request with seq: {self.seq-1}")  # debug级别
            
            msg = await self.conn.receive()
            if msg.type == aiohttp.WSMsgType.BINARY:
                response = ResponseParser.parse_response(msg.data)
                logger.debug(f"Received initial response: {response.to_dict()}")  # debug级别
            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.debug(f"Sent audio segment {i+1}/{total_segments} with seq: {self.seq} (last: {is_last})")  # debug级别
            
            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 = text
                        logger.debug(f"Current recognized text: {text}")  # debug级别
                    
                    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:
            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)
            # ===================== 仅修改这几行，极简逻辑 =====================
            text = self.recognized_text.strip()
            import re
            # 唯一规则：中文句号「。」后强制换行
            text = re.sub(r'。', r'。\n', text)
            # 仅压缩连续空行，无其他操作
            #text = re.sub(r'\n+', '\n', text)
            formatted_text = text.strip()
            # ===================== 修改结束 =====================
            
            with open(self.output_file, 'w', encoding=DEFAULT_OUTPUT_ENCODING) as f:
                f.write(formatted_text)
            
            logger.info(f"Successfully saved final recognized text to {self.output_file}")
            logger.debug(f"Final recognized text length: {len(self.recognized_text)} characters")  # debug级别
            
        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:
            content = await self.read_audio_data(file_path)
            segment_size = self.get_segment_size(content)
            await self.create_connection()
            await self.send_full_client_request()
            
            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=None, help="Output text file path (auto-generated if not set)")
    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")
    parser.add_argument("--debug", action='store_true', help="Enable debug mode (show detailed logs)")  # 新增debug参数
    
    args = parser.parse_args()
    
    # 初始化日志（根据debug参数）
    global logger
    logger = setup_logging(args.debug)
    
    # 自动生成输出文件名
    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, args.debug) as client:
        try:
            async for response in client.execute(args.file):
                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())