stack_search.py

import requests
from typing import List, Optional

from config import settings
from schemas import RetrievedEvidence, SourceType


STACK_API_BASE = "https://api.stackexchange.com/2.3/search/advanced"
STACK_SITE = "stackoverflow"


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


def build_stack_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(clean_text(error_message))

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

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


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

    if not query:
        return []

    params = {
        "order": "desc",
        "sort": "relevance",
        "q": query,
        "site": STACK_SITE,
        "pagesize": max_results or settings.MAX_STACK_RESULTS,
        "filter": "default",
    }

    if settings.STACKOVERFLOW_KEY:
        params["key"] = settings.STACKOVERFLOW_KEY

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

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

    for item in items:
        title = clean_text(item.get("title"))
        link = clean_text(item.get("link"))
        score = item.get("score", 0)

        tags = item.get("tags", [])
        tag_text = ", ".join(tags) if tags else ""
        is_answered = item.get("is_answered", False)

        snippet_parts = []
        if tag_text:
            snippet_parts.append(f"Tags: {tag_text}")
        snippet_parts.append(f"Answered: {'yes' if is_answered else 'no'}")
        snippet_parts.append(f"Score: {score}")

        snippet = " | ".join(snippet_parts)

        if not title:
            continue

        evidence_list.append(
            RetrievedEvidence(
                source_type=SourceType.STACKOVERFLOW,
                title=title,
                snippet=snippet,
                url=link or None,
                score=float(score) if score is not None else None,
            )
        )

    return evidence_list