feat: 支持文件编辑的模糊匹配 && ChatUI工具调用结果美化

astrbot/core/tools/computer_tools/fs.py

@@ -33,6 +33,7 @@
 - In sandbox runtime, relative paths are passed through unchanged.
 """
 
+import base64
 import os
 import uuid
 from dataclasses import dataclass, field
@@ -55,6 +56,12 @@
 
 from ..registry import builtin_tool
 from . import util as computer_util
+from .edit_engine import (
+    EditResult,
+    bytes_edit_file,
+    edit_file,
+    get_file_lock,
+)
 from .util import (
     check_admin_permission,
     is_local_runtime,
@@ -368,30 +375,161 @@ async def call(
             return f"Error writing file: {exc}"
 
 
+async def _sandbox_read_bytes(sb, path: str) -> bytes:
+    """Read a file in binary mode inside the sandbox via sb.python.exec.
+
+    Returns the raw bytes so that original line endings (CRLF/LF) are preserved.
+    Raises IOError on read failure.
+    """
+    code = (
+        "import base64 as _b64, sys\n"
+        f"path = {path!r}\n"
+        "try:\n"
+        "    with open(path, 'rb') as _f:\n"
+        "        _data = _f.read()\n"
+        "    print(_b64.b64encode(_data).decode(), end='')\n"
+        "except Exception as _e:\n"
+        "    print('ERROR:' + str(_e), end='', file=sys.stderr)\n"
+        "    sys.exit(1)\n"
+    )
+    result = await sb.python.exec(code, timeout=30)
+    if not result.get("success", False):
+        error = str(result.get("error", "") or "").strip()
+        if not error:
+            output = result.get("output", "")
+            if isinstance(output, dict):
+                error = str(output.get("error", "") or "").strip()
+        raise OSError(
+            f"Failed to read file in sandbox: {error or 'unknown read error'}"
+        )
+    output = result.get("output", "")
+    if isinstance(output, dict):
+        output = output.get("text", "")
+    b64_text = str(output).strip()
+    return base64.b64decode(b64_text)
+
+
+async def _sandbox_write_bytes(sb, path: str, data: bytes) -> None:
+    """Write raw bytes to a file inside the sandbox via sb.python.exec.
+
+    Raises IOError on write failure.
+    """
+    b64_data = base64.b64encode(data).decode()
+    code = (
+        "import base64 as _b64, sys\n"
+        f"path = {path!r}\n"
+        f"b64 = {b64_data!r}\n"
+        "try:\n"
+        "    _raw = _b64.b64decode(b64)\n"
+        "    with open(path, 'wb') as _f:\n"
+        "        _f.write(_raw)\n"
+        "    print('ok', end='')\n"
+        "except Exception as _e:\n"
+        "    print('ERROR:' + str(_e), end='', file=sys.stderr)\n"
+        "    sys.exit(1)\n"
+    )
+    result = await sb.python.exec(code, timeout=30)
+    if not result.get("success", False):
+        error = str(result.get("error", "") or "").strip()
+        if not error:
+            output = result.get("output", "")
+            if isinstance(output, dict):
+                error = str(output.get("error", "") or "").strip()
+        raise OSError(
+            f"Failed to write file in sandbox: {error or 'unknown write error'}"
+        )
+
+
+def _format_result(
+    path: str,
+    result: EditResult,
+    *,
+    replace_all: bool,
+) -> str:
+    """Build the human-readable / LLM-readable result string with optional diff."""
+    if not result.success:
+        return f"Error editing file: {result.error}"
+
+    mode_text = "all matches" if replace_all else "first match"
+
+    lines = [
+        f"Edited {path}.",
+        f"Replaced {result.replacements} occurrence(s) using {mode_text} mode.",
+    ]
+
+    if result.diff:
+        diff_preview = result.diff
+        if len(diff_preview) > 2000:
+            diff_preview = diff_preview[:2000] + "\n... (diff truncated)"
+        lines.append("")
+        lines.append("Diff:")
+        lines.append("```diff")
+        lines.append(diff_preview)
+        lines.append("```")
+
+    return "\n".join(lines)
+
+
 @builtin_tool(config=_COMPUTER_RUNTIME_TOOL_CONFIG)
 @dataclass
 class FileEditTool(FunctionTool):
+    """
+    Enhanced file editing tool with robust fuzzy matching.
+
+    In local runtime it uses the full robust edit engine (BOM +
+    line-ending preservation, file locks); in sandbox runtimes it
+    mediates reads and writes through the booter's filesystem
+    abstraction while applying the same 9-strategy replacer chain.
+
+    This tool is designed to handle LLM-generated edits that may
+    have minor whitespace, indentation, or escape sequence
+    differences from the actual file content.
+    """
+
     name: str = "astrbot_file_edit_tool"
-    description: str = "Editing files."
+    description: str = (
+        "Editing files with robust fuzzy matching. "
+        "Supports exact match, escape-normalized match, line-trimmed match, block-anchor match, "
+        "whitespace-normalized match, indentation-flexible match, "
+        "trimmed-boundary match, context-aware match, "
+        "and multi-occurrence replacement. "
+        "When editing text from Read tool output, preserve the exact indentation "
+        "(tabs/spaces) as it appears AFTER the line number prefix. "
+        "The line number prefix format is: line number + colon + space (e.g., '1: '). "
+        "Everything after that space is the actual file content to match. "
+        "Never include any part of the line number prefix in oldString or newString. "
+        "The edit will FAIL if oldString is not found. "
+        "The edit will FAIL if oldString is found multiple times and replace_all is false. "
+        "Use replace_all for renaming variables or strings across the file."
+    )
     parameters: dict = field(
         default_factory=lambda: {
             "type": "object",
             "properties": {
                 "path": {
                     "type": "string",
-                    "description": "Path of the file to edit. If relative, will be in workspace root.",
+                    "description": (
+                        "Path of the file to edit. If relative, will be in workspace root."
+                    ),
                 },
                 "old": {
                     "type": "string",
-                    "description": "The exact old text to replace.",
+                    "description": (
+                        "The text to replace. Must be an exact substring of the file content, "
+                        "but the tool will try multiple matching strategies if exact match fails. "
+                        "Include sufficient surrounding context (3-5 lines) to make the match unique."
+                    ),
                 },
                 "new": {
                     "type": "string",
                     "description": "The replacement text.",
                 },
                 "replace_all": {
                     "type": "boolean",
-                    "description": "Whether to replace all matches. Defaults to false.",
+                    "description": (
+                        "Whether to replace all matches. Defaults to false. "
+                        "Useful for renaming variables or strings across the file."
+                    ),
                 },
             },
             "required": ["path", "old", "new"],
@@ -409,45 +547,67 @@ async def call(
         umo = str(context.context.event.unified_msg_origin)
         local_env = is_local_runtime(context)
         restricted = _is_restricted_env(context)
+
         try:
-            normalized_path = (
-                _normalize_rw_path(
+            if local_env:
+                normalized_path = _normalize_rw_path(
                     path,
                     restricted=restricted,
                     local_env=local_env,
                     umo=umo,
                     write=True,
                 )
-                if local_env
-                else path.strip()
-            )
+            else:
+                normalized_path = path.strip()
+
             if not normalized_path:
                 raise ValueError("`path` must be a non-empty string.")
-            normalized_old = _decode_escaped_text(old)
-            normalized_new = _decode_escaped_text(new)
-            sb = await get_booter(
-                context.context.context,
-                context.context.event.unified_msg_origin,
-            )
-            result = await sb.fs.edit_file(
-                path=normalized_path,
-                old_string=normalized_old,
-                new_string=normalized_new,
-                replace_all=replace_all,
-                encoding="utf-8",
-            )
-            if not result.get("success", False):
-                error_detail = str(result.get("error", "") or "").strip()
-                return (
-                    "Error editing file: "
-                    f"{error_detail or 'unknown filesystem edit error'}"
+
+            if local_env:
+                result = await edit_file(
+                    path=normalized_path,
+                    old_string=old,
+                    new_string=new,
+                    replace_all=replace_all,
+                    encoding="utf-8",
+                )
+            else:
+                sb = await get_booter(
+                    context.context.context,
+                    umo,
                 )
-            replacements = int(result.get("replacements", 0) or 0)
-            mode_text = "all matches" if replace_all else "first match"
-            return (
-                f"Edited {normalized_path}. "
-                f"Replaced {replacements} occurrence(s) using {mode_text} mode."
+                lock = get_file_lock(normalized_path)
+                async with lock:
+                    # 1. Binary read — preserves original line endings (CRLF/LF)
+                    try:
+                        raw_bytes = await _sandbox_read_bytes(sb, normalized_path)
+                    except OSError as exc:
+                        return f"Error editing file: {exc}"
+
+                    # 2. Line-ending-aware edit (reuses edit_engine core logic)
+                    try:
+                        write_bytes, result = bytes_edit_file(
+                            raw_bytes,
+                            old,
+                            new,
+                            replace_all=replace_all,
+                            encoding="utf-8",
+                        )
+                    except ValueError as exc:
+                        return f"Error editing file: {exc}"
+
+                    # 3. Binary write — preserves restored line endings
+                    try:
+                        await _sandbox_write_bytes(sb, normalized_path, write_bytes)
+                    except OSError as exc:
+                        return f"Error editing file: {exc}"
+
+            return _format_result(
+                normalized_path,
+                result,
+                replace_all=replace_all,
             )
+
         except PermissionError as exc:
             return f"Error: {exc}"
         except Exception as exc: