Coverage for src/local_deep_research/web_search_engines/engines/search_engine_github.py: 99%

1import base64 

2import json 

3import time 

4from typing import Any, Dict, List, Optional 

5 

6from langchain_core.language_models import BaseLLM 

7from ...security.secure_logging import logger 

8 

9from ...config import llm_config 

10from ...constants import USER_AGENT 

11from ...security.safe_requests import safe_get 

12from ...utilities.json_utils import extract_json, get_llm_response_text 

13from ..search_engine_base import BaseSearchEngine, Exposure, Sensitivity 

14 

15 

16_VALID_SEARCH_TYPES = frozenset({"repositories", "code", "issues", "users"}) 

17 

18 

19class GitHubSearchEngine(BaseSearchEngine): 

20 """ 

21 GitHub search engine implementation. 

22 Provides search across GitHub repositories, code, issues, and users. 

23 """ 

24 

25 is_public = True 

26 egress_sensitivity = Sensitivity.NON_SENSITIVE 

27 egress_exposure = Exposure.EXPOSING 

28 is_code = True 

29 is_lexical = True 

30 needs_llm_relevance_filter = True 

31 

32 def __init__( 

33 self, 

34 max_results: int = 15, 

35 api_key: Optional[str] = None, 

36 search_type: str = "repositories", 

37 include_readme: bool = True, 

38 include_issues: bool = False, 

39 llm: Optional[BaseLLM] = None, 

40 max_filtered_results: Optional[int] = None, 

41 settings_snapshot: Optional[Dict[str, Any]] = None, 

42 ): 

43 """ 

44 Initialize the GitHub search engine. 

45 

46 Args: 

47 max_results: Maximum number of search results 

48 api_key: GitHub API token (can also be set via LDR_SEARCH_ENGINE_WEB_GITHUB_API_KEY env var or in UI settings) 

49 search_type: Type of GitHub search ("repositories", "code", "issues", "users") 

50 include_readme: Whether to include README content for repositories 

51 include_issues: Whether to include recent issues for repositories 

52 llm: Language model for relevance filtering 

53 max_filtered_results: Maximum number of results to keep after filtering 

54 """ 

55 # Initialize the BaseSearchEngine with LLM, max_filtered_results, and max_results 

56 super().__init__( 

57 llm=llm, 

58 max_filtered_results=max_filtered_results, 

59 max_results=max_results, 

60 settings_snapshot=settings_snapshot, 

61 ) 

62 self.api_key = api_key 

63 if search_type not in _VALID_SEARCH_TYPES: 

64 raise ValueError( 

65 f"Invalid GitHub search_type: {search_type!r}. " 

66 f"Must be one of {_VALID_SEARCH_TYPES}" 

67 ) 

68 self.search_type = search_type 

69 self.include_readme = include_readme 

70 self.include_issues = include_issues 

71 

72 self._owns_llm = False 

73 

74 # API endpoints 

75 self.api_base = "https://api.github.com" 

76 self.search_endpoint = f"{self.api_base}/search/{search_type}" 

77 

78 # Set up API headers 

79 self.headers = { 

80 "Accept": "application/vnd.github.v3+json", 

81 "User-Agent": USER_AGENT, 

82 } 

83 

84 # Add authentication if API key provided 

85 if self.api_key: 

86 self.headers["Authorization"] = f"token {self.api_key}" 

87 logger.info("Using authenticated GitHub API requests") 

88 else: 

89 logger.warning( 

90 "No GitHub API key provided. Rate limits will be restricted." 

91 ) 

92 

93 def close(self) -> None: 

94 """Close the lazily-loaded LLM client if this engine created it.""" 

95 from ...utilities.resource_utils import safe_close 

96 

97 if self._owns_llm: 

98 safe_close(self.llm, "GitHub LLM") 

99 super().close() 

100 

101 def _handle_rate_limits(self, response): 

102 """Handle GitHub API rate limits by logging warnings and sleeping if necessary""" 

103 remaining = int(response.headers.get("X-RateLimit-Remaining", 60)) 

104 reset_time = int(response.headers.get("X-RateLimit-Reset", 0)) 

105 

