generated from HoloTech/holotech-plugin-template
83 lines
2.5 KiB
Python
83 lines
2.5 KiB
Python
from typing import Any
|
|
import asyncio
|
|
import discord
|
|
from psycopg import sql
|
|
from urllib.parse import urlparse, parse_qs
|
|
|
|
from data import RawExpr, Expression
|
|
|
|
|
|
def LOWER(expression: Expression) -> RawExpr:
|
|
"""
|
|
Wrap an Expression in the SQL function LOWER().
|
|
"""
|
|
expr, values = expression.as_tuple()
|
|
final_expr = sql.SQL("LOWER({})").format(expr)
|
|
final_values = values
|
|
|
|
return RawExpr(final_expr, final_values)
|
|
|
|
|
|
def asexpr(value: Any) -> RawExpr:
|
|
"""
|
|
Turn a value into an expression.
|
|
"""
|
|
return RawExpr(sql.Placeholder(), (value,))
|
|
|
|
|
|
async def fire_and_forget(awaitable, do_in=1, ignorable=(discord.HTTPException)):
|
|
await asyncio.sleep(do_in)
|
|
try:
|
|
await awaitable
|
|
except ignorable:
|
|
pass
|
|
except Exception as e:
|
|
# TODO: Log unexpected exceptions
|
|
pass
|
|
|
|
|
|
class ThreadedWebhook(discord.Webhook):
|
|
__slots__ = ("thread_id",)
|
|
|
|
def __init__(self, *args, thread_id=None, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
self.thread_id = thread_id
|
|
|
|
@classmethod
|
|
def from_url(cls, url: str, *args, **kwargs):
|
|
self = super().from_url(url, *args, **kwargs)
|
|
parse = urlparse(url)
|
|
if parse.query:
|
|
args = parse_qs(parse.query)
|
|
if "thread_id" in args:
|
|
self.thread_id = int(args["thread_id"][0])
|
|
return self
|
|
|
|
async def send(self, *args, **kwargs):
|
|
if self.thread_id is not None:
|
|
kwargs.setdefault("thread", discord.Object(self.thread_id))
|
|
return await super().send(*args, **kwargs)
|
|
|
|
async def edit_message(self, *args, **kwargs):
|
|
if self.thread_id is not None:
|
|
kwargs.setdefault("thread", discord.Object(self.thread_id))
|
|
return await super().edit_message(*args, **kwargs)
|
|
|
|
async def delete_message(self, *args, **kwargs):
|
|
if self.thread_id is not None:
|
|
kwargs.setdefault("thread", discord.Object(self.thread_id))
|
|
return await super().delete_message(*args, **kwargs)
|
|
|
|
async def fetch_message(self, *args, **kwargs):
|
|
if self.thread_id is not None:
|
|
kwargs.setdefault("thread", discord.Object(self.thread_id))
|
|
return await super().fetch_message(*args, **kwargs)
|
|
|
|
async def test_webhook(self):
|
|
embed = discord.Embed(
|
|
title="Testing", description="Testing logging webhook, feel free to delete."
|
|
)
|
|
result = await self.send(embed=embed, wait=True, silent=True)
|
|
asyncio.create_task(fire_and_forget(result.delete()))
|
|
return result
|