src/audio_processor.py - master - serohvostov/radio_proxy

src/audio_processor.py
474 строки20 KB
serohvostov
version 2025
12 янв 2026, 10:30
12 янв 2026, 10:30cd6cb9b
"""
Обработчик аудиофайлов для управления аудиофайлами и хранилищем
"""

import os
import uuid
import logging
import asyncio
from pathlib import Path
from typing import Optional, Dict, Any, List
from datetime import datetime, timedelta

from src.models import BitstreamPacket, ContentTypes, FileProcessingError


class FileManager:
    """
    Управляет операциями файловой системы для аудиофайлов
    """
    
    def __init__(self, base_directory: str = './audio_files'):
        """
        Инициализация FileManager
        
        Args:
            base_directory: Базовая директория для файловых операций
        """
        self.base_directory = Path(base_directory)
        self.logger = logging.getLogger(__name__)
    
    def ensure_directory_exists(self, path: str) -> None:
        """
        Убедиться, что директория существует, создать при необходимости
        
        Args:
            path: Путь к директории для проверки существования
            
        Raises:
            FileProcessingError: Если директория не может быть создана
        """
        try:
            directory = Path(path)
            directory.mkdir(parents=True, exist_ok=True)
            self.logger.debug(f'Убедились, что директория существует: {directory}')
            
        except Exception as e:
            error_msg = f'Не удалось создать директорию {path}: {e}'
            self.logger.error(error_msg)
            raise FileProcessingError(error_msg) from e
    
    def get_file_path(self, filename: str) -> str:
        """
        Получить полный путь к файлу в базовой директории
        
        Args:
            filename: Имя файла
            
        Returns:
            str: Полный путь к файлу
        """
        return str(self.base_directory / filename)
    
    def delete_file(self, filepath: str) -> bool:
        """
        Удалить файл
        
        Args:
            filepath: Путь к файлу для удаления
            
        Returns:
            bool: True если файл был удален, False если файл не существовал
            
        Raises:
            FileProcessingError: Если файл не может быть удален
        """
        try:
            path = Path(filepath)
            
            if not path.exists():
                self.logger.debug(f'Файл не существует: {filepath}')
                return False
            
            path.unlink()
            self.logger.debug(f'Удален файл: {filepath}')
            return True
            
        except Exception as e:
            error_msg = f'Не удалось удалить файл {filepath}: {e}'
            self.logger.error(error_msg)
            raise FileProcessingError(error_msg) from e
    
    def list_files(self, pattern: str = '*') -> List[str]:
        """
        Список файлов в базовой директории, соответствующих шаблону
        
        Args:
            pattern: Glob-шаблон для поиска файлов
            
        Returns:
            List[str]: Список соответствующих путей к файлам
        """
        try:
            if not self.base_directory.exists():
                return []
            
            files = [str(f) for f in self.base_directory.glob(pattern) if f.is_file()]
            return sorted(files)
            
        except Exception as e:
            self.logger.error(f'Ошибка при получении списка файлов с шаблоном {pattern}: {e}')
            return []
    
    def get_directory_size(self) -> int:
        """
        Получить общий размер всех файлов в базовой директории
        
        Returns:
            int: Общий размер в байтах
        """
        try:
            total_size = 0
            
            if self.base_directory.exists():
                for filepath in self.base_directory.rglob('*'):
                    if filepath.is_file():
                        total_size += filepath.stat().st_size
            
            return total_size
            
        except Exception as e:
            self.logger.error(f'Ошибка при вычислении размера директории: {e}')
            return 0
    
    def cleanup_empty_directories(self) -> int:
        """
        Удалить пустые поддиректории
        
        Returns:
            int: Количество удаленных директорий
        """
        try:
            removed_count = 0
            
            if not self.base_directory.exists():
                return 0
            
            # Проходим по директориям в обратном порядке (сначала самые глубокие)
            for dirpath in sorted(self.base_directory.rglob('*'), reverse=True):
                if dirpath.is_dir() and dirpath != self.base_directory:
                    try:
                        # Пытаемся удалить, если пустая
                        dirpath.rmdir()
                        removed_count += 1
                        self.logger.debug(f'Удалена пустая директория: {dirpath}')
                    except OSError:
                        # Директория не пустая, пропускаем
                        pass
            
            return removed_count
            
        except Exception as e:
            self.logger.error(f'Ошибка при очистке пустых директорий: {e}')
            return 0


