# tools/github_tool.py from .base_tool import BaseTool import requests import os import base64 import logging class GitHubTool(BaseTool): def __init__(self, session=None, token=None, repo=None, base_url=None, initial_branch="main", logger=None): self.base_url = base_url if base_url else "https://api.github.com" self._token = token if token else os.environ.get("GITHUB_TOKEN") self._repo = repo if repo else os.environ.get("GITHUB_REPOSITORY") if not self._token: # In a real scenario, might raise an error or operate in a degraded mode. # For this tool, token is essential. raise ValueError("GitHub token must be provided either as an argument or via GITHUB_TOKEN env var.") if not self._repo: raise ValueError("GitHub repository (e.g., 'owner/repo') must be provided either as an argument or via GITHUB_REPOSITORY env var.") if session: self.session = session else: self.session = requests.Session() self.session.headers.update({ "Authorization": f"token {self._token}", "Accept": "application/vnd.github.v3+json" }) self.current_branch = initial_branch # Use provided logger or get a new one for the module # The application using this tool should configure the logging handlers and formatting. self.logger = logger if logger else logging.getLogger(__name__) # If no handlers are configured by the application, add a NullHandler # to prevent "No handler found" warnings if the tool logs something. if not self.logger.handlers: self.logger.addHandler(logging.NullHandler()) def clear(self): if self.current_branch != "main": self._set_current_branch("main") self.logger.info(f"GitHubTool state cleared. Current branch is {self.current_branch}") def get_functions(self): return [ { "type": "function", "function": { "name": "read_file", "description": "Read a file from the repository", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "Path to the file in the repository"} }, "required": ["path"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "list_files", "description": "List files in a directory of the repository", "parameters": { "type": "object", "properties": { "path": {"type": "string", "description": "Path to the directory in the repository"} }, "required": ["path"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "search_code", "description": "Search for code in the repository", "parameters": { "type": "object", "properties": { "query": {"type": "string", "description": "Search query"} }, "required": ["query"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "create_branch", "description": "Create a new branch in the repository", "parameters": { "type": "object", "properties": { "branch_name": {"type": "string", "description": "Name of the new branch"}, "base_branch": {"type": "string", "description": "Name of the base branch", "default": "main"} }, "required": ["branch_name"] } }, "_tags": ["write"] }, { "type": "function", "function": { "name": "commit_file", "description": "Commit a file to a branch (not main)", "parameters": { "type": "object", "properties": { "file_path": {"type": "string", "description": "Path to the file in the repository"}, "commit_message": {"type": "string", "description": "Commit message"}, "content": {"type": "string", "description": "Content of the file"} }, "required": ["file_path", "commit_message", "content"] } }, "_tags": ["write"] }, { "type": "function", "function": { "name": "create_pull_request", "description": "Create a pull request", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "Title of the pull request"}, "body": {"type": "string", "description": "Body of the pull request"}, "base": {"type": "string", "description": "The name of the branch you want the changes pulled into", "default": "main"} }, "required": ["title", "body"] } }, "_tags": ["write"] }, { "type": "function", "function": { "name": "get_commit_history", "description": "Get commit history for a file", "parameters": { "type": "object", "properties": { "file_path": {"type": "string", "description": "Path to the file in the repository"}, "num_commits": {"type": "integer", "description": "Number of commits to retrieve", "default": 10} }, "required": ["file_path"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "view_commit_details_for_file", "description": "View commit history and details for a specific file, including commit messages.", "parameters": { "type": "object", "properties": { "file_path": {"type": "string", "description": "Path to the file in the repository"}, "num_commits": {"type": "integer", "description": "Number of commits to retrieve", "default": 10} }, "required": ["file_path"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "get_branch_sha", "description": "Get the SHA of the latest commit on a branch", "parameters": { "type": "object", "properties": { "branch": {"type": "string", "description": "Name of the branch"} }, "required": ["branch"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "get_current_branch", "description": "Get the name of the current branch", "parameters": { "type": "object", "properties": {} } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "set_current_branch", "description": "Set the current branch", "parameters": { "type": "object", "properties": { "branch_name": {"type": "string", "description": "Name of the branch to set as current"} }, "required": ["branch_name"] } }, "_tags": ["read", "write"] }, { "type": "function", "function": { "name": "get_file_at_commit", "description": "Get the contents of a file at a specific commit", "parameters": { "type": "object", "properties": { "file_path": {"type": "string", "description": "Path to the file in the repository"}, "commit_sha": {"type": "string", "description": "SHA of the commit to retrieve the file from"} }, "required": ["file_path", "commit_sha"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "list_branches", "description": "List all branches in the repository", "parameters": { "type": "object", "properties": { "per_page": {"type": "integer", "description": "Number of branches to return per page (max 100)", "default": 100}, "all_pages": {"type": "boolean", "description": "Whether to fetch all pages of results", "default": True} } } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "approve_pull_request", "description": "Approve a pull request", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"} }, "required": ["pull_number"] } }, "_tags": ["write"] }, { "type": "function", "function": { "name": "close_pull_request", "description": "Close a pull request", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"} }, "required": ["pull_number"] } }, "_tags": ["write"] }, { "type": "function", "function": { "name": "merge_pull_request", "description": "Merge a pull request", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"}, "commit_title": {"type": "string", "description": "Title for the automatic commit message", "default": "Merge pull request"}, "commit_message": {"type": "string", "description": "Extra detail to append to automatic commit message", "default": ""}, "merge_method": { "type": "string", "description": "Merge method to use", "enum": ["merge", "squash", "rebase"], "default": "merge" } }, "required": ["pull_number"] } }, "_tags": ["write"] }, { "type": "function", "function": { "name": "delete_branch", "description": "Delete a branch", "parameters": { "type": "object", "properties": { "branch_name": {"type": "string", "description": "Name of the branch to delete"} }, "required": ["branch_name"] } }, "_tags": ["write"] }, { "type": "function", "function": { "name": "get_issue_details", "description": "Get details of a specific issue", "parameters": { "type": "object", "properties": { "issue_number": {"type": "integer", "description": "The number of the issue"} }, "required": ["issue_number"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "create_issue", "description": "Create a new issue in the repository", "parameters": { "type": "object", "properties": { "title": {"type": "string", "description": "Title of the issue"}, "body": {"type": "string", "description": "Body of the issue"}, "labels": { "type": "array", "items": {"type": "string"}, "description": "Labels to apply to the issue" } }, "required": ["title", "body"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "list_issues", "description": "List issues in the repository", "parameters": { "type": "object", "properties": { "state": {"type": "string", "enum": ["open", "closed", "all"], "default": "open", "description": "State of the issues to retrieve"}, "per_page": {"type": "integer", "default": 30, "description": "Number of issues to return per page"}, "page": {"type": "integer", "default": 1, "description": "Page number of the results to fetch"} } } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "add_issue_comment", "description": "Add a comment to an issue", "parameters": { "type": "object", "properties": { "issue_number": {"type": "integer", "description": "The number of the issue"}, "comment": {"type": "string", "description": "The comment to add to the issue"} }, "required": ["issue_number", "comment"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "get_issue_comments", "description": "Get comments for an issue", "parameters": { "type": "object", "properties": { "issue_number": {"type": "integer", "description": "The number of the issue"} }, "required": ["issue_number"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "get_pull_request_general_comments", "description": "Get general comments posted on a pull request itself (not specific to file lines).", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request."} }, "required": ["pull_number"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "create_project_board", "description": "Create a new project board", "parameters": { "type": "object", "properties": { "name": {"type": "string", "description": "Name of the project board"}, "body": {"type": "string", "description": "Body of the project board"} }, "required": ["name"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "create_project_column", "description": "Create a new column in a project board", "parameters": { "type": "object", "properties": { "project_id": {"type": "integer", "description": "ID of the project board"}, "column_name": {"type": "string", "description": "Name of the column"} }, "required": ["project_id", "column_name"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "create_project_card", "description": "Create a new card in a project column", "parameters": { "type": "object", "properties": { "column_id": {"type": "integer", "description": "ID of the project column"}, "note": {"type": "string", "description": "Note for the project card"} }, "required": ["column_id", "note"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "move_project_card", "description": "Move a card to a new position", "parameters": { "type": "object", "properties": { "card_id": {"type": "integer", "description": "ID of the project card"}, "position": {"type": "string", "description": "New position of the card"}, "column_id": {"type": "integer", "description": "ID of the target column"} }, "required": ["card_id", "position", "column_id"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "link_issue_to_project_card", "description": "Link an issue or pull request to a project card", "parameters": { "type": "object", "properties": { "card_id": {"type": "integer", "description": "ID of the project card"}, "content_id": {"type": "integer", "description": "ID of the issue or pull request"}, "content_type": {"type": "string", "description": "Type of the content (Issue or PullRequest)"} }, "required": ["card_id", "content_id", "content_type"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "list_project_boards", "description": "List project boards associated with the repository", "parameters": { "type": "object", "properties": {} } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "view_project_board_items", "description": "View items (columns and cards) in a specific project board", "parameters": { "type": "object", "properties": { "project_id": {"type": "integer", "description": "ID of the project board"} }, "required": ["project_id"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "get_pull_request_details", "description": "Get detailed information about a pull request", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"} }, "required": ["pull_number"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "get_pull_request_diff", "description": "Get the diff of a pull request", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"} }, "required": ["pull_number"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "get_pull_request_files", "description": "Get a list of files changed in a pull request, with their patch details", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"} }, "required": ["pull_number"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "create_pull_request_review_comment", "description": "Add a comment to a specific line of a file in a pull request review", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"}, "body": {"type": "string", "description": "The text of the comment"}, "commit_id": {"type": "string", "description": "The SHA of the commit to comment on"}, "path": {"type": "string", "description": "The path to the file being commented on"}, "position": {"type": "integer", "description": "The line index in the diff to comment on (starting at 1)"}, "side": {"type": "string", "enum": ["LEFT", "RIGHT"], "description": "The side of the diff to comment on (LEFT for old file, RIGHT for new file)", "default": "RIGHT"}, "start_line": {"type": "integer", "description": "The start line of the diff if commenting on a range"}, "start_side": {"type": "string", "enum": ["LEFT", "RIGHT"], "description": "The side of the diff for the start line"} }, "required": ["pull_number", "body", "commit_id", "path", "position"] } }, "_tags": ["communicate"] }, { "type": "function", "function": { "name": "list_pull_request_review_comments", "description": "List comments on a pull request review", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"} }, "required": ["pull_number"] } }, "_tags": ["read"] }, { "type": "function", "function": { "name": "submit_pull_request_review", "description": "Submit a formal pull request review (APPROVE, REQUEST_CHANGES, COMMENT)", "parameters": { "type": "object", "properties": { "pull_number": {"type": "integer", "description": "The number of the pull request"}, "event": {"type": "string", "enum": ["APPROVE", "REQUEST_CHANGES", "COMMENT"], "description": "The type of review event"}, "body": {"type": "string", "description": "The body of the review (required for REQUEST_CHANGES, optional for others)"} }, "required": ["pull_number", "event"] } }, "_tags": ["communicate"] } ] def execute(self, function_name, **kwargs): self.logger.info(f"Executing GitHub Tool function: {function_name} with args: {kwargs}") # Dispatch to the appropriate private method method_name = f"_{function_name}" if hasattr(self, method_name): method = getattr(self, method_name) try: return method(**kwargs) # Ensure only expected args are passed if method signature is strict except Exception as e: self.logger.error(f"Error executing {method_name}: {e}", exc_info=True) return f"Error during {function_name} execution: {str(e)}" else: error_message = f"Unknown function: {function_name}" self.logger.error(error_message) return error_message # Private methods for each function, using self.session for HTTP requests def _read_file(self, path): self.logger.info(f"Reading file: {path} from branch: {self.current_branch}") url = f"{self.base_url}/repos/{self._repo}/contents/{path}" response = self.session.get(url, params={"ref": self.current_branch}) if response.status_code == 200: content = response.json()["content"] decoded_content = base64.b64decode(content).decode('utf-8') self.logger.info(f"Successfully read file: {path}") return decoded_content else: error_message = f"Error reading file ({path}): {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _create_branch(self, branch_name, base_branch="main"): self.logger.info(f"Creating branch: {branch_name} from base: {base_branch}") # Get SHA of base branch ref_url = f"{self.base_url}/repos/{self._repo}/git/refs/heads/{base_branch}" response_sha = self.session.get(ref_url) if response_sha.status_code != 200: error_message = f"Error getting base branch SHA ({base_branch}): {response_sha.status_code} - {response_sha.text}" self.logger.error(error_message) return error_message sha = response_sha.json()["object"]["sha"] # Create new branch create_ref_url = f"{self.base_url}/repos/{self._repo}/git/refs" data = {"ref": f"refs/heads/{branch_name}", "sha": sha} response_create = self.session.post(create_ref_url, json=data) if response_create.status_code == 201: self.current_branch = branch_name success_message = f"Branch '{branch_name}' created successfully from '{base_branch}' and set as current branch." self.logger.info(success_message) return success_message else: error_message = f"Error creating branch '{branch_name}': {response_create.status_code} - {response_create.text}" self.logger.error(error_message) return error_message def _commit_file(self, file_path, content, commit_message): self.logger.info(f"Committing file: {file_path} to branch: {self.current_branch} with message: '{commit_message}'") if self.current_branch == "main": error_message = "Action directly to main branch is not allowed. Please create and switch to a new branch first." self.logger.warning(error_message) return error_message url = f"{self.base_url}/repos/{self._repo}/contents/{file_path}" encoded_content = base64.b64encode(content.encode('utf-8')).decode('utf-8') data = { "message": commit_message, "content": encoded_content, "branch": self.current_branch } # Check if file exists to get its SHA for update self.logger.info(f"Checking if file '{file_path}' exists on branch '{self.current_branch}'") get_response = self.session.get(url, params={"ref": self.current_branch}) if get_response.status_code == 200: data["sha"] = get_response.json()["sha"] self.logger.info(f"File '{file_path}' exists, will update.") elif get_response.status_code == 404: self.logger.info(f"File '{file_path}' does not exist, will create.") else: error_message = f"Error checking file existence for '{file_path}': {get_response.status_code} - {get_response.text}" self.logger.error(error_message) return error_message response = self.session.put(url, json=data) if response.status_code in [200, 201]: # 200 for update, 201 for create commit_sha = response.json().get("commit", {}).get("sha", "N/A") success_message = f"File '{file_path}' committed successfully to branch '{self.current_branch}'. Commit SHA: {commit_sha}" self.logger.info(success_message) return success_message else: error_message = f"Error committing file '{file_path}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _create_pull_request(self, title, body, base="main"): self.logger.info(f"Creating pull request: '{title}' from branch '{self.current_branch}' to '{base}'") if self.current_branch == base: error_message = f"Cannot create a pull request from branch '{self.current_branch}' to itself ('{base}')." self.logger.warning(error_message) return error_message url = f"{self.base_url}/repos/{self._repo}/pulls" data = {"title": title, "body": body, "head": self.current_branch, "base": base} response = self.session.post(url, json=data) if response.status_code == 201: pr_html_url = response.json().get("html_url", "N/A") pr_number = response.json().get("number", "N/A") success_message = f"Pull request '{title}' created successfully: {pr_html_url} (Number: {pr_number})" self.logger.info(success_message) return success_message else: error_message = f"Error creating pull request: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _get_branch_sha(self, branch): self.logger.info(f"Getting SHA for branch: {branch}") url = f"{self.base_url}/repos/{self._repo}/git/refs/heads/{branch}" response = self.session.get(url) if response.status_code == 200: sha = response.json()["object"]["sha"] self.logger.info(f"SHA for branch '{branch}' is {sha}") return sha else: error_message = f"Error getting SHA for branch '{branch}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _list_files(self, path): self.logger.info(f"Listing files in path: '{path}' on branch: '{self.current_branch}'") url = f"{self.base_url}/repos/{self._repo}/contents/{path.strip('/')}" # Ensure no leading/trailing slashes for consistency response = self.session.get(url, params={"ref": self.current_branch}) if response.status_code == 200: items = response.json() results = [] if isinstance(items, list): # It's a directory listing for item in items: results.append({"name": item["name"], "type": item["type"], "path": item["path"]}) elif isinstance(items, dict) and 'type' in items: # It's a single file response results.append({"name": items["name"], "type": items["type"], "path": items["path"]}) self.logger.info(f"Successfully listed {len(results)} items in '{path}'.") return results elif response.status_code == 404: self.logger.warning(f"Path '{path}' not found on branch '{self.current_branch}'.") return f"Error: Path '{path}' not found." else: error_message = f"Error listing files in '{path}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _search_code(self, query): self.logger.info(f"Searching code with query: '{query}' in repo: '{self._repo}'") url = f"{self.base_url}/search/code" params = {"q": f"{query} repo:{self._repo}"} response = self.session.get(url, params=params) if response.status_code == 200: search_results = response.json().get("items", []) results = [{"path": item["path"], "url": item["html_url"]} for item in search_results] self.logger.info(f"Code search for '{query}' found {len(results)} items.") return results else: error_message = f"Error searching code for '{query}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _get_commit_history(self, file_path, num_commits=10): self.logger.info(f"Getting last {num_commits} commit(s) for file: '{file_path}' on branch '{self.current_branch}'") url = f"{self.base_url}/repos/{self._repo}/commits" params = {"path": file_path, "sha": self.current_branch, "per_page": num_commits} response = self.session.get(url, params=params) if response.status_code == 200: commits_data = response.json() commits = [{ "sha": commit["sha"], "message": commit["commit"]["message"], "author": commit["commit"]["author"]["name"], "date": commit["commit"]["author"]["date"] } for commit in commits_data] self.logger.info(f"Successfully retrieved {len(commits)} commit(s) for '{file_path}'.") return commits else: error_message = f"Error getting commit history for '{file_path}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _view_commit_details_for_file(self, file_path, num_commits=10): # This function is essentially the same as get_commit_history based on its description. self.logger.info(f"Viewing commit details for file '{file_path}' (last {num_commits} commits) - using _get_commit_history.") return self._get_commit_history(file_path, num_commits) def _get_current_branch(self): self.logger.info(f"Current branch is: {self.current_branch}") return self.current_branch def _set_current_branch(self, branch_name): self.logger.info(f"Attempting to set current branch to: {branch_name}") # Check if branch exists by trying to get its SHA sha_info = self._get_branch_sha(branch_name) if isinstance(sha_info, str) and sha_info.startswith("Error getting SHA"): # Crude check for error string error_message = f"Cannot set current branch: Branch '{branch_name}' not found or error accessing it. Details: {sha_info}" self.logger.warning(error_message) return error_message self.current_branch = branch_name success_message = f"Current branch set to: {self.current_branch}" self.logger.info(success_message) return success_message def _get_file_at_commit(self, file_path, commit_sha): self.logger.info(f"Getting file '{file_path}' at commit SHA: {commit_sha}") url = f"{self.base_url}/repos/{self._repo}/contents/{file_path}" response = self.session.get(url, params={"ref": commit_sha}) if response.status_code == 200: content = response.json()["content"] decoded_content = base64.b64decode(content).decode('utf-8') self.logger.info(f"Successfully retrieved file '{file_path}' at commit {commit_sha}.") return decoded_content else: error_message = f"Error reading file '{file_path}' at commit {commit_sha}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _list_branches(self, per_page=100, all_pages=True): self.logger.info(f"Listing branches for repo '{self._repo}'. Per_page={per_page}, All_pages={all_pages}") url = f"{self.base_url}/repos/{self._repo}/branches" params = {"per_page": min(per_page, 100)} # Respect GitHub API limit branches_list = [] page = 1 while url: self.logger.debug(f"Fetching page {page} from {url} with params {params if page==1 else {}}") response = self.session.get(url, params=params if page == 1 else None) # params only for first page if paginating via links if response.status_code != 200: error_message = f"Error listing branches: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message current_page_branches = [branch["name"] for branch in response.json()] branches_list.extend(current_page_branches) self.logger.debug(f"Fetched {len(current_page_branches)} branches on page {page}.") if not all_pages or not response.links.get("next"): break url = response.links["next"]["url"] page += 1 params = {} # Clear params for subsequent calls using a link that includes them self.logger.info(f"Successfully listed {len(branches_list)} branches.") return branches_list def _approve_pull_request(self, pull_number): self.logger.info(f"Approving pull request #{pull_number}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/reviews" data = {"event": "APPROVE"} response = self.session.post(url, json=data) if response.status_code == 200: success_message = f"Pull request #{pull_number} approved successfully." self.logger.info(success_message) return success_message else: error_message = f"Error approving pull request #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _close_pull_request(self, pull_number): self.logger.info(f"Closing pull request #{pull_number}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}" data = {"state": "closed"} response = self.session.patch(url, json=data) # Use PATCH for update if response.status_code == 200: success_message = f"Pull request #{pull_number} closed successfully." self.logger.info(success_message) return success_message else: error_message = f"Error closing pull request #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _merge_pull_request(self, pull_number, commit_title="Merge pull request", commit_message="", merge_method="merge"): self.logger.info(f"Merging pull request #{pull_number} using method '{merge_method}'") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/merge" data = {"commit_title": commit_title, "commit_message": commit_message, "merge_method": merge_method} response = self.session.put(url, json=data) if response.status_code == 200: success_message = f"Pull request #{pull_number} merged successfully." self.logger.info(success_message) return success_message elif response.status_code == 405: # Method Not Allowed (e.g., PR not mergeable) error_message = f"Error merging pull request #{pull_number}: Not mergeable. {response.json().get('message', response.text)}" self.logger.warning(error_message) return error_message elif response.status_code == 409: # Conflict error_message = f"Error merging pull request #{pull_number}: Merge conflict. {response.json().get('message', response.text)}" self.logger.warning(error_message) return error_message else: error_message = f"Error merging pull request #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _delete_branch(self, branch_name): self.logger.info(f"Deleting branch: {branch_name}") if branch_name == "main" or (hasattr(self, 'default_branch') and branch_name == self.default_branch) : # Add a check for a configurable default branch if necessary error_message = f"Cannot delete protected branch: {branch_name}" self.logger.warning(error_message) return error_message url = f"{self.base_url}/repos/{self._repo}/git/refs/heads/{branch_name}" response = self.session.delete(url) if response.status_code == 204: success_message = f"Branch '{branch_name}' deleted successfully." self.logger.info(success_message) if self.current_branch == branch_name: self.current_branch = "main" # Or some other default self.logger.info(f"Current branch was {branch_name}, reset to {self.current_branch}.") return success_message else: error_message = f"Error deleting branch '{branch_name}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _get_issue_details(self, issue_number): self.logger.info(f"Getting details for issue #{issue_number}") url = f"{self.base_url}/repos/{self._repo}/issues/{issue_number}" response = self.session.get(url) if response.status_code == 200: self.logger.info(f"Successfully retrieved details for issue #{issue_number}.") return response.json() # Return raw JSON data for now else: error_message = f"Error getting details for issue #{issue_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _create_issue(self, title, body, labels=None): self.logger.info(f"Creating new issue with title: '{title}'") url = f"{self.base_url}/repos/{self._repo}/issues" data = {"title": title, "body": body} if labels: # Ensure labels is a list of strings data["labels"] = labels if isinstance(labels, list) else [labels] response = self.session.post(url, json=data) if response.status_code == 201: issue_html_url = response.json().get("html_url", "N/A") issue_number = response.json().get("number", "N/A") success_message = f"Issue '{title}' created successfully: {issue_html_url} (Number: {issue_number})" self.logger.info(success_message) return success_message else: error_message = f"Error creating issue '{title}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _list_issues(self, state="open", per_page=30, page=1): self.logger.info(f"Listing issues with state: {state}, per_page: {per_page}, page: {page}") url = f"{self.base_url}/repos/{self._repo}/issues" params = {"state": state, "per_page": per_page, "page": page} response = self.session.get(url, params=params) if response.status_code == 200: issues_data = response.json() self.logger.info(f"Successfully listed {len(issues_data)} issues.") # Return a summary or full data based on needs return [{ "title": i["title"], "number": i["number"], "state": i["state"], "url": i["html_url"] } for i in issues_data] else: error_message = f"Error listing issues: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _add_issue_comment(self, issue_number, comment): self.logger.info(f"Adding comment to issue #{issue_number}: '{comment[:50]}...'") url = f"{self.base_url}/repos/{self._repo}/issues/{issue_number}/comments" data = {"body": comment} response = self.session.post(url, json=data) if response.status_code == 201: comment_html_url = response.json().get("html_url", "N/A") success_message = f"Comment added to issue #{issue_number} successfully: {comment_html_url}" self.logger.info(success_message) return success_message else: error_message = f"Error adding comment to issue #{issue_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _get_issue_comments(self, issue_number): self.logger.info(f"Getting comments for issue #{issue_number}") url = f"{self.base_url}/repos/{self._repo}/issues/{issue_number}/comments" response = self.session.get(url) if response.status_code == 200: comments_data = response.json() self.logger.info(f"Successfully retrieved {len(comments_data)} comments for issue #{issue_number}.") # Return summary or full data return [{ "user": c["user"]["login"], "body": c["body"], "created_at": c["created_at"] } for c in comments_data] else: error_message = f"Error getting comments for issue #{issue_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _get_pull_request_general_comments(self, pull_number): self.logger.info(f"Getting general comments for pull request #{pull_number}") # In GitHub API, PR comments (general, not review comments on lines) are issue comments. # The PR is also an issue, so use the issue comments endpoint. return self._get_issue_comments(issue_number=pull_number) def _create_project_board(self, name, body=None): self.logger.info(f"Creating project board: '{name}'") url = f"{self.base_url}/repos/{self._repo}/projects" headers = self.session.headers.copy() # Get existing session headers headers["Accept"] = "application/vnd.github.inertia-preview+json" # Required for Projects API data = {"name": name} if body: data["body"] = body response = self.session.post(url, headers=headers, json=data) if response.status_code == 201: project_data = response.json() success_message = f"Project board '{name}' created successfully with ID: {project_data['id']}" self.logger.info(success_message) return project_data # Return full project data else: error_message = f"Error creating project board '{name}': {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _create_project_column(self, project_id, column_name): self.logger.info(f"Creating column '{column_name}' for project ID: {project_id}") url = f"{self.base_url}/projects/{project_id}/columns" headers = self.session.headers.copy() headers["Accept"] = "application/vnd.github.inertia-preview+json" data = {"name": column_name} response = self.session.post(url, headers=headers, json=data) if response.status_code == 201: column_data = response.json() success_message = f"Column '{column_name}' created successfully for project {project_id} with ID: {column_data['id']}" self.logger.info(success_message) return column_data else: error_message = f"Error creating column '{column_name}' for project {project_id}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _create_project_card(self, column_id, note=None, content_id=None, content_type=None): self.logger.info(f"Creating card in column ID: {column_id}") url = f"{self.base_url}/projects/columns/{column_id}/cards" headers = self.session.headers.copy() headers["Accept"] = "application/vnd.github.inertia-preview+json" data = {} if note: data["note"] = note if content_id and content_type: data["content_id"] = content_id data["content_type"] = content_type elif (content_id and not content_type) or (not content_id and content_type): err = "Both content_id and content_type must be provided to link content to a project card." self.logger.warning(err) return err if not data: return "Error: Card must have a note or content to link." response = self.session.post(url, headers=headers, json=data) if response.status_code == 201: card_data = response.json() success_message = f"Card created successfully in column {column_id} with ID: {card_data['id']}" self.logger.info(success_message) return card_data else: error_message = f"Error creating card in column {column_id}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _move_project_card(self, card_id, position, column_id=None): self.logger.info(f"Moving card ID: {card_id} to position: {position}" + (f" in column ID: {column_id}" if column_id else "")) url = f"{self.base_url}/projects/columns/cards/{card_id}/moves" headers = self.session.headers.copy() headers["Accept"] = "application/vnd.github.inertia-preview+json" data = {"position": position} if column_id: data["column_id"] = column_id response = self.session.post(url, headers=headers, json=data) if response.status_code == 201: # Successful move returns 201 with empty body success_message = f"Card {card_id} moved successfully to position {position}" + (f" in column {column_id}" if column_id else ".") self.logger.info(success_message) return success_message # Return success message as body is empty else: error_message = f"Error moving card {card_id}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message # _link_issue_to_project_card is effectively handled by _create_project_card if content_id and content_type are passed. # The API used to have a separate link endpoint, but now it is part of card creation/update. # For updating an existing card to link an issue, one would PATCH the card's content_id/content_type. # Let's assume the function intends to update an existing card if it's a separate function. # However, the provided API spec for `link_issue_to_project_card` uses PATCH on card_id, so let's implement that. def _link_issue_to_project_card(self, card_id, content_id, content_type): self.logger.info(f"Linking content_id {content_id} (type: {content_type}) to card_id {card_id}") url = f"{self.base_url}/projects/cards/{card_id}" # Note: API docs suggest /projects/columns/cards/{card_id} or /projects/cards/{card_id} # Using /projects/cards/{card_id} as it seems more general for card update. headers = self.session.headers.copy() headers["Accept"] = "application/vnd.github.inertia-preview+json" data = {"content_id": content_id, "content_type": content_type} response = self.session.patch(url, headers=headers, json=data) if response.status_code == 200: updated_card = response.json() success_message = f"{content_type} {content_id} linked to card {card_id} successfully." self.logger.info(success_message) return updated_card else: error_message = f"Error linking {content_type} {content_id} to card {card_id}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _list_project_boards(self): self.logger.info(f"Listing project boards for repo: {self._repo}") url = f"{self.base_url}/repos/{self._repo}/projects" headers = self.session.headers.copy() headers["Accept"] = "application/vnd.github.inertia-preview+json" response = self.session.get(url, headers=headers) if response.status_code == 200: projects_data = response.json() self.logger.info(f"Successfully listed {len(projects_data)} project boards.") return projects_data else: error_message = f"Error listing project boards: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _view_project_board_items(self, project_id): self.logger.info(f"Viewing items for project ID: {project_id}") columns_url = f"{self.base_url}/projects/{project_id}/columns" headers = self.session.headers.copy() headers["Accept"] = "application/vnd.github.inertia-preview+json" columns_response = self.session.get(columns_url, headers=headers) if columns_response.status_code != 200: error_message = f"Error fetching columns for project {project_id}: {columns_response.status_code} - {columns_response.text}" self.logger.error(error_message) return error_message columns_data = columns_response.json() project_items = [] for column in columns_data: column_info = {"id": column["id"], "name": column["name"], "cards": []} cards_url = column["cards_url"] cards_response = self.session.get(cards_url, headers=headers) if cards_response.status_code == 200: column_info["cards"] = cards_response.json() else: self.logger.error(f"Error fetching cards for column {column['id']}('{column['name']}'): {cards_response.status_code} - {cards_response.text}") column_info["cards"] = "Error fetching cards" project_items.append(column_info) self.logger.info(f"Successfully retrieved items for project ID: {project_id}.") return project_items def _get_pull_request_details(self, pull_number): self.logger.info(f"Getting details for PR #{pull_number}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}" response = self.session.get(url) if response.status_code == 200: self.logger.info(f"Successfully retrieved details for PR #{pull_number}.") return response.json() else: error_message = f"Error getting details for PR #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _get_pull_request_diff(self, pull_number): self.logger.info(f"Getting diff for PR #{pull_number}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}" diff_headers = self.session.headers.copy() diff_headers["Accept"] = "application/vnd.github.diff" response = self.session.get(url, headers=diff_headers) if response.status_code == 200: self.logger.info(f"Successfully retrieved diff for PR #{pull_number}.") return response.text else: error_message = f"Error getting diff for PR #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _get_pull_request_files(self, pull_number): self.logger.info(f"Getting files for PR #{pull_number}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/files" response = self.session.get(url) if response.status_code == 200: self.logger.info(f"Successfully retrieved files for PR #{pull_number}.") return response.json() else: error_message = f"Error getting files for PR #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _create_pull_request_review_comment(self, pull_number, body, commit_id, path, position, side="RIGHT", start_line=None, start_side=None): self.logger.info(f"Creating review comment on PR #{pull_number}, file '{path}', position {position}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/comments" data = {"body": body, "commit_id": commit_id, "path": path, "position": position, "side": side} if start_line is not None: data["start_line"] = start_line if start_side is not None: data["start_side"] = start_side response = self.session.post(url, json=data) if response.status_code == 201: comment_url = response.json().get("html_url", "N/A") success_message = f"Review comment created successfully on PR #{pull_number}: {comment_url}" self.logger.info(success_message) return success_message else: error_message = f"Error creating review comment on PR #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _list_pull_request_review_comments(self, pull_number): self.logger.info(f"Listing review comments for PR #{pull_number}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/comments" response = self.session.get(url) if response.status_code == 200: self.logger.info(f"Successfully retrieved review comments for PR #{pull_number}.") return response.json() else: error_message = f"Error listing review comments for PR #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message def _submit_pull_request_review(self, pull_number, event, body=None): self.logger.info(f"Submitting '{event}' review for PR #{pull_number}") url = f"{self.base_url}/repos/{self._repo}/pulls/{pull_number}/reviews" data = {"event": event.upper()} # Ensure event is uppercase as per API if body: data["body"] = body response = self.session.post(url, json=data) if response.status_code == 200: review_url = response.json().get("html_url", "N/A") success_message = f"Review ({event}) submitted successfully for PR #{pull_number}: {review_url}" self.logger.info(success_message) return success_message else: error_message = f"Error submitting review for PR #{pull_number}: {response.status_code} - {response.text}" self.logger.error(error_message) return error_message