Spaces:
Sleeping
Sleeping
| import os | |
| import json | |
| from typing import Optional, List | |
| from phi.tools import Toolkit | |
| from phi.utils.log import logger | |
| try: | |
| from github import Github, GithubException, Auth | |
| except ImportError: | |
| raise ImportError("`PyGithub` not installed. Please install using `pip install PyGithub`") | |
| class GithubTools(Toolkit): | |
| def __init__( | |
| self, | |
| access_token: Optional[str] = None, | |
| base_url: Optional[str] = None, | |
| search_repositories: bool = True, | |
| list_repositories: bool = True, | |
| get_repository: bool = True, | |
| list_pull_requests: bool = True, | |
| get_pull_request: bool = True, | |
| get_pull_request_changes: bool = True, | |
| create_issue: bool = True, | |
| create_repository: bool = True, | |
| get_repository_languages: bool = True, | |
| ): | |
| super().__init__(name="github") | |
| self.access_token = access_token or os.getenv("GITHUB_ACCESS_TOKEN") | |
| self.base_url = base_url | |
| self.g = self.authenticate() | |
| if search_repositories: | |
| self.register(self.search_repositories) | |
| if list_repositories: | |
| self.register(self.list_repositories) | |
| if get_repository: | |
| self.register(self.get_repository) | |
| if list_pull_requests: | |
| self.register(self.list_pull_requests) | |
| if get_pull_request: | |
| self.register(self.get_pull_request) | |
| if get_pull_request_changes: | |
| self.register(self.get_pull_request_changes) | |
| if create_issue: | |
| self.register(self.create_issue) | |
| if create_repository: | |
| self.register(self.create_repository) | |
| if get_repository_languages: | |
| self.register(self.get_repository_languages) | |
| def authenticate(self): | |
| """Authenticate with GitHub using the provided access token.""" | |
| if not self.access_token: # Fixes lint type error | |
| raise ValueError("GitHub access token is required") | |
| auth = Auth.Token(self.access_token) | |
| if self.base_url: | |
| logger.debug(f"Authenticating with GitHub Enterprise at {self.base_url}") | |
| return Github(base_url=self.base_url, auth=auth) | |
| else: | |
| logger.debug("Authenticating with public GitHub") | |
| return Github(auth=auth) | |
| def search_repositories(self, query: str, sort: str = "stars", order: str = "desc", per_page: int = 5) -> str: | |
| """Search for repositories on GitHub. | |
| Args: | |
| query (str): The search query keywords. | |
| sort (str, optional): The field to sort results by. Can be 'stars', 'forks', or 'updated'. Defaults to 'stars'. | |
| order (str, optional): The order of results. Can be 'asc' or 'desc'. Defaults to 'desc'. | |
| per_page (int, optional): Number of results per page. Defaults to 5. | |
| Returns: | |
| A JSON-formatted string containing a list of repositories matching the search query. | |
| """ | |
| logger.debug(f"Searching repositories with query: '{query}'") | |
| try: | |
| repositories = self.g.search_repositories(query=query, sort=sort, order=order) | |
| repo_list = [] | |
| for repo in repositories[:per_page]: | |
| repo_info = { | |
| "full_name": repo.full_name, | |
| "description": repo.description, | |
| "url": repo.html_url, | |
| "stars": repo.stargazers_count, | |
| "forks": repo.forks_count, | |
| "language": repo.language, | |
| } | |
| repo_list.append(repo_info) | |
| return json.dumps(repo_list, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error searching repositories: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def list_repositories(self) -> str: | |
| """List all repositories for the authenticated user. | |
| Returns: | |
| A JSON-formatted string containing a list of repository names. | |
| """ | |
| logger.debug("Listing repositories") | |
| try: | |
| repos = self.g.get_user().get_repos() | |
| repo_names = [repo.full_name for repo in repos] | |
| return json.dumps(repo_names, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error listing repositories: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def create_repository( | |
| self, | |
| name: str, | |
| private: bool = False, | |
| description: Optional[str] = None, | |
| auto_init: bool = False, | |
| organization: Optional[str] = None, | |
| ) -> str: | |
| """Create a new repository on GitHub. | |
| Args: | |
| name (str): The name of the repository. | |
| private (bool, optional): Whether the repository is private. Defaults to False. | |
| description (str, optional): A short description of the repository. | |
| auto_init (bool, optional): Whether to initialize the repository with a README. Defaults to False. | |
| organization (str, optional): Name of organization to create repo in. If None, creates in user account. | |
| Returns: | |
| A JSON-formatted string containing the created repository details. | |
| """ | |
| logger.debug(f"Creating repository: {name}") | |
| try: | |
| description = description if description is not None else "" | |
| if organization: | |
| logger.debug(f"Creating in organization: {organization}") | |
| org = self.g.get_organization(organization) | |
| repo = org.create_repo( | |
| name=name, | |
| private=private, | |
| description=description, | |
| auto_init=auto_init, | |
| ) | |
| else: | |
| repo = self.g.get_user().create_repo( | |
| name=name, | |
| private=private, | |
| description=description, | |
| auto_init=auto_init, | |
| ) | |
| repo_info = { | |
| "name": repo.full_name, | |
| "url": repo.html_url, | |
| "private": repo.private, | |
| "description": repo.description, | |
| } | |
| return json.dumps(repo_info, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error creating repository: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def get_repository(self, repo_name: str) -> str: | |
| """Get details of a specific repository. | |
| Args: | |
| repo_name (str): The full name of the repository (e.g., 'owner/repo'). | |
| Returns: | |
| A JSON-formatted string containing repository details. | |
| """ | |
| logger.debug(f"Getting repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| repo_info = { | |
| "name": repo.full_name, | |
| "description": repo.description, | |
| "url": repo.html_url, | |
| "stars": repo.stargazers_count, | |
| "forks": repo.forks_count, | |
| "open_issues": repo.open_issues_count, | |
| "language": repo.language, | |
| "license": repo.license.name if repo.license else None, | |
| "default_branch": repo.default_branch, | |
| } | |
| return json.dumps(repo_info, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error getting repository: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def get_repository_languages(self, repo_name: str) -> str: | |
| """Get the languages used in a repository. | |
| Args: | |
| repo_name (str): The full name of the repository (e.g., 'owner/repo'). | |
| Returns: | |
| A JSON-formatted string containing the list of languages. | |
| """ | |
| logger.debug(f"Getting languages for repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| languages = repo.get_languages() | |
| return json.dumps(languages, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error getting repository languages: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def list_pull_requests(self, repo_name: str, state: str = "open") -> str: | |
| """List pull requests for a repository. | |
| Args: | |
| repo_name (str): The full name of the repository (e.g., 'owner/repo'). | |
| state (str, optional): The state of the PRs to list ('open', 'closed', 'all'). Defaults to 'open'. | |
| Returns: | |
| A JSON-formatted string containing a list of pull requests. | |
| """ | |
| logger.debug(f"Listing pull requests for repository: {repo_name} with state: {state}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| pulls = repo.get_pulls(state=state) | |
| pr_list = [] | |
| for pr in pulls: | |
| pr_info = { | |
| "number": pr.number, | |
| "title": pr.title, | |
| "user": pr.user.login, | |
| "created_at": pr.created_at.isoformat(), | |
| "state": pr.state, | |
| "url": pr.html_url, | |
| } | |
| pr_list.append(pr_info) | |
| return json.dumps(pr_list, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error listing pull requests: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def get_pull_request(self, repo_name: str, pr_number: int) -> str: | |
| """Get details of a specific pull request. | |
| Args: | |
| repo_name (str): The full name of the repository (e.g., 'owner/repo'). | |
| pr_number (int): The number of the pull request. | |
| Returns: | |
| A JSON-formatted string containing pull request details. | |
| """ | |
| logger.debug(f"Getting pull request #{pr_number} for repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| pr = repo.get_pull(pr_number) | |
| pr_info = { | |
| "number": pr.number, | |
| "title": pr.title, | |
| "user": pr.user.login, | |
| "body": pr.body, | |
| "created_at": pr.created_at.isoformat(), | |
| "updated_at": pr.updated_at.isoformat(), | |
| "state": pr.state, | |
| "merged": pr.is_merged(), | |
| "mergeable": pr.mergeable, | |
| "url": pr.html_url, | |
| } | |
| return json.dumps(pr_info, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error getting pull request: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def get_pull_request_changes(self, repo_name: str, pr_number: int) -> str: | |
| """Get the changes (files modified) in a pull request. | |
| Args: | |
| repo_name (str): The full name of the repository (e.g., 'owner/repo'). | |
| pr_number (int): The number of the pull request. | |
| Returns: | |
| A JSON-formatted string containing the list of changed files. | |
| """ | |
| logger.debug(f"Getting changes for pull request #{pr_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| pr = repo.get_pull(pr_number) | |
| files = pr.get_files() | |
| changes = [] | |
| for file in files: | |
| file_info = { | |
| "filename": file.filename, | |
| "status": file.status, | |
| "additions": file.additions, | |
| "deletions": file.deletions, | |
| "changes": file.changes, | |
| "raw_url": file.raw_url, | |
| "blob_url": file.blob_url, | |
| "patch": file.patch, | |
| } | |
| changes.append(file_info) | |
| return json.dumps(changes, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error getting pull request changes: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def create_issue(self, repo_name: str, title: str, body: Optional[str] = None) -> str: | |
| """Create an issue in a repository. | |
| Args: | |
| repo_name (str): The full name of the repository (e.g., 'owner/repo'). | |
| title (str): The title of the issue. | |
| body (str, optional): The body content of the issue. | |
| Returns: | |
| A JSON-formatted string containing the created issue details. | |
| """ | |
| logger.debug(f"Creating issue in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.create_issue(title=title, body=body) | |
| issue_info = { | |
| "id": issue.id, | |
| "number": issue.number, | |
| "title": issue.title, | |
| "body": issue.body, | |
| "url": issue.html_url, | |
| "state": issue.state, | |
| "created_at": issue.created_at.isoformat(), | |
| "user": issue.user.login, | |
| } | |
| return json.dumps(issue_info, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error creating issue: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def list_issues(self, repo_name: str, state: str = "open") -> str: | |
| """List issues for a repository. | |
| Args: | |
| repo_name (str): The full name of the repository (e.g., 'owner/repo'). | |
| state (str, optional): The state of issues to list ('open', 'closed', 'all'). Defaults to 'open'. | |
| Returns: | |
| A JSON-formatted string containing a list of issues. | |
| """ | |
| logger.debug(f"Listing issues for repository: {repo_name} with state: {state}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issues = repo.get_issues(state=state) | |
| # Filter out pull requests after fetching issues | |
| filtered_issues = [issue for issue in issues if not issue.pull_request] | |
| issue_list = [] | |
| for issue in filtered_issues: | |
| issue_info = { | |
| "number": issue.number, | |
| "title": issue.title, | |
| "user": issue.user.login, | |
| "created_at": issue.created_at.isoformat(), | |
| "state": issue.state, | |
| "url": issue.html_url, | |
| } | |
| issue_list.append(issue_info) | |
| return json.dumps(issue_list, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error listing issues: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def get_issue(self, repo_name: str, issue_number: int) -> str: | |
| """Get details of a specific issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| Returns: | |
| A JSON-formatted string containing issue details. | |
| """ | |
| logger.debug(f"Getting issue #{issue_number} for repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| issue_info = { | |
| "number": issue.number, | |
| "title": issue.title, | |
| "body": issue.body, | |
| "user": issue.user.login, | |
| "state": issue.state, | |
| "created_at": issue.created_at.isoformat(), | |
| "updated_at": issue.updated_at.isoformat(), | |
| "url": issue.html_url, | |
| "assignees": [assignee.login for assignee in issue.assignees], | |
| "labels": [label.name for label in issue.labels], | |
| } | |
| return json.dumps(issue_info, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error getting issue: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def comment_on_issue(self, repo_name: str, issue_number: int, comment_body: str) -> str: | |
| """Add a comment to an issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| comment_body (str): The content of the comment. | |
| Returns: | |
| A JSON-formatted string containing the comment details. | |
| """ | |
| logger.debug(f"Adding comment to issue #{issue_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| comment = issue.create_comment(body=comment_body) | |
| comment_info = { | |
| "id": comment.id, | |
| "body": comment.body, | |
| "user": comment.user.login, | |
| "created_at": comment.created_at.isoformat(), | |
| "url": comment.html_url, | |
| } | |
| return json.dumps(comment_info, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error commenting on issue: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def close_issue(self, repo_name: str, issue_number: int) -> str: | |
| """Close an issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| Returns: | |
| A JSON-formatted string confirming the issue is closed. | |
| """ | |
| logger.debug(f"Closing issue #{issue_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| issue.edit(state="closed") | |
| return json.dumps({"message": f"Issue #{issue_number} closed."}, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error closing issue: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def reopen_issue(self, repo_name: str, issue_number: int) -> str: | |
| """Reopen a closed issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| Returns: | |
| A JSON-formatted string confirming the issue is reopened. | |
| """ | |
| logger.debug(f"Reopening issue #{issue_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| issue.edit(state="open") | |
| return json.dumps({"message": f"Issue #{issue_number} reopened."}, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error reopening issue: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def assign_issue(self, repo_name: str, issue_number: int, assignees: List[str]) -> str: | |
| """Assign users to an issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| assignees (List[str]): A list of usernames to assign. | |
| Returns: | |
| A JSON-formatted string confirming the assignees. | |
| """ | |
| logger.debug(f"Assigning users to issue #{issue_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| issue.edit(assignees=assignees) | |
| return json.dumps({"message": f"Issue #{issue_number} assigned to {assignees}."}, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error assigning issue: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def label_issue(self, repo_name: str, issue_number: int, labels: List[str]) -> str: | |
| """Add labels to an issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| labels (List[str]): A list of label names to add. | |
| Returns: | |
| A JSON-formatted string confirming the labels. | |
| """ | |
| logger.debug(f"Labeling issue #{issue_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| issue.edit(labels=labels) | |
| return json.dumps({"message": f"Labels {labels} added to issue #{issue_number}."}, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error labeling issue: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def list_issue_comments(self, repo_name: str, issue_number: int) -> str: | |
| """List comments on an issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| Returns: | |
| A JSON-formatted string containing a list of comments. | |
| """ | |
| logger.debug(f"Listing comments for issue #{issue_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| comments = issue.get_comments() | |
| comment_list = [] | |
| for comment in comments: | |
| comment_info = { | |
| "id": comment.id, | |
| "user": comment.user.login, | |
| "body": comment.body, | |
| "created_at": comment.created_at.isoformat(), | |
| "url": comment.html_url, | |
| } | |
| comment_list.append(comment_info) | |
| return json.dumps(comment_list, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error listing issue comments: {e}") | |
| return json.dumps({"error": str(e)}) | |
| def edit_issue( | |
| self, repo_name: str, issue_number: int, title: Optional[str] = None, body: Optional[str] = None | |
| ) -> str: | |
| """Edit the title or body of an issue. | |
| Args: | |
| repo_name (str): The full name of the repository. | |
| issue_number (int): The number of the issue. | |
| title (str, optional): The new title for the issue. | |
| body (str, optional): The new body content for the issue. | |
| Returns: | |
| A JSON-formatted string confirming the issue has been updated. | |
| """ | |
| logger.debug(f"Editing issue #{issue_number} in repository: {repo_name}") | |
| try: | |
| repo = self.g.get_repo(repo_name) | |
| issue = repo.get_issue(number=issue_number) | |
| issue.edit(title=title, body=body) | |
| return json.dumps({"message": f"Issue #{issue_number} updated."}, indent=2) | |
| except GithubException as e: | |
| logger.error(f"Error editing issue: {e}") | |
| return json.dumps({"error": str(e)}) | |