feat: better history data, refs NOISSUE

2026-04-10 23:52:08 +02:00
parent 032139c14f
commit fd812476cc
11 changed files with 2643 additions and 93 deletions
--- a/ai_software_factory/agents/gitea.py
+++ b/ai_software_factory/agents/gitea.py
@@ -1,6 +1,9 @@
 """Gitea API integration for repository and pull request operations."""

 import os
+import urllib.error
+import urllib.request
+import json


 class GiteaAPI:
@@ -41,6 +44,38 @@ class GiteaAPI:
        """Build a Gitea API URL from a relative path."""
        return f"{self.base_url}/api/v1/{path.lstrip('/')}"

+    def build_repo_git_url(self, owner: str | None = None, repo: str | None = None) -> str | None:
+        """Build the clone URL for a repository."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        if not _owner or not _repo:
+            return None
+        return f"{self.base_url}/{_owner}/{_repo}.git"
+
+    def build_commit_url(self, commit_hash: str, owner: str | None = None, repo: str | None = None) -> str | None:
+        """Build a browser URL for a commit."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        if not _owner or not _repo or not commit_hash:
+            return None
+        return f"{self.base_url}/{_owner}/{_repo}/commit/{commit_hash}"
+
+    def build_compare_url(self, base_ref: str, head_ref: str, owner: str | None = None, repo: str | None = None) -> str | None:
+        """Build a browser URL for a compare view."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        if not _owner or not _repo or not base_ref or not head_ref:
+            return None
+        return f"{self.base_url}/{_owner}/{_repo}/compare/{base_ref}...{head_ref}"
+
+    def build_pull_request_url(self, pr_number: int, owner: str | None = None, repo: str | None = None) -> str | None:
+        """Build a browser URL for a pull request."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        if not _owner or not _repo or not pr_number:
+            return None
+        return f"{self.base_url}/{_owner}/{_repo}/pulls/{pr_number}"
+
    async def _request(self, method: str, path: str, payload: dict | None = None) -> dict:
        """Perform a Gitea API request and normalize the response."""
        try:
@@ -59,6 +94,30 @@ class GiteaAPI:
        except Exception as e:
            return {"error": str(e)}

+    def _request_sync(self, method: str, path: str, payload: dict | None = None) -> dict:
+        """Perform a synchronous Gitea API request."""
+        request = urllib.request.Request(
+            self._api_url(path),
+            headers=self.get_auth_headers(),
+            method=method.upper(),
+        )
+        data = None
+        if payload is not None:
+            data = json.dumps(payload).encode('utf-8')
+            request.data = data
+        try:
+            with urllib.request.urlopen(request) as response:
+                body = response.read().decode('utf-8')
+                return json.loads(body) if body else {}
+        except urllib.error.HTTPError as exc:
+            try:
+                body = exc.read().decode('utf-8')
+            except Exception:
+                body = str(exc)
+            return {'error': body, 'status_code': exc.code}
+        except Exception as exc:
+            return {'error': str(exc)}
+
    def build_project_repo_name(self, project_id: str, project_name: str | None = None) -> str:
        """Build a repository name for a generated project."""
        preferred = (project_name or project_id or "project").strip().lower().replace(" ", "-")
@@ -97,6 +156,10 @@ class GiteaAPI:
            result.setdefault("status", "created")
        return result

+    async def get_current_user(self) -> dict:
+        """Get the user associated with the configured token."""
+        return await self._request("GET", "user")
+
    async def create_branch(self, branch: str, base: str = "main", owner: str | None = None, repo: str | None = None):
        """Create a new branch."""
        _owner = owner or self.owner
@@ -127,6 +190,110 @@ class GiteaAPI:
        }
        return await self._request("POST", f"repos/{_owner}/{_repo}/pulls", payload)

+    async def list_pull_requests(
+        self,
+        owner: str | None = None,
+        repo: str | None = None,
+        state: str = 'open',
+    ) -> dict | list:
+        """List pull requests for a repository."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return await self._request("GET", f"repos/{_owner}/{_repo}/pulls?state={state}")
+
+    def list_pull_requests_sync(
+        self,
+        owner: str | None = None,
+        repo: str | None = None,
+        state: str = 'open',
+    ) -> dict | list:
+        """Synchronously list pull requests for a repository."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return self._request_sync("GET", f"repos/{_owner}/{_repo}/pulls?state={state}")
+
+    async def list_repositories(self, owner: str | None = None) -> dict | list:
+        """List repositories within the configured organization."""
+        _owner = owner or self.owner
+        return await self._request("GET", f"orgs/{_owner}/repos")
+
+    def list_repositories_sync(self, owner: str | None = None) -> dict | list:
+        """Synchronously list repositories within the configured organization."""
+        _owner = owner or self.owner
+        return self._request_sync("GET", f"orgs/{_owner}/repos")
+
+    async def list_branches(self, owner: str | None = None, repo: str | None = None) -> dict | list:
+        """List repository branches."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return await self._request("GET", f"repos/{_owner}/{_repo}/branches")
+
+    def list_branches_sync(self, owner: str | None = None, repo: str | None = None) -> dict | list:
+        """Synchronously list repository branches."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return self._request_sync("GET", f"repos/{_owner}/{_repo}/branches")
+
+    async def list_repo_commits(
+        self,
+        owner: str | None = None,
+        repo: str | None = None,
+        limit: int = 25,
+        branch: str | None = None,
+    ) -> dict | list:
+        """List recent commits for a repository."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        branch_query = f"&sha={branch}" if branch else ""
+        return await self._request("GET", f"repos/{_owner}/{_repo}/commits?limit={limit}{branch_query}")
+
+    def list_repo_commits_sync(
+        self,
+        owner: str | None = None,
+        repo: str | None = None,
+        limit: int = 25,
+        branch: str | None = None,
+    ) -> dict | list:
+        """Synchronously list recent commits for a repository."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        branch_query = f"&sha={branch}" if branch else ""
+        return self._request_sync("GET", f"repos/{_owner}/{_repo}/commits?limit={limit}{branch_query}")
+
+    async def get_commit(
+        self,
+        commit_hash: str,
+        owner: str | None = None,
+        repo: str | None = None,
+    ) -> dict:
+        """Return one commit by hash."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return await self._request("GET", f"repos/{_owner}/{_repo}/git/commits/{commit_hash}")
+
+    def get_commit_sync(
+        self,
+        commit_hash: str,
+        owner: str | None = None,
+        repo: str | None = None,
+    ) -> dict:
+        """Synchronously return one commit by hash."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return self._request_sync("GET", f"repos/{_owner}/{_repo}/git/commits/{commit_hash}")
+
+    async def get_pull_request(self, pr_number: int, owner: str | None = None, repo: str | None = None) -> dict:
+        """Return one pull request by number."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return await self._request("GET", f"repos/{_owner}/{_repo}/pulls/{pr_number}")
+
+    def get_pull_request_sync(self, pr_number: int, owner: str | None = None, repo: str | None = None) -> dict:
+        """Synchronously return one pull request by number."""
+        _owner = owner or self.owner
+        _repo = repo or self.repo
+        return self._request_sync("GET", f"repos/{_owner}/{_repo}/pulls/{pr_number}")
+
    async def push_commit(
        self,
        branch: str,