basicmachines-co-basic-memory/src/basic_memory/file_utils.py

"""Utilities for file operations."""

import asyncio
import hashlib
import shlex
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
import re
from typing import TYPE_CHECKING, Any, Dict, Optional, Union

import aiofiles
import yaml
import frontmatter
from loguru import logger

from basic_memory.utils import FilePath

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


@dataclass
class FileMetadata:
    """File metadata for cloud-compatible file operations.

    This dataclass provides a cloud-agnostic way to represent file metadata,
    enabling S3FileService to return metadata from head_object responses
    instead of mock stat_result with zeros.
    """

    size: int
    created_at: datetime
    modified_at: datetime


class FileError(Exception):
    """Base exception for file operations."""

    pass


class FileWriteError(FileError):
    """Raised when file operations fail."""

    pass


class ParseError(FileError):
    """Raised when parsing file content fails."""

    pass


async def compute_checksum(content: Union[str, bytes]) -> str:
    """
    Compute SHA-256 checksum of content.

    Args:
        content: Content to hash (either text string or bytes)

    Returns:
        SHA-256 hex digest

    Raises:
        FileError: If checksum computation fails
    """
    try:
        if isinstance(content, str):
            content = content.encode()
        return hashlib.sha256(content).hexdigest()
    except Exception as e:  # pragma: no cover
        logger.error(f"Failed to compute checksum: {e}")
        raise FileError(f"Failed to compute checksum: {e}")


# UTF-8 BOM character that can appear at the start of files
UTF8_BOM = "\ufeff"


def strip_bom(content: str) -> str:
    """Strip UTF-8 BOM from the start of content if present.

    BOM (Byte Order Mark) characters can be present in files created on Windows
    or copied from certain sources. They should be stripped before processing
    frontmatter. See issue #452.

    Args:
        content: Content that may start with BOM

    Returns:
        Content with BOM removed if present
    """
    if content and content.startswith(UTF8_BOM):
        return content[1:]
    return content


async def write_file_atomic(path: FilePath, content: str) -> None:
    """
    Write file with atomic operation using temporary file.

    Uses aiofiles for true async I/O (non-blocking).

    Args:
        path: Target file path (Path or string)
        content: Content to write

    Raises:
        FileWriteError: If write operation fails
    """
    # Convert string to Path if needed
    path_obj = Path(path) if isinstance(path, str) else path
    temp_path = path_obj.with_suffix(".tmp")

    try:
        # Trigger: callers hand us normalized Python text, but the final bytes are allowed
        #          to use the host platform's native newline convention during the write.
        # Why: preserving CRLF on Windows keeps local files aligned with editors like
        #      Obsidian, while FileService now hashes the persisted file bytes instead of
        #      the pre-write string.
        # Outcome: this async write stays editor-friendly across platforms without
        #          reintroducing checksum drift in sync or move detection.
        async with aiofiles.open(temp_path, mode="w", encoding="utf-8") as f:
            await f.write(content)

        # Atomic rename (this is fast, doesn't need async)
        temp_path.replace(path_obj)
        logger.debug("Wrote file atomically", path=str(path_obj), content_length=len(content))
    except Exception as e:  # pragma: no cover
        temp_path.unlink(missing_ok=True)
        logger.error("Failed to write file", path=str(path_obj), error=str(e))
        raise FileWriteError(f"Failed to write file {path}: {e}")


async def format_markdown_builtin(path: Path) -> Optional[str]:
    """
    Format a markdown file using the built-in mdformat formatter.

    Uses mdformat with GFM (GitHub Flavored Markdown) support for consistent
    formatting without requiring Node.js or external tools.

    Args:
        path: Path to the markdown file to format

    Returns:
        Formatted content if successful, None if formatting failed.
    """
    try:
        import mdformat
    except ImportError:  # pragma: no cover
        logger.warning(
            "mdformat not installed, skipping built-in formatting",
            path=str(path),
        )
        return None

    try:
        # Read original content
        async with aiofiles.open(path, mode="r", encoding="utf-8") as f:
            content = await f.read()

        # Format using mdformat with GFM and frontmatter extensions
        # mdformat is synchronous, so we run it in a thread executor
        loop = asyncio.get_event_loop()
        formatted_content = await loop.run_in_executor(
            None,
            lambda: mdformat.text(
                content,
                extensions={"gfm", "frontmatter"},  # GFM + YAML frontmatter support
                options={"wrap": "no"},  # Don't wrap lines
            ),
        )

        # Only write if content changed
        if formatted_content != content:
            # Trigger: mdformat may rewrite markdown content, then the host platform
            #          decides the newline bytes for the follow-up async text write.
            # Why: we want formatter output to preserve native newlines instead of
            #      forcing LF, and the authoritative checksum comes from rereading the
            #      stored file bytes later in FileService.
            # Outcome: formatting remains compatible with local editors on Windows while
            #          checksum-based sync logic stays anchored to on-disk bytes.
            async with aiofiles.open(path, mode="w", encoding="utf-8") as f:
                await f.write(formatted_content)

        logger.debug(
            "Formatted file with mdformat",
            path=str(path),
            changed=formatted_content != content,
        )
        return formatted_content

    except Exception as e:  # pragma: no cover
        logger.warning(
            "mdformat formatting failed",
            path=str(path),
            error=str(e),
        )
        return None


