github_pr.py

"""
GitHub PR Tools - Read and comment on pull requests.
"""

import subprocess

from ..base import BaseTool, ToolResult
from ..registry import register_tool
from .base import validate_project_root


@register_tool
class GitHubPRReadTool(BaseTool):
    """List and view GitHub pull requests."""

    @property
    def name(self) -> str:
        return "github_pr_read"

    @property
    def description(self) -> str:
        return (
            "Read GitHub pull requests. List PRs with filters or view a specific PR. "
            "Can filter by state, base branch, author, and labels."
        )

    @property
    def input_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "pr_number": {
                    "type": "integer",
                    "description": "View a specific PR by number",
                },
                "state": {
                    "type": "string",
                    "enum": ["open", "closed", "merged", "all"],
                    "description": "Filter by state (default: open)",
                },
                "base": {
                    "type": "string",
                    "description": "Filter by base branch",
                },
                "author": {
                    "type": "string",
                    "description": "Filter by author username",
                },
                "label": {
                    "type": "string",
                    "description": "Filter by label",
                },
                "search": {
                    "type": "string",
                    "description": "Search in title and body",
                },
                "limit": {
                    "type": "integer",
                    "description": "Number of PRs to list (default: 30)",
                },
            },
        }

    @property
    def requires_grant_metadata(self) -> list[str]:
        return ["project_root"]

    def credential_keys(self) -> list[str]:
        return []

    async def execute(
        self,
        pr_number: int = None,
        state: str = "open",
        base: str = None,
        author: str = None,
        label: str = None,
        search: str = None,
        limit: int = 30,
        **kwargs
    ) -> ToolResult:
        project_root = self.get_grant_metadata("project_root")

        valid, error = validate_project_root(project_root)
        if not valid:
            return ToolResult.fail(error)

        try:
            if pr_number:
                return await self._view_pr(project_root, pr_number)
            else:
                return await self._list_prs(
                    project_root, state, base, author, label, search, limit
                )

        except FileNotFoundError:
            return ToolResult.fail("GitHub CLI (gh) not installed")
        except subprocess.TimeoutExpired:
            return ToolResult.fail("Command timed out")
        except Exception as e:
            return ToolResult.fail(f"GitHub error: {str(e)}")

    async def _list_prs(
        self,
        project_root: str,
        state: str,
        base: str = None,
        author: str = None,
        label: str = None,
        search: str = None,
        limit: int = 30,
    ) -> ToolResult:
        """List pull requests with filters."""
        cmd = ["gh", "pr", "list", f"--state={state}", f"--limit={limit}"]

        if base:
            cmd.extend(["--base", base])
        if author:
            cmd.extend(["--author", author])
        if label:
            cmd.extend(["--label", label])
        if search:
            cmd.extend(["--search", search])

        result = subprocess.run(
            cmd,
            cwd=project_root,
            capture_output=True,
            text=True,
            timeout=30,
        )

        if result.returncode != 0:
            stderr = result.stderr.lower()
            if "not a git repository" in stderr:
                return ToolResult.fail("Not a git repository")
            if "no pull requests" in stderr or "no results" in stderr:
                return ToolResult.ok({
                    "prs": [],
                    "message": "No pull requests found",
                })
            return ToolResult.fail(f"gh error: {result.stderr}")

        output = result.stdout.strip()

        if not output:
            return ToolResult.ok({
                "prs": [],
                "message": "No pull requests found",
            })

        lines = output.split('\n')
        return ToolResult.ok({
            "state": state,
            "count": len(lines),
            "prs": output,
        })

    async def _view_pr(self, project_root: str, pr_number: int) -> ToolResult:
        """View a specific PR with comments."""
        result = subprocess.run(
            ["gh", "pr", "view", str(pr_number), "--comments"],
            cwd=project_root,
            capture_output=True,
            text=True,
            timeout=30,
        )

        if result.returncode != 0:
            stderr = result.stderr.lower()
            if "not found" in stderr or "could not find" in stderr:
                return ToolResult.fail(f"PR #{pr_number} not found")
            return ToolResult.fail(f"gh error: {result.stderr}")

        return ToolResult.ok({
            "pr_number": pr_number,
            "content": result.stdout.strip(),
        })


@register_tool
class GitHubPRCommentTool(BaseTool):
    """Comment on a GitHub pull request."""

    @property
    def name(self) -> str:
        return "github_pr_comment"

    @property
    def description(self) -> str:
        return (
            "Add a comment to a GitHub pull request. "
            "Requires approval from the gatekeeper."
        )

    @property
    def input_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "pr_number": {
                    "type": "integer",
                    "description": "PR number to comment on",
                },
                "body": {
                    "type": "string",
                    "description": "Comment text",
                },
            },
            "required": ["pr_number", "body"],
        }

    @property
    def requires_grant_metadata(self) -> list[str]:
        return ["project_root"]

    def credential_keys(self) -> list[str]:
        return []

    async def execute(
        self,
        pr_number: int,
        body: str,
        **kwargs
    ) -> ToolResult:
        project_root = self.get_grant_metadata("project_root")

        valid, error = validate_project_root(project_root)
        if not valid:
            return ToolResult.fail(error)

        if not body or not body.strip():
            return ToolResult.fail("Comment body cannot be empty")

        try:
            result = subprocess.run(
                ["gh", "pr", "comment", str(pr_number), "--body", body],
                cwd=project_root,
                capture_output=True,
                text=True,
                timeout=30,
            )

            if result.returncode != 0:
                stderr = result.stderr.lower()
                if "not found" in stderr or "could not find" in stderr:
                    return ToolResult.fail(f"PR #{pr_number} not found")
                return ToolResult.fail(f"gh error: {result.stderr}")

            return ToolResult.ok({
                "pr_number": pr_number,
                "status": "commented",
                "message": f"Comment added to PR #{pr_number}",
            })

        except FileNotFoundError:
            return ToolResult.fail("GitHub CLI (gh) not installed")
        except subprocess.TimeoutExpired:
            return ToolResult.fail("Command timed out")
        except Exception as e:
            return ToolResult.fail(f"GitHub error: {str(e)}")