From 5fdf9cf002994037eacd62dc2d0afd0220c3dfa7 Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Sun, 13 Apr 2025 18:31:33 +0200 Subject: [PATCH] Split into multiple files --- aider_gitea/__init__.py | 220 +++++++++++++++++++++++++ aider_gitea/__main__.py | 310 +----------------------------------- aider_gitea/gitea_client.py | 119 ++++++++++++++ 3 files changed, 341 insertions(+), 308 deletions(-) create mode 100644 aider_gitea/gitea_client.py diff --git a/aider_gitea/__init__.py b/aider_gitea/__init__.py index 13339eb..79bc96b 100644 --- a/aider_gitea/__init__.py +++ b/aider_gitea/__init__.py @@ -6,4 +6,224 @@ Use [Aider](https://aider.chat/) by creating issues. The program will then automatically invoke Aider and create a pull request for the issue. """ +import argparse +import logging +import re +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +import requests + +from . import secrets +from .seen_issues_db import SeenIssuesDB + from ._version import __version__ # noqa: F401 + +logger = logging.getLogger(__name__) + + + +def generate_branch_name(issue_number: str, issue_title: str) -> str: + """ + Create a branch name by sanitizing the issue title. + Non-alphanumeric characters (except spaces) are removed, + the text is lowercased, and spaces are replaced with dashes. + """ + sanitized = re.sub(r'[^0-9a-zA-Z ]+', '', issue_title) + parts = ['issue', str(issue_number), *sanitized.lower().split()] + return '-'.join(parts) + + +def bash_cmd(*commands: str) -> str: + commands = ('set -e', *commands) + return 'bash -c "' + ';'.join(commands) + '"' + + +AIDER_TEST = bash_cmd( + 'virtualenv venv', + 'source venv/bin/activate', + 'pip install -e .', + 'pytest test', +) + +RUFF_FORMAT_AND_AUTO_FIX = bash_cmd( + 'ruff format', + 'ruff check --fix --ignore RUF022 --ignore PGH004', + 'ruff format', + 'ruff check --fix --ignore RUF022 --ignore PGH004', +) + +AIDER_LINT = bash_cmd( + RUFF_FORMAT_AND_AUTO_FIX, + 'ruff format', + 'ruff check --ignore RUF022 --ignore PGH004', +) + + +LLM_MESSAGE_FORMAT = """ +{issue} + +# Solution Details + +For code tasks: + +1. Create a plan for how to solve the issue. +2. Write unit tests that proves that your solution works. +3. Then, solve the issue by writing the required code. +""" + + +def create_aider_command(issue: str) -> list[str]: + return [ + 'aider', + '--chat-language', + 'english', + '--test-cmd', + AIDER_TEST, + '--lint-cmd', + AIDER_LINT, + '--auto-test', + '--no-auto-lint', + '--api-key', + secrets.llm_api_key(), + '--read', + 'CONVENTIONS.md', + '--message', + LLM_MESSAGE_FORMAT.format(issue=issue), + '--yes-always', + '--architect', + ] + + +def push_changes( + cwd: Path, + branch_name: str, + issue_number: str, + issue_title: str, + issue_description: str, + base_branch: str, +) -> bool: + # Check if there are any commits on the branch before pushing + if not has_commits_on_branch(cwd, base_branch, branch_name): + logger.info(f'No commits made on branch {branch_name}, skipping push') + return False + cmd = [ + 'git', + 'push', + 'origin', + f'HEAD:refs/for/{base_branch}', + '-o', + f'topic={branch_name}', + '-o', + f'title={issue_title}', + '-o', + f'description=This pull request resolves #{issue_number}', + ] + run_cmd(cmd, cwd) + return True + + +def has_commits_on_branch(cwd: Path, base_branch: str, current_branch: str) -> bool: + """Check if there are any commits on the current branch that aren't in the base branch.""" + try: + result = subprocess.run( + ['git', 'log', f'{base_branch}..{current_branch}', '--oneline'], + check=True, + cwd=cwd, + capture_output=True, + text=True, + ) + return bool(result.stdout.strip()) + except subprocess.CalledProcessError: + logger.exception(f'Failed to check commits on branch {current_branch}') + return False + + +def run_cmd(cmd: list[str], cwd: Path | None = None, check=True) -> bool: + """Returns true if the command succeeded.""" + result = subprocess.run(cmd, check=check, cwd=cwd) + return result.returncode == 0 + + +def solve_issue_in_repository( + args, + tmpdirname: Path, + branch_name: str, + issue_title: str, + issue_description: str, + issue_number: str, +) -> bool: + repo_url = f'{args.gitea_url}:{args.owner}/{args.repo}.git'.replace( + 'https://', + 'git@', + ) + + # Setup repository + run_cmd(['git', 'clone', repo_url, tmpdirname]) + run_cmd(['bash', '-c', AIDER_TEST], tmpdirname) + run_cmd(['git', 'checkout', args.base_branch], tmpdirname) + run_cmd(['git', 'checkout', '-b', branch_name], tmpdirname) + + # Run aider + succeeded = run_cmd( + create_aider_command(f'# {issue_title}\n{issue_description}'), + tmpdirname, + check=False, + ) + if not succeeded: + logger.error('Aider invocation failed for issue #%s', issue_number) + return False + + # Auto-fix standard code quality stuff + run_cmd(['bash', '-c', RUFF_FORMAT_AND_AUTO_FIX], tmpdirname, check=False) + run_cmd(['git', 'add', '.'], tmpdirname) + run_cmd(['git', 'commit', '-m', 'Ruff'], tmpdirname) + + # Push changes + return push_changes( + tmpdirname, + branch_name, + issue_number, + issue_title, + issue_description, + args.base_branch, + ) + + +def handle_issues(args, client, seen_issues_db): + """Process all open issues with the 'aider' label.""" + try: + issues = client.get_issues(args.owner, args.repo) + except Exception: + logger.exception('Failed to retrieve issues') + sys.exit(1) + + if not issues: + logger.info('No issues found.') + return + + for issue in issues: + issue_number = issue.get('number') + issue_description = issue.get('body', '') + title = issue.get('title', f'Issue {issue_number}') + issue_text = f'{title}\n{issue_description}' + if seen_issues_db.has_seen(issue_text): + logger.info(f'Skipping already processed issue #{issue_number}: {title}') + continue + + branch_name = generate_branch_name(issue_number, title) + with tempfile.TemporaryDirectory() as tmpdirname: + solved = solve_issue_in_repository( + args, + Path(tmpdirname), + branch_name, + title, + issue_description, + issue_number, + ) + + if solved: + seen_issues_db.mark_as_seen(issue_text) diff --git a/aider_gitea/__main__.py b/aider_gitea/__main__.py index 176f2a1..fcdc912 100644 --- a/aider_gitea/__main__.py +++ b/aider_gitea/__main__.py @@ -15,188 +15,13 @@ from pathlib import Path import requests -from . import secrets +from . import secrets, handle_issues +from .gitea_client import GiteaClient from .seen_issues_db import SeenIssuesDB - -def generate_branch_name(issue_number: str, issue_title: str) -> str: - """ - Create a branch name by sanitizing the issue title. - Non-alphanumeric characters (except spaces) are removed, - the text is lowercased, and spaces are replaced with dashes. - """ - sanitized = re.sub(r'[^0-9a-zA-Z ]+', '', issue_title) - parts = ['issue', str(issue_number), *sanitized.lower().split()] - return '-'.join(parts) - - logger = logging.getLogger(__name__) -def bash_cmd(*commands: str) -> str: - commands = ('set -e', *commands) - return 'bash -c "' + ';'.join(commands) + '"' - - -AIDER_TEST = bash_cmd( - 'virtualenv venv', - 'source venv/bin/activate', - 'pip install -e .', - 'pytest test', -) - -RUFF_FORMAT_AND_AUTO_FIX = bash_cmd( - 'ruff format', - 'ruff check --fix --ignore RUF022 --ignore PGH004', - 'ruff format', - 'ruff check --fix --ignore RUF022 --ignore PGH004', -) - -AIDER_LINT = bash_cmd( - RUFF_FORMAT_AND_AUTO_FIX, - 'ruff format', - 'ruff check --ignore RUF022 --ignore PGH004', -) - - -LLM_MESSAGE_FORMAT = """ -{issue} - -# Solution Details - -For code tasks: - -1. Create a plan for how to solve the issue. -2. Write unit tests that proves that your solution works. -3. Then, solve the issue by writing the required code. -""" - - -def create_aider_command(issue: str) -> list[str]: - return [ - 'aider', - '--chat-language', - 'english', - '--test-cmd', - AIDER_TEST, - '--lint-cmd', - AIDER_LINT, - '--auto-test', - '--no-auto-lint', - '--api-key', - secrets.llm_api_key(), - '--read', - 'CONVENTIONS.md', - '--message', - LLM_MESSAGE_FORMAT.format(issue=issue), - '--yes-always', - '--architect', - ] - - -class GiteaClient: - """ - Client for interacting with the Gitea API. - - This class provides methods to interact with a Gitea instance's API, - including retrieving repository information, creating branches, and fetching issues. - - Attributes: - gitea_url (str): The base URL for the Gitea API endpoints. - session (requests.Session): HTTP session for making API requests. - """ - - def __init__(self, gitea_url: str, token: str) -> None: - """ - Initialize a new Gitea API client. - - Args: - gitea_url (str): Base URL for the Gitea instance (without '/api/v1'). - token (str): Authentication token for the Gitea API. If empty, requests will be unauthenticated. - - Raises: - AssertionError: If gitea_url ends with '/api/v1'. - """ - assert not gitea_url.endswith('/api/v1') - self.gitea_url = gitea_url + '/api/v1' - self.session = requests.Session() - self.session.headers['Content-Type'] = 'application/json' - if token: - self.session.headers['Authorization'] = f'token {token}' - - def get_default_branch_sha(self, owner: str, repo: str, branch: str) -> str: - """ - Retrieve the commit SHA of the specified branch. - - Args: - owner (str): Owner of the repository. - repo (str): Name of the repository. - branch (str): Name of the branch. - - Returns: - str: The commit SHA of the specified branch. - - Raises: - requests.HTTPError: If the API request fails. - """ - url = f'{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}' - response = self.session.get(url) - response.raise_for_status() - data = response.json() - return data['commit']['sha'] - - def create_branch(self, owner: str, repo: str, new_branch: str, sha: str) -> bool: - """ - Create a new branch from the provided SHA. - - Args: - owner (str): Owner of the repository. - repo (str): Name of the repository. - new_branch (str): Name of the new branch to create. - sha (str): Commit SHA to use as the starting point for the new branch. - - Returns: - bool: True if the branch was created successfully, False if the branch already exists. - - Raises: - requests.HTTPError: If the API request fails for reasons other than branch already existing. - """ - url = f'{self.gitea_url}/repos/{owner}/{repo}/git/refs' - json_data = {'ref': f'refs/heads/{new_branch}', 'sha': sha} - response = self.session.post(url, json=json_data) - if response.status_code == 422: - logger.warning(f'Branch {new_branch} already exists.') - return False - response.raise_for_status() - return True - - def get_issues(self, owner: str, repo: str) -> list: - """ - Download issues from the specified repository and filter those with the 'aider' label. - - Args: - owner (str): Owner of the repository. - repo (str): Name of the repository. - - Returns: - list: A list of issue dictionaries, filtered to only include issues with the 'aider' label. - - Raises: - requests.HTTPError: If the API request fails. - """ - url = f'{self.gitea_url}/repos/{owner}/{repo}/issues' - response = self.session.get(url) - response.raise_for_status() - issues = response.json() - # Filter to only include issues marked with the "aider" label. - issues = [ - issue - for issue in issues - if any(label.get('name') == 'aider' for label in issue.get('labels', [])) - ] - return issues - - def parse_args(): parser = argparse.ArgumentParser( description='Download issues and create pull requests for a Gitea repository.', @@ -227,137 +52,6 @@ def parse_args(): return parser.parse_args() -def push_changes( - cwd: Path, - branch_name: str, - issue_number: str, - issue_title: str, - issue_description: str, - base_branch: str, -) -> bool: - # Check if there are any commits on the branch before pushing - if not has_commits_on_branch(cwd, base_branch, branch_name): - logger.info(f'No commits made on branch {branch_name}, skipping push') - return False - cmd = [ - 'git', - 'push', - 'origin', - f'HEAD:refs/for/{base_branch}', - '-o', - f'topic={branch_name}', - '-o', - f'title={issue_title}', - '-o', - f'description=This pull request resolves #{issue_number}', - ] - run_cmd(cmd, cwd) - return True - - -def has_commits_on_branch(cwd: Path, base_branch: str, current_branch: str) -> bool: - """Check if there are any commits on the current branch that aren't in the base branch.""" - try: - result = subprocess.run( - ['git', 'log', f'{base_branch}..{current_branch}', '--oneline'], - check=True, - cwd=cwd, - capture_output=True, - text=True, - ) - return bool(result.stdout.strip()) - except subprocess.CalledProcessError: - logger.exception(f'Failed to check commits on branch {current_branch}') - return False - - -def run_cmd(cmd: list[str], cwd: Path | None = None, check=True) -> bool: - """Returns true if the command succeeded.""" - result = subprocess.run(cmd, check=check, cwd=cwd) - return result.returncode == 0 - - -def solve_issue_in_repository( - args, - tmpdirname: Path, - branch_name: str, - issue_title: str, - issue_description: str, - issue_number: str, -) -> bool: - repo_url = f'{args.gitea_url}:{args.owner}/{args.repo}.git'.replace( - 'https://', - 'git@', - ) - - # Setup repository - run_cmd(['git', 'clone', repo_url, tmpdirname]) - run_cmd(['bash', '-c', AIDER_TEST], tmpdirname) - run_cmd(['git', 'checkout', args.base_branch], tmpdirname) - run_cmd(['git', 'checkout', '-b', branch_name], tmpdirname) - - # Run aider - succeeded = run_cmd( - create_aider_command(f'# {issue_title}\n{issue_description}'), - tmpdirname, - check=False, - ) - if not succeeded: - logger.error('Aider invocation failed for issue #%s', issue_number) - return False - - # Auto-fix standard code quality stuff - run_cmd(['bash', '-c', RUFF_FORMAT_AND_AUTO_FIX], tmpdirname, check=False) - run_cmd(['git', 'add', '.'], tmpdirname) - run_cmd(['git', 'commit', '-m', 'Ruff'], tmpdirname) - - # Push changes - return push_changes( - tmpdirname, - branch_name, - issue_number, - issue_title, - issue_description, - args.base_branch, - ) - - -def handle_issues(args, client, seen_issues_db): - """Process all open issues with the 'aider' label.""" - try: - issues = client.get_issues(args.owner, args.repo) - except Exception: - logger.exception('Failed to retrieve issues') - sys.exit(1) - - if not issues: - logger.info('No issues found.') - return - - for issue in issues: - issue_number = issue.get('number') - issue_description = issue.get('body', '') - title = issue.get('title', f'Issue {issue_number}') - issue_text = f'{title}\n{issue_description}' - if seen_issues_db.has_seen(issue_text): - logger.info(f'Skipping already processed issue #{issue_number}: {title}') - continue - - branch_name = generate_branch_name(issue_number, title) - with tempfile.TemporaryDirectory() as tmpdirname: - solved = solve_issue_in_repository( - args, - Path(tmpdirname), - branch_name, - title, - issue_description, - issue_number, - ) - - if solved: - seen_issues_db.mark_as_seen(issue_text) - - def main(): logging.basicConfig(level='INFO') args = parse_args() diff --git a/aider_gitea/gitea_client.py b/aider_gitea/gitea_client.py new file mode 100644 index 0000000..f29d16a --- /dev/null +++ b/aider_gitea/gitea_client.py @@ -0,0 +1,119 @@ + +import argparse +import logging +import re +import subprocess +import sys +import tempfile +import time +from pathlib import Path + +import requests + +logger = logging.getLogger(__name__) + + +class GiteaClient: + """ + Client for interacting with the Gitea API. + + This class provides methods to interact with a Gitea instance's API, + including retrieving repository information, creating branches, and fetching issues. + + Attributes: + gitea_url (str): The base URL for the Gitea API endpoints. + session (requests.Session): HTTP session for making API requests. + """ + + def __init__(self, gitea_url: str, token: str) -> None: + """ + Initialize a new Gitea API client. + + Args: + gitea_url (str): Base URL for the Gitea instance (without '/api/v1'). + token (str): Authentication token for the Gitea API. If empty, requests will be unauthenticated. + + Raises: + AssertionError: If gitea_url ends with '/api/v1'. + """ + assert not gitea_url.endswith('/api/v1') + self.gitea_url = gitea_url + '/api/v1' + self.session = requests.Session() + self.session.headers['Content-Type'] = 'application/json' + if token: + self.session.headers['Authorization'] = f'token {token}' + + def get_default_branch_sha(self, owner: str, repo: str, branch: str) -> str: + """ + Retrieve the commit SHA of the specified branch. + + Args: + owner (str): Owner of the repository. + repo (str): Name of the repository. + branch (str): Name of the branch. + + Returns: + str: The commit SHA of the specified branch. + + Raises: + requests.HTTPError: If the API request fails. + """ + url = f'{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}' + response = self.session.get(url) + response.raise_for_status() + data = response.json() + return data['commit']['sha'] + + def create_branch(self, owner: str, repo: str, new_branch: str, sha: str) -> bool: + """ + Create a new branch from the provided SHA. + + Args: + owner (str): Owner of the repository. + repo (str): Name of the repository. + new_branch (str): Name of the new branch to create. + sha (str): Commit SHA to use as the starting point for the new branch. + + Returns: + bool: True if the branch was created successfully, False if the branch already exists. + + Raises: + requests.HTTPError: If the API request fails for reasons other than branch already existing. + """ + url = f'{self.gitea_url}/repos/{owner}/{repo}/git/refs' + json_data = {'ref': f'refs/heads/{new_branch}', 'sha': sha} + response = self.session.post(url, json=json_data) + if response.status_code == 422: + logger.warning(f'Branch {new_branch} already exists.') + return False + response.raise_for_status() + return True + + def get_issues(self, owner: str, repo: str) -> list: + """ + Download issues from the specified repository and filter those with the 'aider' label. + + Args: + owner (str): Owner of the repository. + repo (str): Name of the repository. + + Returns: + list: A list of issue dictionaries, filtered to only include issues with the 'aider' label. + + Raises: + requests.HTTPError: If the API request fails. + """ + url = f'{self.gitea_url}/repos/{owner}/{repo}/issues' + response = self.session.get(url) + response.raise_for_status() + issues = response.json() + # Filter to only include issues marked with the "aider" label. + issues = [ + issue + for issue in issues + if any(label.get('name') == 'aider' for label in issue.get('labels', [])) + ] + return issues + + +