--- /dev/null
+#!/usr/bin/env python3
+"""
+MP3音频转换工具
+将任意MP3文件转换为16kHz采样率、单声道的MP3文件
+"""
+
+import os
+import sys
+import argparse
+import subprocess
+from pathlib import Path
+import logging
+
+# 配置日志
+logging.basicConfig(
+ level=logging.INFO,
+ format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+
+def check_ffmpeg():
+ """检查ffmpeg是否已安装"""
+ try:
+ result = subprocess.run(
+ ['ffmpeg', '-version'],
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ check=False
+ )
+ if result.returncode == 0:
+ logger.info("✓ ffmpeg已安装")
+ return True
+ else:
+ logger.error("✗ ffmpeg未正确安装")
+ return False
+ except FileNotFoundError:
+ logger.error("✗ 未找到ffmpeg，请先安装ffmpeg")
+ logger.error(" Ubuntu/Debian: sudo apt-get install ffmpeg")
+ logger.error(" CentOS/RHEL: sudo yum install ffmpeg")
+ logger.error(" macOS: brew install ffmpeg")
+ return False
+
+
+def convert_mp3(input_file, output_file=None, sample_rate=16000, channels=1, bitrate='64k'):
+ """
+ 转换MP3文件
+
+ Args:
+ input_file: 输入MP3文件路径
+ output_file: 输出MP3文件路径（可选，默认为输入文件名_16k.mp3）
+ sample_rate: 采样率，默认16000 (16kHz)
+ channels: 声道数，默认1（单声道）
+ bitrate: 比特率，默认64k
+
+ Returns:
+ bool: 转换是否成功
+ """
+ input_path = Path(input_file)
+
+ # 检查输入文件是否存在
+ if not input_path.exists():
+ logger.error(f"输入文件不存在: {input_file}")
+ return False
+
+ # 检查文件是否为MP3
+ if input_path.suffix.lower() != '.mp3':
+ logger.warning(f"文件不是MP3格式: {input_file}")
+
+ # 生成输出文件名
+ if output_file is None:
+ output_path = input_path.parent / f"{input_path.stem}_16k.mp3"
+ else:
+ output_path = Path(output_file)
+
+ # 确保输出目录存在
+ output_path.parent.mkdir(parents=True, exist_ok=True)
+
+ # 构建ffmpeg命令
+ cmd = [
+ 'ffmpeg',
+ '-i', str(input_path), # 输入文件
+ '-ar', str(sample_rate), # 采样率
+ '-ac', str(channels), # 声道数
+ '-b:a', bitrate, # 比特率
+ '-y', # 覆盖输出文件
+ str(output_path) # 输出文件
+ ]
+
+ logger.info(f"转换: {input_path.name} -> {output_path.name}")
+ logger.debug(f"命令: {' '.join(cmd)}")
+
+ try:
+ # 执行转换
+ result = subprocess.run(
+ cmd,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.PIPE,
+ check=False
+ )
+
+ if result.returncode == 0:
+ # 获取文件大小
+ input_size = input_path.stat().st_size / 1024 # KB
+ output_size = output_path.stat().st_size / 1024 # KB
+ logger.info(f"✓ 转换成功: {output_path}")
+ logger.info(f" 输入大小: {input_size:.1f} KB")
+ logger.info(f" 输出大小: {output_size:.1f} KB")
+ return True
+ else:
+ logger.error(f"✗ 转换失败: {input_file}")
+ logger.error(f" 错误信息: {result.stderr.decode('utf-8', errors='ignore')}")
+ return False
+
+ except Exception as e:
+ logger.error(f"✗ 转换过程出错: {e}")
+ return False
+
+
+def convert_directory(input_dir, output_dir=None, sample_rate=16000, channels=1, bitrate='64k', recursive=False):
+ """
+ 批量转换目录中的所有MP3文件
+
+ Args:
+ input_dir: 输入目录
+ output_dir: 输出目录（可选）
+ sample_rate: 采样率
+ channels: 声道数
+ bitrate: 比特率
+ recursive: 是否递归处理子目录
+
+ Returns:
+ tuple: (成功数量, 失败数量)
+ """
+ input_path = Path(input_dir)
+
+ if not input_path.exists() or not input_path.is_dir():
+ logger.error(f"输入目录不存在: {input_dir}")
+ return 0, 0
+
+ # 查找所有MP3文件
+ if recursive:
+ mp3_files = list(input_path.rglob('*.mp3'))
+ else:
+ mp3_files = list(input_path.glob('*.mp3'))
+
+ if not mp3_files:
+ logger.warning(f"未找到MP3文件: {input_dir}")
+ return 0, 0
+
+ logger.info(f"找到 {len(mp3_files)} 个MP3文件")
+
+ success_count = 0
+ fail_count = 0
+
+ for mp3_file in mp3_files:
+ # 跳过已经转换过的文件（文件名包含_16k）
+ if '_16k' in mp3_file.stem:
+ logger.info(f"跳过已转换文件: {mp3_file.name}")
+ continue
+
+ # 确定输出文件路径
+ if output_dir:
+ output_path = Path(output_dir)
+ # 保持相对路径结构
+ rel_path = mp3_file.relative_to(input_path)
+ out_file = output_path / rel_path.parent / f"{mp3_file.stem}_16k.mp3"
+ else:
+ out_file = None # 使用默认命名
+
+ # 转换文件
+ if convert_mp3(mp3_file, out_file, sample_rate, channels, bitrate):
+ success_count += 1
+ else:
+ fail_count += 1
+
+ return success_count, fail_count
+
+
+def main():
+ parser = argparse.ArgumentParser(
+ description='将MP3文件转换为16kHz单声道格式',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+示例:
+ # 转换单个文件
+ %(prog)s input.mp3
+
+ # 指定输出文件
+ %(prog)s input.mp3 -o output.mp3
+
+ # 批量转换目录
+ %(prog)s -d /path/to/mp3/
+
+ # 递归转换目录及子目录
+ %(prog)s -d /path/to/mp3/ -r
+
+ # 自定义参数
+ %(prog)s input.mp3 --sample-rate 8000 --bitrate 32k
+ """
+ )
+
+ parser.add_argument('input', nargs='?', help='输入MP3文件或目录')
+ parser.add_argument('-o', '--output', help='输出MP3文件或目录')
+ parser.add_argument('-d', '--directory', help='批量转换目录')
+ parser.add_argument('-r', '--recursive', action='store_true', help='递归处理子目录')
+ parser.add_argument('--sample-rate', type=int, default=16000, help='采样率（默认: 16000）')
+ parser.add_argument('--channels', type=int, default=1, choices=[1, 2], help='声道数（默认: 1单声道）')
+ parser.add_argument('--bitrate', default='64k', help='比特率（默认: 64k）')
+ parser.add_argument('-v', '--verbose', action='store_true', help='详细输出')
+
+ args = parser.parse_args()
+
+ # 设置日志级别
+ if args.verbose:
+ logger.setLevel(logging.DEBUG)
+
+ # 检查ffmpeg
+ if not check_ffmpeg():
+ return 1
+
+ # 处理目录模式
+ if args.directory:
+ logger.info(f"批量转换模式")
+ success, fail = convert_directory(
+ args.directory,
+ args.output,
+ args.sample_rate,
+ args.channels,
+ args.bitrate,
+ args.recursive
+ )
+ logger.info(f"转换完成: 成功 {success}, 失败 {fail}")
+ return 0 if fail == 0 else 1
+
+ # 处理单文件模式
+ elif args.input:
+ input_path = Path(args.input)
+
+ # 如果输入是目录，提示使用-d参数
+ if input_path.is_dir():
+ logger.error("输入是目录，请使用 -d 参数进行批量转换")
+ return 1
+
+ # 转换单个文件
+ success = convert_mp3(
+ args.input,
+ args.output,
+ args.sample_rate,
+ args.channels,
+ args.bitrate
+ )
+ return 0 if success else 1
+
+ else:
+ parser.print_help()
+ return 1
+
+
+if __name__ == '__main__':
+ sys.exit(main())
+
--- /dev/null
+#!/usr/bin/env python3
+"""
+WebSocket MP3流播放客户端
+接收MP3二进制流并实时播放
+"""
+
+import asyncio
+import logging
+import argparse
+import sys
+import io
+import tempfile
+from pathlib import Path
+from datetime import datetime
+
+try:
+ import websockets
+except ImportError:
+ print("错误: 无法导入 websockets 模块")
+ print("请安装: pip install websockets")
+ sys.exit(1)
+
+try:
+ import pygame
+ PYGAME_AVAILABLE = True
+except ImportError:
+ PYGAME_AVAILABLE = False
+ print("警告: pygame 未安装，将保存到文件而不播放")
+ print("安装播放功能: pip install pygame")
+
+# 配置日志
+logging.basicConfig(
+ level=logging.INFO,
+ format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+
+class MP3StreamClient:
+ """WebSocket MP3流播放客户端"""
+
+ def __init__(self, server_url: str, play_mode: str = 'stream',
+ save_dir: str = None, buffer_size: int = 262144):
+ """
+ 初始化客户端
+
+ Args:
+ server_url: WebSocket服务器地址 (如: ws://localhost:8910)
+ play_mode: 播放模式 ('stream'=流式播放, 'buffer'=缓冲播放, 'save'=仅保存)
+ save_dir: 保存目录 (save模式必需)
+ buffer_size: 缓冲区大小（字节，默认256KB，避免播放卡顿）
+ """
+ self.server_url = server_url
+ self.play_mode = play_mode
+ self.save_dir = Path(save_dir) if save_dir else None
+ self.buffer_size = buffer_size
+
+ # 统计信息
+ self.received_bytes = 0
+ self.received_packets = 0
+ self.start_time = None
+
+ # 播放相关
+ self.audio_buffer = bytearray()
+ self.temp_file = None
+ self.is_playing = False
+ self.play_queue = [] # 播放队列
+ self.min_buffer_size = max(buffer_size, 131072) # 最小128KB
+
+ # 验证保存目录
+ if self.play_mode == 'save':
+ if not self.save_dir:
+ raise ValueError("save模式必须指定保存目录")
+ self.save_dir.mkdir(parents=True, exist_ok=True)
+ logger.info(f"保存目录: {self.save_dir.absolute()}")
+
+ # 初始化pygame（增大缓冲区避免卡顿）
+ if PYGAME_AVAILABLE and self.play_mode != 'save':
+ pygame.mixer.init(frequency=44100, size=-16, channels=2, buffer=8192)
+ logger.info("音频播放器已初始化")
+ logger.info(f"建议：为避免卡顿，建议缓冲区≥256KB")
+
+ logger.info(f"服务器地址: {server_url}")
+ logger.info(f"播放模式: {play_mode}")
+ logger.info(f"缓冲大小: {self.format_bytes(buffer_size)}")
+ if play_mode != 'save':
+ logger.info(f"最小缓冲: {self.format_bytes(self.min_buffer_size)}")
+
+ def format_bytes(self, bytes_val):
+ """格式化字节大小"""
+ if bytes_val < 1024:
+ return f"{bytes_val} B"
+ elif bytes_val < 1024 * 1024:
+ return f"{bytes_val / 1024:.2f} KB"
+ else:
+ return f"{bytes_val / (1024 * 1024):.2f} MB"
+
+ def print_stats(self):
+ """打印统计信息"""
+ if self.start_time:
+ elapsed = (datetime.now() - self.start_time).total_seconds()
+ speed = self.received_bytes / elapsed if elapsed > 0 else 0
+ logger.info(f"统计: 接收 {self.format_bytes(self.received_bytes)} "
+ f"({self.received_packets} 包) "
+ f"平均速度 {self.format_bytes(speed)}/s")
+
+ async def play_stream_mode(self, data):
+ """
+ 流式播放模式：累积较大数据块后播放，减少卡顿
+
+ 关键改进：
+ 1. 累积足够大的数据块（默认256KB）再播放
+ 2. 使用队列机制平滑衔接
+ 3. 避免频繁创建小文件导致的停顿
+ """
+ self.audio_buffer.extend(data)
+
+ # 当缓冲区达到足够大时才播放（避免小片段导致卡顿）
+ if len(self.audio_buffer) >= self.min_buffer_size:
+ try:
+ # 创建临时MP3文件
+ with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f:
+ f.write(bytes(self.audio_buffer))
+ temp_path = f.name
+
+ self.play_queue.append(temp_path)
+
+ # 播放队列中的音频
+ if not pygame.mixer.music.get_busy():
+ # 没有音频在播放，立即播放
+ pygame.mixer.music.load(temp_path)
+ pygame.mixer.music.play()
+ logger.info(f"▶ 开始播放: {self.format_bytes(len(self.audio_buffer))}")
+ self.play_queue.pop(0)
+ else:
+ # 正在播放，排队下一个
+ if len(self.play_queue) == 1:
+ pygame.mixer.music.queue(temp_path)
+ logger.debug(f"📋 音频排队: {self.format_bytes(len(self.audio_buffer))}")
+ self.play_queue.pop(0)
+ else:
+ logger.debug(f"⏸ 队列等待: {len(self.play_queue)} 个")
+
+ # 清空缓冲区
+ self.audio_buffer.clear()
+
+ except Exception as e:
+ logger.error(f"播放出错: {e}")
+
+ async def play_buffer_mode(self, data):
+ """
+ 缓冲播放模式：累积更大的数据后再播放
+ 适合网络不稳定的情况，提供最流畅的体验
+ """
+ self.audio_buffer.extend(data)
+
+ # 缓冲更大的数据量（512KB或更多）
+ min_buffer = max(self.buffer_size * 2, 524288) # 至少512KB
+
+ if len(self.audio_buffer) >= min_buffer:
+ try:
+ with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f:
+ f.write(bytes(self.audio_buffer))
+ temp_path = f.name
+
+ # 等待当前播放完成（避免重叠）
+ while pygame.mixer.music.get_busy():
+ await asyncio.sleep(0.1)
+
+ pygame.mixer.music.load(temp_path)
+ pygame.mixer.music.play()
+ logger.info(f"▶ 播放音频块: {self.format_bytes(len(self.audio_buffer))}")
+
+ self.audio_buffer.clear()
+
+ except Exception as e:
+ logger.error(f"播放出错: {e}")
+
+ async def save_mode(self, data):
+ """
+ 仅保存模式：将接收的数据保存到文件
+ """
+ if not self.temp_file:
+ # 创建新文件
+ timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
+ filename = f"stream_{timestamp}.mp3"
+ self.temp_file = self.save_dir / filename
+ logger.info(f"开始保存到: {self.temp_file}")
+
+ # 追加数据
+ with open(self.temp_file, 'ab') as f:
+ f.write(data)
+
+ async def receive_and_play(self):
+ """接收并播放MP3流"""
+ try:
+ logger.info(f"连接到服务器: {self.server_url}")
+
+ async with websockets.connect(self.server_url) as websocket:
+ logger.info("连接成功，开始接收MP3流...")
+ self.start_time = datetime.now()
+
+ # 定期打印统计信息
+ last_stats_time = datetime.now()
+
+ async for message in websocket:
+ # 接收二进制数据
+ if isinstance(message, bytes):
+ self.received_bytes += len(message)
+ self.received_packets += 1
+
+ logger.debug(f"接收包 #{self.received_packets}: {len(message)} 字节")
+
+ # 根据模式处理数据
+ if self.play_mode == 'stream' and PYGAME_AVAILABLE:
+ await self.play_stream_mode(message)
+ elif self.play_mode == 'buffer' and PYGAME_AVAILABLE:
+ await self.play_buffer_mode(message)
+ elif self.play_mode == 'save':
+ await self.save_mode(message)
+ else:
+ # 没有pygame，自动切换到保存模式
+ if not hasattr(self, '_warned_no_pygame'):
+ logger.warning("pygame不可用，切换到保存模式")
+ self._warned_no_pygame = True
+ if not self.save_dir:
+ self.save_dir = Path('./received_mp3')
+ self.save_dir.mkdir(exist_ok=True)
+ await self.save_mode(message)
+
+ # 定期打印统计信息（每5秒）
+ now = datetime.now()
+ if (now - last_stats_time).total_seconds() >= 5:
+ self.print_stats()
+ last_stats_time = now
+
+ else:
+ logger.warning(f"接收到非二进制消息: {message}")
+
+ except websockets.exceptions.ConnectionClosedOK:
+ logger.info("服务器关闭连接")
+ except websockets.exceptions.ConnectionClosedError as e:
+ logger.error(f"连接异常关闭: {e}")
+ except Exception as e:
+ logger.error(f"接收数据出错: {e}", exc_info=True)
+ finally:
+ # 播放剩余缓冲区数据
+ if len(self.audio_buffer) > 0:
+ if self.play_mode in ['stream', 'buffer'] and PYGAME_AVAILABLE:
+ try:
+ with tempfile.NamedTemporaryFile(suffix='.mp3', delete=False) as f:
+ f.write(bytes(self.audio_buffer))
+ temp_path = f.name
+ pygame.mixer.music.load(temp_path)
+ pygame.mixer.music.play()
+ # 等待播放完成
+ while pygame.mixer.music.get_busy():
+ await asyncio.sleep(0.1)
+ except Exception as e:
+ logger.error(f"播放剩余数据出错: {e}")
+ elif self.play_mode == 'save':
+ await self.save_mode(bytes(self.audio_buffer))
+
+ # 打印最终统计
+ self.print_stats()
+
+ if self.play_mode == 'save' and self.temp_file:
+ logger.info(f"文件已保存: {self.temp_file} ({self.format_bytes(self.temp_file.stat().st_size)})")
+
+ async def run(self):
+ """运行客户端"""
+ try:
+ await self.receive_and_play()
+ except KeyboardInterrupt:
+ logger.info("用户中断")
+ finally:
+ # 清理
+ if PYGAME_AVAILABLE:
+ pygame.mixer.quit()
+ logger.info("客户端已停止")
+
+
+def main():
+ """主函数"""
+ parser = argparse.ArgumentParser(
+ description='WebSocket MP3流播放客户端',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+播放模式说明:
+ stream - 流式播放（实时，延迟最低）
+ buffer - 缓冲播放（累积更多数据后播放，更稳定）
+ save - 仅保存到文件，不播放
+
+示例:
+ # 流式播放
+ python3 mp3_stream_client.py ws://localhost:8910
+
+ # 缓冲播放
+ python3 mp3_stream_client.py ws://localhost:8910 --mode buffer
+
+ # 仅保存
+ python3 mp3_stream_client.py ws://localhost:8910 --mode save --save-dir ./downloads
+ """
+ )
+
+ parser.add_argument('server', help='WebSocket服务器地址 (如: ws://snicker.com.cn:8910)')
+ parser.add_argument('--mode', choices=['stream', 'buffer', 'save'],
+ default='buffer', help='播放模式 (默认: buffer，最流畅)')
+ parser.add_argument('--save-dir', help='保存目录 (save模式必需)')
+ parser.add_argument('--buffer', type=int, default=262144,
+ help='缓冲区大小（字节）(默认: 262144=256KB，避免卡顿)')
+ parser.add_argument('--debug', action='store_true', help='启用调试日志')
+
+ args = parser.parse_args()
+
+ # 设置日志级别
+ if args.debug:
+ logging.getLogger().setLevel(logging.DEBUG)
+
+ # 验证参数
+ if args.mode == 'save' and not args.save_dir:
+ parser.error("save模式必须指定 --save-dir")
+
+ # 检查pygame
+ if args.mode in ['stream', 'buffer'] and not PYGAME_AVAILABLE:
+ logger.warning("pygame未安装，将自动切换到save模式")
+ if not args.save_dir:
+ args.save_dir = './received_mp3'
+ logger.info(f"使用默认保存目录: {args.save_dir}")
+
+ # 创建并运行客户端
+ try:
+ client = MP3StreamClient(
+ server_url=args.server,
+ play_mode=args.mode,
+ save_dir=args.save_dir,
+ buffer_size=args.buffer
+ )
+
+ asyncio.run(client.run())
+
+ except Exception as e:
+ logger.error(f"客户端错误: {e}", exc_info=True)
+ sys.exit(1)
+
+
+if __name__ == '__main__':
+ main()
+
+
--- /dev/null
+#!/usr/bin/env python3
+"""
+WebSocket MP3流传输服务器
+直接发送MP3二进制流，客户端可直接播放，无需保存
+"""
+
+import asyncio
+import logging
+import argparse
+import os
+import time
+import struct
+from pathlib import Path
+from typing import Set
+
+try:
+ import websockets
+ from websockets.asyncio.server import serve
+ from websockets.exceptions import ConnectionClosedOK, ConnectionClosedError
+except ImportError as e:
+ print(f"错误: 无法导入 websockets 模块: {e}")
+ print("请安装 websockets: pip install websockets")
+ exit(1)
+
+try:
+ from mutagen.mp3 import MP3
+ HAS_MUTAGEN = True
+except ImportError:
+ HAS_MUTAGEN = False
+ logger = logging.getLogger(__name__)
+ logger.warning("mutagen未安装，将使用比特率估算时长。安装: pip install mutagen")
+
+# 配置日志
+logging.basicConfig(
+ level=logging.INFO,
+ format='%(asctime)s - %(levelname)s - %(message)s'
+)
+logger = logging.getLogger(__name__)
+
+
+class MP3StreamServer:
+ """WebSocket MP3流传输服务器类"""
+
+ def __init__(self, host: str, port: int, file_dir: str,
+ realtime: bool = True, speed_factor: float = 1.0,
+ chunk_size: int = 4096, send_buffer_size: int = 16384):
+ """
+ 初始化服务器
+
+ Args:
+ host: 服务器地址
+ port: 服务器端口
+ file_dir: MP3文件目录
+ realtime: 是否按实时速度发送（根据MP3时长控制）
+ speed_factor: 播放速度倍数（1.0=正常，2.0=2倍速）
+ chunk_size: 每次读取的数据块大小（字节）
+ send_buffer_size: 发送缓存大小（字节）
+ """
+ self.host = host
+ self.port = port
+ self.file_dir = Path(file_dir)
+ self.realtime = realtime
+ self.speed_factor = speed_factor
+ self.chunk_size = chunk_size
+ self.send_buffer_size = send_buffer_size
+ self.clients = set()
+
+ # 验证目录
+ if not self.file_dir.exists():
+ raise ValueError(f"目录不存在: {file_dir}")
+ if not self.file_dir.is_dir():
+ raise ValueError(f"不是目录: {file_dir}")
+
+ logger.info(f"MP3目录: {self.file_dir.absolute()}")
+ logger.info(f"实时播放: {'是' if realtime else '否（最快速度）'}")
+ if realtime:
+ logger.info(f"播放速度: {speed_factor}x")
+ logger.info(f"读取块: {chunk_size}字节")
+ logger.info(f"发送缓存: {send_buffer_size}字节")
+
+ def get_mp3_list(self):
+ """获取目录下所有MP3文件列表"""
+ files = []
+ for item in sorted(self.file_dir.iterdir()):
+ if item.is_file() and item.suffix.lower() in ['.mp3', '.MP3']:
+ files.append(item)
+ return files
+
+ def get_mp3_duration(self, file_path: Path) -> float:
+ """
+ 获取MP3文件时长（秒）
+
+ Args:
+ file_path: MP3文件路径
+
+ Returns:
+ float: 时长（秒），如果无法获取则返回0
+ """
+ try:
+ if HAS_MUTAGEN:
+ # 使用mutagen获取精确时长
+ audio = MP3(str(file_path))
+ duration = audio.info.length
+ logger.debug(f"MP3时长 (mutagen): {duration:.2f}秒")
+ return duration
+ else:
+ # 使用比特率估算（假设64kbps）
+ file_size = file_path.stat().st_size
+ estimated_bitrate = 64000 # 64 kbps
+ duration = (file_size * 8) / estimated_bitrate
+ logger.debug(f"MP3时长 (估算): {duration:.2f}秒")
+ return duration
+ except Exception as e:
+ logger.error(f"无法获取MP3时长 {file_path.name}: {e}")
+ return 0
+
+ async def send_mp3_stream(self, websocket, file_path: Path, send_buffer: bytearray):
+ """
+ 发送MP3文件流（纯二进制，无文件头）
+ 根据MP3实际播放时长控制发送速度
+
+ Args:
+ websocket: WebSocket连接
+ file_path: MP3文件路径
+ send_buffer: 发送缓存（bytearray）
+ """
+ try:
+ file_size = file_path.stat().st_size
+ file_name = file_path.name
+
+ # 获取MP3时长
+ duration = self.get_mp3_duration(file_path) if self.realtime else 0
+
+ if self.realtime and duration > 0:
+ # 计算实时发送速率 (bytes/秒)
+ bytes_per_second = file_size / (duration / self.speed_factor)
+ # 计算每个buffer发送后的等待时间（秒）
+ sleep_per_buffer = self.send_buffer_size / bytes_per_second
+
+ logger.info(f"开始发送MP3流: {file_name}")
+ logger.info(f" 文件大小: {file_size:,} 字节 ({file_size/1024:.1f} KB)")
+ logger.info(f" 播放时长: {duration:.1f} 秒 ({duration/60:.1f} 分钟)")
+ logger.info(f" 发送速率: {bytes_per_second:.0f} bytes/s ({bytes_per_second/1024:.1f} KB/s)")
+ logger.info(f" 预计完成: {duration/self.speed_factor:.1f} 秒")
+ else:
+ sleep_per_buffer = 0
+ logger.info(f"开始发送MP3流: {file_name} ({file_size:,} 字节) - 最快速度")
+
+ # 发送前清空缓存
+ send_buffer.clear()
+
+ # 记录开始时间
+ start_time = time.time()
+ sent_bytes = 0
+ total_sent_packets = 0
+
+ with open(file_path, 'rb') as f:
+ while True:
+ # 读取数据块
+ chunk = f.read(self.chunk_size)
+ if not chunk:
+ break
+
+ # 添加到发送缓存
+ send_buffer.extend(chunk)
+
+ # 当缓存满时发送
+ if len(send_buffer) >= self.send_buffer_size:
+ packet_start = time.time()
+
+ await websocket.send(bytes(send_buffer))
+ sent_bytes += len(send_buffer)
+ total_sent_packets += 1
+
+ logger.debug(f"发送包 #{total_sent_packets}: {len(send_buffer)} 字节 "
+ f"({sent_bytes}/{file_size} = {sent_bytes/file_size*100:.1f}%)")
+
+ # 清空缓存
+ send_buffer.clear()
+
+ # 实时播放：按计算的速率等待
+ if self.realtime and sleep_per_buffer > 0:
+ # 计算实际发送耗时
+ send_elapsed = time.time() - packet_start
+ # 等待剩余时间以匹配播放速率
+ sleep_time = max(0, sleep_per_buffer - send_elapsed)
+ if sleep_time > 0:
+ await asyncio.sleep(sleep_time)
+
+ # 发送缓存中的剩余数据
+ if len(send_buffer) > 0:
+ await websocket.send(bytes(send_buffer))
+ sent_bytes += len(send_buffer)
+ total_sent_packets += 1
+ logger.debug(f"发送最后一包: {len(send_buffer)} 字节")
+ send_buffer.clear()
+
+ # 计算实际耗时
+ elapsed_time = time.time() - start_time
+ actual_speed = sent_bytes / elapsed_time if elapsed_time > 0 else 0
+
+ logger.info(f"MP3流发送完成: {file_name}")
+ logger.info(f" 总计: {sent_bytes:,} 字节, {total_sent_packets} 包")
+ logger.info(f" 耗时: {elapsed_time:.1f} 秒")
+ logger.info(f" 实际速率: {actual_speed:.0f} bytes/s ({actual_speed/1024:.1f} KB/s)")
+
+ except Exception as e:
+ logger.error(f"发送MP3流出错 {file_path.name}: {e}")
+ raise
+
+ async def mp3_sender_loop(self, websocket):
+ """
+ MP3流发送循环
+ 持续循环发送目录下的所有MP3文件
+ """
+ logger.info(f"客户端连接: {websocket.remote_address}")
+
+ # 创建发送缓存（每个客户端独立的缓存）
+ send_buffer = bytearray()
+ logger.info(f"创建发送缓存: {self.send_buffer_size} 字节")
+
+ try:
+ # 循环发送MP3文件
+ loop_count = 0
+ while True:
+ loop_count += 1
+ logger.info(f"===== 第 {loop_count} 轮播放 =====")
+
+ # 获取MP3文件列表
+ files = self.get_mp3_list()
+
+ if not files:
+ logger.warning(f"目录为空或无MP3文件: {self.file_dir}")
+ await asyncio.sleep(5)
+ continue
+
+ logger.info(f"找到 {len(files)} 个MP3文件")
+
+ # 逐个发送MP3文件流
+ for idx, file_path in enumerate(files, 1):
+ logger.info(f"[{idx}/{len(files)}] 播放: {file_path.name}")
+ await self.send_mp3_stream(websocket, file_path, send_buffer)
+
+ logger.info(f"第 {loop_count} 轮播放完成，准备下一轮...")
+
+ except (ConnectionClosedOK, ConnectionClosedError) as e:
+ logger.info(f"客户端断开: {websocket.remote_address} - {e}")
+ except Exception as e:
+ logger.error(f"发送循环出错: {e}", exc_info=True)
+ finally:
+ self.clients.discard(websocket)
+ logger.info(f"客户端已移除，当前连接数: {len(self.clients)}")
+
+ async def handle_client(self, websocket):
+ """
+ 处理客户端连接
+ """
+ self.clients.add(websocket)
+ logger.info(f"新客户端连接: {websocket.remote_address}, 当前连接数: {len(self.clients)}")
+
+ try:
+ # 启动MP3流发送循环
+ await self.mp3_sender_loop(websocket)
+ except Exception as e:
+ logger.error(f"处理客户端出错: {e}")
+ finally:
+ self.clients.discard(websocket)
+
+ async def start(self):
+ """启动服务器"""
+ logger.info(f"启动 WebSocket MP3流服务器: ws://{self.host}:{self.port}")
+
+ async with serve(self.handle_client, self.host, self.port):
+ logger.info("服务器已启动，等待客户端连接...")
+ logger.info("客户端连接后将持续接收MP3二进制流")
+ await asyncio.Future() # 永久运行
+
+
+def main():
+ """主函数"""
+ parser = argparse.ArgumentParser(
+ description='WebSocket MP3流传输服务器（按播放时长实时发送）',
+ formatter_class=argparse.RawDescriptionHelpFormatter,
+ epilog="""
+示例:
+ # 实时播放（根据MP3时长自动控制速度）
+ %(prog)s --dir ./mp3
+
+ # 最快速度发送
+ %(prog)s --dir ./mp3 --no-realtime
+
+ # 2倍速播放
+ %(prog)s --dir ./mp3 --speed 2.0
+
+ # 0.5倍速（慢速播放）
+ %(prog)s --dir ./mp3 --speed 0.5
+ """
+ )
+ parser.add_argument('--host', default='0.0.0.0', help='服务器地址 (默认: 0.0.0.0)')
+ parser.add_argument('--port', type=int, default=8910, help='服务器端口 (默认: 8910)')
+ parser.add_argument('--dir', required=True, help='MP3文件目录')
+ parser.add_argument('--no-realtime', action='store_true',
+ help='禁用实时播放，以最快速度发送')
+ parser.add_argument('--speed', type=float, default=1.0,
+ help='播放速度倍数 (默认: 1.0, 实时播放)')
+ parser.add_argument('--chunk', type=int, default=4096,
+ help='读取块大小（字节）(默认: 4096)')
+ parser.add_argument('--buffer', type=int, default=16384,
+ help='发送缓存大小（字节）(默认: 16384)')
+ parser.add_argument('--debug', action='store_true', help='启用调试日志')
+
+ args = parser.parse_args()
+
+ # 设置日志级别
+ if args.debug:
+ logging.getLogger().setLevel(logging.DEBUG)
+
+ # 创建并启动服务器
+ try:
+ server = MP3StreamServer(
+ host=args.host,
+ port=args.port,
+ file_dir=args.dir,
+ realtime=not args.no_realtime,
+ speed_factor=args.speed,
+ chunk_size=args.chunk,
+ send_buffer_size=args.buffer
+ )
+
+ asyncio.run(server.start())
+
+ except KeyboardInterrupt:
+ logger.info("服务器已停止")
+ except Exception as e:
+ logger.error(f"服务器错误: {e}", exc_info=True)
+ exit(1)
+
+
+if __name__ == '__main__':
+ main()
+
+
if not released: # pragma: no cover
- import pathlib
- import re
- import subprocess
-
- def get_version(tag: str) -> str:
- # Since setup.py executes the contents of src/websockets/version.py,
- # __file__ can point to either of these two files.
- file_path = pathlib.Path(__file__)
- root_dir = file_path.parents[0 if file_path.name == "setup.py" else 2]
-
- # Read version from package metadata if it is installed.
- try:
- version = importlib.metadata.version("websockets")
- except ImportError:
- pass
- else:
- # Check that this file belongs to the installed package.
- files = importlib.metadata.files("websockets")
- if files:
- version_files = [f for f in files if f.name == file_path.name]
- if version_files:
- version_file = version_files[0]
- if version_file.locate() == file_path:
- return version
-
- # Read version from git if available.
- try:
- description = subprocess.run(
- ["git", "describe", "--dirty", "--tags", "--long"],
- capture_output=True,
- cwd=root_dir,
- timeout=1,
- check=True,
- text=True,
- ).stdout.strip()
- # subprocess.run raises FileNotFoundError if git isn't on $PATH.
- except (
- FileNotFoundError,
- subprocess.CalledProcessError,
- subprocess.TimeoutExpired,
- ):
- pass
- else:
- description_re = r"[0-9.]+-([0-9]+)-(g[0-9a-f]{7,}(?:-dirty)?)"
- match = re.fullmatch(description_re, description)
- if match is None:
- raise ValueError(f"Unexpected git description: {description}")
- distance, remainder = match.groups()
- remainder = remainder.replace("-", ".") # required by PEP 440
- return f"{tag}.dev{distance}+{remainder}"
-
- # Avoid crashing if the development version cannot be determined.
- return f"{tag}.dev0+gunknown"
-
- version = get_version(tag)
-
- def get_commit(tag: str, version: str) -> str:
- # Extract commit from version, falling back to tag if not available.
- version_re = r"[0-9.]+\.dev[0-9]+\+g([0-9a-f]{7,}|unknown)(?:\.dirty)?"
- match = re.fullmatch(version_re, version)
- if match is None:
- raise ValueError(f"Unexpected version: {version}")
- (commit,) = match.groups()
- return tag if commit == "unknown" else commit
-
- commit = get_commit(tag, version)
+ # 开发版本：不使用 git 信息，直接使用固定版本
+ version = f"{tag}.dev0"
+ commit = tag
projects / mirrors / FD230.git / commitdiff