File size: 2,922 Bytes
f47f67d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 | 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 |