Modernise text responses

This commit is contained in:
Nexus 2023-05-04 18:03:28 +01:00
parent c418f002ab
commit 2d52addfd6
Signed by: nex
GPG key ID: 0FA334385D0B689F
2 changed files with 278 additions and 229 deletions

View file

@ -1,4 +1,5 @@
import hashlib import hashlib
import inspect
import io import io
import json import json
import os import os
@ -7,11 +8,12 @@ import re
import asyncio import asyncio
import textwrap import textwrap
import subprocess import subprocess
import traceback
import warnings import warnings
from datetime import datetime, timezone, timedelta from datetime import datetime, timezone, timedelta
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from pathlib import Path from pathlib import Path
from typing import Optional, Tuple from typing import Optional, Tuple, Dict, Any
import discord import discord
import httpx import httpx
from discord.ext import commands, pages, tasks from discord.ext import commands, pages, tasks
@ -217,6 +219,145 @@ class Events(commands.Cog):
@commands.Cog.listener() @commands.Cog.listener()
async def on_message(self, message: discord.Message): async def on_message(self, message: discord.Message):
async def it_just_works():
_file = Path.cwd() / "assets" / "it-just-works.ogg"
if message.author.voice is not None and message.author.voice.channel is not None:
voice: discord.VoiceClient | None = None
if message.guild.voice_client is not None:
# noinspection PyUnresolvedReferences
if message.guild.voice_client.is_playing():
return
try:
await _dc(message.guild.voice_client)
except discord.HTTPException:
pass
try:
voice = await message.author.voice.channel.connect(timeout=10, reconnect=False)
except asyncio.TimeoutError:
await message.channel.trigger_typing()
await message.reply(
"I'd play the song but discord's voice servers are shit.",
file=discord.File(_file)
)
region = message.author.voice.channel.rtc_region
# noinspection PyUnresolvedReferences
console.log(
"Timed out connecting to voice channel: {0.name} in {0.guild.name} "
"(region {1})".format(
message.author.voice.channel,
region.name if region else "auto (unknown)"
)
)
return
if voice.channel != message.author.voice.channel:
await voice.move_to(message.author.voice.channel)
if message.guild.me.voice.self_mute or message.guild.me.voice.mute:
await _dc(voice)
await message.channel.trigger_typing()
await message.reply("Unmute me >:(", file=discord.File(_file))
else:
def after(err):
asyncio.run_coroutine_threadsafe(
_dc(voice),
self.bot.loop
)
if err is not None:
console.log(f"Error playing audio: {err}")
# noinspection PyTypeChecker
src = discord.FFmpegPCMAudio(str(_file.resolve()), stderr=subprocess.DEVNULL)
src = discord.PCMVolumeTransformer(src, volume=0.5)
voice.play(
src,
after=after
)
else:
await message.channel.trigger_typing()
await message.reply(file=discord.File(_file))
async def send_smeg():
directory = Path.cwd() / "assets" / "smeg"
if directory:
choice = random.choice(list(directory.iterdir()))
_file = discord.File(
choice,
filename="%s.%s" % (os.urandom(32).hex(), choice.suffix)
)
await message.reply(file=_file)
async def send_what():
msg = message.reference.cached_message
if not msg:
try:
msg = await message.channel.fetch_message(message.reference.message_id)
except discord.HTTPException:
return
if msg.content.count(f"{self.bot.user.mention} said ") >= 2:
await message.reply("You really are deaf, aren't you.")
elif not msg.content:
await message.reply(
"Maybe *I* need to get my hearing checked, I have no idea what {} said.".format(
msg.author.mention
)
)
else:
text = "{0.author.mention} said '{0.content}', you deaf sod.".format(
msg
)
_content = textwrap.shorten(
text, width=2000, placeholder="[...]"
)
await message.reply(_content, allowed_mentions=discord.AllowedMentions.none())
async def send_fuck_you():
student = await get_or_none(Student, user_id=message.author.id)
if student is None:
return await message.reply("You aren't even verified...", delete_after=10)
elif student.ip_info is None:
if OAUTH_REDIRECT_URI:
return await message.reply(
f"Let me see who you are, and then we'll talk... <{OAUTH_REDIRECT_URI}>",
delete_after=30
)
else:
return await message.reply(
"I literally don't even know who you are...",
delete_after=10
)
else:
ip = student.ip_info
is_proxy = ip.get("proxy")
if is_proxy is None:
is_proxy = "?"
else:
is_proxy = "\N{WHITE HEAVY CHECK MARK}" if is_proxy else "\N{CROSS MARK}"
is_hosting = ip.get("hosting")
if is_hosting is None:
is_hosting = "?"
else:
is_hosting = "\N{WHITE HEAVY CHECK MARK}" if is_hosting else "\N{CROSS MARK}"
return await message.reply(
"Nice argument, however,\n"
"IP: {0[query]}\n"
"ISP: {0[isp]}\n"
"Latitude: {0[lat]}\n"
"Longitude: {0[lon]}\n"
"Proxy server: {1}\n"
"VPS (or other hosting) provider: {2}\n\n"
"\N{smiling face with sunglasses}".format(
ip,
is_proxy,
is_hosting
),
delete_after=30
)
if not message.guild: if not message.guild:
return return
@ -233,241 +374,143 @@ class Events(commands.Cog):
await message.delete(delay=1) await message.delete(delay=1)
else: else:
# Respond to shronk bot assets = Path.cwd() / "assets"
if message.author.id == 1063875884274163732 and message.channel.can_send(): responses: Dict[str | tuple, Dict[str, Any]] = {
if "pissylicious 💦💦" in message.content: r"ferdi": {
from dns import asyncresolver "content": "https://ferdi-is.gay/",
import httpx "delete_after": 15,
response = await asyncresolver.resolve("shronkservz.tk", "A") },
ip_info_response = await httpx.AsyncClient().get(f"http://ip-api.com/json/{response[0].address}") r"\bbee(s)*\b": {
if ip_info_response.status_code == 200: "content": "https://ferdi-is.gay/bee",
return await message.reply( },
f"Scattylicious\N{pile of poo}\N{pile of poo}\n" r"it just works": {
"IP: {0[query]}\n" "func": it_just_works
"ISP: {0[isp]}\n" },
"Latitude: {0[lat]}\n" r"^linux$": {
"Longitude: {0[lon]}\n".format( "content": lambda: (assets / "copypasta.txt").read_text(),
ip_info_response.json(), "meta": {
) "needs_mention": True
) }
RESPONSES = { },
"Congratulations!!": "Shut up SHRoNK Bot, nobody loves you.", r"carat": {
"You run on a Raspberry Pi... I run on a real server": "At least my server gets action, " "file": discord.File(assets / "carat.jpg"),
"while yours just sits and collects dust!" "delete_after": None
},
r"[s5]+(m)+[e3]+[g9]+": {
"func": send_smeg,
"meta": {
"sub": {
r"pattern": r"(-_.\s)+",
r"with": ''
}
}
},
r"(what|huh)(\?|!)*": {
"func": send_what,
"meta": {
"check": lambda: message.reference is not None
}
},
("year", "linux", "desktop"): {
"content": lambda: "%s will be the year of the GNU+Linux desktop." % datetime.now().year,
"delete_after": None
},
r"fuck you(\W)*": {
"func": send_fuck_you,
"meta": {
"check": lambda: message.content.startswith(self.bot.user.mention)
}
} }
for k, v in RESPONSES.items(): }
if k in message.content:
await message.reply("shut up", delete_after=3)
await message.delete(delay=3)
break
# Stop responding to any bots # Stop responding to any bots
if message.author.bot is True: if message.author.bot is True:
return return
if message.channel.can_send() and "ferdi" in message.content.lower():
await message.reply("https://ferdi-is.gay/", delete_after=30)
# Only respond if the message has content... # Only respond if the message has content...
if message.content: if message.content and message.channel.can_send(discord.Embed, discord.File):
if message.channel.can_send(): # ... and we can send messages for key, data in responses.items():
if "it just works" in message.content.lower(): meta = data.pop("meta", {})
file = Path.cwd() / "assets" / "it-just-works.ogg" if meta.get("needs_mention"):
if message.author.voice is not None and message.author.voice.channel is not None: if not self.bot.user.mention not in message.mentions:
voice: discord.VoiceClient | None = None continue
if message.guild.voice_client is not None:
# noinspection PyUnresolvedReferences
if message.guild.voice_client.is_playing():
return
try:
await _dc(message.guild.voice_client)
except discord.HTTPException:
pass
try:
voice = await message.author.voice.channel.connect(timeout=10, reconnect=False)
except asyncio.TimeoutError:
await message.channel.trigger_typing()
await message.reply(
"I'd play the song but discord's voice servers are shit.",
file=discord.File(file)
)
region = message.author.voice.channel.rtc_region
# noinspection PyUnresolvedReferences
console.log(
"Timed out connecting to voice channel: {0.name} in {0.guild.name} "
"(region {1})".format(
message.author.voice.channel,
region.name if region else "auto (unknown)"
)
)
return
if voice.channel != message.author.voice.channel:
await voice.move_to(message.author.voice.channel)
if message.guild.me.voice.self_mute or message.guild.me.voice.mute:
await _dc(voice)
await message.channel.trigger_typing()
await message.reply("Unmute me >:(", file=discord.File(file))
else:
def after(err):
asyncio.run_coroutine_threadsafe(
_dc(voice),
self.bot.loop
)
if err is not None:
console.log(f"Error playing audio: {err}")
# noinspection PyTypeChecker if meta.get("check"):
src = discord.FFmpegPCMAudio(str(file.absolute()), stderr=subprocess.DEVNULL)
src = discord.PCMVolumeTransformer(src, volume=0.5)
voice.play(
src,
after=after
)
else:
await message.channel.trigger_typing()
await message.reply(file=discord.File(file))
if "linux" in message.content.lower() and self.bot.user in message.mentions:
try: try:
with open("./assets/copypasta.txt", "r") as f: okay = meta["check"]()
await message.reply(f.read()) except (Exception, RuntimeError):
except FileNotFoundError: traceback.print_exc()
await message.reply( okay = False
"I'd just like to interject for a moment. What you're referring to as Linux, "
"is in fact, uh... I don't know, I forgot." if not okay:
) continue
if "carat" in message.content.lower(): elif meta.get("checks") and isinstance(meta["checks"], list):
file = discord.File(Path.cwd() / "assets" / "carat.png", filename="carat.png") for check in meta["checks"]:
await message.reply(file=file) try:
smeg_regex = r"[s5]+(m)+[e3]+[g9]+" okay = check()
smeg_sub = r"(-_.\s)+" except (Exception, RuntimeError):
if re.match(smeg_regex, re.sub(smeg_sub, "", message.content.lower())): traceback.print_exc()
directory = Path.cwd() / "assets" / "smeg" okay = False
if directory:
choice = random.choice(list(directory.iterdir())) if not okay:
file = discord.File( break
choice, else:
filename="%s.%s" % (os.urandom(32).hex(), choice.suffix) continue
)
await message.reply(file=file) if meta.get("sub") is not None and isinstance(meta["sub"], dict):
if message.reference is not None and message.reference.cached_message is not None: content = re.sub(
if message.content.lower().strip() in ("what", "what?", "huh", "huh?", "?"): meta["sub"]["pattern"],
if f"{message.author.mention} said '" * 3 in message.reference.cached_message.content: meta["sub"]["with"],
await message.reply("You really are deaf aren't you.") message.content
elif not message.reference.cached_message.content: )
await message.reply( else:
"Maybe *I* need to get my hearing checked, I have no idea what {} said.".format( content = message.content
message.reference.cached_message.author.mention
) if isinstance(key, str):
) regex = re.compile(key, re.IGNORECASE)
if not regex.search(content):
continue
elif isinstance(key, tuple):
if not all(k in content for k in key):
continue
if "func" in data:
try:
if inspect.iscoroutinefunction(data["func"]):
await data["func"]()
break
else: else:
text = "{0.author.mention} said %r, you deaf sod.".format( data["func"]()
message.reference.cached_message break
) except (Exception, RuntimeError):
_content = textwrap.shorten( traceback.print_exc()
text % message.reference.cached_message.content, width=2000, placeholder="[...]" continue
) else:
await message.reply(_content) for k, v in data.copy().items():
await self.process_message_for_github_links(message) if callable(v):
data[k] = v()
await message.reply(**data)
break
await self.process_message_for_github_links(message)
T_EMOJI = "\U0001f3f3\U0000fe0f\U0000200d\U000026a7\U0000fe0f"
G_EMOJI = "\U0001f3f3\U0000fe0f\U0000200d\U0001f308"
N_EMOJI = "\U0001f922"
C_EMOJI = "\U0000271d\U0000fe0f"
reactions = {
r"mpreg|lupupa|\U0001fac3": "\U0001fac3", # mpreg
r"(trans(gender)?($|\W+)|%s)" % T_EMOJI: T_EMOJI, # trans
r"gay|%s" % G_EMOJI: G_EMOJI,
r"(femboy|trans(gender)?($|\W+))": C_EMOJI
}
if message.channel.permissions_for(message.guild.me).add_reactions: if message.channel.permissions_for(message.guild.me).add_reactions:
if "mpreg" in message.content.lower() or "\U0001fac3" in message.content.lower():
try:
await message.add_reaction("\U0001fac3")
except discord.HTTPException as e:
console.log("Failed to add mpreg reaction:", e)
if "lupupa" in message.content.lower():
try:
await message.add_reaction("\U0001fac3")
except discord.HTTPException as e:
console.log("Failed to add mpreg reaction:", e)
is_naus = random.randint(1, 100) == 32 is_naus = random.randint(1, 100) == 32
if self.bot.user in message.mentions or message.channel.id == 1032974266527907901 or is_naus: for key, value in reactions.items():
T_EMOJI = "\U0001f3f3\U0000fe0f\U0000200d\U000026a7\U0000fe0f" if re.search(key, message.content, re.IGNORECASE):
G_EMOJI = "\U0001f3f3\U0000fe0f\U0000200d\U0001f308" await message.add_reaction(value)
N_EMOJI = "\U0001f922"
C_EMOJI = "\U0000271d\U0000fe0f"
if any((x in message.content.lower() for x in ("trans", T_EMOJI, "femboy"))) or is_naus:
try:
await message.add_reaction(N_EMOJI)
except discord.HTTPException as e:
console.log("Failed to add trans reaction:", e)
if "gay" in message.content.lower() or G_EMOJI in message.content.lower():
try:
await message.add_reaction(C_EMOJI)
except discord.HTTPException as e:
console.log("Failed to add gay reaction:", e)
if all(x in message.content.lower() for x in ("year", "linux", "desktop")): if is_naus:
date = discord.utils.utcnow() await message.add_reaction(N_EMOJI)
# date = date.replace(year=date.year + 1)
return await message.reply(date.strftime("%Y") + " will be the year of the GNU+Linux desktop.")
if self.bot.user in message.mentions:
if message.content.startswith(self.bot.user.mention):
if message.content.lower().endswith("bot"):
pos, neut, neg, _ = await self.analyse_text(message.content)
if pos > neg:
embed = discord.Embed(description=":D", color=discord.Color.green())
embed.set_footer(
text=f"Pos: {pos*100:.2f}% | Neutral: {neut*100:.2f}% | Neg: {neg*100:.2f}%"
)
elif pos == neg:
embed = discord.Embed(description=":|", color=discord.Color.greyple())
embed.set_footer(
text=f"Pos: {pos * 100:.2f}% | Neutral: {neut * 100:.2f}% | Neg: {neg * 100:.2f}%"
)
else:
embed = discord.Embed(description=":(", color=discord.Color.red())
embed.set_footer(
text=f"Pos: {pos*100:.2f}% | Neutral: {neut*100:.2f}% | Neg: {neg*100:.2f}%"
)
return await message.reply(embed=embed)
if message.content.lower().endswith("fuck you"):
student = await get_or_none(Student, user_id=message.author.id)
if student is None:
return await message.reply("You aren't even verified...", delete_after=10)
elif student.ip_info is None:
if OAUTH_REDIRECT_URI:
return await message.reply(
f"Let me see who you are, and then we'll talk... <{OAUTH_REDIRECT_URI}>",
delete_after=30
)
else:
return await message.reply(
"I literally don't even know who you are...",
delete_after=10
)
else:
ip = student.ip_info
is_proxy = ip.get("proxy")
if is_proxy is None:
is_proxy = "?"
else:
is_proxy = "\N{WHITE HEAVY CHECK MARK}" if is_proxy else "\N{CROSS MARK}"
is_hosting = ip.get("hosting")
if is_hosting is None:
is_hosting = "?"
else:
is_hosting = "\N{WHITE HEAVY CHECK MARK}" if is_hosting else "\N{CROSS MARK}"
return await message.reply(
"Nice argument, however,\n"
"IP: {0[query]}\n"
"ISP: {0[isp]}\n"
"Latitude: {0[lat]}\n"
"Longitude: {0[lon]}\n"
"Proxy server: {1}\n"
"VPS (or other hosting) provider: {2}\n\n"
"\N{smiling face with sunglasses}".format(
ip,
is_proxy,
is_hosting
),
delete_after=30
)
@tasks.loop(minutes=10) @tasks.loop(minutes=10)
async def fetch_discord_atom_feed(self): async def fetch_discord_atom_feed(self):

