Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 25 additions & 2 deletions _legacy/skills/web_search.py
Original file line number Diff line number Diff line change
@@ -1,25 +1,48 @@
import logging
import asyncio
import io
import os
import re
from telegram.ext import ContextTypes

import config

def perform_web_search(query):
def _perform_tavily_search(query):
from tavily import TavilyClient
try:
client = TavilyClient()
response = client.search(query=query, max_results=5, search_depth="basic")
results = response.get("results", [])
if not results:
return "No results found."

summary = ""
for r in results:
summary += f"- [{r['title']}]({r['url']}): {r.get('content', '')}\n"
return summary
except Exception as e:
return f"Error performing search: {e}"
Comment on lines +11 to +24
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

The try...except block does not cover the import statement. If the tavily-python package is not installed, an ImportError will be raised and crash the application instead of being handled gracefully. The import should be moved inside the try block. Additionally, it's good practice to log exceptions before returning an error message to aid in debugging.

Suggested change
from tavily import TavilyClient
try:
client = TavilyClient()
response = client.search(query=query, max_results=5, search_depth="basic")
results = response.get("results", [])
if not results:
return "No results found."
summary = ""
for r in results:
summary += f"- [{r['title']}]({r['url']}): {r.get('content', '')}\n"
return summary
except Exception as e:
return f"Error performing search: {e}"
try:
from tavily import TavilyClient
client = TavilyClient()
response = client.search(query=query, max_results=5, search_depth="basic")
results = response.get("results", [])
if not results:
return "No results found."
summary = ""
for r in results:
summary += f"- [{r['title']}]({r['url']}): {r.get('content', '')}\n"
return summary
except ImportError:
logging.error("Tavily SDK not installed. Please run 'pip install tavily-python'.")
return "Error: Tavily search provider is not available."
except Exception as e:
logging.error(f"Error performing Tavily search: {e}")
return f"Error performing search: {e}"


def _perform_ddgs_search(query):
from ddgs import DDGS
try:
results = DDGS().text(query, max_results=5)
if not results:
return "No results found."

summary = ""
for r in results:
summary += f"- [{r['title']}]({r['href']}): {r['body']}\n"
return summary
except Exception as e:
return f"Error performing search: {e}"
Comment on lines 27 to 38
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

high

There are a couple of improvements that can be made here:

  1. The try...except block does not cover the import statement. If the ddgs package is not installed, an ImportError will be raised and crash the application. The import should be moved inside the try block.
  2. Using r['body'] can raise a KeyError if the body key is missing in a result. It's safer to use r.get('body', '') for robustness, which is consistent with the Tavily implementation.

The suggestion below addresses both points.

Suggested change
from ddgs import DDGS
try:
results = DDGS().text(query, max_results=5)
if not results:
return "No results found."
summary = ""
for r in results:
summary += f"- [{r['title']}]({r['href']}): {r['body']}\n"
return summary
except Exception as e:
return f"Error performing search: {e}"
try:
from ddgs import DDGS
results = DDGS().text(query, max_results=5)
if not results:
return "No results found."
summary = ""
for r in results:
summary += f"- [{r['title']}]({r['href']}): {r.get('body', '')}\n"
return summary
except ImportError:
logging.error("DDGS SDK not installed. Please run 'pip install ddgs'.")
return "Error: DuckDuckGo search provider is not available."
except Exception as e:
logging.error(f"Error performing DDGS search: {e}")
return f"Error performing search: {e}"


def perform_web_search(query):
provider = os.environ.get("SEARCH_PROVIDER", "duckduckgo").lower()
if provider == "tavily":
return _perform_tavily_search(query)
return _perform_ddgs_search(query)
Comment on lines +42 to +44
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The current implementation silently defaults to DuckDuckGo for any SEARCH_PROVIDER value other than "tavily". This could be confusing if a user makes a typo (e.g., "tavilyy"). It would be more user-friendly to explicitly handle the "duckduckgo" case and log a warning for unknown providers before defaulting.

    if provider == "tavily":
        return _perform_tavily_search(query)
    elif provider == "duckduckgo":
        return _perform_ddgs_search(query)
    else:
        logging.warning(f"Unknown SEARCH_PROVIDER '{provider}', defaulting to duckduckgo.")
        return _perform_ddgs_search(query)


def get_youtube_video_id(url):
# Patterns: youtube.com/watch?v=ID, youtu.be/ID, youtube.com/embed/ID
patterns = [
Expand Down
1 change: 1 addition & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ pyyaml
dateparser
pytz
ddgs
tavily-python
youtube-transcript-api
python-dateutil
playwright
Expand Down