osmarks/autobotrobot
1
0
Fork 0
mirror of https://github.com/osmarks/autobotrobot synced 2025-11-13 13:57:21 +00:00
Code Issues Releases Wiki Activity

I forgot what most of these changes are. Also autopraise mechanism.

Browse Source
This commit is contained in:
osmarks
2025-10-16 08:31:22 +01:00
parent ea7fac4274
commit 31e9a31c61
15 changed files with 117 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
requirements.txt
View File

@@ -1,9 +1,8 @@
# TODO
pytio==0.3.1
aiohttp==3.9.3
aiosqlite==0.19.0
nextcord==2.3.2
numpy==1.26
aiosqlite
discord.py
numpy<2
prometheus-async==19.2.0
prometheus-client==0.15.0
pydot==1.4.2
@@ -11,4 +10,4 @@ toml==0.10.2
requests==2.28.1
python-dateutil==2.8.2
irc==20.1.0
parsedatetime
parsedatetime

6
src/achievement.py
View File

@@ -15,7 +15,7 @@ Achievement = collections.namedtuple("Achievement", ["name", "condition", "descr
achievements = {
"spectre_of_communism": Achievement("Containment Efforts Ongoing", "Refer to the 'spectre of communism' in a message.", "A spectre is haunting Europe. The spectre of communism. Containment efforts are ongoing and full containment is projected by 2036."),
"test": Achievement("Test", "Test achievement. Obtained for testing.", "Great job, you ran the test command!"),
"spam": Achievement("beesbeesbeesbeesbeesbeesbeesbeesbees", "Send a long message containing the same thing repeatedly.", "You should probably not do this, nobody* likes spam!"),
"spam": Achievement("beesbeesbeesbeesbeesbeesbeesbeesbees", "Send a long message containing the same thing repeatedly.", "You should probably not do this; nobody* likes spam!"),
"unicode_abuse": Achievement("Anomalous Unicode", "Send a high proportion of weird Unicode characters in a message.", "h̵͖̻̮̗̹̆͛͆̎ͮͤͫ͛ͦ̓̅ͤ́͢é͒ͧ̌̀ͪ̈͂̈́̉ͣ̅̿̄̌̋̿̽̚͏̛͏͚̯͉̟͇̼̹͎ͅa̠̹̘͎̫̜̞̩͖̟̟͍͇͈͍̝͕͛ͥ͊̾̈́ͩͯͩͭ̆̋͐͗̉͋̓̀͝v͎͖̜͎͔̞͚͉̺̞̘̥͖̝͚̺̍ͤ̌͂ͨ̃̅ͫ̿͛ͯ̓̉̆̎͊̀̚̕͟s̪̠̟̣̝̹̭̻̈́ͤ͗̏ͮ̂ͯ̈́̊ͩ̓̆̌̆͌̽̓̈́̚͢͞e̛̞̙̜̗̰͕͕͎̺͍̭̲̟̭̲̫̬͓ͯ̅̓̆̂̔̃͟r̷̛̮̮͇̳̳̾ͯͮͩ̏͂ͤ̿̽ͧ͒͋́̕ͅͅv̴̠͉̼̮̭̘ͪͯͦ͌́ͯ̒̃̀́̃͜͝ͅe̵̷̢͕̣̻̥̲͓̼͍̱͕̮̯̱̤̹̱̝̎̓̈́̿ͤ̔̍ͭͭ͐ͅŗ̔ͮͯ͂́͏̻͈̱ͅ ̣͇̼͊̄ͫ̆̍̄̀̀̓͊͐͋̌͘͠į̱͔̰̭̫̱̫̊ͪ̅ͥ̈́ͥ̐͌̅ͪ̅ͨ̎̀͘͝s̍͑̌̋̅͌͂ͨͬͯ̇͊҉̛̱̺͕̰͓̗̖̬͡͡ ̥̤̺̖̪̪́ͯͣ̏̅̈ͣ̿̀͠͠͞i̢̛̭̰̻͈̦̣̮̞̤̩̊̌̾͛ͭͦ̆ͮ̃̎ͪ̔ͬ͊̆͂ͫͅn̸̖͚̣̪̩̏ͥ̈́̅ͯ̔͆́ͦ͗͛͒̃̃ͫ͟͜͝͠ȩ̸͎̟̣̞͉̫̗̙̻̯͍̰̣̌ͪͨ͛̆̕͡v̙͙̲͕͔̦̣̺͔̖͉̜̲̩̈̿ͥ̎͊̈́̊ͯͯ͒ͭ̊̀͢i̪͈̣̱̞̥̰̟̣̩̼̻̪̳̤͇̻̹͉͗ͭ͆̆̎̀͑͑̆͋̏̏͊ͣͦ͆ͣ̈́̓͟͢ţ̵̘̫̯͓̻̗͕̘͙̯̞̪̪̲̤̬̜͕ͫ̄̌̓̎͌ͧ̔͟͢ͅa̸̧̭̲̯̳̔́͋̐͂̇ͪ̔̐́̚͢b͐̅̔ͭ͗̊̂̾̀̓ͭͭ͑ͤ̏̐̃ͩͬ҉̞̼̮̤̝̲̳͓̗̤̫̭̝̹̙͘͟͝ļ̷͈̭̖͓̜̬͔̻͔̀̎ͯ͗̐̽̏ͦ̊͗ͧ́͘ͅe̢͍̦̗̬̝̠͔̳̣̯̮̣̹͍͙̞̜ͣ̉͆̊̀̎ͦ͌̂̋̊ͨ͛́"),
"rtfm": Achievement("RTFM", "Tell someone to read the documentation.", "Apparently, people won't do this without prompting half the time."),
"error": Achievement("You broke it", "Cause an internal error in the bot", "I should probably fix this.")
@@ -48,7 +48,7 @@ async def achieve(bot: commands.Bot, message: discord.Message, achievement):
await bot.database.commit()
logging.info("Awarded achievement %s to %s", achievement, message.author.name)
def setup(bot):
async def setup(bot):
@bot.group(name="achievements", aliases=["ach", "achieve", "achievement"], brief="Achieve a wide variety of fun achievements!", help=f"""
Do things and get arbitrary achievements for them!
Note that due to reasons messages for achievements will not be shown except in opted-in servers, although achievements will be gained regardless.
@@ -73,4 +73,4 @@ def setup(bot):
if re.match("spect(re|er).{,20}(communism|☭)", content, re.IGNORECASE): await achieve(bot, msg, "spectre_of_communism")
if re.match(r"^(.+)\1+$", content) and len(content) >= 1950: await achieve(bot, msg, "spam")
if content_len > 30 and (len(re.findall("[\u0300-\u036f\U00040000-\U0010FFFF]", content)) / content_len) > 0.35: await achieve(bot, msg, "unicode_abuse")
if re.match("(RTFM|(read|look|view|use).{,8}(document|man(page|ual)s?|instruction))", content, re.IGNORECASE): await achieve(bot, msg, "rtfm")
if re.match("(RTFM|(read|look|view|use).{,8}(document|man(page|ual)s?|instruction))", content, re.IGNORECASE): await achieve(bot, msg, "rtfm")

6
src/commands.py
View File

@@ -101,7 +101,7 @@ class GeneralCommands(commands.Cog):
await ctx.send(embed=util.error_embed(util.gen_codeblock(result), "Execution failed"))
else:
out = result
if flags.verbose:
if flags.verbose:
debug_block = "\n" + util.gen_codeblock(f"""{debug}\nLanguage: {real_lang}""")
out = out[:2000 - len(debug_block)] + debug_block
else:
@@ -195,5 +195,5 @@ AutoBotRobot is operated by gollark/osmarks.
await ctx.send("\n".join(map(lambda x: f"{x[0]} x{x[1]}", results)))
def setup(bot):
bot.add_cog(GeneralCommands(bot))
async def setup(bot):
await bot.add_cog(GeneralCommands(bot))

6
src/debug.py
View File

@@ -6,7 +6,7 @@ from discord.ext import commands
import util
import eventbus
def setup(bot):
async def setup(bot):
@bot.group(help="Debug/random messing around utilities. Owner-only.")
@commands.check(util.admin_check)
async def magic(ctx):
@@ -76,6 +76,6 @@ def setup(bot):
@magic.command(help="Reload extensions (all or the specified one).")
async def reload_ext(ctx, ext="all"):
if ext == "all":
for ext in util.extensions: bot.reload_extension(ext)
else: bot.reload_extension(ext)
for ext in util.extensions: await bot.reload_extension(ext)
else: await bot.reload_extension(ext)
await ctx.send("Done!")

4
src/esoserver.py
View File

@@ -8,7 +8,7 @@ import metrics
role_transfer_lock = asyncio.Lock()
def setup(bot):
async def setup(bot):
@bot.listen()
async def on_message(message):
if message.guild and message.guild.id == util.config["esoserver"]["id"]:
@@ -26,4 +26,4 @@ def setup(bot):
await user.remove_roles(role, reason="untransfer unrole")
await message.author.add_roles(role, reason="transfer role")
metrics.role_transfers.inc()
return
return

4
src/heavserver.py
View File

@@ -5,11 +5,11 @@ import discord
import metrics
def setup(bot):
async def setup(bot):
@bot.listen()
async def on_member_join(member):
if member.guild and member.guild.id == util.config["heavserver"]["id"]:
logging.info("%s (%d) joined heavserver", member.display_name, member.id)
if member.bot:
await member.add_roles(discord.utils.get(member.guild.roles, id=util.config["heavserver"]["quarantine_role"]))
await member.add_roles(discord.utils.get(member.guild.roles, id=random.choice(util.config["heavserver"]["moderator_roles"][:])))
await member.add_roles(discord.utils.get(member.guild.roles, id=random.choice(util.config["heavserver"]["moderator_roles"][:])))

4
src/irc_link.py
View File

@@ -113,10 +113,10 @@ async def initialize():
global unlisten
unlisten = eventbus.add_listener(util.config["irc"]["name"], on_bridge_message)
def setup(bot):
async def setup(bot):
asyncio.create_task(initialize())
def teardown(bot=None):
async def teardown(bot=None):
if global_conn:
global_conn.planned_disconnection = True
global_conn.disconnect()

4
src/main.py
View File

@@ -74,7 +74,7 @@ async def on_command_error(ctx, err):
@bot.check
async def andrew_bad(ctx):
return ctx.message.author.id != 543131534685765673
return ctx.message.author.id != 543131534685765673 or ctx.message.author.id != 739032871087374408 or ctx.message.author.id in config.get("bans", [])
@bot.event
async def on_ready():
@@ -108,7 +108,7 @@ async def run_bot():
await eventbus.initial_load(bot.database)
for ext in util.extensions:
logging.info("Loaded %s", ext)
bot.load_extension(ext)
await bot.load_extension(ext)
asyncio.create_task(autogollark.run_bot())
await bot.start(config["token"])

10
src/reminders.py
View File

@@ -71,7 +71,7 @@ class Reminders(commands.Cog):
await ctx.send(embed=util.error_embed("Invalid time (wrong format/too large months or years)"))
return
utc_time, local_time = util.in_timezone(time, tz)
id = (await self.bot.database.execute_insert("INSERT INTO reminders (remind_timestamp, created_timestamp, reminder, expired, extra) VALUES (?, ?, ?, ?, ?)",
id = (await self.bot.database.execute_insert("INSERT INTO reminders (remind_timestamp, created_timestamp, reminder, expired, extra) VALUES (?, ?, ?, ?, ?)",
(utc_time.timestamp(), now.timestamp(), reminder, 0, util.json_encode(extra_data))))["last_insert_rowid()"]
await self.bot.database.commit()
await ctx.send(f"Reminder scheduled for {util.format_time(local_time)} ({util.format_timedelta(now, utc_time)}).")
@@ -178,10 +178,10 @@ class Reminders(commands.Cog):
self.reminder_queue.pop(0)
await self.fire_reminder(next_id)
def setup(bot):
async def setup(bot):
cog = Reminders(bot)
await bot.add_cog(cog)
asyncio.create_task(cog.init_reminders())
bot.add_cog(cog)
def teardown(bot):
bot.rloop_task.cancel()
async def teardown(bot):
bot.rloop_task.cancel()

12
src/search.py
View File

@@ -18,7 +18,7 @@ class Parser(html.parser.HTMLParser):
attrs = dict(attrs)
if tag == "a" and attrs.get("class") == "result__a" and "https://duckduckgo.com/y.js?ad_provider" not in attrs["href"]:
self.links.append(attrs["href"])
class Search(commands.Cog):
def __init__(self, bot):
self.bot = bot
@@ -43,7 +43,7 @@ class Search(commands.Cog):
return await ctx.send(p.links[0], reference=ctx.message)
except IndexError:
return await ctx.send("No results.", reference=ctx.message)
async def wp_search(self, query):
async with self.session.get("https://en.wikipedia.org/w/api.php",
params={ "action": "query", "list": "search", "srsearch": query, "utf8": "1", "format": "json", "srlimit": 1 }) as resp:
@@ -62,10 +62,10 @@ class Search(commands.Cog):
return await self.wp_fetch(new_page, fallback=False)
if page in self.wp_cache: return self.wp_cache[page]
if page in self.wp_search_cache:
if page in self.wp_search_cache:
if self.wp_search_cache[page] is None: return None
return await self.wp_fetch(self.wp_search_cache[page], fallback=False)
async with self.session.get("https://en.wikipedia.org/w/api.php",
async with self.session.get("https://en.wikipedia.org/w/api.php",
params={ "action": "query", "format": "json", "titles": page, "prop": "extracts", "exintro": 1, "explaintext": 1 }) as resp:
data = (await resp.json())["query"]
if "-1" in data["pages"]:
@@ -94,6 +94,6 @@ class Search(commands.Cog):
if self.pool is not None:
self.pool.shutdown()
def setup(bot):
async def setup(bot):
cog = Search(bot)
bot.add_cog(cog)
await bot.add_cog(cog)

75
src/sentience.py
View File

@@ -1,16 +1,14 @@
import asyncio
import argparse
import random
from numpy.random import default_rng
import re
import aiohttp
import subprocess
from collections import defaultdict, deque
import discord.ext.commands as commands
import discord
from datetime import datetime
from datetime import datetime, timedelta, timezone
from pathlib import Path
import asyncio
import logging
import tio
import util
cleaner = commands.clean_content()
@@ -20,7 +18,11 @@ def clean(ctx, text):
class Sentience(commands.Cog):
def __init__(self, bot):
self.bot = bot
self.timeouts = {}
self.session = aiohttp.ClientSession()
self.autopraise_spontaneous_times = {}
self.autopraise_triggered_times = {}
self.praise_context_buffers = defaultdict(deque)
async def serialize_history(self, ctx, n=20):
PREFIXES = [ ctx.prefix + "ai", ctx.prefix + "ag", ctx.prefix + "autogollark", ctx.prefix + "gollark" ]
@@ -52,11 +54,19 @@ class Sentience(commands.Cog):
@commands.command(help="Highly advanced AI Assistant.")
async def ai(self, ctx, *, query=None):
if timeout := self.timeouts.get(ctx.channel.id):
if timeout > datetime.now():
return
prompt = await self.serialize_history(ctx)
prompt.append(f'[{util.render_time(datetime.utcnow())}] {util.config["ai"]["own_name"]}:')
prompt.append(f'[{util.render_time(datetime.now(timezone.utc))}] {util.config["ai"]["own_name"]}:')
generation = await util.generate(self.session, util.config["ai"]["prompt_start"] + "".join(prompt))
if generation.strip():
await ctx.send(generation.strip())
assert generation, "backend failed"
generation = generation.strip()
if generation:
await ctx.send(generation)
if generation.endswith("/quit"):
await ctx.send("Disconnecting AI as requested.")
self.timeouts[ctx.channel.id] = datetime.now() + timedelta(seconds=1200)
@commands.command(help="Search meme library.", aliases=["memes"])
async def meme(self, ctx, *, query=None):
@@ -74,5 +84,46 @@ class Sentience(commands.Cog):
o_files = [ discord.File(Path(util.config["memetics"]["memes_local"]) / util.meme_thumbnail(results, m)) for m in mat ]
await ctx.send(files=o_files)
def setup(bot):
bot.add_cog(Sentience(bot))
async def spontaneous_praise(self, target, delay):
await asyncio.sleep(delay)
del self.autopraise_spontaneous_times[target["user"]]
await self.praise(target, target["spontaneous_channel"], util.config["autopraise"]["spontaneous_prompt"])
async def praise(self, target, channel, prompt):
chan = self.bot.get_channel(channel)
if chan:
context = "\n".join(self.praise_context_buffers[target["user"]])
praise_message = await util.generate_raw_chatcompletion(self.session, util.config["ai"]["chat_completions"], prompt + "\n" + context)
praise_message = praise_message.strip()
if praise_message and praise_message != util.config["autopraise"]["no_praise"]:
await chan.send(praise_message)
else:
# if no praise occurred, reset the timer
del self.autopraise_triggered_times[target["user"]]
@commands.Cog.listener("on_message")
async def auto_praise(self, msg):
now = util.timestamp()
# if anyone uses this, rearrange to dict users → spec, for efficiency
for target in util.config["autopraise"]["targets"]:
if target["guild"] == msg.guild.id and target["user"] == msg.author.id:
if msg.channel.id in target["channels"]:
if msg.content and msg.content.strip(): self.praise_context_buffers[msg.author.id].append(f"{msg.author.name}: {msg.content.strip()}")
if len(self.praise_context_buffers[msg.author.id]) >= target["context_length"]:
self.praise_context_buffers[msg.author.id].popleft()
# no spontaneous praise event within window: dispatch
if msg.author.id not in self.autopraise_spontaneous_times:
logging.info("Scheduling spontaneous praise for %d", msg.author.id)
spontaneous_praise_delay = random.expovariate(target["spontaneous_interval"] / 2) + target["spontaneous_interval"] / 2
self.autopraise_spontaneous_times[msg.author.id] = now + spontaneous_praise_delay
asyncio.create_task(self.spontaneous_praise(target, spontaneous_praise_delay))
may_praise_at = self.autopraise_triggered_times.get(msg.author.id)
if may_praise_at is None or may_praise_at < now:
logging.info("Triggered praise event for %d", msg.author.id)
self.autopraise_triggered_times[msg.author.id] = now + target["triggered_interval"]
await self.praise(target, msg.channel.id, util.config["autopraise"]["triggered_prompt"])
async def setup(bot):
await bot.add_cog(Sentience(bot))

8
src/telephone.py
View File

@@ -135,7 +135,7 @@ class Telephone(commands.Cog):
reply = (eventbus.AuthorInfo(replying_to.author.name, replying_to.author.id, str(replying_to.author.display_avatar.url), replying_to.author.bot), parse_formatting(self.bot, replying_to.content))
else:
reply = (None, None)
msg = eventbus.Message(eventbus.AuthorInfo(msg.author.name, msg.author.id, str(msg.author.display_avatar.url), msg.author.bot),
msg = eventbus.Message(eventbus.AuthorInfo(msg.author.name, msg.author.id, str(msg.author.display_avatar.url), msg.author.bot),
parse_formatting(self.bot, msg.content), ("discord", channel_id), msg.id, [ at for at in msg.attachments if not at.is_spoiler() ], reply=reply)
await eventbus.push(msg)
@@ -315,7 +315,7 @@ When you want to end a call, use hangup.
return await ctx.send(embed=util.error_embed("Target channel no longer exists."))
_, call_message = await asyncio.gather(
ctx.send(embed=util.info_embed("Outgoing call", f"Dialing {address}...")),
recv_channel.send(embed=util.info_embed("Incoming call",
recv_channel.send(embed=util.info_embed("Incoming call",
f"Call from {originating_address}. Click :white_check_mark: to accept or :negative_squared_cross_mark: to decline."))
)
# add clickable reactions to it
@@ -429,7 +429,7 @@ When you want to end a call, use hangup.
finally:
os.unlink(tmppath)
def setup(bot):
async def setup(bot):
cog = Telephone(bot)
bot.add_cog(cog)
await bot.add_cog(cog)
asyncio.create_task(cog.initial_load_webhooks())

4
src/userdata.py
View File

@@ -87,5 +87,5 @@ class Userdata(commands.Cog):
await self.bot.database.commit()
await ctx.send(f"**{key}** deleted")
def setup(bot):
bot.add_cog(Userdata(bot))
async def setup(bot):
await bot.add_cog(Userdata(bot))

13
src/util.py
View File

@@ -378,7 +378,8 @@ async def generate(sess: aiohttp.ClientSession, prompt, stop=["\n"]):
# high to low
def sort_key(backend):
failure_stats = last_failures[backend["url"]]
return (failure_stats.avoid_until is None or failure_stats.avoid_until < now), backend["priority"], -failure_stats.consecutive_failures
currently_ok = failure_stats.avoid_until is None or failure_stats.avoid_until < now
return currently_ok, -(not currently_ok and failure_stats.consecutive_failures), backend["priority"]
backends = sorted(backends, key=sort_key, reverse=True)
@@ -396,6 +397,16 @@ async def generate(sess: aiohttp.ClientSession, prompt, stop=["\n"]):
failure_stats.avoid_until = now + datetime.timedelta(seconds=2 ** failure_stats.consecutive_failures)
failure_stats.consecutive_failures += 1
async def generate_raw_chatcompletion(sess: aiohttp.ClientSession, backend, prompt):
async with sess.post(backend["url"], json={
"messages": [{"role": "user", "content": prompt}],
"client": "abr",
**backend.get("params", {})
}, headers=backend.get("headers", {}), timeout=aiohttp.ClientTimeout(total=30)) as res:
data = await res.json()
print(data)
return data["choices"][0]["message"]["content"]
filesafe_charset = string.ascii_letters + string.digits + "-"
TARGET_FORMAT = "jpegh"

8
src/voice.py
View File

@@ -16,7 +16,7 @@ class HTTPSource(discord.AudioSource):
def read(self): return next(self.packets, b"")
def is_opus(self): return True
def setup(bot):
async def setup(bot):
# experimental, thus limit to me only
@bot.group()
@commands.check(util.admin_check)
@@ -43,11 +43,11 @@ def setup(bot):
@radio.command()
async def disconnect(ctx):
if ctx.guild.voice_client:
if ctx.guild.voice_client:
ctx.guild.voice_client.stop()
await ctx.guild.voice_client.disconnect()
def teardown(bot):
async def teardown(bot):
for guild in bot.guilds:
if guild.voice_client:
guild.voice_client.stop()
guild.voice_client.stop()