src/analyzer.py - master - xcenia/order_analysis

src/analyzer.py
137 строк5 KB
xcenia
14 июл 2026, 18:42
14 июл 2026, 18:421157727
import os
import pandas as pd
import logging
import config


class OrderAnalyzer:
    def __init__(self):
        self.status_column = config.STATUS_COLUMN
        self.filter_status = config.FILTER_STATUS
        self.results = []
        self.processed_files = 0
        self.error_files = 0
        self._setup_logging()

    def _setup_logging(self):
        log_file = os.path.join(config.LOGS_DIR, config.LOG_FILE)
        logging.basicConfig(
            level=logging.ERROR,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[logging.FileHandler(log_file, encoding=config.ENCODING)]
        )
        self.logger = logging.getLogger(__name__)

    def _log_error(self, file_name, error_message):
        self.logger.error(f"Файл: {file_name} - {error_message}")
        self.error_files += 1

    def load_file(self, file_path):
        try:
            if not os.path.exists(file_path):
                raise FileNotFoundError(f"Файл не найден: {file_path}")
            if os.path.getsize(file_path) == 0:
                raise ValueError("Файл пустой")
            df = pd.read_csv(file_path, encoding=config.ENCODING, sep=config.SEPARATOR)
            return df
        except Exception as e:
            self._log_error(os.path.basename(file_path), f"Ошибка загрузки: {e}")
            return None

    def filter_delivered_orders(self, df, file_name):
        try:
            if self.status_column not in df.columns:
                raise KeyError(f"Колонка '{self.status_column}' не найдена")

            if 'total_amount' not in df.columns:
                raise KeyError("Колонка 'total_amount' не найдена")

            if not pd.api.types.is_numeric_dtype(df['total_amount']):
                try:
                    df['total_amount'] = pd.to_numeric(df['total_amount'], errors='raise')
                except ValueError:
                    self._log_error(file_name, "Неверный тип данных в колонке total_amount")
                    raise ValueError("Неверный тип данных")

            filtered_df = df[df[self.status_column] == self.filter_status].copy()

            return filtered_df

        except Exception as e:
            self._log_error(file_name, f"Ошибка фильтрации: {e}")
            return None

    def calculate_metrics(self, df, file_name):
        if df is None or df.empty:
            return {
                'file_name': file_name,
                'total_revenue': 0,
                'average_order_value': 0,
                'order_count': 0
            }

        return {
            'file_name': file_name,
            'total_revenue': round(df['total_amount'].sum(), 2),
            'average_order_value': round(df['total_amount'].mean(), 2),
            'order_count': len(df)
        }

    def process_file(self, file_path):
        file_name = os.path.basename(file_path)
        print(f"Обработка файла: {file_name}")

        df = self.load_file(file_path)
        if df is None:
            return None


        filtered_df = self.filter_delivered_orders(df, file_name)
        if filtered_df is None:
            return None

        metrics = self.calculate_metrics(filtered_df, file_name)

        self.processed_files += 1
        print(f"  Обработано: {metrics['order_count']} заказов, выручка: ${metrics['total_revenue']:,.2f}")

        return metrics

    def process_all_files(self, data_dir=None):
        if data_dir is None:
            data_dir = config.DATA_DIR

        csv_files = [f for f in os.listdir(data_dir) if f.endswith('.csv')]

        print(f"Найдено {len(csv_files)} CSV-файлов")

        for file_name in csv_files:
            file_path = os.path.join(data_dir, file_name)
            result = self.process_file(file_path)
            if result is not None:
                self.results.append(result)

        print(f"Обработано файлов: {self.processed_files}")
        print(f"С ошибками: {self.error_files}")

        return self.results

    def save_results(self, output_file=None):
        if output_file is None:
            output_file = config.OUTPUT_FILE

        if not self.results:
            print("Нет данных для сохранения")
            return False

        results_df = pd.DataFrame(self.results)
        output_path = os.path.join(config.REPORTS_DIR, output_file)
        results_df.to_csv(output_path, index=False, encoding=config.ENCODING)
        print(f"Отчет сохранен в {output_path}")

    def get_summary(self):
        return {
            'processed_files': self.processed_files,
            'error_files': self.error_files,
            'total_results': len(self.results)
        }