39 lines
1.4 KiB
Python
39 lines
1.4 KiB
Python
"""Custom JARVIS client."""
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from interactions.ext.prefixed_commands.context import PrefixedContext
|
|
from interactions.models.internal.context import BaseContext, InteractionContext
|
|
from jarvis_core.util.ansi import Fore, Format, fmt
|
|
from statipy import StatipyClient
|
|
|
|
from jarvis.client.errors import ErrorMixin
|
|
from jarvis.client.events import EventMixin
|
|
from jarvis.client.tasks import TaskMixin
|
|
|
|
if TYPE_CHECKING:
|
|
from redis import asyncio as aioredis
|
|
|
|
KEY_FMT = fmt(Fore.GRAY)
|
|
VAL_FMT = fmt(Fore.WHITE)
|
|
CMD_FMT = fmt(Fore.GREEN, Format.BOLD)
|
|
|
|
|
|
class Jarvis(StatipyClient, ErrorMixin, EventMixin, TaskMixin):
|
|
def __init__(self, redis: "aioredis.Redis", *args, **kwargs): # noqa: ANN002 ANN003
|
|
super().__init__(*args, **kwargs)
|
|
self.redis = redis
|
|
self.logger = logging.getLogger(__name__)
|
|
self.phishing_domains = []
|
|
self.pre_run_callback = self._prerun
|
|
self.synced = False
|
|
|
|
async def _prerun(self, ctx: BaseContext, *args, **kwargs) -> None:
|
|
name = ctx.invoke_target
|
|
cargs = ""
|
|
if isinstance(ctx, InteractionContext) and hasattr(ctx, "target_id"):
|
|
kwargs["context target"] = ctx.target
|
|
cargs = " ".join(f"{k}:{v}" for k, v in kwargs.items())
|
|
elif isinstance(ctx, PrefixedContext):
|
|
cargs = " ".join(args)
|
|
self.logger.debug(f"Running command `{name}` with args: {cargs or 'None'}")
|