271 lines
11 KiB
Python
271 lines
11 KiB
Python
"""
|
||
Скрипт для загрузки изображений из папки images/ в базу данных
|
||
"""
|
||
|
||
import os
|
||
import mimetypes
|
||
from pathlib import Path
|
||
from database.database import db, get_db_session
|
||
from database.services import ImageService
|
||
from database.repositories import ImageRepository
|
||
import logging
|
||
|
||
logging.basicConfig(level=logging.INFO)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
def get_image_content_type(filename: str) -> str:
|
||
"""Определение MIME типа изображения по расширению файла"""
|
||
mime_type, _ = mimetypes.guess_type(filename)
|
||
|
||
# Если mimetypes не смог определить, делаем по расширению
|
||
if not mime_type:
|
||
ext = filename.lower().split('.')[-1]
|
||
mime_types = {
|
||
'jpg': 'image/jpeg',
|
||
'jpeg': 'image/jpeg',
|
||
'png': 'image/png',
|
||
'gif': 'image/gif',
|
||
'bmp': 'image/bmp',
|
||
'webp': 'image/webp'
|
||
}
|
||
mime_type = mime_types.get(ext, 'image/jpeg')
|
||
|
||
return mime_type
|
||
|
||
|
||
def load_image_to_db(image_path: str, alt_text: str = None) -> bool:
|
||
"""
|
||
Загрузка одного изображения в базу данных
|
||
|
||
Args:
|
||
image_path: Путь к файлу изображения
|
||
alt_text: Альтернативный текст для изображения
|
||
|
||
Returns:
|
||
bool: True если загрузка успешна, False если ошибка
|
||
"""
|
||
try:
|
||
# Проверяем, существует ли файл
|
||
if not os.path.isfile(image_path):
|
||
logger.error(f"Файл не найден: {image_path}")
|
||
return False
|
||
|
||
# Получаем информацию о файле
|
||
filename = os.path.basename(image_path)
|
||
content_type = get_image_content_type(filename)
|
||
|
||
# Читаем бинарные данные
|
||
with open(image_path, 'rb') as f:
|
||
image_data = f.read()
|
||
|
||
# Проверяем размер файла (ограничение PostgreSQL ~1GB, но лучше меньше)
|
||
file_size = len(image_data)
|
||
if file_size > 10 * 1024 * 1024: # 10MB лимит
|
||
logger.warning(f"Файл слишком большой ({file_size} байт): {filename}")
|
||
return False
|
||
|
||
# Сохраняем в базу данных
|
||
with get_db_session() as session:
|
||
# Проверяем, не существует ли уже такой файл
|
||
existing_image = ImageRepository.get_image_by_filename(session, filename)
|
||
if existing_image:
|
||
logger.info(f"Изображение уже существует в БД: {filename}")
|
||
return True
|
||
|
||
# Создаем новую запись
|
||
image = ImageRepository.create_image(
|
||
session=session,
|
||
filename=filename,
|
||
content_type=content_type,
|
||
image_data=image_data,
|
||
alt_text=alt_text or f"Изображение {filename}"
|
||
)
|
||
|
||
session.commit()
|
||
logger.info(f"Загружено изображение: {filename} (ID: {image.id}, размер: {file_size} байт)")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при загрузке изображения {image_path}: {e}")
|
||
return False
|
||
|
||
|
||
def load_images_from_directory(images_dir: str = "images") -> dict:
|
||
"""
|
||
Загрузка всех изображений из указанной директории
|
||
|
||
Args:
|
||
images_dir: Путь к директории с изображениями
|
||
|
||
Returns:
|
||
dict: Статистика загрузки
|
||
"""
|
||
if not os.path.isdir(images_dir):
|
||
logger.error(f"Директория не найдена: {images_dir}")
|
||
return {"success": 0, "failed": 0, "skipped": 0}
|
||
|
||
# Поддерживаемые форматы изображений
|
||
supported_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.webp'}
|
||
|
||
stats = {"success": 0, "failed": 0, "skipped": 0}
|
||
|
||
logger.info(f"Начинаем загрузку изображений из директории: {images_dir}")
|
||
|
||
# Проходим по всем файлам в директории
|
||
for filename in os.listdir(images_dir):
|
||
file_path = os.path.join(images_dir, filename)
|
||
|
||
# Пропускаем директории
|
||
if os.path.isdir(file_path):
|
||
continue
|
||
|
||
# Проверяем расширение файла
|
||
file_ext = Path(filename).suffix.lower()
|
||
if file_ext not in supported_extensions:
|
||
logger.info(f"Пропускаем файл неподдерживаемого формата: {filename}")
|
||
stats["skipped"] += 1
|
||
continue
|
||
|
||
# Загружаем изображение
|
||
if load_image_to_db(file_path):
|
||
stats["success"] += 1
|
||
else:
|
||
stats["failed"] += 1
|
||
|
||
logger.info(f"Загрузка завершена. Успешно: {stats['success']}, "
|
||
f"Ошибок: {stats['failed']}, Пропущено: {stats['skipped']}")
|
||
|
||
return stats
|
||
|
||
|
||
def list_images_in_db():
|
||
"""Показать все изображения в базе данных"""
|
||
try:
|
||
with get_db_session() as session:
|
||
images = ImageRepository.get_all_images(session)
|
||
|
||
if not images:
|
||
logger.info("В базе данных нет изображений")
|
||
return
|
||
|
||
logger.info(f"Изображения в базе данных ({len(images)} шт.):")
|
||
for image in images:
|
||
size_kb = image.file_size / 1024
|
||
logger.info(f" ID: {image.id}, Файл: {image.filename}, "
|
||
f"Тип: {image.content_type}, Размер: {size_kb:.1f} KB")
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при получении списка изображений: {e}")
|
||
|
||
|
||
def delete_image_from_db(filename: str = None, image_id: int = None) -> bool:
|
||
"""
|
||
Удаление изображения из базы данных
|
||
|
||
Args:
|
||
filename: Имя файла для удаления
|
||
image_id: ID изображения для удаления
|
||
|
||
Returns:
|
||
bool: True если удаление успешно
|
||
"""
|
||
try:
|
||
with get_db_session() as session:
|
||
if filename:
|
||
image = ImageRepository.get_image_by_filename(session, filename)
|
||
elif image_id:
|
||
image = ImageRepository.get_image_by_id(session, image_id)
|
||
else:
|
||
logger.error("Необходимо указать filename или image_id")
|
||
return False
|
||
|
||
if not image:
|
||
logger.error("Изображение не найдено")
|
||
return False
|
||
|
||
if ImageRepository.delete_image(session, image.id):
|
||
session.commit()
|
||
logger.info(f"Изображение удалено: {image.filename}")
|
||
return True
|
||
else:
|
||
logger.error("Ошибка при удалении изображения")
|
||
return False
|
||
|
||
except Exception as e:
|
||
logger.error(f"Ошибка при удалении изображения: {e}")
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
"""Главная функция для запуска скрипта"""
|
||
|
||
print("🖼️ Скрипт загрузки изображений в базу данных")
|
||
print("=" * 50)
|
||
|
||
# Проверяем подключение к БД
|
||
try:
|
||
from database.database import test_db_connection
|
||
if not test_db_connection():
|
||
print("❌ Ошибка подключения к базе данных!")
|
||
exit(1)
|
||
print("✅ Подключение к базе данных успешно")
|
||
except Exception as e:
|
||
print(f"❌ Ошибка: {e}")
|
||
exit(1)
|
||
|
||
# Меню
|
||
while True:
|
||
print("\nВыберите действие:")
|
||
print("1. Загрузить все изображения из папки images/")
|
||
print("2. Загрузить конкретное изображение")
|
||
print("3. Показать все изображения в БД")
|
||
print("4. Удалить изображение из БД")
|
||
print("5. Выход")
|
||
|
||
choice = input("\nВведите номер (1-5): ").strip()
|
||
|
||
if choice == "1":
|
||
stats = load_images_from_directory("images")
|
||
print(f"\n📊 Результат: загружено {stats['success']}, "
|
||
f"ошибок {stats['failed']}, пропущено {stats['skipped']}")
|
||
|
||
elif choice == "2":
|
||
image_path = input("Введите путь к изображению: ").strip()
|
||
alt_text = input("Введите описание (необязательно): ").strip() or None
|
||
if load_image_to_db(image_path, alt_text):
|
||
print("✅ Изображение загружено!")
|
||
else:
|
||
print("❌ Ошибка загрузки")
|
||
|
||
elif choice == "3":
|
||
list_images_in_db()
|
||
|
||
elif choice == "4":
|
||
print("Удаление по:")
|
||
print("1. Имени файла")
|
||
print("2. ID")
|
||
sub_choice = input("Выберите (1-2): ").strip()
|
||
|
||
if sub_choice == "1":
|
||
filename = input("Введите имя файла: ").strip()
|
||
if delete_image_from_db(filename=filename):
|
||
print("✅ Изображение удалено!")
|
||
else:
|
||
print("❌ Ошибка удаления")
|
||
elif sub_choice == "2":
|
||
try:
|
||
image_id = int(input("Введите ID: ").strip())
|
||
if delete_image_from_db(image_id=image_id):
|
||
print("✅ Изображение удалено!")
|
||
else:
|
||
print("❌ Ошибка удаления")
|
||
except ValueError:
|
||
print("❌ Неверный ID")
|
||
|
||
elif choice == "5":
|
||
print("👋 До свидания!")
|
||
break
|
||
|
||
else:
|
||
print("❌ Неверный выбор") |