Hamza4100/ai-workflow-agent
0
1# GitHub Search Tool
2"""
3Search GitHub repositories and provide recommendations.
4Simple keyword search with README summarization.
5"""
6
7import httpx
8import logging
9from typing import List, Dict, Any, Optional
10
11from config import settings
12
13logger = logging.getLogger(__name__)
14
15
16class GitHubSearchTool:
17 """
18 GitHub repository search tool.
19 Searches for relevant projects and provides recommendations.
20 """
21
22 def __init__(self):
23 self.api_base = "https://api.github.com"
24 self.token = settings.GITHUB_TOKEN
25 self.client = httpx.AsyncClient(timeout=30.0)
26
27 # Set headers
28 self.headers = {
29 "Accept": "application/vnd.github.v3+json",
30 "User-Agent": "AI-Workflow-Agent"
31 }
32 if self.token:
33 self.headers["Authorization"] = f"token {self.token}"
34
35 async def search(
36 self,
37 keywords: str,
38 max_results: int = 3,
39 language: Optional[str] = None
40 ) -> List[Dict[str, Any]]:
41 """
42 Search GitHub repositories by keywords.
43
44 Args:
45 keywords: Search terms
46 max_results: Maximum number of results (default 3)
47 language: Optional language filter
48
49 Returns:
50 List of repository info dictionaries
51 """
52 try:
53 # Build search query
54 query = keywords
55 if language:
56 query += f" language:{language}"
57
58 # Search repositories
59 response = await self.client.get(
60 f"{self.api_base}/search/repositories",
61 params={
62 "q": query,
63 "sort": "stars",
64 "order": "desc",
65 "per_page": max_results
66 },
67 headers=self.headers
68 )
69
70 if response.status_code == 200:
71 data = response.json()
72 repos = data.get("items", [])
73
74 # Extract relevant info
75 results = []
76 for repo in repos[:max_results]:
77 repo_info = await self._extract_repo_info(repo)
78 results.append(repo_info)
79
80 return results
81 else:
82 logger.error(f"GitHub search failed: {response.status_code}")
83 return []
84
85 except Exception as e:
86 logger.error(f"GitHub search error: {e}")
87 return []
88
89 async def _extract_repo_info(self, repo: Dict[str, Any]) -> Dict[str, Any]:
90 """Extract relevant information from repository data."""
91 # Get README summary
92 readme_summary = await self._get_readme_summary(
93 repo.get("owner", {}).get("login", ""),
94 repo.get("name", "")
95 )
96
97 return {
98 "name": repo.get("name", ""),
99 "full_name": repo.get("full_name", ""),
100 "url": repo.get("html_url", ""),
101 "clone_url": repo.get("clone_url", ""),
102 "description": repo.get("description", "No description"),
103 "stars": repo.get("stargazers_count", 0),
104 "forks": repo.get("forks_count", 0),
105 "language": repo.get("language", "Unknown"),
106 "topics": repo.get("topics", []),
107 "updated_at": repo.get("updated_at", ""),
108 "has_docker": await self._check_has_docker(
109 repo.get("owner", {}).get("login", ""),
110 repo.get("name", "")
111 ),
112 "readme_summary": readme_summary
113 }
114
115 async def _get_readme_summary(self, owner: str, repo: str) -> str:
116 """Fetch and summarize repository README."""
117 try:
118 response = await self.client.get(
119 f"{self.api_base}/repos/{owner}/{repo}/readme",
120 headers=self.headers
121 )
122
123 if response.status_code == 200:
124 data = response.json()
125 # README is base64 encoded
126 import base64
127 content = base64.b64decode(data.get("content", "")).decode("utf-8")
128
129 # Simple summary: first 500 chars
130 summary = content[:500].replace("\n", " ").strip()
131 if len(content) > 500:
132 summary += "..."
133 return summary
134 return "README not available"
135
136 except Exception as e:
137 logger.warning(f"README fetch error: {e}")
138 return "README not available"
139
140 async def _check_has_docker(self, owner: str, repo: str) -> bool:
141 """Check if repository has Dockerfile or docker-compose."""
142 try:
143 # Check for Dockerfile
144 response = await self.client.get(
145 f"{self.api_base}/repos/{owner}/{repo}/contents/Dockerfile",
146 headers=self.headers
147 )
148 if response.status_code == 200:
149 return True
150
151 # Check for docker-compose
152 response = await self.client.get(
153 f"{self.api_base}/repos/{owner}/{repo}/contents/docker-compose.yml",
154 headers=self.headers
155 )
156 if response.status_code == 200:
157 return True
158
159 # Check for docker-compose.yaml
160 response = await self.client.get(
161 f"{self.api_base}/repos/{owner}/{repo}/contents/docker-compose.yaml",
162 headers=self.headers
163 )
164 return response.status_code == 200
165
166 except Exception:
167 return False
168
169 async def generate_recommendation(self, repos: List[Dict[str, Any]]) -> str:
170 """
171 Generate a recommendation summary for found repositories.
172
173 Args:
174 repos: List of repository info dictionaries
175
176 Returns:
177 Recommendation text
178 """
179 if not repos:
180 return "No repositories found. Try different keywords."
181
182 # Build recommendation
183 lines = ["📦 **Repository Recommendations:**\n"]
184
185 for i, repo in enumerate(repos, 1):
186 stars = repo.get("stars", 0)
187 docker_status = "✅ Docker" if repo.get("has_docker") else "❌ No Docker"
188
189 lines.append(f"**{i}. {repo.get('name', 'Unknown')}** ⭐ {stars}")
190 lines.append(f" {repo.get('description', 'No description')}")
191 lines.append(f" Language: {repo.get('language', 'Unknown')} | {docker_status}")
192 lines.append(f" URL: {repo.get('url', '')}")
193 lines.append("")
194
195 # Add best pick recommendation
196 if repos:
197 best = max(repos, key=lambda r: r.get("stars", 0))
198 if best.get("has_docker"):
199 lines.append(f"💡 **Recommended:** {best.get('name')} (most stars + Docker support)")
200 else:
201 docker_repos = [r for r in repos if r.get("has_docker")]
202 if docker_repos:
203 best_docker = max(docker_repos, key=lambda r: r.get("stars", 0))
204 lines.append(f"💡 **Recommended:** {best_docker.get('name')} (has Docker support)")
205 else:
206 lines.append(f"💡 **Recommended:** {best.get('name')} (most stars, but needs Docker setup)")
207
208 return "\n".join(lines)
209
210 async def get_repo_details(self, repo_url: str) -> Dict[str, Any]:
211 """
212 Get detailed information about a specific repository.
213
214 Args:
215 repo_url: GitHub repository URL or owner/repo format
216
217 Returns:
218 Repository details dictionary
219 """
220 try:
221 # Parse owner/repo from URL or direct format
222 if "github.com" in repo_url:
223 parts = repo_url.rstrip("/").split("/")
224 owner = parts[-2]
225 repo = parts[-1]
226 else:
227 owner, repo = repo_url.split("/")
228
229 response = await self.client.get(
230 f"{self.api_base}/repos/{owner}/{repo}",
231 headers=self.headers
232 )
233
234 if response.status_code == 200:
235 return await self._extract_repo_info(response.json())
236 else:
237 return {"error": f"Repository not found: {response.status_code}"}
238
239 except Exception as e:
240 logger.error(f"Repo details error: {e}")
241 return {"error": str(e)}
242