src/analyzer.py - master - broot/order_analysis

src/analyzer.py
240 строк10 KB
broot
pycharm patch commit
13 июн 2026, 19:55
13 июн 2026, 19:55d81498a
import os
import logging
import pandas as pd


class OrderAnalyzer:
    """Класс для пакетного анализа файлов с заказами."""

    def __init__(self, data_dir, reports_dir, logs_dir,
                 status_col, status_val, amount_col):
        """
        Инициализирует анализатор, сохраняет настройки и готовит окружение.

        Параметры:
            data_dir (str): путь к папке с CSV-файлами.
            reports_dir (str): путь к папке для сохранения отчётов.
            logs_dir (str): путь к папке для логов.
            status_col (str): название колонки со статусом заказа.
            status_val (str): значение статуса для фильтрации (по условию - 'Delivered').
            amount_col (str): название колонки с суммой заказа.
        """
        self.data_dir = data_dir
        self.reports_dir = reports_dir
        self.logs_dir = logs_dir
        self.status_col = status_col
        self.status_val = status_val
        self.amount_col = amount_col

        # Создаём служебные папки, если их ещё нет
        os.makedirs(self.reports_dir, exist_ok=True)
        os.makedirs(self.logs_dir, exist_ok=True)

        # Настраиваем логирование
        self._setup_logging()

    def _setup_logging(self):
        """Настраивает запись логов в файл внутри папки logs/."""
        log_file = os.path.join(self.logs_dir, "errors.log")

        # Создаём логгер с именем класса
        self.logger = logging.getLogger("OrderAnalyzer")
        self.logger.setLevel(logging.INFO)

        # Чтобы не дублировались сообщения при повторных вызовах,
        # удаляем старые обработчики (на случай, если __init__ вызван несколько раз)
        if self.logger.handlers:
            self.logger.handlers.clear()

        # Файловый обработчик
        file_handler = logging.FileHandler(log_file, encoding='utf-8')
        file_handler.setLevel(logging.INFO)

        # Формат сообщений: время - уровень - текст
        formatter = logging.Formatter(
            '%(asctime)s - %(levelname)-8s - %(message)s',
            datefmt='%Y-%m-%d %H:%M:%S'
        )
        file_handler.setFormatter(formatter)

        self.logger.addHandler(file_handler)

    def load_file(self, filepath):
        """
        Загружает CSV-файл в DataFrame.

        Параметры:
            filepath (str): полный путь к файлу.

        Возвращает:
            pandas.DataFrame или None, если файл не удалось прочитать.
        """
        filename = os.path.basename(filepath)
        try:
            df = pd.read_csv(filepath)
            self.logger.info(f"Файл '{filename}' успешно загружен: {df.shape[0]} строк, {df.shape[1]} столбцов.")
            return df
        except FileNotFoundError:
            self.logger.error(f"Файл '{filename}' не найден.")
        except pd.errors.EmptyDataError:
            self.logger.error(f"Файл '{filename}' пуст.")
        except pd.errors.ParserError:
            self.logger.error(f"Файл '{filename}' повреждён или имеет неверную структуру CSV.")
        except Exception as e:
            self.logger.error(f"Не удалось загрузить файл '{filename}': {e}")
        return None

    def _validate_columns(self, df, filename):
        """
        Проверяет наличие обязательных колонок: status и total_amount.

        Параметры:
            df (pd.DataFrame): загруженный DataFrame.
            filename (str): имя файла для сообщения в лог.

        Возвращает:
            bool: True, если все обязательные колонки присутствуют, иначе False.
        """
        required = [self.status_col, self.amount_col]
        missing = [col for col in required if col not in df.columns]
        if missing:
            self.logger.error(
                f"Файл '{filename}' пропущен: отсутствуют обязательные колонки: {', '.join(missing)}. "
                f"Найдены колонки: {list(df.columns)}"
            )
            return False
        return True

    def filter_delivered(self, df):
        """
        Фильтрует DataFrame, оставляя только доставленные заказы.

        Параметры:
            df (pd.DataFrame): данные для фильтрации.

        Возвращает:
            pd.DataFrame: отфильтрованный DataFrame.
        """
        return df[df[self.status_col] == self.status_val]

    def calculate_metrics(self, df, filename):
        """
        Рассчитывает метрики для доставленных заказов.

        Параметры:
            df (pd.DataFrame): отфильтрованный DataFrame (только status == 'Delivered').
            filename (str): имя файла для сообщений в лог.

        Возвращает:
            dict или None: словарь с ключами total_revenue, average_order_value, order_count.
                           None в случае ошибки (нечисловые данные).
        """
        try:
            total_revenue = df[self.amount_col].sum()
            order_count = len(df)
            average_order_value = total_revenue / order_count if order_count > 0 else 0.0
        except (TypeError, ValueError) as e:
            self.logger.error(
                f"Файл '{filename}' пропущен: ошибка при расчёте метрик — {e}"
            )
            return None

        return {
            'total_revenue': round(total_revenue, 2),
            'average_order_value': round(average_order_value, 2),
            'order_count': order_count
        }

    def process_file(self, filepath):
        """
        Обрабатывает один CSV-файл: загружает, проверяет, фильтрует, считает метрики.

        Параметры:
            filepath (str): полный путь к файлу.

        Возвращает:
            dict или None: словарь с метриками и именем файла, либо None при ошибке.
        """
        filename = os.path.basename(filepath)
        try:
            # 1. Загрузка
            df = self.load_file(filepath)
            if df is None:
                return None

            # 2. Проверка колонок
            if not self._validate_columns(df, filename):
                return None

            # 3. Фильтрация доставленных
            delivered = self.filter_delivered(df)

            # 4. Расчёт метрик
            metrics = self.calculate_metrics(delivered, filename)
            if metrics is None:
                return None

            # Добавляем имя файла в результат
            metrics['file_name'] = filename
            self.logger.info(f"Файл '{filename}' успешно обработан: {metrics}")
            return metrics

        except Exception as e:
            self.logger.error(f"Файл '{filename}' пропущен из-за неожиданной ошибки: {e}")
            return None

    def process_all(self):
        """
        Обрабатывает все CSV-файлы в папке data_dir, сохраняет сводный отчёт.

        Возвращает:
            tuple: (processed_count, error_count) — количество успешно обработанных и пропущенных файлов.
        """
        # Получаем список всех CSV-файлов в папке с данными
        try:
            files = [f for f in os.listdir(self.data_dir) if f.endswith('.csv')]
        except FileNotFoundError:
            self.logger.error(f"Папка '{self.data_dir}' не найдена.")
            return 0, 0

        results = []
        processed = 0
        errors = 0

        for filename in files:
            filepath = os.path.join(self.data_dir, filename)
            metrics = self.process_file(filepath)
            if metrics:
                results.append(metrics)
                processed += 1
            else:
                errors += 1

        # Сохраняем отчёт, если есть успешные результаты
        if results:
            df_report = pd.DataFrame(results)
            # Задаём желаемый порядок колонок
            columns_order = ['file_name', 'total_revenue', 'average_order_value', 'order_count']
            df_report = df_report[columns_order]
            report_path = os.path.join(self.reports_dir, "summary_report.csv")
            df_report.to_csv(report_path, index=False)
            self.logger.info(f"Отчёт сохранён в '{report_path}'. Обработано файлов: {processed}, пропущено: {errors}.")
        else:
            self.logger.warning("Нет успешно обработанных файлов для сохранения отчёта.")

        return processed, errors

    def run(self):
        """
        Запускает полный цикл обработки: все файлы -> отчёт -> вывод итогов.
        """
        self.logger.info("=" * 40)
        self.logger.info("Начало новой обработки файлов")
        processed, errors = self.process_all()
        print(f"\nОбработка завершена.")
        print(f"Успешно обработано файлов: {processed}")
        print(f"Файлов с ошибками: {errors}")
        if processed > 0:
            print(f"Отчёт сохранён в: {os.path.join(self.reports_dir, 'summary_report.csv')}")
        else:
            print("Отчёт не был создан.")