""" Скрипт для загрузки изображений из папки 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("❌ Неверный выбор")