feat: GOD MODE+ v4.0 - tools/github_tool.py

tools/github_tool.py

@@ -0,0 +1,310 @@
+"""
+GitHub Tool — Real Autonomous GitHub Operations
+cloneRepo / createRepo / createBranch / commitChanges / pushChanges / openPR / readIssues
+"""
+
+import asyncio
+import os
+import subprocess
+from typing import Dict, List, Optional
+
+import httpx
+import structlog
+
+log = structlog.get_logger()
+
+GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN", "")
+WORKSPACE = os.environ.get("WORKSPACE_DIR", "/tmp/god_workspace")
+
+
+class GitHubTool:
+    def __init__(self):
+        self.token = GITHUB_TOKEN or os.environ.get("GITHUB_TOKEN", "")
+        self.api = "https://api.github.com"
+        self.headers = {
+            "Authorization": f"token {self.token}",
+            "Accept": "application/vnd.github.v3+json",
+            "Content-Type": "application/json",
+        }
+
+    def _refresh_token(self):
+        self.token = os.environ.get("GITHUB_TOKEN", "")
+        self.headers["Authorization"] = f"token {self.token}"
+
+    async def _api(self, method: str, path: str, **kwargs) -> Dict:
+        self._refresh_token()
+        url = f"{self.api}{path}" if not path.startswith("http") else path
+        async with httpx.AsyncClient(timeout=30) as client:
+            resp = await client.request(method, url, headers=self.headers, **kwargs)
+            if resp.status_code >= 400:
+                return {"success": False, "error": resp.text, "status": resp.status_code}
+            try:
+                return {"success": True, "data": resp.json()}
+            except Exception:
+                return {"success": True, "data": resp.text}
+
+    # ─── Run Git Command ───────────────────────────────────────────────────────
+
+    async def _git(self, cmd: str, cwd: str = WORKSPACE, timeout: int = 60) -> Dict:
+        try:
+            env = os.environ.copy()
+            env["GIT_ASKPASS"] = "echo"
+            env["GIT_TERMINAL_PROMPT"] = "0"
+            if self.token:
+                env["GIT_TOKEN"] = self.token
+            proc = await asyncio.create_subprocess_shell(
+                cmd,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+                cwd=cwd,
+                env=env,
+            )
+            stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+            out = stdout.decode("utf-8", errors="replace")
+            err = stderr.decode("utf-8", errors="replace")
+            return {
+                "success": proc.returncode == 0,
+                "stdout": out,
+                "stderr": err,
+                "returncode": proc.returncode,
+                "output": out or err,
+            }
+        except asyncio.TimeoutError:
+            return {"success": False, "error": f"Git command timed out: {cmd[:60]}"}
+        except Exception as e:
+            return {"success": False, "error": str(e)}
+
+    # ─── Clone Repo ───────────────────────────────────────────────────────────
+
+    async def clone_repo(self, repo_url: str, dest: str = "", branch: str = "") -> Dict:
+        """Clone a GitHub repository into workspace."""
+        if self.token and "github.com" in repo_url:
+            if repo_url.startswith("https://github.com/"):
+                repo_url = repo_url.replace(
+                    "https://github.com/",
+                    f"https://{self.token}@github.com/",
+                )
+        dest_path = dest or os.path.join(WORKSPACE, repo_url.split("/")[-1].replace(".git", ""))
+        os.makedirs(dest_path, exist_ok=True)
+        branch_flag = f"--branch {branch}" if branch else ""
+        result = await self._git(f"git clone --depth 1 {branch_flag} {repo_url} .", cwd=dest_path, timeout=120)
+        if result["success"]:
+            return {
+                "success": True,
+                "repo_url": repo_url,
+                "local_path": dest_path,
+                "action": "cloned",
+            }
+        return result
+
+    # ─── Create Repo ──────────────────────────────────────────────────────────
+
+    async def create_repo(
+        self,
+        name: str,
+        description: str = "",
+        private: bool = False,
+        auto_init: bool = True,
+    ) -> Dict:
+        """Create a new GitHub repo via API."""
+        result = await self._api("POST", "/user/repos", json={
+            "name": name,
+            "description": description,
+            "private": private,
+            "auto_init": auto_init,
+        })
+        if result["success"]:
+            data = result["data"]
+            return {
+                "success": True,
+                "repo_name": name,
+                "url": data.get("html_url", ""),
+                "clone_url": data.get("clone_url", ""),
+                "action": "created",
+            }
+        return result
+
+    # ─── Create Branch ────────────────────────────────────────────────────────
+
+    async def create_branch(self, branch: str, cwd: str = WORKSPACE, from_branch: str = "main") -> Dict:
+        """Create and checkout a new git branch."""
+        # First fetch to ensure we have latest
+        await self._git(f"git fetch origin {from_branch} --quiet", cwd=cwd)
+        result = await self._git(f"git checkout -b {branch}", cwd=cwd)
+        if result["success"]:
+            return {"success": True, "branch": branch, "action": "created_and_checked_out"}
+        # Try checkout if branch exists
+        result2 = await self._git(f"git checkout {branch}", cwd=cwd)
+        return result2 if result2["success"] else result
+
+    # ─── Commit Changes ───────────────────────────────────────────────────────
+
+    async def commit_changes(
+        self,
+        message: str,
+        cwd: str = WORKSPACE,
+        files: Optional[List[str]] = None,
+    ) -> Dict:
+        """Stage and commit changes."""
+        if files:
+            for f in files:
+                await self._git(f"git add {f}", cwd=cwd)
+        else:
+            await self._git("git add -A", cwd=cwd)
+
+        result = await self._git(f'git commit -m "{message}"', cwd=cwd)
+        return {
+            "success": result["success"],
+            "message": message,
+            "output": result.get("output", ""),
+            "action": "committed",
+        }
+
+    # ─── Push Changes ─────────────────────────────────────────────────────────
+
+    async def push_changes(self, branch: str = "", cwd: str = WORKSPACE, force: bool = False) -> Dict:
+        """Push commits to remote."""
+        self._refresh_token()
+        branch_arg = branch or ""
+        force_flag = "--force" if force else ""
+        if branch_arg:
+            cmd = f"git push origin {branch_arg} {force_flag}".strip()
+        else:
+            cmd = f"git push {force_flag}".strip()
+        result = await self._git(cmd, cwd=cwd, timeout=60)
+        return {
+            "success": result["success"],
+            "branch": branch_arg,
+            "output": result.get("output", ""),
+            "action": "pushed",
+        }
+
+    # ─── Open PR ──────────────────────────────────────────────────────────────
+
+    async def open_pr(
+        self,
+        owner: str,
+        repo: str,
+        title: str,
+        body: str = "",
+        head: str = "genspark_ai_developer",
+        base: str = "main",
+    ) -> Dict:
+        """Create a pull request via GitHub API."""
+        result = await self._api("POST", f"/repos/{owner}/{repo}/pulls", json={
+            "title": title,
+            "body": body,
+            "head": head,
+            "base": base,
+        })
+        if result["success"]:
+            data = result["data"]
+            return {
+                "success": True,
+                "pr_url": data.get("html_url", ""),
+                "pr_number": data.get("number"),
+                "title": title,
+                "action": "pr_opened",
+            }
+        return result
+
+    # ─── Read Issues ──────────────────────────────────────────────────────────
+
+    async def read_issues(self, owner: str, repo: str, state: str = "open", limit: int = 10) -> Dict:
+        result = await self._api("GET", f"/repos/{owner}/{repo}/issues?state={state}&per_page={limit}")
+        if result["success"]:
+            issues = result["data"]
+            return {
+                "success": True,
+                "issues": [
+                    {
+                        "number": i.get("number"),
+                        "title": i.get("title"),
+                        "state": i.get("state"),
+                        "body": (i.get("body") or "")[:500],
+                        "url": i.get("html_url"),
+                        "labels": [l["name"] for l in i.get("labels", [])],
+                    }
+                    for i in issues
+                ],
+            }
+        return result
+
+    # ─── Get Repo Info ────────────────────────────────────────────────────────
+
+    async def get_repo_info(self, owner: str, repo: str) -> Dict:
+        result = await self._api("GET", f"/repos/{owner}/{repo}")
+        if result["success"]:
+            d = result["data"]
+            return {
+                "success": True,
+                "name": d.get("name"),
+                "full_name": d.get("full_name"),
+                "description": d.get("description"),
+                "url": d.get("html_url"),
+                "clone_url": d.get("clone_url"),
+                "default_branch": d.get("default_branch"),
+                "stars": d.get("stargazers_count"),
+                "language": d.get("language"),
+                "private": d.get("private"),
+            }
+        return result
+
+    # ─── Get File from GitHub ─────────────────────────────────────────────────
+
+    async def get_file(self, owner: str, repo: str, path: str, branch: str = "main") -> Dict:
+        import base64
+        result = await self._api("GET", f"/repos/{owner}/{repo}/contents/{path}?ref={branch}")
+        if result["success"]:
+            d = result["data"]
+            content = base64.b64decode(d.get("content", "")).decode("utf-8", errors="replace")
+            return {"success": True, "path": path, "content": content, "sha": d.get("sha")}
+        return result
+
+    # ─── Setup Git Config ─────────────────────────────────────────────────────
+
+    async def setup_git_config(self, cwd: str = WORKSPACE, name: str = "God Agent", email: str = "god-agent@ai.com") -> Dict:
+        await self._git(f'git config user.name "{name}"', cwd=cwd)
+        await self._git(f'git config user.email "{email}"', cwd=cwd)
+        return {"success": True, "action": "git_config_set"}
+
+    # ─── Get Status ───────────────────────────────────────────────────────────
+
+    async def status(self, cwd: str = WORKSPACE) -> Dict:
+        result = await self._git("git status --short", cwd=cwd)
+        log_result = await self._git("git log --oneline -5", cwd=cwd)
+        branch_result = await self._git("git branch --show-current", cwd=cwd)
+        return {
+            "success": True,
+            "status": result.get("stdout", ""),
+            "recent_commits": log_result.get("stdout", ""),
+            "current_branch": branch_result.get("stdout", "").strip(),
+        }
+
+    # ─── Init + Setup Repo ────────────────────────────────────────────────────
+
+    async def init_repo(self, cwd: str = WORKSPACE, remote_url: str = "") -> Dict:
+        await self._git("git init", cwd=cwd)
+        await self.setup_git_config(cwd=cwd)
+        if remote_url:
+            await self._git(f"git remote add origin {remote_url}", cwd=cwd)
+        return {"success": True, "cwd": cwd, "action": "initialized"}
+
+    # ─── Full Autonomous Commit + Push ────────────────────────────────────────
+
+    async def auto_commit_push(
+        self,
+        message: str,
+        branch: str = "main",
+        cwd: str = WORKSPACE,
+    ) -> Dict:
+        """All-in-one: add → commit → push."""
+        await self.setup_git_config(cwd=cwd)
+        commit_r = await self.commit_changes(message, cwd=cwd)
+        push_r = await self.push_changes(branch, cwd=cwd)
+        return {
+            "success": push_r.get("success", False),
+            "commit": commit_r,
+            "push": push_r,
+            "action": "committed_and_pushed",
+        }