src/analyzer.py - master - melvia/project_lection_08062026

src/analyzer.py
363 строки14 KB
oleg.gezhin
Create: errors_20260608_165450.log, summary_report_20260608_165554.csv, analyzer.py, analyzer.cpython-313.pyc, config.py, main.py
08 июн 2026, 17:40
Верифицирован
08 июн 2026, 17:40bc391ed
import pandas as pd
import config
import logging
from datetime import datetime
from pathlib import Path
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass, asdict


# Конфигурация по умолчанию
DEFAULT_CONFIG = {
    'input_folder': 'data',
    'output_folder': 'reports',
    'logs_folder': 'logs',
    'status_column': 'status',
    'target_status': 'Delivered',
    'amount_column': 'total_amount',
    'encoding': 'utf-8'
}

@dataclass
class FileProcessingResult:
    """Результат обработки одного файла"""
    file_name: str
    total_orders: int = 0
    delivered_orders: int = 0
    total_revenue: float = 0.0
    average_bill: float = 0.0
    success: bool = False
    error_message: str = ""

class CSVFileProcessor:
    """Класс для загрузки и валидации CSV файлов"""

    def __init__(self, encoding: str = config.DEFAULT_ENCODING):
        self.encoding = encoding

    def load_file(self, file_path: Path) -> Optional[pd.DataFrame]:
        """Безопасная загрузка CSV файла"""
        try:
            return pd.read_csv(file_path, encoding=self.encoding)
        except UnicodeDecodeError:
            # Пробуем другую кодировку
            try:
                return pd.read_csv(file_path, encoding='utf-8')
            except Exception as e:
                logging.error(f"Не удалось загрузить {file_path.name}: {e}")
                return None
        except Exception as e:
            logging.error(f"Ошибка загрузки {file_path.name}: {e}")
            return None
        
    def validate_columns(self, df: pd.DataFrame, required_cols: List[str]) -> Tuple[bool, List[str]]:
        """Проверка наличия необходимых колонок"""
        missing = [col for col in required_cols if col not in df.columns]
        return len(missing) == 0, missing
    
class OrderDataAnalyzer:
    """Класс для анализа данных заказов"""

    def __init__(self, status_col: str = 'status', amount_col: str = 'total_amount'):
        self.status_col = status_col
        self.amount_col = amount_col

    def analyze_delivered_orders(self, df: pd.DataFrame, target_status: str = 'Delivered') -> Dict:
        """Анализирует доставленные заказы"""
        delivered_df = df[df[self.status_col] == target_status]

        return {
            'total_orders': len(df),
            'delivered_orders': len(delivered_df),
            'total_revenue': delivered_df[self.amount_col].sum() if len(delivered_df) > 0 else 0,
            'average_bill': delivered_df[self.amount_col].mean() if len(delivered_df) > 0 else 0
        }
    
