dragon-bot/app/cogs/actual_utils.py
Luke R 8b44daa832
All checks were successful
Build and push / changes (push) Successful in 4s
Build and push / Lint-Python (push) Successful in 2s
Build and push / Build-and-Push-Docker (push) Successful in 3m12s
Build and push / sync-argocd-app (push) Successful in 2s
Build and push / post-failure-to-discord (push) Has been skipped
Build and push / post-success-to-discord (push) Successful in 2s
Adding fstrings and also a message if ollama is off
2025-04-22 12:38:32 -07:00

359 lines
12 KiB
Python
Executable File

from discord import option
from discord.ext import commands
import discord
import os
import requests
import socket
import tempfile
class ActualUtils(commands.Cog):
def __init__(self, bot):
self.bot: commands.Bot = bot
@commands.slash_command(
guild_ids=None,
name="translate",
description="Translate a string",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
"query",
description="What you're tryna translate",
requred=True,
)
async def translate(self, ctx, query: str):
from googletrans import Translator
translator = Translator()
result = translator.translate(query, dest="en").text
await ctx.defer()
await ctx.followup.send(result)
@commands.slash_command(
guild_ids=None,
name="youtube",
description="Search youtube for the passed in query",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
"query",
description="The search string you want to enter on youtube",
requred=True,
)
async def youtube(self, ctx, query: str):
import re
from urllib import parse, request
query_string = parse.urlencode({"search_query": query})
html_content = request.urlopen("http://www.youtube.com/results?" + query_string)
search_results = re.findall("\/watch\?v=(.{11})", html_content.read().decode())
result = "https://www.youtube.com/watch?v=" + search_results[0]
await ctx.defer()
await ctx.followup.send(result)
@commands.slash_command(
guild_ids=None,
name="define",
description="Grabs the highest rated definition from urban dictionary",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
name="word",
description="The word to define. Tries UD first then an actual dictionary",
required=True,
)
async def define(self, ctx, word):
import define_word
await ctx.defer()
await ctx.send_followup(embed=define_word.get_definition(word))
@commands.slash_command(
name="tts",
description="Use google's text-to-speech to make an mp3",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
name="input",
description="The text to turn into an mp3",
required=True,
)
async def tts(self, ctx: discord.ApplicationContext, input: str):
import tts
file_path = tts.text_to_speech(input)
await ctx.defer()
await ctx.send_followup(
file=discord.File(
file_path,
filename=f"A Message From {ctx.author.name}.mp3",
)
)
os.remove(file_path)
@commands.slash_command(
guild_ids=None, name="wolfram", description="Send a query to wolfram alpha"
)
@option(
name="query",
required=True,
description="The query you want to pass to wolfram alpha",
)
async def wolfram(self, ctx, query):
import wolframalpha
client = wolframalpha.Client(os.getenv("wolfram_token"))
await ctx.defer()
try:
res = client.query(query)
return await ctx.send_followup(
"You asked: %s\n\n%s" % (query, next(res.results).text)
)
except Exception:
return await ctx.send_followup("Sorry, I'm unable to answer that")
@commands.slash_command(
guild_ids=None,
name="weather",
description="Get current weather for a location",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
name="location",
required=True,
description="City name or location to get weather for",
)
async def weather(self, ctx, location):
api_key = os.getenv("openweather_api_key")
if not api_key:
return await ctx.respond("OpenWeather API key not configured")
await ctx.defer()
# Weather condition to emoji mapping
weather_emojis = {
# Clear
"01d": "☀️", # clear sky (day)
"01n": "🌙", # clear sky (night)
# Clouds
"02d": "", # few clouds (day)
"02n": "☁️", # few clouds (night)
"03d": "☁️", # scattered clouds
"03n": "☁️", # scattered clouds
"04d": "☁️", # broken clouds
"04n": "☁️", # broken clouds
# Rain
"09d": "🌧️", # shower rain
"09n": "🌧️", # shower rain
"10d": "🌦️", # rain (day)
"10n": "🌧️", # rain (night)
# Thunderstorm
"11d": "⛈️", # thunderstorm
"11n": "⛈️", # thunderstorm
# Snow
"13d": "❄️", # snow
"13n": "❄️", # snow
# Mist/Fog
"50d": "🌫️", # mist
"50n": "🌫️", # mist
}
# Default emoji if condition code not found
default_emoji = "🌡️"
try:
# Get weather data from OpenWeather API
url = f"https://api.openweathermap.org/data/2.5/weather?q={location}&appid={api_key}&units=imperial"
response = requests.get(url)
data = response.json()
if response.status_code != 200:
return await ctx.send_followup(
f"Error: {data.get('message', 'Unknown error')}"
)
# Extract weather information
weather_main = data["weather"][0]["main"]
weather_desc = data["weather"][0]["description"]
weather_icon = data["weather"][0]["icon"]
temp = data["main"]["temp"]
feels_like = data["main"]["feels_like"]
humidity = data["main"]["humidity"]
wind_speed = data["wind"]["speed"]
# Get appropriate emoji
emoji = weather_emojis.get(weather_icon, default_emoji)
# Create embed
embed = discord.Embed(
title=f"Weather in {data['name']}, {data.get('sys', {}).get('country', '')}",
description=f"{emoji} **{weather_main}** - {weather_desc}",
color=discord.Color.blue(),
)
embed.add_field(name="Temperature", value=f"{temp}°F", inline=True)
embed.add_field(name="Feels Like", value=f"{feels_like}°F", inline=True)
embed.add_field(name="Humidity", value=f"{humidity}%", inline=True)
embed.add_field(name="Wind Speed", value=f"{wind_speed} m/s", inline=True)
# Add timestamp
embed.set_footer(text="Data from OpenWeather API")
await ctx.send_followup(embed=embed)
except Exception as e:
await ctx.send_followup(f"Error fetching weather data: {str(e)}")
@commands.slash_command(
guild_ids=None,
name="wifiqr",
description="Generate a qr code for your wifi network",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
name="network_name",
required=True,
description="The name of your network",
)
@option(
name="password", required=True, description="Your wifi password", min_length=8
)
async def wifiqr(self, ctx, network_name, password):
import wifi_qrcode_generator.generator
qr_code = wifi_qrcode_generator.generator.wifi_qrcode(
ssid=network_name,
hidden=False,
authentication_type="WPA",
password=password,
)
await ctx.defer()
try:
file, file_path = tempfile.mkstemp()
qr_code.make_image().save(file_path + ".png")
return await ctx.send_followup(file=discord.File(file_path + ".png"))
os.remove(file_path + ".png")
except Exception as e:
return await ctx.send_followup(f"Something went wrong,\n{e}")
@commands.slash_command(
guld_ids=None,
name="stock",
description="Enter one or many stock tickers to get info about them",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
name="symbols", description="Stocks to look up the price for", required=True
)
@option(
name="verbose",
description="Return extended info about the stocks",
type=bool,
default=False,
)
async def stock(self, ctx, symbols, verbose):
import stock
results = stock.parse_message(symbols, verbose)
await ctx.defer()
for res in results:
await ctx.respond(embed=res)
@commands.slash_command(
guld_ids=None,
name="llm",
description="Send a question to the LLM",
contexts={
discord.InteractionContextType.guild,
discord.InteractionContextType.bot_dm,
discord.InteractionContextType.private_channel,
},
integration_types={
discord.IntegrationType.guild_install,
discord.IntegrationType.user_install,
},
)
@option(
name="question", description="The qustion to send to the LLM", required=True
)
async def send_to_llm(self, ctx, question: str):
import core_utils
# check if its running
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(1)
try:
s.connect((core_utils.ein_ip, int(7869)))
except Exception:
await ctx.respond("The LLM is not running right now")
return
await ctx.defer()
await ctx.send_followup(await core_utils.send_to_llm(ctx, question))
def setup(bot):
bot.add_cog(ActualUtils(bot))