core/file_processor.py - main - AntiManager/table_structure_analyzer

core/file_processor.py
289 строк12 KB
Eugene
fix big data
26 дек 2025, 19:16
26 дек 2025, 19:165be946c
"""
Обработчик файлов и папок
Оптимизировано для работы с большими объемами данных
"""
from pathlib import Path
from typing import Dict, List, Any, Optional, Tuple
import pandas as pd
import logging
from concurrent.futures import ThreadPoolExecutor, as_completed
import time

from .excel_analyzer import ExcelAnalyzer
from .config import CACHE_DIR, MAX_FOLDER_FILES

class FileProcessor:
    """Обработчик файлов и папок для анализа"""

    def __init__(self):
        self._analyzer_classes = {"excel": ExcelAnalyzer}
        self._analyzer_cache = {}

    def get_analyzer(self, file_path: str):
        """Возвращает подходящий анализатор для файла"""
        extension = Path(file_path).suffix.lower()
        for cls in self._analyzer_classes.values():
            if cls.supports_extension(extension):
                # Кэшируем анализаторы
                if cls not in self._analyzer_cache:
                    self._analyzer_cache[cls] = cls()
                return self._analyzer_cache[cls]
        return None

    def process_single_file(
        self,
        file_path: str,
        sheet_names: Optional[List[str]] = None,
        deep: bool = False,
    ) -> Optional[Dict[str, Any]]:
        """Обрабатывает одиночный файл с оптимизацией"""
        analyzer = self.get_analyzer(file_path)
        if not analyzer:
            logging.error(f"Неподдерживаемый формат файла: {file_path}")
            return None

        try:
            logging.info(f"Начало анализа файла: {Path(file_path).name}")
            start_time = time.time()
            
            # Используем быстрый режим по умолчанию
            structure = analyzer.analyze_structure(file_path, sheet_names, deep=deep)
            
            elapsed = time.time() - start_time
            logging.info(f"Анализ завершен за {elapsed:.1f} секунд")
            
            return structure
        except Exception as e:
            logging.error(f"Ошибка обработки файла {file_path}: {e}")
            return None

    def get_folder_stats(self, folder_path: str) -> Optional[Dict[str, Any]]:
        """Собирает статистику по папке БЕЗ чтения всех данных"""
        folder = Path(folder_path)
        
        if not folder.exists() or not folder.is_dir():
            logging.error(f"Папка не существует: {folder_path}")
            return None
        
        # Получаем все Excel файлы
        excel_extensions = ExcelAnalyzer.SUPPORTED_EXTENSIONS
        files = []
        for ext in excel_extensions:
            files.extend(folder.glob(f"*{ext}"))
        
        if not files:
            return {
                "total_files": 0,
                "total_estimated_rows": 0,
                "total_columns": 0,
                "processed_files": 0,
                "file_list": [],
                "analysis_note": "Файлы не найдены"
            }
        
        # Ограничиваем количество обрабатываемых файлов для скорости
        if len(files) > MAX_FOLDER_FILES:
            files = files[:MAX_FOLDER_FILES]
            logging.warning(f"Ограничение: анализируем только первые {MAX_FOLDER_FILES} файлов")
        
        total_estimated_rows = 0
        total_columns = 0
        processed_files = 0
        file_list = []
        
        logging.info(f"Сбор статистики для {len(files)} файлов...")
        
        # Используем многопоточность для параллельной обработки
        with ThreadPoolExecutor(max_workers=4) as executor:
            future_to_file = {
                executor.submit(self._process_file_stats, str(file_path)): file_path
                for file_path in files
            }
            
            for future in as_completed(future_to_file):
                file_path = future_to_file[future]
                try:
                    result = future.result(timeout=30)  # Таймаут 30 секунд на файл
                    if result:
                        total_estimated_rows += result.get("estimated_rows", 0)
                        total_columns += result.get("columns", 0)
                        processed_files += 1
                        file_list.append(result)
                except Exception as e:
                    logging.warning(f"Ошибка обработки {file_path.name}: {e}")
        
        # Сортируем файлы по размеру
        file_list.sort(key=lambda x: x.get("size", 0), reverse=True)
        
        return {
            "total_files": len(files),
            "total_estimated_rows": total_estimated_rows,
            "total_columns": total_columns,
            "processed_files": processed_files,
            "file_list": file_list[:10],  # Только 10 самых больших файлов
            "analysis_note": f"Проанализировано {processed_files} из {len(files)} файлов",
            "estimation_note": "Количество строк - приблизительная оценка",
        }
    
    def _process_file_stats(self, file_path: str) -> Optional[Dict[str, Any]]:
        """Быстрая обработка статистики одного файла"""
        try:
            analyzer = self.get_analyzer(file_path)
            if not analyzer:
                return None
            
            path_obj = Path(file_path)
            file_size = path_obj.stat().st_size
            
            # Получаем информацию о листах
            sheet_info = analyzer.get_sheet_info(file_path)
            if not sheet_info:
                return None
            
            # Быстрая оценка размеров для первого листа (или самого большого)
            estimated_rows = 0
            total_columns = 0
            
            for sheet_name in list(sheet_info.keys())[:3]:  # Анализируем только первые 3 листа
                try:
                    rows, cols = analyzer.get_sheet_dimensions(file_path, sheet_name)
                    estimated_rows += rows
                    total_columns += cols
                except:
                    # Если не удалось, оцениваем по количеству колонок
                    total_columns += len(sheet_info[sheet_name])
            
            return {
                "name": path_obj.name,
                "path": str(file_path),
                "size": file_size,
                "size_mb": round(file_size / 1024 / 1024, 2),
                "sheets": list(sheet_info.keys()),
                "total_sheets": len(sheet_info),
                "estimated_rows": estimated_rows,
                "columns": total_columns,
            }
        except Exception as e:
            logging.debug(f"Ошибка статистики для {Path(file_path).name}: {e}")
            return None

    def process_folder(
        self, folder_path: str, file_pattern: str = "*", deep: bool = False
    ) -> Dict[str, Dict[str, Any]]:
        """Обрабатывает файлы в папке с оптимизацией"""
        results = {}
        folder = Path(folder_path)

        if not folder.exists() or not folder.is_dir():
            logging.error(f"Папка не существует: {folder_path}")
            return results

        # Ограничиваем количество файлов
        excel_extensions = ExcelAnalyzer.SUPPORTED_EXTENSIONS
        files = []
        for ext in excel_extensions:
            files.extend(folder.glob(f"*{ext}"))
        
        if len(files) > MAX_FOLDER_FILES:
            files = files[:MAX_FOLDER_FILES]
            logging.warning(f"Ограничение: обрабатываем только {MAX_FOLDER_FILES} файлов")

        if not files:
            return results

        # Получаем статистику
        folder_stats = self.get_folder_stats(folder_path)
        
        # Берем самый маленький файл как образец (для скорости)
        files_sorted = sorted(files, key=lambda x: x.stat().st_size)
        sample_file = files_sorted[0] if len(files_sorted) > 0 else files[0]
        
        logging.info(f"Анализируем образец: {sample_file.name}")
        sample_structure = self.process_single_file(str(sample_file), deep=deep)

        if sample_structure:
            # Добавляем статистику
            sample_structure["folder_stats"] = folder_stats
            results[str(sample_file)] = sample_structure

        return results

    def validate_files_structure(
        self, folder_path: str, deep: bool = False
    ) -> Tuple[bool, Dict[str, Any]]:
        """Проверяет структуру файлов с оптимизацией"""
        folder = Path(folder_path)
        excel_extensions = ExcelAnalyzer.SUPPORTED_EXTENSIONS
        files = []
        for ext in excel_extensions:
            files.extend(folder.glob(f"*{ext}"))

        if len(files) < 2:
            return True, {
                "message": "Для проверки нужно как минимум 2 файла",
                "total_files": len(files),
                "sample_structure": None,
            }

        # Ограничиваем количество анализируемых файлов
        max_files_to_analyze = min(5, len(files))
        files_to_analyze = files[:max_files_to_analyze]
        
        logging.info(f"Проверка структуры {len(files_to_analyze)} файлов...")

        structures = {}
        for file in files_to_analyze:
            # Используем быстрый режим для проверки
            structure = self.process_single_file(str(file), deep=False)
            if structure:
                structures[file.name] = structure

        if len(structures) < 2:
            return False, {
                "error": "Не удалось проанализировать файлы",
                "total_files": len(files),
                "sample_structure": None,
            }

        first_file = list(structures.keys())[0]
        first_structure = structures[first_file]

        differences = []
        for file_name, structure in list(structures.items())[1:]:
            diff = self._compare_structures(first_structure, structure)
            if diff:
                differences.append({"file": file_name, "differences": diff})

        # Получаем статистику
        folder_stats = self.get_folder_stats(folder_path)
        
        return len(differences) == 0, {
            "total_files": len(files),
            "analyzed_files": len(structures),
            "sample_structure": first_structure,
            "differences": differences if differences else None,
            "folder_stats": folder_stats,
            "analysis_note": f"Проверено {len(structures)} из {len(files)} файлов",
        }

    @staticmethod
    def _compare_structures(struct1: Dict, struct2: Dict) -> List[str]:
        """Быстрое сравнение структур"""
        differences = []

        sheets1 = set(struct1.get("sheets", {}).keys())
        sheets2 = set(struct2.get("sheets", {}).keys())

        if sheets1 != sheets2:
            differences.append(f"Разные листы: {sheets1} vs {sheets2}")
        else:
            # Сравниваем только первые 3 листа для скорости
            for sheet in list(sheets1)[:3]:
                cols1 = set(struct1["sheets"][sheet].get("columns", {}).keys())
                cols2 = set(struct2["sheets"][sheet].get("columns", {}).keys())
                if cols1 != cols2:
                    differences.append(
                        f"Лист '{sheet}': разные колонки ({len(cols1)} vs {len(cols2)})"
                    )
        
        return differences