src/hermes/tools/github.py

"""GitHub tool implementation."""

from __future__ import annotations

import logging
from typing import Any

import httpx

from hermes.tools.base.tool import BaseTool, ToolSchema

logger = logging.getLogger(__name__)


class GitHubTool(BaseTool):
    """Tool for interacting with GitHub."""

    def __init__(self, token: str | None = None) -> None:
        super().__init__()
        self.token = token
        self.base_url = "https://api.github.com"

    def _define_schema(self) -> ToolSchema:
        return ToolSchema(
            name="github_repo_reader",
            description="Read GitHub repository information, files, and analysis",
            parameters={
                "action": {
                    "type": "string",
                    "description": "Action to perform: get_repo, list_files, read_file, get_readme, search_code",
                },
                "owner": {
                    "type": "string",
                    "description": "Repository owner",
                },
                "repo": {
                    "type": "string",
                    "description": "Repository name",
                },
                "path": {
                    "type": "string",
                    "description": "File path (for read_file action)",
                },
                "query": {
                    "type": "string",
                    "description": "Search query (for search_code action)",
                },
                "branch": {
                    "type": "string",
                    "description": "Branch name",
                    "default": "main",
                },
            },
            required=["action", "owner", "repo"],
            category="research",
            tags=["github", "code", "research"],
        )

    def _get_headers(self) -> dict[str, str]:
        """Get request headers."""
        headers = {"Accept": "application/vnd.github.v3+json"}
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers

    async def execute(self, **kwargs: Any) -> dict[str, Any]:
        """Execute GitHub API call."""
        action = kwargs["action"]
        owner = kwargs["owner"]
        repo = kwargs["repo"]

        try:
            if action == "get_repo":
                return await self._get_repo(owner, repo)
            elif action == "list_files":
                return await self._list_files(owner, repo, kwargs.get("path", ""), kwargs.get("branch", "main"))
            elif action == "read_file":
                return await self._read_file(owner, repo, kwargs["path"], kwargs.get("branch", "main"))
            elif action == "get_readme":
                return await self._get_readme(owner, repo)
            elif action == "search_code":
                return await self._search_code(owner, repo, kwargs.get("query", ""))
            else:
                return {"error": f"Unknown action: {action}"}

        except Exception as e:
            logger.error(f"GitHub API error: {e}")
            return {"error": str(e)}

    async def _get_repo(self, owner: str, repo: str) -> dict[str, Any]:
        """Get repository information."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/repos/{owner}/{repo}",
                headers=self._get_headers(),
                timeout=30.0,
            )
            response.raise_for_status()
            data = response.json()

            return {
                "name": data.get("name", ""),
                "full_name": data.get("full_name", ""),
                "description": data.get("description", ""),
                "language": data.get("language", ""),
                "stars": data.get("stargazers_count", 0),
                "forks": data.get("forks_count", 0),
                "open_issues": data.get("open_issues_count", 0),
                "created_at": data.get("created_at", ""),
                "updated_at": data.get("updated_at", ""),
                "topics": data.get("topics", []),
                "license": data.get("license", {}).get("name", "") if data.get("license") else "",
                "default_branch": data.get("default_branch", "main"),
            }

    async def _list_files(
        self, owner: str, repo: str, path: str, branch: str
    ) -> dict[str, Any]:
        """List files in a directory."""
        async with httpx.AsyncClient() as client:
            url = f"{self.base_url}/repos/{owner}/{repo}/contents/{path}"
            response = await client.get(
                url,
                headers=self._get_headers(),
                params={"ref": branch},
                timeout=30.0,
            )
            response.raise_for_status()
            data = response.json()

            files = []
            for item in data if isinstance(data, list) else [data]:
                files.append(
                    {
                        "name": item.get("name", ""),
                        "path": item.get("path", ""),
                        "type": item.get("type", ""),
                        "size": item.get("size", 0),
                        "url": item.get("html_url", ""),
                    }
                )

            return {"path": path, "files": files, "count": len(files)}

    async def _read_file(
        self, owner: str, repo: str, path: str, branch: str
    ) -> dict[str, Any]:
        """Read a file's contents."""
        import base64

        async with httpx.AsyncClient() as client:
            url = f"{self.base_url}/repos/{owner}/{repo}/contents/{path}"
            response = await client.get(
                url,
                headers=self._get_headers(),
                params={"ref": branch},
                timeout=30.0,
            )
            response.raise_for_status()
            data = response.json()

            content = ""
            if data.get("encoding") == "base64" and data.get("content"):
                content = base64.b64decode(data["content"]).decode("utf-8", errors="replace")

            return {
                "path": path,
                "content": content,
                "size": data.get("size", 0),
                "sha": data.get("sha", ""),
            }

    async def _get_readme(self, owner: str, repo: str) -> dict[str, Any]:
        """Get repository README."""
        import base64

        async with httpx.AsyncClient() as client:
            url = f"{self.base_url}/repos/{owner}/{repo}/readme"
            response = await client.get(
                url,
                headers=self._get_headers(),
                timeout=30.0,
            )
            response.raise_for_status()
            data = response.json()

            content = ""
            if data.get("encoding") == "base64" and data.get("content"):
                content = base64.b64decode(data["content"]).decode("utf-8", errors="replace")

            return {
                "name": data.get("name", ""),
                "content": content,
                "size": data.get("size", 0),
            }

    async def _search_code(
        self, owner: str, repo: str, query: str
    ) -> dict[str, Any]:
        """Search code in repository."""
        async with httpx.AsyncClient() as client:
            response = await client.get(
                f"{self.base_url}/search/code",
                headers=self._get_headers(),
                params={"q": f"{query} repo:{owner}/{repo}"},
                timeout=30.0,
            )
            response.raise_for_status()
            data = response.json()

            results = []
            for item in data.get("items", []):
                results.append(
                    {
                        "name": item.get("name", ""),
                        "path": item.get("path", ""),
                        "url": item.get("html_url", ""),
                        "score": item.get("score", 0),
                    }
                )

            return {"query": query, "results": results, "total": data.get("total_count", 0)}