web_search.py

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())