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