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 

14from ...utilities.llm_utils import invoke_llm_sync 

15 

16 

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

18 

19 

20class GitHubSearchEngine(BaseSearchEngine): 

21 """ 

22 GitHub search engine implementation. 

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

24 """ 

25 

26 is_public = True 

27 egress_sensitivity = Sensitivity.NON_SENSITIVE 

28 egress_exposure = Exposure.EXPOSING 

29 is_code = True 

30 is_lexical = True 

31 needs_llm_relevance_filter = True 

32 

33 def __init__( 

34 self, 

35 max_results: int = 15, 

36 api_key: Optional[str] = None, 

37 search_type: str = "repositories", 

38 include_readme: bool = True, 

39 include_issues: bool = False, 

40 llm: Optional[BaseLLM] = None, 

41 max_filtered_results: Optional[int] = None, 

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

43 ): 

44 """ 

45 Initialize the GitHub search engine. 

46 

47 Args: 

48 max_results: Maximum number of search results 

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

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

51 include_readme: Whether to include README content for repositories 

52 include_issues: Whether to include recent issues for repositories 

53 llm: Language model for relevance filtering 

54 max_filtered_results: Maximum number of results to keep after filtering 

55 """ 

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

57 super().__init__( 

58 llm=llm, 

59 max_filtered_results=max_filtered_results, 

60 max_results=max_results, 

61 settings_snapshot=settings_snapshot, 

62 ) 

63 self.api_key = api_key 

64 if search_type not in _VALID_SEARCH_TYPES: 

65 raise ValueError( 

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

67 f"Must be one of {_VALID_SEARCH_TYPES}" 

68 ) 

69 self.search_type = search_type 

70 self.include_readme = include_readme 

71 self.include_issues = include_issues 

72 

73 self._owns_llm = False 

74 

75 # API endpoints 

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

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

78 

79 # Set up API headers 

80 self.headers = { 

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

82 "User-Agent": USER_AGENT, 

83 } 

84 

85 # Add authentication if API key provided 

86 if self.api_key: 

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

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

89 else: 

90 logger.warning( 

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

92 ) 

93 

94 def close(self) -> None: 

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

96 from ...utilities.resource_utils import safe_close 

97 

98 if self._owns_llm: 

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

100 super().close() 

101 

102 def _handle_rate_limits(self, response): 

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

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

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

106 

107 if remaining < 5: 

108 current_time = time.time() 

109 wait_time = max(reset_time - current_time, 0) 

110 logger.warning( 

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

112 ) 

113 

114 if wait_time > 0 and remaining == 0: 

115 logger.warning( 

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

117 ) 

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

119 

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

121 """ 

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

123 

124 Args: 

125 query: Original search query 

126 

127 Returns: 

128 Optimized GitHub search query 

129 """ 

130 # Get LLM from config if not already set 

131 if not self.llm: 

132 try: 

133 self.llm = llm_config.get_llm() 

134 self._owns_llm = True 

135 if not self.llm: 

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

137 return query 

138 except Exception as e: 

139 safe_msg = self._scrub_error(e) 

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

141 return query 

142 

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

144 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. 

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

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

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

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

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

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

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

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

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

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

155 - Exclude irrelevant languages based on context 

156 - Use 'NOT' to exclude competing terms 

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

158 

159 

160 Original query: "{query}" 

161 

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

163 

164 try: 

165 response = invoke_llm_sync(self.llm, prompt) 

166 

167 optimized_query = get_llm_response_text(response).strip() 

168 

169 # Validate the optimized query 

170 if optimized_query and len(optimized_query) > 0: 

171 logger.info( 

172 f"LLM optimized GitHub query " 

173 f"({len(query)} -> {len(optimized_query)} chars)" 

174 ) 

175 return optimized_query 

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

177 return query 

178 

179 except Exception as e: 

180 safe_msg = self._scrub_error(e) 

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

182 return query 

183 

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

185 """ 

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

187 

188 Args: 

189 query: The search query 

190 

191 Returns: 

192 List of GitHub search result items 

193 """ 

194 results = [] 

195 

196 try: 

197 # Optimize GitHub query using LLM 

198 github_query = self._optimize_github_query(query) 

199 

200 logger.info(f"Final GitHub query built ({len(github_query)} chars)") 

201 

202 # Construct search parameters 

203 params = { 

204 "q": github_query, 

205 "per_page": min( 

206 self.max_results, 100 

207 ), # GitHub API max is 100 per page 

208 "page": 1, 

209 } 

210 

211 # Add sort parameters based on search type 

