utils.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-

"""
工具模块
包含辅助功能如音频播放和文件操作
"""

import os
import shutil
import threading
import tempfile
import time
from typing import Optional, Callable

# 导入音频处理库
try:
    import pyaudio
    import wave
    PYAUDIO_AVAILABLE = True
except ImportError:
    PYAUDIO_AVAILABLE = False
    print("警告: 未能导入PyAudio模块，音频播放功能将不可用")

try:
    import numpy as np
    NUMPY_AVAILABLE = True
except ImportError:
    NUMPY_AVAILABLE = False
    print("警告: 未能导入numpy模块，部分功能将不可用")
    # 创建一个假的numpy模块，以便在导入失败时仍能继续
    class FakeNumpy:
        def __init__(self):
            pass
        def array(self, *args, **kwargs):
            return []
    np = FakeNumpy()

try:
    import soundfile as sf
    SOUNDFILE_AVAILABLE = True
except ImportError:
    SOUNDFILE_AVAILABLE = False
    print("警告: 未能导入soundfile模块，音频格式转换功能将不可用")

class AudioPlayer:
    """音频播放器类"""
    
    def __init__(self):
        """初始化播放器"""
        self.playing = False
        self.stop_event = threading.Event()
        self.play_thread = None
        self.temp_wav = None
        self.total_frames = 0     # 总帧数
        self.current_frame = 0    # 当前帧位置
        self.frame_rate = 0       # 采样率
        self.wf = None            # 当前打开的WAV文件
        self.progress_callback = None  # 进度回调函数
        self.seeking = False      # 正在跳转标志
        self.pyaudio_instance = None  # 保存PyAudio实例的引用
        self.stream = None        # 保存流的引用
        self._file_path = None    # 保存当前文件路径
        self._update_timer = None # 进度更新定时器
        self._chunk_size = 4096   # 缓冲区大小
        self._lock = threading.RLock()  # 添加线程锁，确保线程安全
    
    def play(self, file_path: str, progress_callback=None):
        """
        播放音频文件
        
        Args:
            file_path: 音频文件路径
            progress_callback: 播放进度回调函数，接收当前位置百分比参数
        """
        with self._lock:  # 使用线程锁保护关键部分
            # 检查PyAudio是否可用
            if not PYAUDIO_AVAILABLE:
                print("错误: 未能导入PyAudio模块，无法播放音频")
                return
                
            # 检查文件是否存在
            if not os.path.exists(file_path):
                print(f"错误: 文件不存在: {file_path}")
                return
                
            # 如果已经在播放，先停止
            if self.playing:
                self.stop()
                time.sleep(0.2)  # 等待一小段时间确保之前的资源被释放
            
            # 保存文件路径
            self._file_path = file_path
                
            # 保存进度回调
            self.progress_callback = progress_callback
            
            # 初始化播放线程
            self.stop_event.clear()
            self.playing = True
            
            # 启动播放线程
            try:
                self.play_thread = threading.Thread(
                    target=self._play_audio,
                    args=(file_path,),
                    daemon=True  # 确保线程是守护线程
                )
                self.play_thread.start()
                print(f"开始播放音频: {file_path}")
            except Exception as e:
                print(f"启动播放线程时出错: {e}")
                self.playing = False
    
    def stop(self):
        """
        停止播放
        """
        with self._lock:  # 使用线程锁保护关键部分
            if not self.playing:
                return
                
            # 设置停止标志
            self.stop_event.set()
            self.playing = False
            
            # 关闭并清理资源
            try:
                if self.stream:
                    try:
                        self.stream.stop_stream()
                        self.stream.close()
                    except Exception as e:
                        print(f"关闭音频流时出错: {str(e)}")
                    finally:
                        self.stream = None
                
                if self.pyaudio_instance:
                    try:
                        self.pyaudio_instance.terminate()
                    except Exception as e:
                        print(f"终止PyAudio实例时出错: {str(e)}")
                    finally:
                        self.pyaudio_instance = None
                    
                if self.wf:
                    try:
                        self.wf.close()
                    except Exception as e:
                        print(f"关闭WAV文件时出错: {str(e)}")
                    finally:
                        self.wf = None
                    
                # 删除临时文件
                if self.temp_wav and os.path.exists(self.temp_wav):
                    try:
                        os.remove(self.temp_wav)
                    except Exception as e:
                        print(f"删除临时文件时出错: {str(e)}")
                    finally:
                        self.temp_wav = None
                        
                print("成功停止音频播放")
            
            except Exception as e:
                print(f"停止播放时出错: {str(e)}")
            finally:
                # 重置状态
                self.stop_event.clear()
                self.seeking = False
                
                # 等待播放线程结束
                if self.play_thread and self.play_thread.is_alive():
                    self.play_thread.join(timeout=0.5)  # 等待最多0.5秒
                    
                # 重置播放线程引用
                self.play_thread = None
    
    def get_duration(self):
        """获取音频总时长（秒）"""
        if self.frame_rate > 0 and self.total_frames > 0:
            return self.total_frames / self.frame_rate
        return 0
    
    def get_position(self):
        """获取当前播放位置（秒）"""
        if self.frame_rate > 0:
            return self.current_frame / self.frame_rate
        return 0
    
    def get_position_percent(self):
        """获取当前播放位置百分比（0-100）"""
        if self.total_frames > 0:
            return (self.current_frame * 100) / self.total_frames
        return 0
    
    def seek(self, percent):
        """
        跳转到指定位置
        
        Args:
            percent: 位置百分比（0-100）
        """
        if not self.wf or not self.playing:
            return False
            
        try:
            self.seeking = True
            
            # 确保百分比在有效范围内
            percent = max(0, min(99.5, percent))
            
            # 计算目标帧位置，确保在有效范围内
            target_frame = int((percent / 100.0) * self.total_frames)
            
            # 确保目标帧位置不超出总帧数
            target_frame = max(0, min(target_frame, self.total_frames - 1))
            
            print(f"尝试跳转到位置: {target_frame}/{self.total_frames} ({percent:.1f}%)")
            
            # 设置文件位置
            self.wf.setpos(target_frame)
            self.current_frame = target_frame
            
            # 立即更新进度显示
            if self.progress_callback:
                self.progress_callback(percent)
                
            print(f"成功跳转到位置: {target_frame}")
            self.seeking = False
            return True
        except Exception as e:
            print(f"跳转音频位置时出错: {str(e)}")
            # 发生错误时，尝试跳转到文件开头
            try:
                self.wf.rewind()
                self.current_frame = 0
                if self.progress_callback:
                    self.progress_callback(0)
                print("出错后重置到文件开头")
            except:
                pass
            
            self.seeking = False
            return False
    
    def is_playing_file(self, file_path):
        """
        检查是否正在播放指定文件
        
        Args:
            file_path: 要检查的文件路径
            
        Returns:
            bool: 如果正在播放指定文件，返回True
        """
        return self.playing and self._file_path == file_path
    
    def _play_audio(self, file_path: str):
        """
        实际播放音频的内部方法
        
        Args:
            file_path: 音频文件路径
        """
        try:
            # 检查文件格式，如果不是WAV格式，则转换
            file_ext = os.path.splitext(file_path)[1].lower()
            wav_file = file_path
            
            if file_ext != '.wav':
                # 转换为临时WAV文件
                self.temp_wav = self._convert_to_wav(file_path)
                if not self.temp_wav:
                    print(f"无法转换音频文件: {file_path}")
                    self.playing = False
                    return
                wav_file = self.temp_wav
            
            # 使用wave模块打开WAV文件
            try:
                self.wf = wave.open(wav_file, 'rb')
            except Exception as e:
                print(f"打开WAV文件失败: {str(e)}")
                self.playing = False
                return
                
            # 初始化PyAudio
            try:
                self.pyaudio_instance = pyaudio.PyAudio()
            except Exception as e:
                print(f"初始化PyAudio失败: {str(e)}")
                self.playing = False
                return
            
            # 获取音频信息
            self.total_frames = self.wf.getnframes()
            self.frame_rate = self.wf.getframerate()
            self.current_frame = 0
            
            # 打开音频流
            try:
                self.stream = self.pyaudio_instance.open(
                    format=self.pyaudio_instance.get_format_from_width(self.wf.getsampwidth()),
                    channels=self.wf.getnchannels(),
                    rate=self.frame_rate,
                    output=True
                )
            except Exception as e:
                print(f"打开音频流失败: {str(e)}")
                self.playing = False
                self._cleanup_resources()
                return
            
            # 读取音频数据块并播放
            chunk_size = self._chunk_size
            data = self.wf.readframes(chunk_size)
            last_update_time = time.time()
            last_frame_position = 0
            no_change_count = 0  # 定义位置未变化计数器
            
            print(f"开始播放循环，总帧数: {self.total_frames}, 帧率: {self.frame_rate}")
            
            # 主播放循环
            play_count = 0
            
            while len(data) > 0 and not self.stop_event.is_set() and self.playing:
                play_count += 1
                if play_count % 100 == 0:
                    print(f"播放进度: {self.current_frame}/{self.total_frames}, "
                          f"{(self.current_frame/self.total_frames*100):.1f}%")
                
                if not self.seeking:  # 避免跳转时更新位置
                    try:
                        self.stream.write(data)
                        
                        # 更新当前位置（只有在没有seek操作时）
                        self.current_frame = self.wf.tell()
                        
                        # 检查位置是否正在变化，避免卡住
                        if self.current_frame == last_frame_position:
                            # 如果位置没有变化，可能出现了文件读取问题
                            no_change_count += 1
                            if no_change_count > 5:
                                print(f"警告: 播放位置未变化，尝试读取下一块")
                                no_change_count = 0
                        else:
                            no_change_count = 0
                            
                        last_frame_position = self.current_frame
                        
                        # 调用进度回调，但限制更新频率（每100ms更新一次）
                        current_time = time.time()
                        if self.progress_callback and self.total_frames > 0 and (current_time - last_update_time) >= 0.1:
                            percent = (self.current_frame * 100) / self.total_frames
                            self.progress_callback(percent)
                            last_update_time = current_time
                        
                        # 读取下一块数据
                        data = self.wf.readframes(chunk_size)
                    except Exception as e:
                        print(f"播放数据块时出错: {str(e)}")
                        break
                else:
                    # 如果正在跳转，暂停一下，然后重新读取当前位置的数据
                    time.sleep(0.01)
                    try:
                        # 跳转后需要重新读取数据
                        data = self.wf.readframes(chunk_size)
                    except Exception as e:
                        print(f"跳转后读取数据出错: {str(e)}")
                        data = b""  # 空数据会导致循环退出
            
            print(f"播放完成或被停止，播放了{play_count}个数据块")
            
            # 播放结束或停止，发送100%进度信号
            if self.progress_callback and not self.seeking and not self.stop_event.is_set():
                self.progress_callback(100.0)
            
            # 标记播放完成
            if not self.stop_event.is_set():
                self.playing = False
                
        except Exception as e:
            print(f"播放音频时出错: {str(e)}")
            import traceback
            traceback.print_exc()
            self.playing = False
        finally:
            self._cleanup_resources()
            
    def _cleanup_resources(self):
        """清理资源"""
        try:
            if self.stream:
                try:
                    self.stream.stop_stream()
                    self.stream.close()
                except Exception as e:
                    print(f"关闭音频流时出错: {str(e)}")
                finally:
                    self.stream = None
            
            if self.pyaudio_instance:
                try:
                    self.pyaudio_instance.terminate()
                except Exception as e:
                    print(f"终止PyAudio实例时出错: {str(e)}")
                finally:
                    self.pyaudio_instance = None
                
            if self.wf:
                try:
                    self.wf.close()
                except Exception as e:
                    print(f"关闭WAV文件时出错: {str(e)}")
                finally:
                    self.wf = None
        except Exception as e:
            print(f"清理资源时出错: {str(e)}")
    
    def _convert_to_wav(self, file_path: str) -> Optional[str]:
        """
        将音频文件转换为WAV格式
        
        Args:
            file_path: 输入音频文件路径
            
        Returns:
            str: 转换后的WAV文件路径，如果转换失败则返回None
        """
        # 检查soundfile是否可用
        if not SOUNDFILE_AVAILABLE or not NUMPY_AVAILABLE:
            print("警告: 缺少soundfile或numpy模块，尝试使用wave直接加载")
            # 尝试直接打开，确认是否为WAV格式
            try:
                test_wf = wave.open(file_path, 'rb')
                test_wf.close()
                return file_path  # 如果可以直接打开，则直接返回原文件
            except Exception:
                print("错误: 无法直接打开音频文件，且缺少格式转换依赖")
                return None
            
        try:
            # 创建临时文件
            temp_fd, temp_wav = tempfile.mkstemp(suffix='.wav')
            os.close(temp_fd)
            
            # 使用soundfile进行转换
            try:
                print(f"正在转换音频文件: {file_path} -> {temp_wav}")
                data, samplerate = sf.read(file_path)
                sf.write(temp_wav, data, samplerate)
                return temp_wav
            except Exception as e:
                print(f"使用soundfile转换音频失败: {str(e)}")
                # 尝试使用librosa(如果可用)
                try:
                    import librosa
                    import soundfile as sf
                    print("尝试使用librosa加载音频...")
                    y, sr = librosa.load(file_path, sr=None)
                    sf.write(temp_wav, y, sr)
                    return temp_wav
                except ImportError:
                    print("librosa模块不可用，无法进行备选转换")
                except Exception as e2:
                    print(f"使用librosa转换音频失败: {str(e2)}")
                    
                # 如果所有转换方法都失败，返回None
                return None
        except Exception as e:
            print(f"转换音频文件时出错: {str(e)}")
            return None

