1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
|
"""Module containing the core bot base for King Arthur."""
from pathlib import Path
from typing import Any, Union
from botcore import BotBase
from botcore.utils import scheduling
from discord import Interaction, Member, User
from discord.ext import commands
from kubernetes_asyncio import config
from arthur import exts, logger
from arthur.config import CONFIG
class KingArthur(BotBase):
"""Base bot class for King Arthur."""
def __init__(self, *args: list[Any], **kwargs: dict[str, Any]) -> None:
super().__init__(*args, **kwargs)
self.add_check(self._is_devops)
@staticmethod
def _is_devops(ctx: Union[commands.Context, Interaction]) -> bool:
"""Check all commands are executed by authorised personnel."""
if isinstance(ctx, Interaction):
if isinstance(ctx.user, Member):
return CONFIG.devops_role in [r.id for r in ctx.user.roles]
else:
return False
if ctx.command.name == "ed":
return True
if not ctx.guild:
return False
return CONFIG.devops_role in [r.id for r in ctx.author.roles]
async def setup_hook(self) -> None:
"""Async initialisation method for discord.py."""
await super().setup_hook()
# Authenticate with Kubernetes
if (Path.home() / ".kube/config").exists():
await config.load_kube_config()
else:
config.load_incluster_config()
logger.info(f"Logged in <red>{self.user}</>")
# This is not awaited to avoid a deadlock with any cogs that have
# wait_until_guild_available in their cog_load method.
scheduling.create_task(self.load_extensions(exts))
logger.info("Loading <red>jishaku</red>")
await self.load_extension("jishaku")
logger.info("Loaded <red>jishaku</red>")
async def is_owner(self, user: Union[User, Member]) -> bool:
"""Check if the invoker is a bot owner."""
if not user.guild:
return False
return CONFIG.devops_role in [r.id for r in user.roles]
|
