483 lines
19 KiB
Python
483 lines
19 KiB
Python
"""
|
||
This plugin provides DuckDuckGo search functionality using the DuckDuckGo Instant Answer API.
|
||
"""
|
||
|
||
import logging
|
||
import requests
|
||
import json
|
||
import simplematrixbotlib as botlib
|
||
from urllib.parse import quote, urlencode
|
||
import html
|
||
|
||
DDG_API_URL = "https://api.duckduckgo.com/"
|
||
DDG_SEARCH_URL = "https://html.duckduckgo.com/html/"
|
||
|
||
async def handle_command(room, message, bot, prefix, config):
|
||
"""
|
||
Function to handle DuckDuckGo search commands.
|
||
|
||
Args:
|
||
room (Room): The Matrix room where the command was invoked.
|
||
message (RoomMessage): The message object containing the command.
|
||
bot (Bot): The bot object.
|
||
prefix (str): The command prefix.
|
||
config (dict): Configuration parameters.
|
||
|
||
Returns:
|
||
None
|
||
"""
|
||
match = botlib.MessageMatch(room, message, bot, prefix)
|
||
if match.is_not_from_this_bot() and match.prefix() and match.command("ddg"):
|
||
logging.info("Received !ddg command")
|
||
|
||
args = match.args()
|
||
|
||
if len(args) < 1:
|
||
await show_usage(room, bot)
|
||
return
|
||
|
||
subcommand = args[0].lower()
|
||
|
||
if subcommand == "search":
|
||
if len(args) < 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !ddg search <query>")
|
||
return
|
||
query = ' '.join(args[1:])
|
||
await ddg_search(room, bot, query)
|
||
|
||
elif subcommand == "instant":
|
||
if len(args) < 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !ddg instant <query>")
|
||
return
|
||
query = ' '.join(args[1:])
|
||
await ddg_instant_answer(room, bot, query)
|
||
|
||
elif subcommand == "image":
|
||
if len(args) < 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !ddg image <query>")
|
||
return
|
||
query = ' '.join(args[1:])
|
||
await ddg_image_search(room, bot, query)
|
||
|
||
elif subcommand == "news":
|
||
if len(args) < 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !ddg news <query>")
|
||
return
|
||
query = ' '.join(args[1:])
|
||
await ddg_news_search(room, bot, query)
|
||
|
||
elif subcommand == "video":
|
||
if len(args) < 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !ddg video <query>")
|
||
return
|
||
query = ' '.join(args[1:])
|
||
await ddg_video_search(room, bot, query)
|
||
|
||
elif subcommand == "bang":
|
||
if len(args) < 2:
|
||
await show_bang_help(room, bot)
|
||
return
|
||
bang_query = ' '.join(args[1:])
|
||
await ddg_bang_search(room, bot, bang_query)
|
||
|
||
elif subcommand == "define":
|
||
if len(args) < 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !ddg define <word>")
|
||
return
|
||
word = ' '.join(args[1:])
|
||
await ddg_definition(room, bot, word)
|
||
|
||
elif subcommand == "calc":
|
||
if len(args) < 2:
|
||
await bot.api.send_text_message(room.room_id, "Usage: !ddg calc <expression>")
|
||
return
|
||
expression = ' '.join(args[1:])
|
||
await ddg_calculator(room, bot, expression)
|
||
|
||
elif subcommand == "weather":
|
||
location = ' '.join(args[1:]) if len(args) > 1 else ""
|
||
await ddg_weather(room, bot, location)
|
||
|
||
elif subcommand == "help":
|
||
await show_usage(room, bot)
|
||
|
||
else:
|
||
# Default to instant answer search
|
||
query = ' '.join(args)
|
||
await ddg_instant_answer(room, bot, query)
|
||
|
||
async def show_usage(room, bot):
|
||
"""Display DuckDuckGo command usage."""
|
||
usage = """
|
||
<strong>🦆 DuckDuckGo Search Commands</strong>
|
||
|
||
<strong>!ddg <query></strong> - Instant answer search (default)
|
||
<strong>!ddg search <query></strong> - Web search with results
|
||
<strong>!ddg instant <query></strong> - Instant answer with detailed info
|
||
<strong>!ddg image <query></strong> - Image search
|
||
<strong>!ddg news <query></strong> - News search
|
||
<strong>!ddg video <query></strong> - Video search
|
||
<strong>!ddg bang <!bang query></strong> - Use DuckDuckGo bangs
|
||
<strong>!ddg define <word></strong> - Word definitions
|
||
<strong>!ddg calc <expression></strong> - Calculator
|
||
<strong>!ddg weather [location]</strong> - Weather information
|
||
<strong>!ddg help</strong> - Show this help
|
||
|
||
<strong>Examples:</strong>
|
||
• <code>!ddg python programming</code>
|
||
• <code>!ddg search matrix protocol</code>
|
||
• <code>!ddg image cute cats</code>
|
||
• <code>!ddg bang !w matrix</code>
|
||
• <code>!ddg define serendipity</code>
|
||
• <code>!ddg calc 2+2*5</code>
|
||
• <code>!ddg weather London</code>
|
||
|
||
<strong>Popular Bangs:</strong>
|
||
• <code>!w</code> - Wikipedia
|
||
• <code>!g</code> - Google
|
||
• <code>!yt</code> - YouTube
|
||
• <code>!aw</code> - ArchWiki
|
||
• <code>!gh</code> - GitHub
|
||
• <code>!so</code> - Stack Overflow
|
||
"""
|
||
await bot.api.send_markdown_message(room.room_id, usage)
|
||
|
||
async def show_bang_help(room, bot):
|
||
"""Display DuckDuckGo bang help."""
|
||
bang_help = """
|
||
<strong>🦆 DuckDuckGo Bangs</strong>
|
||
|
||
<strong>Usage:</strong> <code>!ddg bang <!bang query></code>
|
||
|
||
<strong>Popular Bangs:</strong>
|
||
• <code>!ddg bang !w matrix</code> - Search Wikipedia
|
||
• <code>!ddg bang !g python</code> - Search Google
|
||
• <code>!ddg bang !yt music</code> - Search YouTube
|
||
• <code>!ddg bang !aw arch</code> - Search ArchWiki
|
||
• <code>!ddg bang !gh repository</code> - Search GitHub
|
||
• <code>!ddg bang !so error</code> - Search Stack Overflow
|
||
• <code>!ddg bang !amazon book</code> - Search Amazon
|
||
• <code>!ddg bang !imdb movie</code> - Search IMDb
|
||
• <code>!ddg bang !reddit topic</code> - Search Reddit
|
||
• <code>!ddg bang !tw tweet</code> - Search Twitter
|
||
|
||
<strong>More Bangs:</strong>
|
||
• <code>!ddg</code> - DuckDuckGo
|
||
• <code>!bing</code> - Bing
|
||
• <code>!ddgimages</code> - DuckDuckGo Images
|
||
• <code>!npm</code> - npm packages
|
||
• <code>!cpp</code> - C++ reference
|
||
• <code>!python</code> - Python docs
|
||
• <code>!rust</code> - Rust docs
|
||
• <code>!mdn</code> - MDN Web Docs
|
||
|
||
<em>Thousands of bangs available! See: https://duckduckgo.com/bangs</em>
|
||
"""
|
||
await bot.api.send_markdown_message(room.room_id, bang_help)
|
||
|
||
async def ddg_instant_answer(room, bot, query):
|
||
"""Get DuckDuckGo instant answer."""
|
||
try:
|
||
params = {
|
||
'q': query,
|
||
'format': 'json',
|
||
'no_html': '1',
|
||
'skip_disambig': '1',
|
||
'no_redirect': '1'
|
||
}
|
||
|
||
logging.info(f"Fetching DuckDuckGo instant answer for: {query}")
|
||
|
||
response = requests.get(DDG_API_URL, params=params, timeout=10)
|
||
|
||
if response.status_code != 200:
|
||
# If API fails, provide direct search link
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🦆 DuckDuckGo: {html.escape(query)}</strong><br><br>"
|
||
f"API temporarily unavailable. <a href='{search_url}'>Search on DuckDuckGo</a>"
|
||
)
|
||
return
|
||
|
||
data = response.json()
|
||
|
||
output = f"<strong>🦆 DuckDuckGo: {html.escape(query)}</strong><br><br>"
|
||
|
||
# Handle different answer types
|
||
if data.get('AbstractText'):
|
||
# Wikipedia-style answer
|
||
output += f"<strong>📚 {data.get('Heading', 'Definition')}</strong><br>"
|
||
output += f"{html.escape(data['AbstractText'])}<br>"
|
||
if data.get('AbstractURL'):
|
||
output += f"<a href='{data['AbstractURL']}'>Read more on {data.get('AbstractSource', 'Wikipedia')}</a><br>"
|
||
|
||
elif data.get('Answer'):
|
||
# Direct answer
|
||
output += f"<strong>💡 Answer</strong><br>"
|
||
output += f"{html.escape(data['Answer'])}<br>"
|
||
|
||
elif data.get('Definition'):
|
||
# Definition
|
||
output += f"<strong>📖 Definition</strong><br>"
|
||
output += f"{html.escape(data['Definition'])}<br>"
|
||
if data.get('DefinitionSource'):
|
||
output += f"<em>Source: {data['DefinitionSource']}</em><br>"
|
||
|
||
elif data.get('Results'):
|
||
# List of results
|
||
output += f"<strong>🔍 Results</strong><br>"
|
||
for result in data['Results'][:3]:
|
||
output += f"• <a href='{result.get('FirstURL', '#')}'>{html.escape(result.get('Text', 'Result'))}</a><br>"
|
||
|
||
elif data.get('RelatedTopics'):
|
||
# Related topics
|
||
output += f"<strong>🔗 Related Topics</strong><br>"
|
||
for topic in data['RelatedTopics'][:3]:
|
||
if isinstance(topic, dict) and topic.get('FirstURL'):
|
||
output += f"• <a href='{topic['FirstURL']}'>{html.escape(topic.get('Text', 'Topic'))}</a><br>"
|
||
elif isinstance(topic, dict) and topic.get('Name'):
|
||
output += f"• {html.escape(topic['Name'])}<br>"
|
||
|
||
else:
|
||
# No instant answer found, show search results
|
||
output += "<strong>🔍 No instant answer found.</strong><br>"
|
||
|
||
# Add search link
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}"
|
||
output += f"<br><a href='{search_url}'>View all results on DuckDuckGo</a>"
|
||
|
||
await bot.api.send_markdown_message(room.room_id, output)
|
||
|
||
except Exception as e:
|
||
# Fallback to direct search link
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🦆 DuckDuckGo: {html.escape(query)}</strong><br><br>"
|
||
f"Error accessing API. <a href='{search_url}'>Search on DuckDuckGo</a>"
|
||
)
|
||
logging.error(f"Error in ddg_instant_answer: {e}")
|
||
|
||
async def ddg_search(room, bot, query):
|
||
"""Perform web search with multiple results."""
|
||
try:
|
||
await ddg_web_search(room, bot, query, limit=5)
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id, f"Error performing search: {str(e)}")
|
||
|
||
async def ddg_web_search(room, bot, query, limit=5):
|
||
"""Perform web search and return results."""
|
||
try:
|
||
params = {
|
||
'q': query,
|
||
'format': 'json'
|
||
}
|
||
|
||
response = requests.get(DDG_API_URL, params=params, timeout=10)
|
||
|
||
if response.status_code != 200:
|
||
# Fallback to direct search
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🔍 DuckDuckGo Search: {html.escape(query)}</strong><br><br>"
|
||
f"API temporarily unavailable. <a href='{search_url}'>Search on DuckDuckGo</a>"
|
||
)
|
||
return
|
||
|
||
data = response.json()
|
||
|
||
output = f"<strong>🔍 DuckDuckGo Search: {html.escape(query)}</strong><br><br>"
|
||
|
||
results_shown = 0
|
||
|
||
# Show instant answer if available
|
||
if data.get('AbstractText') and results_shown < limit:
|
||
output += f"<strong>💡 {data.get('Heading', 'Instant Answer')}</strong><br>"
|
||
abstract = data['AbstractText'][:200] + "..." if len(data['AbstractText']) > 200 else data['AbstractText']
|
||
output += f"{html.escape(abstract)}<br>"
|
||
if data.get('AbstractURL'):
|
||
output += f"<a href='{data['AbstractURL']}'>Read more</a><br>"
|
||
output += "<br>"
|
||
results_shown += 1
|
||
|
||
# Show web results
|
||
if data.get('Results') and results_shown < limit:
|
||
output += "<strong>🌐 Web Results</strong><br>"
|
||
for result in data['Results'][:limit - results_shown]:
|
||
output += f"• <a href='{result.get('FirstURL', '#')}'>{html.escape(result.get('Text', 'Result'))}</a><br>"
|
||
results_shown += 1
|
||
|
||
# Show related topics
|
||
if data.get('RelatedTopics') and results_shown < limit:
|
||
output += "<strong>🔗 Related Topics</strong><br>"
|
||
for topic in data['RelatedTopics'][:limit - results_shown]:
|
||
if isinstance(topic, dict) and topic.get('FirstURL'):
|
||
output += f"• <a href='{topic['FirstURL']}'>{html.escape(topic.get('Text', 'Topic'))}</a><br>"
|
||
results_shown += 1
|
||
|
||
# Add search link
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}"
|
||
output += f"<br><a href='{search_url}'>View all results on DuckDuckGo</a>"
|
||
|
||
await bot.api.send_markdown_message(room.room_id, output)
|
||
|
||
except Exception as e:
|
||
# Fallback to direct search
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🔍 DuckDuckGo Search: {html.escape(query)}</strong><br><br>"
|
||
f"Error accessing API. <a href='{search_url}'>Search on DuckDuckGo</a>"
|
||
)
|
||
logging.error(f"Error in ddg_web_search: {e}")
|
||
|
||
async def ddg_image_search(room, bot, query):
|
||
"""Perform image search."""
|
||
try:
|
||
params = {
|
||
'q': query,
|
||
'format': 'json',
|
||
'iax': 'images',
|
||
'ia': 'images'
|
||
}
|
||
|
||
response = requests.get(DDG_API_URL, params=params, timeout=10)
|
||
|
||
if response.status_code != 200:
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}&iax=images&ia=images"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🖼️ DuckDuckGo Images: {html.escape(query)}</strong><br><br>"
|
||
f"API temporarily unavailable. <a href='{search_url}'>Search images on DuckDuckGo</a>"
|
||
)
|
||
return
|
||
|
||
data = response.json()
|
||
|
||
output = f"<strong>🖼️ DuckDuckGo Images: {html.escape(query)}</strong><br><br>"
|
||
|
||
if data.get('Results'):
|
||
output += "<strong>📸 Image Results</strong><br>"
|
||
for image in data['Results'][:3]:
|
||
output += f"• <a href='{image.get('Image', '#')}'>{html.escape(image.get('Title', 'Image'))}</a><br>"
|
||
if image.get('Width') and image.get('Height'):
|
||
output += f" Size: {image['Width']}×{image['Height']}<br>"
|
||
else:
|
||
output += "No image results found.<br>"
|
||
|
||
# Add search link
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}&iax=images&ia=images"
|
||
output += f"<br><a href='{search_url}'>View all images on DuckDuckGo</a>"
|
||
|
||
await bot.api.send_markdown_message(room.room_id, output)
|
||
|
||
except Exception as e:
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}&iax=images&ia=images"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🖼️ DuckDuckGo Images: {html.escape(query)}</strong><br><br>"
|
||
f"Error accessing API. <a href='{search_url}'>Search images on DuckDuckGo</a>"
|
||
)
|
||
|
||
async def ddg_news_search(room, bot, query):
|
||
"""Perform news search."""
|
||
try:
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}&iar=news"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>📰 DuckDuckGo News: {html.escape(query)}</strong><br><br>"
|
||
f"<a href='{search_url}'>View news on DuckDuckGo</a>"
|
||
)
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id, f"Error performing news search: {str(e)}")
|
||
|
||
async def ddg_video_search(room, bot, query):
|
||
"""Perform video search."""
|
||
try:
|
||
search_url = f"https://duckduckgo.com/?q={quote(query)}&iar=videos"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🎬 DuckDuckGo Videos: {html.escape(query)}</strong><br><br>"
|
||
f"<a href='{search_url}'>View videos on DuckDuckGo</a>"
|
||
)
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id, f"Error performing video search: {str(e)}")
|
||
|
||
async def ddg_bang_search(room, bot, bang_query):
|
||
"""Perform search using DuckDuckGo bangs."""
|
||
try:
|
||
# Create search URL directly - this is more reliable than API for bangs
|
||
search_url = f"https://duckduckgo.com/?q={quote(bang_query)}"
|
||
|
||
# Common bangs with descriptions
|
||
bang_descriptions = {
|
||
'!w': 'Wikipedia',
|
||
'!g': 'Google',
|
||
'!yt': 'YouTube',
|
||
'!aw': 'ArchWiki',
|
||
'!gh': 'GitHub',
|
||
'!so': 'Stack Overflow',
|
||
'!amazon': 'Amazon',
|
||
'!imdb': 'IMDb',
|
||
'!reddit': 'Reddit',
|
||
'!tw': 'Twitter'
|
||
}
|
||
|
||
# Extract bang for description
|
||
bang = bang_query.split(' ')[0] if ' ' in bang_query else bang_query
|
||
description = bang_descriptions.get(bang, 'Site-specific search')
|
||
|
||
output = f"<strong>🎯 DuckDuckGo Bang: {html.escape(bang)}</strong><br>"
|
||
output += f"<strong>Description:</strong> {description}<br>"
|
||
|
||
if ' ' in bang_query:
|
||
output += f"<strong>Query:</strong> {html.escape(bang_query.split(' ', 1)[1])}<br><br>"
|
||
|
||
output += f"<a href='{search_url}'>Search with {html.escape(bang)} on DuckDuckGo</a>"
|
||
|
||
await bot.api.send_markdown_message(room.room_id, output)
|
||
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id, f"Error with bang search: {str(e)}")
|
||
|
||
async def ddg_definition(room, bot, word):
|
||
"""Get word definition."""
|
||
try:
|
||
search_url = f"https://duckduckgo.com/?q=define+{quote(word)}"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>📖 Definition: {html.escape(word)}</strong><br><br>"
|
||
f"<a href='{search_url}'>Get definition on DuckDuckGo</a>"
|
||
)
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id, f"Error getting definition: {str(e)}")
|
||
|
||
async def ddg_calculator(room, bot, expression):
|
||
"""Use DuckDuckGo as a calculator."""
|
||
try:
|
||
search_url = f"https://duckduckgo.com/?q={quote(expression)}"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🧮 Calculator: {html.escape(expression)}</strong><br><br>"
|
||
f"<a href='{search_url}'>Calculate on DuckDuckGo</a>"
|
||
)
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id, f"Error with calculator: {str(e)}")
|
||
|
||
async def ddg_weather(room, bot, location):
|
||
"""Get weather information."""
|
||
try:
|
||
if not location:
|
||
location = "current location"
|
||
|
||
search_url = f"https://duckduckgo.com/?q=weather+{quote(location)}"
|
||
await bot.api.send_markdown_message(
|
||
room.room_id,
|
||
f"<strong>🌤️ Weather: {html.escape(location)}</strong><br><br>"
|
||
f"<a href='{search_url}'>Get weather on DuckDuckGo</a>"
|
||
)
|
||
except Exception as e:
|
||
await bot.api.send_text_message(room.room_id, f"Error getting weather: {str(e)}")
|