diff --git a/audio_processor.py b/audio_processor.py index 2a013c9..b8a2289 100644 --- a/audio_processor.py +++ b/audio_processor.py @@ -1,18 +1,48 @@ """ 音频处理模块 支持从MP4、MOV、AVI等视频文件中提取音频，以及处理MP3、WAV等音频文件 +改进版：增强FFmpeg查找逻辑，提供更友好的错误信息和解决方案 """ import os +import sys import subprocess import tempfile +import shutil from pathlib import Path -from typing import Optional, Tuple +from typing import Optional, Tuple, List import logging logger = logging.getLogger(__name__) +class FFmpegNotFoundError(Exception): + """FFmpeg未找到错误""" + + def __init__(self, message: str, solutions: List[str] = None): + super().__init__(message) + self.solutions = solutions or [] + + def get_detailed_message(self) -> str: + """获取详细的错误信息""" + msg = str(self) + if self.solutions: + msg += "\n\n解决方案:\n" + for i, solution in enumerate(self.solutions, 1): + msg += f" {i}. {solution}\n" + return msg + + +class FFmpegError(Exception): + """FFmpeg执行错误""" + + def __init__(self, command: List[str], return_code: int, stderr: str): + self.command = command + self.return_code = return_code + self.stderr = stderr + super().__init__(f"FFmpeg执行失败 (返回码: {return_code})") + + class AudioProcessor: """ 音频处理器类 @@ -20,34 +50,177 @@ class AudioProcessor: """ # 支持的视频格式 - SUPPORTED_VIDEO_FORMATS = {'.mp4', '.mov', '.avi', '.mkv', '.flv', '.wmv', '.webm'} + SUPPORTED_VIDEO_FORMATS = {'.mp4', '.mov', '.avi', '.mkv', '.flv', '.wmv', '.webm', '.m4v', '.mpeg', '.mpg', '.3gp'} # 支持的音频格式 - SUPPORTED_AUDIO_FORMATS = {'.mp3', '.wav', '.m4a', '.flac', '.aac', '.ogg', '.wma'} + SUPPORTED_AUDIO_FORMATS = {'.mp3', '.wav', '.m4a', '.flac', '.aac', '.ogg', '.wma', '.opus', '.amr'} + + # FFmpeg详细安装指南 + FFMPEG_INSTALL_GUIDE = { + 'windows': [ + "方法1: 使用 winget 安装 (推荐)", + " 以管理员身份打开 PowerShell，运行:", + " winget install Gyan.FFmpeg", + "", + "方法2: 使用 Chocolatey 安装", + " choco install ffmpeg", + "", + "方法3: 手动安装", + " 1. 访问 https://www.gyan.dev/ffmpeg/builds/", + " 2. 下载 'ffmpeg-release-full.7z'", + " 3. 解压到 C:\\ffmpeg", + " 4. 将 C:\\ffmpeg\\bin 添加到系统 PATH 环境变量", + "", + "安装后需要重启命令行/IDE才能生效！" + ], + 'macos': [ + "方法1: 使用 Homebrew 安装 (推荐)", + " brew install ffmpeg", + "", + "方法2: 使用 MacPorts 安装", + " sudo port install ffmpeg", + "", + "方法3: 手动安装", + " 从 https://evermeet.cx/ffmpeg/ 下载并安装" + ], + 'linux': [ + "Ubuntu/Debian:", + " sudo apt update", + " sudo apt install ffmpeg", + "", + "Fedora/RHEL:", + " sudo dnf install ffmpeg", + "", + "Arch Linux:", + " sudo pacman -S ffmpeg" + ] + } - def __init__(self, ffmpeg_path: Optional[str] = None): + def __init__(self, ffmpeg_path: Optional[str] = None, auto_download: bool = False): """ 初始化音频处理器 Args: ffmpeg_path: FFmpeg可执行文件路径，如为None则尝试从PATH中查找 + auto_download: 是否自动下载FFmpeg（预留功能） """ - self.ffmpeg_path = ffmpeg_path or self._find_ffmpeg() - if not self.ffmpeg_path: - raise RuntimeError("未找到FFmpeg，请确保FFmpeg已安装并添加到系统PATH中") + self.ffmpeg_path = None + self.ffprobe_path = None + self._ffmpeg_version = None + + # 尝试查找 FFmpeg + if ffmpeg_path: + # 用户指定了路径 + self.ffmpeg_path = self._validate_ffmpeg_path(ffmpeg_path) + if not self.ffmpeg_path: + raise FFmpegNotFoundError( + f"指定的FFmpeg路径无效: {ffmpeg_path}", + self._get_install_solutions() + ) + else: + # 自动查找 + self.ffmpeg_path = self._find_ffmpeg() + if not self.ffmpeg_path: + raise FFmpegNotFoundError( + "未找到FFmpeg！FFmpeg是音视频处理必需的工具。", + self._get_install_solutions() + ) + + # 查找 ffprobe + self.ffprobe_path = self._find_ffprobe() + + # 验证 FFmpeg 是否可用 + self._validate_ffmpeg() + + logger.info(f"FFmpeg初始化成功: {self.ffmpeg_path}") + logger.info(f"FFmpeg版本: {self._ffmpeg_version}") + + def _get_platform(self) -> str: + """ + 获取当前平台 + + Returns: + 'windows', 'macos', 或 'linux' + """ + if sys.platform == 'win32': + return 'windows' + elif sys.platform == 'darwin': + return 'macos' + else: + return 'linux' + + def _get_install_solutions(self) -> List[str]: + """ + 获取安装解决方案列表 + + Returns: + 解决方案列表 + """ + platform = self._get_platform() + return self.FFMPEG_INSTALL_GUIDE.get(platform, []) + + def _validate_ffmpeg_path(self, path: str) -> Optional[str]: + """ + 验证 FFmpeg 路径 + + Args: + path: FFmpeg 可执行文件或目录路径 + + Returns: + 验证后的完整路径或 None + """ + path = Path(path).resolve() + + if path.is_file(): + # 是文件路径 + if path.name.lower() in ['ffmpeg', 'ffmpeg.exe']: + if self._test_ffmpeg_executable(str(path)): + return str(path) + elif path.is_dir(): + # 是目录路径 + ffmpeg_exe = 'ffmpeg.exe' if sys.platform == 'win32' else 'ffmpeg' + candidate = path / ffmpeg_exe + if candidate.exists() and self._test_ffmpeg_executable(str(candidate)): + return str(candidate) + + return None + + def _test_ffmpeg_executable(self, path: str) -> bool: + """ + 测试 FFmpeg 可执行文件是否可用 + + Args: + path: FFmpeg 可执行文件路径 + + Returns: + 是否可用 + """ + try: + result = subprocess.run( + [path, '-version'], + capture_output=True, + text=True, + timeout=10 + ) + return result.returncode == 0 + except (subprocess.SubprocessError, FileNotFoundError, OSError): + return False def _find_ffmpeg(self) -> Optional[str]: """ 在系统中查找FFmpeg可执行文件 查找顺序： - 1. 环境变量 FFMPEG_PATH 或 FFMPEG_BIN + 1. 环境变量 FFMPEG_PATH, FFMPEG_BIN, FFMPEG 2. 常见的FFmpeg安装位置 3. 系统PATH环境变量 + 4. 项目目录下的 ffmpeg 文件夹 Returns: FFmpeg路径或None """ - ffmpeg_exe = 'ffmpeg.exe' if os.name == 'nt' else 'ffmpeg' + ffmpeg_exe = 'ffmpeg.exe' if sys.platform == 'win32' else 'ffmpeg' + + logger.info("正在查找FFmpeg...") # 1. 检查环境变量 env_paths = [ @@ -58,60 +231,225 @@ def _find_ffmpeg(self) -> Optional[str]: for env_path in env_paths: if env_path: - env_path = os.path.expanduser(env_path) - # 如果是目录，拼接ffmpeg.exe - if os.path.isdir(env_path): - candidate = os.path.join(env_path, ffmpeg_exe) - if os.path.exists(candidate): - return candidate - # 如果是文件路径 - elif os.path.exists(env_path): - return env_path - - # 2. 检查常见的FFmpeg安装位置（Windows） - if os.name == 'nt': - common_paths = [ + validated = self._validate_ffmpeg_path(env_path) + if validated: + logger.info(f"从环境变量找到FFmpeg: {validated}") + return validated + + # 2. 检查常见的FFmpeg安装位置 + common_locations = self._get_common_locations() + + for location in common_locations: + if location and os.path.exists(location): + validated = self._validate_ffmpeg_path(location) + if validated: + logger.info(f"从常见位置找到FFmpeg: {validated}") + return validated + + # 3. 检查项目目录下的 ffmpeg 文件夹 + project_ffmpeg = Path(__file__).parent / 'ffmpeg' + if project_ffmpeg.exists(): + validated = self._validate_ffmpeg_path(str(project_ffmpeg)) + if validated: + logger.info(f"从项目目录找到FFmpeg: {validated}") + return validated + + # 4. 检查系统PATH环境变量 + path_dirs = os.environ.get('PATH', '').split(os.pathsep) + + for path_dir in path_dirs: + if not path_dir: + continue + try: + path_dir = Path(path_dir).resolve() + if path_dir.exists(): + ffmpeg_file = path_dir / ffmpeg_exe + if ffmpeg_file.exists(): + if self._test_ffmpeg_executable(str(ffmpeg_file)): + logger.info(f"从系统PATH找到FFmpeg: {ffmpeg_file}") + return str(ffmpeg_file) + except Exception: + continue + + # 5. 尝试使用 shutil.which + which_result = shutil.which('ffmpeg') + if which_result: + if self._test_ffmpeg_executable(which_result): + logger.info(f"从shutil.which找到FFmpeg: {which_result}") + return which_result + + logger.warning("未找到FFmpeg") + return None + + def _get_common_locations(self) -> List[str]: + """ + 获取常见的FFmpeg安装位置 + + Returns: + 路径列表 + """ + locations = [] + + if sys.platform == 'win32': + # Windows 常见位置 + user_profile = os.environ.get('USERPROFILE', '') + local_app_data = os.environ.get('LOCALAPPDATA', '') + program_files = os.environ.get('PROGRAMFILES', '') + program_files_x86 = os.environ.get('PROGRAMFILES(X86)', '') + + locations.extend([ r'C:\ffmpeg\bin', - r'C:\Program Files\ffmpeg\bin', - r'C:\Program Files (x86)\ffmpeg\bin', - os.path.join(os.environ.get('LOCALAPPDATA', ''), 'ffmpeg', 'bin'), - os.path.join(os.environ.get('USERPROFILE', ''), 'ffmpeg', 'bin'), - os.path.join(os.environ.get('USERPROFILE', ''), 'scoop', 'shims'), - os.path.join(os.environ.get('USERPROFILE', ''), 'scoop', 'apps', 'ffmpeg', 'current', 'bin'), + r'C:\ffmpeg', + os.path.join(program_files, 'ffmpeg', 'bin'), + os.path.join(program_files, 'ffmpeg'), + os.path.join(program_files_x86, 'ffmpeg', 'bin'), + os.path.join(program_files_x86, 'ffmpeg'), + os.path.join(local_app_data, 'ffmpeg', 'bin'), + os.path.join(local_app_data, 'ffmpeg'), + os.path.join(user_profile, 'ffmpeg', 'bin'), + os.path.join(user_profile, 'ffmpeg'), + os.path.join(user_profile, 'scoop', 'shims'), + os.path.join(user_profile, 'scoop', 'apps', 'ffmpeg', 'current', 'bin'), r'C:\ProgramData\chocolatey\bin', - ] - - for path in common_paths: - candidate = os.path.join(path, ffmpeg_exe) - if os.path.exists(candidate): - return candidate + r'C:\tools\ffmpeg\bin', + r'C:\tools\ffmpeg', + ]) - # 检查 WinGet 安装的 FFmpeg（Gyan.FFmpeg） - winget_packages = os.path.join( - os.environ.get('LOCALAPPDATA', ''), - 'Microsoft', 'WinGet', 'Packages' - ) + # WinGet 包位置 + winget_packages = os.path.join(local_app_data, 'Microsoft', 'WinGet', 'Packages') if os.path.exists(winget_packages): try: - # 查找包含 ffmpeg 的文件夹 for item in os.listdir(winget_packages): item_path = os.path.join(winget_packages, item) if os.path.isdir(item_path) and 'ffmpeg' in item.lower(): - # 在该目录下查找 ffmpeg.exe + locations.append(item_path) + # 也检查子目录 for root, dirs, files in os.walk(item_path): - if ffmpeg_exe in files: - return os.path.join(root, ffmpeg_exe) + if 'bin' in dirs: + locations.append(os.path.join(root, 'bin')) except Exception: pass - # 3. 检查系统PATH环境变量 - for path in os.environ.get('PATH', '').split(os.pathsep): - ffmpeg_full_path = os.path.join(path, ffmpeg_exe) - if os.path.exists(ffmpeg_full_path): - return ffmpeg_full_path + elif sys.platform == 'darwin': + # macOS 常见位置 + locations.extend([ + '/usr/local/bin', + '/usr/local/ffmpeg/bin', + '/opt/homebrew/bin', + '/opt/homebrew/ffmpeg/bin', + '/Applications/ffmpeg', + os.path.expanduser('~/Applications/ffmpeg'), + ]) + + else: + # Linux 常见位置 + locations.extend([ + '/usr/bin', + '/usr/local/bin', + '/opt/ffmpeg/bin', + '/snap/bin', + ]) + + return locations + + def _find_ffprobe(self) -> Optional[str]: + """ + 查找 ffprobe 工具 + + Returns: + ffprobe路径或None + """ + if not self.ffmpeg_path: + return None + + ffprobe_exe = 'ffprobe.exe' if sys.platform == 'win32' else 'ffprobe' + + # 首先在 FFmpeg 同一目录查找 + ffmpeg_dir = Path(self.ffmpeg_path).parent + ffprobe_candidate = ffmpeg_dir / ffprobe_exe + + if ffprobe_candidate.exists(): + try: + result = subprocess.run( + [str(ffprobe_candidate), '-version'], + capture_output=True, + text=True, + timeout=5 + ) + if result.returncode == 0: + return str(ffprobe_candidate) + except Exception: + pass + + # 在系统 PATH 中查找 + which_result = shutil.which('ffprobe') + if which_result: + return which_result return None + def _validate_ffmpeg(self): + """ + 验证 FFmpeg 是否可用并获取版本信息 + """ + try: + result = subprocess.run( + [self.ffmpeg_path, '-version'], + capture_output=True, + text=True, + timeout=10 + ) + + if result.returncode == 0: + # 解析版本信息 + first_line = result.stdout.split('\n')[0] if result.stdout else '' + # 典型输出: "ffmpeg version 6.0-full_build-www.gyan.dev Copyright (c) 2000-2023 the FFmpeg developers" + import re + version_match = re.search(r'ffmpeg version (\S+)', first_line, re.IGNORECASE) + if version_match: + self._ffmpeg_version = version_match.group(1) + else: + self._ffmpeg_version = 'unknown' + else: + raise FFmpegError( + [self.ffmpeg_path, '-version'], + result.returncode, + result.stderr + ) + + except subprocess.TimeoutExpired: + raise FFmpegNotFoundError( + "FFmpeg响应超时", + ["请检查FFmpeg是否正确安装", "尝试重新安装FFmpeg"] + ) + except FileNotFoundError as e: + raise FFmpegNotFoundError( + f"FFmpeg可执行文件不存在: {self.ffmpeg_path}", + self._get_install_solutions() + ) from e + except OSError as e: + if 'WinError 2' in str(e) or 'The system cannot find the file specified' in str(e): + raise FFmpegNotFoundError( + f"无法执行FFmpeg: {e}\n路径: {self.ffmpeg_path}", + self._get_install_solutions() + ) from e + else: + raise + + def get_ffmpeg_info(self) -> dict: + """ + 获取 FFmpeg 信息 + + Returns: + 包含FFmpeg信息的字典 + """ + return { + 'ffmpeg_path': self.ffmpeg_path, + 'ffprobe_path': self.ffprobe_path, + 'version': self._ffmpeg_version, + 'platform': self._get_platform(), + } + def is_supported_format(self, file_path: str) -> bool: """ 检查文件是否为支持的音视频格式 @@ -138,6 +476,71 @@ def is_video_file(self, file_path: str) -> bool: ext = Path(file_path).suffix.lower() return ext in self.SUPPORTED_VIDEO_FORMATS + def _run_ffmpeg_command( + self, + cmd: List[str], + description: str = "执行FFmpeg命令", + timeout: Optional[int] = None, + ) -> Tuple[int, str, str]: + """ + 执行 FFmpeg 命令 + + Args: + cmd: 命令参数列表 + description: 命令描述（用于错误信息） + timeout: 超时时间（秒） + + Returns: + (返回码, 标准输出, 标准错误) + """ + cmd = [self.ffmpeg_path] + cmd + + logger.debug(f"执行FFmpeg命令: {' '.join(cmd)}") + + try: + # 使用 startupinfo 隐藏 Windows 控制台窗口 + startupinfo = None + if sys.platform == 'win32': + startupinfo = subprocess.STARTUPINFO() + startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW + startupinfo.wShowWindow = subprocess.SW_HIDE + + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=timeout, + startupinfo=startupinfo, + ) + + if result.returncode != 0: + logger.error(f"FFmpeg命令失败: {description}") + logger.error(f"命令: {' '.join(cmd)}") + logger.error(f"错误输出: {result.stderr}") + + raise FFmpegError(cmd, result.returncode, result.stderr) + + return result.returncode, result.stdout, result.stderr + + except FileNotFoundError as e: + raise FFmpegNotFoundError( + f"执行FFmpeg时找不到文件: {e}\n命令: {' '.join(cmd)}", + self._get_install_solutions() + ) from e + except OSError as e: + if 'WinError 2' in str(e) or 'The system cannot find the file specified' in str(e): + raise FFmpegNotFoundError( + f"无法执行FFmpeg: {e}\n" + f"FFmpeg路径: {self.ffmpeg_path}\n" + f"命令: {' '.join(cmd)}", + self._get_install_solutions() + ) from e + else: + raise + except subprocess.TimeoutExpired: + logger.error(f"FFmpeg命令超时: {description}") + raise RuntimeError(f"FFmpeg命令超时: {description}") + def extract_audio_from_video( self, video_path: str, @@ -156,9 +559,6 @@ def extract_audio_from_video( Returns: 提取的音频文件路径 - - Raises: - RuntimeError: 提取失败时抛出 """ video_path = Path(video_path).resolve() if not video_path.exists(): @@ -171,33 +571,26 @@ def extract_audio_from_video( output_path = Path(output_path).resolve() + logger.info(f"正在从视频提取音频: {video_path.name} -> {output_path.name}") + # 构建FFmpeg命令 + # -vn: 不处理视频 + # -acodec pcm_s16le: PCM 16位小端编码 + # -ar: 采样率 + # -ac: 声道数 + # -y: 覆盖输出文件 cmd = [ - self.ffmpeg_path, '-i', str(video_path), - '-vn', # 只处理音频 - '-acodec', 'pcm_s16le', # PCM编码，16位 - '-ar', str(sample_rate), # 采样率 - '-ac', str(channels), # 声道数 - '-y', # 覆盖输出文件 + '-vn', + '-acodec', 'pcm_s16le', + '-ar', str(sample_rate), + '-ac', str(channels), + '-y', str(output_path) ] - logger.info(f"正在从视频提取音频: {video_path} -> {output_path}") - logger.info(f"执行命令: {' '.join(cmd)}") - try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - check=False - ) - - if result.returncode != 0: - error_msg = f"音频提取失败: {result.stderr}" - logger.error(error_msg) - raise RuntimeError(error_msg) + self._run_ffmpeg_command(cmd, f"提取音频: {video_path.name}") if not output_path.exists(): raise RuntimeError(f"音频提取失败，输出文件不存在: {output_path}") @@ -205,9 +598,9 @@ def extract_audio_from_video( logger.info(f"音频提取成功: {output_path}") return str(output_path) - except Exception as e: - logger.exception(f"音频提取过程中发生错误: {e}") - raise + except FFmpegError as e: + logger.error(f"音频提取失败: {e}") + raise RuntimeError(f"音频提取失败: {e.stderr}") from e def convert_audio_format( self, @@ -236,6 +629,7 @@ def convert_audio_format( if (input_path.suffix.lower() == '.wav' and sample_rate == 16000 and channels == 1): + logger.info(f"音频已是所需格式，跳过转换: {input_path}") return str(input_path) # 如果未指定输出路径，创建临时文件 @@ -245,9 +639,10 @@ def convert_audio_format( output_path = Path(output_path).resolve() + logger.info(f"正在转换音频: {input_path.name} -> {output_path.name}") + # 构建FFmpeg命令 cmd = [ - self.ffmpeg_path, '-i', str(input_path), '-acodec', 'pcm_s16le', '-ar', str(sample_rate), @@ -256,27 +651,15 @@ def convert_audio_format( str(output_path) ] - logger.info(f"正在转换音频: {input_path} -> {output_path}") - try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - check=False - ) - - if result.returncode != 0: - error_msg = f"音频转换失败: {result.stderr}" - logger.error(error_msg) - raise RuntimeError(error_msg) + self._run_ffmpeg_command(cmd, f"转换音频: {input_path.name}") logger.info(f"音频转换成功: {output_path}") return str(output_path) - except Exception as e: - logger.exception(f"音频转换过程中发生错误: {e}") - raise + except FFmpegError as e: + logger.error(f"音频转换失败: {e}") + raise RuntimeError(f"音频转换失败: {e.stderr}") from e def get_audio_duration(self, audio_path: str) -> float: """ @@ -292,34 +675,66 @@ def get_audio_duration(self, audio_path: str) -> float: if not audio_path.exists(): raise FileNotFoundError(f"音频文件不存在: {audio_path}") - # 使用ffprobe获取时长 - ffprobe_path = self.ffmpeg_path.replace('ffmpeg', 'ffprobe') - cmd = [ - ffprobe_path, - '-v', 'error', - '-show_entries', 'format=duration', - '-of', 'default=noprint_wrappers=1:nokey=1', - str(audio_path) - ] - + # 优先使用 ffprobe + if self.ffprobe_path: + try: + cmd = [ + self.ffprobe_path, + '-v', 'error', + '-show_entries', 'format=duration', + '-of', 'default=noprint_wrappers=1:nokey=1', + str(audio_path) + ] + + # 使用 startupinfo 隐藏 Windows 控制台窗口 + startupinfo = None + if sys.platform == 'win32': + startupinfo = subprocess.STARTUPINFO() + startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW + startupinfo.wShowWindow = subprocess.SW_HIDE + + result = subprocess.run( + cmd, + capture_output=True, + text=True, + timeout=10, + startupinfo=startupinfo, + ) + + if result.returncode == 0: + duration_str = result.stdout.strip() + if duration_str: + return float(duration_str) + except Exception as e: + logger.warning(f"使用ffprobe获取时长失败: {e}，将使用备用方法") + + # 备用方法：使用 ffmpeg 解析 try: - result = subprocess.run( - cmd, - capture_output=True, - text=True, - check=False - ) + cmd = [ + '-i', str(audio_path), + '-f', 'null', + '-' + ] - if result.returncode == 0: - duration = float(result.stdout.strip()) - return duration - else: - logger.warning(f"获取音频时长失败: {result.stderr}") - return 0.0 + _, _, stderr = self._run_ffmpeg_command(cmd, f"获取音频时长: {audio_path.name}") + + # 解析输出中的 Duration + import re + match = re.search(r'Duration:\s*(\d{2}):(\d{2}):(\d{2})\.(\d+)', stderr) + if match: + hours = int(match.group(1)) + minutes = int(match.group(2)) + seconds = int(match.group(3)) + milliseconds = int(match.group(4)) + + total_seconds = hours * 3600 + minutes * 60 + seconds + milliseconds / 100.0 + return total_seconds except Exception as e: - logger.exception(f"获取音频时长时发生错误: {e}") + logger.error(f"获取音频时长失败: {e}") return 0.0 + + return 0.0 def process_media_file( self, @@ -342,11 +757,18 @@ def process_media_file( raise FileNotFoundError(f"媒体文件不存在: {media_path}") if not self.is_supported_format(str(media_path)): - raise ValueError(f"不支持的文件格式: {media_path.suffix}") + raise ValueError( + f"不支持的文件格式: {media_path.suffix}\n" + f"支持的视频格式: {', '.join(sorted(self.SUPPORTED_VIDEO_FORMATS))}\n" + f"支持的音频格式: {', '.join(sorted(self.SUPPORTED_AUDIO_FORMATS))}" + ) is_video = self.is_video_file(str(media_path)) is_temp = temp_output + logger.info(f"处理媒体文件: {media_path.name}") + logger.info(f"文件类型: {'视频' if is_video else '音频'}") + if is_video: # 从视频提取音频 audio_path = self.extract_audio_from_video(str(media_path)) @@ -369,3 +791,73 @@ def cleanup_temp_file(self, file_path: str): logger.info(f"已清理临时文件: {file_path}") except Exception as e: logger.warning(f"清理临时文件失败: {file_path}, 错误: {e}") + + def get_media_info(self, media_path: str) -> dict: + """ + 获取媒体文件信息 + + Args: + media_path: 媒体文件路径 + + Returns: + 包含媒体信息的字典 + """ + media_path = Path(media_path).resolve() + if not media_path.exists(): + raise FileNotFoundError(f"媒体文件不存在: {media_path}") + + info = { + 'path': str(media_path), + 'name': media_path.name, + 'extension': media_path.suffix.lower(), + 'is_video': self.is_video_file(str(media_path)), + 'is_audio': not self.is_video_file(str(media_path)) and self.is_supported_format(str(media_path)), + 'size_bytes': media_path.stat().st_size, + 'size_mb': round(media_path.stat().st_size / (1024 * 1024), 2), + } + + # 尝试获取时长 + try: + duration = self.get_audio_duration(str(media_path)) + info['duration_seconds'] = duration + info['duration_formatted'] = self._format_duration(duration) + except Exception: + pass + + return info + + def _format_duration(self, seconds: float) -> str: + """ + 格式化时长 + + Args: + seconds: 秒数 + + Returns: + 格式化的时长字符串 + """ + hours = int(seconds // 3600) + minutes = int((seconds % 3600) // 60) + secs = int(seconds % 60) + + if hours > 0: + return f"{hours}:{minutes:02d}:{secs:02d}" + else: + return f"{minutes}:{secs:02d}" + + +# 便捷函数：检查FFmpeg是否可用 +def check_ffmpeg_available() -> Tuple[bool, Optional[str]]: + """ + 检查FFmpeg是否可用 + + Returns: + (是否可用, 错误信息或None) + """ + try: + processor = AudioProcessor() + return True, None + except FFmpegNotFoundError as e: + return False, e.get_detailed_message() + except Exception as e: + return False, str(e) diff --git a/dependency_checker.py b/dependency_checker.py new file mode 100644 index 0000000..42ecd50 --- /dev/null +++ b/dependency_checker.py @@ -0,0 +1,569 @@ +""" +依赖检查模块 +在程序启动时自动验证所有必需依赖 +提供详细的错误信息和安装指南 +""" + +import os +import sys +import importlib.util +from typing import Dict, List, Tuple, Optional +from dataclasses import dataclass, field +from enum import Enum + + +class DependencyStatus(Enum): + """依赖状态枚举""" + OK = "ok" + MISSING = "missing" + VERSION_TOO_OLD = "version_too_old" + ERROR = "error" + + +@dataclass +class DependencyInfo: + """依赖信息""" + name: str + display_name: str + required: bool = True + installed: bool = False + version: Optional[str] = None + status: DependencyStatus = DependencyStatus.MISSING + error_message: Optional[str] = None + install_guide: List[str] = field(default_factory=list) + + def is_ok(self) -> bool: + """检查是否满足要求""" + return self.status == DependencyStatus.OK + + +class DependencyChecker: + """ + 依赖检查器 + 验证所有必需的依赖是否已安装 + """ + + # 必需的 Python 包 + REQUIRED_PACKAGES = [ + { + 'name': 'torch', + 'display_name': 'PyTorch', + 'required': True, + 'min_version': '2.0.0', + 'install_guide': [ + "GPU版本 (推荐):", + " pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118", + "", + "CPU版本:", + " pip3 install torch torchvision torchaudio", + ] + }, + { + 'name': 'whisper', + 'display_name': 'OpenAI Whisper', + 'required': True, + 'min_version': None, + 'install_guide': [ + "pip install openai-whisper", + ] + }, + { + 'name': 'numpy', + 'display_name': 'NumPy', + 'required': True, + 'min_version': '1.20.0', + 'install_guide': [ + "pip install numpy", + ] + }, + { + 'name': 'tqdm', + 'display_name': 'tqdm (进度条)', + 'required': False, + 'min_version': None, + 'install_guide': [ + "pip install tqdm", + ] + }, + ] + + # 系统工具 + SYSTEM_TOOLS = [ + { + 'name': 'ffmpeg', + 'display_name': 'FFmpeg', + 'required': True, + 'install_guide_windows': [ + "方法1: 使用 winget 安装 (推荐)", + " 以管理员身份打开 PowerShell，运行:", + " winget install Gyan.FFmpeg", + "", + "方法2: 使用 Chocolatey 安装", + " choco install ffmpeg", + "", + "方法3: 手动安装", + " 1. 访问 https://www.gyan.dev/ffmpeg/builds/", + " 2. 下载 'ffmpeg-release-full.7z'", + " 3. 解压到 C:\\ffmpeg", + " 4. 将 C:\\ffmpeg\\bin 添加到系统 PATH 环境变量", + "", + "⚠️ 安装后需要重启命令行/IDE才能生效！", + ], + 'install_guide_macos': [ + "方法1: 使用 Homebrew 安装 (推荐)", + " brew install ffmpeg", + "", + "方法2: 使用 MacPorts 安装", + " sudo port install ffmpeg", + "", + "方法3: 手动安装", + " 从 https://evermeet.cx/ffmpeg/ 下载并安装", + ], + 'install_guide_linux': [ + "Ubuntu/Debian:", + " sudo apt update", + " sudo apt install ffmpeg", + "", + "Fedora/RHEL:", + " sudo dnf install ffmpeg", + "", + "Arch Linux:", + " sudo pacman -S ffmpeg", + ], + } + ] + + def __init__(self): + """初始化依赖检查器""" + self.packages: Dict[str, DependencyInfo] = {} + self.system_tools: Dict[str, DependencyInfo] = {} + self.python_version_ok: bool = True + self.python_version_info: Optional[Tuple[int, int, int]] = None + self._checked: bool = False + + def _get_platform(self) -> str: + """ + 获取当前平台 + + Returns: + 'windows', 'macos', 或 'linux' + """ + if sys.platform == 'win32': + return 'windows' + elif sys.platform == 'darwin': + return 'macos' + else: + return 'linux' + + def check_python_version(self) -> bool: + """ + 检查 Python 版本 + + Returns: + 版本是否满足要求 + """ + self.python_version_info = sys.version_info[:3] + major, minor, micro = self.python_version_info + + # 要求 Python 3.8 - 3.11 + # Whisper 在 Python 3.12+ 可能有兼容性问题 + if (major, minor) < (3, 8): + self.python_version_ok = False + return False + + if (major, minor) >= (3, 12): + # Python 3.12+ 可能有兼容性问题，但不直接阻止 + self.python_version_ok = True + return True + + self.python_version_ok = True + return True + + def check_package(self, package_info: dict) -> DependencyInfo: + """ + 检查单个 Python 包 + + Args: + package_info: 包信息字典 + + Returns: + 依赖信息对象 + """ + name = package_info['name'] + display_name = package_info['display_name'] + required = package_info['required'] + min_version = package_info.get('min_version') + install_guide = package_info.get('install_guide', []) + + info = DependencyInfo( + name=name, + display_name=display_name, + required=required, + install_guide=install_guide, + ) + + try: + # 尝试导入包 + spec = importlib.util.find_spec(name) + if spec is None: + info.status = DependencyStatus.MISSING + info.installed = False + info.error_message = f"包 {display_name} ({name}) 未安装" + return info + + # 尝试获取版本 + module = importlib.import_module(name) + info.installed = True + + # 获取版本号 + version = None + if hasattr(module, '__version__'): + version = module.__version__ + elif name == 'whisper': + # whisper 可能没有 __version__ + try: + import openai_whisper + if hasattr(openai_whisper, '__version__'): + version = openai_whisper.__version__ + except ImportError: + pass + + info.version = version + + # 检查版本要求 + if min_version and version: + try: + from packaging.version import parse as parse_version + if parse_version(version) < parse_version(min_version): + info.status = DependencyStatus.VERSION_TOO_OLD + info.error_message = f"包 {display_name} 版本 {version} 太旧，要求 >= {min_version}" + return info + except ImportError: + # 如果没有 packaging 模块，跳过版本检查 + pass + + info.status = DependencyStatus.OK + + except ImportError as e: + info.status = DependencyStatus.MISSING + info.installed = False + info.error_message = f"导入包 {display_name} 失败: {e}" + + except Exception as e: + info.status = DependencyStatus.ERROR + info.error_message = f"检查包 {display_name} 时发生错误: {e}" + + return info + + def check_all_packages(self) -> Dict[str, DependencyInfo]: + """ + 检查所有 Python 包 + + Returns: + 包信息字典 + """ + for package_info in self.REQUIRED_PACKAGES: + name = package_info['name'] + self.packages[name] = self.check_package(package_info) + + return self.packages + + def check_ffmpeg(self) -> DependencyInfo: + """ + 检查 FFmpeg + + Returns: + FFmpeg 依赖信息 + """ + tool_info = self.SYSTEM_TOOLS[0] + name = tool_info['name'] + display_name = tool_info['display_name'] + + platform = self._get_platform() + if platform == 'windows': + install_guide = tool_info['install_guide_windows'] + elif platform == 'macos': + install_guide = tool_info['install_guide_macos'] + else: + install_guide = tool_info['install_guide_linux'] + + info = DependencyInfo( + name=name, + display_name=display_name, + required=True, + install_guide=install_guide, + ) + + try: + # 使用 audio_processor 中的检查 + from audio_processor import check_ffmpeg_available, FFmpegNotFoundError + + available, error_msg = check_ffmpeg_available() + + if available: + info.status = DependencyStatus.OK + info.installed = True + + # 尝试获取版本 + try: + from audio_processor import AudioProcessor + processor = AudioProcessor() + ffmpeg_info = processor.get_ffmpeg_info() + info.version = ffmpeg_info.get('version') + except Exception: + pass + else: + info.status = DependencyStatus.MISSING + info.installed = False + info.error_message = error_msg + + except ImportError: + # 如果无法导入 audio_processor，使用 shutil.which 检查 + import shutil + ffmpeg_path = shutil.which('ffmpeg') + + if ffmpeg_path: + # 验证是否可执行 + try: + import subprocess + result = subprocess.run( + [ffmpeg_path, '-version'], + capture_output=True, + text=True, + timeout=10 + ) + if result.returncode == 0: + info.status = DependencyStatus.OK + info.installed = True + # 尝试解析版本 + import re + first_line = result.stdout.split('\n')[0] + version_match = re.search(r'ffmpeg version (\S+)', first_line, re.IGNORECASE) + if version_match: + info.version = version_match.group(1) + else: + info.status = DependencyStatus.ERROR + info.error_message = f"FFmpeg 执行失败: {result.stderr}" + except Exception as e: + info.status = DependencyStatus.ERROR + info.error_message = f"检查 FFmpeg 时发生错误: {e}" + else: + info.status = DependencyStatus.MISSING + info.installed = False + info.error_message = "未找到 FFmpeg" + + self.system_tools[name] = info + return info + + def check_all(self) -> bool: + """ + 执行所有依赖检查 + + Returns: + 所有必需依赖是否都满足 + """ + self._checked = True + + # 检查 Python 版本 + self.check_python_version() + + # 检查 Python 包 + self.check_all_packages() + + # 检查系统工具 + self.check_ffmpeg() + + return self.is_all_ok() + + def is_all_ok(self) -> bool: + """ + 检查所有必需依赖是否都满足 + + Returns: + 是否满足 + """ + if not self.python_version_ok: + return False + + # 检查必需的包 + for name, info in self.packages.items(): + if info.required and not info.is_ok(): + return False + + # 检查必需的系统工具 + for name, info in self.system_tools.items(): + if info.required and not info.is_ok(): + return False + + return True + + def get_missing_dependencies(self) -> List[DependencyInfo]: + """ + 获取缺失或有问题的依赖 + + Returns: + 有问题的依赖列表 + """ + missing = [] + + # 检查必需的包 + for name, info in self.packages.items(): + if info.required and not info.is_ok(): + missing.append(info) + + # 检查必需的系统工具 + for name, info in self.system_tools.items(): + if info.required and not info.is_ok(): + missing.append(info) + + return missing + + def get_warning_dependencies(self) -> List[DependencyInfo]: + """ + 获取可选但缺失的依赖（警告级别） + + Returns: + 可选但缺失的依赖列表 + """ + warnings = [] + + for name, info in self.packages.items(): + if not info.required and not info.is_ok(): + warnings.append(info) + + return warnings + + def generate_report(self, show_ok: bool = False) -> str: + """ + 生成依赖检查报告 + + Args: + show_ok: 是否显示状态正常的依赖 + + Returns: + 报告字符串 + """ + lines = [] + lines.append("=" * 60) + lines.append(" VideoSubtitleAI 依赖检查报告") + lines.append("=" * 60) + lines.append("") + + # Python 版本 + lines.append("【Python 版本】") + if self.python_version_info: + version_str = f"{self.python_version_info[0]}.{self.python_version_info[1]}.{self.python_version_info[2]}" + status = "✓ 兼容" if self.python_version_ok else "✗ 不兼容" + lines.append(f" 当前版本: {version_str}") + lines.append(f" 状态: {status}") + if not self.python_version_ok: + lines.append(f" 要求: Python 3.8 - 3.11") + lines.append("") + + # Python 包 + lines.append("【Python 包】") + for name, info in self.packages.items(): + if not show_ok and info.is_ok(): + continue + + status_icon = "✓" if info.is_ok() else "✗" + status_text = "OK" if info.is_ok() else f"问题: {info.error_message}" + required_text = "[必需]" if info.required else "[可选]" + + version_str = f" v{info.version}" if info.version else "" + lines.append(f" {status_icon} {info.display_name}{version_str} {required_text}") + if not info.is_ok(): + lines.append(f" {status_text}") + if info.install_guide: + lines.append(f" 安装指南:") + for guide_line in info.install_guide: + lines.append(f" {guide_line}") + lines.append("") + + # 系统工具 + lines.append("【系统工具】") + for name, info in self.system_tools.items(): + if not show_ok and info.is_ok(): + continue + + status_icon = "✓" if info.is_ok() else "✗" + status_text = "OK" if info.is_ok() else f"问题: {info.error_message}" + + version_str = f" v{info.version}" if info.version else "" + lines.append(f" {status_icon} {info.display_name}{version_str}") + if not info.is_ok(): + lines.append(f" {status_text}") + if info.install_guide: + lines.append(f" 安装指南:") + for guide_line in info.install_guide: + lines.append(f" {guide_line}") + lines.append("") + + # 总结 + lines.append("【总结】") + if self.is_all_ok(): + lines.append(" ✓ 所有必需依赖已满足！") + warnings = self.get_warning_dependencies() + if warnings: + lines.append(f" ⚠ {len(warnings)} 个可选依赖未安装（不影响核心功能）") + else: + missing = self.get_missing_dependencies() + lines.append(f" ✗ {len(missing)} 个必需依赖未满足，请安装后重试") + lines.append("") + lines.append("=" * 60) + + return '\n'.join(lines) + + def print_report(self, show_ok: bool = False): + """ + 打印依赖检查报告到控制台 + + Args: + show_ok: 是否显示状态正常的依赖 + """ + report = self.generate_report(show_ok=show_ok) + + # 添加颜色（如果支持） + try: + from colorama import init, Fore, Style + init() + + # 简单的颜色替换 + colored_report = report + colored_report = colored_report.replace('✓ ', Fore.GREEN + '✓ ' + Style.RESET_ALL) + colored_report = colored_report.replace('✗ ', Fore.RED + '✗ ' + Style.RESET_ALL) + colored_report = colored_report.replace('⚠ ', Fore.YELLOW + '⚠ ' + Style.RESET_ALL) + + print(colored_report) + except ImportError: + print(report) + + +# 便捷函数 +def check_dependencies() -> Tuple[bool, DependencyChecker]: + """ + 便捷函数：检查所有依赖 + + Returns: + (是否所有必需依赖都满足, 依赖检查器实例) + """ + checker = DependencyChecker() + checker.check_all() + return checker.is_all_ok(), checker + + +def print_dependency_report(show_ok: bool = False): + """ + 便捷函数：打印依赖检查报告 + + Args: + show_ok: 是否显示状态正常的依赖 + """ + checker = DependencyChecker() + checker.check_all() + checker.print_report(show_ok=show_ok) + + +if __name__ == '__main__': + # 当直接运行此模块时，执行依赖检查 + print_dependency_report(show_ok=True) diff --git a/main.py b/main.py index 2195e9e..af7ce2f 100644 --- a/main.py +++ b/main.py @@ -1,20 +1,83 @@ """ VideoSubtitleAI - 本地离线音视频转字幕工具 主程序入口，提供命令行接口 +改进版：添加依赖检查，提供更友好的错误信息和安装指南 """ import os import sys import argparse +import traceback from pathlib import Path from typing import Optional, List, Dict, Any, Callable -# 导入自定义模块 -from audio_processor import AudioProcessor -from speech_recognizer import SpeechRecognizer -from subtitle_generator import SubtitleGenerator -from batch_processor import BatchProcessor, BatchProgress -from progress_logger import ProgressLogger, ConsoleLogger + +def check_dependencies_and_exit() -> bool: + """ + 检查依赖并在必要时退出 + + Returns: + 所有依赖是否满足 + """ + try: + from dependency_checker import check_dependencies, DependencyChecker, DependencyStatus + + all_ok, checker = check_dependencies() + + if not all_ok: + # 有缺失的依赖 + print("\n" + "=" * 60) + print(" ⚠️ 依赖检查失败") + print("=" * 60) + + missing = checker.get_missing_dependencies() + warnings = checker.get_warning_dependencies() + + for dep in missing: + print(f"\n✗ {dep.display_name}: {dep.error_message}") + if dep.install_guide: + print(" 安装指南:") + for line in dep.install_guide: + print(f" {line}") + + if warnings: + print(f"\n⚠ {len(warnings)} 个可选依赖未安装（不影响核心功能）:") + for dep in warnings: + print(f" - {dep.display_name}") + + print("\n" + "=" * 60) + print("请安装上述缺失的依赖后重试。") + print("=" * 60 + "\n") + return False + else: + print("✓ 所有必需依赖已满足") + return True + + except ImportError as e: + print(f"\n✗ 依赖检查模块加载失败: {e}") + print("这通常表示 Python 包未正确安装。") + print("\n请确保已安装以下必需包:") + print(" 1. PyTorch: pip install torch torchvision torchaudio") + print(" 2. Whisper: pip install openai-whisper") + print(" 3. NumPy: pip install numpy") + print("\n同时请确保 FFmpeg 已安装:") + print(" Windows: winget install Gyan.FFmpeg") + print(" macOS: brew install ffmpeg") + print(" Linux: sudo apt install ffmpeg") + return False + + +# 导入自定义模块（在依赖检查后） +try: + from audio_processor import AudioProcessor, FFmpegNotFoundError, FFmpegError + from speech_recognizer import SpeechRecognizer + from subtitle_generator import SubtitleGenerator + from batch_processor import BatchProcessor, BatchProgress + from progress_logger import ProgressLogger, ConsoleLogger + MODULES_LOADED = True +except ImportError as e: + print(f"\n✗ 模块加载失败: {e}") + MODULES_LOADED = False class VideoSubtitleAI: @@ -75,22 +138,54 @@ def initialize(self): # 初始化音频处理器 try: + self.progress_logger.info("正在初始化音频处理器...", "初始化") self.audio_processor = AudioProcessor(ffmpeg_path=self.ffmpeg_path) - self.progress_logger.info("音频处理器初始化成功", "初始化") + + # 显示 FFmpeg 信息 + ffmpeg_info = self.audio_processor.get_ffmpeg_info() + if ffmpeg_info['version']: + self.progress_logger.info(f"FFmpeg版本: {ffmpeg_info['version']}", "初始化") + self.progress_logger.success("音频处理器初始化成功", "初始化") + + except FFmpegNotFoundError as e: + # 显示详细的错误信息和解决方案 + print("\n" + "=" * 60) + print(" ❌ FFmpeg 未找到") + print("=" * 60) + print(f"\n错误: {e}") + print("\n" + e.get_detailed_message()) + print("=" * 60 + "\n") + + if self.progress_logger: + self.progress_logger.error(f"音频处理器初始化失败: {e}", "初始化") + raise + except Exception as e: - self.progress_logger.error(f"音频处理器初始化失败: {e}", "初始化") + if self.progress_logger: + self.progress_logger.error(f"音频处理器初始化失败: {e}", "初始化") raise # 初始化语音识别器 try: + self.progress_logger.info("正在初始化语音识别器...", "初始化") self.speech_recognizer = SpeechRecognizer( model_name=self.model_name, device=self.device, model_dir=self.model_dir, ) - self.progress_logger.info(f"语音识别器初始化成功，模型: {self.model_name}", "初始化") + self.progress_logger.success(f"语音识别器初始化成功，模型: {self.model_name}", "初始化") + except ImportError as e: + print("\n" + "=" * 60) + print(" ❌ Whisper 模型加载失败") + print("=" * 60) + print(f"\n错误: {e}") + print("\n请确保已安装 Whisper:") + print(" pip install openai-whisper") + print("=" * 60 + "\n") + raise except Exception as e: - self.progress_logger.error(f"语音识别器初始化失败: {e}", "初始化") + if self.progress_logger: + self.progress_logger.error(f"语音识别器初始化失败: {e}", "初始化") raise # 初始化字幕生成器 @@ -114,6 +209,14 @@ def _show_device_info(self): self.progress_logger.info(f"GPU: {info['gpu_name']} ({info['gpu_memory']})", "设备信息") else: self.progress_logger.warning("GPU不可用，将使用CPU运行（速度较慢）", "设备信息") + print("\n" + "=" * 60) + print(" ⚠️ GPU 不可用") + print("=" * 60) + print("\n未检测到可用的 NVIDIA GPU，将使用 CPU 运行。") + print("CPU 模式运行速度较慢，建议安装 CUDA 版本的 PyTorch。") + print("\n安装 CUDA 版本 PyTorch:") + print(" pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118") + print("=" * 60 + "\n") def load_model(self, force_reload: bool = False): """ @@ -123,8 +226,27 @@ def load_model(self, force_reload: bool = False): self.initialize() self.progress_logger.info(f"正在加载模型: {self.model_name}...", "模型加载") - self.speech_recognizer.load_model(force_reload=force_reload) - self.progress_logger.success("模型加载完成", "模型加载") + self.progress_logger.info("首次运行会自动下载模型，请耐心等待...", "模型加载") + + try: + self.speech_recognizer.load_model(force_reload=force_reload) + self.progress_logger.success("模型加载完成", "模型加载") + except Exception as e: + self.progress_logger.error(f"模型加载失败: {e}", "模型加载") + print("\n" + "=" * 60) + print(" ❌ 模型加载失败") + print("=" * 60) + print(f"\n错误: {e}") + print("\n可能的原因:") + print(" 1. 网络连接问题（首次运行需要下载模型）") + print(" 2. 磁盘空间不足") + print(" 3. 内存不足") + print("\n解决方案:") + print(" - 检查网络连接") + print(" - 确保有足够的磁盘空间（模型大小从 39MB 到 1.5GB）") + print(" - 尝试使用更小的模型: -m tiny 或 -m base") + print("=" * 60 + "\n") + raise def unload_model(self): """ @@ -132,7 +254,8 @@ def unload_model(self): """ if self.speech_recognizer: self.speech_recognizer.unload_model() - self.progress_logger.info("模型已卸载", "模型管理") + if self.progress_logger: + self.progress_logger.info("模型已卸载", "模型管理") def process_single_file( self, @@ -165,6 +288,14 @@ def process_single_file( if not input_path.exists(): raise FileNotFoundError(f"输入文件不存在: {input_path}") + # 检查文件格式 + if not self.audio_processor.is_supported_format(str(input_path)): + raise ValueError( + f"不支持的文件格式: {input_path.suffix}\n" + f"支持的视频格式: {', '.join(sorted(self.audio_processor.SUPPORTED_VIDEO_FORMATS))}\n" + f"支持的音频格式: {', '.join(sorted(self.audio_processor.SUPPORTED_AUDIO_FORMATS))}" + ) + # 确定输出路径 if output_path is None: output_path = input_path.parent / f"{input_path.stem}.srt" @@ -223,6 +354,19 @@ def process_single_file( return result + except FFmpegError as e: + self.progress_logger.error(f"FFmpeg执行失败: {e}", "错误") + print("\n" + "=" * 60) + print(" ❌ 音频处理失败") + print("=" * 60) + print(f"\nFFmpeg执行错误 (返回码: {e.return_code})") + print(f"错误信息: {e.stderr}") + print("\n可能的原因:") + print(" 1. 输入文件损坏或格式不兼容") + print(" 2. FFmpeg版本太旧") + print(" 3. 磁盘空间不足") + print("=" * 60 + "\n") + raise except Exception as e: self.progress_logger.error(f"处理失败: {e}", "错误") raise @@ -295,6 +439,13 @@ def process_folder( if not file_pairs: self.progress_logger.warning("没有找到待处理的文件", "批量处理") + print("\n" + "=" * 60) + print(" ⚠️ 未找到待处理文件") + print("=" * 60) + print(f"\n在 {input_folder} 中未找到支持的音视频文件。") + print(f"\n支持的视频格式: {', '.join(sorted(self.audio_processor.SUPPORTED_VIDEO_FORMATS))}") + print(f"支持的音频格式: {', '.join(sorted(self.audio_processor.SUPPORTED_AUDIO_FORMATS))}") + print("=" * 60 + "\n") return { 'success': True, 'total_files': 0, @@ -417,6 +568,9 @@ def get_info(self) -> Dict[str, Any]: info['device_info'] = self.speech_recognizer.get_device_info() info['model_info'] = self.speech_recognizer.get_model_info() + if self.audio_processor: + info['ffmpeg_info'] = self.audio_processor.get_ffmpeg_info() + return info @@ -429,9 +583,15 @@ def parse_args(): formatter_class=argparse.RawDescriptionHelpFormatter, epilog=''' 示例: + # 检查依赖 + python main.py --check-deps + # 处理单个视频 python main.py -i video.mp4 -m base + # 指定FFmpeg路径 + python main.py -i video.mp4 --ffmpeg-path "C:\\ffmpeg\\bin\\ffmpeg.exe" + # 处理整个文件夹 python main.py -i ./videos -o ./subtitles --recursive -m small @@ -441,7 +601,8 @@ def parse_args(): ) # 输入输出参数 - parser.add_argument('-i', '--input', required=True, help='输入文件或文件夹路径') + parser.add_argument('-i', '--input', required=False, default=None, + help='输入文件或文件夹路径（使用 --check-deps 时可选）') parser.add_argument('-o', '--output', default=None, help='输出文件或文件夹路径（可选）') # 模型参数 @@ -452,6 +613,10 @@ def parse_args(): parser.add_argument('--device', default=None, choices=['cpu', 'cuda'], help='运行设备 (默认: 自动选择)') + # FFmpeg 参数 + parser.add_argument('--ffmpeg-path', default=None, + help='FFmpeg可执行文件路径（可选，如未指定则从系统PATH中查找）') + # 语言参数 parser.add_argument('-l', '--language', default='auto', help='语言代码 (默认: auto自动检测，可选: zh, en, ja等)') @@ -473,77 +638,137 @@ def parse_args(): help='仅检测语言，不生成字幕') parser.add_argument('--info', action='store_true', help='显示系统信息') - parser.add_argument('--ffmpeg-path', default=None, - help='FFmpeg可执行文件路径（可选，如未指定则从系统PATH中查找）') + parser.add_argument('--check-deps', action='store_true', + help='检查依赖并退出') return parser.parse_args() +def handle_exception(e: Exception, verbose: bool = False): + """ + 处理异常，提供友好的错误信息 + + Args: + e: 异常对象 + verbose: 是否显示详细堆栈 + """ + print("\n" + "=" * 60) + print(" ❌ 程序执行失败") + print("=" * 60) + print(f"\n错误类型: {type(e).__name__}") + print(f"错误信息: {e}") + + if verbose: + print("\n详细错误信息:") + traceback.print_exc() + + print("\n" + "=" * 60) + print("如有问题，请检查:") + print(" 1. 所有依赖是否已正确安装") + print(" 2. 输入文件是否存在且格式正确") + print(" 3. 是否有足够的磁盘空间") + print("=" * 60 + "\n") + + def main(): """ 主函数入口 """ args = parse_args() + # 检查依赖 + if args.check_deps: + print("\n" + "=" * 60) + print(" VideoSubtitleAI 依赖检查") + print("=" * 60 + "\n") + + check_dependencies_and_exit() + + # 尝试打印详细报告 + try: + from dependency_checker import print_dependency_report + print("") + print_dependency_report(show_ok=True) + except Exception: + pass + + sys.exit(0) + + # 检查必需参数 + if args.input is None: + print("\n错误: 缺少必需参数 -i/--input") + print("使用 --help 查看帮助信息") + sys.exit(1) + # 检查输入 input_path = Path(args.input) if not input_path.exists(): - print(f"错误: 输入路径不存在: {input_path}") + print(f"\n错误: 输入路径不存在: {input_path}") sys.exit(1) # 确定任务类型 task = 'translate' if args.translate else 'transcribe' - # 创建实例 - ai = VideoSubtitleAI( - model_name=args.model, - device=args.device, - model_dir=args.model_dir, - log_dir=args.log_dir, - verbose=args.verbose, - ffmpeg_path=args.ffmpeg_path, - ) - - # 显示信息 - if args.info: - ai.initialize() - info = ai.get_info() - print("\n" + "=" * 60) - print(" 系统信息") - print("=" * 60) - print(f" 版本: {info['version']}") - print(f" 模型: {info['model_name']}") - if 'device_info' in info: - dev = info['device_info'] - print(f" 设备: {dev['current_device']}") - if dev['gpu_available']: - print(f" GPU: {dev['gpu_name']}") - print("=" * 60 + "\n") - return - - # 仅检测语言 - if args.detect_language_only: - ai.initialize() - if input_path.is_file(): - result = ai.detect_language(str(input_path)) - print(f"\n检测结果:") - print(f" 语言: {result['language_name']}") - print(f" 代码: {result['detected_language']}") - print(f" 置信度: {result['confidence']:.2%}") - print("\nTop 5 可能语言:") - for i, lang in enumerate(result['top_languages'], 1): - print(f" {i}. {lang['language_name']}: {lang['confidence']:.2%}") - else: - print("错误: --detect-language-only 仅支持单个文件") - sys.exit(1) - return - - # 初始化并加载模型 - ai.initialize() - ai.load_model() + # 检查模块是否加载 + if not MODULES_LOADED: + print("\n错误: 核心模块加载失败，请确保所有依赖已正确安装。") + print("运行 --check-deps 检查依赖状态。") + sys.exit(1) - # 处理 try: + # 创建实例 + ai = VideoSubtitleAI( + model_name=args.model, + device=args.device, + model_dir=args.model_dir, + log_dir=args.log_dir, + verbose=args.verbose, + ffmpeg_path=args.ffmpeg_path, + ) + + # 显示信息 + if args.info: + ai.initialize() + info = ai.get_info() + print("\n" + "=" * 60) + print(" 系统信息") + print("=" * 60) + print(f" 版本: {info['version']}") + print(f" 模型: {info['model_name']}") + if 'device_info' in info: + dev = info['device_info'] + print(f" 设备: {dev['current_device']}") + if dev['gpu_available']: + print(f" GPU: {dev['gpu_name']}") + if 'ffmpeg_info' in info: + ffmpeg = info['ffmpeg_info'] + print(f" FFmpeg: {ffmpeg.get('version', 'unknown')}") + print(f" FFmpeg路径: {ffmpeg.get('ffmpeg_path', 'unknown')}") + print("=" * 60 + "\n") + return + + # 仅检测语言 + if args.detect_language_only: + ai.initialize() + if input_path.is_file(): + result = ai.detect_language(str(input_path)) + print(f"\n检测结果:") + print(f" 语言: {result['language_name']}") + print(f" 代码: {result['detected_language']}") + print(f" 置信度: {result['confidence']:.2%}") + print("\nTop 5 可能语言:") + for i, lang in enumerate(result['top_languages'], 1): + print(f" {i}. {lang['language_name']}: {lang['confidence']:.2%}") + else: + print("错误: --detect-language-only 仅支持单个文件") + sys.exit(1) + return + + # 初始化并加载模型 + ai.initialize() + ai.load_model() + + # 处理 if input_path.is_file(): # 单个文件 result = ai.process_single_file( @@ -552,11 +777,15 @@ def main(): language=args.language, task=task, ) - print(f"\n处理完成!") - print(f" 输入: {result['input_path']}") - print(f" 输出: {result['output_path']}") - print(f" 语言: {result['language_name']}") + print(f"\n" + "=" * 60) + print(" ✅ 处理完成!") + print("=" * 60) + print(f"\n 输入文件: {result['input_path']}") + print(f" 输出字幕: {result['output_path']}") + print(f" 检测语言: {result['language_name']}") print(f" 字幕段数: {result['segment_count']}") + print(f" 总字数: {result['total_characters']}") + print("\n" + "=" * 60 + "\n") else: # 文件夹 result = ai.process_folder( @@ -567,22 +796,37 @@ def main(): task=task, skip_existing=args.skip_existing, ) - print(f"\n批量处理完成!") - print(f" 总文件数: {result['total_files']}") - print(f" 成功: {result['completed_files']}") - print(f" 失败: {result['failed_files']}") + print(f"\n" + "=" * 60) + print(" ✅ 批量处理完成!") + print("=" * 60) + print(f"\n 总文件数: {result['total_files']}") + print(f" 成功处理: {result['completed_files']}") + print(f" 处理失败: {result['failed_files']}") + print(f" 跳过: {result['skipped_files']}") print(f" 成功率: {result['success_rate']}") - print(f" 耗时: {result['elapsed_time']}") + print(f" 总耗时: {result['elapsed_time']}") + + if result['errors']: + print(f"\n 失败的文件:") + for err in result['errors']: + print(f" - {Path(err['input_path']).name}: {err.get('error', '未知错误')}") + print("\n" + "=" * 60 + "\n") + + except FFmpegNotFoundError as e: + # 已经在 initialize 中处理过了 + sys.exit(1) except Exception as e: - print(f"\n错误: {e}") - import traceback - traceback.print_exc() + handle_exception(e, args.verbose) sys.exit(1) finally: # 卸载模型 - ai.unload_model() + try: + if 'ai' in locals(): + ai.unload_model() + except Exception: + pass if __name__ == '__main__': diff --git a/requirements.txt b/requirements.txt index 96c87d8..bf40f5e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,25 +1,101 @@ # VideoSubtitleAI - 本地离线音视频转字幕工具 # 依赖包列表 +# +# ⚠️ 重要提示：在安装这些包之前，请先安装 FFmpeg +# Windows: winget install Gyan.FFmpeg +# macOS: brew install ffmpeg +# Linux: sudo apt install ffmpeg +# +# 快速安装所有依赖： +# pip install -r requirements.txt +# -# 核心依赖 +# ============================================================================ +# 必需依赖（必须安装） +# ============================================================================ + +# OpenAI Whisper - 核心语音识别模型 openai-whisper>=20231117 -# PyTorch (Whisper需要，根据系统环境选择合适版本) -# 安装GPU版本 (推荐，速度快): -# pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 +# PyTorch - Whisper 依赖的深度学习框架 +# ⚠️ 强烈建议安装 CUDA 版本以获得 GPU 加速 +# +# 安装 GPU 版本 (推荐，速度提升 10-100 倍): +# pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 +# +# 或者安装 CUDA 12.1 版本: +# pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu121 # -# 或者安装CPU版本: -# pip3 install torch torchvision torchaudio +# 安装 CPU 版本 (速度较慢): +# pip3 install torch torchvision torchaudio # -# 注意: 以下仅为最小依赖，实际安装时请根据需求安装完整版PyTorch +# 注意: 以下仅为最小依赖，实际使用时请安装完整版 torch>=2.0.0 -# 音频处理 (可选，FFmpeg通常需要单独安装) -# ffmpeg-python>=0.2.0 -# pydub>=0.25.1 +# NumPy - 数值计算库 +numpy>=1.24.0 + +# ============================================================================ +# 可选依赖（推荐安装） +# ============================================================================ # 进度条显示 tqdm>=4.66.0 -# 其他工具 -numpy>=1.24.0 +# 颜色输出（用于更友好的终端显示） +colorama>=0.4.6 + +# 版本比较（用于依赖检查） +packaging>=23.0 + +# ============================================================================ +# 系统工具（必须单独安装） +# ============================================================================ +# +# FFmpeg - 音视频处理工具（必需） +# +# Windows: +# 方法1: 使用 winget (推荐) +# winget install Gyan.FFmpeg +# +# 方法2: 使用 Chocolatey +# choco install ffmpeg +# +# 方法3: 手动安装 +# 1. 访问 https://www.gyan.dev/ffmpeg/builds/ +# 2. 下载 ffmpeg-release-full.7z +# 3. 解压到 C:\ffmpeg +# 4. 将 C:\ffmpeg\bin 添加到系统 PATH +# 5. 重启命令行/IDE +# +# macOS: +# brew install ffmpeg +# +# Linux (Ubuntu/Debian): +# sudo apt update +# sudo apt install ffmpeg +# +# Linux (Fedora): +# sudo dnf install ffmpeg +# +# ============================================================================ + +# ============================================================================ +# 安装步骤总结 +# ============================================================================ +# +# 1. 安装 FFmpeg (必需) +# Windows: winget install Gyan.FFmpeg +# macOS: brew install ffmpeg +# Linux: sudo apt install ffmpeg +# +# 2. 安装 PyTorch (推荐 GPU 版本) +# pip3 install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu118 +# +# 3. 安装其他依赖 +# pip install -r requirements.txt +# +# 4. 验证安装 +# python main.py --check-deps +# +# ============================================================================