basicmachines-co-basic-memory/src/basic_memory/markdown/markdown_processor.py

from pathlib import Path
from typing import TYPE_CHECKING, Optional
from collections import OrderedDict

from frontmatter import Post
from loguru import logger


from basic_memory import file_utils
from basic_memory.file_utils import dump_frontmatter
from basic_memory.markdown.entity_parser import EntityParser
from basic_memory.markdown.schemas import EntityMarkdown, Observation, Relation

if TYPE_CHECKING:  # pragma: no cover
    from basic_memory.config import BasicMemoryConfig


class DirtyFileError(Exception):
    """Raised when attempting to write to a file that has been modified."""

    pass


class MarkdownProcessor:
    """Process markdown files while preserving content and structure.

    used only for import

    This class handles the file I/O aspects of our markdown processing. It:
    1. Uses EntityParser for reading/parsing files into our schema
    2. Handles writing files with proper frontmatter
    3. Formats structured sections (observations/relations) consistently
    4. Preserves user content exactly as written
    5. Performs atomic writes using temp files

    It does NOT:
    1. Modify the schema directly (that's done by services)
    2. Handle in-place updates (everything is read->modify->write)
    3. Track schema changes (that's done by the database)
    """

    def __init__(
        self,
        entity_parser: EntityParser,
        app_config: Optional["BasicMemoryConfig"] = None,
    ):
        """Initialize processor with parser and optional config."""
        self.entity_parser = entity_parser
        self.app_config = app_config

    async def read_file(self, path: Path) -> EntityMarkdown:
        """Read and parse file into EntityMarkdown schema.

        This is step 1 of our read->modify->write pattern.
        We use EntityParser to handle all the markdown parsing.
        """
        return await self.entity_parser.parse_file(path)

    async def write_file(
        self,
        path: Path,
        markdown: EntityMarkdown,
        expected_checksum: Optional[str] = None,
    ) -> str:
        """Write EntityMarkdown schema back to file.

        This is step 3 of our read->modify->write pattern.
        The entire file is rewritten atomically on each update.

        File Structure:
        ---
        frontmatter fields
        ---
        user content area (preserved exactly)

        ## Observations (if any)
        formatted observations

        ## Relations (if any)
        formatted relations

        Args:
            path: Where to write the file
            markdown: Complete schema to write
            expected_checksum: If provided, verify file hasn't changed

        Returns:
            Checksum of written file

        Raises:
            DirtyFileError: If file has been modified (when expected_checksum provided)
        """
        # Dirty check if needed
        if expected_checksum is not None:
            current_content = path.read_text(encoding="utf-8")
            current_checksum = await file_utils.compute_checksum(current_content)
            if current_checksum != expected_checksum:
                raise DirtyFileError(f"File {path} has been modified")

        # Convert frontmatter to dict
        frontmatter_dict = OrderedDict()
        frontmatter_dict["title"] = markdown.frontmatter.title
        frontmatter_dict["type"] = markdown.frontmatter.type
        frontmatter_dict["permalink"] = markdown.frontmatter.permalink

        metadata = markdown.frontmatter.metadata or {}
        for k, v in metadata.items():
            frontmatter_dict[k] = v

        # Start with user content (or minimal title for new files)
        content = markdown.content or f"# {markdown.frontmatter.title}\n"

        # Add structured sections with proper spacing
        content = content.rstrip()  # Remove trailing whitespace

        # add a blank line if we have semantic content
        if markdown.observations or markdown.relations:
            content += "\n"

        if markdown.observations:
            content += self.format_observations(markdown.observations)
        if markdown.relations:
            content += self.format_relations(markdown.relations)

        # Create Post object for frontmatter
        post = Post(content, **frontmatter_dict)
        final_content = dump_frontmatter(post)

        logger.debug(f"writing file {path} with content:\n{final_content}")

        # Write atomically and return checksum of updated file
        path.parent.mkdir(parents=True, exist_ok=True)
        await file_utils.write_file_atomic(path, final_content)

        # Format file if configured (MarkdownProcessor always handles markdown files)
        content_for_checksum = final_content
        if self.app_config:
            formatted_content = await file_utils.format_file(  # pragma: no cover
                path, self.app_config, is_markdown=True
            )
            if formatted_content is not None:  # pragma: no cover
                content_for_checksum = formatted_content  # pragma: no cover

        return await file_utils.compute_checksum(content_for_checksum)

    def to_markdown_string(self, markdown: EntityMarkdown) -> str:
        """Convert EntityMarkdown to markdown string with frontmatter.

        This method handles serialization only - it does not write to files.
        Use FileService.write_file() to persist the output.

        This enables cloud environments to override file operations via
        dependency injection while reusing the serialization logic.

        Args:
            markdown: EntityMarkdown schema to serialize

        Returns:
            Complete markdown string with frontmatter, content, and structured sections
        """
        # Convert frontmatter to dict
        frontmatter_dict = OrderedDict()
        frontmatter_dict["title"] = markdown.frontmatter.title
        frontmatter_dict["type"] = markdown.frontmatter.type
        frontmatter_dict["permalink"] = markdown.frontmatter.permalink

        metadata = markdown.frontmatter.metadata or {}
        for k, v in metadata.items():
            frontmatter_dict[k] = v

        # Start with user content (or minimal title for new files)
        content = markdown.content or f"# {markdown.frontmatter.title}\n"

        # Add structured sections with proper spacing
        content = content.rstrip()  # Remove trailing whitespace

        # Add a blank line if we have semantic content
        if markdown.observations or markdown.relations:
            content += "\n"

        if markdown.observations:
            content += self.format_observations(markdown.observations)
        if markdown.relations:
            content += self.format_relations(markdown.relations)

        # Create Post object for frontmatter
        post = Post(content, **frontmatter_dict)
        return dump_frontmatter(post)

    def format_observations(self, observations: list[Observation]) -> str:
        """Format observations section in standard way.

        Format: - [category] content #tag1 #tag2 (context)
        """
        lines = [f"{obs}" for obs in observations]
        return "\n".join(lines) + "\n"

    def format_relations(self, relations: list[Relation]) -> str:
        """Format relations section in standard way.

        Format: - relation_type [[target]] (context)
        """
        lines = [f"{rel}" for rel in relations]
        return "\n".join(lines) + "\n"