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 # Handle different response formats (string or object with content attribute) 

167 if hasattr(response, "content"): 

168 optimized_query = str(response.content).strip() 

169 else: 

170 # Handle string responses 

171 optimized_query = str(response).strip() 

172 

173 # Validate the optimized query 

174 if optimized_query and len(optimized_query) > 0: 

175 logger.info( 

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

177 ) 

178 return optimized_query 

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

180 return query 

181 

182 except Exception as e: 

183 safe_msg = self._scrub_error(e) 

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

185 return query 

186 

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

188 """ 

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

190 

191 Args: 

192 query: The search query 

193 

194 Returns: 

195 List of GitHub search result items 

196 """ 

197 results = [] 

198 

199 try: 

200 # Optimize GitHub query using LLM 

201 github_query = self._optimize_github_query(query) 

202 

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

204 

205 # Construct search parameters 

206 params = { 

207 "q": github_query, 

208 "per_page": min( 

209 self.max_results, 100 

210 ), # GitHub API max is 100 per page 

211 "page": 1, 

212 } 

213 

214 # Add sort parameters based on search type 

215 if self.search_type == "repositories": 

216 params["sort"] = "stars" 

217 params["order"] = "desc" 

218 elif self.search_type == "code": 

219 params["sort"] = "indexed" 

220 params["order"] = "desc" 

221 elif self.search_type == "issues": 

222 params["sort"] = "updated" 

223 params["order"] = "desc" 

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

225 params["sort"] = "followers" 

226 params["order"] = "desc" 

227 

228 # Apply rate limiting before request 

229 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

230 self.engine_type 

231 ) 

232 

233 # Execute the API request 

234 response = safe_get( 

235 self.search_endpoint, headers=self.headers, params=params 

236 ) 

237 

238 # Check for rate limiting 

239 self._handle_rate_limits(response) 

240 

241 # Handle response with detailed logging 

242 if response.status_code == 200: 

243 data = response.json() 

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

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

246 logger.info( 

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

248 ) 

249 

250 # Log the rate limit information 

251 rate_limit_remaining = response.headers.get( 

252 "X-RateLimit-Remaining", "unknown" 

253 ) 

254 logger.info( 

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

256 ) 

257 

258 # If no results, try to provide more guidance 

259 if not results: 

260 logger.warning( 

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

262 ) 

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

264 logger.warning( 

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

266 ) 

267 logger.warning( 

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

269 ) 

270 else: 

271 logger.error( 

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

273 ) 

274 

275 except Exception as e: 

276 safe_msg = self._scrub_error(e) 

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

278 

279 return results 

280 

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

282 """ 

283 Get README content for a repository. 

284 

285 Args: 

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

287 

288 Returns: 

289 Decoded README content or empty string if not found 

290 """ 

291 try: 

292 # Get README 

293 # Apply rate limiting before request 

294 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

295 self.engine_type 

296 ) 

297 

298 response = safe_get( 

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

300 headers=self.headers, 

301 ) 

302 

303 # Check for rate limiting 

304 self._handle_rate_limits(response) 

305 

306 if response.status_code == 200: 

307 data = response.json() 

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

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

310 

311 if encoding == "base64" and content: 

312 return base64.b64decode(content).decode( 

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

314 ) 

315 return content 

316 logger.warning( 

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

318 ) 

319 return "" 

320 

321 except Exception as e: 

322 safe_msg = self._scrub_error(e) 

323 logger.warning( 

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

325 ) 

326 return "" 

327 

328 def _get_recent_issues( 

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

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

331 """ 

332 Get recent issues for a repository. 

333 

334 Args: 

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

336 limit: Maximum number of issues to return 

337 

338 Returns: 

339 List of recent issues 

340 """ 

341 issues = [] 

342 

343 try: 

344 # Get recent issues 

345 # Apply rate limiting before request 

346 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

347 self.engine_type 

348 ) 

349 

350 response = safe_get( 

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

352 headers=self.headers, 

353 params={ 

354 "state": "all", 

355 "per_page": limit, 

356 "sort": "updated", 

357 "direction": "desc", 

358 }, 

359 ) 

360 

361 # Check for rate limiting 

362 self._handle_rate_limits(response) 

363 

364 if response.status_code == 200: 

365 issues = response.json() 

366 logger.info( 

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

368 ) 

369 else: 

370 logger.warning( 

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

372 ) 

373 

374 except Exception as e: 

375 safe_msg = self._scrub_error(e) 

376 logger.warning( 

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

378 ) 

379 

380 return issues 

381 

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

383 """ 

384 Get content of a file from GitHub. 

385 

386 Args: 

387 file_url: API URL for the file 

388 

389 Returns: 

390 Decoded file content or empty string if not found 

391 """ 

