src/analyzer.py - master - AnnaSav/order_analysis

src/analyzer.py
122 строки4 KB
AnnaSav
Добавлен скрипт
10 авг 2026, 15:08
10 авг 2026, 15:087fc5065
import logging
from pathlib import Path
import pandas as pd

class OrderAnalyzer:
 # задаем параметры класса
    def __init__(
        self,
        data_dir,
        reports_dir,
        logs_dir,
        output_file,
        error_log_name,
        status_column,
        filter_status,
    ):
        self._data_dir = Path(data_dir)
        self._reports_dir = Path(reports_dir)
        self._logs_dir = Path(logs_dir)
        self._report_path = self._reports_dir / output_file

        self._status_column = status_column
        self._filter_status = filter_status

        self._prepare_folders()
        self._logger = self._setup_logging(error_log_name)
    
    # создаем папки если они отсутствуют
    def _prepare_folders(self):
        self._reports_dir.mkdir(parents=True, exist_ok=True)
        self._logs_dir.mkdir(parents=True, exist_ok=True)

    # параметры логирования
    def _setup_logging(self, error_log_name):
        file_handler = logging.FileHandler(self._logs_dir / error_log_name, encoding="utf-8")
        file_handler.setLevel(logging.ERROR)

        logging.basicConfig(
            level=logging.INFO,
            format="%(asctime)s - %(levelname)s - %(message)s",
            handlers=[file_handler, logging.StreamHandler()],
            force=True,
        )
        return logging.getLogger("order_analyzer")

    # проверям данные в файле (наличие данных, нужные колонки, формат данных)
    def load_file(self, file_path):
        df = pd.read_csv(file_path, encoding="utf-8")

        if df.empty:
            raise ValueError("Файл пустой")

        if self._status_column not in df.columns:
            raise ValueError(f"Нет колонки '{self._status_column}'")

        if "total_amount" not in df.columns:
            raise ValueError("Нет колонки 'total_amount'")

        df["total_amount"] = pd.to_numeric(df["total_amount"], errors="coerce")
        not_number = df["total_amount"].isna().sum()
        if not_number > 0:
            raise ValueError(
                f"В колонке 'total_amount' есть значения не в виде числа: {not_number} шт"
            )

        return df

    # фильтр на статус заказа
    def filter_for_status(self, df):
        return df[df[self._status_column] == self._filter_status]

    # расчет метрик (количество заказов, выручка, средний чек)
    def calculate_metrics(self, df):
        orders_count = len(df)
        total_revenue = float(df["total_amount"].sum())
        average_check = float(df["total_amount"].mean()) if orders_count > 0 else 0.0

        return {
            "total_revenue": round(total_revenue, 2),
            "average_check": round(average_check, 2),
            "orders_count": orders_count,
        }

    # применяем к одному файлу чтение, фильтрацияю и расчет метрик
    def process_file(self, file_path):
        df = self.load_file(file_path)
        delivered_df = self.filter_for_status(df)
        metrics = self.calculate_metrics(delivered_df)

        result = {"file_name": Path(file_path).name}
        result.update(metrics)
        return result

    # собираем все файлы
    def process_all_files(self):
        csv_files = list(self._data_dir.glob("*.csv"))
        self._logger.info(f"Найдено файлов: {len(csv_files)}")

        results = []
        success = 0
        errors = 0

        for file in csv_files:
            try:
                result = self.process_file(file)
                results.append(result)
                success += 1
                self._logger.info(f"{file.name} — успешно")
            except Exception as e:
                errors += 1
                self._logger.error(f"{file.name}: {e}")

        if results:
            report_df = pd.DataFrame(results)
            columns_order = ["file_name", "total_revenue", "average_check", "orders_count"]
            report_df = report_df[columns_order]
            report_df.to_csv(self._report_path, index=False, encoding="utf-8")
            self._logger.info(f"Отчет сохранен: {self._report_path}")

        self._logger.info(f"Завершено. Успешно: {success}, Ошибок: {errors}")

        return {"total": len(csv_files), "success": success, "errors": errors}