aider-gitea/aider_gitea/__init__.py

"""Aider Gitea.

A code automation tool that integrates Gitea with Aider to automatically solve issues.

This program monitors your [Gitea](https://about.gitea.com/) repository for issues with the 'aider' label.
When such an issue is found, it:

1. Creates a new branch.
2. Invokes [Aider](https://aider.chat/) to solve the issue using a Large-Language Model.
3. Runs tests and code quality checks.
4. Creates a pull request with the solution.

## Usage

An application token must be supplied for the `gitea_token` secret. This must
have the following permissions:

- `read:issue`: To be able to read issues on the specified repository.
- `write:repository`: To be able to create pull requests.
- `read:user`: Needed to iterate all user's repositories.

### Command Line

```bash
# Run with default settings
python -m aider_gitea

# Specify custom repository and owner
python -m aider_gitea --owner myorg --repo myproject

# Use a custom Gitea URL
python -m aider_gitea --gitea-url https://gitea.example.com

# Specify a different base branch
python -m aider_gitea --base-branch develop
```

### Python API

```python
from aider_gitea import solve_issue_in_repository
from pathlib import Path

# Solve an issue programmatically
args = argparse.Namespace(
    gitea_url="https://gitea.example.com",
    owner="myorg",
    repo="myproject",
    base_branch="main"
)

solve_issue_in_repository(
    args,
    Path("/path/to/repo"),
    "issue-123-fix-bug",
    "Fix critical bug",
    "The application crashes when processing large files",
    "123"
)
```

### Environment Configuration

The tool uses environment variables for sensitive information:
- `GITEA_TOKEN`: Your Gitea API token
- `LLM_API_KEY`: API key for the language model used by Aider
```
"""

import logging
import re
import subprocess
import sys
import tempfile
from pathlib import Path

from . import secrets
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.

    Args:
        issue_number: The issue number to include in the branch name.
        issue_title: The issue title to sanitize and include in the branch name.

    Returns:
        A sanitized branch name combining the issue number and title.
    """
    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.
"""

# MODEL = 'ollama/gemma3:27b'
# MODEL = 'ollama_chat/gemma3:27b'


def create_aider_command(issue: str) -> list[str]:
    return [
        'aider',
        '--chat-language',
        'english',
        '--cache-prompts',
        '--no-stream',
        '--no-analytics',
        #'--model',
        # MODEL,
        '--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,
    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('No commits made on branch %s, skipping push', branch_name)
        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.

    Args:
        cwd: The current working directory (repository path).
        base_branch: The name of the base branch to compare against.
        current_branch: The name of the current branch to check for commits.

    Returns:
        True if there are commits on the current branch not in the base branch, False otherwise.
    """
    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:
    """Run a shell command and return its success status.

    Args:
        cmd: The command to run as a list of strings.
        cwd: The directory to run the command in.
        check: Whether to raise an exception if the command fails.

    Returns:
        True if the command succeeded, False otherwise.
    """
    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, check=False)

    # Push changes
    return push_changes(
        tmpdirname,
        branch_name,
        issue_number,
        issue_title,
        args.base_branch,
    )


def handle_issues(args, client, seen_issues_db):
    """Process all open issues with the 'aider' label.

    Args:
        args: Command line arguments.
        client: The Gitea client instance.
        seen_issues_db: Database of previously processed issues.
    """
    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 for %s', args.repo)
        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)