mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 05:30:39 +00:00
359 lines
12 KiB
Python
359 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
GitHub API client for deep research.
|
|
Uses requests for HTTP operations.
|
|
"""
|
|
|
|
import os
|
|
import json
|
|
import sys
|
|
from typing import Any, Dict, List, Optional
|
|
from urllib.parse import quote
|
|
|
|
try:
|
|
import requests
|
|
except ImportError:
|
|
# Fallback to urllib if requests not available
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
class RequestsFallback:
|
|
"""Minimal requests-like interface using urllib."""
|
|
|
|
class Response:
|
|
def __init__(self, data: bytes, status: int):
|
|
self._data = data
|
|
self.status_code = status
|
|
self.text = data.decode("utf-8", errors="replace")
|
|
|
|
def json(self):
|
|
return json.loads(self._data)
|
|
|
|
def raise_for_status(self):
|
|
if self.status_code >= 400:
|
|
raise Exception(f"HTTP {self.status_code}")
|
|
|
|
@staticmethod
|
|
def get(url: str, headers: dict = None, params: dict = None, timeout: int = 30):
|
|
if params:
|
|
query = urllib.parse.urlencode(params)
|
|
url = f"{url}?{query}"
|
|
|
|
req = urllib.request.Request(url, headers=headers or {})
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=timeout) as resp:
|
|
return RequestsFallback.Response(resp.read(), resp.status)
|
|
except urllib.error.HTTPError as e:
|
|
return RequestsFallback.Response(e.read(), e.code)
|
|
|
|
requests = RequestsFallback()
|
|
|
|
|
|
class GitHubAPIError(RuntimeError):
|
|
"""A fetch this client could not complete.
|
|
|
|
The two content readers used to answer with strings such as
|
|
``"[README not found: HTTP 404]"``, which put a failure on the same channel and with the
|
|
same exit code as a successful body: a rate-limited run and a missing README looked alike
|
|
to whoever read stdout. Raising keeps the HTTP status visible on stderr and out of the
|
|
content path.
|
|
"""
|
|
|
|
|
|
class GitHubAPI:
|
|
"""GitHub API client for repository analysis."""
|
|
|
|
BASE_URL = "https://api.github.com"
|
|
|
|
def __init__(self, token: Optional[str] = None):
|
|
"""
|
|
Initialize GitHub API client.
|
|
|
|
Args:
|
|
token:
|
|
Optional GitHub personal access token for higher rate limits.
|
|
User can set it in .env by uncommenting the line "GITHUB_TOKEN=your-github-token".
|
|
"""
|
|
self.headers = {
|
|
"Accept": "application/vnd.github.v3+json",
|
|
"User-Agent": "Deep-Research-Bot/1.0",
|
|
}
|
|
if token:
|
|
self.headers["Authorization"] = f"token {token}"
|
|
|
|
def _get(
|
|
self, endpoint: str, params: Optional[Dict] = None, accept: Optional[str] = None
|
|
) -> Any:
|
|
"""Make GET request to GitHub API."""
|
|
url = f"{self.BASE_URL}{endpoint}"
|
|
headers = self.headers.copy()
|
|
if accept:
|
|
headers["Accept"] = accept
|
|
|
|
resp = requests.get(url, headers=headers, params=params, timeout=30)
|
|
resp.raise_for_status()
|
|
|
|
if "application/vnd.github.raw" in (accept or ""):
|
|
return resp.text
|
|
return resp.json()
|
|
|
|
def get_repo_info(self, owner: str, repo: str) -> Dict:
|
|
"""Get basic repository information."""
|
|
return self._get(f"/repos/{owner}/{repo}")
|
|
|
|
def get_readme(self, owner: str, repo: str) -> str:
|
|
"""Get repository README content as markdown."""
|
|
try:
|
|
return self._get(
|
|
f"/repos/{owner}/{repo}/readme", accept="application/vnd.github.raw"
|
|
)
|
|
except Exception as e:
|
|
raise GitHubAPIError(f"README fetch failed for {owner}/{repo}: {e}") from e
|
|
|
|
def get_tree(
|
|
self, owner: str, repo: str, branch: str = "HEAD", recursive: bool = True
|
|
) -> Dict:
|
|
"""Get repository directory tree.
|
|
|
|
``HEAD`` is resolved by GitHub to the repository's default branch, so a
|
|
repository that uses neither ``main`` nor ``master`` gets its current tree.
|
|
Guessing those two names instead served the tree of an abandoned branch
|
|
whenever the repository kept a stale ``master`` around.
|
|
"""
|
|
params = {"recursive": "1"} if recursive else {}
|
|
return self._get(f"/repos/{owner}/{repo}/git/trees/{branch}", params)
|
|
|
|
def get_file_content(self, owner: str, repo: str, path: str) -> str:
|
|
"""Get content of a file using its raw repository-relative path."""
|
|
try:
|
|
return self._get(
|
|
f"/repos/{owner}/{repo}/contents/{quote(path, safe='/')}",
|
|
accept="application/vnd.github.raw",
|
|
)
|
|
except Exception as e:
|
|
raise GitHubAPIError(
|
|
f"File fetch failed for {owner}/{repo}:{path}: {e}"
|
|
) from e
|
|
|
|
def get_languages(self, owner: str, repo: str) -> Dict[str, int]:
|
|
"""Get repository languages and their bytes."""
|
|
return self._get(f"/repos/{owner}/{repo}/languages")
|
|
|
|
def get_contributors(self, owner: str, repo: str, limit: int = 30) -> List[Dict]:
|
|
"""Get repository contributors."""
|
|
return self._get(
|
|
f"/repos/{owner}/{repo}/contributors", params={"per_page": min(limit, 100)}
|
|
)
|
|
|
|
def get_recent_commits(
|
|
self, owner: str, repo: str, limit: int = 50, since: Optional[str] = None
|
|
) -> List[Dict]:
|
|
"""
|
|
Get recent commits.
|
|
|
|
Args:
|
|
owner: Repository owner
|
|
repo: Repository name
|
|
limit: Max commits to fetch
|
|
since: ISO date string to fetch commits since
|
|
"""
|
|
params = {"per_page": min(limit, 100)}
|
|
if since:
|
|
params["since"] = since
|
|
return self._get(f"/repos/{owner}/{repo}/commits", params)
|
|
|
|
def get_issues(
|
|
self,
|
|
owner: str,
|
|
repo: str,
|
|
state: str = "all",
|
|
limit: int = 30,
|
|
labels: Optional[str] = None,
|
|
) -> List[Dict]:
|
|
"""
|
|
Get repository issues, excluding pull requests.
|
|
|
|
Args:
|
|
state: 'open', 'closed', or 'all'
|
|
labels: Comma-separated label names
|
|
"""
|
|
limit = min(limit, 100)
|
|
if limit <= 0:
|
|
return []
|
|
params = {"state": state, "per_page": limit}
|
|
if labels:
|
|
params["labels"] = labels
|
|
issues = []
|
|
page = 1
|
|
while len(issues) < limit:
|
|
items = self._get(f"/repos/{owner}/{repo}/issues", params)
|
|
issues.extend(item for item in items if "pull_request" not in item)
|
|
if len(items) < limit:
|
|
break
|
|
page += 1
|
|
params["page"] = page
|
|
return issues[:limit]
|
|
|
|
def get_pull_requests(
|
|
self, owner: str, repo: str, state: str = "all", limit: int = 30
|
|
) -> List[Dict]:
|
|
"""Get repository pull requests."""
|
|
return self._get(
|
|
f"/repos/{owner}/{repo}/pulls",
|
|
params={"state": state, "per_page": min(limit, 100)},
|
|
)
|
|
|
|
def get_releases(self, owner: str, repo: str, limit: int = 10) -> List[Dict]:
|
|
"""Get repository releases."""
|
|
return self._get(
|
|
f"/repos/{owner}/{repo}/releases", params={"per_page": min(limit, 100)}
|
|
)
|
|
|
|
def get_tags(self, owner: str, repo: str, limit: int = 20) -> List[Dict]:
|
|
"""Get repository tags."""
|
|
return self._get(
|
|
f"/repos/{owner}/{repo}/tags", params={"per_page": min(limit, 100)}
|
|
)
|
|
|
|
def search_issues(self, owner: str, repo: str, query: str, limit: int = 30) -> Dict:
|
|
"""Search issues and PRs in repository."""
|
|
q = f"repo:{owner}/{repo} {query}"
|
|
return self._get("/search/issues", params={"q": q, "per_page": min(limit, 100)})
|
|
|
|
def get_commit_activity(self, owner: str, repo: str) -> List[Dict]:
|
|
"""Get weekly commit activity for the last year."""
|
|
return self._get(f"/repos/{owner}/{repo}/stats/commit_activity")
|
|
|
|
def get_code_frequency(self, owner: str, repo: str) -> List[List[int]]:
|
|
"""Get weekly additions/deletions."""
|
|
return self._get(f"/repos/{owner}/{repo}/stats/code_frequency")
|
|
|
|
def format_tree(self, tree_data: Dict, max_depth: int = 3) -> str:
|
|
"""
|
|
Format tree data as text directory structure.
|
|
|
|
Args:
|
|
tree_data: Response from get_tree()
|
|
max_depth: Maximum depth to display
|
|
"""
|
|
if "tree" not in tree_data:
|
|
return "[Unable to parse tree]"
|
|
|
|
lines = []
|
|
for item in tree_data["tree"]:
|
|
path = item["path"]
|
|
depth = path.count("/")
|
|
if depth < max_depth:
|
|
indent = " " * depth
|
|
name = path.split("/")[-1]
|
|
if item["type"] == "tree":
|
|
lines.append(f"{indent}{name}/")
|
|
else:
|
|
lines.append(f"{indent}{name}")
|
|
|
|
return "\n".join(lines[:100]) # Limit output
|
|
|
|
def summarize_repo(self, owner: str, repo: str) -> Dict:
|
|
"""
|
|
Get comprehensive repository summary.
|
|
|
|
Returns dict with: info, languages, contributor_count,
|
|
recent_activity, top_issues, latest_release
|
|
"""
|
|
info = self.get_repo_info(owner, repo)
|
|
|
|
summary = {
|
|
"name": info.get("full_name"),
|
|
"description": info.get("description"),
|
|
"url": info.get("html_url"),
|
|
"stars": info.get("stargazers_count"),
|
|
"forks": info.get("forks_count"),
|
|
"open_issues": info.get("open_issues_count"),
|
|
"language": info.get("language"),
|
|
"license": info.get("license", {}).get("spdx_id")
|
|
if info.get("license")
|
|
else None,
|
|
"created_at": info.get("created_at"),
|
|
"updated_at": info.get("updated_at"),
|
|
"pushed_at": info.get("pushed_at"),
|
|
"default_branch": info.get("default_branch"),
|
|
"topics": info.get("topics", []),
|
|
}
|
|
|
|
# Add languages
|
|
try:
|
|
summary["languages"] = self.get_languages(owner, repo)
|
|
except Exception:
|
|
summary["languages"] = {}
|
|
|
|
# Add contributor count
|
|
try:
|
|
contributors = self.get_contributors(owner, repo, limit=1)
|
|
# GitHub returns Link header with total, but we approximate
|
|
summary["contributor_count"] = len(
|
|
self.get_contributors(owner, repo, limit=100)
|
|
)
|
|
except Exception:
|
|
summary["contributor_count"] = "N/A"
|
|
|
|
# Latest release
|
|
try:
|
|
releases = self.get_releases(owner, repo, limit=1)
|
|
if releases:
|
|
summary["latest_release"] = {
|
|
"tag": releases[0].get("tag_name"),
|
|
"name": releases[0].get("name"),
|
|
"date": releases[0].get("published_at"),
|
|
}
|
|
except Exception:
|
|
summary["latest_release"] = None
|
|
|
|
return summary
|
|
|
|
|
|
def main():
|
|
"""CLI interface for testing."""
|
|
if len(sys.argv) < 3:
|
|
print("Usage: python github_api.py <owner> <repo> [command]")
|
|
print("Commands: info, readme, tree, languages, contributors,")
|
|
print(" commits, issues, prs, releases, summary")
|
|
sys.exit(1)
|
|
|
|
owner, repo = sys.argv[1], sys.argv[2]
|
|
command = sys.argv[3] if len(sys.argv) > 3 else "summary"
|
|
|
|
token = os.getenv("GITHUB_TOKEN")
|
|
api = GitHubAPI(token=token)
|
|
|
|
commands = {
|
|
"info": lambda: api.get_repo_info(owner, repo),
|
|
"readme": lambda: api.get_readme(owner, repo),
|
|
"tree": lambda: api.format_tree(api.get_tree(owner, repo)),
|
|
"languages": lambda: api.get_languages(owner, repo),
|
|
"contributors": lambda: api.get_contributors(owner, repo),
|
|
"commits": lambda: api.get_recent_commits(owner, repo),
|
|
"issues": lambda: api.get_issues(owner, repo),
|
|
"prs": lambda: api.get_pull_requests(owner, repo),
|
|
"releases": lambda: api.get_releases(owner, repo),
|
|
"summary": lambda: api.summarize_repo(owner, repo),
|
|
}
|
|
|
|
if command not in commands:
|
|
print(f"Unknown command: {command}")
|
|
sys.exit(1)
|
|
|
|
try:
|
|
result = commands[command]()
|
|
if isinstance(result, str):
|
|
print(result)
|
|
else:
|
|
print(json.dumps(result, indent=2, default=str))
|
|
except Exception as e:
|
|
print(f"Error: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|