excel_manager.py - master - valuevvik/wbAnalytics

excel_manager.py
80 строк3 KB
valuevvik
initial commit
05 дек 2025, 17:36
05 дек 2025, 17:36f514336
import pandas as pd
import os
from openpyxl import load_workbook
from openpyxl.utils.dataframe import dataframe_to_rows
from typing import Callable


class ExcelManager:
    def save_to_excel(self, df: pd.DataFrame, output_file: str,
                      save_mode: str, log_callback: Callable = None) -> bool:
        if save_mode == "append" and os.path.exists(output_file):
            return self._append_to_excel(df, output_file, log_callback)
        else:
            return self._create_new_excel(df, output_file, log_callback)

    def _create_new_excel(self, df: pd.DataFrame, output_file: str,
                          log_callback: Callable = None) -> bool:
        try:
            df.to_excel(output_file, index=False, engine='openpyxl')
            if log_callback:
                log_callback(f"💾 Отчет сохранен в: {output_file}")
            return True
        except Exception as e:
            if log_callback:
                log_callback(f"❌ Ошибка при создании файла: {str(e)}")
            return False

    def _append_to_excel(self, df: pd.DataFrame, output_file: str,
                         log_callback: Callable = None) -> bool:
        try:
            wb = load_workbook(output_file)

            if 'DATA' not in wb.sheetnames:
                if log_callback:
                    log_callback("❌ Ошибка: лист 'DATA' не найден в файле")
                return False

            ws = wb['DATA']

            last_row = 1
            while ws.cell(row=last_row, column=1).value is not None:
                last_row += 1

            if log_callback:
                log_callback(f"📊 Последняя заполненная строка: {last_row - 1}")

            start_row = last_row if last_row > 1 else last_row

            for r_idx, row in enumerate(dataframe_to_rows(df, index=False, header=False), start_row):
                for c_idx, value in enumerate(row, 1):
                    ws.cell(row=r_idx, column=c_idx, value=value)

            if ws.tables:
                for table_name in ws.tables:
                    table = ws.tables[table_name]
                    new_ref = f"{table.ref.split(':')[0]}:{ws.cell(row=last_row + len(df) - 1, column=len(df.columns)).coordinate}"
                    table.ref = new_ref
                    if log_callback:
                        log_callback(f"🔄 Обновлен диапазон таблицы: {new_ref}")

            wb.save(output_file)
            if log_callback:
                log_callback(f"✅ Добавлено {len(df)} записей в лист DATA")
            return True

        except Exception as e:
            if log_callback:
                log_callback(f"❌ Ошибка при добавлении данных: {str(e)}")
            return False

    def read_existing_data(self, file_path: str, log_callback: Callable = None) -> pd.DataFrame:
        try:
            df = pd.read_excel(file_path, sheet_name='DATA', engine='openpyxl')
            if log_callback:
                log_callback(f"📁 Загружено записей из файла: {len(df)}")
            return df
        except Exception as e:
            if log_callback:
                log_callback(f"❌ Ошибка при чтении файла: {str(e)}")
            raise e