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

"""Tools for Gitea pull request operations."""

import json
import logging
from typing import Any
from gitea.client import GiteaClient
from gitea.models import PullRequestModel, CommentModel

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


_MAX_DIFF_CHARS: int = 15_000


def _truncate_diff(
    text: str,
    max_chars: int = _MAX_DIFF_CHARS,
    char_offset: int = 0,
) -> str:
    """Slice [char_offset : char_offset+max_chars] from text, cutting at a hunk boundary."""
    total: int = len(text)
    chunk: str = text[char_offset : char_offset + max_chars]
    if char_offset > 0 or (char_offset + max_chars) < total:
        # Try to cut at a diff hunk boundary (@@) for coherence
        hunk_boundary: int = chunk.rfind("\n@@")
        if hunk_boundary > int(len(chunk) * 0.6):
            chunk = chunk[:hunk_boundary]
        next_offset: int = char_offset + len(chunk)
        chunk += (
            f"\n\n[Diff truncated — {total} chars total. "
            f"Showing chars {char_offset}–{next_offset}. "
            f"Re-call with char_offset={next_offset} to read more.]"
        )
    return chunk


class PRTools:
    """Tools for Gitea pull request operations."""

    def __init__(self, client: GiteaClient) -> None:
        self._client = client

    def get_pull_request(self, owner: str, repo: str, pull_number: int) -> str:
        try:
            pr: PullRequestModel = self._client.get_pull_request(owner, repo, pull_number)
            return pr.model_dump_json(indent=2)
        except Exception as e:
            return f"Error getting pull request: {str(e)}"

    def close_pull_request(self, owner: str, repo: str, pull_number: int) -> str:
        try:
            self._client.close_pull_request(owner, repo, pull_number)
            return f"Pull request #{pull_number} closed successfully."
        except Exception as e:
            return f"Error closing pull request: {str(e)}"

    def get_pull_request_comments(
        self,
        owner: str,
        repo: str,
        pull_number: int,
        limit: int = 20,
        offset: int = 0,
    ) -> str:
        """Get comments on a pull request with optional paging.

        Args:
            limit: Maximum number of comments to return (default 20).
            offset: Zero-based comment index to start from (default 0).
        """
        try:
            comments: list[CommentModel] = self._client.get_pull_request_comments(
                owner, repo, pull_number
            )
            total: int = len(comments)
            page: list[CommentModel] = comments[offset : offset + limit]
            result: str = json.dumps([c.model_dump() for c in page], indent=2)
            if total > offset + limit:
                next_offset: int = offset + limit
                result += (
                    f"\n\n[{total} comments total — showing {offset}–{offset + len(page)}. "
                    f"Re-call with offset={next_offset} to see more.]"
                )
            return result
        except Exception as e:
            return f"Error getting PR comments: {str(e)}"

    def list_assigned_pull_requests(self) -> list[dict[str, Any]]:
        try:
            repos = self._client.list_all_user_repos()
            all_prs: list[dict[str, Any]] = []
            for repo_info in repos:
                repo_owner = repo_info.owner
                repo_name = repo_info.name
                prs = self._client.list_assigned_pull_requests(repo_owner, repo_name)
                if prs:
                    all_prs.extend([pr.model_dump() if hasattr(pr, 'model_dump') else pr for pr in prs])
            return all_prs
        except Exception as e:
            logger.error(f"Error listing assigned pull requests: {e}", exc_info=True)
            return []

    def list_pull_requests(self, owner: str, repo: str, state: str = "open") -> str:
        try:
            prs = self._client.list_repo_pull_requests(owner, repo, state)
            if not prs:
                return f"No PRs in {owner}/{repo}."
            summary = [f"#{pr.number}: {pr.title}" for pr in prs]
            return "\n".join(summary)
        except Exception as e:
            return f"Error listing PRs: {str(e)}"

    def create_pull_request(self, owner: str, repo: str, head: str, base: str, title: str, description: str = "") -> str:
        try:
            pr = self._client.create_pr_via_tea(owner, repo, title, description, head, base)
            return pr.model_dump_json(indent=2)
        except Exception as e:
            return f"Error creating PR: {str(e)}"

    def update_pull_request(
        self,
        owner: str,
        repo: str,
        pull_number: int,
        title: str | None = None,
        body: str | None = None,
        state: str | None = None,
    ) -> str:
        try:
            pr = self._client.update_pull_request(owner, repo, pull_number, title, body, state)
            return pr.model_dump_json(indent=2)
        except Exception as e:
            return f"Error updating PR #{pull_number}: {str(e)}"

    def add_label_to_pr(self, owner: str, repo: str, pr_number: int, label: str) -> str:
        try:
            self._client.add_label_pr(owner, repo, pr_number, label)
            return f"Label '{label}' added to PR #{pr_number}."
        except Exception as e:
            return f"Error adding label to PR #{pr_number}: {e}"

    def get_pull_request_diff(
        self,
        owner: str,
        repo: str,
        pull_number: int,
        max_chars: int = _MAX_DIFF_CHARS,
        char_offset: int = 0,
    ) -> str:
        """Get the diff of a pull request, with truncation and offset paging.

        Args:
            max_chars: Maximum characters to return (default 15 000).
            char_offset: Character offset to start reading from (default 0).
                         Increment by max_chars to page through a large diff.
        """
        try:
            diff: str = self._client.get_pull_request_diff(owner, repo, pull_number)
            return _truncate_diff(diff, max_chars, char_offset)
        except Exception as e:
            return f"Error getting PR diff: {str(e)}"

    def get_pull_request_patch(
        self,
        owner: str,
        repo: str,
        pull_number: int,
        max_chars: int = _MAX_DIFF_CHARS,
        char_offset: int = 0,
    ) -> str:
        """Get the patch of a pull request, with truncation and offset paging.

        Args:
            max_chars: Maximum characters to return (default 15 000).
            char_offset: Character offset to start reading from (default 0).
                         Increment by max_chars to page through a large patch.
        """
        try:
            patch: str = self._client.get_pull_request_patch(owner, repo, pull_number)
            return _truncate_diff(patch, max_chars, char_offset)
        except Exception as e:
            return f"Error getting PR patch: {str(e)}"

    def approve_pull_request(self, owner: str, repo: str, pull_number: int, comment: str) -> str:
        try:
            self._client.approve_pr(owner, repo, pull_number, comment)
            return f"Approved PR #{pull_number}."
        except Exception as e:
            return f"Error approving PR: {str(e)}"

    def request_changes(self, owner: str, repo: str, pull_number: int, comment: str) -> str:
        try:
            self._client.request_changes_pr(owner, repo, pull_number, comment)
            return f"Requested changes on PR #{pull_number}."
        except Exception as e:
            return f"Error requesting changes: {str(e)}"