212 if self.search_type == "repositories": 

213 params["sort"] = "stars" 

214 params["order"] = "desc" 

215 elif self.search_type == "code": 

216 params["sort"] = "indexed" 

217 params["order"] = "desc" 

218 elif self.search_type == "issues": 

219 params["sort"] = "updated" 

220 params["order"] = "desc" 

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

222 params["sort"] = "followers" 

223 params["order"] = "desc" 

224 

225 # Apply rate limiting before request 

226 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

227 self.engine_type 

228 ) 

229 

230 # Execute the API request 

231 response = safe_get( 

232 self.search_endpoint, headers=self.headers, params=params 

233 ) 

234 

235 # Check for rate limiting 

236 self._handle_rate_limits(response) 

237 

238 # Handle response with detailed logging 

239 if response.status_code == 200: 

240 data = response.json() 

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

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

243 logger.info( 

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

245 ) 

246 

247 # Log the rate limit information 

248 rate_limit_remaining = response.headers.get( 

249 "X-RateLimit-Remaining", "unknown" 

250 ) 

251 logger.info( 

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

253 ) 

254 

255 # If no results, try to provide more guidance 

256 if not results: 

257 logger.warning( 

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

259 ) 

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

261 logger.warning( 

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

263 ) 

264 logger.warning( 

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

266 ) 

267 else: 

268 logger.error( 

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

270 ) 

271 

272 except Exception as e: 

273 safe_msg = self._scrub_error(e) 

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

275 

276 return results 

277 

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

279 """ 

280 Get README content for a repository. 

281 

282 Args: 

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

284 

285 Returns: 

286 Decoded README content or empty string if not found 

287 """ 

288 try: 

289 # Get README 

290 # Apply rate limiting before request 

291 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

292 self.engine_type 

293 ) 

294 

295 response = safe_get( 

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

297 headers=self.headers, 

298 ) 

299 

300 # Check for rate limiting 

301 self._handle_rate_limits(response) 

302 

303 if response.status_code == 200: 

304 data = response.json() 

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

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

307 

308 if encoding == "base64" and content: 

309 return base64.b64decode(content).decode( 

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

311 ) 

312 return content 

313 logger.warning( 

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

315 ) 

316 return "" 

317 

318 except Exception as e: 

319 safe_msg = self._scrub_error(e) 

320 logger.warning( 

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

322 ) 

323 return "" 

324 

325 def _get_recent_issues( 

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

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

328 """ 

329 Get recent issues for a repository. 

330 

331 Args: 

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

333 limit: Maximum number of issues to return 

334 

335 Returns: 

336 List of recent issues 

337 """ 

338 issues = [] 

339 

340 try: 

341 # Get recent issues 

342 # Apply rate limiting before request 

343 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

344 self.engine_type 

345 ) 

346 

347 response = safe_get( 

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

349 headers=self.headers, 

350 params={ 

351 "state": "all", 

352 "per_page": limit, 

353 "sort": "updated", 

354 "direction": "desc", 

355 }, 

356 ) 

357 

358 # Check for rate limiting 

359 self._handle_rate_limits(response) 

360 

361 if response.status_code == 200: 

362 issues = response.json() 

363 logger.info( 

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

365 ) 

366 else: 

367 logger.warning( 

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

369 ) 

370 

371 except Exception as e: 

372 safe_msg = self._scrub_error(e) 

373 logger.warning( 

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

375 ) 

376 

377 return issues 

378 

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

380 """ 

381 Get content of a file from GitHub. 

382 

383 Args: 

384 file_url: API URL for the file 

385 

386 Returns: 

387 Decoded file content or empty string if not found 

388 """ 

389 try: 

390 # Apply rate limiting before request 

391 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

392 self.engine_type 

393 ) 

394 

395 # Get file content 

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

397 

398 # Check for rate limiting 

399 self._handle_rate_limits(response) 

400 

401 if response.status_code == 200: 

402 data = response.json() 

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

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

405 

406 if encoding == "base64" and content2: 

407 return base64.b64decode(content2).decode( 

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

409 ) 

410 return content2 

411 logger.warning( 

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

413 ) 

414 return "" 

415 

416 except Exception as e: 

417 safe_msg = self._scrub_error(e) 

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

419 return "" 

420 

421 def _format_repository_preview( 

422 self, repo: Dict[str, Any] 

423 ) -> Dict[str, Any]: 

424 """Format repository search result as preview""" 

425 return { 

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

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

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

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

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

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

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

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

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

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

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

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

438 "search_type": "repository", 

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

440 } 

441 

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