106 if remaining < 5: 

107 current_time = time.time() 

108 wait_time = max(reset_time - current_time, 0) 

109 logger.warning( 

110 f"GitHub API rate limit almost reached. {remaining} requests remaining." 

111 ) 

112 

113 if wait_time > 0 and remaining == 0: 

114 logger.warning( 

115 f"GitHub API rate limit exceeded. Waiting {wait_time:.0f} seconds." 

116 ) 

117 time.sleep(min(wait_time, 60)) # Wait at most 60 seconds 

118 

119 def _optimize_github_query(self, query: str) -> str: 

120 """ 

121 Optimize the GitHub search query using LLM to improve search results. 

122 

123 Args: 

124 query: Original search query 

125 

126 Returns: 

127 Optimized GitHub search query 

128 """ 

129 # Get LLM from config if not already set 

130 if not self.llm: 

131 try: 

132 self.llm = llm_config.get_llm() 

133 self._owns_llm = True 

134 if not self.llm: 

135 logger.warning("No LLM available for query optimization") 

136 return query 

137 except Exception as e: 

138 safe_msg = self._scrub_error(e) 

139 logger.warning(f"Error getting LLM from config: {safe_msg}") 

140 return query 

141 

142 prompt = f"""Transform this GitHub search query into an optimized version for the GitHub search API. Follow these steps: 

143 1. Strip question words (e.g., 'what', 'are', 'is'), stop words (e.g., 'and', 'as', 'of', 'on'), and redundant terms (e.g., 'repositories', 'repos', 'github') since they're implied by the search context. 

144 2. Keep only domain-specific keywords and avoid using "-related" terms. 

145 3. Add GitHub-specific filters with dynamic thresholds based on query context: 

146 - For stars: Use higher threshold (e.g., 'stars:>1000') for mainstream topics, lower (e.g., 'stars:>50') for specialized topics 

147 - For language: Detect programming language from query or omit if unclear 

148 - For search scope: Use 'in:name,description,readme' for general queries, 'in:file' for code-specific queries 

149 4. For date ranges, adapt based on query context: 

150 - For emerging: Use 'created:>2024-01-01' 

151 - For mature: Use 'pushed:>2023-01-01' 

152 - For historical research: Use 'created:2020-01-01..2024-01-01' 

153 5. For excluding results, adapt based on query: 

154 - Exclude irrelevant languages based on context 

155 - Use 'NOT' to exclude competing terms 

156 6. Ensure the output is a concise, space-separated string with no punctuation or extra text beyond keywords and filters. 

157 

158 

159 Original query: "{query}" 

160 

161 Return ONLY the optimized query, ready for GitHub's search API. Do not include explanations or additional text.""" 

162 

163 try: 

164 response = self.llm.invoke(prompt) 

165 

166 optimized_query = get_llm_response_text(response).strip() 

167 

168 # Validate the optimized query 

169 if optimized_query and len(optimized_query) > 0: 

170 logger.info( 

171 f"LLM optimized query from '{query}' to '{optimized_query}'" 

172 ) 

173 return optimized_query 

174 logger.warning("LLM returned empty query, using original") 

175 return query 

176 

177 except Exception as e: 

178 safe_msg = self._scrub_error(e) 

179 logger.warning(f"Error optimizing query with LLM: {safe_msg}") 

180 return query 

181 

182 def _search_github(self, query: str) -> List[Dict[str, Any]]: 

183 """ 

184 Perform a GitHub search based on the configured search type. 

185 

186 Args: 

187 query: The search query 

188 

189 Returns: 

190 List of GitHub search result items 

191 """ 

192 results = [] 

193 

194 try: 

195 # Optimize GitHub query using LLM 

196 github_query = self._optimize_github_query(query) 

197 

198 logger.info(f"Final GitHub query: {github_query}") 

199 

200 # Construct search parameters 

201 params = { 

202 "q": github_query, 

203 "per_page": min( 

204 self.max_results, 100 

205 ), # GitHub API max is 100 per page 

206 "page": 1, 

207 } 

208 

209 # Add sort parameters based on search type 

210 if self.search_type == "repositories": 

211 params["sort"] = "stars" 

