admin
21b0fff6c4
- New function commit_file_patch(file_path, commit_message, edits, base_sha=None) - Applies line-range edits on the current file content and commits the result - Preserves original newline style and supports insert/replace/delete via ranges - Prevents committing directly to main and detects stale base via optional base_sha This enables the agent to send only the minimal edits rather than entire files.
1429 lines
71 KiB
Python
1429 lines
71 KiB
Python
# 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": "read_readme",
|
|
"description": "Read the README.md file from the root of the repository",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {}
|
|
}
|
|
},
|
|
"_tags": ["read", "communicate"]
|
|
},
|
|
{
|
|
"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": "commit_file_patch",
|
|
"description": "Apply partial line-based edits to a file and commit the result (without requiring the caller to upload the entire file).",
|
|
"parameters": {
|
|
"type": "object",
|
|
"properties": {
|
|
"file_path": {"type": "string", "description": "Path to the file in the repository"},
|
|
"commit_message": {"type": "string", "description": "Commit message"},
|
|
"edits": {
|
|
"type": "array",
|
|
"description": "List of line-based edits. Each edit replaces lines [start_line..end_line] (inclusive) with 'replacement'. If end_line < start_line, the 'replacement' is inserted before start_line.",
|
|
"items": {
|
|
"type": "object",
|
|
"properties": {
|
|
"start_line": {"type": "integer", "description": "1-based start line"},
|
|
"end_line": {"type": "integer", "description": "1-based end line (inclusive), or set less than start_line for insertion"},
|
|
"replacement": {"type": "string", "description": "Replacement text for the specified range (can be multi-line)"}
|
|
},
|
|
"required": ["start_line", "end_line", "replacement"]
|
|
}
|
|
},
|
|
"base_sha": {"type": "string", "description": "Optional expected current blob SHA for the file; if provided and does not match, the operation aborts to prevent overwriting newer changes."}
|
|
},
|
|
"required": ["file_path", "commit_message", "edits"]
|
|
}
|
|
},
|
|
"_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", "communicate"]
|
|
},
|
|
{
|
|
"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", "communicate"]
|
|
},
|
|
{
|
|
"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", "communicate"]
|
|
},
|
|
{
|
|
"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", "communicate"]
|
|
},
|
|
{
|
|
"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", "communicate"]
|
|
},
|
|
{
|
|
"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", "communicate"]
|
|
},
|
|
{
|
|
"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", "communicate"]
|
|
},
|
|
{
|
|
"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", "communicate"]
|
|
},
|
|
{
|
|
"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 _read_readme(self):
|
|
self.logger.info("Reading README.md from the root of the repository.")
|
|
return self._read_file("README.md")
|
|
|
|
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):
|
|
content = content.strip("'''")
|
|
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 _commit_file_patch(self, file_path, edits, commit_message, base_sha=None):
|
|
self.logger.info(f"Committing partial edits to file: {file_path} on branch: {self.current_branch}")
|
|
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
|
|
|
|
# Fetch current file content and sha
|
|
url = f"{self.base_url}/repos/{self._repo}/contents/{file_path}"
|
|
get_response = self.session.get(url, params={"ref": self.current_branch})
|
|
if get_response.status_code == 404:
|
|
error_message = f"File '{file_path}' not found on branch '{self.current_branch}'. Cannot apply partial edits to a non-existent file."
|
|
self.logger.error(error_message)
|
|
return error_message
|
|
if get_response.status_code != 200:
|
|
error_message = f"Error reading file '{file_path}' for patching: {get_response.status_code} - {get_response.text}"
|
|
self.logger.error(error_message)
|
|
return error_message
|
|
|
|
current_sha = get_response.json()["sha"]
|
|
if base_sha and base_sha != current_sha:
|
|
error_message = (
|
|
f"Abort: base_sha mismatch for '{file_path}'. Expected {base_sha}, current {current_sha}. "
|
|
"Please refresh the file content and re-apply your edits."
|
|
)
|
|
self.logger.warning(error_message)
|
|
return error_message
|
|
|
|
content_b64 = get_response.json()["content"]
|
|
decoded_content = base64.b64decode(content_b64).decode('utf-8')
|
|
|
|
# Apply edits
|
|
try:
|
|
new_content = self._apply_line_edits(decoded_content, edits)
|
|
except Exception as e:
|
|
self.logger.error(f"Failed to apply edits to '{file_path}': {e}", exc_info=True)
|
|
return f"Error applying edits: {str(e)}"
|
|
|
|
if new_content == decoded_content:
|
|
msg = f"No changes detected after applying edits to '{file_path}'. Skipping commit."
|
|
self.logger.info(msg)
|
|
return msg
|
|
|
|
# Commit updated content using the Contents API
|
|
encoded_content = base64.b64encode(new_content.encode('utf-8')).decode('utf-8')
|
|
data = {
|
|
"message": commit_message,
|
|
"content": encoded_content,
|
|
"branch": self.current_branch,
|
|
"sha": current_sha
|
|
}
|
|
put_response = self.session.put(url, json=data)
|
|
if put_response.status_code in [200, 201]:
|
|
commit_sha = put_response.json().get("commit", {}).get("sha", "N/A")
|
|
success_message = f"Partial edits committed to '{file_path}' on branch '{self.current_branch}'. Commit SHA: {commit_sha}"
|
|
self.logger.info(success_message)
|
|
return success_message
|
|
else:
|
|
error_message = f"Error committing partial edits to '{file_path}': {put_response.status_code} - {put_response.text}"
|
|
self.logger.error(error_message)
|
|
return error_message
|
|
|
|
def _apply_line_edits(self, original_text, edits):
|
|
"""
|
|
Apply line-based edits to the provided text.
|
|
- Lines are 1-based.
|
|
- For each edit: replace lines [start_line..end_line] inclusive with 'replacement'.
|
|
If end_line < start_line, insert 'replacement' before start_line (no deletion).
|
|
- Preserves the file's original newline style (\n or \r\n) for joins.
|
|
"""
|
|
# Determine newline style
|
|
newline = "\r\n" if "\r\n" in original_text else "\n"
|
|
ends_with_newline = original_text.endswith("\n") or original_text.endswith("\r\n")
|
|
|
|
# Normalize the working representation to a list of lines without newlines
|
|
lines = original_text.replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
|
if ends_with_newline and (len(lines) == 0 or lines[-1] != ""):
|
|
# split drops the trailing empty element when text ends with newline; add it back as an empty logical line
|
|
lines.append("")
|
|
|
|
# Validate and apply edits in reverse order to avoid index shifts
|
|
sorted_edits = sorted(edits, key=lambda e: (e.get('start_line', 1), e.get('end_line', 0)), reverse=True)
|
|
|
|
for idx, edit in enumerate(sorted_edits):
|
|
start_line = edit.get('start_line')
|
|
end_line = edit.get('end_line')
|
|
replacement = edit.get('replacement', "")
|
|
|
|
if start_line is None or end_line is None:
|
|
raise ValueError(f"Edit #{idx+1} missing start_line or end_line")
|
|
if not isinstance(start_line, int) or not isinstance(end_line, int):
|
|
raise ValueError(f"Edit #{idx+1} start_line and end_line must be integers")
|
|
if start_line < 1:
|
|
raise ValueError(f"Edit #{idx+1} start_line must be >= 1")
|
|
|
|
# Normalize replacement to list of lines (without trailing newline)
|
|
rep_lines = replacement.replace("\r\n", "\n").replace("\r", "\n").split("\n")
|
|
|
|
if end_line >= start_line:
|
|
# Replace lines in the inclusive range [start_line..end_line]
|
|
if end_line > len(lines):
|
|
raise ValueError(f"Edit #{idx+1} end_line {end_line} exceeds file length {len(lines)}")
|
|
# Python slice end is exclusive; convert to 0-based indices
|
|
s = start_line - 1
|
|
e = end_line
|
|
lines[s:e] = rep_lines
|
|
else:
|
|
# Insertion before start_line (no deletion)
|
|
if start_line > len(lines) + 1:
|
|
raise ValueError(f"Edit #{idx+1} start_line {start_line} is beyond end of file {len(lines)}")
|
|
insert_at = start_line - 1
|
|
lines[insert_at:insert_at] = rep_lines
|
|
|
|
# Reconstruct the text using the original newline style
|
|
text = newline.join(lines)
|
|
if not ends_with_newline and text.endswith(newline):
|
|
# Original did not end with newline; remove any trailing newline we may have introduced
|
|
text = text[:-len(newline)]
|
|
elif ends_with_newline and not text.endswith(newline):
|
|
# Original ended with newline; ensure we keep it
|
|
text += newline
|
|
return text
|
|
|
|
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 |