class AudioProcessor:
    """
    Обрабатывает обработку и хранение аудиофайлов
    """
    
    def __init__(self, files_directory: str = './audio_files', debug_keep_audio_files: bool = False):
        """
        Инициализация AudioProcessor
        
        Args:
            files_directory: Директория для хранения аудиофайлов
            debug_keep_audio_files: Отладочный режим - не удалять аудиофайлы
        """
        self.files_directory = Path(files_directory)
        self.debug_keep_audio_files = debug_keep_audio_files
        self.logger = logging.getLogger(__name__)
        self.file_manager = FileManager(files_directory)
        
        # Убеждаемся, что директория существует
        self.file_manager.ensure_directory_exists(str(self.files_directory))
        
        # Соответствие типа контента расширению файла
        self.content_type_extensions = {
            ContentTypes.WAV: '.wav',
            ContentTypes.AMBE: '.ambe', 
            ContentTypes.JPEG: '.jpg'
        }
    
    def save_audio_file(self, audio_data: bytes, content_type: int, unique_id: Optional[int] = None) -> str:
        """
        Сохранить аудиофайл с уникальным именем
        
        Args:
            audio_data: Бинарные аудиоданные
            content_type: Тип контента (1=WAV, 2=AMBE, 3=JPEG)
            unique_id: Опциональный уникальный ID для включения в имя файла
            
        Returns:
            str: Путь к сохраненному файлу
            
        Raises:
            FileProcessingError: Если файл не может быть сохранен
        """
        try:
            # Валидируем тип контента
            if content_type not in self.content_type_extensions:
                raise FileProcessingError(f'Неподдерживаемый тип контента: {content_type}')
            
            # Валидируем аудиоданные
            if not audio_data or len(audio_data) == 0:
                raise FileProcessingError('Аудиоданные пусты')
            
            # Для WAV-файлов выполняем базовую валидацию и извлекаем параметры
            if content_type == ContentTypes.WAV:
                if not self.validate_wav_format(audio_data):
                    self.logger.warning('Валидация WAV-файла не прошла, но продолжаем сохранение')
                else:
                    # Извлекаем и логируем параметры WAV
                    wav_params = self.extract_wav_parameters(audio_data)
                    if wav_params:
                        self.logger.debug(f'Параметры WAV: {wav_params["sample_rate"]}Hz, '
                                       f'{wav_params["channels"]} каналов, '
                                       f'{wav_params["bits_per_sample"]} бит, '
                                       f'формат={wav_params["audio_format"]}, '
                                       f'размер_данных={wav_params["data_size"]} байт')
            
            # Генерируем уникальное имя файла
            filename = self.generate_unique_filename(content_type, unique_id)
            filepath = self.files_directory / filename
            
            # Сохраняем файл
            with open(filepath, 'wb') as f:
                f.write(audio_data)
            
            self.logger.info(f'Сохранен аудиофайл: {filepath} ({len(audio_data)} байт)')
            return str(filepath)
            
        except Exception as e:
            error_msg = f'Не удалось сохранить аудиофайл: {e}'
            self.logger.error(error_msg)
            raise FileProcessingError(error_msg) from e
    
    def generate_unique_filename(self, content_type: int, unique_id: Optional[int] = None) -> str:
        """
        Генерировать уникальное имя файла для аудиофайла
        
        Args:
            content_type: Тип контента для расширения файла
            unique_id: Опциональный уникальный ID для включения
            
        Returns:
            str: Уникальное имя файла
        """
        # Получаем расширение файла
        extension = self.content_type_extensions.get(content_type, '.bin')
        
        # Генерируем временную метку
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        
        # Генерируем UUID
        file_uuid = str(uuid.uuid4())[:8]
        
        # Включаем unique_id если предоставлен
        if unique_id is not None:
            filename = f'audio_{timestamp}_{unique_id}_{file_uuid}{extension}'
        else:
            filename = f'audio_{timestamp}_{file_uuid}{extension}'
        
        return filename
    
    def validate_wav_format(self, audio_data: bytes) -> bool:
        """
        Валидировать формат WAV-файла путем проверки заголовка
        
        Args:
            audio_data: Бинарные аудиоданные
            
        Returns:
            bool: True если формат WAV валиден
        """
        try:
            # WAV-файлы должны быть не менее 44 байт (минимальный размер заголовка)
            if len(audio_data) < 44:
                return False
            
            # Проверяем заголовок RIFF
            if audio_data[:4] != b'RIFF':
                return False
            
            # Проверяем формат WAVE
            if audio_data[8:12] != b'WAVE':
                return False
            
            # Проверяем chunk fmt
            if audio_data[12:16] != b'fmt ':
                return False
            
            return True
            
        except Exception as e:
            self.logger.warning(f"Ошибка валидации WAV: {e}")
            return False
    
    def extract_wav_parameters(self, audio_data: bytes) -> Optional[Dict[str, Any]]:
        """
        Извлечь параметры аудио из WAV заголовка
        
        Args:
            audio_data: Бинарные аудиоданные WAV файла
            
        Returns:
            Dict с параметрами WAV или None если не удалось извлечь
            {
                'sample_rate': int,      # Частота дискретизации (Hz)
                'channels': int,         # Количество каналов
                'bits_per_sample': int,  # Битность (8, 16, 24, 32)
                'audio_format': int,     # Аудио формат (1 = PCM)
                'byte_rate': int,        # Байт в секунду
                'block_align': int,      # Выравнивание блока
                'data_size': int         # Размер аудио данных
            }
        """
        try:
            # Проверяем минимальный размер
            if len(audio_data) < 44:
                self.logger.warning("WAV файл слишком короткий для извлечения параметров")
                return None
            
            # Проверяем базовую валидность
            if not self.validate_wav_format(audio_data):
                self.logger.warning("Неверный формат WAV файла")
                return None
            
            import struct
            
            # Извлекаем параметры из заголовка
            # Байты 16-19: Размер fmt chunk
            fmt_size = struct.unpack('<I', audio_data[16:20])[0]
            
            # Байты 20-21: Аудио формат (1 = PCM)
            audio_format = struct.unpack('<H', audio_data[20:22])[0]
            
            # Байты 22-23: Количество каналов
            channels = struct.unpack('<H', audio_data[22:24])[0]
            
            # Байты 24-27: Частота дискретизации
            sample_rate = struct.unpack('<I', audio_data[24:28])[0]
            
            # Байты 28-31: Byte rate
            byte_rate = struct.unpack('<I', audio_data[28:32])[0]
            
            # Байты 32-33: Block align
            block_align = struct.unpack('<H', audio_data[32:34])[0]
            
            # Байты 34-35: Bits per sample
            bits_per_sample = struct.unpack('<H', audio_data[34:36])[0]
            
            # Ищем data chunk для определения размера аудио данных
            data_size = 0
            offset = 36  # Начинаем после fmt chunk
            
            while offset < len(audio_data) - 8:
                chunk_id = audio_data[offset:offset+4]
                chunk_size = struct.unpack('<I', audio_data[offset+4:offset+8])[0]
                
                if chunk_id == b'data':
                    data_size = chunk_size
                    break
                
                # Переходим к следующему chunk
                offset += 8 + chunk_size
                # Выравнивание на четные байты
                if chunk_size % 2 == 1:
                    offset += 1
            
            parameters = {
                'sample_rate': sample_rate,
                'channels': channels,
                'bits_per_sample': bits_per_sample,
                'audio_format': audio_format,
                'byte_rate': byte_rate,
                'block_align': block_align,
                'data_size': data_size,
                'fmt_size': fmt_size
            }
            
            # Логируем параметры для отладки
            self.logger.debug(f"Извлечены параметры WAV: {parameters}")
            
            # Базовая валидация параметров
            if sample_rate == 0 or channels == 0 or bits_per_sample == 0:
                self.logger.warning("Неверные параметры WAV файла")
                return None
            
            return parameters
            
        except Exception as e:
            self.logger.warning(f"Ошибка извлечения параметров WAV: {e}")
            return None
    
    async def cleanup_old_files(self, max_age_hours: int = 24) -> int:
        """
        Очистить старые аудиофайлы
        
        Args:
            max_age_hours: Максимальный возраст файлов для сохранения в часах
            
        Returns:
            int: Количество удаленных файлов
        """
        if self.debug_keep_audio_files:
            self.logger.info("Отладочный режим: очистка старых аудиофайлов пропущена")
            return 0
            
        try:
            cutoff_time = datetime.now() - timedelta(hours=max_age_hours)
            deleted_count = 0
            
            for filepath in self.files_directory.iterdir():
                if filepath.is_file():
                    # Получаем время модификации файла
                    file_mtime = datetime.fromtimestamp(filepath.stat().st_mtime)
                    
                    if file_mtime < cutoff_time:
                        try:
                            filepath.unlink()
                            deleted_count += 1
                            self.logger.debug(f"Удален старый файл: {filepath}")
                        except Exception as e:
                            self.logger.warning(f"Не удалось удалить файл {filepath}: {e}")
            
            if deleted_count > 0:
                self.logger.info(f"Очищено {deleted_count} старых аудиофайлов")
            
            return deleted_count
            
        except Exception as e:
            self.logger.error(f"Ошибка при очистке файлов: {e}")
            return 0
    
    def get_file_info(self, filepath: str) -> Dict[str, Any]:
        """
        Получить информацию об аудиофайле
        
        Args:
            filepath: Путь к аудиофайлу
            
        Returns:
            dict: Информация о файле
        """
        try:
            path = Path(filepath)
            if not path.exists():
                raise FileProcessingError(f"Файл не найден: {filepath}")
            
            stat = path.stat()
            return {
                "filepath": str(path),
                "filename": path.name,
                "size_bytes": stat.st_size,
                "created": datetime.fromtimestamp(stat.st_ctime).isoformat(),
                "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
                "extension": path.suffix.lower()
            }
            
        except Exception as e:
            error_msg = f"Не удалось получить информацию о файле {filepath}: {e}"
            self.logger.error(error_msg)
            raise FileProcessingError(error_msg) from e