Add/Update wolf/gitea.py by Wolf

wolf/gitea.py

@@ -0,0 +1,95 @@
+import requests
+import logging
+import base64
+import json
+
+class GiteaClient:
+    """
+    Gitea API client for Wolf.
+    """
+    def __init__(self, base_url, token):
+        self.base_url = base_url.rstrip('/')
+        self.token = token
+        self.headers = {
+            "Authorization": f"token {self.token}",
+            "Content-Type": "application/json"
+        }
+
+    def get_issues(self, owner, repo, state="open"):
+        url = f"{self.base_url}/repos/{owner}/{repo}/issues?state={state}"
+        response = requests.get(url, headers=self.headers)
+        response.raise_for_status()
+        return response.json()
+
+    def create_branch(self, owner, repo, branch_name, old_branch="main"):
+        url = f"{self.base_url}/repos/{owner}/{repo}/branches"
+        data = {
+            "new_branch_name": branch_name,
+            "old_branch_name": old_branch
+        }
+        response = requests.post(url, headers=self.headers, json=data)
+        if response.status_code == 409: # Branch already exists
+            logging.info(f"Branch {branch_name} already exists in {owner}/{repo}")
+            return True
+        response.raise_for_status()
+        return True
+
+    def create_file(self, owner, repo, path, content, branch, message="Commit by Wolf"):
+        url = f"{self.base_url}/repos/{owner}/{repo}/contents/{path}"
+        data = {
+            "branch": branch,
+            "content": base64.b64encode(content.encode()).decode(),
+            "message": message
+        }
+        response = requests.post(url, headers=self.headers, json=data)
+        response.raise_for_status()
+        return response.json()
+
+    def update_file(self, owner, repo, path, content, branch, sha, message="Update by Wolf"):
+        url = f"{self.base_url}/repos/{owner}/{repo}/contents/{path}"
+        data = {
+            "branch": branch,
+            "content": base64.b64encode(content.encode()).decode(),
+            "sha": sha,
+            "message": message
+        }
+        response = requests.put(url, headers=self.headers, json=data)
+        response.raise_for_status()
+        return response.json()
+
+    def get_file(self, owner, repo, path, ref="main"):
+        url = f"{self.base_url}/repos/{owner}/{repo}/contents/{path}?ref={ref}"
+        response = requests.get(url, headers=self.headers)
+        if response.status_code == 404:
+            return None
+        response.raise_for_status()
+        return response.json()
+
+    def create_pull_request(self, owner, repo, title, body, head, base="main"):
+        url = f"{self.base_url}/repos/{owner}/{repo}/pulls"
+        data = {
+            "title": title,
+            "body": body,
+            "head": head,
+            "base": base
+        }
+        response = requests.post(url, headers=self.headers, json=data)
+        response.raise_for_status()
+        return response.json()
+
+    def get_pull_request(self, owner, repo, index):
+        url = f"{self.base_url}/repos/{owner}/{repo}/pulls/{index}"
+        response = requests.get(url, headers=self.headers)
+        response.raise_for_status()
+        return response.json()
+
+    def get_pr_status(self, owner, repo, index):
+        # In Gitea, status can be checked via the combined status API or the PR itself
+        url = f"{self.base_url}/repos/{owner}/{repo}/pulls/{index}/status"
+        response = requests.get(url, headers=self.headers)
+        # If status endpoint doesn't exist, fallback to checking if it's mergeable
+        if response.status_code == 404:
+            pr = self.get_pull_request(owner, repo, index)
+            return {"state": "success" if pr.get("mergeable") else "pending"}
+        response.raise_for_status()
+        return response.json()