212 params["order"] = "desc" 

213 elif self.search_type == "code": 

214 params["sort"] = "indexed" 

215 params["order"] = "desc" 

216 elif self.search_type == "issues": 

217 params["sort"] = "updated" 

218 params["order"] = "desc" 

219 elif self.search_type == "users": 219 ↛ 224line 219 didn't jump to line 224 because the condition on line 219 was always true

220 params["sort"] = "followers" 

221 params["order"] = "desc" 

222 

223 # Apply rate limiting before request 

224 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

225 self.engine_type 

226 ) 

227 

228 # Execute the API request 

229 response = safe_get( 

230 self.search_endpoint, headers=self.headers, params=params 

231 ) 

232 

233 # Check for rate limiting 

234 self._handle_rate_limits(response) 

235 

236 # Handle response with detailed logging 

237 if response.status_code == 200: 

238 data = response.json() 

239 total_count = data.get("total_count", 0) 

240 results = data.get("items", []) 

241 logger.info( 

242 f"GitHub search returned {len(results)} results (total available: {total_count})" 

243 ) 

244 

245 # Log the rate limit information 

246 rate_limit_remaining = response.headers.get( 

247 "X-RateLimit-Remaining", "unknown" 

248 ) 

249 logger.info( 

250 f"GitHub API rate limit: {rate_limit_remaining} requests remaining" 

251 ) 

252 

253 # If no results, try to provide more guidance 

254 if not results: 

255 logger.warning( 

256 "No results found. Consider these search tips:" 

257 ) 

258 logger.warning("1. Use shorter, more specific queries") 

259 logger.warning( 

260 "2. For repositories, try adding 'stars:>100' or 'language:python'" 

261 ) 

262 logger.warning( 

263 "3. For contribution opportunities, search for 'good-first-issue' or 'help-wanted'" 

264 ) 

265 else: 

266 logger.error( 

267 f"GitHub API error: {response.status_code} - {response.text}" 

268 ) 

269 

270 except Exception as e: 

271 safe_msg = self._scrub_error(e) 

272 logger.warning(f"Error searching GitHub: {safe_msg}") 

273 

274 return results 

275 

276 def _get_readme_content(self, repo_full_name: str) -> str: 

277 """ 

278 Get README content for a repository. 

279 

280 Args: 

281 repo_full_name: Full name of the repository (owner/repo) 

282 

283 Returns: 

284 Decoded README content or empty string if not found 

285 """ 

286 try: 

287 # Get README 

288 # Apply rate limiting before request 

289 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

290 self.engine_type 

291 ) 

292 

293 response = safe_get( 

294 f"{self.api_base}/repos/{repo_full_name}/readme", 

295 headers=self.headers, 

296 ) 

297 

298 # Check for rate limiting 

299 self._handle_rate_limits(response) 

300 

301 if response.status_code == 200: 

302 data = response.json() 

303 content: str = data.get("content", "") 

304 encoding = data.get("encoding", "") 

305 

306 if encoding == "base64" and content: 

307 return base64.b64decode(content).decode( 

308 "utf-8", errors="replace" 

309 ) 

310 return content 

311 logger.warning( 

312 f"Could not get README for {repo_full_name}: {response.status_code}" 

313 ) 

314 return "" 

315 

316 except Exception as e: 

317 safe_msg = self._scrub_error(e) 

318 logger.warning( 

319 f"Error getting README for {repo_full_name}: {safe_msg}" 

320 ) 

321 return "" 

322 

323 def _get_recent_issues( 

324 self, repo_full_name: str, limit: int = 5 

325 ) -> List[Dict[str, Any]]: 

326 """ 

327 Get recent issues for a repository. 

328 

329 Args: 

330 repo_full_name: Full name of the repository (owner/repo) 

331 limit: Maximum number of issues to return 

332 

333 Returns: 

334 List of recent issues 

335 """ 

336 issues = [] 

337 

338 try: 

339 # Get recent issues 

340 # Apply rate limiting before request 

341 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

342 self.engine_type 

343 ) 

344 

