Python/excel-to-bd/main.py - main - Demek/DSam621

Python/excel-to-bd/main.py
122 строки5 KB
d_e_m_e_k
теперь загружается excel файл с произвольным количеством листов. для каждого листа создается своя таблица и в нее загружаются данные.
07 авг 2026, 15:55
07 авг 2026, 15:55a8ce725
import logging
import sys
from pathlib import Path
from excel_analyzer import ExcelAnalyzer
from excel_loader import ExcelLoader
from db_manager import DatabaseManager
from config import Config
import time

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('project.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

def main():
    logger.info("🚀 Запуск проекта Excel -> PostgreSQL")
    start_time = time.time()
    
    try:
        # Проверяем наличие Excel файла
        excel_file = Config.EXCEL_FILE
        if not Path(excel_file).exists():
            logger.error(f"❌ Файл не найден: {excel_file}")
            logger.info(f"📁 Пожалуйста, поместите Excel файл в папку {Config.DATA_DIR}")
            logger.info(f"   или укажите путь в .env (EXCEL_FILE=...)")
            
            logger.info("\n💡 Для создания тестовых данных выполните:")
            logger.info("   python generate_excel.py")
            return False
        
        # Шаг 1: Загрузка Excel
        logger.info(f"📂 Загрузка Excel файла: {excel_file}")
        loader = ExcelLoader(excel_file)
        
        # Показываем все листы
        sheets = loader.get_all_sheets()
        logger.info(f"📋 Найдено листов: {len(sheets)}")
        for i, sheet in enumerate(sheets, 1):
            table_name = loader.get_table_name_from_sheet(sheet)
            logger.info(f"   {i}. {sheet} -> таблица: {table_name}")
        
        # Шаг 2: Подключение к PostgreSQL
        logger.info(f"🗄️ Подключение к PostgreSQL...")
        logger.info(f"   Host: {Config.DB_HOST}:{Config.DB_PORT}")
        logger.info(f"   Database: {Config.DB_NAME}")
        logger.info(f"   User: {Config.DB_USER}")
        
        db = DatabaseManager()
        
        if not db.connect():
            logger.error("❌ Не удалось подключиться к PostgreSQL")
            logger.info("\n💡 Проверьте:")
            logger.info("   1. Запущен ли PostgreSQL сервер")
            logger.info(f"   2. Правильные ли настройки в .env файле")
            return False
        
        # Показываем превью каждого листа
        loader.preview_all_sheets(rows=3)
        
        # Спрашиваем подтверждение
        print("\n" + "="*80)
        print(f"📊 Данные для загрузки:")
        print(f"   - Листов в Excel: {len(sheets)}")
        print(f"   - Будет создано таблиц: {len(sheets)}")
        print("="*80)
        
        response = input("\nЗагрузить все листы в отдельные таблицы PostgreSQL? (y/n): ")
        if response.lower() != 'y':
            logger.info("⏹️ Загрузка отменена пользователем")
            db.close()
            return True
        
        # Шаг 3: Загрузка всех листов
        logger.info("📤 Загрузка данных в PostgreSQL...")
        
        results = loader.load_all_sheets(db)
        
        # Сохраняем отчет о загрузке
        loader.save_loading_report(results)
        
        # Шаг 4: Показываем созданные таблицы
        logger.info("🔍 Проверка созданных таблиц...")
        
        tables = db.get_table_list()
        if tables is not None and len(tables) > 0:
            logger.info(f"📋 Таблицы в базе данных:")
            for _, row in tables.iterrows():
                table_name = row['table_name']
                # Получаем количество записей в каждой таблице
                count_query = f"SELECT COUNT(*) as count FROM {table_name};"
                count_result = db.execute_query(count_query)
                if count_result is not None and len(count_result) > 0:
                    count = count_result['count'].iloc[0]
                    logger.info(f"   - {table_name}: {count} записей")
        
        db.close()
        
        elapsed_time = time.time() - start_time
        logger.info(f"✅ Проект успешно завершен за {elapsed_time:.2f} секунд")
        return True
        
    except FileNotFoundError as e:
        logger.error(f"❌ Ошибка: {e}")
        logger.info("📁 Убедитесь, что Excel файл существует и путь указан верно")
        return False
    except KeyboardInterrupt:
        logger.info("⏹️ Прервано пользователем")
        return False
    except Exception as e:
        logger.error(f"❌ Критическая ошибка: {e}")
        import traceback
        traceback.print_exc()
        return False

if __name__ == "__main__":
    success = main()
    sys.exit(0 if success else 1)