breaking up parsing logic for reuse

src/basic_memory/markdown/base_parser.py

@@ -0,0 +1,141 @@
+"""Base parser for markdown files with frontmatter."""
+from abc import ABC, abstractmethod
+from pathlib import Path
+from typing import List, TypeVar, Generic, Optional, Dict, Any, Tuple
+
+from loguru import logger
+
+from basic_memory.utils.file_utils import parse_frontmatter, ParseError, FileError
+
+T = TypeVar('T')  # The parsed document type
+
+
+class MarkdownParser(ABC, Generic[T]):
+    """
+    Base parser for markdown files with frontmatter.
+    
+    Every supported document type should subclass this with their specific
+    parsing and document creation logic.
+    """
+
+    async def parse_file(self, path: Path, encoding: str = "utf-8") -> T:
+        """
+        Parse a markdown file with frontmatter.
+        
+        Args:
+            path: Path to markdown file
+            encoding: File encoding to use
+            
+        Returns:
+            Parsed document of type T
+            
+        Raises:
+            FileError: If file cannot be read
+            ParseError: If content cannot be parsed
+        """
+        if not path.exists():
+            raise FileError(f"File does not exist: {path}")
+
+        try:
+            content = path.read_text(encoding=encoding)
+            return await self.parse_content_str(content)
+            
+        except UnicodeError as e:
+            if encoding == "utf-8":
+                return await self.parse_file(path, encoding="utf-16")
+            raise ParseError(f"Failed to decode {path}: {str(e)}") from e
+        except Exception as e:
+            if not isinstance(e, (FileError, ParseError)):
+                logger.error(f"Failed to parse {path}: {e}")
+                raise ParseError(f"Failed to parse {path}: {str(e)}") from e
+            raise
+
+    async def parse_content_str(self, content: str) -> T:
+        """
+        Parse raw content string into document.
+        
+        Args:
+            content: Raw file content
+            
+        Returns:
+            Parsed document
+            
+        Raises:
+            ParseError: If parsing fails
+        """
+        try:
+            # Split into frontmatter and content
+            frontmatter, remaining = await parse_frontmatter(content)
+            
+            # Parse sections with concrete implementation
+            parsed_frontmatter = await self.parse_frontmatter(frontmatter)
+            parsed_content = await self.parse_content(remaining)
+            parsed_metadata = await self.parse_metadata(frontmatter.get("metadata"))
+            
+            # Create document from parts
+            return await self.create_document(parsed_frontmatter, parsed_content, parsed_metadata)
+            
+        except Exception as e:
+            if not isinstance(e, ParseError):
+                logger.error(f"Failed to parse content: {e}")
+                raise ParseError(f"Failed to parse content: {str(e)}") from e
+            raise
+
+    @abstractmethod
+    async def parse_frontmatter(self, frontmatter: Dict[str, Any]) -> Any:
+        """
+        Parse frontmatter section.
+        
+        Args:
+            frontmatter: Parsed YAML frontmatter
+            
+        Returns:
+            Parsed frontmatter in format needed by document
+        """
+        pass
+
+    @abstractmethod 
+    async def parse_content(self, content: str) -> Any:
+        """
+        Parse main content section.
+        
+        Args:
+            content: Content section text
+            
+        Returns:
+            Parsed content in format needed by document
+        """
+        pass
+
+    @abstractmethod
+    async def parse_metadata(self, metadata: Optional[Dict[str, Any]]) -> Any:
+        """
+        Parse optional metadata section.
+        
+        Args:
+            metadata: Optional metadata dictionary
+            
+        Returns:
+            Parsed metadata in format needed by document
+        """
+        pass
+
+    @abstractmethod
+    async def create_document(
+        self, 
+        frontmatter: Any, 
+        content: Any, 
+        metadata: Optional[Any]
+    ) -> T:
+        """
+        Create document from parsed sections.
+        
+        Args:
+            frontmatter: Parsed frontmatter
+            content: Parsed content
+            metadata: Optional parsed metadata
+            
+        Returns:
+            Complete document of type T
+        """
+        pass

src/basic_memory/markdown/parser.py

@@ -1,9 +1,11 @@
 """Parser for Basic Memory entity markdown files."""
 
-import logging
 from pathlib import Path
+from typing import Dict, Any, Optional
 
