src/analyzer.py - W2804-patch - W2804/Total_analysis

src/analyzer.py
115 строк5 KB
W2804
Create: run.py, config.py, requirements.txt, analyzer.py, analyzer.cpython-314.pyc, summary_report.csv, errors.log, order_10000.csv, order_100000.csv, order_file.csv
13 июн 2026, 21:50
Верифицирован
13 июн 2026, 21:504e8cfbe
import os
import pandas as pd
import logging
from glob import glob

import config

class OrderAnalyzer:
    def __init__(self, data_dir, reports_dir, logs_dir):
        self.data_dir = data_dir
        self.reports_dir = reports_dir
        self.logs_dir = logs_dir
        self.processed_count = 0
        self.error_count = 0
        self.results = []

        # Создаем директории для отчетов и логов
        os.makedirs(self.logs_dir, exist_ok=True)
        os.makedirs(self.reports_dir, exist_ok=True)

        # Настройка логирования
        log_file = os.path.join(self.logs_dir, config.LOG_FILENAME)
        logging.basicConfig(
            filename=log_file,
            level=logging.ERROR,
            format='%(asctime)s - %(levelname)s - %(message)s',
            encoding='utf-8'
        )

    def _load_data(self, file_path):
       # Загрузка файла и первичная проверка структуры
        df = pd.read_csv(file_path)
        if config.STATUS_COLUMN not in df.columns or config.AMOUNT_COLUMN not in df.columns:
            raise ValueError(f"Отсутствуют обязательные колонки: {config.STATUS_COLUMN} или {config.AMOUNT_COLUMN}")
        return df

    def _filter_delivered(self, df):
        # Фильтрация доставленных заказов
        filtered_df = df[df[config.STATUS_COLUMN] == config.TARGET_STATUS].copy()

        # Преобразуем колонку с суммой в числа. Ошибочные данные станут Nan
        filtered_df[config.AMOUNT_COLUMN] = pd.to_numeric(filtered_df[config.AMOUNT_COLUMN], errors='coerce')

        # Удаляем строки, где сумма стала Nan
        filtered_df = filtered_df.dropna(subset=[config.AMOUNT_COLUMN])
        return filtered_df

    def _calculate_metrics(self, df):
        
        orders_count = len(df)
        if orders_count == 0:
            return 0, 0.0, 0.0

        total_revenue = df[config.AMOUNT_COLUMN].sum()
        average_check = df[config.AMOUNT_COLUMN].mean()
        return orders_count, total_revenue, average_check

    def process_single_file(self, file_path):
        # Обработка одного файла с перехватом ошибок
        file_name = os.path.basename(file_path)
        try:
            # 1. Загрузка
            df = self._load_data(file_path)

            # 2. Фильтрация
            delivered_df = self._filter_delivered(df)

            # 3. Расчет метрик
            orders_count, total_revenue, average_check = self._calculate_metrics(delivered_df)

            # Сохранение результатов
            self.results.append({
                'filename': file_name,
                'orders_count': orders_count,
                'total_revenue': round(total_revenue, 2),
                'average_check': round(average_check, 2)
            })
            self.processed_count += 1

        except Exception as e:
            # Записываем ошибку и увеличиваем счетчик ошибок.
            self.error_count += 1
            logging.error(f"Ошибка при обработке файла {file_name}: {e}")

    def process_all_files(self):
        # Поиск и обработка всех CSV-файлов.
        search_pattern = os.path.join(self.data_dir, "*.csv")
        file_list = glob(search_pattern)

        if not file_list:
            print(f"В папке {self.data_dir} не найдено CSV-файлов.")
            return

        for file_path in file_list:
            self.process_single_file(file_path)

        self._save_report()

    def _save_report(self):
        # Сохранение результатов в общий CSV.
        if not self.results:
            print("Нет успешных результатов для сохранения отчета.")
            return

        report_df = pd.DataFrame(self.results)
        report_path = os.path.join(self.reports_dir, config.REPORT_FILENAME)
        report_df.to_csv(report_path, index=False)

    def print_summary(self):
        #Вывод сводки в консоль
        print("\n=== РЕЗУЛЬТАТЫ ОБРАБОТКИ ===")
        print(f"Файлов успешно обработано: {self.processed_count}")
        print(f"Файлов с ошибками (пропущено): {self.error_count}")
        if self.results:
            print(f"Отчет сохранен в папку: {self.reports_dir}")