feat: initial release, refs NOISSUE

ai_software_factory/agents/gitea.py

@@ -0,0 +1,115 @@
+"""Gitea API integration for commits and PRs."""
+
+import json
+import os
+from typing import Optional
+
+
+class GiteaAPI:
+    """Gitea API client for repository operations."""
+    
+    def __init__(self, token: str, base_url: str):
+        self.token = token
+        self.base_url = base_url.rstrip("/")
+        self.headers = {
+            "Authorization": f"token {token}",
+            "Content-Type": "application/json"
+        }
+    
+    async def create_branch(self, owner: str, repo: str, branch: str, base: str = "main"):
+        """Create a new branch."""
+        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.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,
+        owner: str,
+        repo: str,
+        title: str,
+        body: str,
+        base: str = "main",
+        head: str = None
+    ) -> dict:
+        """Create a pull request."""
+        url = f"{self.base_url}/repos/{owner}/{repo}/pulls"
+        
+        payload = {
+            "title": title,
+            "body": body,
+            "base": {"branch": base},
+            "head": head or f"ai-gen-{hash(title) % 10000}"
+        }
+        
+        try:
+            import aiohttp
+            async with aiohttp.ClientSession() as session:
+                async with session.post(url, headers=self.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,
+        owner: str,
+        repo: str,
+        branch: str,
+        files: list[dict],
+        message: str
+    ) -> dict:
+        """
+        Push files to a branch.
+        
+        In production, this would use gitea's API or git push.
+        For now, we'll simulate the operation.
+        """
+        # In reality, you'd need to:
+        # 1. Clone repo
+        # 2. Create branch
+        # 3. Add files
+        # 4. Commit
+        # 5. Push
+        
+        return {
+            "status": "simulated",
+            "branch": branch,
+            "message": message,
+            "files": files
+        }
+    
+    async def get_repo_info(self, owner: str, repo: str) -> dict:
+        """Get repository information."""
+        url = f"{self.base_url}/repos/{owner}/{repo}"
+        
+        try:
+            import aiohttp
+            async with aiohttp.ClientSession() as session:
+                async with session.get(url, headers=self.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)}
+    
+    def get_config(self) -> dict:
+        """Load configuration from environment."""
+        return {
+            "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", "ai-test")
+        }