fastapi-v2 / app /web_search.py
Shageenderan Sapai
Align with staging branch web_search
1aeb534
raw
history blame
12.7 kB
import requests
from dotenv import load_dotenv
import os
import logging
logger = logging.getLogger(__name__)
load_dotenv()
class SearchEngine:
BING_API_KEY = os.getenv("BING_API_KEY")
BING_ENDPOINT = 'https://api.bing.microsoft.com/v7.0'
@staticmethod
def search(feedback_type_name, search_term, user_id):
logger.info(f"User {user_id}: Initiating search for type '{feedback_type_name}' with term '{search_term}'")
"""
Public method to perform a search based on the feedback type.
"""
search_methods = {
"General": SearchEngine._search_general,
"Resource Links": SearchEngine._search_relevant_links,
"Book/Podcast Recommendations": SearchEngine._search_books_or_podcasts,
"Inspirational Stories or Case Studies": SearchEngine._search_inspirational_stories,
"Fun Facts": SearchEngine._search_fun_facts,
"Personalised Recommendations": SearchEngine._search_personalized_recommendations,
"Videos": SearchEngine._search_videos
}
search_method = search_methods.get(feedback_type_name)
if search_method:
return search_method(search_term)
else:
return (feedback_type_name, search_term)
@staticmethod
def _search_relevant_links(search_term):
logger.debug(f"Searching relevant links for term: {search_term}")
"""
Uses Bing Web Search API to search for relevant links.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
params = {'q': search_term, 'textDecorations': True, 'textFormat': 'HTML', 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Web Search API.")
data = response.json()
links = []
if 'webPages' in data and 'value' in data['webPages']:
for result in data['webPages']['value']:
links.append(result)
return links
else:
logger.error(f"Bing Web Search API returned status code {response.status_code}")
return ["No relevant links found."]
@staticmethod
def _search_books_or_podcasts(search_term):
logger.debug(f"Searching books or podcasts for term: {search_term}")
"""
Uses Bing Web Search API to search for books or podcasts.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
query = f"{search_term} book OR podcast"
params = {'q': query, 'textDecorations': True, 'textFormat': 'HTML', 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Web Search API for books/podcasts.")
data = response.json()
recommendations = []
if 'webPages' in data and 'value' in data['webPages']:
for result in data['webPages']['value']:
title = result.get('name', 'Unknown Title')
url = result.get('url', '')
recommendations.append(f"{title}: {url}")
return recommendations
else:
logger.error(f"Bing Web Search API returned status code {response.status_code} for books/podcasts search.")
return ["No book or podcast recommendations found."]
@staticmethod
def _search_success_stories(search_term):
logger.debug(f"Searching success stories for term: {search_term}")
"""
Uses Bing Web Search API to search for success stories.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
query = f"{search_term} success stories"
params = {'q': query, 'textDecorations': True, 'textFormat': 'HTML', 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Web Search API for success stories.")
data = response.json()
stories = []
if 'webPages' in data and 'value' in data['webPages']:
for result in data['webPages']['value']:
title = result.get('name', 'Unknown Title')
url = result.get('url', '')
stories.append(f"{title}: {url}")
return stories
else:
logger.error(f"Bing Web Search API returned status code {response.status_code} for success stories search.")
return ["No success stories found."]
@staticmethod
def _search_inspirational_stories(search_term):
logger.debug(f"Searching inspirational stories or case studies for term: {search_term}")
"""
Uses Bing Web Search API to search for inspirational stories or case studies.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
query = f"{search_term} inspirational stories OR case studies"
params = {'q': query, 'textDecorations': True, 'textFormat': 'HTML', 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Web Search API for inspirational stories.")
data = response.json()
stories = []
if 'webPages' in data and 'value' in data['webPages']:
for result in data['webPages']['value']:
title = result.get('name', 'Unknown Title')
url = result.get('url', '')
stories.append(f"{title}: {url}")
return stories
else:
logger.error(f"Bing Web Search API returned status code {response.status_code} for inspirational stories search.")
return ["No inspirational stories found."]
@staticmethod
def _search_fun_facts(search_term):
logger.debug(f"Searching fun facts for term: {search_term}")
"""
Uses Bing Web Search API to search for fun facts related to personal growth.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
query = f"{search_term} fun facts"
params = {'q': query, 'textDecorations': True, 'textFormat': 'HTML', 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Web Search API for fun facts.")
data = response.json()
facts = []
if 'webPages' in data and 'value' in data['webPages']:
for result in data['webPages']['value']:
snippet = result.get('snippet', '')
facts.append(snippet)
return facts
else:
logger.error(f"Bing Web Search API returned status code {response.status_code} for fun facts search.")
return ["No fun facts found."]
@staticmethod
def _search_visual_content(search_term):
logger.debug(f"Searching visual content for term: {search_term}")
"""
Uses Bing Image Search API to search for images or infographics.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
params = {'q': search_term, 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/images/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Image Search API.")
data = response.json()
images = []
if 'value' in data:
for result in data['value']:
image_url = result.get('contentUrl', '')
images.append(image_url)
return images
else:
logger.error(f"Bing Image Search API returned status code {response.status_code} for visual content search.")
return ["No visual content found."]
@staticmethod
def _search_personalized_recommendations(search_term):
logger.debug(f"Searching personalized recommendations for term: {search_term}")
"""
Uses Bing Web Search API to provide personalized recommendations.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
query = f"tips for {search_term}"
params = {'q': query, 'textDecorations': True, 'textFormat': 'HTML', 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Web Search API for personalized recommendations.")
data = response.json()
recommendations = []
if 'webPages' in data and 'value' in data['webPages']:
for result in data['webPages']['value']:
title = result.get('name', 'Unknown Title')
url = result.get('url', '')
recommendations.append(f"{title}: {url}")
return recommendations
else:
logger.error(f"Bing Web Search API returned status code {response.status_code} for personalized recommendations search.")
return ["No personalized recommendations found."]
@staticmethod
def _search_videos(search_term):
logger.debug(f"Searching videos for term: {search_term}")
"""
Uses Bing Video Search API to search for videos, prioritizing YouTube results.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
query = f"site:youtube.com {search_term}"
params = {'q': query, 'textDecorations': True, 'textFormat': 'HTML', 'count': 3}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/videos/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Video Search API.")
data = response.json()
videos = []
if 'value' in data:
for result in data['value']:
title = result.get('name', 'Unknown Title')
url = result.get('contentUrl', '')
# Prioritize YouTube results
if 'youtube.com' in url.lower():
videos.append(f"{title}: {url}")
if len(videos) >= 3:
break
# If we don't have enough YouTube results, add other video results
if len(videos) < 3:
for result in data['value']:
title = result.get('name', 'Unknown Title')
url = result.get('contentUrl', '')
if url not in [v.split(': ')[1] for v in videos]:
videos.append(f"{title}: {url}")
if len(videos) >= 3:
break
return videos
else:
logger.error(f"Bing Video Search API returned status code {response.status_code} for video search.")
return ["No video content found."]
@staticmethod
def _search_general(search_term):
logger.debug(f"Performing a general search for term: {search_term}")
"""
Uses Bing Web Search API to perform a general search.
"""
headers = {'Ocp-Apim-Subscription-Key': SearchEngine.BING_API_KEY}
params = {'q': search_term, 'textDecorations': True, 'textFormat': 'HTML', 'count': 5}
response = requests.get(f"{SearchEngine.BING_ENDPOINT}/search", headers=headers, params=params)
if response.status_code == 200:
logger.debug("Received successful response from Bing Web Search API for general search.")
data = response.json()
results = []
if 'webPages' in data and 'value' in data['webPages']:
for result in data['webPages']['value']:
results.append(result)
return results
else:
logger.error(f"Bing Web Search API returned status code {response.status_code} for general search.")
return ["No results found for general search."]