-from basic_memory.markdown.exceptions import ParseError
+from loguru import logger
+
+from basic_memory.markdown.base_parser import MarkdownParser, ParseError
 from basic_memory.markdown.schemas import (
     Entity,
     EntityFrontmatter,
@@ -11,49 +13,95 @@ from basic_memory.markdown.schemas import (
     EntityMetadata,
 )
 
-logging.basicConfig(level=logging.DEBUG)
-logger = logging.getLogger(__name__)
 
+class EntityParser(MarkdownParser[Entity]):
+    """
+    Parser for entity markdown files.
+    
+    Parses files in the Knowledge Format, which includes:
+    - YAML frontmatter
+    - Markdown content with observations
+    - Optional metadata section
+    """
 
-def debug_sections(text):
-    """Debug helper to show section contents."""
-    # Split on triple-dash and newline combinations to be more lenient
-    sections = [s.strip() for s in text.replace("\r\n", "\n").split("---")]
-    logger.debug("File sections:")
-    for i, section in enumerate(sections):
-        logger.debug(f"\n=== Section {i} ===\n{section.strip()}\n")
-    return [s for s in sections if s.strip()]  # Remove empty sections
-
-
-class EntityParser:
-    """Parser for entity markdown files."""
-
-    def parse_file(self, path: Path, encoding: str = "utf-8") -> Entity:
-        """Parse an entity markdown file."""
-        if not path.exists():
-            raise ParseError(f"File does not exist: {path}")
-
-        try:
-            # Read file content and split sections
-            with open(path, "r", encoding=encoding) as f:
-                raw_content = f.read()
-            sections = debug_sections(raw_content)
-
-            if len(sections) < 2:  # Need at least frontmatter and content
-                raise ParseError("Missing required document sections")
-
-            # Parse each section using schema methods
-            frontmatter = EntityFrontmatter.from_text(sections[0])
-            content = EntityContent.from_markdown(sections[1])
+    async def parse_frontmatter(self, frontmatter: Dict[str, Any]) -> EntityFrontmatter:
+        """
+        Parse entity frontmatter.
+        
+        Args:
+            frontmatter: Parsed YAML frontmatter
             
-            # Handle optional metadata section
-            metadata = EntityMetadata.from_text(sections[2] if len(sections) > 2 else "")
-
-            return Entity(frontmatter=frontmatter, content=content, metadata=metadata)
-
-        except UnicodeError as e:
-            if encoding == "utf-8":
-                return self.parse_file(path, encoding="utf-16")
-            raise ParseError(f"Failed to parse {path}: {str(e)}") from e
+        Returns:
+            Parsed EntityFrontmatter
+            
+        Raises:
+            ParseError: If frontmatter doesn't match schema
+        """
+        try:
+            return EntityFrontmatter(**frontmatter)
         except Exception as e:
-            raise ParseError(f"Failed to parse {path}: {str(e)}") from e
+            logger.error(f"Invalid entity frontmatter: {e}")
+            raise ParseError(f"Invalid entity frontmatter: {str(e)}") from e
+
+    async def parse_content(self, content: str) -> EntityContent:
+        """
+        Parse entity content section.
+        
+        Args:
+            content: Content section text
+            
+        Returns:
+            Parsed EntityContent
+            
+        Raises:
+            ParseError: If content doesn't match schema
+        """
+        try:
+            return EntityContent.from_markdown(content)
+        except Exception as e:
+            logger.error(f"Invalid entity content: {e}")
+            raise ParseError(f"Invalid entity content: {str(e)}") from e
+
+    async def parse_metadata(self, metadata: Optional[Dict[str, Any]]) -> EntityMetadata:
+        """
+        Parse entity metadata section.
+        
+        Args:
+            metadata: Optional metadata dictionary
+            
+        Returns:
+            Parsed EntityMetadata
+            
+        Raises:
+            ParseError: If metadata doesn't match schema
+        """
+        try:
+            if not metadata:
+                return EntityMetadata()
+            return EntityMetadata(**metadata)
+        except Exception as e:
+            logger.error(f"Invalid entity metadata: {e}")
+            raise ParseError(f"Invalid entity metadata: {str(e)}") from e
+
+    async def create_document(
+        self, 
+        frontmatter: EntityFrontmatter, 
+        content: EntityContent, 
+        metadata: EntityMetadata
+    ) -> Entity:
+        """
+        Create entity from parsed sections.
+        
+        Args:
+            frontmatter: Parsed frontmatter
+            content: Parsed content
+            metadata: Parsed metadata
+            
+        Returns:
+            Complete entity
+        """
+        return Entity(
+            frontmatter=frontmatter,
+            content=content,
+            metadata=metadata
+        )