class FileOperations:
    """文件操作工具类"""
    
    def copy_file(self, src: str, dst: str):
        """
        复制文件
        
        Args:
            src: 源文件路径
            dst: 目标文件路径
        """
        # 检查源文件是否存在
        if not os.path.exists(src):
            raise FileNotFoundError(f"源文件不存在: {src}")
            
        # 确保目标目录存在
        dst_dir = os.path.dirname(dst)
        if dst_dir and not os.path.exists(dst_dir):
            os.makedirs(dst_dir)
            
        shutil.copy2(src, dst)
    
    def delete_file(self, path: str) -> bool:
        """
        删除文件
        
        Args:
            path: 文件路径
            
        Returns:
            删除成功返回True，否则返回False
        """
        try:
            if os.path.exists(path):
                os.remove(path)
                return True
            return False
        except Exception as e:
            print(f"删除文件时出错: {str(e)}")
            return False
    
    def ensure_dir(self, directory: str) -> bool:
        """
        确保目录存在，不存在则创建
        
        Args:
            directory: 目录路径
            
        Returns:
            操作成功返回True，否则返回False
        """
        try:
            if not os.path.exists(directory):
                os.makedirs(directory)
            return True
        except Exception as e:
            print(f"创建目录时出错: {str(e)}")
            return False
    
    @staticmethod
    def get_file_size(path: str) -> str:
        """
        获取文件大小的人类可读表示
        
        Args:
            path: 文件路径
            
        Returns:
            格式化的文件大小字符串
        """
        if not os.path.exists(path) or not os.path.isfile(path):
            return "不存在"
            
        size_bytes = os.path.getsize(path)
        
        for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
            if size_bytes < 1024.0 or unit == 'TB':
                break
            size_bytes /= 1024.0
            
        return f"{size_bytes:.2f} {unit}"