src/services/file_validator.py

"""文件验证模块

提供音频文件格式验证、大小检查等功能。
"""

import magic
from pathlib import Path
from typing import List, Optional, Tuple
import mimetypes

from ..core.config import get_config
from ..utils.logger import get_task_logger


class FileValidator:
    """文件验证器"""
    
    # 支持的音频文件格式
    SUPPORTED_EXTENSIONS = {
        '.aac', '.amr', '.avi', '.flac', '.flv', '.m4a', '.mkv', 
        '.mov', '.mp3', '.mp4', '.mpeg', '.ogg', '.opus', '.wav', 
        '.webm', '.wma', '.wmv'
    }
    
    # 支持的MIME类型
    SUPPORTED_MIME_TYPES = {
        'audio/aac', 'audio/amr', 'audio/flac', 'audio/mp3', 'audio/mpeg',
        'audio/mp4', 'audio/ogg', 'audio/opus', 'audio/wav', 'audio/webm',
        'audio/x-wav', 'audio/x-flac', 'audio/x-m4a',
        'video/mp4', 'video/avi', 'video/x-flv', 'video/quicktime',
        'video/x-msvideo', 'video/webm', 'video/x-ms-wmv'
    }
    
    def __init__(self):
        """初始化文件验证器"""
        self.config = get_config()
        self.logger = get_task_logger(logger_name="transcript_service.validator")
        
        # 初始化libmagic
        try:
            self.magic = magic.Magic(mime=True)
        except Exception as e:
            self.logger.warning(f"无法初始化libmagic: {str(e)}, 将使用基础验证")
            self.magic = None
    
    def validate_file(self, file_path: Path) -> Tuple[bool, Optional[str]]:
        """验证单个文件
        
        Args:
            file_path: 文件路径
            
        Returns:
            (是否有效, 错误信息)
        """
        try:
            # 检查文件是否存在
            if not file_path.exists():
                return False, f"文件不存在: {file_path}"
            
            # 检查是否是文件
            if not file_path.is_file():
                return False, f"不是有效的文件: {file_path}"
            
            # 检查文件大小
            file_size = file_path.stat().st_size
            if file_size == 0:
                return False, f"文件为空: {file_path.name}"
            
            if file_size > self.config.app.max_file_size:
                size_mb = file_size / (1024 * 1024)
                max_size_mb = self.config.app.max_file_size / (1024 * 1024)
                return False, f"文件大小 {size_mb:.1f}MB 超过限制 {max_size_mb:.1f}MB: {file_path.name}"
            
            # 检查文件扩展名
            file_ext = file_path.suffix.lower()
            if file_ext not in self.SUPPORTED_EXTENSIONS:
                return False, f"不支持的文件格式 {file_ext}: {file_path.name}"
            
            # 检查MIME类型
            if not self._check_mime_type(file_path):
                return False, f"文件内容与扩展名不匹配: {file_path.name}"
            
            # 检查文件完整性
            if not self._check_file_integrity(file_path):
                return False, f"文件可能损坏或不完整: {file_path.name}"
            
            self.logger.info(f"文件验证通过: {file_path.name}")
            return True, None
            
        except Exception as e:
            error_msg = f"验证文件时发生错误: {file_path.name}, 错误: {str(e)}"
            self.logger.exception(error_msg)
            return False, error_msg
    
    def validate_multiple_files(self, file_paths: List[Path]) -> Tuple[List[Path], List[Tuple[Path, str]]]:
        """验证多个文件
        
        Args:
            file_paths: 文件路径列表
            
        Returns:
            (有效文件列表, 无效文件列表[(文件路径, 错误信息)])
        """
        # 检查文件数量
        if len(file_paths) > self.config.app.max_files_count:
            self.logger.warning(f"文件数量 {len(file_paths)} 超过限制 {self.config.app.max_files_count}")
        
        valid_files = []
        invalid_files = []
        
        for file_path in file_paths[:self.config.app.max_files_count]:
            is_valid, error_msg = self.validate_file(file_path)
            if is_valid:
                valid_files.append(file_path)
            else:
                invalid_files.append((file_path, error_msg))
        
        # 如果超过限制，记录被跳过的文件
        if len(file_paths) > self.config.app.max_files_count:
            skipped_count = len(file_paths) - self.config.app.max_files_count
            self.logger.warning(f"跳过了 {skipped_count} 个文件（超过批处理限制）")
        
        self.logger.info(f"文件验证完成: {len(valid_files)} 个有效文件, {len(invalid_files)} 个无效文件")
        return valid_files, invalid_files
    
    def _check_mime_type(self, file_path: Path) -> bool:
        """检查文件MIME类型
        
        Args:
            file_path: 文件路径
            
        Returns:
            MIME类型是否匹配
        """
        try:
            # 使用libmagic检查
            if self.magic:
                mime_type = self.magic.from_file(str(file_path))
                if mime_type in self.SUPPORTED_MIME_TYPES:
                    return True
            
            # 使用mimetypes作为备选方案
            mime_type, _ = mimetypes.guess_type(str(file_path))
            if mime_type and mime_type in self.SUPPORTED_MIME_TYPES:
                return True
            
            # 对于某些格式，检查文件头
            return self._check_file_header(file_path)
            
        except Exception as e:
            self.logger.warning(f"检查MIME类型时发生错误: {file_path.name}, 错误: {str(e)}")
            # 如果MIME检查失败，只要扩展名正确就通过
            return True
    
    def _check_file_header(self, file_path: Path) -> bool:
        """检查文件头部特征
        
        Args:
            file_path: 文件路径
            
        Returns:
            文件头是否匹配
        """
        try:
            with open(file_path, 'rb') as f:
                header = f.read(16)
            
            if not header:
                return False
            
            # 检查常见音频格式的文件头
            if header.startswith(b'ID3') or header[4:8] == b'ftyp':  # MP3, MP4
                return True
            elif header.startswith(b'RIFF') and b'WAVE' in header:  # WAV
                return True
            elif header.startswith(b'fLaC'):  # FLAC
                return True
            elif header.startswith(b'OggS'):  # OGG
                return True
            elif header.startswith(b'\xff\xfb') or header.startswith(b'\xff\xfa'):  # MP3
                return True
            
            # 如果无法识别文件头，但扩展名正确，就通过验证
            return True
            
        except Exception as e:
            self.logger.warning(f"检查文件头时发生错误: {file_path.name}, 错误: {str(e)}")
            return True
    
    def _check_file_integrity(self, file_path: Path) -> bool:
        """检查文件完整性
        
        Args:
            file_path: 文件路径
            
        Returns:
            文件是否完整
        """
        try:
            # 基础完整性检查：确保文件可以完全读取
            with open(file_path, 'rb') as f:
                # 读取文件开头和结尾
                f.read(1024)  # 读取前1KB
                f.seek(-min(1024, file_path.stat().st_size), 2)  # 读取后1KB
                f.read()
            
            return True
            
        except Exception as e:
            self.logger.warning(f"检查文件完整性时发生错误: {file_path.name}, 错误: {str(e)}")
            return False
    
    def get_file_info(self, file_path: Path) -> dict:
        """获取文件信息
        
        Args:
            file_path: 文件路径
            
        Returns:
            文件信息字典
        """
        try:
            stat = file_path.stat()
            
            # 获取MIME类型
            mime_type = None
            if self.magic:
                try:
                    mime_type = self.magic.from_file(str(file_path))
                except:
                    pass
            
            if not mime_type:
                mime_type, _ = mimetypes.guess_type(str(file_path))
            
            return {
                'name': file_path.name,
                'size': stat.st_size,
                'size_mb': round(stat.st_size / (1024 * 1024), 2),
                'extension': file_path.suffix.lower(),
                'mime_type': mime_type,
                'modified_time': stat.st_mtime,
                'is_supported': file_path.suffix.lower() in self.SUPPORTED_EXTENSIONS
            }
            
        except Exception as e:
            self.logger.error(f"获取文件信息失败: {file_path.name}, 错误: {str(e)}")
            return {
                'name': file_path.name,
                'error': str(e)
            }
    
    def get_supported_formats(self) -> dict:
        """获取支持的文件格式信息
        
        Returns:
            支持的格式信息
        """
        return {
            'extensions': sorted(list(self.SUPPORTED_EXTENSIONS)),
            'mime_types': sorted(list(self.SUPPORTED_MIME_TYPES)),
            'max_file_size_mb': self.config.app.max_file_size / (1024 * 1024),
            'max_files_count': self.config.app.max_files_count
        }


# 全局文件验证器实例
file_validator = FileValidator()


def get_file_validator() -> FileValidator:
    """获取文件验证器实例
    
    Returns:
        文件验证器实例
    """
    return file_validator