from __future__ import annotations
import importlib
import discord.errors
import traceback
import abc
from .command import Command, CommandCheck
from .config import Config
from .context import MessageContext
from .errors import *
from .response import Response
import aiohttp
from . import utils
import asyncio as aio
from . import argh
import re
import discord.ext
from aurcore.event import *
import inspect
import logging
logging.getLogger("discord.client").addFilter(lambda r: r.getMessage() != "PyNaCl is not installed, voice will NOT be supported")
logger = logging.getLogger(__name__)
if ty.TYPE_CHECKING:
import discord
from .types import *
class AurfluxEvent(Event):
def __init__(self, aurflux, __event_name, *args, **kwargs):
super().__init__(__event_name, *args, **kwargs)
self.bot: Aurflux = aurflux
class AurfluxCog:
def __init__(self, aurflux: Aurflux):
self.aurflux = aurflux
self.router = EventRouter(self.__class__.__name__, self.aurflux.router)
self.listeners: ty.Dict[ty.Union[EventFunction, EventRouter, EventWaiter], EventMuxer] = {}
self.commands = set()
self.route()
logging.info(f"[AurfluxCog] {self.__class__.__name__} loaded!")
async def startup(self):
pass
def register(self, cog_member: ty.Union[Command, ty.Tuple[EventMuxer, ty.Union[EventFunction, EventRouter, EventWaiter]]]):
print(f"Cog registering!")
print(cog_member)
if isinstance(cog_member, Command):
self.commands.add(cog_member)
else:
muxer, listener = cog_member
self.listeners[listener] = muxer
def teardown(self):
for listener, muxer in self.listeners.items():
print(f"Deregistering {listener} on {muxer}")
if isinstance(listener, EventRouter):
muxer.router = None
else:
muxer.remove_listener(listener)
for command in self.commands:
del self.aurflux.commands[command.name]
self.router.detatch()
@abc.abstractmethod
def route(self):
...
def ms(key):
try:
return dict(inspect.getmembers(
inspect.stack()[-1][0]))["f_globals"][key]
except KeyError:
for i in inspect.stack()[::-1]:
try:
return dict(inspect.getmembers(i[0]))["f_locals"][key]
except KeyError:
pass
try:
return dict(inspect.getmembers(i[0]))["f_globals"][key]
except KeyError:
pass
raise KeyError("Could not find key " + key)
def __aiterify(obj: ty.Union[ty.Coroutine, ty.AsyncIterable]):
if asyncio.iscoroutine(obj) or asyncio.isfuture(obj):
class AiterCoro:
def __aiter__(self):
async def gen():
yield await obj
return gen()
return AiterCoro()
else:
return obj
def register_builtins(aurflux: Aurflux, builtins=True):
@aurflux.router.endpoint(":ready", decompose=True)
async def _():
print("Ready!")
@aurflux.router.endpoint(":message", decompose=True)
async def command_listener(message: discord.Message):
if not message.content or message.author is aurflux.user:
return
ctx = MessageContext(bot=aurflux, message=message)
prefix = aurflux.CONFIG.of(ctx)["prefix"]
cmd = message.content.split(" ", 1)[0][len(prefix):]
if cmd not in aurflux.commands:
return
if not message.content.startswith(prefix):
return
try:
if cmd not in aurflux.commands:
return
async for response in __aiterify(await aurflux.commands[cmd].execute(ctx)):
await response.execute(ctx)
except CommandError as e:
info_message = f"{e}"
if argparser := aurflux.commands[cmd].argparser:
info_message += f"\n```{argparser.format_help()}```"
await Response(content=info_message, errored=True).execute(ctx)
except CommandInfo as e:
info_message = f"{e}"
if argparser := aurflux.commands[cmd].argparser:
info_message += f"\n```{argparser.format_help()}```"
await Response(content=info_message).execute(ctx)
@CommandCheck.check(lambda ctx: ctx.author.id == aurflux.admin_id)
@aurflux.commandeer(name="reload", parsed=False, private=True)
async def reload(ctx: MessageContext, cog_name: str):
reloaded_cogs = []
for cog in aurflux.cogs:
new_cog = cog
if cog.__class__.__name__.lower() == cog_name:
cog.teardown()
module = importlib.reload(inspect.getmodule(cog))
new_cog = getattr(module, cog.__class__.__name__)(ctx.aurflux)
await new_cog.startup()
reloaded_cogs.append(new_cog)
ctx.aurflux.cogs = reloaded_cogs
return Response()
@CommandCheck.check(CommandCheck.has_permissions(discord.Permissions(manage_guild=True )))
@aurflux.commandeer(name="setprefix", parsed=False)
async def set_prefix(ctx: MessageContext, prefix: str, *_):
"""
Sets the prefix to [prefix]
..setprefix !!
:param ctx:
:param prefix:
:param _:
:return:
"""
async with aurflux.CONFIG.writeable_conf(ctx) as cfg:
cfg["prefix"] = prefix
if not builtins:
@CommandCheck.check(lambda ctx: ctx.author.id == aurflux.admin_id)
@aurflux.commandeer(name="exec", parsed=False, private=True)
async def exec_(ctx: MessageContext, script: str):
exec_func = utils.sexec
if "await " in script:
exec_func = utils.aexec
with utils.Timer() as t:
# noinspection PyBroadException
try:
res = await exec_func(script, globals(), locals())
except Exception as e:
res = re.sub(r'File ".*[\\/]([^\\/]+.py)"', r'File "\1"', traceback.format_exc(limit=1))
return Response((f""
f"Ran in {t.elapsed * 1000:.2f} ms\n"
f"**IN**:\n"
f"```py\n{script}\n```\n"
f"**OUT**:\n"
f"```py\n{res}\n```"), trashable=True)
@aurflux.commandeer(name="help", parsed=False)
async def get_help(ctx: MessageContext, help_target: ty.Optional[str]):
"""
help [command_name]
:param ctx:
:param args:
:return:
"""
configs = aurflux.CONFIG.of(ctx)
public_cmds = {name: command for name, command in aurflux.commands.items() if not command.private and name != "help"}
if not help_target:
help_embed = discord.Embed(title=f"{utils.EMOJIS['question']} Command Help", description=f"{configs['prefix']}help <command> for more info")
for cmd_name, command in public_cmds.items():
help_embed.add_field(name=cmd_name, value=f"{configs['prefix']}{command.short_usage}", inline=False)
return Response(
embed=help_embed
)
else:
if help_target not in public_cmds:
return Response(f"No command `{help_target}` to show help for", errored=True)
embed = discord.Embed(
title="\U00002754 Command Help",
description=f"Help for `{configs['prefix']}{help_target}`")
if public_cmds[help_target].argparser:
embed.add_field(name="Usage", value=f"{public_cmds[help_target].short_usage}\n{public_cmds[help_target].long_usage}")
return Response(embed=embed)
class Aurflux(discord.Client):
CONFIG: Config
def __init__(self, name, admin_id: int, parent_router: EventRouter = None, buil
aurflux-1.2.80.tar.gz （14个子文件） 
aurflux-1.2.80 
PKG-INFO 263B pyproject.toml 364B aurflux aurflux.py 9KB utils.py 1KB errors.py 2KB types.py 57B __init__.py 200B context.py 2KB argh.py 4KB ext.py 589B command.py 5KB config.py 4KB response.py 3KB setup.py 594B资源评论
