core/excel_analyzer.py - main - AntiManager/table_structure_analyzer

core/excel_analyzer.py
463 строки19 KB
Eugene
fix gui bugs
28 дек 2025, 19:13
28 дек 2025, 19:1397257de
"""
Анализатор Excel-файлов
Оптимизировано для работы с очень большими файлами.
"""
import pandas as pd
import numpy as np
from typing import Dict, List, Any, Optional, Tuple
from pathlib import Path
from datetime import datetime
import logging
import re
import warnings
from openpyxl import load_workbook

from .base_analyzer import BaseAnalyzer
from .config import (
    TYPE_ANALYSIS_ROWS,
    PREVIEW_ROWS,
    MAX_FILE_SIZE_MB,
)

# Отключаем предупреждения pandas
warnings.filterwarnings('ignore')


class ExcelAnalyzer(BaseAnalyzer):
    SUPPORTED_EXTENSIONS = [".xlsx", ".xls", ".xlsm"]
    
    DATE_PATTERNS = [
        r'\d{2}\.\d{2}\.\d{4} \d{1,2}:\d{2}:\d{2}',
        r'\d{4}-\d{2}-\d{2} \d{1,2}:\d{2}:\d{2}',
        r'\d{2}/\d{2}/\d{4} \d{1,2}:\d{2}:\d{2}',
        r'\d{2}\.\d{2}\.\d{4}',
        r'\d{4}-\d{2}-\d{2}',
        r'\d{2}/\d{2}/\d{4}',
    ]

    def __init__(self):
        pass

    def get_sheet_info(self, file_path: str) -> Dict[str, List[str]]:
        """Быстрое получение информации о листах и их колонках"""
        try:
            # Сначала пробуем openpyxl для быстрого чтения
            wb = load_workbook(filename=file_path, read_only=True, data_only=True)
            sheet_info = {}
            
            for sheet_name in wb.sheetnames:
                try:
                    ws = wb[sheet_name]
                    headers = []
                    
                    # Получаем первую строку (заголовки)
                    max_column = ws.max_column
                    if max_column > 0:
                        for col in range(1, min(max_column, 100) + 1):  # Ограничиваем 100 колонками
                            cell = ws.cell(row=1, column=col)
                            if cell.value:
                                headers.append(str(cell.value).strip())
                            else:
                                headers.append(f"Column_{col}")
                    else:
                        # Если нет колонок, создаем пустой список
                        headers = []
                    
                    sheet_info[sheet_name] = headers
                    
                except Exception as e:
                    logging.warning(f"Не удалось прочитать лист {sheet_name}: {e}")
                    sheet_info[sheet_name] = []
            
            wb.close()
            return sheet_info
            
        except Exception as e:
            logging.error(f"Ошибка openpyxl при чтении {file_path}: {e}")
            # Fallback: используем pandas
            try:
                excel_file = pd.ExcelFile(file_path)
                sheet_info = {}
                
                for sheet_name in excel_file.sheet_names:
                    try:
                        # Читаем только заголовки
                        df = pd.read_excel(file_path, sheet_name=sheet_name, nrows=0)
                        sheet_info[sheet_name] = df.columns.tolist()
                    except Exception as e2:
                        logging.warning(f"Не удалось прочитать колонки для листа {sheet_name}: {e2}")
                        sheet_info[sheet_name] = []
                
                return sheet_info
                
            except Exception as e2:
                logging.error(f"Ошибка pandas при чтении {file_path}: {e2}")
                return {}

    def get_sheet_data(
        self, file_path: str, sheet_name: str, max_rows: int = PREVIEW_ROWS
    ) -> List[Dict[str, Any]]:
        """Получение данных для предпросмотра"""
        try:
            df = pd.read_excel(file_path, sheet_name=sheet_name, nrows=max_rows)
            return self._convert_to_serializable(df.fillna("").to_dict("records"))
        except Exception as e:
            logging.error(f"get_sheet_data({sheet_name}) -> {e}")
            return []

    def fast_row_count(self, file_path: str, sheet_name: str) -> int:
        """Быстрая оценка количества строк"""
        try:
            file_size_mb = Path(file_path).stat().st_size / 1024 / 1024
            
            if file_size_mb > 10:
                wb = load_workbook(filename=file_path, read_only=True)
                ws = wb[sheet_name]
                row_count = ws.max_row
                wb.close()
                
                try:
                    df = pd.read_excel(file_path, sheet_name=sheet_name, 
                                     skiprows=max(0, row_count - 100))
                    actual_rows = len(df) + max(0, row_count - 100)
                    return int(actual_rows)
                except:
                    return int(row_count)
            else:
                df = pd.read_excel(file_path, sheet_name=sheet_name)
                return int(len(df))
                
        except Exception as e:
            logging.warning(f"fast_row_count для {sheet_name}: {e}")
            try:
                df = pd.read_excel(file_path, sheet_name=sheet_name, nrows=1000)
                return int(len(df))
            except:
                return 0

    def get_sheet_dimensions(self, file_path: str, sheet_name: str) -> Tuple[int, int]:
        """Быстрое получение размеров листа"""
        try:
            wb = load_workbook(filename=file_path, read_only=True)
            ws = wb[sheet_name]
            rows = int(ws.max_row)
            cols = int(ws.max_column)
            wb.close()
            return rows, cols
        except Exception as e:
            logging.warning(f"get_sheet_dimensions для {sheet_name}: {e}")
            try:
                df = pd.read_excel(file_path, sheet_name=sheet_name, nrows=0)
                return 0, int(len(df.columns))
            except:
                return 0, 0

    def analyze_structure(
        self,
        file_path: str,
        sheet_names: Optional[List[str]] = None,
        deep: bool = False,
    ) -> Dict[str, Any]:
        """
        Оптимизированный анализ структуры
        """
        file_path_obj = Path(file_path)
        
        result = {
            "file_info": {
                "file_name": file_path_obj.name,
                "file_path": str(file_path_obj.absolute()),
                "file_size": int(file_path_obj.stat().st_size),
                "modified_time": datetime.fromtimestamp(
                    file_path_obj.stat().st_mtime
                ).isoformat(),
                "analyzed_at": datetime.now().isoformat(),
                "analysis_type": "full" if deep else "quick",
            },
            "sheets": {},
        }

        sheet_info = self.get_sheet_info(file_path)
        if not sheet_info:
            return result

        want_sheets = sheet_names or list(sheet_info.keys())
        
        for sh in want_sheets:
            if sh not in sheet_info:
                continue
                
            try:
                total_rows, total_cols = self.get_sheet_dimensions(file_path, sh)
                
                analysis_rows = TYPE_ANALYSIS_ROWS
                
                if deep:
                    file_size_mb = file_path_obj.stat().st_size / 1024 / 1024
                    if file_size_mb > MAX_FILE_SIZE_MB:
                        analysis_rows = min(TYPE_ANALYSIS_ROWS * 10, 10000)
                    else:
                        analysis_rows = None
                
                if analysis_rows:
                    df_analysis = pd.read_excel(
                        file_path, sheet_name=sh, nrows=analysis_rows
                    )
                else:
                    df_analysis = pd.read_excel(file_path, sheet_name=sh)
                
                df_preview = df_analysis.head(PREVIEW_ROWS) if len(df_analysis) > 0 else pd.DataFrame()
                
                sheet_result = {
                    "name": sh,
                    "dimensions": {
                        "total_rows": int(total_rows),
                        "total_columns": int(total_cols),
                        "analyzed_rows": int(len(df_analysis)),
                        "total_cells": int(total_rows * total_cols),
                    },
                    "columns": self._infer_column_types(df_analysis),
                    "data_preview": self._convert_to_serializable(
                        df_preview.fillna("").to_dict("records")
                    ),
                    "metadata": {
                        "has_headers": True,
                        "empty_rows_percent": float(self._calculate_empty_percentage(df_analysis)),
                        "estimated_total_rows": int(total_rows),
                        "analysis_note": f"Проанализировано {len(df_analysis)} из ~{total_rows} строк" 
                                       if total_rows > len(df_analysis) else "Полный анализ",
                    },
                }
                sheet_result["summary"] = self._make_summary(sheet_result)
                result["sheets"][sh] = self._convert_to_serializable(sheet_result)
                
                logging.info(f"Лист '{sh}': {total_rows} строк, {total_cols} колонок (проанализировано {len(df_analysis)} строк)")
                
            except Exception as e:
                logging.exception(f"Ошибка анализа листа {sh}: {e}")
                result["sheets"][sh] = {
                    "name": sh,
                    "error": str(e),
                    "dimensions": {
                        "total_rows": 0,
                        "total_columns": 0,
                        "analyzed_rows": 0,
                    }
                }
        
        total_estimated_rows = sum(
            s.get("dimensions", {}).get("total_rows", 0) 
            for s in result["sheets"].values() 
            if isinstance(s.get("dimensions", {}).get("total_rows", 0), (int, float))
        )
        
        result["summary"] = self._convert_to_serializable({
            "total_sheets": len(result["sheets"]),
            "total_estimated_rows": int(total_estimated_rows),
            "total_columns": sum(
                len(s.get("columns", {})) 
                for s in result["sheets"].values()
            ),
            "analysis_time": datetime.now().isoformat(),
        })
        
        return self._convert_to_serializable(result)

    # ---------- вспомогательные методы ----------
    @staticmethod
    def _convert_to_serializable(obj: Any) -> Any:
        """Конвертирует объект в сериализуемый формат"""
        if isinstance(obj, (np.integer, np.int64, np.int32, np.int16, np.int8)):
            return int(obj)
        elif isinstance(obj, (np.floating, np.float64, np.float32, np.float16)):
            return float(obj)
        elif isinstance(obj, np.bool_):
            return bool(obj)
        elif isinstance(obj, np.ndarray):
            return obj.tolist()
        elif isinstance(obj, (pd.Timestamp, datetime)):
            return obj.isoformat()
        elif isinstance(obj, dict):
            # Преобразуем все ключи в строки
            new_dict = {}
            for k, v in obj.items():
                if not isinstance(k, (str, int, float, bool)) and k is not None:
                    k = str(k)
                new_dict[k] = ExcelAnalyzer._convert_to_serializable(v)
            return new_dict
        elif isinstance(obj, list):
            return [ExcelAnalyzer._convert_to_serializable(v) for v in obj]
        elif isinstance(obj, tuple):
            return tuple(ExcelAnalyzer._convert_to_serializable(v) for v in obj)
        elif pd.isna(obj):
            return None
        elif obj == "":
            return ""
        else:
            return obj

    @staticmethod
    def _calculate_empty_percentage(df: pd.DataFrame) -> float:
        """Рассчитывает процент пустых строк"""
        if len(df) == 0:
            return 100.0
        empty_rows = int(df.isnull().all(axis=1).sum())
        return float(round((empty_rows / len(df)) * 100, 2))

    def _infer_column_types(self, df: pd.DataFrame) -> Dict[str, Dict[str, Any]]:
        """Анализ типов столбцов на основе выборки"""
        if len(df) == 0:
            return {}
            
        cols = {}
        for c in df.columns:
            # Преобразуем имя столбца в строку (может быть datetime или другой тип)
            col_name = str(c)  # Эта строка уже делает преобразование
            
            # Учитываем пустые строки как отдельный случай
            non_empty = df[c].notna()
            empty_count = len(df) - non_empty.sum()
            
            if empty_count == len(df):
                cols[col_name] = {
                    "type": "empty", 
                    "sample_values": [], 
                    "unique_count": 0,
                    "filled_percent": 0.0,
                    "empty_count": empty_count
                }
                continue

            sample = df[c].dropna()
            if len(sample) == 0:
                cols[col_name] = {
                    "type": "empty", 
                    "sample_values": [], 
                    "unique_count": 0,
                    "filled_percent": 0.0,
                    "empty_count": empty_count
                }
                continue

            col_type = self._detect_column_type(sample)
            
            # Для смешанных типов (пустые строки + значения) отмечаем это
            unique_vals = sample.value_counts().head(20)
            
            cols[col_name] = {
                "type": col_type,
                "sample_values": self._convert_to_serializable(
                    unique_vals.index[:3].tolist()
                ),
                "unique_count": int(min(len(unique_vals), 1000)),
                "total_count": int(len(sample)),
                "filled_percent": float(round((len(sample) / len(df)) * 100, 1)),
                "empty_count": int(empty_count),
                "has_empty_strings": any(str(val).strip() == "" for val in df[c].dropna()[:10])
            }
                       
        return cols

    @staticmethod
    def _detect_column_type(sample: pd.Series) -> str:
        """Определяет тип столбца на основе выборки"""
        values = sample.head(10).tolist()
        
        # Проверяем на пустые строки
        if any(isinstance(v, str) and v.strip() == "" for v in values):
            return "mixed_empty_string"
        
        for val in values:
            if isinstance(val, (datetime, pd.Timestamp)):
                return "datetime"
        
        numeric_count = 0
        string_count = 0
        for val in values:
            if isinstance(val, (int, float, np.integer, np.floating)):
                numeric_count += 1
            elif isinstance(val, str):
                string_count += 1
                # Пытаемся распарсить как число
                try:
                    clean_val = str(val).replace(',', '.').replace(' ', '').replace(' ', '')
                    float(clean_val)
                    numeric_count += 1
                except:
                    pass
        
        if numeric_count == len(values):
            if all(isinstance(v, int) or (isinstance(v, float) and v.is_integer()) 
                   for v in values if isinstance(v, (int, float))):
                return "integer"
            return "float"
        
        for val in values:
            if isinstance(val, str):
                for pattern in ExcelAnalyzer.DATE_PATTERNS:
                    if re.fullmatch(pattern, str(val).strip()):
                        return "datetime_string"
        
        return "string"

    @staticmethod
    def _suggest_column_purpose(name: str, info: Dict) -> str:
        """Предлагает назначение столбца"""
        # Преобразуем name в строку, если это не строка (может быть datetime или другой тип)
        if not isinstance(name, str):
            name = str(name)
        n = name.lower()
        t = info["type"]

        purpose_keywords = {
            "id": ["id", "код", "uuid", "guid", "артикул", "номер", "№"],
            "date": ["дата", "date", "время", "time", "период", "срок"],
            "amount": ["сумма", "цена", "amount", "price", "вес", "стоимость", "руб", "стоимость"],
            "quantity": ["количество", "qty", "шт", "м", "метров", "штук", "объем", "вес"],
            "status": ["статус", "status", "flag", "признак", "флаг"],
            "category": ["вид", "тип", "категория", "группа", "класс", "вид", "тип"],
            "email": ["email", "e-mail", "почта", "mail"],
            "phone": ["телефон", "phone", "тел.", "мобильный"],
        }
        
        for purpose, keywords in purpose_keywords.items():
            if any(kw in n for kw in keywords):
                return purpose
        
        if t in ("integer", "float"):
            if info.get("unique_count", 0) < 100:
                return "code"
            return "numeric_value"
        elif t == "string":
            if info.get("unique_count", 0) <= 50:
                return "category"
            if "наименование" in n or "название" in n or "name" in n:
                return "name"
        elif t in ("datetime", "datetime_string"):
            return "timestamp"
            
        return "text"

    @staticmethod
    def _make_summary(info: Dict) -> Dict[str, Any]:
        """Создает сводку по листу"""
        cols = info.get("columns", {})
        dimensions = info.get("dimensions", {})
        
        return {
            "total_columns": len(cols),
            "estimated_rows": dimensions.get("total_rows", 0),
            "analyzed_rows": dimensions.get("analyzed_rows", 0),
            "numeric_columns": sum(1 for c in cols.values() 
                                 if c.get("type") in ("integer", "float")),
            "date_columns": sum(1 for c in cols.values() 
                               if c.get("type") in ("datetime", "datetime_string")),
            "empty_columns": sum(1 for c in cols.values() 
                                if c.get("filled_percent", 100) == 0),
            "mixed_type_columns": sum(1 for c in cols.values() 
                                     if c.get("type") == "mixed_empty_string"),
            "key_candidates": [
                k for k, v in cols.items() 
                if v.get("suggested_purpose") in ("id", "code")
            ],
        }