coding-agent-gitea/gitea/tools/file_tools.py

import logging
import os
from pathlib import Path
from typing import Any
from gitea.client import GiteaClient

logger: logging.Logger = logging.getLogger("gitea-file-tools")


class FileTools:
    """Tools for Gitea file/content operations."""

    def __init__(self, client: GiteaClient, repo_path: str | None = None) -> None:
        self._client = client
        self._repo_path: str | None = repo_path

    def _paginate_lines(
        self,
        content: str,
        offset: int,
        limit: int,
    ) -> str:
        """Return lines[offset-1 : offset-1+limit] with a paging footer if truncated.

        Uses the same 1-indexed convention as CodingTools.read_file.
        """
        lines: list[str] = content.splitlines()
        total: int = len(lines)
        start: int = offset - 1  # convert to 0-indexed
        page: list[str] = lines[start : start + limit]
        formatted: list[str] = [
            f"{start + i + 1}: {line}" for i, line in enumerate(page)
        ]
        result: str = "\n".join(formatted)
        end_line: int = start + len(page)
        if end_line < total:
            next_offset: int = end_line + 1
            result += (
                f"\n\n[{total} lines total — showing lines {offset}–{end_line}. "
                f"Re-call with offset={next_offset} to read more.]"
            )
        return result

    def _resolve_local_path(self, owner: str, repo: str, path: str) -> str | None:
        """Resolve owner/repo/path to a local filesystem path if the repo is cloned."""
        if not self._repo_path:
            return None
        local_repo: str = os.path.join(self._repo_path, owner, repo)
        if os.path.isdir(local_repo):
            return os.path.join(local_repo, path)
        return None

    def get_file_content(
        self,
        owner: str,
        repo: str,
        path: str,
        offset: int = 1,
        limit: int = 250,
    ) -> str:
        """Get the content of a file from a Gitea repository with line paging.

        Checks the local workspace first if repo_path is configured, falling
        back to the remote API when the file is not available locally.

        Args:
            offset: 1-indexed line to start from (default 1).
            limit: Maximum number of lines to return (default 250).
        """
        local_path: str | None = self._resolve_local_path(owner, repo, path)
        if local_path and os.path.isfile(local_path):
            try:
                with open(local_path, "r", encoding="utf-8", errors="replace") as f:
                    raw: str = f.read()
                return self._paginate_lines(raw, offset, limit)
            except Exception as exc:
                logger.debug(
                    f"Local read failed for {owner}/{repo}/{path}: {exc}",
                    exc_info=True,
                )

        try:
            content = self._client.files.get_file_content(owner, repo, path)
            raw: str = "\n".join(content) if isinstance(content, list) else content
            return self._paginate_lines(raw, offset, limit)
        except Exception as e:
            logger.error(
                f"Failed to get file content for {owner}/{repo}/{path}: {e}",
                exc_info=True,
            )
            return (
                f"Error: Could not retrieve file '{path}' from {owner}/{repo}. "
                f"Verify the file path and branch are correct. Details: {e}"
            )

    def get_file_content_with_ref(
        self,
        owner: str,
        repo: str,
        path: str,
        ref: str = "master",
        offset: int = 1,
        limit: int = 250,
    ) -> str:
        """Get file content at a specific git ref with line paging.

        Checks the local workspace first using ``git show`` if the repo is
        cloned locally, falling back to the remote API.

        Args:
            ref: Branch, tag, or commit SHA (default 'master').
            offset: 1-indexed line to start from (default 1).
            limit: Maximum number of lines to return (default 250).
        """
        if self._repo_path:
            local_repo: str = os.path.join(self._repo_path, owner, repo)
            if os.path.isdir(local_repo):
                try:
                    import subprocess

                    result = subprocess.run(
                        ["git", "-C", local_repo, "show", f"{ref}:{path}"],
                        capture_output=True,
                        text=True,
                        timeout=15,
                    )
                    if result.returncode == 0:
                        return self._paginate_lines(result.stdout, offset, limit)
                except Exception as exc:
                    logger.debug(
                        f"Local git show failed for {owner}/{repo}/{path}@{ref}: {exc}",
                        exc_info=True,
                    )

        try:
            content = self._client.files.get_file_content(owner, repo, path, ref)
            raw: str = "\n".join(content) if isinstance(content, list) else content
            return self._paginate_lines(raw, offset, limit)
        except Exception as e:
            logger.error(
                f"Failed to get file content for {owner}/{repo}/{path}@{ref}: {e}",
                exc_info=True,
            )
            return (
                f"Error: Could not retrieve file '{path}' at ref '{ref}' from {owner}/{repo}. "
                f"Verify the file path and ref are correct. Details: {e}"
            )

    def commit_file(
        self, owner: str, repo: str, path: str, message: str, content: str, branch: str
    ) -> str:
        try:
            self._client.files.update_file(owner, repo, path, message, content, branch)
            return f"File '{path}' committed successfully to {owner}/{repo}."
        except Exception as e:
            logger.error(
                f"Failed to commit file '{path}' to {owner}/{repo}@{branch}: {e}",
                exc_info=True,
            )
            return (
                f"Error: Could not commit file '{path}' to {owner}/{repo} on branch '{branch}'. "
                f"Check for conflicts or permission issues. Details: {e}"
            )

    def update_file(
        self, owner: str, repo: str, path: str, message: str, content: str, branch: str
    ) -> str:
        try:
            self._client.files.update_file(owner, repo, path, message, content, branch)
            return f"File '{path}' updated in {owner}/{repo}."
        except Exception as e:
            logger.error(
                f"Failed to update file '{path}' in {owner}/{repo}@{branch}: {e}",
                exc_info=True,
            )
            return (
                f"Error: Could not update file '{path}' in {owner}/{repo} on branch '{branch}'. "
                f"Check for conflicts or permission issues. Details: {e}"
            )