Spaces:
Sleeping
Sleeping
| """GitHub tools — PR diffs, repo info.""" | |
| import os | |
| import re | |
| from typing import Optional | |
| async def github_pr_diff(pr_url: str) -> dict: | |
| """Fetch a GitHub PR's diff and metadata. | |
| Args: | |
| pr_url: Full GitHub PR URL (e.g. https://github.com/owner/repo/pull/123) | |
| """ | |
| import httpx | |
| # Parse owner/repo/number from URL | |
| match = re.match(r"https?://github\.com/([^/]+)/([^/]+)/pull/(\d+)", pr_url) | |
| if not match: | |
| return {"error": f"Invalid PR URL format: {pr_url}"} | |
| owner, repo, number = match.group(1), match.group(2), match.group(3) | |
| api_base = f"https://api.github.com/repos/{owner}/{repo}/pulls/{number}" | |
| headers = {"Accept": "application/vnd.github.v3+json"} | |
| token = os.environ.get("GITHUB_TOKEN") | |
| if token: | |
| headers["Authorization"] = f"token {token}" | |
| async with httpx.AsyncClient(timeout=20) as client: | |
| # Get PR metadata | |
| meta_res = await client.get(api_base, headers=headers) | |
| if meta_res.status_code != 200: | |
| return {"error": f"GitHub API {meta_res.status_code}: {meta_res.text[:200]}"} | |
| meta = meta_res.json() | |
| # Get diff | |
| diff_headers = {**headers, "Accept": "application/vnd.github.v3.diff"} | |
| diff_res = await client.get(api_base, headers=diff_headers) | |
| diff = diff_res.text if diff_res.status_code == 200 else "Could not fetch diff" | |
| # Truncate large diffs | |
| if len(diff) > 20000: | |
| diff = diff[:20000] + "\n\n[diff truncated — too large]" | |
| return { | |
| "title": meta.get("title"), | |
| "author": meta.get("user", {}).get("login"), | |
| "state": meta.get("state"), | |
| "additions": meta.get("additions"), | |
| "deletions": meta.get("deletions"), | |
| "changed_files": meta.get("changed_files"), | |
| "base_branch": meta.get("base", {}).get("ref"), | |
| "head_branch": meta.get("head", {}).get("ref"), | |
| "description": (meta.get("body") or "")[:1000], | |
| "diff": diff, | |
| } | |
| async def github_repo_info(repo_url: str) -> dict: | |
| """Get basic info about a GitHub repository. | |
| Args: | |
| repo_url: GitHub repo URL (e.g. https://github.com/owner/repo) | |
| """ | |
| import httpx | |
| match = re.match(r"https?://github\.com/([^/]+)/([^/]+)", repo_url) | |
| if not match: | |
| return {"error": f"Invalid repo URL: {repo_url}"} | |
| owner, repo = match.group(1), match.group(2) | |
| headers = {"Accept": "application/vnd.github.v3+json"} | |
| token = os.environ.get("GITHUB_TOKEN") | |
| if token: | |
| headers["Authorization"] = f"token {token}" | |
| async with httpx.AsyncClient(timeout=15) as client: | |
| res = await client.get( | |
| f"https://api.github.com/repos/{owner}/{repo}", | |
| headers=headers, | |
| ) | |
| if res.status_code != 200: | |
| return {"error": f"GitHub API {res.status_code}"} | |
| data = res.json() | |
| return { | |
| "name": data.get("full_name"), | |
| "description": data.get("description"), | |
| "language": data.get("language"), | |
| "stars": data.get("stargazers_count"), | |
| "forks": data.get("forks_count"), | |
| "open_issues": data.get("open_issues_count"), | |
| "topics": data.get("topics", []), | |
| "default_branch": data.get("default_branch"), | |
| "updated_at": data.get("updated_at"), | |
| } | |