order_analysis/src/analyzer.py - master - Shcod/netology_python_automation

order_analysis/src/analyzer.py
168 строк7 KB
Shcod
change logic for load file
22 июл 2026, 22:45
22 июл 2026, 22:45d45df51
import logging
from pathlib import Path
import pandas as pd

import config


class OrderAnalyzer:
    def __init__(self):
        self.logger = self._setup_logger()
        self.results = []
        self.processed_files = 0
        self.error_files = 0
        self.file_errors = []

    def _setup_logger(self):
        logger = logging.getLogger("OrderAnalyzer")
        logger.setLevel(logging.ERROR)
        log_file = config.LOGS_DIR / config.ERROR_LOG_FILENAME
        file_handler = logging.FileHandler(log_file, encoding="utf-8")
        file_handler.setLevel(logging.ERROR)
        format = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s", datefmt="%d - %m - %Y %H:%M:%S")
        file_handler.setFormatter(format)
        logger.addHandler(file_handler)
        return logger

    def load_file(self, file_path: Path):
        try:
            if not file_path.exists():
                error_msg = f"Файл не найден: {file_path.name}"
                self.logger.error(error_msg)
                self.file_errors.append((file_path.name, error_msg))
                return None

            df = pd.read_csv(file_path)

            if df.empty:
                error_msg = f"Файл не содержит данных: {file_path.name}"
                self.logger.error(error_msg)
                self.file_errors.append((file_path.name, error_msg))
                return None
            required_col = [config.STATUS_COL, config.AMOUNT_COL]
            missing_col = []
            for col in required_col:
                if col not in df.columns:
                    missing_col.append(col)
            if missing_col:
                error_msg = f"В файле {file_path.name} нет необходимых колонок: {missing_col}"
                self.logger.error(error_msg)
                self.file_errors.append((file_path.name, error_msg))
                return None
            amount_series = pd.to_numeric(df[config.AMOUNT_COL], errors='coerce')
            if amount_series.isna().any():
                invalid_rows = df[amount_series.isna()].index.tolist()
                error_msg = (f"Файл поврежден: {file_path.name} содержит нечисловые значения в колонке {config.AMOUNT_COL} "
                             f"в строках: {invalid_rows}. Файл полностью пропускается.")
                self.logger.error(error_msg)
                self.file_errors.append((file_path.name, error_msg))
                return None
            return df

        except Exception as e:
            error_msg = f"Ошибка при загрузке файла {file_path.name}: {str(e)}"
            self.logger.error(error_msg)
            self.file_errors.append((file_path.name, error_msg))
            return None

    def filter_delivered_orders(self, df):
        try:
            if config.STATUS_COL not in df.columns:
                self.logger.error(f"Колонка '{config.STATUS_COL}' не найдена. Доступные колонки: {list(df.columns)}")
                return pd.DataFrame()
            mask = df[config.STATUS_COL] == config.TARGET_COL

            if not mask.any():
                self.logger.warning(f"Заказы со статусом '{config.TARGET_COL}' не найдены")
                return pd.DataFrame()
            delivered = df[mask].copy()
            return delivered
        except Exception as e:
            self.logger.error(f"Ошибка при фильтрации данных: {str(e)}")
            return pd.DataFrame()

    def calculate_metrics(self, df):
        try:
            if df.empty:
                return {
                    "total_revenue": 0.0,
                    "average_order_value": 0.0,
                    "order_count": 0
                }
            total_revenue = df[config.AMOUNT_COL].sum()
            order_count = len(df)
            average_order_value = total_revenue / order_count if order_count > 0 else 0.0

            return {
                "total_revenue": round(total_revenue, 2),
                "average_order_value": round(average_order_value, 2),
                "order_count": order_count
            }
        except Exception as e:
            self.logger.error(f"Ошибка при расчете метрик: {str(e)}")
            return {
                "total_revenue": 0.0,
                "average_order_value": 0.0,
                "order_count": 0
            }
    def process_file(self, file_path):
        try:
            df = self.load_file(file_path)
            if df is None:
                self.error_files += 1
                return None
            delivered_orders = self.filter_delivered_orders(df)
            metrics = self.calculate_metrics(delivered_orders)
            result = {
                "file_name": file_path.name,
                "total_revenue": metrics["total_revenue"],
                "average_order_value": metrics["average_order_value"],
                "order_count": metrics["order_count"]
            }
            self.processed_files += 1
            return result
        except Exception as e:
            error_msg = f"Критическая ошибка при обработке файла {file_path.name}: {str(e)}"
            self.logger.error(error_msg)
            self.file_errors.append((file_path.name, error_msg))
            self.error_files += 1
            return None

    def save_results(self, results_df):
        try:
            if results_df.empty:
                self.logger.warning("Нет данных для сохранения")
                return
            output_file = config.REPORTS_DIR / config.OUTPUT_FILENAME
            results_df.to_csv(output_file, index=False, encoding='utf-8')

        except Exception as e:
            self.logger.error(f"Ошибка при сохранении результатов: {str(e)}")

    def process_all_files(self, data_dir):
        self.processed_files = 0
        self.error_files = 0
        self.results = []

        csv_files = list(data_dir.glob("*.csv"))

        if not csv_files:
            self.logger.error(f"CSV файлы не найдены в директории {data_dir}")
            return pd.DataFrame()

        for file_path in csv_files:
            result = self.process_file(file_path)
            if result:
                self.results.append(result)

        results_df = pd.DataFrame(self.results)
        self.save_results(results_df)
        return results_df

    def get_summary(self):
        summary = f"Обработано файлов: {self.processed_files}\nФайлов с ошибкой: {self.error_files}"
        if self.file_errors:
            summary += "\n\nФайлы с ошибками:"
            for file_name, error in self.file_errors:
                summary += f"\n- {file_name}: {error}"
        return summary