345 response = safe_get( 

346 f"{self.api_base}/repos/{repo_full_name}/issues", 

347 headers=self.headers, 

348 params={ 

349 "state": "all", 

350 "per_page": limit, 

351 "sort": "updated", 

352 "direction": "desc", 

353 }, 

354 ) 

355 

356 # Check for rate limiting 

357 self._handle_rate_limits(response) 

358 

359 if response.status_code == 200: 

360 issues = response.json() 

361 logger.info( 

362 f"Got {len(issues)} recent issues for {repo_full_name}" 

363 ) 

364 else: 

365 logger.warning( 

366 f"Could not get issues for {repo_full_name}: {response.status_code}" 

367 ) 

368 

369 except Exception as e: 

370 safe_msg = self._scrub_error(e) 

371 logger.warning( 

372 f"Error getting issues for {repo_full_name}: {safe_msg}" 

373 ) 

374 

375 return issues 

376 

377 def _get_file_content(self, file_url: str) -> str: 

378 """ 

379 Get content of a file from GitHub. 

380 

381 Args: 

382 file_url: API URL for the file 

383 

384 Returns: 

385 Decoded file content or empty string if not found 

386 """ 

387 try: 

388 # Apply rate limiting before request 

389 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

390 self.engine_type 

391 ) 

392 

393 # Get file content 

394 response = safe_get(file_url, headers=self.headers) 

395 

396 # Check for rate limiting 

397 self._handle_rate_limits(response) 

398 

399 if response.status_code == 200: 

400 data = response.json() 

401 content2: str = data.get("content", "") 

402 encoding = data.get("encoding", "") 

403 

404 if encoding == "base64" and content2: 

405 return base64.b64decode(content2).decode( 

406 "utf-8", errors="replace" 

407 ) 

408 return content2 

409 logger.warning( 

410 f"Could not get file content: {response.status_code}" 

411 ) 

412 return "" 

413 

414 except Exception as e: 

415 safe_msg = self._scrub_error(e) 

416 logger.warning(f"Error getting file content: {safe_msg}") 

417 return "" 

418 

419 def _format_repository_preview( 

420 self, repo: Dict[str, Any] 

421 ) -> Dict[str, Any]: 

422 """Format repository search result as preview""" 

423 return { 

424 "id": str(repo.get("id", "")), 

425 "title": repo.get("full_name", ""), 

426 "link": repo.get("html_url", ""), 

427 "snippet": repo.get("description", "No description provided"), 

428 "stars": repo.get("stargazers_count", 0), 

429 "forks": repo.get("forks_count", 0), 

430 "language": repo.get("language", ""), 

431 "updated_at": repo.get("updated_at", ""), 

432 "created_at": repo.get("created_at", ""), 

433 "topics": repo.get("topics", []), 

434 "owner": repo.get("owner", {}).get("login", ""), 

435 "is_fork": repo.get("fork", False), 

436 "search_type": "repository", 

437 "repo_full_name": repo.get("full_name", ""), 

438 } 

439 

440 def _format_code_preview(self, code: Dict[str, Any]) -> Dict[str, Any]: 

441 """Format code search result as preview""" 

442 repo = code.get("repository", {}) 

443 return { 

444 "id": f"code_{code.get('sha', '')}", 

445 "title": f"{code.get('name', '')} in {repo.get('full_name', '')}", 

446 "link": code.get("html_url", ""), 

447 "snippet": f"Match in {code.get('path', '')}", 

448 "path": code.get("path", ""), 

449 "repo_name": repo.get("full_name", ""), 

450 "repo_url": repo.get("html_url", ""), 

451 "search_type": "code", 

452 "file_url": code.get("url", ""), 

453 } 

454 

455 def _format_issue_preview(self, issue: Dict[str, Any]) -> Dict[str, Any]: 

456 """Format issue search result as preview""" 

457 repo = ( 

458 issue.get("repository", {}) 

459 if "repository" in issue 

460 else {"full_name": ""} 

461 ) 

