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