from typing import Dict, List, Any
import logging
from fastapi import APIRouter, Request, HTTPException, Query
from fastapi.responses import HTMLResponse, JSONResponse
from ..core import Cache, ExaClient, OpenRouterClient, MarkdownRenderer, settings
logger = logging.getLogger(__name__)
def _format_results_as_markdown(query: str, results: List[Dict]) -> str:
if not results:
return f"No search results found for: {query}"
output = f"# Search Results for: {query}\n\n"
for i, result in enumerate(results, 1):
output += f"## {i}. {result.get('title', 'Untitled')}\n\n"
output += f"**URL:** {result.get('url', '')}\n\n"
text = result.get('text', '')
if text:
output += f"{text[:500]}...\n\n"
highlights = result.get('highlights', [])
if highlights:
output += f"**Key points:**\n"
for highlight in highlights:
output += f"- {highlight}\n"
output += "\n"
return output
router = APIRouter()
cache = Cache(settings.CACHE_DB_PATH)
exa = ExaClient()
openrouter = OpenRouterClient()
renderer = MarkdownRenderer()
@router.get("/", response_class=HTMLResponse)
async def home(request: Request):
return request.state.templates.TemplateResponse("home.html", {"request": request})
@router.get("/ai", response_class=HTMLResponse)
async def search_ai(request: Request, q: str = Query(""), num: int = 10):
query = q.strip() if q else ""
if not query:
return request.state.templates.TemplateResponse("home.html", {"request": request})
logger.info(f"AI search request for: '{query}' (num={num})")
try:
cached = cache.get(query, "ai", num)
if cached:
logger.info(f"Cache hit for AI search: '{query}'")
markdown_answer = cached.get("markdown_answer", "")
html_answer = renderer.render(markdown_answer)
search_results = cached.get("search_results", [])
search_type = cached.get("search_type", "web")
else:
logger.info(f"Cache miss for AI search: '{query}'. Fetching from Exa.")
try:
search_data = exa.answer_ai(query, num)
search_results = search_data.get("results", [])
search_type = search_data.get("search_type", "web")
if not search_results:
logger.warning(f"No results found from Exa for: '{query}'")
raise Exception("No results found from Exa")
logger.info(f"Generating AI answer with OpenRouter for: '{query}'")
markdown_answer = await openrouter.format_results_as_markdown(query, search_results)
html_answer = renderer.render(markdown_answer)
cache.set(query, "ai", num, {
"html_answer": html_answer,
"search_results": search_results,
"search_type": search_type,
"markdown_answer": markdown_answer
})
except Exception as e:
logger.error(f"AI search failed, falling back to web search for: '{query}'. Error: {e}")
# Fallback to standard web search
fallback_results = exa.search_web(query, num)
markdown_answer = _format_results_as_markdown(query, fallback_results)
html_answer = renderer.render(markdown_answer)
result_data = {
"html_answer": html_answer,
"search_results": fallback_results,
"search_type": "web",
"markdown_answer": markdown_answer,
"fallback": True
}
cache.set(query, "ai", num, result_data)
search_results = fallback_results
search_type = "web"
return request.state.templates.TemplateResponse(
"results_ai.html",
{
"request": request,
"query": query,
"answer": html_answer,
"search_results": search_results,
"search_type": search_type
}
)
except Exception as e:
logger.error(f"Critical error in /ai endpoint: {e}")
return request.state.templates.TemplateResponse(
"home.html",
{"request": request, "error": "An internal error occurred. Please try again."}
)
@router.get("/api/ai")
async def api_ai(q: str = Query(""), num: int = Query(10)):
query = q.strip() if q else ""
if not query:
return {
"error": "Query parameter 'q' is required",
"query": ""
}
try:
cached = cache.get(query, "ai", num)
if cached:
markdown_answer = cached.get("markdown_answer", "")
html_answer = renderer.render(markdown_answer)
search_results = cached.get("search_results", [])
search_type = cached.get("search_type", "web")
return {
"html_answer": html_answer,
"markdown_answer": markdown_answer,
"search_results": search_results,
"search_type": search_type,
"cached": True
}
try:
search_data = exa.answer_ai(query, num)
search_results = search_data.get("results", [])
search_type = search_data.get("search_type", "web")
if not search_results:
raise Exception("No results found from Exa")
markdown_answer = await openrouter.format_results_as_markdown(query, search_results)
html_answer = renderer.render(markdown_answer)
result = {
"html_answer": html_answer,
"markdown_answer": markdown_answer,
"search_results": search_results,
"search_type": search_type,
"cached": False
}
cache.set(query, "ai", num, result)
return result
except Exception as e:
logger.error(f"API AI search failed, falling back: {e}")
fallback_results = exa.search_web(query, num)
markdown_answer = _format_results_as_markdown(query, fallback_results)
html_answer = renderer.render(markdown_answer)
result = {
"html_answer": html_answer,
"markdown_answer": markdown_answer,
"search_results": fallback_results,
"search_type": "web",
"cached": False,
"error": str(e),
"fallback": True
}
cache.set(query, "ai", num, result)
return result
except Exception as e:
logger.error(f"Critical error in /api/ai endpoint: {e}")
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(e)}
)
@router.get("/search", response_class=HTMLResponse)
async def search_web(request: Request, q: str = Query(""), num: int = 10):
query = q.strip() if q else ""
if not query:
return request.state.templates.TemplateResponse("home.html", {"request": request})
logger.info(f"Web search request for: '{query}' (num={num})")
try:
cached_results = cache.get(query, "web", num)
if cached_results:
logger.info(f"Cache hit for web search: '{query}'")
results = cached_results.get("results", [])
else:
logger.info(f"Cache miss for web search: '{query}'. Fetching from Exa.")
results = exa.search_web(query, num)
cache.set(query, "web", num, {"results": results})
return request.state.templates.TemplateResponse(
"results_web.html",
{"request": request, "query": query, "results": results},
)
except Exception as e:
logger.error(f"Error in /search endpoint: {e}")
return request.state.templates.TemplateResponse(
"home.html",
{"request": request, "error": "Search failed. Please try again."}
)
@router.get("/images", response_class=HTMLResponse)
async def search_images(request: Request, q: str = Query(""), num: int = 20):
query = q.strip() if q else ""
if not query:
return request.state.templates.TemplateResponse("home.html", {"request": request})
try:
cached_results = cache.get(query, "images", num)
if cached_results:
results = cached_results.get("results", [])
else:
results = exa.search_images(query, num)
cache.set(query, "images", num, {"results": results})
return request.state.templates.TemplateResponse(
"results_images.html",
{"request": request, "query": query, "results": results},
)
except Exception as e:
logger.error(f"Error in /images endpoint: {e}")
return request.state.templates.TemplateResponse(
"home.html",
{"request": request, "error": "Image search failed. Please try again."}
)
@router.get("/api/search")
async def api_search(q: str = Query(""), num: int = Query(10)):
query = q.strip() if q else ""
if not query:
return {
"error": "Query parameter 'q' is required",
"results": []
}
try:
cached_results = cache.get(query, "web", num)
if cached_results:
return {"results": cached_results.get("results", []), "cached": True}
results = exa.search_web(query, num)
cache.set(query, "web", num, {"results": results})
return {"results": results, "cached": False}
except Exception as e:
logger.error(f"Error in /api/search endpoint: {e}")
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(e)}
)
@router.get("/api/images")
async def api_images(q: str = Query(""), num: int = Query(20)):
query = q.strip() if q else ""
if not query:
return {
"error": "Query parameter 'q' is required",
"results": []
}
try:
cached_results = cache.get(query, "images", num)
if cached_results:
return {"results": cached_results.get("results", []), "cached": True}
results = exa.search_images(query, num)
cache.set(query, "images", num, {"results": results})
return {"results": results, "cached": False}
except Exception as e:
logger.error(f"Error in /api/images endpoint: {e}")
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(e)}
)
@router.get("/api/mixed")
async def api_mixed(q: str = Query(...), num: int = Query(10)):
if not q.strip():
raise HTTPException(status_code=400, detail="Query parameter 'q' is required")
try:
cached_results = cache.get(q, "mixed", num)
if cached_results:
return cached_results
results = exa.search_mixed(q, num)
cache.set(q, "mixed", num, {"results": results})
return results
except Exception as e:
logger.error(f"Error in /api/mixed endpoint: {e}")
return JSONResponse(
status_code=500,
content={"error": "Internal server error", "detail": str(e)}
)
Reference in New Issue View Git Blame Copy Permalink