Wow, it actually works
This commit is contained in:
parent
dbc86c85d4
commit
59166b6d8c
|
@ -5,6 +5,7 @@ It assumes that the default branch (default "main") exists and that you have a v
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
from pathlib import Path
|
||||||
import argparse
|
import argparse
|
||||||
import requests
|
import requests
|
||||||
import sys
|
import sys
|
||||||
|
@ -13,13 +14,41 @@ import tempfile
|
||||||
import subprocess
|
import subprocess
|
||||||
import os
|
import os
|
||||||
|
|
||||||
|
import secret_loader
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
AIDER_TEST="pytest test"
|
||||||
|
AIDER_LINT="ruff format; ruff check --fix --ignore RUF022 --ignore PGH004; ruff format; ruff check --ignore RUF022 --ignore PGH004;"
|
||||||
|
|
||||||
|
MODEL = 'o3-mini'
|
||||||
|
|
||||||
|
SECRETS = secret_loader.SecretLoader()
|
||||||
|
LLM_API_KEY = SECRETS.load_or_fail('LLM_API_KEY')
|
||||||
|
GITEA_TOKEN = SECRETS.load_or_fail('GITEA_TOKEN')
|
||||||
|
|
||||||
|
|
||||||
|
def create_aider_command(issue: str) -> list[str]:
|
||||||
|
return [
|
||||||
|
'aider',
|
||||||
|
'--model', MODEL,
|
||||||
|
'--chat-language', 'english',
|
||||||
|
'--test-cmd', AIDER_TEST,
|
||||||
|
'--lint-cmd', AIDER_LINT,
|
||||||
|
'--auto-test',
|
||||||
|
'--no-auto-lint',
|
||||||
|
'--api-key', LLM_API_KEY,
|
||||||
|
'--read', 'CONVENTIONS.md',
|
||||||
|
'--message', issue,
|
||||||
|
'--yes-always',
|
||||||
|
'--architect',
|
||||||
|
]
|
||||||
|
|
||||||
class GiteaClient:
|
class GiteaClient:
|
||||||
|
|
||||||
def __init__(self, gitea_url: str, token: str) -> None:
|
def __init__(self, gitea_url: str, token: str) -> None:
|
||||||
self.gitea_url = gitea_url
|
assert not gitea_url.endswith('/api/v1')
|
||||||
|
self.gitea_url = gitea_url + '/api/v1'
|
||||||
self.session = requests.Session()
|
self.session = requests.Session()
|
||||||
self.session.headers["Content-Type"] = "application/json"
|
self.session.headers["Content-Type"] = "application/json"
|
||||||
if token:
|
if token:
|
||||||
|
@ -31,7 +60,6 @@ class GiteaClient:
|
||||||
response = self.session.get(url)
|
response = self.session.get(url)
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
data = response.json()
|
data = response.json()
|
||||||
print(data)
|
|
||||||
return data['commit']['sha']
|
return data['commit']['sha']
|
||||||
|
|
||||||
def create_branch(self, owner, repo, new_branch, sha):
|
def create_branch(self, owner, repo, new_branch, sha):
|
||||||
|
@ -40,7 +68,7 @@ class GiteaClient:
|
||||||
json_data = {"ref": f"refs/heads/{new_branch}", "sha": sha}
|
json_data = {"ref": f"refs/heads/{new_branch}", "sha": sha}
|
||||||
response = self.session.post(url, json=json_data)
|
response = self.session.post(url, json=json_data)
|
||||||
if response.status_code == 422:
|
if response.status_code == 422:
|
||||||
print(f"Branch {new_branch} already exists.")
|
logger.warning(f"Branch {new_branch} already exists.")
|
||||||
return False
|
return False
|
||||||
response.raise_for_status()
|
response.raise_for_status()
|
||||||
return True
|
return True
|
||||||
|
@ -66,56 +94,59 @@ def parse_args():
|
||||||
parser.add_argument("--owner", required=True, help="Owner of the repository")
|
parser.add_argument("--owner", required=True, help="Owner of the repository")
|
||||||
parser.add_argument("--repo", required=True, help="Repository name")
|
parser.add_argument("--repo", required=True, help="Repository name")
|
||||||
parser.add_argument("--base-branch", default="main", help="Base branch to use for new branches (default: main)")
|
parser.add_argument("--base-branch", default="main", help="Base branch to use for new branches (default: main)")
|
||||||
parser.add_argument("--token", default="", help="Authentication token if required")
|
|
||||||
return parser.parse_args()
|
return parser.parse_args()
|
||||||
|
|
||||||
|
def run_cmd(cmd: list[str], cwd:Path|None=None) -> None:
|
||||||
|
print(cmd)
|
||||||
|
subprocess.run(cmd, check=True, cwd=cwd)
|
||||||
|
|
||||||
|
|
||||||
|
def process_issue(args, tmpdirname: Path, branch_name: str, issue_description: str, issue_number: str):
|
||||||
|
repo_url = f"{args.gitea_url}:{args.owner}/{args.repo}.git".replace('https://', 'git@')
|
||||||
|
run_cmd(["git", "clone", repo_url, tmpdirname])
|
||||||
|
run_cmd(["git", "checkout", args.base_branch], tmpdirname)
|
||||||
|
run_cmd(["git", "checkout", "-b", branch_name], tmpdirname)
|
||||||
|
run_cmd(create_aider_command(issue_description), tmpdirname)
|
||||||
|
run_cmd(["git", "add", "."], tmpdirname)
|
||||||
|
run_cmd(["git", "commit", "-m", f"Apply aider for issue {issue_number}"], tmpdirname)
|
||||||
|
run_cmd(["git", "push", "origin", branch_name], tmpdirname)
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
logging.basicConfig()
|
logging.basicConfig(level='INFO')
|
||||||
args = parse_args()
|
args = parse_args()
|
||||||
|
|
||||||
client = GiteaClient(args.gitea_url, args.token)
|
client = GiteaClient(args.gitea_url, GITEA_TOKEN )
|
||||||
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
issues = client.get_issues(args.owner, args.repo)
|
issues = client.get_issues(args.owner, args.repo)
|
||||||
except Exception as e:
|
except Exception:
|
||||||
logger.exception('Failed to retrieve issues')
|
logger.exception('Failed to retrieve issues')
|
||||||
sys.exit(1)
|
sys.exit(1)
|
||||||
|
|
||||||
if not issues:
|
if not issues:
|
||||||
print("No issues found.")
|
logger.info("No issues found.")
|
||||||
return
|
return
|
||||||
|
|
||||||
try:
|
|
||||||
base_sha = client.get_default_branch_sha(args.owner, args.repo, args.base_branch)
|
|
||||||
except Exception as e:
|
|
||||||
logger.exception('Failed to retrieve base branch SHA')
|
|
||||||
sys.exit(1)
|
|
||||||
|
|
||||||
for issue in issues:
|
for issue in issues:
|
||||||
issue_number = issue.get("number")
|
issue_number = issue.get("number")
|
||||||
|
issue_description = issue.get("body", "")
|
||||||
title = issue.get("title", f"Issue {issue_number}")
|
title = issue.get("title", f"Issue {issue_number}")
|
||||||
branch_name = f"issue-{issue_number}"
|
branch_name = f"issue-{issue_number}"
|
||||||
try:
|
try:
|
||||||
with tempfile.TemporaryDirectory() as tmpdirname:
|
with tempfile.TemporaryDirectory() as tmpdirname:
|
||||||
repo_url = f"{args.gitea_url}/repos/{args.owner}/{args.repo}.git"
|
process_issue(args, Path(tmpdirname), branch_name, issue_description, issue_number)
|
||||||
subprocess.run(["git", "clone", repo_url, tmpdirname], check=True)
|
logger.info(f"Created branch {branch_name} for issue {issue_number}.")
|
||||||
subprocess.run(["git", "-C", tmpdirname, "checkout", args.base_branch], check=True)
|
except Exception:
|
||||||
subprocess.run(["git", "-C", tmpdirname, "checkout", "-b", branch_name], check=True)
|
logger.exception('Error processing issue')
|
||||||
message = issue.get("body", "")
|
sys.exit(1)
|
||||||
subprocess.run(["aider", message], check=True)
|
|
||||||
subprocess.run(["git", "-C", tmpdirname, "add", "."], check=True)
|
|
||||||
subprocess.run(["git", "-C", tmpdirname, "commit", "-m", f"Apply aider for issue {issue_number}"], check=True)
|
|
||||||
subprocess.run(["git", "-C", tmpdirname, "push", "origin", branch_name], check=True)
|
|
||||||
print(f"Created branch {branch_name} for issue {issue_number}.")
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Error processing issue {issue_number}: {e}")
|
|
||||||
body = f"Automatically generated pull request for issue: {issue.get('html_url', 'unknown')}"
|
body = f"Automatically generated pull request for issue: {issue.get('html_url', 'unknown')}"
|
||||||
try:
|
try:
|
||||||
pr = client.create_pull_request(args.owner, args.repo, f"[Issue {issue_number}] {title}", branch_name, args.base_branch, body)
|
pr = client.create_pull_request(args.owner, args.repo, f"[Issue {issue_number}] {title}", branch_name, args.base_branch, body)
|
||||||
print(f"Created pull request: {pr.get('html_url', 'unknown')} for issue {issue_number}.")
|
logger.info(f"Created pull request: {pr.get('html_url', 'unknown')} for issue {issue_number}.")
|
||||||
except Exception as e:
|
except Exception:
|
||||||
print(f"Error creating pull request for branch {branch_name}: {e}")
|
logger.exception('"Error creating pull request for branch')
|
||||||
|
sys.exit(1)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
main()
|
main()
|
||||||
|
|
Loading…
Reference in New Issue
Block a user