462 return { 

463 "id": f"issue_{issue.get('number', '')}", 

464 "title": issue.get("title", ""), 

465 "link": issue.get("html_url", ""), 

466 "snippet": ( 

467 issue.get("body", "")[:200] + "..." 

468 if len(issue.get("body", "")) > 200 

469 else issue.get("body", "") 

470 ), 

471 "state": issue.get("state", ""), 

472 "created_at": issue.get("created_at", ""), 

473 "updated_at": issue.get("updated_at", ""), 

474 "user": issue.get("user", {}).get("login", ""), 

475 "comments": issue.get("comments", 0), 

476 "search_type": "issue", 

477 "repo_name": repo.get("full_name", ""), 

478 } 

479 

480 def _format_user_preview(self, user: Dict[str, Any]) -> Dict[str, Any]: 

481 """Format user search result as preview""" 

482 return { 

483 "id": f"user_{user.get('id', '')}", 

484 "title": user.get("login", ""), 

485 "link": user.get("html_url", ""), 

486 "snippet": user.get("bio", "No bio provided"), 

487 "name": user.get("name", ""), 

488 "followers": user.get("followers", 0), 

489 "public_repos": user.get("public_repos", 0), 

490 "location": user.get("location", ""), 

491 "search_type": "user", 

492 "user_login": user.get("login", ""), 

493 } 

494 

495 def _get_previews(self, query: str) -> List[Dict[str, Any]]: 

496 """ 

497 Get preview information for GitHub search results. 

498 

499 Args: 

500 query: The search query 

501 

502 Returns: 

503 List of preview dictionaries 

504 """ 

505 logger.info(f"Getting GitHub previews for query: {query}") 

506 

507 # For contribution-focused queries, automatically adjust search type and add filters 

508 if any( 

509 term in query.lower() 

510 for term in [ 

511 "contribute", 

512 "contributing", 

513 "contribution", 

514 "beginner", 

515 "newcomer", 

516 ] 

517 ): 

518 # Use repositories search with help-wanted or good-first-issue labels 

519 original_search_type = self.search_type 

520 self.search_type = "repositories" 

521 self.search_endpoint = f"{self.api_base}/search/repositories" 

522 

523 # Create a specialized query for finding beginner-friendly projects 

524 specialized_query = "good-first-issues:>5 is:public archived:false" 

525 

526 # Extract language preferences if present 

527 languages = [] 

528 for lang in [ 

529 "python", 

530 "javascript", 

531 "java", 

532 "rust", 

533 "go", 

534 "typescript", 

535 "c#", 

536 "c++", 

537 "ruby", 

538 ]: 

539 if lang in query.lower(): 

540 languages.append(lang) 

541 

542 if languages: 

543 specialized_query += f" language:{' language:'.join(languages)}" 

544 

545 # Extract keywords 

546 keywords = [ 

547 word 

548 for word in query.split() 

549 if len(word) > 3 

550 and word.lower() 

551 not in [ 

552 "recommend", 

553 "recommended", 

554 "github", 

555 "repositories", 

556 "looking", 

557 "developers", 

558 "contribute", 

559 "contributing", 

560 "beginner", 

561 "newcomer", 

562 ] 

563 ] 

564 

565 if keywords: 565 ↛ 570line 565 didn't jump to line 570 because the condition on line 565 was always true

566 specialized_query += " " + " ".join( 

567 keywords[:5] 

568 ) # Add up to 5 keywords 

569 

570 logger.info( 

571 f"Using specialized contribution query: {specialized_query}" 

572 ) 

573 

574 # Perform GitHub search with specialized query 

575 results = self._search_github(specialized_query) 

576 

577 # Restore original search type 

578 self.search_type = original_search_type 

579 self.search_endpoint = f"{self.api_base}/search/{self.search_type}" 

580 else: 

581 # Perform standard GitHub search 

582 results = self._search_github(query) 

583 

584 if not results: 

585 logger.warning(f"No GitHub results found for query: {query}") 

586 return [] 

587 

588 # Format results as previews 

589 previews = [] 

590 for result in results: 

591 # Format based on search type 

592 if self.search_type == "repositories": 

593 preview = self._format_repository_preview(result) 

594 elif self.search_type == "code": 

595 preview = self._format_code_preview(result) 

596 elif self.search_type == "issues": 

597 preview = self._format_issue_preview(result) 

