mirror of
https://github.com/agentscope-ai/ReMe.git
synced 2026-08-28 05:25:04 +00:00
f34dcdb09b
* feat(read): add white/black path prefix permission filtering to ReadStep * feat: add PrefixCheck mixin for path-prefix permission in file I/O steps * feat: add injected_job_kwargs mechanism and refine path-prefix permission * refactor(file_io): consolidate prefix_check into _path module
123 lines
4.9 KiB
Python
123 lines
4.9 KiB
Python
"""Find-and-replace text in a markdown file body (front matter is preserved)."""
|
|
|
|
import frontmatter
|
|
import yaml
|
|
|
|
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 ..base_step import BaseStep
|
|
from ...components import R
|
|
|
|
|
|
@R.register("edit_step")
|
|
class EditStep(BaseStep):
|
|
"""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)."""
|
|
|
|
def _fail(self, message: str, **meta) -> None:
|
|
assert self.context is not None
|
|
self.context.response.success = False
|
|
self.context.response.answer = f"Error: {message}"
|
|
if meta:
|
|
self.context.response.metadata.update(meta)
|
|
|
|
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 old_str not in body:
|
|
self._fail(not_found_msg, path=str(target))
|
|
return None
|
|
|
|
count = body.count(old_str)
|
|
new_body = body.replace(old_str, new_str)
|
|
|
|
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
|