src/analyzer.py - master - user1234567/dz

src/analyzer.py
118 строк4 KB
user1234567
Create: analyzer.py
06 июл 2026, 22:44
06 июл 2026, 22:44a0e83b5
import logging
from pathlib import Path

import pandas as pd


class OrderAnalyzer:
    def __init__(
        self,
        data_dir,
        output_file,
        log_file,
        file_pattern,
        status_column,
        delivered_status,
        amount_column,
    ):
        self.data_dir = Path(data_dir)
        self.output_file = Path(output_file)
        self.log_file = Path(log_file)
        self.file_pattern = file_pattern

        self.status_column = status_column
        self.delivered_status = delivered_status
        self.amount_column = amount_column

        self.results = []
        self.processed_files = 0
        self.error_files = 0

        self.output_file.parent.mkdir(parents=True, exist_ok=True)
        self.log_file.parent.mkdir(parents=True, exist_ok=True)

        logging.basicConfig(
            filename=self.log_file,
            level=logging.ERROR,
            format="%(asctime)s | %(levelname)s | %(message)s",
            encoding="utf-8",
        )

    def load_file(self, file_path):
        """Загружает CSV-файл."""
        return pd.read_csv(file_path)

    def validate_data(self, df, file_path):
        """Проверяет, что в файле есть нужные колонки и корректные данные."""
        required_columns = [
            self.status_column,
            self.amount_column,
        ]

        for column in required_columns:
            if column not in df.columns:
                raise ValueError(f"В файле {file_path.name} нет колонки {column}")

        df[self.amount_column] = pd.to_numeric(df[self.amount_column], errors="raise")

        return df

    def filter_delivered_orders(self, df):
        """Оставляет только доставленные заказы."""
        return df[df[self.status_column] == self.delivered_status]

    def calculate_metrics(self, df):
        """Считает общую выручку, средний чек и количество заказов."""
        orders_count = len(df)
        total_revenue = df[self.amount_column].sum()

        if orders_count == 0:
            average_order_value = 0
        else:
            average_order_value = df[self.amount_column].mean()

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

    def process_file(self, file_path):
        """Обрабатывает один CSV-файл."""
        try:
            df = self.load_file(file_path)
            df = self.validate_data(df, file_path)
            delivered_orders = self.filter_delivered_orders(df)
            metrics = self.calculate_metrics(delivered_orders)

            result = {
                "file_name": file_path.name,
                **metrics,
            }

            self.results.append(result)
            self.processed_files += 1

        except Exception as error:
            self.error_files += 1
            logging.error(f"Файл {file_path.name} пропущен. Ошибка: {error}")

    def process_all_files(self):
        """Обрабатывает все CSV-файлы в папке data."""
        files = list(self.data_dir.glob(self.file_pattern))

        for file_path in files:
            self.process_file(file_path)

        self.save_report()

        return {
            "processed_files": self.processed_files,
            "error_files": self.error_files,
        }

    def save_report(self):
        """Сохраняет общий отчёт по всем обработанным файлам."""
        report_df = pd.DataFrame(self.results)

        report_df.to_csv(self.output_file, index=False, encoding="utf-8")