598 elif self.search_type == "users": 598 ↛ 601line 598 didn't jump to line 601 because the condition on line 598 was always true

599 preview = self._format_user_preview(result) 

600 else: 

601 logger.warning(f"Unknown search type: {self.search_type}") 

602 continue 

603 

604 previews.append(preview) 

605 

606 logger.info(f"Formatted {len(previews)} GitHub preview results") 

607 return previews 

608 

609 def _get_full_content( 

610 self, relevant_items: List[Dict[str, Any]] 

611 ) -> List[Dict[str, Any]]: 

612 """ 

613 Get full content for the relevant GitHub search results. 

614 

615 Args: 

616 relevant_items: List of relevant preview dictionaries 

617 

618 Returns: 

619 List of result dictionaries with full content 

620 """ 

621 logger.info( 

622 f"Getting full content for {len(relevant_items)} GitHub results" 

623 ) 

624 

625 results = [] 

626 for item in relevant_items: 

627 result = item.copy() 

628 search_type = item.get("search_type", "") 

629 

630 # Add content based on search type 

631 if search_type == "repository" and self.include_readme: 

632 repo_full_name = item.get("repo_full_name", "") 

633 if repo_full_name: 

634 # Get README content 

635 readme_content = self._get_readme_content(repo_full_name) 

636 result["full_content"] = readme_content 

637 result["content_type"] = "readme" 

638 

639 # Get recent issues if requested 

640 if self.include_issues: 

641 issues = self._get_recent_issues(repo_full_name) 

642 result["recent_issues"] = issues 

643 

644 elif search_type == "code": 

645 file_url = item.get("file_url", "") 

646 if file_url: 

647 # Get file content 

648 file_content = self._get_file_content(file_url) 

649 result["full_content"] = file_content 

650 result["content_type"] = "file" 

651 

652 elif search_type == "issue": 

653 # For issues, the snippet usually contains a summary already 

654 # We'll just keep it as is 

655 result["full_content"] = item.get("snippet", "") 

656 result["content_type"] = "issue" 

657 

658 elif search_type == "user": 

659 # For users, construct a profile summary 

660 profile_summary = f"GitHub user: {item.get('title', '')}\n" 

661 

662 if item.get("name"): 

663 profile_summary += f"Name: {item.get('name')}\n" 

664 

665 if item.get("location"): 

666 profile_summary += f"Location: {item.get('location')}\n" 

667 

668 profile_summary += f"Followers: {item.get('followers', 0)}\n" 

669 profile_summary += ( 

670 f"Public repositories: {item.get('public_repos', 0)}\n" 

671 ) 

672 

673 if ( 

674 item.get("snippet") 

675 and item.get("snippet") != "No bio provided" 

676 ): 

677 profile_summary += f"\nBio: {item.get('snippet')}\n" 

678 

679 result["full_content"] = profile_summary 

680 result["content_type"] = "user_profile" 

681 

682 results.append(result) 

683 

684 return results 

685 

686 def search_repository( 

687 self, repo_owner: str, repo_name: str 

688 ) -> Dict[str, Any]: 

689 """ 

690 Get detailed information about a specific repository. 

691 

692 Args: 

693 repo_owner: Owner of the repository 

694 repo_name: Name of the repository 

695 

696 Returns: 

697 Dictionary with repository information 

698 """ 

699 repo_full_name = f"{repo_owner}/{repo_name}" 

700 logger.info(f"Getting details for repository: {repo_full_name}") 

701 

702 try: 

703 # Get repository details 

704 # Apply rate limiting before request 

705 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

706 self.engine_type 

707 ) 

708 

709 response = safe_get( 

710 f"{self.api_base}/repos/{repo_full_name}", headers=self.headers 

711 ) 

712 

713 # Check for rate limiting 

714 self._handle_rate_limits(response) 

715 

716 if response.status_code == 200: 

717 repo = response.json() 

718 

719 # Format as repository preview 

720 result = self._format_repository_preview(repo) 

721 

722 # Add README content if requested 

723 if self.include_readme: 

724 readme_content = self._get_readme_content(repo_full_name) 

725 result["full_content"] = readme_content 

726 result["content_type"] = "readme" 

727 