392 try: 

393 # Apply rate limiting before request 

394 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

395 self.engine_type 

396 ) 

397 

398 # Get file content 

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

400 

401 # Check for rate limiting 

402 self._handle_rate_limits(response) 

403 

404 if response.status_code == 200: 

405 data = response.json() 

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

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

408 

409 if encoding == "base64" and content2: 

410 return base64.b64decode(content2).decode( 

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

412 ) 

413 return content2 

414 logger.warning( 

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

416 ) 

417 return "" 

418 

419 except Exception as e: 

420 safe_msg = self._scrub_error(e) 

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

422 return "" 

423 

424 def _format_repository_preview( 

425 self, repo: Dict[str, Any] 

426 ) -> Dict[str, Any]: 

427 """Format repository search result as preview""" 

428 return { 

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

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

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

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

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

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

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

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

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

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

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

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

441 "search_type": "repository", 

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

443 } 

444 

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

446 """Format code search result as preview""" 

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

448 return { 

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

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

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

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

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

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

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

456 "search_type": "code", 

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

458 } 

459 

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

461 """Format issue search result as preview""" 

462 repo = ( 

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

464 if "repository" in issue 

465 else {"full_name": ""} 

466 ) 

467 return { 

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

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

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

471 "snippet": ( 

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

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

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

475 ), 

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

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

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

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

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

481 "search_type": "issue", 

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

483 } 

484 

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

486 """Format user search result as preview""" 

487 return { 

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

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

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

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

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

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

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

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

496 "search_type": "user", 

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

498 } 

499 

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

501 """ 

502 Get preview information for GitHub search results. 

503 

504 Args: 

505 query: The search query 

506 

507 Returns: 

508 List of preview dictionaries 

509 """ 

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

511 

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

513 if any( 

514 term in query.lower() 

515 for term in [ 

516 "contribute", 

517 "contributing", 

518 "contribution", 

519 "beginner", 

520 "newcomer", 

521 ] 

522 ): 

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

524 original_search_type = self.search_type 

525 self.search_type = "repositories" 

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

527 

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

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

530 

531 # Extract language preferences if present 

532 languages = [] 

533 for lang in [ 

534 "python", 

535 "javascript", 

536 "java", 

537 "rust", 

538 "go", 

539 "typescript", 

540 "c#", 

541 "c++", 

542 "ruby", 

543 ]: 

544 if lang in query.lower(): 

545 languages.append(lang) 

546 

547 if languages: 

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

549 

550 # Extract keywords 

551 keywords = [ 

552 word 

553 for word in query.split() 

554 if len(word) > 3 

555 and word.lower() 

556 not in [ 

557 "recommend", 

558 "recommended", 

559 "github", 

560 "repositories", 

561 "looking", 

562 "developers", 

563 "contribute", 

564 "contributing", 

565 "beginner", 

566 "newcomer", 

567 ] 

568 ] 

569 

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

571 specialized_query += " " + " ".join( 

572 keywords[:5] 

573 ) # Add up to 5 keywords 

574 

575 logger.info( 

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

577 ) 

578 

579 # Perform GitHub search with specialized query 

580 results = self._search_github(specialized_query) 

581 

582 # Restore original search type 

583 self.search_type = original_search_type 

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

585 else: 

586 # Perform standard GitHub search 

587 results = self._search_github(query) 

588 

589 if not results: 

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

591 return [] 

592 

593 # Format results as previews 

594 previews = [] 

595 for result in results: 

596 # Format based on search type 

597 if self.search_type == "repositories": 

598 preview = self._format_repository_preview(result) 

599 elif self.search_type == "code": 

600 preview = self._format_code_preview(result) 

601 elif self.search_type == "issues": 

602 preview = self._format_issue_preview(result) 

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

604 preview = self._format_user_preview(result) 

605 else: 

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

607 continue 

608 

609 previews.append(preview) 

610 

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

612 return previews 

613 