View file

@ -20,15 +20,16 @@ class Bot(commands.Bot):
if TYPE_CHECKING: if TYPE_CHECKING:
web: Optional[Dict[str, Union[Server, Config, Task]]] web: Optional[Dict[str, Union[Server, Config, Task]]]
def __init__(self, intents: discord.Intents, guilds: list[int], extensions: list[str]): def __init__(self, intents: discord.Intents, guilds: list[int], extensions: list[str], prefixes: list[str]):
from .db import registry from .db import registry
from .console import console from .console import console
super().__init__( super().__init__(
command_prefix=commands.when_mentioned_or("h!", "r!"), command_prefix=commands.when_mentioned_or(*prefixes),
debug_guilds=guilds, debug_guilds=guilds,
allowed_mentions=discord.AllowedMentions.none(), allowed_mentions=discord.AllowedMentions.none(),
intents=intents, intents=intents,
max_messages=5000 max_messages=5000,
case_insensitive=True,
) )
self.loop.run_until_complete(registry.create_all()) self.loop.run_until_complete(registry.create_all())
self.training_lock = Lock() self.training_lock = Lock()
@ -103,5 +104,10 @@ except ImportError:
"cogs.voice" "cogs.voice"
] ]
try:
from config import prefixes as _prefixes
except ImportError:
_prefixes = ("h!", "r!")
bot = Bot(_intents, config.guilds, _extensions)
bot = Bot(_intents, config.guilds, _extensions, list(_prefixes))