Spaces:
Sleeping
Sleeping
File size: 3,053 Bytes
3869fd1 59b66b3 3869fd1 59b66b3 3869fd1 546cf90 3869fd1 5620621 3869fd1 59b66b3 3869fd1 546cf90 5620621 546cf90 b34be46 |
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 |
from dotenv import load_dotenv
import os
import aiohttp
class GoogleSearch:
def __init__(self):
load_dotenv()
self.counter = 0
self.api_key = os.environ["GOOGLE_API_KEY"]
self.cse_id = os.getenv("GOOGLE_CSE_ID")
async def google_search(self, query: str, num_results: int = 5) -> str:
"""
Args:
query: Search query
num_results: Max results to return
Returns:
dict: JSON response from Google API.
"""
if self.counter > 1:
return "No more searches, move on"
self.counter += 1
if not self.api_key or not self.cse_id:
raise ValueError(
"GOOGLE_API_KEY and GOOGLE_CSE_ID must be set in environment variables."
)
url = "https://www.googleapis.com/customsearch/v1"
params = {"key": self.api_key, "cx": self.cse_id, "q": query}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
results = "Web Search results:\n\n" + "\n\n".join(
[
f"Link:{result['link']}\nTitle:{result['title']}\nSnippet:{result['snippet']}"
for result in data["items"][:num_results]
]
)
return results
else:
return f"Search failed with status {response.status}"
async def google_image_search(self, query: str) -> str:
"""
Args:
query: Search query
Returns:
dict: JSON response from Google API.
"""
if self.counter > 2:
return "No more searches, move on"
self.counter += 1
if not self.api_key or not self.cse_id:
raise ValueError(
"GOOGLE_API_KEY and GOOGLE_CSE_ID must be set in environment variables."
)
url = "https://www.googleapis.com/customsearch/v1"
params = {"key": self.api_key, "cx": self.cse_id, "q": query}
async with aiohttp.ClientSession() as session:
async with session.get(url, params=params) as response:
if response.status == 200:
data = await response.json()
results = "Web Search results:\n\n" + "\n\n".join(
[
f"Link:{result['link']}\nTitle:{result['title']}"
for result in data["items"][:4]
]
)
return results
else:
return f"Search failed with status {response.status}"
async def main():
google_search = GoogleSearch()
results = await google_search.google_search("Fresident of Prance?")
print(results)
if __name__ == "__main__":
import asyncio
asyncio.run(main())
|