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

Python/Archaeology/template_parser.py
410 строк18 KB
d_e_m_e_k
еще один заход на анализ и загрузку данных из excel по керамике в БД
10 авг 2026, 13:33
10 авг 2026, 13:33b15832f
# template_parser.py
import pandas as pd
import re
from typing import List, Dict, Any, Optional, Tuple
import logging
import numpy as np

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

class TemplateParser:
    """Класс для парсинга шаблонов керамики из Excel файлов"""
    
    # Типы шаблонов и их структура
    TEMPLATE_TYPES = {
        'лепная': {
            'header_rows': 2,
            'subheader_row': 3,
            'data_start': 4,
            'columns': {
                'thin_walls': 'B',
                'thick_walls': 'C',
                'simple_rims': 'D',
                'profiled_rims': 'F',
                'nail_rope': 'G',
                'bottoms': 'J',
                'total_count': 'M',
                'total_weight': 'O',
                'linear_ornament': 'Q',
                'wave_zigzag': 'R',
                'nail_ornament': 'S',
                'rope_ornament': 'T',
                'stamp_ornament': 'U',
                'combined_ornament': 'V',
                'total_ornamented': 'W'
            }
        },
        'раннекруговая': {
            'header_rows': 2,
            'subheader_row': 3,
            'data_start': 4,
            'columns': {
                'thin_walls': 'B',
                'thick_walls': 'C',
                'rims_v0': 'D',
                'rims_v1': 'E',
                'rims_v2': 'F',
                'rims_v4': 'G',
                'rims_v_unknown': 'H',
                'bowls': 'I',
                'other': 'J',
                'bottoms_without': 'K',
                'bottoms_with': 'L',
                'stamp': 'M',
                'total_count': 'N',
                'total_weight': 'P',
                'linear_ornament': 'R',
                'wave_zigzag': 'S',
                'nail_ornament': 'T',
                'rope_ornament': 'U',
                'stamp_ornament': 'V',
                'combined_ornament': 'W',
                'total_ornamented': 'X'
            }
        },
        'развитая круговая': {
            'header_rows': 2,
            'subheader_row': 3,
            'data_start': 4,
            'columns': {
                'thin_walls': 'B',
                'thick_walls': 'C',
                'rims_v0': 'D',
                'rims_v1': 'E',
                'rims_v3': 'F',
                'rims_v4': 'G',
                'rims_v_unknown': 'H',
                'bowls': 'I',
                'other': 'J',
                'bottoms_without': 'K',
                'bottoms_with': 'L',
                'stamp': 'M',
                'total_count': 'N',
                'total_weight': 'P',
                'linear_ornament': 'R',
                'wave_zigzag': 'S',
                'nail_ornament': 'T',
                'rope_ornament': 'U',
                'stamp_ornament': 'V',
                'combined_ornament': 'W',
                'total_ornamented': 'X'
            }
        },
        'поздняя круговая': {
            'header_rows': 2,
            'subheader_row': 3,
            'data_start': 4,
            'columns': {
                'thin_walls': 'B',
                'thick_walls': 'C',
                'rims_v10': 'G',
                'rims_v11': 'H',
                'handles': 'I',
                'bottoms_without': 'K',
                'bottoms_with': 'L',
                'stamp': 'M',
                'total_count': 'N',
                'total_weight': 'P',
                'linear_ornament': 'R',
                'wave_zigzag': 'S',
                'nail_ornament': 'T',
                'rope_ornament': 'U',
                'glaze_ornament': 'V',
                'combined_ornament': 'W',
                'total_ornamented': 'X'
            }
        }
    }
    
    # Паттерны для поиска шаблонов
    TEMPLATE_PATTERNS = {
        'лепная': r'Лепная керамика',
        'раннекруговая': r'Раннекруговая керамика',
        'развитая круговая': r'Развитая круговая керамика',
        'поздняя круговая': r'Поздняя круговая керамика'
    }
    
    def __init__(self):
        pass
    
    def _safe_row_to_string(self, row: pd.Series) -> str:
        """Безопасно преобразует строку DataFrame в строку"""
        row_values = []
        for val in row.values:
            if pd.isna(val):
                row_values.append('')
            elif isinstance(val, (int, float)):
                row_values.append(str(int(val)) if val == int(val) else str(val))
            else:
                row_values.append(str(val))
        return ' '.join(row_values)
    
    def _letter_to_index(self, letter: str) -> int:
        """Преобразует букву колонки в индекс"""
        if not letter:
            return 0
        letter = letter.upper()
        index = 0
        for char in letter:
            index = index * 26 + (ord(char) - ord('A') + 1)
        return index - 1
    
    def _get_block_name_from_row(self, row: pd.Series, col_index: int = 0) -> Tuple[str, int]:
        """Извлекает название блока из строки и возвращает его и индекс колонки"""
        for idx in range(col_index, len(row)):
            val = row.iloc[idx]
            if pd.notna(val) and isinstance(val, str):
                val_str = str(val).strip()
                # Проверяем паттерны названий блоков
                block_patterns = [
                    r'кв\.\s*[\d\-]+,?\s*гл\.\s*[\d\-,\.]+',
                    r'кв\.\s*[\d\-]+\s*[\d\-]+',
                    r'кв\.\s*[\d\-,]+\s*[\d\-]+'
                ]
                for pattern in block_patterns:
                    if re.search(pattern, val_str, re.IGNORECASE):
                        return val_str, idx
        return "Без названия", col_index
    
    def _find_block_boundaries(self, df: pd.DataFrame) -> List[Dict[str, Any]]:
        """Находит границы всех блоков на листе"""
        blocks = []
        
        # Сначала находим все строки с названиями блоков
        block_starts = []
        for idx, row in df.iterrows():
            row_str = self._safe_row_to_string(row)
            if re.search(r'кв\.', row_str, re.IGNORECASE):
                # Находим все названия блоков в строке
                for col_idx in range(len(row)):
                    val = row.iloc[col_idx]
                    if pd.notna(val) and isinstance(val, str):
                        val_str = str(val).strip()
                        if re.search(r'кв\.', val_str, re.IGNORECASE):
                            block_starts.append({
                                'row': idx,
                                'col': col_idx,
                                'name': val_str
                            })
        
        # Если не нашли блоки, создаем один блок на основе шаблонов
        if not block_starts:
            # Находим все строки с шаблонами
            template_rows = []
            for idx, row in df.iterrows():
                row_str = self._safe_row_to_string(row)
                for pattern in self.TEMPLATE_PATTERNS.values():
                    if re.search(pattern, row_str, re.IGNORECASE):
                        template_rows.append(idx)
                        break
            
            if template_rows:
                min_row = min(template_rows)
                max_row = max(template_rows)
                # Ищем название блока
                block_name = "Блок"
                for idx in range(max(0, min_row - 3), min_row):
                    row = df.iloc[idx]
                    for val in row.values:
                        if pd.notna(val) and isinstance(val, str):
                            if re.search(r'кв\.', str(val), re.IGNORECASE):
                                block_name = str(val).strip()
                                break
                
                blocks.append({
                    'row_start': min_row,
                    'row_end': max_row + 10,  # Примерный конец
                    'col_start': 0,
                    'col_end': len(df.columns),
                    'name': block_name
                })
            return blocks
        
        # Группируем блоки по строкам
        rows_with_blocks = {}
        for block in block_starts:
            row = block['row']
            if row not in rows_with_blocks:
                rows_with_blocks[row] = []
            rows_with_blocks[row].append(block)
        
        # Для каждой строки с блоками определяем границы
        for row, blocks_in_row in rows_with_blocks.items():
            # Сортируем по колонкам
            blocks_in_row.sort(key=lambda x: x['col'])
            
            for i, block in enumerate(blocks_in_row):
                col_start = block['col']
                
                # Определяем конец блока
                if i + 1 < len(blocks_in_row):
                    col_end = blocks_in_row[i + 1]['col']
                else:
                    # Ищем конец блока по наличию данных
                    col_end = len(df.columns)
                    # Проверяем, есть ли данные дальше
                    for col in range(col_start + 25, min(col_start + 40, len(df.columns))):
                        has_data = False
                        for row_offset in range(1, min(20, len(df) - row)):
                            check_row = row + row_offset
                            if check_row < len(df):
                                val = df.iloc[check_row, col] if col < len(df.columns) else None
                                if pd.notna(val) and str(val).strip() not in ['', '0', '0.0']:
                                    has_data = True
                                    break
                        if not has_data:
                            col_end = col
                            break
                
                # Определяем конец блока по строкам
                row_end = row + 50  # Примерная высота блока
                for r in range(row + 1, min(row + 50, len(df))):
                    # Проверяем, не начался ли новый блок в этой строке
                    row_str = self._safe_row_to_string(df.iloc[r])
                    if re.search(r'кв\.', row_str, re.IGNORECASE):
                        # Проверяем, не является ли это продолжением текущего блока
                        is_new_block = False
                        for col in range(col_start, min(col_end, len(df.columns))):
                            val = df.iloc[r, col] if col < len(df.columns) else None
                            if pd.notna(val) and isinstance(val, str):
                                if re.search(r'кв\.', str(val), re.IGNORECASE):
                                    is_new_block = True
                                    break
                        if is_new_block:
                            row_end = r
                            break
                    
                    # Проверяем, есть ли данные в этом блоке
                    has_data = False
                    for col in range(col_start, min(col_end, len(df.columns))):
                        val = df.iloc[r, col] if col < len(df.columns) else None
                        if pd.notna(val) and str(val).strip() not in ['', '0', '0.0']:
                            has_data = True
                            break
                    
                    if not has_data and r > row + 20:
                        row_end = r
                        break
                
                blocks.append({
                    'row_start': row,
                    'row_end': row_end,
                    'col_start': col_start,
                    'col_end': col_end,
                    'name': block['name']
                })
        
        return blocks
    
    def find_templates(self, df: pd.DataFrame, sheet_name: str) -> List[Dict[str, Any]]:
        """Находит все шаблоны в DataFrame"""
        templates = []
        
        # Находим границы всех блоков
        blocks = self._find_block_boundaries(df)
        
        # Для каждого блока извлекаем данные
        for block in blocks:
            block_name = block['name']
            row_start = block['row_start']
            row_end = block['row_end']
            col_start = block['col_start']
            col_end = block['col_end']
            
            logger.info(f"Processing block '{block_name}' at rows {row_start}-{row_end}, cols {col_start}-{col_end}")
            
            # Собираем все данные из блока
            all_data_rows = []
            
            # Проходим по всем строкам блока
            for idx in range(row_start + 1, min(row_end, len(df))):
                row = df.iloc[idx]
                first_val = str(row.iloc[0]) if len(row) > 0 and pd.notna(row.iloc[0]) else ''
                
                # Проверяем, не строка ли с итогами
                if 'Кол-во' in first_val:
                    continue
                
                # Проверяем, есть ли данные в колонках блока
                has_data = False
                for col in range(col_start, min(col_end, len(row))):
                    val = row.iloc[col]
                    if pd.notna(val) and str(val).strip() not in ['', '0', '0.0']:
                        has_data = True
                        break
                
                if has_data and first_val:
                    # Проверяем, что это не заголовок
                    is_header = False
                    for pattern in self.TEMPLATE_PATTERNS.values():
                        if re.search(pattern, first_val, re.IGNORECASE):
                            is_header = True
                            break
                    
                    if not is_header and first_val not in ['', 'Лепная керамика', 'Раннекруговая керамика', 
                                                          'Развитая круговая керамика', 'Поздняя круговая керамика',
                                                          'Неопределенные ф-ты:']:
                        # Парсим строку данных с учетом колонок блока
                        parsed_row = self._parse_data_row_in_block(row, col_start, col_end)
                        if parsed_row and parsed_row.get('subcategory'):
                            all_data_rows.append(parsed_row)
            
            if all_data_rows:
                # Определяем тип шаблона
                template_type = 'лепная'  # По умолчанию
                for idx in range(row_start, min(row_end, len(df))):
                    row_str = self._safe_row_to_string(df.iloc[idx])
                    for t_type, pattern in self.TEMPLATE_PATTERNS.items():
                        if re.search(pattern, row_str, re.IGNORECASE):
                            template_type = t_type
                            break
                    if template_type != 'лепная':
                        break
                
                template = {
                    'sheet_name': sheet_name,
                    'block_name': block_name,
                    'template_type': template_type,
                    'start_row': row_start,
                    'end_row': row_end,
                    'start_col': col_start,
                    'end_col': col_end,
                    'data_rows': all_data_rows,
                    'is_block': True
                }
                templates.append(template)
                logger.info(f"Found block '{block_name}' with {len(all_data_rows)} data rows")
        
        return templates
    
    def _parse_data_row_in_block(self, row: pd.Series, col_start: int, col_end: int) -> Dict[str, Any]:
        """Парсит строку данных в пределах блока"""
        result = {}
        
        # Извлекаем название подкатегории из первой колонки
        if len(row) > 0 and pd.notna(row.iloc[0]):
            result['subcategory'] = str(row.iloc[0]).strip()
        else:
            return {}
        
        # Парсим числовые данные из колонок блока
        col_idx = 1
        for col in range(col_start, min(col_end, len(row))):
            val = row.iloc[col]
            if pd.notna(val) and str(val).strip() not in ['', '0', '0.0']:
                try:
                    if isinstance(val, (int, float)):
                        result[f'col_{col_idx}'] = float(val)
                    else:
                        val_str = str(val).strip().replace(' ', '').replace(',', '.')
                        if val_str:
                            try:
                                result[f'col_{col_idx}'] = float(val_str)
                            except:
                                result[f'col_{col_idx}'] = val_str
                except:
                    result[f'col_{col_idx}'] = val
            col_idx += 1
        
        return result if result.get('subcategory') else {}