async def format_file(
    path: Path,
    config: "BasicMemoryConfig",
    is_markdown: bool = False,
) -> Optional[str]:
    """
    Format a file using configured formatter.

    By default, uses the built-in mdformat formatter for markdown files (pure Python,
    no Node.js required). External formatters like Prettier can be configured via
    formatter_command or per-extension formatters.

    Args:
        path: File to format
        config: Configuration with formatter settings
        is_markdown: Whether this is a markdown file (caller should use FileService.is_markdown)

    Returns:
        Formatted content if successful, None if formatting was skipped or failed.
        Failures are logged as warnings but don't raise exceptions.
    """
    if not config.format_on_save:
        return None

    extension = path.suffix.lstrip(".")
    formatter = config.formatters.get(extension) or config.formatter_command

    # Use built-in mdformat for markdown files when no external formatter configured
    if not formatter:
        if is_markdown:
            return await format_markdown_builtin(path)
        else:
            logger.debug("No formatter configured for extension", extension=extension)
            return None

    # Use external formatter
    # Replace {file} placeholder with the actual path
    cmd = formatter.replace("{file}", str(path))

    try:
        # Parse command into args list for safer execution (no shell=True)
        args = shlex.split(cmd)

        proc = await asyncio.create_subprocess_exec(
            *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        try:
            stdout, stderr = await asyncio.wait_for(
                proc.communicate(),
                timeout=config.formatter_timeout,
            )
        except asyncio.TimeoutError:
            proc.kill()
            await proc.wait()
            logger.warning(
                "Formatter timed out",
                path=str(path),
                timeout=config.formatter_timeout,
            )
            return None

        if proc.returncode != 0:
            logger.warning(
                "Formatter exited with non-zero status",
                path=str(path),
                returncode=proc.returncode,
                stderr=stderr.decode("utf-8", errors="replace") if stderr else "",
            )
            # Still try to read the file - formatter may have partially worked
            # or the file may be unchanged

        # Read formatted content
        async with aiofiles.open(path, mode="r", encoding="utf-8") as f:
            formatted_content = await f.read()

        logger.debug(
            "Formatted file successfully",
            path=str(path),
            formatter=args[0] if args else formatter,
        )
        return formatted_content

    except FileNotFoundError:
        # Formatter executable not found
        logger.warning(
            "Formatter executable not found",
            command=cmd.split()[0] if cmd else "",
            path=str(path),
        )
        return None
    except Exception as e:  # pragma: no cover
        logger.warning(
            "Formatter failed",
            path=str(path),
            error=str(e),
        )
        return None


def has_frontmatter(content: str) -> bool:
    """
    Check if content contains valid YAML frontmatter.

    Args:
        content: Content to check

    Returns:
        True if content has valid frontmatter markers (---), False otherwise
    """
    if not content:
        return False

    # Strip BOM before checking for frontmatter markers
    content = strip_bom(content).strip()
    if not content.startswith("---"):
        return False

    return "---" in content[3:]


def parse_frontmatter(content: str) -> Dict[str, Any]:
    """
    Parse YAML frontmatter from content.

    Args:
        content: Content with YAML frontmatter

    Returns:
        Dictionary of frontmatter values

    Raises:
        ParseError: If frontmatter is invalid or parsing fails
    """
    try:
        # Strip BOM before parsing frontmatter
        content = strip_bom(content)
        if not content.strip().startswith("---"):
            raise ParseError("Content has no frontmatter")

        # Split on first two occurrences of ---
        parts = content.split("---", 2)
        if len(parts) < 3:
            raise ParseError("Invalid frontmatter format")

        # Parse YAML
        try:
            frontmatter = yaml.safe_load(parts[1])
            # Handle empty frontmatter (None from yaml.safe_load)
            if frontmatter is None:
                return {}
            if not isinstance(frontmatter, dict):
                raise ParseError("Frontmatter must be a YAML dictionary")
            return frontmatter

        except yaml.YAMLError as e:
            raise ParseError(f"Invalid YAML in frontmatter: {e}")

    except Exception as e:  # pragma: no cover
        if not isinstance(e, ParseError):
            logger.error(f"Failed to parse frontmatter: {e}")
            raise ParseError(f"Failed to parse frontmatter: {e}")
        raise


def remove_frontmatter(content: str) -> str:
    """
    Remove YAML frontmatter from content.

    Args:
        content: Content with frontmatter

    Returns:
        Content with frontmatter removed, or original content if no frontmatter

    Raises:
        ParseError: If content starts with frontmatter marker but is malformed
    """
    # Strip BOM before processing
    content = strip_bom(content).strip()

    # Return as-is if no frontmatter marker
    if not content.startswith("---"):
        return content

    # Split on first two occurrences of ---
    parts = content.split("---", 2)
    if len(parts) < 3:
        raise ParseError("Invalid frontmatter format")

    return parts[2].strip()


def dump_frontmatter(post: frontmatter.Post) -> str:
    """
    Serialize frontmatter.Post to markdown with Obsidian-compatible YAML format.

    This function ensures that:
    1. Tags are formatted as YAML lists instead of JSON arrays
    2. String values are properly quoted to handle special characters (colons, etc.)

    Good (Obsidian compatible):
    ---
    title: "L2 Governance Core (Split: Core)"
    tags:
    - system
    - overview
    - reference
    ---

    Bad (causes parsing errors):
    ---
    title: L2 Governance Core (Split: Core)  # Unquoted colon breaks YAML
    tags: ["system", "overview", "reference"]
    ---

    Args:
        post: frontmatter.Post object to serialize

    Returns:
        String containing markdown with properly formatted YAML frontmatter
    """
    if not post.metadata:
        # No frontmatter, just return content
        return post.content

    # Serialize YAML with block style for lists
    # SafeDumper automatically quotes values with special characters (colons, etc.)
    yaml_str = yaml.dump(
        post.metadata,
        sort_keys=False,
        allow_unicode=True,
        default_flow_style=False,
        Dumper=yaml.SafeDumper,
    )

    # Construct the final markdown with frontmatter
    if post.content:
        return f"---\n{yaml_str}---\n\n{post.content}"
    else:
        return f"---\n{yaml_str}---\n"


def sanitize_for_filename(text: str, replacement: str = "-") -> str:
    """
    Sanitize string to be safe for use as a note title
    Replaces path separators and other problematic characters
    with hyphens.
    """
    # replace both POSIX and Windows path separators
    text = re.sub(r"[/\\]", replacement, text)

    # replace some other problematic chars
    text = re.sub(r'[<>:"|?*]', replacement, text)

    # compress multiple, repeated replacements
    text = re.sub(f"{re.escape(replacement)}+", replacement, text)

    # Strip trailing periods — they cause "hi-everyone..md" double-dot filenames
    # when ".md" is appended, which triggers path traversal false positives.
    # Trailing periods are also invalid on Windows filesystems.
    text = text.strip(".")

    return text.strip(replacement)


def sanitize_for_directory(directory: str) -> str:
    """
    Sanitize directory path to be safe for use in file system paths.
    Removes leading/trailing whitespace, compresses multiple slashes,
    and removes special characters except for /, -, and _.
    """
    if not directory:
        return ""

    sanitized = directory.strip()

    if sanitized.startswith("./"):
        sanitized = sanitized[2:]

    # ensure no special characters (except for a few that are allowed)
    sanitized = "".join(
        c for c in sanitized if c.isalnum() or c in (".", " ", "-", "_", "\\", "/")
    ).rstrip()

    # compress multiple, repeated instances of path separators
    sanitized = re.sub(r"[\\/]+", "/", sanitized)

    # trim any leading/trailing path separators
    sanitized = sanitized.strip("\\/")

    return sanitized