ai-software-factory/ai_software_factory/agents/gitea.py

"""Gitea API integration for commits and PRs."""

import os
from typing import Optional


class GiteaAPI:
    """Gitea API client for repository operations."""
    
    def __init__(self, token: str, base_url: str, owner: str | None = None, repo: str | None = None):
        self.token = token
        self.base_url = base_url.rstrip("/")
        self.owner = owner
        self.repo = repo
        self.headers = {
            "Authorization": f"token {token}",
            "Content-Type": "application/json"
        }
    
    def get_config(self) -> dict:
        """Load configuration from environment."""
        base_url = os.getenv("GITEA_URL", "https://gitea.local")
        token = os.getenv("GITEA_TOKEN", "")
        owner = os.getenv("GITEA_OWNER", "ai-test")
        repo = os.getenv("GITEA_REPO", "")
        
        # Allow empty repo for any repo mode (org/repo pattern)
        if not repo:
            repo = "any-repo"  # Use this as a placeholder for org/repo operations
        
        # Check for repo suffix pattern (e.g., repo-* for multiple repos)
        repo_suffix = os.getenv("GITEA_REPO_SUFFIX", "")
        
        return {
            "base_url": base_url.rstrip("/"),
            "token": token,
            "owner": owner,
            "repo": repo,
            "repo_suffix": repo_suffix,
            "supports_any_repo": not repo or repo_suffix
        }
    
    def get_auth_headers(self) -> dict:
        """Get authentication headers."""
        return {
            "Authorization": f"token {self.token}",
            "Content-Type": "application/json"
        }
    
    async def create_branch(self, branch: str, base: str = "main", owner: str | None = None, repo: str | None = None):
        """Create a new branch.
        
        Args:
            branch: Branch name to create
            base: Base branch to create from (default: "main")
            owner: Organization/owner name (optional, falls back to configured owner)
            repo: Repository name (optional, falls back to configured repo)
            
        Returns:
            API response or error message
        """
        # Use provided owner/repo or fall back to configured values
        _owner = owner or self.owner
        _repo = repo or self.repo
        
        url = f"{self.base_url}/repos/{_owner}/{_repo}/branches/{branch}"
        payload = {"base": base}
        
        try:
            import aiohttp
            async with aiohttp.ClientSession() as session:
                async with session.post(url, headers=self.get_auth_headers(), json=payload) as resp:
                    if resp.status == 201:
                        return await resp.json()
                    else:
                        return {"error": await resp.text()}
        except Exception as e:
            return {"error": str(e)}
    
    async def create_pull_request(
        self,
        title: str,
        body: str,
        owner: str,
        repo: str,
        base: str = "main",
        head: str | None = None
    ) -> dict:
        """Create a pull request.
        
        Args:
            title: PR title
            body: PR description
            owner: Organization/owner name
            repo: Repository name
            base: Base branch (default: "main")
            head: Head branch (optional, auto-generated if not provided)
            
        Returns:
            API response or error message
        """
        _owner = owner or self.owner
        _repo = repo or self.repo
        
        url = f"{self.base_url}/repos/{_owner}/{_repo}/pulls"
        
        payload = {
            "title": title,
            "body": body,
            "base": {"branch": base},
            "head": head or f"{_owner}-{_repo}-ai-gen-{hash(title) % 10000}"
        }
        
        try:
            import aiohttp
            async with aiohttp.ClientSession() as session:
                async with session.post(url, headers=self.get_auth_headers(), json=payload) as resp:
                    if resp.status == 201:
                        return await resp.json()
                    else:
                        return {"error": await resp.text()}
        except Exception as e:
            return {"error": str(e)}
    
    async def push_commit(
        self,
        branch: str,
        files: list[dict],
        message: str,
        owner: str | None = None,
        repo: str | None = None
    ) -> dict:
        """
        Push files to a branch.
        
        In production, this would use gitea's API or git push.
        For now, we'll simulate the operation.
        
        Args:
            branch: Branch name
            files: List of files to push
            message: Commit message
            owner: Organization/owner name (optional, falls back to configured owner)
            repo: Repository name (optional, falls back to configured repo)
            
        Returns:
            Status response
        """
        # Use provided owner/repo or fall back to configured values
        _owner = owner or self.owner
        _repo = repo or self.repo
        
        return {
            "status": "simulated",
            "branch": branch,
            "message": message,
            "files": files,
            "owner": _owner,
            "repo": _repo
        }
    
    async def get_repo_info(self, owner: str | None = None, repo: str | None = None) -> dict:
        """Get repository information.
        
        Args:
            owner: Organization/owner name (optional, falls back to configured owner)
            repo: Repository name (optional, falls back to configured repo)
            
        Returns:
            Repository info or error message
        """
        # Use provided owner/repo or fall back to configured values
        _owner = owner or self.owner
        _repo = repo or self.repo
        
        if not _repo:
            return {"error": "Repository name required for org operations"}
        
        url = f"{self.base_url}/repos/{_owner}/{_repo}"
        
        try:
            import aiohttp
            async with aiohttp.ClientSession() as session:
                async with session.get(url, headers=self.get_auth_headers()) as resp:
                    if resp.status == 200:
                        return await resp.json()
                    else:
                        return {"error": await resp.text()}
        except Exception as e:
            return {"error": str(e)}