ReMe/reme/steps/file_io/edit.py

"""Find-and-replace text in a markdown file body (front matter is preserved)."""

import re

import frontmatter
import yaml

from ._base import _FileIOStep
from ._file_io import get_path_lock, read_file_safe, write_file_safe
from ._path import _check_path_permission, NON_MD_WARNING, gate_md, resolve_path
from ...components import R


def _replace_normalized_newlines(body: str, old: str, new: str) -> tuple[str, int]:
    """Find literal LF text, mapping its spans back without normalizing untouched text."""
    normalized = body.replace("\r\n", "\n").replace("\r", "\n")
    position = normalized.find(old)
    if position < 0:
        return body, 0

    # Only CRLF pairs change offsets. Consume them once as match boundaries advance;
    # no per-character offset array or regex search for the user-supplied needle.
    crlf_starts = (match.start() for match in re.finditer("\r\n", body))
    next_crlf = next(crlf_starts, len(body))
    removed = 0
    source_cursor = 0
    count = 0
    parts = []
    while position >= 0:
        source_start = position + removed
        while next_crlf < source_start:
            removed += 1
            source_start += 1
            next_crlf = next(crlf_starts, len(body))

        end = position + len(old)
        source_end = end + removed
        while next_crlf < source_end:
            removed += 1
            source_end += 1
            next_crlf = next(crlf_starts, len(body))

        parts.extend((body[source_cursor:source_start], new))
        source_cursor = source_end
        count += 1
        position = normalized.find(old, end)

    parts.append(body[source_cursor:])
    return "".join(parts), count


@R.register("edit_step")
class EditStep(_FileIOStep):
    """Replace every occurrence of ``old`` with ``new`` inside the file body.

    The YAML front matter block (if any) is parsed out, kept verbatim and
    re-emitted unchanged — matches that fall inside front matter are ignored,
    so a typo in `old` cannot corrupt structured metadata.

    Permission: honors the request-scoped ``_allowed_paths`` constraint
    injected by the server into the RuntimeContext.

    Concurrency: in-process per-path ``asyncio.Lock`` serializes the
    read-modify-write cycle against the same file (multi-worker / multi-
    process safety is out of scope)."""

    async def execute(self):  # pylint: disable=too-many-return-statements
        assert self.context is not None
        raw = str(self.context.get("path") or "")
        old = self.context.get("old")
        new = self.context.get("new")

        if old is None or str(old) == "":
            self._fail("`old` is required and must be non-empty")
            return None
        if new is None:
            self._fail("`new` is required")
            return None
        old_str = str(old)
        new_str = str(new)

        target, err = resolve_path(self.workspace_path, raw)
        if err:
            self._fail(err)
            return None

        target, is_md = gate_md(target)

        if not _check_path_permission(self.workspace_path, target, self.context.get("_allowed_paths")):
            self._fail("no permission to edit this file", path=str(target))
            return None

        lock = await get_path_lock(target)
        async with lock:
            if not target.exists():
                self._fail(f"file {target} does not exist", path=str(target))
                return None
            if not target.is_file():
                self._fail(f"path {target} is not a file", path=str(target))
                return None

            try:
                raw_text, encoding = await read_file_safe(target)
            except Exception as e:  # pylint: disable=broad-except
                self._fail(f"read failed: {e}", path=str(target))
                return None

            # Markdown: parse frontmatter and operate on body only. Non-markdown:
            # there's no frontmatter convention, so operate on the full text.
            if is_md:
                try:
                    post = frontmatter.loads(raw_text)
                except yaml.YAMLError as exc:
                    self._fail(f"failed to parse frontmatter in {target}: {exc}", path=str(target))
                    return None
                body = post.content
                not_found_msg = (
                    f"text to replace was not found in the body of {target} (front matter is excluded from edit)"
                )
            else:
                post = None
                body = raw_text
                not_found_msg = f"text to replace was not found in {target}"

            if "\r" in body and "\n" in old_str and "\r" not in old_str:
                new_body, count = _replace_normalized_newlines(body, old_str, new_str)
            else:
                count = body.count(old_str)
                new_body = body.replace(old_str, new_str)

            if count == 0:
                self._fail(not_found_msg, path=str(target))
                return None

            if is_md and post is not None:
                post.content = new_body
                # Re-serialize: keep front matter when present, otherwise emit body alone
                # so we don't introduce an empty `---\n---\n` block.
                new_text = frontmatter.dumps(post) if post.metadata else post.content
                if not new_text.endswith("\n"):
                    new_text += "\n"
            else:
                new_text = new_body

            # Preserve the file's original encoding (returned by read_file_safe above)
            # so edits don't silently re-encode non-UTF-8 files (e.g. GBK CSV) to UTF-8.
            try:
                await write_file_safe(target, new_text, encoding=encoding)
            except Exception as e:  # pylint: disable=broad-except
                self._fail(f"write failed: {e}", path=str(target))
                return None

        self.context.response.success = True
        answer = f"Replaced {count} occurrence(s) in {target}"
        if not is_md:
            answer = f"{answer} [system notice: {NON_MD_WARNING}]"
        self.context.response.answer = answer
        self.logger.info(f"[{self.name}] edited path={target} count={count} is_md={is_md}")
        return self.context.response