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

Python/excel-to-bd/excel_loader.py
578 строк26 KB
d_e_m_e_k
поправил логику. теперь таблицы не пересоздаются каждый запуск, а дописываются новыми данными (делаем проверу на дубликаты)
07 авг 2026, 16:07
07 авг 2026, 16:079414e18
import pandas as pd
import numpy as np
from sqlalchemy import create_engine, text
from config import Config
import logging
from pathlib import Path
from typing import List, Dict, Optional, Tuple
import re
from sqlalchemy.exc import IntegrityError

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ExcelLoader:
    def __init__(self, file_path=None):
        self.file_path = file_path or Config.EXCEL_FILE
        self.table_name = Config.TABLE_NAME
        self.df = None
        self.engine = None
        self.all_sheets_data = {}
        
    def validate_file(self):
        """Проверяет существование файла"""
        if not Path(self.file_path).exists():
            raise FileNotFoundError(f"Файл {self.file_path} не найден")
        
        if not self.file_path.endswith(('.xlsx', '.xls')):
            raise ValueError(f"Неподдерживаемый формат файла: {self.file_path}")
        
        logger.info(f"✅ Файл {self.file_path} найден")
        return True
    
    def get_all_sheets(self) -> List[str]:
        """Получает список всех листов в Excel файле"""
        try:
            excel_file = pd.ExcelFile(self.file_path)
            sheets = excel_file.sheet_names
            logger.info(f"📋 Найдено листов в Excel: {len(sheets)}")
            for i, sheet in enumerate(sheets, 1):
                logger.info(f"   {i}. {sheet}")
            return sheets
        except Exception as e:
            logger.error(f"❌ Ошибка получения списка листов: {e}")
            return []
    
    def load_sheet(self, sheet_name: str) -> Optional[pd.DataFrame]:
        """Загружает конкретный лист Excel"""
        try:
            df = pd.read_excel(self.file_path, sheet_name=sheet_name)
            logger.info(f"✅ Загружен лист '{sheet_name}': {len(df)} строк, {len(df.columns)} колонок")
            return df
        except Exception as e:
            logger.error(f"❌ Ошибка загрузки листа '{sheet_name}': {e}")
            return None
    
    def transliterate(self, text: str) -> str:
        """Транслитерирует русский текст в английский"""
        translit_dict = {
            'а': 'a', 'б': 'b', 'в': 'v', 'г': 'g', 'д': 'd', 'е': 'e', 'ё': 'e',
            'ж': 'zh', 'з': 'z', 'и': 'i', 'й': 'y', 'к': 'k', 'л': 'l', 'м': 'm',
            'н': 'n', 'о': 'o', 'п': 'p', 'р': 'r', 'с': 's', 'т': 't', 'у': 'u',
            'ф': 'f', 'х': 'kh', 'ц': 'ts', 'ч': 'ch', 'ш': 'sh', 'щ': 'shch',
            'ъ': '', 'ы': 'y', 'ь': '', 'э': 'e', 'ю': 'yu', 'я': 'ya',
            'А': 'A', 'Б': 'B', 'В': 'V', 'Г': 'G', 'Д': 'D', 'Е': 'E', 'Ё': 'E',
            'Ж': 'Zh', 'З': 'Z', 'И': 'I', 'Й': 'Y', 'К': 'K', 'Л': 'L', 'М': 'M',
            'Н': 'N', 'О': 'O', 'П': 'P', 'Р': 'R', 'С': 'S', 'Т': 'T', 'У': 'U',
            'Ф': 'F', 'Х': 'Kh', 'Ц': 'Ts', 'Ч': 'Ch', 'Ш': 'Sh', 'Щ': 'Shch',
            'Ъ': '', 'Ы': 'Y', 'Ь': '', 'Э': 'E', 'Ю': 'Yu', 'Я': 'Ya'
        }
        
        result = ''
        for char in text:
            if char in translit_dict:
                result += translit_dict[char]
            elif char.isalnum() or char == '_':
                result += char
            else:
                if char.isspace() or char in '.,;:!?-()[]{}':
                    result += '_'
                else:
                    result += '_'
        
        result = re.sub(r'_+', '_', result)
        result = result.strip('_')
        result = result.lower()
        
        return result
    
    def clean_column_name(self, col_name: str) -> str:
        """Очищает и транслитерирует название колонки"""
        col_clean = self.transliterate(col_name)
        col_clean = re.sub(r'[^a-zA-Z0-9_]', '_', col_clean)
        col_clean = re.sub(r'_+', '_', col_clean)
        col_clean = col_clean.strip('_')
        if col_clean and col_clean[0].isdigit():
            col_clean = 'col_' + col_clean
        return col_clean
    
    def get_table_name_from_sheet(self, sheet_name: str) -> str:
        """Генерирует имя таблицы из имени листа"""
        table_name = self.transliterate(sheet_name)
        if len(table_name) > 63:
            table_name = table_name[:63]
        return table_name
    
    def get_sql_type(self, series: pd.Series, column_name: str) -> str:
        """Определяет SQL тип для колонки на основе данных"""
        
        # Специальная обработка для колонки id
        if column_name.lower() == 'id':
            return 'INTEGER'
        
        # Проверяем, является ли колонка датой
        if 'date' in column_name.lower() or 'dt' in column_name.lower():
            try:
                sample = series.dropna()
                if len(sample) > 0:
                    pd.to_datetime(sample.iloc[0])
                    try:
                        pd.to_datetime(sample, errors='raise')
                        return 'DATE'
                    except:
                        pass
            except:
                pass
        
        # Проверяем, является ли колонка булевой
        if 'active' in column_name.lower() or 'status' in column_name.lower():
            unique_values = series.dropna().unique()
            if len(unique_values) <= 2:
                bool_like = all(
                    isinstance(v, bool) or 
                    str(v).lower() in ['true', 'false', 'active', 'inactive', 'активный', 'неактивный', 'yes', 'no', 'y', 'n', '1', '0']
                    for v in unique_values
                )
                if bool_like:
                    return 'BOOLEAN'
        
        # Проверяем, является ли колонка числовой
        if pd.api.types.is_numeric_dtype(series):
            if pd.api.types.is_integer_dtype(series):
                return 'INTEGER'
            else:
                # Проверяем размер чисел
                max_val = series.max()
                if pd.isna(max_val):
                    return 'DECIMAL(10,2)'
                
                # Определяем нужную точность
                if max_val > 1000000000:  # > 1 миллиарда
                    return 'DECIMAL(20,2)'
                elif max_val > 10000000:  # > 10 миллионов
                    return 'DECIMAL(15,2)'
                else:
                    return 'DECIMAL(10,2)'
        
        # Определяем длину строки
        if series.dtype == 'object':
            str_series = series.astype(str)
            max_len = str_series.str.len().max()
            if pd.isna(max_len) or max_len == 0:
                max_len = 255
            if max_len > 1000:
                return 'TEXT'
            else:
                return f'VARCHAR({max(1, min(max_len, 1000))})'
        
        return 'TEXT'
    
    def table_exists(self, table_name: str, db_manager) -> bool:
        """Проверяет существование таблицы"""
        query = """
        SELECT EXISTS (
            SELECT 1 
            FROM information_schema.tables 
            WHERE table_name = %s
            AND table_schema = 'public'
        );
        """
        try:
            result = db_manager.execute_query(query, (table_name,))
            if result is not None and len(result) > 0:
                return result['exists'].iloc[0] if 'exists' in result.columns else False
            return False
        except Exception as e:
            logger.warning(f"⚠️ Ошибка проверки таблицы {table_name}: {e}")
            return False
    
    def get_table_primary_key(self, table_name: str, db_manager) -> Optional[str]:
        """Получает имя первичного ключа таблицы"""
        query = """
        SELECT 
            kcu.column_name
        FROM information_schema.table_constraints tc
        JOIN information_schema.key_column_usage kcu
            ON tc.constraint_name = kcu.constraint_name
            AND tc.table_schema = kcu.table_schema
        WHERE tc.constraint_type = 'PRIMARY KEY'
            AND tc.table_name = %s
            AND tc.table_schema = 'public'
        LIMIT 1;
        """
        try:
            result = db_manager.execute_query(query, (table_name,))
            if result is not None and len(result) > 0:
                return result['column_name'].iloc[0]
            return None
        except Exception as e:
            logger.warning(f"⚠️ Ошибка получения первичного ключа таблицы {table_name}: {e}")
            return None
    
    def check_duplicates(self, df: pd.DataFrame, table_name: str, db_manager, pk_column: str) -> pd.DataFrame:
        """Проверяет дубликаты в данных по сравнению с существующей таблицей"""
        try:
            # Получаем существующие значения первичного ключа из таблицы
            query = f"SELECT {pk_column} FROM {table_name};"
            existing_data = db_manager.execute_query(query)
            
            if existing_data is None or len(existing_data) == 0:
                # Таблица пуста - все данные новые
                logger.info(f"   Таблица {table_name} пуста, все данные будут загружены")
                return df
            
            # Получаем множество существующих id
            existing_ids = set(existing_data[pk_column].astype(str).values)
            
            # Фильтруем новые записи (которых нет в таблице)
            df['_temp_id_str'] = df[pk_column].astype(str)
            new_df = df[~df['_temp_id_str'].isin(existing_ids)].copy()
            new_df = new_df.drop(columns=['_temp_id_str'])
            
            duplicates_count = len(df) - len(new_df)
            if duplicates_count > 0:
                logger.info(f"   Найдено дубликатов: {duplicates_count} (пропущены)")
                logger.info(f"   Будет загружено новых записей: {len(new_df)}")
            
            return new_df
            
        except Exception as e:
            logger.warning(f"⚠️ Ошибка проверки дубликатов: {e}")
            logger.info("   Все данные будут загружены (проверка дубликатов пропущена)")
            return df
    
    def create_table_from_sheet(self, df: pd.DataFrame, table_name: str, db_manager) -> bool:
        """Создает таблицу на основе данных листа (только если её нет)"""
        try:
            # Проверяем, существует ли таблица
            if self.table_exists(table_name, db_manager):
                logger.info(f"   Таблица {table_name} уже существует, пропускаем создание")
                return True
            
            # Очищаем названия колонок
            df.columns = [self.clean_column_name(col) for col in df.columns]
            
            # Формируем SQL для создания таблицы
            columns_sql = []
            has_id = False
            
            for col in df.columns:
                if col.lower() == 'id':
                    has_id = True
                    columns_sql.append(f"{col} INTEGER PRIMARY KEY")
                elif col.lower() in ['created_at', 'updated_at']:
                    continue
                else:
                    sql_type = self.get_sql_type(df[col], col)
                    columns_sql.append(f"{col} {sql_type}")
            
            # Добавляем служебные колонки
            if not has_id:
                columns_sql.insert(0, "id SERIAL PRIMARY KEY")
            columns_sql.append("created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP")
            columns_sql.append("updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP")
            
            create_table_sql = f"""
            CREATE TABLE IF NOT EXISTS {table_name} (
                {', '.join(columns_sql)}
            );
            """
            
            # Выполняем создание таблицы
            db_manager.execute_query(create_table_sql)
            logger.info(f"✅ Таблица {table_name} создана с {len(columns_sql)} колонками")
            
            return True
            
        except Exception as e:
            logger.error(f"❌ Ошибка создания таблицы {table_name}: {e}")
            return False
    
    def prepare_sheet_for_database(self, df: pd.DataFrame) -> pd.DataFrame:
        """Подготавливает данные листа для загрузки в БД"""
        df_clean = df.copy()
        
        # Очищаем названия колонок
        df_clean.columns = [self.clean_column_name(col) for col in df_clean.columns]
        
        # Добавляем id если нет
        if 'id' not in df_clean.columns:
            df_clean.insert(0, 'id', range(1, len(df_clean) + 1))
        
        # Преобразуем типы данных
        for col in df_clean.columns:
            if col == 'id':
                df_clean[col] = df_clean[col].astype(int)
                continue
            
            # Дата
            if 'date' in col.lower() or 'dt' in col.lower():
                try:
                    df_clean[col] = pd.to_datetime(df_clean[col]).dt.date
                except:
                    logger.warning(f"⚠️ Не удалось преобразовать {col} в дату")
                    df_clean[col] = None
            
            # Булевы значения
            if 'active' in col.lower() or 'status' in col.lower():
                if df_clean[col].dtype == 'object':
                    df_clean[col] = df_clean[col].map({
                        'Активный': True, 'Неактивный': False,
                        'Active': True, 'Inactive': False,
                        'True': True, 'False': False,
                        'Yes': True, 'No': False,
                        'Y': True, 'N': False,
                        1: True, 0: False,
                        'true': True, 'false': False,
                        'TRUE': True, 'FALSE': False
                    }).fillna(False)
                elif pd.api.types.is_numeric_dtype(df_clean[col]):
                    df_clean[col] = df_clean[col].astype(bool)
            
            # Числовые поля
            if pd.api.types.is_numeric_dtype(df_clean[col]):
                if col not in ['id']:
                    continue
                # Если все значения целые, преобразуем в int
                if df_clean[col].dropna().apply(lambda x: x == int(x)).all():
                    df_clean[col] = df_clean[col].astype(float).fillna(0).astype(int)
        
        # Заменяем NaN на None
        df_clean = df_clean.where(pd.notnull(df_clean), None)
        
        return df_clean
    
    def load_sheet_to_database(self, df: pd.DataFrame, sheet_name: str, db_manager) -> Tuple[bool, int, str]:
        """Загружает лист в отдельную таблицу"""
        try:
            # Подготавливаем данные
            df_clean = self.prepare_sheet_for_database(df)
            
            # Получаем имя таблицы из имени листа
            table_name = self.get_table_name_from_sheet(sheet_name)
            
            logger.info(f"   Имя таблицы: {table_name}")
            logger.info(f"   Колонки: {', '.join(df_clean.columns)}")
            
            # Проверяем существование таблицы
            table_exists = self.table_exists(table_name, db_manager)
            
            if not table_exists:
                # Создаем таблицу
                if not self.create_table_from_sheet(df_clean, table_name, db_manager):
                    return False, 0, "Failed to create table"
            else:
                logger.info(f"   Таблица {table_name} уже существует")
                
                # Проверяем дубликаты
                pk_column = self.get_table_primary_key(table_name, db_manager)
                
                if pk_column and pk_column in df_clean.columns:
                    logger.info(f"   Проверка дубликатов по колонке: {pk_column}")
                    df_clean = self.check_duplicates(df_clean, table_name, db_manager, pk_column)
                    
                    if len(df_clean) == 0:
                        logger.info(f"   Все записи уже существуют в таблице, загрузка не требуется")
                        return True, 0, "All records already exist"
                else:
                    logger.warning(f"   ⚠️ Не удалось определить первичный ключ для проверки дубликатов")
                    logger.info("   Будет выполнена попытка вставки всех данных (дубликаты будут пропущены)")
            
            # Если нет данных для загрузки
            if len(df_clean) == 0:
                return True, 0, "No new records to load"
            
            # Загружаем данные
            engine = create_engine(Config.get_db_url())
            
            try:
                df_clean.to_sql(
                    table_name,
                    engine,
                    if_exists='append',
                    index=False,
                    method='multi',
                    chunksize=Config.CHUNK_SIZE
                )
                
                logger.info(f"✅ Загружено {len(df_clean)} новых записей в таблицу {table_name}")
                return True, len(df_clean), "Success"
                
            except IntegrityError as e:
                # Если возникла ошибка уникальности, загружаем по одной записи
                logger.warning(f"⚠️ Ошибка уникальности при массовой загрузке: {e}")
                logger.info("   Выполняется поштучная загрузка...")
                
                loaded = 0
                for index, row in df_clean.iterrows():
                    try:
                        row_df = pd.DataFrame([row])
                        row_df.to_sql(
                            table_name,
                            engine,
                            if_exists='append',
                            index=False
                        )
                        loaded += 1
                    except IntegrityError:
                        # Пропускаем дубликат
                        continue
                    except Exception as e:
                        logger.warning(f"   Ошибка загрузки строки {index}: {e}")
                        continue
                
                if loaded > 0:
                    logger.info(f"✅ Загружено {loaded} новых записей (пропущено дубликатов: {len(df_clean) - loaded})")
                    return True, loaded, f"Loaded {loaded} records (skipped duplicates)"
                else:
                    logger.info(f"   Все записи уже существуют в таблице")
                    return True, 0, "All records already exist"
            
        except Exception as e:
            error_msg = str(e)
            logger.error(f"❌ Ошибка загрузки листа '{sheet_name}': {e}")
            return False, 0, f"Error: {error_msg[:100]}"
    
    def load_all_sheets(self, db_manager) -> Dict[str, Dict]:
        """Загружает все листы Excel в отдельные таблицы"""
        results = {}
        
        sheets = self.get_all_sheets()
        
        if not sheets:
            logger.error("❌ В Excel файле нет листов")
            return results
        
        logger.info(f"🔄 Начинаем загрузку {len(sheets)} листов...")
        logger.info("="*60)
        
        for sheet_name in sheets:
            logger.info(f"\n📄 Обработка листа: {sheet_name}")
            logger.info("-" * 50)
            
            df = self.load_sheet(sheet_name)
            if df is None or len(df) == 0:
                results[sheet_name] = {
                    'status': 'empty',
                    'records': 0,
                    'table_name': '',
                    'message': 'Sheet is empty or could not be loaded'
                }
                logger.warning(f"⚠️ Лист '{sheet_name}' пуст или не загружен")
                continue
            
            success, count, message = self.load_sheet_to_database(df, sheet_name, db_manager)
            
            table_name = self.get_table_name_from_sheet(sheet_name)
            
            if success:
                results[sheet_name] = {
                    'status': 'success',
                    'records': count,
                    'table_name': table_name,
                    'columns': len(df.columns),
                    'message': message
                }
                if count > 0:
                    logger.info(f"✅ Лист '{sheet_name}': загружено {count} записей в таблицу {table_name}")
                else:
                    logger.info(f"ℹ️ Лист '{sheet_name}': новые записи не найдены")
            else:
                results[sheet_name] = {
                    'status': 'failed',
                    'records': 0,
                    'table_name': table_name,
                    'columns': len(df.columns),
                    'message': message
                }
                logger.error(f"❌ Лист '{sheet_name}': ошибка загрузки")
        
        total_success = sum(1 for r in results.values() if r['status'] == 'success')
        total_failed = sum(1 for r in results.values() if r['status'] == 'failed')
        total_empty = sum(1 for r in results.values() if r['status'] == 'empty')
        total_records = sum(r['records'] for r in results.values() if r['status'] == 'success')
        
        logger.info("\n" + "="*60)
        logger.info(f"📊 ИТОГИ ЗАГРУЗКИ:")
        logger.info(f"   Всего листов: {len(sheets)}")
        logger.info(f"   ✅ Успешно загружено: {total_success}")
        logger.info(f"   ❌ Ошибок: {total_failed}")
        logger.info(f"   📭 Пустых: {total_empty}")
        logger.info(f"   📝 Всего новых записей: {total_records}")
        logger.info("="*60)
        
        return results
    
    def preview_all_sheets(self, rows=5):
        """Показывает превью всех листов"""
        sheets = self.get_all_sheets()
        
        print("\n" + "="*80)
        print(f"📊 ПРЕВЬЮ ВСЕХ ЛИСТОВ (файл: {self.file_path})")
        print("="*80)
        
        for sheet_name in sheets:
            print(f"\n📄 Лист: {sheet_name}")
            print("-"*60)
            
            df = self.load_sheet(sheet_name)
            if df is None:
                print("   ❌ Не удалось загрузить")
                continue
            
            table_name = self.get_table_name_from_sheet(sheet_name)
            
            print(f"   Таблица: {table_name}")
            print(f"   Всего строк: {len(df)}")
            print(f"   Всего колонок: {len(df.columns)}")
            print(f"   Колонки: {', '.join(df.columns[:10])}")
            if len(df.columns) > 10:
                print(f"   ... и еще {len(df.columns) - 10} колонок")
            
            print(f"\n   Превью (первые {rows} строк):")
            print(df.head(rows).to_string())
            print("\n   Типы данных:")
            print(df.dtypes.to_string())
    
    def generate_loading_report(self, results: Dict) -> str:
        """Генерирует отчет о загрузке"""
        report = f"""
# Excel Loading Report
**File:** {self.file_path}
**Date:** {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')}

## Summary
- Total Sheets Processed: {len(results)}
- Successfully Loaded: {sum(1 for r in results.values() if r['status'] == 'success')}
- Failed: {sum(1 for r in results.values() if r['status'] == 'failed')}
- Empty: {sum(1 for r in results.values() if r['status'] == 'empty')}
- Total New Records Loaded: {sum(r['records'] for r in results.values() if r['status'] == 'success')}

## Detailed Results
"""
        for sheet_name, info in results.items():
            if info['status'] == 'success':
                status_icon = '✅'
                status_text = 'SUCCESS'
            elif info['status'] == 'failed':
                status_icon = '❌'
                status_text = 'FAILED'
            else:
                status_icon = '📭'
                status_text = 'EMPTY'
            
            report += f"\n### {status_icon} {sheet_name}\n"
            report += f"- Status: **{status_text}**\n"
            report += f"- Table: `{info.get('table_name', 'N/A')}`\n"
            report += f"- New Records: {info.get('records', 0)}\n"
            report += f"- Total Columns: {info.get('columns', 0)}\n"
            report += f"- Message: {info.get('message', 'N/A')}\n"
        
        return report
    
    def save_loading_report(self, results: Dict, filename='loading_report.md'):
        """Сохраняет отчет о загрузке"""
        report = self.generate_loading_report(results)
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(report)
        logger.info(f"✅ Отчет о загрузке сохранен в {filename}")