src/analyzer.py - main - kingsize/order-analysis

src/analyzer.py
106 строк4 KB
dleskov
Готов к сдаче: структура папок, корректная обработка ошибок
01 авг 2026, 13:13
01 авг 2026, 13:1346750b7
import logging
from typing import Optional, Dict, Any, List
import pandas as pd
from pathlib import Path
from config import STATUS_COLUMN, DELIVERED_STATUS, REQUIRED_COLUMNS

logger = logging.getLogger(__name__)

class OrderAnalyzer:
    @staticmethod
    def _read_csv_safe(file_path: str) -> Optional[pd.DataFrame]:
        try:
            df = pd.read_csv(file_path)
            if df.empty:
                logger.warning(f"Файл пуст: {file_path}")
                return None
            return df
        except Exception as e:
            logger.error(f"Ошибка чтения файла {file_path}: {e}")
            return None

    @staticmethod
    def _validate_columns(df: pd.DataFrame, file_path: str) -> bool:
        missing_cols = REQUIRED_COLUMNS - set(df.columns)
        if missing_cols:
            logger.error(f"В файле {file_path} отсутствуют колонки: {missing_cols}. Файл полностью пропущен.")
            return False
        return True

    @classmethod
    def _check_numeric_column(cls, df: pd.DataFrame, column: str, file_path: str) -> bool:
        if column not in df.columns:
            logger.error(f"Колонка '{column}' отсутствует в файле {file_path}. Файл полностью пропущен.")
            return False

        converted = pd.to_numeric(df[column], errors="coerce")
        bad_mask = converted.isna() & df[column].notna()

        if bad_mask.any():
            bad_indices = bad_mask[bad_mask].index.tolist()
            total_bad = len(bad_indices)
            preview_count = 10
            preview = bad_indices[:preview_count]

            if total_bad <= preview_count:
                logger.error(
                    f"Файл {file_path} содержит некорректные данные в колонке '{column}'. "
                    f"Проблемные строки: {preview}. Файл полностью пропущен."
                )
            else:
                logger.error(
                    f"Файл {file_path} содержит некорректные данные в колонке '{column}'. "
                    f"Найдено {total_bad} проблемных строк. Для примера: строки {preview}... "
                    "Файл полностью пропущен."
                )
            return False
        return True

    @classmethod
    def _calculate_metrics(cls, df: pd.DataFrame) -> Dict[str, Any]:
        delivered = df[df[STATUS_COLUMN] == DELIVERED_STATUS].copy()
        delivered["total_amount"] = pd.to_numeric(delivered["total_amount"], errors="raise")

        total_revenue = delivered["total_amount"].sum() if not delivered.empty else 0.0
        avg_check = delivered["total_amount"].mean() if not delivered.empty else 0.0
        count_orders = len(delivered)

        return {
            "total_revenue": total_revenue,
            "avg_check": avg_check,
            "count_orders": count_orders,
        }

    def analyze_file(self, file_path: str) -> Optional[Dict[str, Any]]:
        df = self._read_csv_safe(file_path)
        if df is None:
            return None
        if not self._validate_columns(df, file_path):
            return None
        if not self._check_numeric_column(df, "total_amount", file_path):
            return None
        return self._calculate_metrics(df)

    def analyze_all_files(self, data_dir: str, output_path: Optional[str] = None) -> List[Dict[str, Any]]:
        data_path = Path(data_dir)
        csv_files = list(data_path.glob("*.csv"))

        results = []
        for file_path in csv_files:
            metrics = self.analyze_file(str(file_path))
            if metrics is not None:
                results.append({
                    "file_name": file_path.name,
                    "total_revenue": metrics["total_revenue"],
                    "avg_check": metrics["avg_check"],
                    "count_orders": metrics["count_orders"],
                })

        if output_path is not None:
            output_path = Path(output_path)
            output_path.parent.mkdir(parents=True, exist_ok=True)
            df_report = pd.DataFrame(results)
            df_report.to_csv(output_path, index=False)
            logger.info(f"Итоговый отчёт сохранён: {output_path}")

        return results