Jmaa/aider-gitea
1
0
Fork 0
Code Issues 11 Pull Requests 3 Actions Activity

Ruff

Browse Source
This commit is contained in:
Jon Michael Aanes 2025-04-13 18:06:44 +02:00
parent 8c75a10b3a
commit bdee056b67
7 changed files with 174 additions and 95 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

214
aider_gitea/__main__.py
View File

@ -19,33 +19,37 @@ import re
from . import secrets
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)
sanitized = re.sub(r'[^0-9a-zA-Z ]+', '', issue_title)
parts = ['issue', str(issue_number), *sanitized.lower().split()]
return "-".join(parts)
return '-'.join(parts)
logger = logging.getLogger(__name__)
def bash_cmd(*commands: str) -> str:
return "bash -c \""+';'.join(commands)+"\""
AIDER_TEST=bash_cmd(
"virtualenv venv",
"source venv/bin/activate",
"pip install -e .",
"pytest test",
def bash_cmd(*commands: str) -> str:
return 'bash -c "' + ';'.join(commands) + '"'
AIDER_TEST = bash_cmd(
'virtualenv venv',
'source venv/bin/activate',
'pip install -e .',
'pytest test',
)
AIDER_LINT=bash_cmd(
"ruff format",
"ruff check --fix --ignore RUF022 --ignore PGH004",
"ruff format",
"ruff check --ignore RUF022 --ignore PGH004",
AIDER_LINT = bash_cmd(
'ruff format',
'ruff check --fix --ignore RUF022 --ignore PGH004',
'ruff format',
'ruff check --ignore RUF022 --ignore PGH004',
)
@ -61,34 +65,41 @@ For code tasks:
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,
'--chat-language',
'english',
'--test-cmd',
AIDER_TEST,
'--lint-cmd',
AIDER_LINT,
'--auto-test',
'--auto-lint',
'--api-key', secrets.llm_api_key(),
'--read', 'CONVENTIONS.md',
'--message', LLM_MESSAGE_FORMAT.format(issue=issue),
'--api-key',
secrets.llm_api_key(),
'--read',
'CONVENTIONS.md',
'--message',
LLM_MESSAGE_FORMAT.format(issue=issue),
'--yes-always',
'--architect',
]
class GiteaClient:
class GiteaClient:
def __init__(self, gitea_url: str, token: str) -> None:
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"
self.session.headers['Content-Type'] = 'application/json'
if token:
self.session.headers["Authorization"] = f"token {token}"
self.session.headers['Authorization'] = f'token {token}'
def get_default_branch_sha(self, owner, repo, branch):
"""Retrieve the commit SHA of the default branch."""
url = f"{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}"
url = f'{self.gitea_url}/repos/{owner}/{repo}/branches/{branch}'
response = self.session.get(url)
response.raise_for_status()
data = response.json()
@ -96,87 +107,135 @@ class GiteaClient:
def create_branch(self, owner, repo, new_branch, sha):
"""Create a new branch from the provided SHA."""
url = f"{self.gitea_url}/repos/{owner}/{repo}/git/refs"
json_data = {"ref": f"refs/heads/{new_branch}", "sha": sha}
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.")
logger.warning(f'Branch {new_branch} already exists.')
return False
response.raise_for_status()
return True
def get_issues(self, owner, repo):
"""Download issues from the specified repository and filter those with the aider label."""
url = f"{self.gitea_url}/repos/{owner}/{repo}/issues"
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", []))
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.")
parser.add_argument("--gitea-url", required=True, help="Base URL for the Gitea instance, e.g., https://gitfub.space/api/v1")
parser.add_argument("--owner", required=True, help="Owner of the repository")
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("--daemon", action="store_true", help="Run in daemon mode to continuously monitor for new issues")
parser.add_argument("--interval", type=int, default=300, help="Interval in seconds between checks in daemon mode (default: 300)")
parser = argparse.ArgumentParser(
description='Download issues and create pull requests for a Gitea repository.'
)
parser.add_argument(
'--gitea-url',
required=True,
help='Base URL for the Gitea instance, e.g., https://gitfub.space/api/v1',
)
parser.add_argument('--owner', required=True, help='Owner of the repository')
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(
'--daemon',
action='store_true',
help='Run in daemon mode to continuously monitor for new issues',
)
parser.add_argument(
'--interval',
type=int,
default=300,
help='Interval in seconds between checks in daemon mode (default: 300)',
)
return parser.parse_args()
def push_changes(cwd: Path, branch_name: str, issue_number: str, issue_title: str, issue_description: str, base_branch: str) -> None:
def push_changes(
cwd: Path,
branch_name: str,
issue_number: str,
issue_title: str,
issue_description: str,
base_branch: str,
) -> None:
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}",
'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)
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"],
['git', 'log', f'{base_branch}..{current_branch}', '--oneline'],
check=True,
cwd=cwd,
capture_output=True,
text=True
text=True,
)
return bool(result.stdout.strip())
except subprocess.CalledProcessError:
logger.exception(f"Failed to check commits on branch {current_branch}")
logger.exception(f'Failed to check commits on branch {current_branch}')
return False
def run_cmd(cmd: list[str], cwd:Path|None=None) -> None:
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_title: 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])
def process_issue(
args,
tmpdirname: Path,
branch_name: str,
issue_title: 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(['bash', '-c', AIDER_TEST], tmpdirname)
run_cmd(["git", "checkout", args.base_branch], tmpdirname)
run_cmd(["git", "checkout", "-b", branch_name], tmpdirname)
run_cmd(['git', 'checkout', args.base_branch], tmpdirname)
run_cmd(['git', 'checkout', '-b', branch_name], tmpdirname)
run_cmd(create_aider_command(f'# {issue_title}\n{issue_description}'), tmpdirname)
run_cmd(["git", "add", "."], tmpdirname)
run_cmd(['git', 'add', '.'], tmpdirname)
# Check if there are any commits on the branch before pushing
if has_commits_on_branch(tmpdirname, args.base_branch, branch_name):
push_changes(tmpdirname, branch_name, issue_number, issue_title, issue_description, args.base_branch)
push_changes(
tmpdirname,
branch_name,
issue_number,
issue_title,
issue_description,
args.base_branch,
)
else:
logger.info(f"No commits made on branch {branch_name}, skipping push")
logger.info(f'No commits made on branch {branch_name}, skipping push')
def handle_issues(args, client, seen_issues_db):
"""Process all open issues with the 'aider' label."""
@ -187,28 +246,36 @@ def handle_issues(args, client, seen_issues_db):
sys.exit(1)
if not issues:
logger.info("No issues found.")
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}"
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}")
logger.info(f'Skipping already processed issue #{issue_number}: {title}')
continue
branch_name = generate_branch_name(issue_number, title)
try:
with tempfile.TemporaryDirectory() as tmpdirname:
process_issue(args, Path(tmpdirname), branch_name, title, issue_description, issue_number)
process_issue(
args,
Path(tmpdirname),
branch_name,
title,
issue_description,
issue_number,
)
except Exception:
logger.exception('Error processing issue')
sys.exit(1)
seen_issues_db.mark_as_seen(issue_text)
def main():
logging.basicConfig(level='INFO')
args = parse_args()
@ -217,18 +284,21 @@ def main():
client = GiteaClient(args.gitea_url, secrets.gitea_token())
if args.daemon:
logger.info(f"Starting daemon mode, checking for new issues every {args.interval} seconds")
logger.info(
f'Starting daemon mode, checking for new issues every {args.interval} seconds'
)
try:
while True:
logger.info("Checking for new issues...")
logger.info('Checking for new issues...')
handle_issues(args, client, seen_issues_db)
logger.info(f"Sleeping for {args.interval} seconds...")
logger.info(f'Sleeping for {args.interval} seconds...')
time.sleep(args.interval)
except KeyboardInterrupt:
logger.info("Daemon stopped by user")
logger.info('Daemon stopped by user')
else:
# One-off run
handle_issues(args, client, seen_issues_db)
if __name__ == "__main__":
if __name__ == '__main__':
main()

