github_search.py

import requests
from typing import List, Optional

from config import settings
from schemas import RetrievedEvidence, SourceType


GITHUB_SEARCH_API = "https://api.github.com/search/issues"


def clean_text(text: Optional[str]) -> str:
    if not text:
        return ""
    return str(text).strip()


def build_github_query(
    message: str,
    error_message: Optional[str] = None,
    language: Optional[str] = None,
    framework: Optional[str] = None,
) -> str:
    parts = []

    if framework:
        parts.append(clean_text(framework))

    if language:
        parts.append(clean_text(language))

    if error_message:
        parts.append(f'"{clean_text(error_message)}"')

    if message:
        parts.append(clean_text(message))

    parts.append("is:issue")

    return " ".join(part for part in parts if part).strip()


def search_github(
    message: str,
    error_message: Optional[str] = None,
    language: Optional[str] = None,
    framework: Optional[str] = None,
    max_results: Optional[int] = None,
) -> List[RetrievedEvidence]:
    query = build_github_query(
        message=message,
        error_message=error_message,
        language=language,
        framework=framework,
    )

    if not query:
        return []

    headers = {
        "Accept": "application/vnd.github+json",
    }

    if settings.GITHUB_TOKEN:
        headers["Authorization"] = f"Bearer {settings.GITHUB_TOKEN}"

    params = {
        "q": query,
        "sort": "reactions",
        "order": "desc",
        "per_page": max_results or settings.MAX_GITHUB_RESULTS,
    }

    try:
        response = requests.get(
            GITHUB_SEARCH_API,
            headers=headers,
            params=params,
            timeout=settings.SEARCH_TIMEOUT_SECONDS,
        )
        response.raise_for_status()
        data = response.json()
    except Exception as e:
        print(f"GitHub search failed: {e}")
        return []

    items = data.get("items", [])
    evidence_list: List[RetrievedEvidence] = []

    for item in items:
        title = clean_text(item.get("title"))
        url = clean_text(item.get("html_url"))
        score = item.get("score")
        state = clean_text(item.get("state"))
        repo_full_name = clean_text(item.get("repository_url", "").split("/repos/")[-1])

        snippet_parts = []
        if repo_full_name:
            snippet_parts.append(f"Repo: {repo_full_name}")
        if state:
            snippet_parts.append(f"State: {state}")

        comments = item.get("comments")
        if comments is not None:
            snippet_parts.append(f"Comments: {comments}")

        snippet = " | ".join(snippet_parts)

        if not title:
            continue

        evidence_list.append(
            RetrievedEvidence(
                source_type=SourceType.GITHUB,
                title=title,
                snippet=snippet or "GitHub issue/discussion result",
                url=url or None,
                score=float(score) if score is not None else None,
            )
        )

    return evidence_list