tools/github_search.py

# GitHub Search Tool
"""
Search GitHub repositories and provide recommendations.
Simple keyword search with README summarization.
"""

import httpx
import logging
from typing import List, Dict, Any, Optional

from config import settings

logger = logging.getLogger(__name__)


class GitHubSearchTool:
    """
    GitHub repository search tool.
    Searches for relevant projects and provides recommendations.
    """
    
    def __init__(self):
        self.api_base = "https://api.github.com"
        self.token = settings.GITHUB_TOKEN
        self.client = httpx.AsyncClient(timeout=30.0)
        
        # Set headers
        self.headers = {
            "Accept": "application/vnd.github.v3+json",
            "User-Agent": "AI-Workflow-Agent"
        }
        if self.token:
            self.headers["Authorization"] = f"token {self.token}"
    
    async def search(
        self,
        keywords: str,
        max_results: int = 3,
        language: Optional[str] = None
    ) -> List[Dict[str, Any]]:
        """
        Search GitHub repositories by keywords.
        
        Args:
            keywords: Search terms
            max_results: Maximum number of results (default 3)
            language: Optional language filter
            
        Returns:
            List of repository info dictionaries
        """
        try:
            # Build search query
            query = keywords
            if language:
                query += f" language:{language}"
            
            # Search repositories
            response = await self.client.get(
                f"{self.api_base}/search/repositories",
                params={
                    "q": query,
                    "sort": "stars",
                    "order": "desc",
                    "per_page": max_results
                },
                headers=self.headers
            )
            
            if response.status_code == 200:
                data = response.json()
                repos = data.get("items", [])
                
                # Extract relevant info
                results = []
                for repo in repos[:max_results]:
                    repo_info = await self._extract_repo_info(repo)
                    results.append(repo_info)
                
                return results
            else:
                logger.error(f"GitHub search failed: {response.status_code}")
                return []
                
        except Exception as e:
            logger.error(f"GitHub search error: {e}")
            return []
    
    async def _extract_repo_info(self, repo: Dict[str, Any]) -> Dict[str, Any]:
        """Extract relevant information from repository data."""
        # Get README summary
        readme_summary = await self._get_readme_summary(
            repo.get("owner", {}).get("login", ""),
            repo.get("name", "")
        )
        
        return {
            "name": repo.get("name", ""),
            "full_name": repo.get("full_name", ""),
            "url": repo.get("html_url", ""),
            "clone_url": repo.get("clone_url", ""),
            "description": repo.get("description", "No description"),
            "stars": repo.get("stargazers_count", 0),
            "forks": repo.get("forks_count", 0),
            "language": repo.get("language", "Unknown"),
            "topics": repo.get("topics", []),
            "updated_at": repo.get("updated_at", ""),
            "has_docker": await self._check_has_docker(
                repo.get("owner", {}).get("login", ""),
                repo.get("name", "")
            ),
            "readme_summary": readme_summary
        }
    
    async def _get_readme_summary(self, owner: str, repo: str) -> str:
        """Fetch and summarize repository README."""
        try:
            response = await self.client.get(
                f"{self.api_base}/repos/{owner}/{repo}/readme",
                headers=self.headers
            )
            
            if response.status_code == 200:
                data = response.json()
                # README is base64 encoded
                import base64
                content = base64.b64decode(data.get("content", "")).decode("utf-8")
                
                # Simple summary: first 500 chars
                summary = content[:500].replace("\n", " ").strip()
                if len(content) > 500:
                    summary += "..."
                return summary
            return "README not available"
            
        except Exception as e:
            logger.warning(f"README fetch error: {e}")
            return "README not available"
    
    async def _check_has_docker(self, owner: str, repo: str) -> bool:
        """Check if repository has Dockerfile or docker-compose."""
        try:
            # Check for Dockerfile
            response = await self.client.get(
                f"{self.api_base}/repos/{owner}/{repo}/contents/Dockerfile",
                headers=self.headers
            )
            if response.status_code == 200:
                return True
            
            # Check for docker-compose
            response = await self.client.get(
                f"{self.api_base}/repos/{owner}/{repo}/contents/docker-compose.yml",
                headers=self.headers
            )
            if response.status_code == 200:
                return True
                
            # Check for docker-compose.yaml
            response = await self.client.get(
                f"{self.api_base}/repos/{owner}/{repo}/contents/docker-compose.yaml",
                headers=self.headers
            )
            return response.status_code == 200
            
        except Exception:
            return False
    
    async def generate_recommendation(self, repos: List[Dict[str, Any]]) -> str:
        """
        Generate a recommendation summary for found repositories.
        
        Args:
            repos: List of repository info dictionaries
            
        Returns:
            Recommendation text
        """
        if not repos:
            return "No repositories found. Try different keywords."
        
        # Build recommendation
        lines = ["📦 **Repository Recommendations:**\n"]
        
        for i, repo in enumerate(repos, 1):
            stars = repo.get("stars", 0)
            docker_status = "✅ Docker" if repo.get("has_docker") else "❌ No Docker"
            
            lines.append(f"**{i}. {repo.get('name', 'Unknown')}** ⭐ {stars}")
            lines.append(f"   {repo.get('description', 'No description')}")
            lines.append(f"   Language: {repo.get('language', 'Unknown')} | {docker_status}")
            lines.append(f"   URL: {repo.get('url', '')}")
            lines.append("")
        
        # Add best pick recommendation
        if repos:
            best = max(repos, key=lambda r: r.get("stars", 0))
            if best.get("has_docker"):
                lines.append(f"💡 **Recommended:** {best.get('name')} (most stars + Docker support)")
            else:
                docker_repos = [r for r in repos if r.get("has_docker")]
                if docker_repos:
                    best_docker = max(docker_repos, key=lambda r: r.get("stars", 0))
                    lines.append(f"💡 **Recommended:** {best_docker.get('name')} (has Docker support)")
                else:
                    lines.append(f"💡 **Recommended:** {best.get('name')} (most stars, but needs Docker setup)")
        
        return "\n".join(lines)
    
    async def get_repo_details(self, repo_url: str) -> Dict[str, Any]:
        """
        Get detailed information about a specific repository.
        
        Args:
            repo_url: GitHub repository URL or owner/repo format
            
        Returns:
            Repository details dictionary
        """
        try:
            # Parse owner/repo from URL or direct format
            if "github.com" in repo_url:
                parts = repo_url.rstrip("/").split("/")
                owner = parts[-2]
                repo = parts[-1]
            else:
                owner, repo = repo_url.split("/")
            
            response = await self.client.get(
                f"{self.api_base}/repos/{owner}/{repo}",
                headers=self.headers
            )
            
            if response.status_code == 200:
                return await self._extract_repo_info(response.json())
            else:
                return {"error": f"Repository not found: {response.status_code}"}
                
        except Exception as e:
            logger.error(f"Repo details error: {e}")
            return {"error": str(e)}