ruby2210/rag-chatbot

1"""2File processing utilities for the RAG Chatbot application.3Handles markdown parsing and other file operations for content ingestion.4"""5import markdown6from bs4 import BeautifulSoup7import re8from pathlib import Path9from typing import List, Dict, Any, Optional10from ..utils.logging import get_logger11 12 13logger = get_logger(__name__)14 15 16class FileProcessor:17    """18    Utility class for processing various file types, primarily markdown files for book content.19    """20    def __init__(self):21        pass22 23    def extract_text_from_markdown(self, file_path: str) -> str:24        """25        Extract plain text from a markdown file, preserving the structure.26        """27        try:28            with open(file_path, 'r', encoding='utf-8') as file:29                markdown_content = file.read()30 31            # Convert markdown to HTML32            html_content = markdown.markdown(markdown_content)33 34            # Parse HTML and extract text35            soup = BeautifulSoup(html_content, 'html.parser')36 37            # Extract text while preserving paragraph structure38            text_parts = []39            for element in soup.descendants:40                if element.name in ['p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'li', 'blockquote']:41                    text = element.get_text().strip()42                    if text:43                        text_parts.append(text)44                elif element.name is None and element.strip():  # Text nodes45                    text = element.strip()46                    if text:47                        text_parts.append(text)48 49            return '\n\n'.join(text_parts)50        except Exception as e:51            logger.error(f"Error extracting text from markdown {file_path}: {e}")52            return ""53 54    def extract_structured_content_from_markdown(self, file_path: str) -> Dict[str, Any]:55        """56        Extract structured content from a markdown file, preserving headings and sections.57        """58        try:59            with open(file_path, 'r', encoding='utf-8') as file:60                markdown_content = file.read()61 62            # Convert markdown to HTML63            html_content = markdown.markdown(markdown_content)64 65            # Parse HTML66            soup = BeautifulSoup(html_content, 'html.parser')67 68            # Extract structured content69            content_structure = {70                "title": "",71                "sections": [],72                "headings": [],73                "paragraphs": [],74                "raw_text": ""75            }76 77            # Find the main title (h1)78            h1_tag = soup.find('h1')79            if h1_tag:80                content_structure["title"] = h1_tag.get_text().strip()81 82            # Process all elements in order83            for element in soup.find_all(['h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'p', 'li', 'blockquote']):84                element_data = {85                    "tag": element.name,86                    "text": element.get_text().strip(),87                    "level": int(element.name[1]) if element.name.startswith('h') else 088                }89 90                if element.name.startswith('h'):91                    content_structure["headings"].append(element_data)92                elif element.name == 'p':93                    content_structure["paragraphs"].append(element_data)94 95            # Extract raw text96            content_structure["raw_text"] = soup.get_text(separator=' ', strip=True)97 98            return content_structure99        except Exception as e:100            logger.error(f"Error extracting structured content from markdown {file_path}: {e}")101            # Return a minimal structure in case of error102            return {103                "title": "",104                "sections": [],105                "headings": [],106                "paragraphs": [],107                "raw_text": self.extract_text_from_markdown(file_path)108            }109 110    def get_files_by_extension(self, source_path: str, extensions: List[str]) -> List[str]:111        """112        Get all files with specified extensions from the source path recursively.113        """114        try:115            source_dir = Path(source_path)116            if not source_dir.exists():117                logger.error(f"Source path does not exist: {source_path}")118                return []119 120            files = []121            for ext in extensions:122                # Find all files with the extension recursively123                ext_files = list(source_dir.rglob(f"*.{ext.lstrip('.')}"))124                files.extend([str(file_path) for file_path in ext_files])125 126            return files127        except Exception as e:128            logger.error(f"Error getting files from {source_path}: {e}")129            return []130 131    def read_file(self, file_path: str) -> Optional[str]:132        """133        Read the content of a file with proper encoding handling.134        """135        try:136            with open(file_path, 'r', encoding='utf-8') as file:137                return file.read()138        except UnicodeDecodeError:139            # Try with different encoding140            try:141                with open(file_path, 'r', encoding='latin-1') as file:142                    return file.read()143            except Exception:144                logger.error(f"Error reading file {file_path} with multiple encodings")145                return None146        except Exception as e:147            logger.error(f"Error reading file {file_path}: {e}")148            return None149 150 151# Create a singleton instance152file_processor = FileProcessor()