614 def _get_full_content( 

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

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

617 """ 

618 Get full content for the relevant GitHub search results. 

619 

620 Args: 

621 relevant_items: List of relevant preview dictionaries 

622 

623 Returns: 

624 List of result dictionaries with full content 

625 """ 

626 logger.info( 

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

628 ) 

629 

630 results = [] 

631 for item in relevant_items: 

632 result = item.copy() 

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

634 

635 # Add content based on search type 

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

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

638 if repo_full_name: 

639 # Get README content 

640 readme_content = self._get_readme_content(repo_full_name) 

641 result["full_content"] = readme_content 

642 result["content_type"] = "readme" 

643 

644 # Get recent issues if requested 

645 if self.include_issues: 

646 issues = self._get_recent_issues(repo_full_name) 

647 result["recent_issues"] = issues 

648 

649 elif search_type == "code": 

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

651 if file_url: 

652 # Get file content 

653 file_content = self._get_file_content(file_url) 

654 result["full_content"] = file_content 

655 result["content_type"] = "file" 

656 

657 elif search_type == "issue": 

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

659 # We'll just keep it as is 

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

661 result["content_type"] = "issue" 

662 

663 elif search_type == "user": 

664 # For users, construct a profile summary 

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

666 

667 if item.get("name"): 

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

669 

670 if item.get("location"): 

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

672 

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

674 profile_summary += ( 

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

676 ) 

677 

678 if ( 

679 item.get("snippet") 

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

681 ): 

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

683 

684 result["full_content"] = profile_summary 

685 result["content_type"] = "user_profile" 

686 

687 results.append(result) 

688 

689 return results 

690 

691 def search_repository( 

692 self, repo_owner: str, repo_name: str 

693 ) -> Dict[str, Any]: 

694 """ 

695 Get detailed information about a specific repository. 

696 

697 Args: 

698 repo_owner: Owner of the repository 

699 repo_name: Name of the repository 

700 

701 Returns: 

702 Dictionary with repository information 

703 """ 

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

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

706 

707 try: 

708 # Get repository details 

709 # Apply rate limiting before request 

710 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

711 self.engine_type 

712 ) 

713 

714 response = safe_get( 

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

716 ) 

717 

718 # Check for rate limiting 

719 self._handle_rate_limits(response) 

720 

721 if response.status_code == 200: 

722 repo = response.json() 

723 

724 # Format as repository preview 

725 result = self._format_repository_preview(repo) 

726 

727 # Add README content if requested 

728 if self.include_readme: 

729 readme_content = self._get_readme_content(repo_full_name) 

730 result["full_content"] = readme_content 

731 result["content_type"] = "readme" 

732 

733 # Add recent issues if requested 

734 if self.include_issues: 

735 issues = self._get_recent_issues(repo_full_name) 

736 result["recent_issues"] = issues 

737 

738 return result 

739 logger.error( 

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

741 ) 

742 return {} 

743 

744 except Exception as e: 

745 safe_msg = self._scrub_error(e) 

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

747 return {} 

748 

749 def search_code( 

750 self, 

751 query: str, 

752 language: Optional[str] = None, 

753 user: Optional[str] = None, 

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

755 """ 

756 Search for code with more specific parameters. 

757 

758 Args: 

759 query: Code search query 

760 language: Filter by programming language 

761 user: Filter by GitHub username/organization 

762 

763 Returns: 

764 List of code search results 

765 """ 

766 # Build advanced query 

767 advanced_query = query 

768 

769 if language: 

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

771 

772 if user: 

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

774 

775 # Save current search type 

776 original_search_type = self.search_type 

777 

778 try: 

779 # Set search type to code 

780 self.search_type = "code" 

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

782 

783 # Perform search 

784 results = self._search_github(advanced_query) 

785 

786 # Format results 

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

788 

789 return self._get_full_content(previews) 

790 

791 finally: 

792 # Restore original search type 

793 self.search_type = original_search_type 

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

795 

796 def search_issues( 

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

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

799 """ 

800 Search for issues with more specific parameters. 

801 

802 Args: 

803 query: Issue search query 

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

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

806 

807 Returns: 

808 List of issue search results 

809 """ 

810 # Build advanced query 

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

812 

813 # Save current search type 

814 original_search_type = self.search_type 

815 

816 try: 

817 # Set search type to issues 

818 self.search_type = "issues" 

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

820 

821 # Set sort parameter 

822 params = { 

823 "q": advanced_query, 

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

825 "page": 1, 

826 "sort": sort, 

827 "order": "desc", 

828 } 

829 

830 # Perform search 

831 response = safe_get( 

832 self.search_endpoint, headers=self.headers, params=params 

833 ) 

834 

835 # Check for rate limiting 

836 self._handle_rate_limits(response) 

837 

838 if response.status_code == 200: 

839 data = response.json() 

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

841 

842 # Format results 

843 return [ 

844 self._format_issue_preview(result) for result in results 

845 ] 

846 

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

848 logger.error( 

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

850 ) 

851 return [] 

852 

853 finally: 

854 # Restore original search type 

855 self.search_type = original_search_type 

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

857 

858 def set_search_type(self, search_type: str): 

859 """ 

860 Set the search type for subsequent searches. 

861 

862 Args: 

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

864 """ 

865 if search_type not in _VALID_SEARCH_TYPES: 

866 raise ValueError( 

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

868 f"Must be one of {_VALID_SEARCH_TYPES}"