Creates an agent bot that reviews code to repositories that is has access to if it gets added as code reviewer Co-authored-by: Copilot <copilot@github.com>
94 lines
4.0 KiB
Python
94 lines
4.0 KiB
Python
import requests
|
|
from typing import Iterator, List
|
|
|
|
|
|
class GiteaClient:
|
|
def __init__(self, api_url: str, token: str):
|
|
self.api_url = api_url.rstrip("/")
|
|
self.token = token
|
|
|
|
def _headers(self):
|
|
return {"Authorization": f"token {self.token}", "Content-Type": "application/json"}
|
|
|
|
def available_repositories(self) -> Iterator[tuple[str, str]]:
|
|
"""List all repository URLs available to the token."""
|
|
url = f"{self.api_url}/user/repos"
|
|
r = requests.get(url, headers=self._headers(), timeout=30)
|
|
r.raise_for_status()
|
|
|
|
for repo in r.json():
|
|
owner = repo.get("owner", {}).get("login")
|
|
name = repo.get("name")
|
|
if owner and name: # Skip repos with missing owner or name
|
|
yield owner, name
|
|
else:
|
|
print(f"Warning: Skipping repo with missing owner or name: {repo}")
|
|
|
|
def list_pull_request_files(self, owner: str, repo: str, pr_number: int) -> List[dict]:
|
|
"""Try to list changed files for a pull request. If the endpoint differs, adjust."""
|
|
# Many Gitea instances expose PR files at /repos/{owner}/{repo}/pulls/{index}/files
|
|
url = f"{self.api_url}/repos/{owner}/{repo}/pulls/{pr_number}/files"
|
|
r = requests.get(url, headers=self._headers(), timeout=30)
|
|
if r.status_code == 200:
|
|
return r.json()
|
|
# Fallback: try issues comments or single PR object
|
|
r.raise_for_status()
|
|
|
|
def get_pull_request_diff(self, owner: str, repo: str, pr_number: int) -> str:
|
|
"""Fetch unified diff text for a pull request."""
|
|
url = f"{self.api_url}/repos/{owner}/{repo}/pulls/{pr_number}.diff"
|
|
headers = {"Authorization": f"token {self.token}"}
|
|
r = requests.get(url, headers=headers, timeout=30)
|
|
r.raise_for_status()
|
|
return r.text
|
|
|
|
def create_issue_comment(self, owner: str, repo: str, issue_index: int, body: str) -> dict:
|
|
url = f"{self.api_url}/repos/{owner}/{repo}/issues/{issue_index}/comments"
|
|
r = requests.post(url, headers=self._headers(), json={"body": body}, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def list_open_pull_requests(self, owner: str, repo: str) -> List[dict]:
|
|
"""List open pull requests for a repository."""
|
|
url = f"{self.api_url}/repos/{owner}/{repo}/pulls?state=open"
|
|
r = requests.get(url, headers=self._headers(), timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def list_repos_for_owner(self, owner: str) -> List[dict]:
|
|
"""Try to list repos for an owner (org or user). Returns list of repo dicts."""
|
|
# Try orgs endpoint first
|
|
url_org = f"{self.api_url}/orgs/{owner}/repos"
|
|
r = requests.get(url_org, headers=self._headers(), timeout=30)
|
|
if r.status_code == 200:
|
|
return r.json()
|
|
# Fallback to users endpoint
|
|
url_user = f"{self.api_url}/users/{owner}/repos"
|
|
r = requests.get(url_user, headers=self._headers(), timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
def create_pull_request_review(self, owner: str, repo: str, pr_number: int, body: str, comments: List[dict] = None) -> dict:
|
|
"""Create a PR review with optional line-specific comments.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
pr_number: PR number/index
|
|
body: General review comment
|
|
comments: List of line comments. Each comment dict should have:
|
|
- path: file path
|
|
- new_position: line number in new version
|
|
- body: comment text
|
|
"""
|
|
url = f"{self.api_url}/repos/{owner}/{repo}/pulls/{pr_number}/reviews"
|
|
payload = {
|
|
"body": body,
|
|
"event": "COMMENT"
|
|
}
|
|
if comments:
|
|
payload["comments"] = comments
|
|
r = requests.post(url, headers=self._headers(), json=payload, timeout=30)
|
|
r.raise_for_status()
|
|
return r.json()
|