2
aider_gitea/secrets.py
View File

@ -2,8 +2,10 @@ import secret_loader
SECRETS = secret_loader.SecretLoader()
def llm_api_key():
return SECRETS.load_or_fail('LLM_API_KEY')
def gitea_token():
return SECRETS.load_or_fail('GITEA_TOKEN')

14
aider_gitea/seen_issues_db.py
View File

@ -3,6 +3,7 @@ from hashlib import sha256
DEFAULT_DB_PATH = 'output/seen_issues.db'
class SeenIssuesDB:
def __init__(self, db_path=DEFAULT_DB_PATH):
self.conn = sqlite3.connect(db_path)
@ -10,20 +11,25 @@ class SeenIssuesDB:
def _create_table(self):
with self.conn:
self.conn.execute('''
self.conn.execute("""
CREATE TABLE IF NOT EXISTS seen_issues (
issue_hash TEXT PRIMARY KEY
)
''')
""")
def mark_as_seen(self, issue_text: str):
issue_hash = self._compute_hash(issue_text)
with self.conn:
self.conn.execute('INSERT OR IGNORE INTO seen_issues (issue_hash) VALUES (?)', (issue_hash,))
self.conn.execute(
'INSERT OR IGNORE INTO seen_issues (issue_hash) VALUES (?)',
(issue_hash,),
)
def has_seen(self, issue_text: str) -> bool:
issue_hash = self._compute_hash(issue_text)
cursor = self.conn.execute('SELECT 1 FROM seen_issues WHERE issue_hash = ?', (issue_hash,))
cursor = self.conn.execute(
'SELECT 1 FROM seen_issues WHERE issue_hash = ?', (issue_hash,)
)
return cursor.fetchone() is not None
def _compute_hash(self, text: str) -> str:

15
test/test_generate_branch_name.py
View File

@ -1,16 +1,19 @@
import pytest
from aider_gitea.__main__ import generate_branch_name
def test_generate_branch_name_normal():
# Normal case with alphanumeric title.
branch = generate_branch_name("123", "Some Issue Title")
assert branch == "issue-123-some-issue-title"
branch = generate_branch_name('123', 'Some Issue Title')
assert branch == 'issue-123-some-issue-title'
def test_generate_branch_name_special_characters():
branch = generate_branch_name("45", "Issue @ Special!")
assert branch == "issue-45-issue-special"
branch = generate_branch_name('45', 'Issue @ Special!')
assert branch == 'issue-45-issue-special'
def test_generate_branch_name_numeric_title():
# Test where the title starts with numbers.
branch = generate_branch_name("789", "123 Numbers Here")
assert branch == "issue-789-123-numbers-here"
branch = generate_branch_name('789', '123 Numbers Here')
assert branch == 'issue-789-123-numbers-here'

6
test/test_init.py
View File

@ -1,5 +1,3 @@
def test_init():
import aider_gitea #noqa: F401
import aider_gitea.secrets #noqa: F401
import aider_gitea # noqa: F401
import aider_gitea.secrets # noqa: F401

6
test/test_seen_issues_db.py
View File

@ -3,17 +3,17 @@ import unittest
from pathlib import Path
from aider_gitea.seen_issues_db import SeenIssuesDB
class TestSeenIssuesDB(unittest.TestCase):
class TestSeenIssuesDB(unittest.TestCase):
def setUp(self):
self.db = SeenIssuesDB(':memory:')
def test_mark_and_check_seen_issue(self):
issue_text = "Test issue"
issue_text = 'Test issue'
self.assertFalse(self.db.has_seen(issue_text))
self.db.mark_as_seen(issue_text)
self.assertTrue(self.db.has_seen(issue_text))
def test_unseen_issue(self):
issue_text = "Unseen issue"
issue_text = 'Unseen issue'
self.assertFalse(self.db.has_seen(issue_text))