CodeWeaver/src/tools/search_tools.py

import logging
import os
import time
from typing import List

import requests
from tavily import TavilyClient  # type: ignore[import]

from src.agent.state import SearchResult


logger = logging.getLogger(__name__)


def search_stackoverflow(query: str, limit: int = 3) -> List[SearchResult]:
    """Stack Overflow에서 관련 질문을 검색한다.
    
    Args:
        query: 검색 쿼리
        limit: 반환할 최대 결과 수
        
    Returns:
        SearchResult 리스트 (실패 시 빈 리스트)
    """
    if not query.strip():
        logger.warning("Stack Overflow 검색: 빈 쿼리")
        return []
    
    try:
        url = "https://api.stackexchange.com/2.3/search/advanced"
        params = {
            "q": query,
            "order": "desc",
            "sort": "votes",
            "site": "stackoverflow",
            "pagesize": limit,
            "filter": "withbody",
        }
        
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        items = data.get("items", [])
        
        results = []
        max_score = max((item.get("score", 0) for item in items), default=1)
        
        for item in items:
            title = item.get("title", "")
            body = item.get("body", "")[:500]  # 본문 일부만 포함
            content = f"{title}\n\n{body}"
            
            score = item.get("score", 0)
            # 정규화: 0-1 범위로 변환
            relevance = min(score / max(max_score, 1), 1.0) if max_score > 0 else 0.5
            
            results.append(
                SearchResult(
                    source="Stack Overflow",
                    content=content,
                    url=item.get("link"),
                    relevance_score=relevance,
                )
            )
        
        logger.info("Stack Overflow 검색 성공: %d개 결과", len(results))
        
        # Rate limit 준수
        time.sleep(1)
        
        return results
        
    except Exception as e:
        logger.error("Stack Overflow 검색 실패: %s", e, exc_info=True)
        return []


def search_github(query: str, limit: int = 3) -> List[SearchResult]:
    """GitHub에서 관련 코드를 검색한다.
    
    Args:
        query: 검색 쿼리
        limit: 반환할 최대 결과 수
        
    Returns:
        SearchResult 리스트 (실패 시 빈 리스트)
    """
    if not query.strip():
        logger.warning("GitHub 검색: 빈 쿼리")
        return []
    
    try:
        url = "https://api.github.com/search/code"
        
        # Python 코드로 제한 (언어 감지 로직은 추후 확장 가능)
        search_query = f"{query} language:python"
        
        params = {
            "q": search_query,
            "sort": "indexed",
            "per_page": limit,
        }
        
        headers = {
            "Accept": "application/vnd.github.v3+json",
        }
        
        # GitHub 토큰이 있으면 Authorization 헤더 추가
        github_token = os.getenv("GITHUB_TOKEN")
        if github_token:
            headers["Authorization"] = f"token {github_token}"
            logger.debug("GitHub 토큰 사용 (인증된 요청)")
        else:
            logger.warning(
                "GITHUB_TOKEN이 설정되지 않음 - rate limit 제한적 (60 req/hr). "
                "토큰 설정 시 5,000 req/hr로 증가"
            )
        
        response = requests.get(url, params=params, headers=headers, timeout=10)
        response.raise_for_status()
        
        data = response.json()
        items = data.get("items", [])
        
        results = []
        for item in items:
            repo_name = item.get("repository", {}).get("full_name", "unknown")
            path = item.get("path", "")
            content = f"Repository: {repo_name}\nFile: {path}"
            
            results.append(
                SearchResult(
                    source="GitHub",
                    content=content,
                    url=item.get("html_url"),
                    relevance_score=0.8,  # GitHub 결과는 일반적으로 높은 관련도
                )
            )
        
        logger.info("GitHub 검색 성공: %d개 결과", len(results))
        
        # Rate limit 준수
        time.sleep(1)
        
        return results
        
    except requests.exceptions.HTTPError as e:
        if e.response.status_code == 403:
            logger.warning("GitHub API rate limit 초과")
        else:
            logger.error("GitHub 검색 HTTP 에러: %s", e, exc_info=True)
        return []
    except Exception as e:
        logger.error("GitHub 검색 실패: %s", e, exc_info=True)
        return []


def search_official_docs(query: str, limit: int = 3) -> List[SearchResult]:
    """Tavily API를 사용해 공식 문서를 검색한다.
    
    Args:
        query: 검색 쿼리
        limit: 반환할 최대 결과 수
        
    Returns:
        SearchResult 리스트 (실패 시 빈 리스트)
    """
    if not query.strip():
        logger.warning("Official Docs 검색: 빈 쿼리")
        return []
    
    api_key = os.getenv("TAVILY_API_KEY")
    if not api_key:
        logger.error("TAVILY_API_KEY 환경 변수가 설정되어 있지 않습니다.")
        return []
    
    try:
        client = TavilyClient(api_key=api_key)
        
        response = client.search(
            query=query,
            search_depth="basic",
            max_results=limit,
            include_domains=[
                "docs.python.org",
                "docs.oracle.com",
                "spring.io/guides",
                "developer.mozilla.org",
                "reactjs.org/docs",
            ],
        )
        
        results = []
        for item in response.get("results", []):
            content = item.get("content", "")
            url = item.get("url", "")
            score = item.get("score", 0.5)  # Tavily가 제공하는 관련도 점수
            
            results.append(
                SearchResult(
                    source="Official Docs",
                    content=content,
                    url=url,
                    relevance_score=score,
                )
            )
        
        logger.info("Tavily 검색 성공: %d개 결과", len(results))
        return results
        
    except Exception as e:
        logger.error("Tavily 검색 실패: %s", e, exc_info=True)
        return []