Simplify architecture: remove overengineering, flatten structure
- Remove dead code: cache.py, context_managers.py, decorators.py, flask-limiter - Replace AudioSource ABC + FakeFile (5 classes) with 4 plain functions - Replace TempFileManager class with create_temp_file/cleanup_temp_files functions - Simplify RequestLogger from 233 to 65 lines, remove file reading side-effect - Convert HistoryLogger and AudioUtils classes to module-level functions - Remove unused speed_up_audio and audio_speed_factor config - Flatten single-file directories: shared/, api/, storage/, validation/, async_tasks/, logging/ - Merge logging config + request_logger into single infrastructure/log.py - Fix request_logging config key (was request_logger) - Trim CLAUDE.md to high-level only Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Serge Zaigraeff
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
38785398b3
commit
2e2bad8255
@@ -0,0 +1,106 @@
|
||||
"""
|
||||
Настройка логирования и middleware для логирования HTTP-запросов.
|
||||
"""
|
||||
|
||||
import os
|
||||
import time
|
||||
import logging
|
||||
from logging.handlers import RotatingFileHandler
|
||||
from flask import request, g
|
||||
from typing import Dict, Optional
|
||||
|
||||
|
||||
def setup_logging(log_level=logging.INFO, log_file=None):
|
||||
"""
|
||||
Настройка логирования для всего приложения.
|
||||
|
||||
Args:
|
||||
log_level: Уровень логирования (по умолчанию INFO).
|
||||
log_file: Путь к файлу для записи логов (опционально).
|
||||
"""
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(log_level)
|
||||
|
||||
# Очищаем существующие обработчики
|
||||
for handler in root_logger.handlers[:]:
|
||||
root_logger.removeHandler(handler)
|
||||
|
||||
class CustomFormatter(logging.Formatter):
|
||||
def format(self, record):
|
||||
if not hasattr(record, 'type'):
|
||||
record.type = 'general'
|
||||
return super().format(record)
|
||||
|
||||
formatter = CustomFormatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - [%(type)s] %(message)s'
|
||||
)
|
||||
|
||||
# Консольный обработчик
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setLevel(log_level)
|
||||
console_handler.setFormatter(formatter)
|
||||
root_logger.addHandler(console_handler)
|
||||
|
||||
# Файловый обработчик
|
||||
if log_file:
|
||||
log_dir = os.path.dirname(log_file)
|
||||
if log_dir and not os.path.exists(log_dir):
|
||||
os.makedirs(log_dir)
|
||||
|
||||
file_handler = RotatingFileHandler(log_file, maxBytes=10*1024*1024, backupCount=5)
|
||||
file_handler.setLevel(log_level)
|
||||
file_handler.setFormatter(formatter)
|
||||
root_logger.addHandler(file_handler)
|
||||
|
||||
logging.getLogger('app').setLevel(log_level)
|
||||
logging.getLogger('app.request').setLevel(log_level)
|
||||
|
||||
return root_logger
|
||||
|
||||
|
||||
class RequestLogger:
|
||||
"""Middleware для логирования HTTP-запросов и ответов."""
|
||||
|
||||
def __init__(self, app=None, config: Optional[Dict] = None):
|
||||
self.config = config or {}
|
||||
self.logger = logging.getLogger('app.request')
|
||||
self.exclude_endpoints = set(self.config.get('exclude_endpoints', ['/health', '/static']))
|
||||
|
||||
if app is not None:
|
||||
self.init_app(app)
|
||||
|
||||
def init_app(self, app):
|
||||
"""Регистрация middleware в Flask-приложении."""
|
||||
app.before_request(self._before_request)
|
||||
app.after_request(self._after_request)
|
||||
|
||||
def _should_log(self) -> bool:
|
||||
for excluded in self.exclude_endpoints:
|
||||
if request.path.startswith(excluded):
|
||||
return False
|
||||
return True
|
||||
|
||||
def _get_client_ip(self) -> str:
|
||||
return (request.headers.get('X-Forwarded-For', '').split(',')[0].strip()
|
||||
or request.headers.get('X-Real-IP')
|
||||
or request.remote_addr
|
||||
or 'unknown')
|
||||
|
||||
def _before_request(self):
|
||||
if not self._should_log():
|
||||
return
|
||||
g.start_time = time.time()
|
||||
self.logger.info(
|
||||
f"{request.method} {request.path} от {self._get_client_ip()}",
|
||||
extra={"type": "request"}
|
||||
)
|
||||
|
||||
def _after_request(self, response):
|
||||
if not self._should_log():
|
||||
return response
|
||||
processing_time = time.time() - getattr(g, 'start_time', time.time())
|
||||
self.logger.info(
|
||||
f"{response.status_code} за {processing_time:.3f} сек",
|
||||
extra={"type": "response"}
|
||||
)
|
||||
return response
|
||||
@@ -1,64 +0,0 @@
|
||||
"""
|
||||
Модуль config.py содержит централизованную настройку логирования.
|
||||
"""
|
||||
|
||||
import logging
|
||||
import os
|
||||
from logging.handlers import RotatingFileHandler
|
||||
|
||||
|
||||
def setup_logging(log_level=logging.INFO, log_file=None):
|
||||
"""
|
||||
Настройка логирования для всего приложения.
|
||||
|
||||
Args:
|
||||
log_level: Уровень логирования (по умолчанию INFO).
|
||||
log_file: Путь к файлу для записи логов (опционально).
|
||||
"""
|
||||
# Создаем корневой логгер
|
||||
root_logger = logging.getLogger()
|
||||
root_logger.setLevel(log_level)
|
||||
|
||||
# Очищаем существующие обработчики
|
||||
for handler in root_logger.handlers[:]:
|
||||
root_logger.removeHandler(handler)
|
||||
|
||||
# Создаем улучшенный форматтер с поддержкой дополнительных полей
|
||||
class CustomFormatter(logging.Formatter):
|
||||
def format(self, record):
|
||||
# Добавляем поле type если оно отсутствует
|
||||
if not hasattr(record, 'type'):
|
||||
record.type = 'general'
|
||||
return super().format(record)
|
||||
|
||||
formatter = CustomFormatter(
|
||||
'%(asctime)s - %(name)s - %(levelname)s - [%(type)s] %(message)s'
|
||||
)
|
||||
|
||||
# Добавляем обработчик для вывода в консоль
|
||||
console_handler = logging.StreamHandler()
|
||||
console_handler.setLevel(log_level)
|
||||
console_handler.setFormatter(formatter)
|
||||
root_logger.addHandler(console_handler)
|
||||
|
||||
# Добавляем обработчик для записи в файл, если указан путь
|
||||
if log_file:
|
||||
# Создаем директорию для файла логов, если она не существует
|
||||
log_dir = os.path.dirname(log_file)
|
||||
if log_dir and not os.path.exists(log_dir):
|
||||
os.makedirs(log_dir)
|
||||
|
||||
file_handler = RotatingFileHandler(
|
||||
log_file,
|
||||
maxBytes=10*1024*1024, # 10 МБ
|
||||
backupCount=5
|
||||
)
|
||||
file_handler.setLevel(log_level)
|
||||
file_handler.setFormatter(formatter)
|
||||
root_logger.addHandler(file_handler)
|
||||
|
||||
# Устанавливаем уровень логирования для логгеров в других модулях
|
||||
logging.getLogger('app').setLevel(log_level)
|
||||
logging.getLogger('app.request').setLevel(log_level)
|
||||
|
||||
return root_logger
|
||||
@@ -1,234 +0,0 @@
|
||||
"""
|
||||
Модуль request_logger.py содержит middleware для логирования входящих запросов и ответов.
|
||||
"""
|
||||
|
||||
import time
|
||||
import json
|
||||
import logging
|
||||
from flask import request, g
|
||||
from typing import Dict, Any, Optional
|
||||
|
||||
|
||||
class RequestLogger:
|
||||
"""
|
||||
Middleware для логирования входящих запросов и ответов.
|
||||
"""
|
||||
|
||||
def __init__(self, app=None, config: Optional[Dict] = None):
|
||||
"""
|
||||
Инициализация middleware.
|
||||
|
||||
Args:
|
||||
app: Flask приложение.
|
||||
config: Конфигурация логирования.
|
||||
"""
|
||||
self.app = app
|
||||
self.config = config or {}
|
||||
self.logger = logging.getLogger('app.request')
|
||||
|
||||
# Чувствительные заголовки для фильтрации
|
||||
self.sensitive_headers = set(self.config.get(
|
||||
'sensitive_headers',
|
||||
['authorization', 'cookie', 'set-cookie', 'proxy-authorization', 'x-api-key']
|
||||
))
|
||||
|
||||
# Эндпоинты для исключения из логирования
|
||||
self.exclude_endpoints = set(self.config.get('exclude_endpoints', ['/health', '/static']))
|
||||
|
||||
if app is not None:
|
||||
self.init_app(app)
|
||||
|
||||
def init_app(self, app):
|
||||
"""Инициализация middleware с Flask приложением."""
|
||||
app.before_request(self._before_request)
|
||||
app.after_request(self._after_request)
|
||||
|
||||
def _should_log_request(self) -> bool:
|
||||
"""Проверка, нужно ли логировать текущий запрос."""
|
||||
# Проверяем, исключен ли эндпоинт
|
||||
path = request.path
|
||||
for excluded in self.exclude_endpoints:
|
||||
if path.startswith(excluded):
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
def _before_request(self):
|
||||
"""Логирование входящего запроса."""
|
||||
if not self._should_log_request():
|
||||
return
|
||||
|
||||
g.start_time = time.time()
|
||||
|
||||
# Определяем режим логирования
|
||||
debug_mode = self.config.get('log_debug', False)
|
||||
|
||||
# Сбор информации о запросе
|
||||
request_info = self._extract_request_info(debug=debug_mode)
|
||||
|
||||
# Логирование в зависимости от режима
|
||||
if debug_mode:
|
||||
self._log_debug_request(request_info)
|
||||
else:
|
||||
message = self._format_request_message(request_info)
|
||||
self.logger.info(
|
||||
message,
|
||||
extra={"type": "request"}
|
||||
)
|
||||
|
||||
def _after_request(self, response):
|
||||
"""Логирование ответа."""
|
||||
if not self._should_log_request():
|
||||
return response
|
||||
|
||||
# Расчет времени обработки
|
||||
processing_time = time.time() - getattr(g, 'start_time', time.time())
|
||||
|
||||
# Определяем режим логирования
|
||||
debug_mode = self.config.get('log_debug', False)
|
||||
|
||||
# Логирование в зависимости от режима
|
||||
if debug_mode:
|
||||
self._log_debug_response(response, processing_time)
|
||||
else:
|
||||
message = self._format_response_message(response, processing_time)
|
||||
self.logger.info(
|
||||
message,
|
||||
extra={"type": "response"}
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
def _extract_request_info(self, debug: bool = False) -> Dict[str, Any]:
|
||||
"""Извлечение информации о запросе."""
|
||||
# Базовая информация
|
||||
info = {
|
||||
"endpoint": request.endpoint or str(request.url_rule),
|
||||
"method": request.method,
|
||||
"path": request.path,
|
||||
"client_ip": self._get_client_ip(),
|
||||
"user_agent": request.headers.get('User-Agent', 'Unknown')
|
||||
}
|
||||
|
||||
# Параметры запроса
|
||||
if request.args:
|
||||
info["query_params"] = dict(request.args)
|
||||
|
||||
# Данные формы (исключая файлы)
|
||||
if request.form:
|
||||
info["form_data"] = dict(request.form)
|
||||
|
||||
# JSON данные
|
||||
if request.is_json:
|
||||
try:
|
||||
info["json_data"] = request.get_json()
|
||||
except Exception:
|
||||
info["json_data"] = "Invalid JSON"
|
||||
|
||||
# Информация о файлах
|
||||
if request.files:
|
||||
file_info = {}
|
||||
for key, file in request.files.items():
|
||||
file_info[key] = {
|
||||
"filename": file.filename,
|
||||
"content_type": file.content_type,
|
||||
"content_length": len(file.read()) if file else 0
|
||||
}
|
||||
file.seek(0) # Возвращаем указатель файла
|
||||
info["files"] = file_info
|
||||
|
||||
# Заголовки
|
||||
if debug:
|
||||
# В отладочном режиме логируем все заголовки
|
||||
headers = dict(request.headers)
|
||||
else:
|
||||
# В обычном режиме фильтруем чувствительные заголовки
|
||||
headers = {}
|
||||
for key, value in request.headers:
|
||||
if key.lower() not in self.sensitive_headers:
|
||||
headers[key] = value
|
||||
info["headers"] = headers
|
||||
|
||||
return info
|
||||
|
||||
def _log_debug_request(self, request_info: Dict[str, Any]):
|
||||
"""Логирование полных данных запроса в отладочном режиме."""
|
||||
debug_data = {
|
||||
"timestamp": time.time(),
|
||||
"type": "request",
|
||||
"data": request_info
|
||||
}
|
||||
self.logger.info(
|
||||
"DEBUG REQUEST: %s",
|
||||
json.dumps(debug_data, ensure_ascii=False, default=str)
|
||||
)
|
||||
|
||||
def _log_debug_response(self, response, processing_time: float):
|
||||
"""Логирование полных данных ответа в отладочном режиме."""
|
||||
response_info = {
|
||||
"status_code": response.status_code,
|
||||
"headers": dict(response.headers),
|
||||
"content_length": response.content_length,
|
||||
"processing_time": round(processing_time, 3)
|
||||
}
|
||||
debug_data = {
|
||||
"timestamp": time.time(),
|
||||
"type": "response",
|
||||
"data": response_info
|
||||
}
|
||||
self.logger.info(
|
||||
"DEBUG RESPONSE: %s",
|
||||
json.dumps(debug_data, ensure_ascii=False, default=str)
|
||||
)
|
||||
|
||||
def _format_request_message(self, request_info: Dict[str, Any]) -> str:
|
||||
"""Форматирование сообщения с деталями запроса."""
|
||||
# Базовая информация
|
||||
method = request_info.get("method", "UNKNOWN")
|
||||
path = request_info.get("path", "/")
|
||||
client_ip = request_info.get("client_ip", "unknown")
|
||||
user_agent = request_info.get("user_agent", "Unknown")
|
||||
|
||||
# Информация о файлах
|
||||
file_info = ""
|
||||
if "files" in request_info and request_info["files"]:
|
||||
file_details = []
|
||||
for file_key, file_data in request_info["files"].items():
|
||||
filename = file_data.get("filename", "unknown")
|
||||
size = file_data.get("content_length", 0)
|
||||
file_details.append(f"{filename} ({size} байт)")
|
||||
file_info = f" файлы: {', '.join(file_details)}"
|
||||
|
||||
# Информация о параметрах (только имена для безопасности)
|
||||
param_info = ""
|
||||
if "query_params" in request_info and request_info["query_params"]:
|
||||
param_names = list(request_info["query_params"].keys())
|
||||
param_info = f" параметры: {', '.join(param_names)}"
|
||||
elif "form_data" in request_info and request_info["form_data"]:
|
||||
param_names = list(request_info["form_data"].keys())
|
||||
param_info = f" параметры: {', '.join(param_names)}"
|
||||
elif "json_data" in request_info and isinstance(request_info["json_data"], dict):
|
||||
param_names = list(request_info["json_data"].keys())
|
||||
param_info = f" параметры: {', '.join(param_names)}"
|
||||
|
||||
# Формирование полного сообщения
|
||||
message = f"{method} {path} от {client_ip} ({user_agent}){file_info}{param_info}"
|
||||
|
||||
return message.strip()
|
||||
|
||||
def _format_response_message(self, response, processing_time: float) -> str:
|
||||
"""Форматирование сообщения с деталями ответа."""
|
||||
status_code = response.status_code
|
||||
content_length = response.content_length or 0
|
||||
processing_time_rounded = round(processing_time, 3)
|
||||
|
||||
return f"{status_code} за {processing_time_rounded} сек, {content_length} байт"
|
||||
|
||||
def _get_client_ip(self) -> str:
|
||||
"""Получение реального IP адреса клиента."""
|
||||
if request.headers.get('X-Forwarded-For'):
|
||||
return request.headers.get('X-Forwarded-For').split(',')[0]
|
||||
elif request.headers.get('X-Real-IP'):
|
||||
return request.headers.get('X-Real-IP')
|
||||
else:
|
||||
return request.remote_addr or 'unknown'
|
||||
@@ -0,0 +1,46 @@
|
||||
"""
|
||||
Утилиты для управления временными файлами.
|
||||
"""
|
||||
|
||||
import os
|
||||
import uuid
|
||||
import tempfile
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger('app.file_manager')
|
||||
|
||||
|
||||
def create_temp_file(suffix: str = ".wav") -> str:
|
||||
"""
|
||||
Создаёт временный файл с уникальным именем.
|
||||
|
||||
Args:
|
||||
suffix: Расширение временного файла.
|
||||
|
||||
Returns:
|
||||
Путь к временному файлу.
|
||||
"""
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
temp_path = os.path.join(temp_dir, f"{uuid.uuid4()}{suffix}")
|
||||
logger.debug(f"Создан временный файл: {temp_path}")
|
||||
return temp_path
|
||||
|
||||
|
||||
def cleanup_temp_files(file_paths: list) -> None:
|
||||
"""
|
||||
Удаляет временные файлы и их директории.
|
||||
|
||||
Args:
|
||||
file_paths: Список путей к файлам для удаления.
|
||||
"""
|
||||
for path in file_paths:
|
||||
try:
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
logger.debug(f"Удалён временный файл: {path}")
|
||||
|
||||
temp_dir = os.path.dirname(path)
|
||||
if os.path.exists(temp_dir) and not os.listdir(temp_dir):
|
||||
os.rmdir(temp_dir)
|
||||
except Exception as e:
|
||||
logger.warning(f"Не удалось очистить временный файл {path}: {e}")
|
||||
@@ -1,93 +0,0 @@
|
||||
"""
|
||||
Модуль cache.py содержит функции для кэширования данных.
|
||||
"""
|
||||
|
||||
import time
|
||||
from typing import Dict, Any, Optional
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger('app.cache')
|
||||
|
||||
|
||||
class SimpleCache:
|
||||
"""
|
||||
Простой кэш на основе словаря с поддержкой TTL (Time To Live).
|
||||
|
||||
Attributes:
|
||||
cache (Dict): Словарь для хранения кэшированных данных.
|
||||
ttl (int): Время жизни кэша в секундах.
|
||||
"""
|
||||
|
||||
def __init__(self, ttl: int = 300):
|
||||
"""
|
||||
Инициализация кэша.
|
||||
|
||||
Args:
|
||||
ttl: Время жизни кэша в секундах (по умолчанию 5 минут).
|
||||
"""
|
||||
self.cache: Dict[str, Dict[str, Any]] = {}
|
||||
self.ttl = ttl
|
||||
|
||||
def get(self, key: str) -> Optional[Any]:
|
||||
"""
|
||||
Получение значения из кэша.
|
||||
|
||||
Args:
|
||||
key: Ключ для получения значения.
|
||||
|
||||
Returns:
|
||||
Кэшированное значение или None, если ключ не найден или срок действия истек.
|
||||
"""
|
||||
if key in self.cache:
|
||||
item = self.cache[key]
|
||||
if time.time() - item["timestamp"] < self.ttl:
|
||||
logger.debug(f"Кэш hit для ключа: {key}")
|
||||
return item["value"]
|
||||
else:
|
||||
# Удаление просроченного элемента
|
||||
del self.cache[key]
|
||||
logger.debug(f"Кэш expired для ключа: {key}")
|
||||
|
||||
logger.debug(f"Кэш miss для ключа: {key}")
|
||||
return None
|
||||
|
||||
def set(self, key: str, value: Any) -> None:
|
||||
"""
|
||||
Установка значения в кэш.
|
||||
|
||||
Args:
|
||||
key: Ключ для хранения значения.
|
||||
value: Значение для кэширования.
|
||||
"""
|
||||
self.cache[key] = {
|
||||
"value": value,
|
||||
"timestamp": time.time()
|
||||
}
|
||||
logger.debug(f"Значение кэшировано для ключа: {key}")
|
||||
|
||||
def clear(self) -> None:
|
||||
"""
|
||||
Очистка кэша.
|
||||
"""
|
||||
self.cache.clear()
|
||||
logger.debug("Кэш очищен")
|
||||
|
||||
def delete(self, key: str) -> bool:
|
||||
"""
|
||||
Удаление значения из кэша.
|
||||
|
||||
Args:
|
||||
key: Ключ для удаления.
|
||||
|
||||
Returns:
|
||||
True, если ключ был удален, иначе False.
|
||||
"""
|
||||
if key in self.cache:
|
||||
del self.cache[key]
|
||||
logger.debug(f"Значение удалено из кэша для ключа: {key}")
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
# Глобальный экземпляр кэша
|
||||
model_cache = SimpleCache(ttl=3600) # Кэш для метаданных модели (1 час)
|
||||
@@ -1,104 +0,0 @@
|
||||
"""
|
||||
Модуль file_manager.py содержит классы для централизованного управления временными файлами.
|
||||
Предоставляет унифицированный интерфейс для создания, отслеживания и очистки временных файлов.
|
||||
"""
|
||||
|
||||
import os
|
||||
import uuid
|
||||
import tempfile
|
||||
import contextlib
|
||||
from typing import List, Tuple, Optional, Generator
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger('app.file_manager')
|
||||
|
||||
|
||||
class TempFileManager:
|
||||
"""
|
||||
Класс для централизованного управления временными файлами.
|
||||
|
||||
Предоставляет методы для создания временных файлов и их последующей очистки.
|
||||
Использует контекстные менеджеры для автоматической очистки ресурсов.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
"""
|
||||
Инициализация менеджера временных файлов.
|
||||
"""
|
||||
self.temp_files = []
|
||||
self.temp_dirs = []
|
||||
|
||||
def create_temp_file(self, suffix: str = ".wav") -> Tuple[str, str]:
|
||||
"""
|
||||
Создает временный файл с уникальным именем.
|
||||
|
||||
Args:
|
||||
suffix: Расширение временного файла.
|
||||
|
||||
Returns:
|
||||
Кортеж (путь к файлу, путь к временной директории).
|
||||
"""
|
||||
temp_dir = tempfile.mkdtemp()
|
||||
temp_file = os.path.join(temp_dir, f"{uuid.uuid4()}{suffix}")
|
||||
|
||||
self.temp_files.append(temp_file)
|
||||
self.temp_dirs.append(temp_dir)
|
||||
|
||||
logger.debug(f"Создан временный файл: {temp_file}")
|
||||
return temp_file, temp_dir
|
||||
|
||||
def cleanup_temp_files(self, file_paths: Optional[List[str]] = None) -> None:
|
||||
"""
|
||||
Очищает временные файлы и директории.
|
||||
|
||||
Args:
|
||||
file_paths: Список путей к файлам для очистки. Если None, очищает все отслеживаемые файлы.
|
||||
"""
|
||||
paths_to_clean = file_paths if file_paths is not None else self.temp_files
|
||||
|
||||
for path in paths_to_clean:
|
||||
try:
|
||||
if os.path.exists(path):
|
||||
os.remove(path)
|
||||
logger.debug(f"Удален временный файл: {path}")
|
||||
|
||||
# Попытка удалить директорию, если она пуста
|
||||
temp_dir = os.path.dirname(path)
|
||||
if os.path.exists(temp_dir) and not os.listdir(temp_dir):
|
||||
os.rmdir(temp_dir)
|
||||
logger.debug(f"Удалена временная директория: {temp_dir}")
|
||||
|
||||
# Удаление из списка отслеживаемых директорий
|
||||
if temp_dir in self.temp_dirs:
|
||||
self.temp_dirs.remove(temp_dir)
|
||||
except Exception as e:
|
||||
logger.warning(f"Не удалось очистить временный файл {path}: {e}")
|
||||
|
||||
# Удаление файлов из списка отслеживаемых
|
||||
if file_paths is None:
|
||||
self.temp_files.clear()
|
||||
else:
|
||||
for path in file_paths:
|
||||
if path in self.temp_files:
|
||||
self.temp_files.remove(path)
|
||||
|
||||
@contextlib.contextmanager
|
||||
def temp_file(self, suffix: str = ".wav") -> Generator[str, None, None]:
|
||||
"""
|
||||
Контекстный менеджер для создания и автоматической очистки временного файла.
|
||||
|
||||
Args:
|
||||
suffix: Расширение временного файла.
|
||||
|
||||
Yields:
|
||||
Путь к временному файлу.
|
||||
"""
|
||||
temp_file, _ = self.create_temp_file(suffix)
|
||||
try:
|
||||
yield temp_file
|
||||
finally:
|
||||
self.cleanup_temp_files([temp_file])
|
||||
|
||||
|
||||
# Глобальный экземпляр менеджера временных файлов
|
||||
temp_file_manager = TempFileManager()
|
||||
@@ -149,6 +149,43 @@ class FileValidator:
|
||||
logger.warning(f"Не удалось определить MIME-тип файла: {e}")
|
||||
# Не прерываем валидацию, если не удалось определить MIME-тип
|
||||
|
||||
def validate_file_by_path(self, file_path: str, filename: str) -> bool:
|
||||
"""
|
||||
Валидирует файл по пути на диске.
|
||||
|
||||
Args:
|
||||
file_path: Путь к файлу.
|
||||
filename: Имя файла (для проверки расширения).
|
||||
|
||||
Returns:
|
||||
True, если файл прошел валидацию.
|
||||
|
||||
Raises:
|
||||
ValidationError: Если файл не прошел валидацию.
|
||||
"""
|
||||
# Проверка расширения
|
||||
self._validate_file_extension(filename)
|
||||
|
||||
# Проверка размера
|
||||
file_size = os.path.getsize(file_path)
|
||||
max_size_bytes = self.max_file_size_mb * 1024 * 1024
|
||||
if file_size > max_size_bytes:
|
||||
raise ValidationError(f"Размер файла ({file_size / (1024*1024):.2f} МБ) "
|
||||
f"превышает максимально допустимый ({self.max_file_size_mb} МБ)")
|
||||
|
||||
# Проверка MIME-типа
|
||||
try:
|
||||
mime_type = magic.from_file(file_path, mime=True)
|
||||
if mime_type not in self.allowed_mime_types:
|
||||
raise ValidationError(f"MIME-тип файла ({mime_type}) не разрешен. "
|
||||
f"Разрешенные MIME-типы: {', '.join(self.allowed_mime_types)}")
|
||||
except ValidationError:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.warning(f"Не удалось определить MIME-тип файла: {e}")
|
||||
|
||||
return True
|
||||
|
||||
@staticmethod
|
||||
def validate_local_file_path(file_path: str, allowed_directories: Optional[List[str]] = None) -> str:
|
||||
"""
|
||||
Reference in New Issue
Block a user