728 # Add recent issues if requested 

729 if self.include_issues: 

730 issues = self._get_recent_issues(repo_full_name) 

731 result["recent_issues"] = issues 

732 

733 return result 

734 logger.error( 

735 f"Error getting repository details: {response.status_code} - {response.text}" 

736 ) 

737 return {} 

738 

739 except Exception as e: 

740 safe_msg = self._scrub_error(e) 

741 logger.warning(f"Error getting repository details: {safe_msg}") 

742 return {} 

743 

744 def search_code( 

745 self, 

746 query: str, 

747 language: Optional[str] = None, 

748 user: Optional[str] = None, 

749 ) -> List[Dict[str, Any]]: 

750 """ 

751 Search for code with more specific parameters. 

752 

753 Args: 

754 query: Code search query 

755 language: Filter by programming language 

756 user: Filter by GitHub username/organization 

757 

758 Returns: 

759 List of code search results 

760 """ 

761 # Build advanced query 

762 advanced_query = query 

763 

764 if language: 

765 advanced_query += f" language:{language}" 

766 

767 if user: 

768 advanced_query += f" user:{user}" 

769 

770 # Save current search type 

771 original_search_type = self.search_type 

772 

773 try: 

774 # Set search type to code 

775 self.search_type = "code" 

776 self.search_endpoint = f"{self.api_base}/search/code" 

777 

778 # Perform search 

779 results = self._search_github(advanced_query) 

780 

781 # Format results 

782 previews = [self._format_code_preview(result) for result in results] 

783 

784 return self._get_full_content(previews) 

785 

786 finally: 

787 # Restore original search type 

788 self.search_type = original_search_type 

789 self.search_endpoint = f"{self.api_base}/search/{self.search_type}" 

790 

791 def search_issues( 

792 self, query: str, state: str = "open", sort: str = "updated" 

793 ) -> List[Dict[str, Any]]: 

794 """ 

795 Search for issues with more specific parameters. 

796 

797 Args: 

798 query: Issue search query 

799 state: Filter by issue state ("open", "closed", "all") 

800 sort: Sort order ("updated", "created", "comments") 

801 

802 Returns: 

803 List of issue search results 

804 """ 

805 # Build advanced query 

806 advanced_query = query + f" state:{state}" 

807 

808 # Save current search type 

809 original_search_type = self.search_type 

810 

811 try: 

812 # Set search type to issues 

813 self.search_type = "issues" 

814 self.search_endpoint = f"{self.api_base}/search/issues" 

815 

816 # Set sort parameter 

817 params = { 

818 "q": advanced_query, 

819 "per_page": min(self.max_results, 100), 

820 "page": 1, 

821 "sort": sort, 

822 "order": "desc", 

823 } 

824 

825 # Perform search 

826 response = safe_get( 

827 self.search_endpoint, headers=self.headers, params=params 

828 ) 

829 

830 # Check for rate limiting 

831 self._handle_rate_limits(response) 

832 

833 if response.status_code == 200: 

834 data = response.json() 

835 results = data.get("items", []) 

836 

837 # Format results 

838 return [ 

839 self._format_issue_preview(result) for result in results 

840 ] 

841 

842 # For issues, we don't need to get full content 

843 logger.error( 

844 f"GitHub API error: {response.status_code} - {response.text}" 

845 ) 

846 return [] 

847 

848 finally: 

849 # Restore original search type 

850 self.search_type = original_search_type 

851 self.search_endpoint = f"{self.api_base}/search/{self.search_type}" 

852 

853 def set_search_type(self, search_type: str): 

854 """ 

855 Set the search type for subsequent searches. 

856 

857 Args: 

858 search_type: Type of GitHub search ("repositories", "code", "issues", "users") 

859 """ 

860 if search_type not in _VALID_SEARCH_TYPES: 

861 raise ValueError( 

862 f"Invalid GitHub search_type: {search_type!r}. " 

863 f"Must be one of {_VALID_SEARCH_TYPES}" 

864 ) 

865 self.search_type = search_type 

866 self.search_endpoint = f"{self.api_base}/search/{search_type}" 

867 logger.info(f"Set GitHub search type to: {search_type}")