github_pr.py

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

import httpx

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


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

    @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 with comments."

    @property
    def input_schema(self) -> dict:
        return {
            "type": "object",
            "properties": {
                "repo": {"type": "string", "description": "Repository in owner/repo format"},
                "pr_number": {"type": "integer", "description": "View a specific PR by number"},
                "state": {"type": "string", "enum": ["open", "closed", "all"], "description": "Filter by state (default: open)"},
                "base": {"type": "string", "description": "Filter by base branch"},
                "head": {"type": "string", "description": "Filter by head branch (user:branch format)"},
                "limit": {"type": "integer", "description": "Number of PRs to list (default: 30)"},
            },
            "required": ["repo"],
        }

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

    async def execute(self, repo: str, pr_number: int = None, state: str = "open",
                      base: str = None, head: str = None, limit: int = 30, **kwargs) -> ToolResult:
        valid, access, error = await check_github_access(repo, "read")
        if not valid:
            return ToolResult.fail(error)

        token = self.get_credential("GITHUB_TOKEN")
        headers = {
            "Authorization": f"Bearer {token}",
            "Accept": "application/vnd.github+json",
            "X-GitHub-Api-Version": "2022-11-28",
        }

        try:
            async with httpx.AsyncClient(timeout=30) as client:
                if pr_number:
                    return await self._view_pr(client, repo, pr_number, headers)
                else:
                    return await self._list_prs(client, repo, state, base, head, limit, headers)
        except httpx.TimeoutException:
            return ToolResult.fail("GitHub API request timed out")
        except Exception as e:
            return ToolResult.fail(f"GitHub error: {str(e)}")

    async def _list_prs(self, client, repo, state, base, head, limit, headers) -> ToolResult:
        params = {"state": state, "per_page": min(limit, 100)}
        if base:
            params["base"] = base
        if head:
            params["head"] = head

        resp = await client.get(f"https://api.github.com/repos/{repo}/pulls", headers=headers, params=params)
        if resp.status_code in (401, 403):
            return ToolResult.fail("GitHub authentication failed — check GITHUB_TOKEN")
        if resp.status_code == 404:
            return ToolResult.fail(f"Repository '{repo}' not found")
        if resp.status_code != 200:
            return ToolResult.fail(f"GitHub API error {resp.status_code}: {resp.text[:300]}")

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

        formatted = [{
            "number": pr["number"], "title": pr["title"], "state": pr["state"],
            "draft": pr.get("draft", False), "author": pr["user"]["login"],
            "head": pr["head"]["ref"], "base": pr["base"]["ref"],
            "created_at": pr["created_at"], "mergeable": pr.get("mergeable"),
            "url": pr["html_url"],
        } for pr in prs]

        return ToolResult.ok({"state": state, "count": len(formatted), "prs": formatted})

    async def _view_pr(self, client, repo, pr_number, headers) -> ToolResult:
        resp = await client.get(f"https://api.github.com/repos/{repo}/pulls/{pr_number}", headers=headers)
        if resp.status_code == 404:
            return ToolResult.fail(f"PR #{pr_number} not found")
        if resp.status_code != 200:
            return ToolResult.fail(f"GitHub API error {resp.status_code}")

        pr = resp.json()
        result = {
            "number": pr["number"], "title": pr["title"], "state": pr["state"],
            "draft": pr.get("draft", False), "body": pr.get("body", ""),
            "author": pr["user"]["login"], "head": pr["head"]["ref"], "base": pr["base"]["ref"],
            "mergeable": pr.get("mergeable"),
            "additions": pr.get("additions", 0), "deletions": pr.get("deletions", 0),
            "changed_files": pr.get("changed_files", 0),
            "created_at": pr["created_at"], "updated_at": pr["updated_at"],
            "url": pr["html_url"],
        }

        comments_resp = await client.get(
            f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments",
            headers=headers, params={"per_page": 50},
        )
        if comments_resp.status_code == 200:
            comments = comments_resp.json()
            if comments:
                result["comments"] = [
                    {"author": c["user"]["login"], "body": c["body"], "created_at": c["created_at"]}
                    for c in comments
                ]

        return ToolResult.ok(result)


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

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

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

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

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

    async def execute(self, repo: str, pr_number: int, body: str, **kwargs) -> ToolResult:
        valid, access, error = await check_github_access(repo, "comment")
        if not valid:
            return ToolResult.fail(error)

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

        token = self.get_credential("GITHUB_TOKEN")

        try:
            async with httpx.AsyncClient(timeout=30) as client:
                resp = await client.post(
                    f"https://api.github.com/repos/{repo}/issues/{pr_number}/comments",
                    headers={
                        "Authorization": f"Bearer {token}",
                        "Accept": "application/vnd.github+json",
                        "X-GitHub-Api-Version": "2022-11-28",
                    },
                    json={"body": body},
                )

                if resp.status_code == 201:
                    data = resp.json()
                    return ToolResult.ok({
                        "pr_number": pr_number, "comment_id": data["id"],
                        "status": "commented", "url": data["html_url"],
                    })
                if resp.status_code == 404:
                    return ToolResult.fail(f"PR #{pr_number} not found")
                if resp.status_code in (401, 403):
                    return ToolResult.fail("GitHub authentication failed — check GITHUB_TOKEN")
                return ToolResult.fail(f"GitHub API error {resp.status_code}: {resp.text[:300]}")

        except httpx.TimeoutException:
            return ToolResult.fail("GitHub API request timed out")
        except Exception as e:
            return ToolResult.fail(f"GitHub error: {str(e)}")