443 """Format code search result as preview""" 

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

445 return { 

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

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

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

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

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

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

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

453 "search_type": "code", 

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

455 } 

456 

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

458 """Format issue search result as preview""" 

459 repo = ( 

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

461 if "repository" in issue 

462 else {"full_name": ""} 

463 ) 

464 return { 

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

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

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

468 "snippet": ( 

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

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

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

472 ), 

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

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

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

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

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

478 "search_type": "issue", 

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

480 } 

481 

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

483 """Format user search result as preview""" 

484 return { 

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

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

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

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

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

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

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

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

493 "search_type": "user", 

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

495 } 

496 

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

498 """ 

499 Get preview information for GitHub search results. 

500 

501 Args: 

502 query: The search query 

503 

504 Returns: 

505 List of preview dictionaries 

506 """ 

507 logger.info("Getting GitHub previews") 

508 

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

510 if any( 

511 term in query.lower() 

512 for term in [ 

513 "contribute", 

514 "contributing", 

515 "contribution", 

516 "beginner", 

517 "newcomer", 

518 ] 

519 ): 

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

521 original_search_type = self.search_type 

522 self.search_type = "repositories" 

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

524 

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

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

527 

528 # Extract language preferences if present 

529 languages = [] 

530 for lang in [ 

531 "python", 

532 "javascript", 

533 "java", 

534 "rust", 

535 "go", 

536 "typescript", 

537 "c#", 

538 "c++", 

539 "ruby", 

540 ]: 

541 if lang in query.lower(): 

542 languages.append(lang) 

543 

544 if languages: 

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

546 

547 # Extract keywords 

548 keywords = [ 

549 word 

550 for word in query.split() 

551 if len(word) > 3 

552 and word.lower() 

553 not in [ 

554 "recommend", 

555 "recommended", 

556 "github", 

557 "repositories", 

558 "looking", 

559 "developers", 

560 "contribute", 

561 "contributing", 

562 "beginner", 

563 "newcomer", 

564 ] 

565 ] 

566 

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

568 specialized_query += " " + " ".join( 

569 keywords[:5] 

570 ) # Add up to 5 keywords 

571 

572 logger.info("Using specialized contribution query") 

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("No GitHub results found") 

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["full_content_is_complete"] = bool(readme_content) 

638 result["content_type"] = "readme" 

639 

640 # Get recent issues if requested 

641 if self.include_issues: 

642 issues = self._get_recent_issues(repo_full_name) 

643 result["recent_issues"] = issues 

644 

645 elif search_type == "code": 

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

647 if file_url: 

648 # Get file content 

649 file_content = self._get_file_content(file_url) 

650 result["full_content"] = file_content 

651 result["full_content_is_complete"] = bool(file_content) 

652 result["content_type"] = "file" 

653 

654 elif search_type == "issue": 

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

656 # We'll just keep it as is 

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

658 result["full_content_is_complete"] = False 

659 result["content_type"] = "issue" 

660 

661 elif search_type == "user": 

662 # For users, construct a profile summary 

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

664 

665 if item.get("name"): 

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

667 

668 if item.get("location"): 

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

670 

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

672 profile_summary += ( 

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

674 ) 

675 

676 if ( 

677 item.get("snippet") 

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

679 ): 

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

681 

682 result["full_content"] = profile_summary 

683 result["full_content_is_complete"] = False 

684 result["content_type"] = "user_profile" 

685 

686 results.append(result) 

687 

688 return results 

689 

690 def search_repository( 

691 self, repo_owner: str, repo_name: str 

692 ) -> Dict[str, Any]: 

693 """ 

694 Get detailed information about a specific repository. 

695 

696 Args: 

697 repo_owner: Owner of the repository 

698 repo_name: Name of the repository 

699 

700 Returns: 

701 Dictionary with repository information 

702 """ 

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

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

705 

706 try: 

707 # Get repository details 

708 # Apply rate limiting before request 

709 self._last_wait_time = self.rate_tracker.apply_rate_limit( 

710 self.engine_type 

711 ) 

712 

713 response = safe_get( 

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

715 ) 

716 

717 # Check for rate limiting 

718 self._handle_rate_limits(response) 

719 

720 if response.status_code == 200: 

721 repo = response.json() 

722 

723 # Format as repository preview 

724 result = self._format_repository_preview(repo) 

725 

726 # Add README content if requested 

727 if self.include_readme: 

728 readme_content = self._get_readme_content(repo_full_name) 

729 result["full_content"] = readme_content 

730 result["full_content_is_complete"] = bool(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_cached(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}. "