class BatchOrderProcessor:
    """Класс для пакетной обработки файлов заказов"""

    def __init__(self, config: Dict = None):
        self.config = {**DEFAULT_CONFIG, **(config or {})}
        self.loader = CSVFileProcessor(self.config['encoding'])
        self.analyzer = OrderDataAnalyzer(
            self.config['status_column'],
            self.config['amount_column']
        )
        self.results: List[FileProcessingResult] = []
        self.processed = False  # Флаг, чтобы предотвратить двойную обработку

        # Настройка логирования
        self._setup_logging()

    def _setup_logging(self):
        """Настройка системы логирования"""
        log_dir = Path(self.config['logs_folder'])
        log_dir.mkdir(exist_ok=True)

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        log_file = log_dir / f"errors_{timestamp}.log"

        logging.basicConfig(
            level=logging.ERROR,
            format='%(asctime)s - %(levelname)s - %(message)s',
            handlers=[
                logging.FileHandler(log_file, encoding='utf-8'),
                logging.StreamHandler()
            ]
        )

        self.log_file = log_file

    def find_csv_files(self) -> List[Path]:
        """Находит все CSV файлы в папке input_folder"""
        input_path = Path(self.config['input_folder'])

        if not input_path.exists():
            logging.warning(f"Папка '{input_path}' не существует. Создаю...")
            input_path.mkdir(parents=True, exist_ok=True)
            return []

        # Используем только один паттерн (на Windows это не важно)
        files = list(input_path.glob("*.csv"))

        # Если нет .csv файлов, пробуем .CSV
        if not files:
            files = list(input_path.glob("*.CSV"))

        # Убираем дубликаты по абсолютному пути
        unique_files = []
        seen = set()
        for f in files:
            if f.resolve() not in seen:
                seen.add(f.resolve())
                unique_files.append(f)

        return unique_files

    def process_single_file(self, file_path: Path) -> FileProcessingResult:
        """Обрабатывает один файл"""
        result = FileProcessingResult(file_name=file_path.name)

        # Загрузка файла
        df = self.loader.load_file(file_path)
        if df is None:
            result.error_message = "Не удалось загрузить файл"
            return result

        # Проверка колонок
        required_cols = [self.config['status_column'], self.config['amount_column']]
        is_valid, missing = self.loader.validate_columns(df, required_cols)

        if not is_valid:
            result.error_message = f"Отсутствуют колонки: {missing}"
            return result

        # Анализ данных
        try:
            analysis = self.analyzer.analyze_delivered_orders(df, self.config['target_status'])
            result.total_orders = analysis['total_orders']
            result.delivered_orders = analysis['delivered_orders']
            result.total_revenue = analysis['total_revenue']
            result.average_bill = analysis['average_bill']
            result.success = True
        except Exception as e:
            result.error_message = f"Ошибка анализа: {str(e)}"

        return result

    def ask_processing_mode(self) -> Tuple[str, Optional[Path]]:
        """Запрашивает у пользователя режим обработки"""
        print("\n" + "=" * 60)
        print("ВЫБОР РЕЖИМА ОБРАБОТКИ")
        print("=" * 60)
        print("1. Обработать один конкретный файл")
        print("2. Пакетная обработка всех CSV файлов в папке")
        print("3. Выйти")
        print("-" * 60)

        while True:
            choice = input("Ваш выбор (1/2/3): ").strip()

            if choice == '1':
                # Режим обработки одного файла
                file_name = input("Введите имя файла (например, orders_10k.csv): ").strip()
                file_path = Path(self.config['input_folder']) / file_name

                if not file_path.exists():
                    print(f"❌ Файл '{file_name}' не найден в папке '{self.config['input_folder']}'")
                    return self.ask_processing_mode()  # Повторный запрос

                print(f"\n✅ Выбран файл: {file_name}")
                return 'single', file_path

            elif choice == '2':
                # Режим пакетной обработки
                print("\n✅ Выбрана пакетная обработка всех CSV файлов")
                return 'batch', None

            elif choice == '3':
                print("\n👋 До свидания!")
                exit(0)
            else:
                print("❌ Неверный выбор. Пожалуйста, введите 1, 2 или 3.")

    def process_single_file_mode(self, file_path: Path) -> Tuple[List[FileProcessingResult], Dict]:
        """Обрабатывает только один указанный файл"""
        print(f"\n📁 Обработка файла: {file_path.name}")
        print("-" * 50)

        stats = {'total': 1, 'success': 0, 'errors': 0}

        result = self.process_single_file(file_path)
        self.results.append(result)

        if result.success:
            stats['success'] += 1
            print(f"✅ {result.file_name}: {result.delivered_orders:,} доставленных заказов, "
                  f"выручка: ${result.total_revenue:,.2f}")
        else:
            stats['errors'] += 1
            logging.error(f"❌ {result.file_name}: {result.error_message}")
            print(f"❌ {result.file_name}: ОШИБКА - {result.error_message[:50]}")

        return self.results, stats

    def process_all_files_mode(self) -> Tuple[List[FileProcessingResult], Dict]:
        """Обрабатывает все CSV файлы в папке"""
        files = self.find_csv_files()

        if not files:
            logging.warning("CSV файлы не найдены")
            return [], {'total': 0, 'success': 0, 'errors': 0}

        print(f"\n📁 Найдено файлов: {len(files)}")
        print("-" * 50)

        stats = {'total': len(files), 'success': 0, 'errors': 0}

        for file_path in files:
            result = self.process_single_file(file_path)
            self.results.append(result)

            if result.success:
                stats['success'] += 1
                print(f"✅ {result.file_name}: {result.delivered_orders:,} доставленных заказов, "
                      f"выручка: ${result.total_revenue:,.2f}")
            else:
                stats['errors'] += 1
                logging.error(f"❌ {result.file_name}: {result.error_message}")
                print(f"❌ {result.file_name}: ОШИБКА - {result.error_message[:50]}")

        return self.results, stats

    def save_results(self) -> Path:
        """Сохраняет результаты обработки в CSV файл"""
        if not self.results:
            logging.warning("Нет результатов для сохранения")
            return None

        # Конвертируем результаты в DataFrame
        data = []
        for r in self.results:
            if r.success:
                data.append(asdict(r))

        if not data:
            logging.warning("Нет успешных результатов для сохранения")
            return None

        df_result = pd.DataFrame(data)

        # Выбираем колонки для вывода
        output_columns = ['file_name', 'total_orders', 'delivered_orders',
                          'total_revenue', 'average_bill']
        df_result = df_result[output_columns]

        # Сохраняем результат
        output_dir = Path(self.config['output_folder'])
        output_dir.mkdir(exist_ok=True)

        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        output_path = output_dir / f"summary_report_{timestamp}.csv"

        df_result.to_csv(output_path, index=False, encoding=self.config['encoding'])

        print(f"\n💾 Результаты сохранены: {output_path}")

        return output_path

    def print_statistics(self, stats: Dict):
        """Выводит статистику обработки в консоль"""
        print("\n" + "=" * 50)
        print("📊 ИТОГОВАЯ СТАТИСТИКА")
        print("=" * 50)
        print(f"✅ Обработано файлов: {stats['success']}")
        print(f"❌ Файлов с ошибками: {stats['errors']}")
        print(f"📁 Всего файлов: {stats['total']}")

        if stats['errors'] > 0:
            print(f"\n⚠️ Файлы с ошибками:")
            for result in self.results:
                if not result.success:
                    print(f"  • {result.file_name}: {result.error_message}")

            print(f"\n📝 Детали ошибок сохранены в: {self.log_file}")

        print("=" * 50)

    def run(self) -> Tuple[Optional[Path], Dict]:
        """Запускает полный цикл обработки с выбором режима"""
        print("\n" + "=" * 70)
        print("🚀 ЗАПУСК ПАКЕТНОЙ ОБРАБОТКИ ЗАКАЗОВ")
        print("=" * 70)

        # Запрашиваем режим обработки
        mode, file_path = self.ask_processing_mode()

        # Обработка в выбранном режиме
        if mode == 'single':
            results, stats = self.process_single_file_mode(file_path)
        else:  # batch mode
            results, stats = self.process_all_files_mode()

        if stats['success'] > 0:
            # Сохранение результатов
            output_path = self.save_results()
        else:
            output_path = None

        # Вывод статистики
        self.print_statistics(stats)

        return output_path, stats


# Упрощенная функция для быстрого запуска
def process_files_with_stats(input_folder: str = 'data', ask_mode: bool = True):
    """Обрабатывает CSV файлы и выводит статистику в консоль

    Args:
        input_folder: Папка с CSV файлами
        ask_mode: Запрашивать ли режим обработки (True/False)
    """
    config = {
        'input_folder': input_folder,
        'output_folder': 'reports',
        'logs_folder': 'logs'
    }

    processor = BatchOrderProcessor(config)

    if ask_mode:
        # С запросом режима
        output_path, stats = processor.run()
    else:
        # Без запроса - пакетная обработка
        results, stats = processor.process_all_files_mode()
        if stats['success'] > 0:
            output_path = processor.save_results()
        else:
            output_path = None
        processor.print_statistics(stats)

    return output_path, stats