utils/file_processor.py - master - triv/GigaDocGen

utils/file_processor.py
210 строк7 KB
triv
расширен дебаг и изменен подход работы с файлами
23 окт 2025, 14:53
23 окт 2025, 14:5389fd19b
"""
Утилиты для работы с файлами и обработки .groovy скриптов
"""

import os
import re
import hashlib
from pathlib import Path
from typing import List, Dict, Tuple, Optional
import logging

logger = logging.getLogger(__name__)


class FileProcessor:
    """Класс для обработки файлов и поиска .groovy скриптов"""
    
    def __init__(self, config: Dict):
        self.config = config
        # Безопасное получение конфигурации с значениями по умолчанию
        processing = config.get('processing', {})
        self.supported_extensions = processing.get('supported_extensions', ['.groovy', '.jenkinsfile'])
        self.directories = config.get('directories', [])
    
    def find_groovy_files(self, base_path: str) -> List[Path]:
        """
        Поиск всех .groovy файлов в указанных директориях
        
        Args:
            base_path: Базовый путь к репозиторию
            
        Returns:
            List[Path]: Список найденных .groovy файлов
        """
        groovy_files = []
        base_path = Path(base_path)
        
        for directory in self.directories:
            search_path = base_path / directory
            if search_path.exists():
                for ext in self.supported_extensions:
                    pattern = f"**/*{ext}"
                    files = list(search_path.glob(pattern))
                    groovy_files.extend(files)
                    logger.info(f"Найдено {len(files)} файлов {ext} в {directory}")
            else:
                logger.warning(f"Директория {search_path} не найдена")
        
        return groovy_files
    
    def read_file_content(self, file_path: Path) -> str:
        """
        Чтение содержимого файла
        
        Args:
            file_path: Путь к файлу
            
        Returns:
            str: Содержимое файла
        """
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            logger.info(f"Прочитан файл: {file_path} ({len(content)} символов)")
            return content
        except Exception as e:
            logger.error(f"Ошибка чтения файла {file_path}: {e}")
            return ""
    
    def calculate_context_size(self, content: str) -> int:
        """
        Расчет размера контекста (примерное количество токенов)
        
        Args:
            content: Содержимое файла
            
        Returns:
            int: Примерное количество токенов
        """
        # Примерная оценка: 1 токен ≈ 4 символа
        return len(content) // 4
    
    def split_content_into_chunks(self, content: str, chunk_size: int = 1000, overlap: int = 200) -> List[str]:
        """
        Разбиение контента на чанки для обработки больших файлов
        
        Args:
            content: Содержимое файла
            chunk_size: Размер чанка
            overlap: Перекрытие между чанками
            
        Returns:
            List[str]: Список чанков
        """
        if len(content) <= chunk_size:
            return [content]
        
        chunks = []
        start = 0
        
        while start < len(content):
            end = start + chunk_size
            
            # Пытаемся разбить по строкам
            if end < len(content):
                # Ищем ближайший перенос строки
                last_newline = content.rfind('\n', start, end)
                if last_newline > start:
                    end = last_newline
            
            chunk = content[start:end]
            chunks.append(chunk)
            
            # Следующий чанк начинается с перекрытием
            start = end - overlap
            if start >= len(content):
                break
        
        logger.info(f"Разбито на {len(chunks)} чанков")
        return chunks
    
    def extract_jenkins_info(self, content: str) -> Dict[str, any]:
        """
        Извлечение информации о Jenkins pipeline из содержимого
        
        Args:
            content: Содержимое .groovy файла
            
        Returns:
            Dict: Извлеченная информация
        """
        info = {
            'pipeline_name': '',
            'parameters': [],
            'stages': [],
            'triggers': [],
            'tools': []
        }
        
        # Поиск имени pipeline
        pipeline_match = re.search(r'pipeline\s*\{', content, re.IGNORECASE)
        if pipeline_match:
            info['pipeline_name'] = 'Jenkins Pipeline'
        
        # Поиск параметров
        parameters_match = re.search(r'parameters\s*\{([^}]+)\}', content, re.DOTALL)
        if parameters_match:
            params_text = parameters_match.group(1)
            # Извлечение типов параметров
            param_types = re.findall(r'(\w+)\s*\([^)]+\)', params_text)
            info['parameters'] = param_types
        
        # Поиск stages
        stages_match = re.search(r'stages\s*\{([^}]+)\}', content, re.DOTALL)
        if stages_match:
            stages_text = stages_match.group(1)
            stage_names = re.findall(r'stage\s*\(\s*[\'"]([^\'"]+)[\'"]\s*\)', stages_text)
            info['stages'] = stage_names
        
        # Поиск триггеров
        triggers = re.findall(r'(triggers|when|if)\s*\{', content, re.IGNORECASE)
        info['triggers'] = triggers
        
        # Поиск инструментов
        tools = re.findall(r'tool\s*\(\s*[\'"]([^\'"]+)[\'"]\s*\)', content)
        info['tools'] = tools
        
        return info
    
    def create_file_hash(self, content: str) -> str:
        """
        Создание хэша содержимого файла для кэширования
        
        Args:
            content: Содержимое файла
            
        Returns:
            str: MD5 хэш
        """
        return hashlib.md5(content.encode('utf-8')).hexdigest()
    
    def is_file_processed(self, file_path: Path, output_dir: Path) -> bool:
        """
        Проверка, был ли файл уже обработан
        
        Args:
            file_path: Путь к исходному файлу
            output_dir: Директория с результатами
            
        Returns:
            bool: True если файл уже обработан
        """
        doc_file = output_dir / f"{file_path.stem}.md"
        return doc_file.exists()
    
    def get_relative_path(self, file_path: Path, base_path: Path) -> str:
        """
        Получение относительного пути файла
        
        Args:
            file_path: Путь к файлу
            base_path: Базовый путь
            
        Returns:
            str: Относительный путь
        """
        try:
            return str(file_path.relative_to(base_path))
        except ValueError:
            return str(file_path)