file_io.py

"""File I/O operations with a configurable working directory."""

import os
from pathlib import Path
from typing import Optional

from agentscope.message import TextBlock
from agentscope.tool import ToolResponse

from ..utils import read_file_safe, truncate_text_output, TRUNCATION_NOTICE_MARKER


class FileIO:
    """File I/O operations with a configurable working directory."""

    def __init__(self, working_dir: str | Path):
        """Initialize FileIO with a working directory.

        Args:
            working_dir (`str`):
                The working directory for resolving relative paths.
        """
        self.working_dir = Path(working_dir)

    def _resolve_file_path(self, file_path: str) -> str:
        """Resolve file path: use absolute path as-is,
        resolve relative path from working_dir.

        Args:
            file_path: The input file path (absolute or relative).

        Returns:
            The resolved absolute file path as string.
        """
        path = Path(file_path).expanduser()
        if path.is_absolute():
            return str(path)
        else:
            return str(self.working_dir / file_path)

    async def read_file(  # pylint: disable=too-many-return-statements
        self,
        file_path: str,
        start_line: Optional[int] = None,
        end_line: Optional[int] = None,
    ) -> ToolResponse:
        """Read a file. Relative paths resolve from WORKING_DIR.

        Use start_line/end_line to read a specific line range (output includes
        line numbers). Omit both to read the full file.

        Args:
            file_path (`str`):
                Path to the file.
            start_line (`int`, optional):
                First line to read (1-based, inclusive).
            end_line (`int`, optional):
                Last line to read (1-based, inclusive).
        """

        # Convert start_line/end_line to int if they are strings
        if start_line is not None:
            try:
                start_line = int(start_line)
            except (ValueError, TypeError):
                return ToolResponse(
                    content=[
                        TextBlock(
                            type="text",
                            text=f"Error: start_line must be an integer, got {start_line!r}.",
                        ),
                    ],
                )

        if end_line is not None:
            try:
                end_line = int(end_line)
            except (ValueError, TypeError):
                return ToolResponse(
                    content=[
                        TextBlock(
                            type="text",
                            text=f"Error: end_line must be an integer, got {end_line!r}.",
                        ),
                    ],
                )

        file_path = self._resolve_file_path(file_path)

        if not os.path.exists(file_path):
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: The file {file_path} does not exist.",
                    ),
                ],
            )

        if not os.path.isfile(file_path):
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: The path {file_path} is not a file.",
                    ),
                ],
            )

        try:
            content = read_file_safe(file_path)
            all_lines = content.split("\n")
            total = len(all_lines)

            # Determine read range
            s = max(1, start_line if start_line is not None else 1)
            e = min(total, end_line if end_line is not None else total)

            if s > total:
                return ToolResponse(
                    content=[
                        TextBlock(
                            type="text",
                            text=f"Error: start_line {s} exceeds file length ({total} lines).",
                        ),
                    ],
                )

            if s > e:
                return ToolResponse(
                    content=[
                        TextBlock(
                            type="text",
                            text=f"Error: start_line ({s}) > end_line ({e}).",
                        ),
                    ],
                )

            # Extract selected lines
            selected_content = "\n".join(all_lines[s - 1 : e])

            # Apply smart truncation (consistent with shell output format)
            text = truncate_text_output(
                selected_content,
                start_line=s,
                total_lines=total,
                file_path=file_path,
            )

            # Add continuation hint if partial read without truncation.
            # Use TRUNCATION_NOTICE_MARKER format so ToolResultCompactor can
            # re-truncate with the correct start_line when compacting old messages.
            if text == selected_content and e < total:
                content_bytes = len(text.encode("utf-8"))
                notice = (
                    TRUNCATION_NOTICE_MARKER + f"\nThe output above was truncated."
                    f"\nThe full content is saved to the file "
                    f"and contains {total} lines in total."
                    f"\nThis excerpt starts at line {s} and "
                    f"covers the next {content_bytes} bytes."
                    "\nIf the current content is not enough, "
                    f"call `read_file` with file_path={file_path} start_line={e + 1} to read more."
                )
                text = text + notice

            return ToolResponse(
                content=[TextBlock(type="text", text=text)],
            )

        except Exception as e:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: Read file failed due to \n{e}",
                    ),
                ],
            )

    async def write_file(
        self,
        file_path: str,
        content: str,
    ) -> ToolResponse:
        """Create or overwrite a file. Relative paths resolve from working_dir.

        Args:
            file_path (`str`):
                Path to the file.
            content (`str`):
                Content to write.
        """
        if not file_path:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text="Error: No `file_path` provided.",
                    ),
                ],
            )

        file_path = self._resolve_file_path(file_path)

        try:
            with open(file_path, "w", encoding="utf-8") as file:
                file.write(content)
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Wrote {len(content)} bytes to {file_path}.",
                    ),
                ],
            )
        except Exception as e:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: Write file failed due to \n{e}",
                    ),
                ],
            )

    # pylint: disable=too-many-return-statements
    async def edit_file(
        self,
        file_path: str,
        old_text: str,
        new_text: str,
    ) -> ToolResponse:
        """Find-and-replace text in a file. All occurrences of old_text are
        replaced with new_text. Relative paths resolve from working_dir.

        Args:
            file_path (`str`):
                Path to the file.
            old_text (`str`):
                Exact text to find.
            new_text (`str`):
                Replacement text.
        """
        if not file_path:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text="Error: No `file_path` provided.",
                    ),
                ],
            )

        resolved_path = self._resolve_file_path(file_path)

        if not os.path.exists(resolved_path):
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: The file {resolved_path} does not exist.",
                    ),
                ],
            )

        if not os.path.isfile(resolved_path):
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: The path {resolved_path} is not a file.",
                    ),
                ],
            )

        try:
            content = read_file_safe(resolved_path)
        except Exception as e:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: Read file failed due to \n{e}",
                    ),
                ],
            )

        if old_text not in content:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: The text to replace was not found in {file_path}.",
                    ),
                ],
            )

        new_content = content.replace(old_text, new_text)
        write_response = await self.write_file(file_path=resolved_path, content=new_content)

        if write_response.content and len(write_response.content) > 0:
            write_text = write_response.content[0].get("text", "")
            if write_text.startswith("Error:"):
                return write_response

        return ToolResponse(
            content=[
                TextBlock(
                    type="text",
                    text=f"Successfully replaced text in {file_path}.",
                ),
            ],
        )

    async def append_file(
        self,
        file_path: str,
        content: str,
    ) -> ToolResponse:
        """Append content to the end of a file. Relative paths resolve from
        working_dir.

        Args:
            file_path (`str`):
                Path to the file.
            content (`str`):
                Content to append.
        """
        if not file_path:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text="Error: No `file_path` provided.",
                    ),
                ],
            )

        file_path = self._resolve_file_path(file_path)

        try:
            with open(file_path, "a", encoding="utf-8") as file:
                file.write(content)
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Appended {len(content)} bytes to {file_path}.",
                    ),
                ],
            )
        except Exception as e:
            return ToolResponse(
                content=[
                    TextBlock(
                        type="text",
                        text=f"Error: Append file failed due to \n{e}",
                    ),
                ],
            )