File size: 2,319 Bytes
712579e |
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 |
import os
import asyncio
import httpx
import nest_asyncio
from dotenv import load_dotenv
from logger.custom_logger import CustomLoggerTracker
custom_logger = CustomLoggerTracker()
logger = custom_logger.get_logger("web_search")
load_dotenv()
nest_asyncio.apply()
async def search_autism(query: str) -> dict:
base_url = os.environ["WEBSEARCH_BASE_URL"]
api_key = os.environ["GOOGLE_API_KEY"]
search_engine_id = os.environ["GOOGLE_SEARCH_ENGINE_ID"]
if not api_key or not search_engine_id:
logger.error(f"results: [], answer: Missing GOOGLE_API_KEY or GOOGLE_SEARCH_ENGINE_ID in .env file.")
return {"results": [], "answer": "Missing GOOGLE_API_KEY or GOOGLE_SEARCH_ENGINE_ID in .env file."}
params = {
"key": api_key,
"cx": search_engine_id,
"q": query}
async with httpx.AsyncClient() as client:
try:
response = await client.get(base_url, params=params)
response.raise_for_status()
data = response.json()
results = [
{
"title": item.get("title"),
"url": item.get("link"),
"snippet": item.get("snippet")
}
for item in data.get("items", [])
]
snippets = [res["snippet"] for res in results if res.get("snippet")]
answer = "\n".join(snippets) if snippets else "No answer could be constructed from search results."
return {"results": results, "answer": answer}
except httpx.HTTPStatusError as e:
return {"results": [], "answer": f"HTTP Error: {str(e)} - Response: {e.response.text}"}
except Exception as e:
return {"results": [], "answer": f"An unexpected error occurred: {str(e)}"}
async def main():
"""Main async function to run the search test."""
query = "autism symptoms and treatments"
logger.info(f"Searching for: '{query}'")
result = await search_autism(query)
logger.info("\n--- Search Results ---")
for res in result.get("results", []):
logger.info(f"- {res.get('title')} ({res.get('url')})")
logger.info("\n--- Generated Answer ---")
logger.info(result.get("answer", "No answer provided."))
if __name__ == "__main__":
asyncio.run(main()) |