Python/Archaeology/excel_processor.py - main - Demek/DSam621

Python/Archaeology/excel_processor.py
316 строк14 KB
d_e_m_e_k
еще один заход на анализ и загрузку данных из excel по керамике в БД
10 авг 2026, 13:33
10 авг 2026, 13:33b15832f
# excel_processor.py
import pandas as pd
from typing import List, Dict, Any
from template_parser import TemplateParser
from db_manager import DatabaseManager
import logging
import os

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class ExcelProcessor:
    def __init__(self):
        self.parser = TemplateParser()
        self.db_manager = DatabaseManager()
    
    def _read_excel_file(self, file_path: str, sheet_name: str = None):
        """
        Читает Excel файл с поддержкой разных форматов
        """
        file_ext = os.path.splitext(file_path)[1].lower()
        
        try:
            if file_ext == '.xls':
                # Для старых .xls файлов используем xlrd
                try:
                    import xlrd
                    if sheet_name:
                        return pd.read_excel(file_path, sheet_name=sheet_name, header=None, engine='xlrd')
                    else:
                        return pd.read_excel(file_path, header=None, engine='xlrd')
                except ImportError:
                    logger.warning("xlrd not installed, trying to use openpyxl for .xls file")
                    # Некоторые .xls файлы могут быть открыты openpyxl
                    if sheet_name:
                        return pd.read_excel(file_path, sheet_name=sheet_name, header=None, engine='openpyxl')
                    else:
                        return pd.read_excel(file_path, header=None, engine='openpyxl')
            else:
                # Для .xlsx файлов используем openpyxl
                if sheet_name:
                    return pd.read_excel(file_path, sheet_name=sheet_name, header=None, engine='openpyxl')
                else:
                    return pd.read_excel(file_path, header=None, engine='openpyxl')
        except Exception as e:
            logger.error(f"Error reading Excel file: {e}")
            # Пробуем альтернативные методы
            try:
                if sheet_name:
                    return pd.read_excel(file_path, sheet_name=sheet_name, header=None)
                else:
                    return pd.read_excel(file_path, header=None)
            except:
                raise Exception(f"Could not read Excel file. Please ensure the file is not corrupted. Error: {e}")
    
    def process_file(self, file_path: str) -> Dict[str, Any]:
        """Обрабатывает Excel файл"""
        logger.info(f"Processing file: {file_path}")
        
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"File not found: {file_path}")
        
        try:
            # Проверяем расширение файла
            file_ext = os.path.splitext(file_path)[1].lower()
            if file_ext not in ['.xlsx', '.xls']:
                raise ValueError(f"Unsupported file format: {file_ext}. Only .xlsx and .xls are supported.")
            
            # Загружаем все листы
            try:
                # Пробуем открыть через ExcelFile
                if file_ext == '.xlsx':
                    excel_file = pd.ExcelFile(file_path, engine='openpyxl')
                else:
                    excel_file = pd.ExcelFile(file_path, engine='xlrd')
                sheet_names = excel_file.sheet_names
            except Exception as e:
                logger.warning(f"Could not open with specified engine: {e}")
                # Пробуем альтернативный способ
                try:
                    excel_file = pd.ExcelFile(file_path)
                    sheet_names = excel_file.sheet_names
                except:
                    # Если не удалось открыть через ExcelFile, пробуем через read_excel
                    excel_file = None
                    sheets_dict = pd.read_excel(file_path, sheet_name=None, header=None)
                    sheet_names = list(sheets_dict.keys())
            
            results = {
                'file_name': os.path.basename(file_path),
                'file_format': file_ext,
                'sheets_processed': 0,
                'templates_found': 0,
                'templates_saved': 0,
                'details': []
            }
            
            for sheet_name in sheet_names:
                logger.info(f"Processing sheet: {sheet_name}")
                try:
                    # Читаем лист
                    if excel_file:
                        df = pd.read_excel(excel_file, sheet_name=sheet_name, header=None)
                    else:
                        df = sheets_dict[sheet_name]
                    
                    if df.empty:
                        logger.warning(f"Sheet '{sheet_name}' is empty")
                        results['sheets_processed'] += 1
                        continue
                    
                    # Находим шаблоны на листе
                    templates = self.parser.find_templates(df, sheet_name)
                    results['sheets_processed'] += 1
                    results['templates_found'] += len(templates)
                    
                    if templates:
                        logger.info(f"Found {len(templates)} templates on sheet '{sheet_name}'")
                    
                    for template in templates:
                        template['file_name'] = os.path.basename(file_path)
                        
                        # Сохраняем в базу данных
                        try:
                            template_id = self.db_manager.save_template(template)
                            results['templates_saved'] += 1
                            results['details'].append({
                                'sheet': sheet_name,
                                'block': template.get('block_name', ''),
                                'type': template.get('template_type', ''),
                                'rows': len(template.get('data_rows', [])),
                                'id': template_id,
                                'status': 'saved'
                            })
                            logger.info(f"Saved template from sheet '{sheet_name}', block '{template.get('block_name', '')}'")
                        except Exception as e:
                            logger.error(f"Error saving template: {e}")
                            results['details'].append({
                                'sheet': sheet_name,
                                'block': template.get('block_name', ''),
                                'type': template.get('template_type', ''),
                                'error': str(e),
                                'status': 'error'
                            })
                except Exception as e:
                    logger.error(f"Error processing sheet {sheet_name}: {e}")
                    results['details'].append({
                        'sheet': sheet_name,
                        'error': str(e),
                        'status': 'error'
                    })
                    results['sheets_processed'] += 1
                    continue
            
            logger.info(f"Processing complete. Found {results['templates_found']}, saved {results['templates_saved']}")
            return results
            
        except Exception as e:
            logger.error(f"Error processing file: {e}")
            raise
    
    def process_directory(self, directory_path: str) -> List[Dict[str, Any]]:
        """Обрабатывает все Excel файлы в директории"""
        results = []
        
        if not os.path.exists(directory_path):
            raise FileNotFoundError(f"Directory not found: {directory_path}")
        
        if not os.path.isdir(directory_path):
            raise ValueError(f"Path is not a directory: {directory_path}")
        
        excel_files = []
        for file_name in os.listdir(directory_path):
            if file_name.endswith(('.xlsx', '.xls')):
                excel_files.append(file_name)
        
        if not excel_files:
            logger.warning(f"No Excel files found in directory: {directory_path}")
            return results
        
        logger.info(f"Found {len(excel_files)} Excel files to process")
        
        for file_name in excel_files:
            file_path = os.path.join(directory_path, file_name)
            try:
                logger.info(f"Processing file: {file_name}")
                result = self.process_file(file_path)
                results.append(result)
            except Exception as e:
                logger.error(f"Error processing {file_name}: {e}")
                results.append({
                    'file_name': file_name,
                    'error': str(e),
                    'status': 'error'
                })
        
        return results
    
    def process_files(self, file_paths: List[str]) -> List[Dict[str, Any]]:
        """Обрабатывает список Excel файлов"""
        results = []
        
        for file_path in file_paths:
            try:
                logger.info(f"Processing file: {file_path}")
                result = self.process_file(file_path)
                results.append(result)
            except Exception as e:
                logger.error(f"Error processing {file_path}: {e}")
                results.append({
                    'file_name': os.path.basename(file_path) if os.path.exists(file_path) else file_path,
                    'error': str(e),
                    'status': 'error'
                })
        
        return results
    
    def process_file_with_encoding(self, file_path: str, encoding: str = 'utf-8') -> Dict[str, Any]:
        """Обрабатывает Excel файл с указанной кодировкой"""
        logger.info(f"Processing file with encoding {encoding}: {file_path}")
        
        if not os.path.exists(file_path):
            raise FileNotFoundError(f"File not found: {file_path}")
        
        try:
            # Пытаемся прочитать файл с указанной кодировкой
            excel_file = pd.ExcelFile(file_path)
            sheet_names = excel_file.sheet_names
            
            results = {
                'file_name': os.path.basename(file_path),
                'encoding': encoding,
                'sheets_processed': 0,
                'templates_found': 0,
                'templates_saved': 0,
                'details': []
            }
            
            for sheet_name in sheet_names:
                logger.info(f"Processing sheet: {sheet_name}")
                try:
                    # Читаем лист с указанной кодировкой
                    df = pd.read_excel(file_path, sheet_name=sheet_name, header=None)
                    
                    # Находим шаблоны на листе
                    templates = self.parser.find_templates(df, sheet_name)
                    results['sheets_processed'] += 1
                    results['templates_found'] += len(templates)
                    
                    for template in templates:
                        template['file_name'] = os.path.basename(file_path)
                        
                        try:
                            template_id = self.db_manager.save_template(template)
                            results['templates_saved'] += 1
                            results['details'].append({
                                'sheet': sheet_name,
                                'block': template.get('block_name', ''),
                                'type': template.get('template_type', ''),
                                'rows': len(template.get('data_rows', [])),
                                'id': template_id,
                                'status': 'saved'
                            })
                        except Exception as e:
                            logger.error(f"Error saving template: {e}")
                            results['details'].append({
                                'sheet': sheet_name,
                                'error': str(e),
                                'status': 'error'
                            })
                except Exception as e:
                    logger.error(f"Error processing sheet {sheet_name}: {e}")
                    results['details'].append({
                        'sheet': sheet_name,
                        'error': str(e),
                        'status': 'error'
                    })
                    results['sheets_processed'] += 1
                    continue
            
            return results
            
        except Exception as e:
            logger.error(f"Error processing file: {e}")
            raise
    
    def get_statistics(self) -> Dict[str, Any]:
        """Получает статистику по обработанным файлам"""
        try:
            templates = self.db_manager.get_templates()
            
            stats = {
                'total_templates': len(templates),
                'by_type': {},
                'by_file': {},
                'by_sheet': {}
            }
            
            for template in templates:
                # Статистика по типам
                template_type = template.get('template_type', 'unknown')
                stats['by_type'][template_type] = stats['by_type'].get(template_type, 0) + 1
                
                # Статистика по файлам
                file_name = template.get('file_name', 'unknown')
                stats['by_file'][file_name] = stats['by_file'].get(file_name, 0) + 1
                
                # Статистика по листам
                sheet_name = template.get('sheet_name', 'unknown')
                stats['by_sheet'][sheet_name] = stats['by_sheet'].get(sheet_name, 0) + 1
            
            return stats
            
        except Exception as e:
            logger.error(f"Error getting statistics: {e}")
            return {'error': str(e)}