aider-gitea/aider_gitea/__main__.py

#!/usr/bin/env python3
"""
This script downloads issues from a given Gitea repository and produces a pull request for each issue.
It assumes that the default branch (default "main") exists and that you have a valid API token if authentication is required.
"""

import logging
from pathlib import Path
import argparse
import requests
import sys
import dataclasses
import tempfile
import subprocess
import os

import secret_loader
import re

def generate_branch_name(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)
    return "issue-" + "-".join(sanitized.lower().split())

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:

    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"
        if 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}"
        response = self.session.get(url)
        response.raise_for_status()
        data = response.json()
        return data['commit']['sha']

    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}
        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, repo):
        """Download issues from the specified repository."""
        url = f"{self.gitea_url}/repos/{owner}/{repo}/issues"
        response = self.session.get(url)
        response.raise_for_status()
        return response.json()

    def create_pull_request(self, owner, repo, title, head, base, body):
        """Create a pull request for the given branch."""
        url = f"{self.gitea_url}/repos/{owner}/{repo}/pulls"
        json_data = {"title": title, "head": head, "base": base, "body": body}
        response = self.session.post(url, json=json_data)
        response.raise_for_status()
        return response.json()

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)")
    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_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(["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", "push", "origin", branch_name], tmpdirname)

def main():
    logging.basicConfig(level='INFO')
    args = parse_args()

    client = GiteaClient(args.gitea_url, GITEA_TOKEN )

    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}")
        branch_name = generate_branch_name(title)
        try:
            with tempfile.TemporaryDirectory() as tmpdirname:
                process_issue(args, Path(tmpdirname), branch_name, title, issue_description, issue_number)
            logger.info(f"Created branch {branch_name} for issue {issue_number}.")
        except Exception:
            logger.exception('Error processing issue')
            sys.exit(1)

        body = f"Automatically generated pull request for issue: {issue.get('html_url', 'unknown')}"
        try:
            pr = client.create_pull_request(args.owner, args.repo, f"[Issue {issue_number}] {title}", branch_name, args.base_branch, body)
            logger.info(f"Created pull request: {pr.get('html_url', 'unknown')} for issue {issue_number}.")
        except Exception:
            logger.exception('"Error creating pull request for branch')
            sys.exit(1)

if __name__ == "__main__":
    main()