PCNUSMSE/transcript_service

1"""文件验证模块2 3提供音频文件格式验证、大小检查等功能。4"""5 6import magic7from pathlib import Path8from typing import List, Optional, Tuple9import mimetypes10 11from ..core.config import get_config12from ..utils.logger import get_task_logger13 14 15class FileValidator:16    """文件验证器"""17    18    # 支持的音频文件格式19    SUPPORTED_EXTENSIONS = {20        '.aac', '.amr', '.avi', '.flac', '.flv', '.m4a', '.mkv', 21        '.mov', '.mp3', '.mp4', '.mpeg', '.ogg', '.opus', '.wav', 22        '.webm', '.wma', '.wmv'23    }24    25    # 支持的MIME类型26    SUPPORTED_MIME_TYPES = {27        'audio/aac', 'audio/amr', 'audio/flac', 'audio/mp3', 'audio/mpeg',28        'audio/mp4', 'audio/ogg', 'audio/opus', 'audio/wav', 'audio/webm',29        'audio/x-wav', 'audio/x-flac', 'audio/x-m4a',30        'video/mp4', 'video/avi', 'video/x-flv', 'video/quicktime',31        'video/x-msvideo', 'video/webm', 'video/x-ms-wmv'32    }33    34    def __init__(self):35        """初始化文件验证器"""36        self.config = get_config()37        self.logger = get_task_logger(logger_name="transcript_service.validator")38        39        # 初始化libmagic40        try:41            self.magic = magic.Magic(mime=True)42        except Exception as e:43            self.logger.warning(f"无法初始化libmagic: {str(e)}, 将使用基础验证")44            self.magic = None45    46    def validate_file(self, file_path: Path) -> Tuple[bool, Optional[str]]:47        """验证单个文件48        49        Args:50            file_path: 文件路径51            52        Returns:53            (是否有效, 错误信息)54        """55        try:56            # 检查文件是否存在57            if not file_path.exists():58                return False, f"文件不存在: {file_path}"59            60            # 检查是否是文件61            if not file_path.is_file():62                return False, f"不是有效的文件: {file_path}"63            64            # 检查文件大小65            file_size = file_path.stat().st_size66            if file_size == 0:67                return False, f"文件为空: {file_path.name}"68            69            if file_size > self.config.app.max_file_size:70                size_mb = file_size / (1024 * 1024)71                max_size_mb = self.config.app.max_file_size / (1024 * 1024)72                return False, f"文件大小 {size_mb:.1f}MB 超过限制 {max_size_mb:.1f}MB: {file_path.name}"73            74            # 检查文件扩展名75            file_ext = file_path.suffix.lower()76            if file_ext not in self.SUPPORTED_EXTENSIONS:77                return False, f"不支持的文件格式 {file_ext}: {file_path.name}"78            79            # 检查MIME类型80            if not self._check_mime_type(file_path):81                return False, f"文件内容与扩展名不匹配: {file_path.name}"82            83            # 检查文件完整性84            if not self._check_file_integrity(file_path):85                return False, f"文件可能损坏或不完整: {file_path.name}"86            87            self.logger.info(f"文件验证通过: {file_path.name}")88            return True, None89            90        except Exception as e:91            error_msg = f"验证文件时发生错误: {file_path.name}, 错误: {str(e)}"92            self.logger.exception(error_msg)93            return False, error_msg94    95    def validate_multiple_files(self, file_paths: List[Path]) -> Tuple[List[Path], List[Tuple[Path, str]]]:96        """验证多个文件97        98        Args:99            file_paths: 文件路径列表100            101        Returns:102            (有效文件列表, 无效文件列表[(文件路径, 错误信息)])103        """104        # 检查文件数量105        if len(file_paths) > self.config.app.max_files_count:106            self.logger.warning(f"文件数量 {len(file_paths)} 超过限制 {self.config.app.max_files_count}")107        108        valid_files = []109        invalid_files = []110        111        for file_path in file_paths[:self.config.app.max_files_count]:112            is_valid, error_msg = self.validate_file(file_path)113            if is_valid:114                valid_files.append(file_path)115            else:116                invalid_files.append((file_path, error_msg))117        118        # 如果超过限制，记录被跳过的文件119        if len(file_paths) > self.config.app.max_files_count:120            skipped_count = len(file_paths) - self.config.app.max_files_count121            self.logger.warning(f"跳过了 {skipped_count} 个文件（超过批处理限制）")122        123        self.logger.info(f"文件验证完成: {len(valid_files)} 个有效文件, {len(invalid_files)} 个无效文件")124        return valid_files, invalid_files125    126    def _check_mime_type(self, file_path: Path) -> bool:127        """检查文件MIME类型128        129        Args:130            file_path: 文件路径131            132        Returns:133            MIME类型是否匹配134        """135        try:136            # 使用libmagic检查137            if self.magic:138                mime_type = self.magic.from_file(str(file_path))139                if mime_type in self.SUPPORTED_MIME_TYPES:140                    return True141            142            # 使用mimetypes作为备选方案143            mime_type, _ = mimetypes.guess_type(str(file_path))144            if mime_type and mime_type in self.SUPPORTED_MIME_TYPES:145                return True146            147            # 对于某些格式，检查文件头148            return self._check_file_header(file_path)149            150        except Exception as e:151            self.logger.warning(f"检查MIME类型时发生错误: {file_path.name}, 错误: {str(e)}")152            # 如果MIME检查失败，只要扩展名正确就通过153            return True154    155    def _check_file_header(self, file_path: Path) -> bool:156        """检查文件头部特征157        158        Args:159            file_path: 文件路径160            161        Returns:162            文件头是否匹配163        """164        try:165            with open(file_path, 'rb') as f:166                header = f.read(16)167            168            if not header:169                return False170            171            # 检查常见音频格式的文件头172            if header.startswith(b'ID3') or header[4:8] == b'ftyp':  # MP3, MP4173                return True174            elif header.startswith(b'RIFF') and b'WAVE' in header:  # WAV175                return True176            elif header.startswith(b'fLaC'):  # FLAC177                return True178            elif header.startswith(b'OggS'):  # OGG179                return True180            elif header.startswith(b'\xff\xfb') or header.startswith(b'\xff\xfa'):  # MP3181                return True182            183            # 如果无法识别文件头，但扩展名正确，就通过验证184            return True185            186        except Exception as e:187            self.logger.warning(f"检查文件头时发生错误: {file_path.name}, 错误: {str(e)}")188            return True189    190    def _check_file_integrity(self, file_path: Path) -> bool:191        """检查文件完整性192        193        Args:194            file_path: 文件路径195            196        Returns:197            文件是否完整198        """199        try:200            # 基础完整性检查：确保文件可以完全读取201            with open(file_path, 'rb') as f:202                # 读取文件开头和结尾203                f.read(1024)  # 读取前1KB204                f.seek(-min(1024, file_path.stat().st_size), 2)  # 读取后1KB205                f.read()206            207            return True208            209        except Exception as e:210            self.logger.warning(f"检查文件完整性时发生错误: {file_path.name}, 错误: {str(e)}")211            return False212    213    def get_file_info(self, file_path: Path) -> dict:214        """获取文件信息215        216        Args:217            file_path: 文件路径218            219        Returns:220            文件信息字典221        """222        try:223            stat = file_path.stat()224            225            # 获取MIME类型226            mime_type = None227            if self.magic:228                try:229                    mime_type = self.magic.from_file(str(file_path))230                except:231                    pass232            233            if not mime_type:234                mime_type, _ = mimetypes.guess_type(str(file_path))235            236            return {237                'name': file_path.name,238                'size': stat.st_size,239                'size_mb': round(stat.st_size / (1024 * 1024), 2),240                'extension': file_path.suffix.lower(),241                'mime_type': mime_type,242                'modified_time': stat.st_mtime,243                'is_supported': file_path.suffix.lower() in self.SUPPORTED_EXTENSIONS244            }245            246        except Exception as e:247            self.logger.error(f"获取文件信息失败: {file_path.name}, 错误: {str(e)}")248            return {249                'name': file_path.name,250                'error': str(e)251            }252    253    def get_supported_formats(self) -> dict:254        """获取支持的文件格式信息255        256        Returns:257            支持的格式信息258        """259        return {260            'extensions': sorted(list(self.SUPPORTED_EXTENSIONS)),261            'mime_types': sorted(list(self.SUPPORTED_MIME_TYPES)),262            'max_file_size_mb': self.config.app.max_file_size / (1024 * 1024),263            'max_files_count': self.config.app.max_files_count264        }265 266 267# 全局文件验证器实例268file_validator = FileValidator()269 270 271def get_file_validator() -> FileValidator:272    """获取文件验证器实例273    274    Returns:275        文件验证器实例276    """277    return file_validator