Spaces:
Paused
Paused
| """Markdown parser for extracting content and metadata from RAG database files.""" | |
| from __future__ import annotations | |
| import re | |
| from pathlib import Path | |
| from typing import Any | |
| import yaml | |
| # Constants | |
| DEFAULT_CHUNK_SIZE = 1000 | |
| PARAGRAPH_SEPARATOR = "\n\n" | |
| PARAGRAPH_SEPARATOR_LENGTH = len(PARAGRAPH_SEPARATOR) | |
| class MarkdownParser: | |
| """Parser for markdown files with YAML frontmatter.""" | |
| def parse_frontmatter(self, content: str) -> dict[str, Any]: | |
| """ | |
| Extract YAML frontmatter from markdown content. | |
| Args: | |
| content: Markdown file content | |
| Returns: | |
| Dictionary of metadata from frontmatter, or empty dict if none | |
| """ | |
| # Match content between --- delimiters at start of file | |
| # Allow Windows (CRLF) or Unix (LF) line endings | |
| # Allow EOF after closing --- (no trailing newline required) | |
| match = re.match(r"^---\s*\r?\n(.*?)\r?\n---(?:\s*\r?\n|$)", content, re.DOTALL) | |
| if not match: | |
| return {} | |
| frontmatter_text = match.group(1) | |
| try: | |
| return yaml.safe_load(frontmatter_text) or {} | |
| except yaml.YAMLError: | |
| return {} | |
| def extract_sections(self, content: str) -> list[dict[str, str]]: | |
| """ | |
| Extract markdown sections by ## and ### headers. | |
| Extracts both ## sections and ### subsections as separate chunks | |
| to create more granular retrieval units. | |
| Args: | |
| content: Markdown file content | |
| Returns: | |
| List of dicts with {title, content} for each section/subsection | |
| """ | |
| # Remove frontmatter first (support CRLF and allow EOF after ---) | |
| content = re.sub(r"^---\s*\r?\n.*?\r?\n---(?:\s*\r?\n|$)", "", content, flags=re.DOTALL) | |
| sections = [] | |
| # Find all ## headers | |
| h2_pattern = r"^## (.+)$" | |
| h2_matches = list(re.finditer(h2_pattern, content, re.MULTILINE)) | |
| for i, h2_match in enumerate(h2_matches): | |
| h2_title = h2_match.group(1).strip() | |
| h2_start = h2_match.end() | |
| # Content ends at next ## header or end of file | |
| if i + 1 < len(h2_matches): | |
| h2_end = h2_matches[i + 1].start() | |
| else: | |
| h2_end = len(content) | |
| h2_section_content = content[h2_start:h2_end] | |
| # Find ### subsections within this ## section | |
| h3_pattern = r"^### (.+)$" | |
| h3_matches = list(re.finditer(h3_pattern, h2_section_content, re.MULTILINE)) | |
| if h3_matches: | |
| # Check if there's content before the first ### subsection | |
| first_h3_start = h3_matches[0].start() | |
| intro_content = h2_section_content[:first_h3_start].strip() | |
| if intro_content: | |
| # Create a chunk for the intro content | |
| sections.append({"title": h2_title, "content": intro_content}) | |
| # Process each ### subsection | |
| for j, h3_match in enumerate(h3_matches): | |
| h3_title = h3_match.group(1).strip() | |
| h3_start = h3_match.end() | |
| # Content ends at next ### or end of section | |
| if j + 1 < len(h3_matches): | |
| h3_end = h3_matches[j + 1].start() | |
| else: | |
| h3_end = len(h2_section_content) | |
| h3_content = h2_section_content[h3_start:h3_end].strip() | |
| # Use combined title for context | |
| combined_title = f"{h2_title}: {h3_title}" | |
| sections.append({"title": combined_title, "content": h3_content}) | |
| else: | |
| # No subsections, use the whole ## section | |
| section_content = h2_section_content.strip() | |
| sections.append({"title": h2_title, "content": section_content}) | |
| return sections | |
| def _determine_doc_type(self, metadata: dict[str, Any]) -> str: | |
| """ | |
| Determine document type from metadata fields. | |
| Args: | |
| metadata: Parsed frontmatter metadata | |
| Returns: | |
| Document type: "lesson", "exercise", or "unknown" | |
| """ | |
| if "lesson_number" in metadata and "lesson_name" in metadata: | |
| return "lesson" | |
| if "exercise_type" in metadata: | |
| return "exercise" | |
| return "unknown" | |
| def chunk_content(self, content: str, max_chunk_size: int = DEFAULT_CHUNK_SIZE) -> list[str]: | |
| """ | |
| Split long content into chunks at paragraph boundaries. | |
| Args: | |
| content: Text content to chunk | |
| max_chunk_size: Maximum characters per chunk | |
| Returns: | |
| List of content chunks | |
| """ | |
| if len(content) <= max_chunk_size: | |
| return [content] | |
| paragraphs = content.split(PARAGRAPH_SEPARATOR) | |
| return self._build_chunks_from_paragraphs(paragraphs, max_chunk_size) | |
| def _build_chunks_from_paragraphs( | |
| self, paragraphs: list[str], max_chunk_size: int | |
| ) -> list[str]: | |
| """Build chunks from paragraphs respecting max size.""" | |
| chunks = [] | |
| current_chunk = "" | |
| for paragraph in paragraphs: | |
| if len(paragraph) > max_chunk_size: | |
| if current_chunk: | |
| chunks.append(current_chunk.strip()) | |
| current_chunk = "" | |
| chunks.extend(self._split_large_paragraph(paragraph, max_chunk_size)) | |
| continue | |
| if len(current_chunk) + len(paragraph) + PARAGRAPH_SEPARATOR_LENGTH > max_chunk_size: | |
| if current_chunk: | |
| chunks.append(current_chunk.strip()) | |
| current_chunk = paragraph | |
| else: | |
| current_chunk = ( | |
| paragraph | |
| if not current_chunk | |
| else f"{current_chunk}{PARAGRAPH_SEPARATOR}{paragraph}" | |
| ) | |
| if current_chunk: | |
| chunks.append(current_chunk.strip()) | |
| return chunks | |
| def _split_large_paragraph(self, paragraph: str, max_chunk_size: int) -> list[str]: | |
| """Split a single large paragraph into fixed-size chunks.""" | |
| return [paragraph[i : i + max_chunk_size] for i in range(0, len(paragraph), max_chunk_size)] | |
| def parse_file(self, file_path: Path) -> list[dict[str, Any]]: | |
| """ | |
| Parse a markdown file into chunks with metadata. | |
| Args: | |
| file_path: Path to markdown file | |
| Returns: | |
| List of dicts with {text, metadata} | |
| Raises: | |
| FileNotFoundError: If file doesn't exist | |
| """ | |
| if not file_path.exists(): | |
| raise FileNotFoundError(f"File not found: {file_path}") | |
| content = file_path.read_text(encoding="utf-8") | |
| # Extract metadata from frontmatter | |
| metadata = self.parse_frontmatter(content) | |
| # Add source file and document type to metadata | |
| metadata["source_file"] = file_path.name | |
| metadata["doc_type"] = self._determine_doc_type(metadata) | |
| # Extract sections | |
| sections = self.extract_sections(content) | |
| # Create chunks from sections | |
| chunks = [] | |
| for section in sections: | |
| section_text = f"{section['title']}\n\n{section['content']}" | |
| # Add section-specific metadata | |
| section_metadata = metadata.copy() | |
| section_metadata["section_title"] = section["title"] | |
| chunks.append({"text": section_text, "metadata": section_metadata}) | |
| return chunks | |
| def parse_directory(self, directory: Path, recursive: bool = True) -> list[dict[str, Any]]: | |
| """ | |
| Parse all markdown files in a directory. | |
| Args: | |
| directory: Path to directory containing markdown files | |
| recursive: If True, search subdirectories recursively | |
| Returns: | |
| List of all chunks from all files | |
| """ | |
| all_chunks = [] | |
| # Use rglob for recursive search, glob for non-recursive | |
| file_paths = directory.rglob("*.md") if recursive else directory.glob("*.md") | |
| for file_path in file_paths: | |
| try: | |
| chunks = self.parse_file(file_path) | |
| all_chunks.extend(chunks) | |
| except (OSError, UnicodeDecodeError): | |
| # Skip files with I/O or encoding errors | |
| # TODO: Add logging to track parse failures for debugging | |
| # logger.warning(f"Failed to parse {file_path}: {e}") | |
| continue | |
| return all_chunks | |