HoloUA/Discord
HoloUA/Discord
1
0
Fork 0
Code Issues 1 Pull Requests Releases 3 Activity

Improved type-hinting and overall sanity checks implemented.

Browse Source
This commit is contained in:
kku 2024-12-16 20:34:37 +01:00
parent 5c763fc02e
commit c05cf64ae0
14 changed files with 193 additions and 165 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

76
classes/holo_user.py
View File

@ -1,55 +1,41 @@
import logging import logging
from typing import Any, Union from typing import Any, Union, Dict
import discord from bson import ObjectId
import discord.member from discord import User, Member
from libbot import config_get from libbot import config_get
from errors import UserNotFoundError
from modules.database import col_warnings, sync_col_users, sync_col_warnings, col_users from modules.database import col_warnings, sync_col_users, sync_col_warnings, col_users
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class UserNotFoundError(Exception):
"""HoloUser could not find user with such an ID in database"""
def __init__(self, user, user_id):
self.user = user
self.user_id = user_id
super().__init__(
f"User of type {type(self.user)} with id {self.user_id} was not found"
)
class HoloUser: class HoloUser:
def __init__( def __init__(self, user: Union[User, Member, int]) -> None:
self, user: Union[discord.User, discord.Member, discord.member.Member, int]
) -> None:
"""Get an object that has a proper binding between Discord ID and database """Get an object that has a proper binding between Discord ID and database
### Args: ### Args:
* `user` (Union[discord.User, discord.Member, discord.member.Member, int]): Object from which ID can be extracted * `user` (Union[User, Member, int]): Object from which ID can be extracted
### Raises: ### Raises:
* `UserNotFoundError`: User with such ID does not seem to exist in database * `UserNotFoundError`: User with such ID does not seem to exist in database
""" """
if hasattr(user, "id"): self.id: int = user if not hasattr(user, "id") else user.id
self.id = user.id # type: ignore
else:
self.id = user
jav_user = sync_col_users.find_one({"user": self.id}) jav_user: Union[Dict[str, Any], None] = sync_col_users.find_one(
{"user": self.id}
)
if jav_user is None: if jav_user is None:
raise UserNotFoundError(user=user, user_id=self.id) raise UserNotFoundError(user=user, user_id=self.id)
self.db_id = jav_user["_id"] self.db_id: ObjectId = jav_user["_id"]
self.customrole = jav_user["customrole"] self.customrole: Union[int, None] = jav_user["customrole"]
self.customchannel = jav_user["customchannel"] self.customchannel: Union[int, None] = jav_user["customchannel"]
self.warnings = self.warns() self.warnings: int = self.warns()
def warns(self) -> int: def warns(self) -> int:
"""Get number of warnings user has """Get number of warnings user has
@ -57,7 +43,9 @@ class HoloUser:
### Returns: ### Returns:
* `int`: Number of warnings * `int`: Number of warnings
""" """
warns = sync_col_warnings.find_one({"user": self.id}) warns: Union[Dict[str, Any], None] = sync_col_warnings.find_one(
{"user": self.id}
)
return 0 if warns is None else warns["warns"] return 0 if warns is None else warns["warns"]
@ -67,7 +55,9 @@ class HoloUser:
### Args: ### Args:
* `count` (int, optional): Count of warnings to be added. Defaults to 1. * `count` (int, optional): Count of warnings to be added. Defaults to 1.
""" """
warns = await col_warnings.find_one({"user": self.id}) warns: Union[Dict[str, Any], None] = await col_warnings.find_one(
{"user": self.id}
)
if warns is not None: if warns is not None:
await col_warnings.update_one( await col_warnings.update_one(
@ -77,7 +67,7 @@ class HoloUser:
else: else:
await col_warnings.insert_one(document={"user": self.id, "warns": count}) await col_warnings.insert_one(document={"user": self.id, "warns": count})
logger.info(f"User {self.id} was warned {count} times due to: {reason}") logger.info("User %s was warned %s times due to: %s", self.id, count, reason)
async def set(self, key: str, value: Any) -> None: async def set(self, key: str, value: Any) -> None:
"""Set attribute data and save it into database """Set attribute data and save it into database
@ -95,45 +85,41 @@ class HoloUser:
{"_id": self.db_id}, {"$set": {key: value}}, upsert=True {"_id": self.db_id}, {"$set": {key: value}}, upsert=True
) )
logger.info(f"Set attribute {key} of user {self.id} to {value}") logger.info("Set attribute %s of user %s to %s", key, self.id, value)
@staticmethod @staticmethod
async def is_moderator( async def is_moderator(member: Union[User, Member]) -> bool:
member: Union[discord.User, discord.Member, discord.member.Member]
) -> bool:
"""Check if user is moderator or council member """Check if user is moderator or council member
### Args: ### Args:
* `member` (Union[discord.User, discord.Member, discord.member.Member]): Member object * `member` (Union[User, Member]): Member object
### Returns: ### Returns:
`bool`: `True` if member is a moderator or member of council and `False` if not `bool`: `True` if member is a moderator or member of council and `False` if not
""" """
if isinstance(member, discord.User): if isinstance(member, User):
return False return False
moderator_role = await config_get("moderators", "roles") moderator_role: Union[int, None] = await config_get("moderators", "roles")
council_role = await config_get("council", "roles") council_role: Union[int, None] = await config_get("council", "roles")
for role in member.roles: for role in member.roles:
if role.id == moderator_role or role.id == council_role: if role.id in (moderator_role, council_role):
return True return True
return False return False
@staticmethod @staticmethod
async def is_council( async def is_council(member: Union[User, Member]) -> bool:
member: Union[discord.User, discord.Member, discord.member.Member]
) -> bool:
"""Check if user is a member of council """Check if user is a member of council
### Args: ### Args:
* `member` (Union[discord.User, discord.Member, discord.member.Member]): Member object * `member` (Union[User, Member]): Member object
### Returns: ### Returns:
`bool`: `True` if member is a member of council and `False` if not `bool`: `True` if member is a member of council and `False` if not
""" """
if isinstance(member, discord.User): if isinstance(member, User):
return False return False
council_role = await config_get("council", "roles") council_role = await config_get("council", "roles")

71
cogs/admin.py
View File

@ -1,14 +1,23 @@
import logging import logging
import sys import sys
from typing import Union
from discord import ApplicationContext, Embed, User, option, slash_command from discord import (
ApplicationContext,
Embed,
User,
option,
slash_command,
Role,
TextChannel,
)
from discord import utils as ds_utils from discord import utils as ds_utils
from discord.ext import commands from discord.ext import commands
from libbot import config_get from libbot import config_get
from libbot.pycord.classes import PycordBot from libbot.pycord.classes import PycordBot
from libbot.sync import config_get as sync_config_get from libbot.sync import config_get as sync_config_get
from enums.colors import Color from enums import Color
from modules.scheduled import scheduler from modules.scheduled import scheduler
from modules.utils_sync import guild_name from modules.utils_sync import guild_name
from modules.waifu_pics import waifu_pics from modules.waifu_pics import waifu_pics
@ -20,7 +29,7 @@ class Admin(commands.Cog):
"""Cog with utility commands for admins.""" """Cog with utility commands for admins."""
def __init__(self, client: PycordBot): def __init__(self, client: PycordBot):
self.client = client self.client: PycordBot = client
# Disabled because warning functionality is temporarily not needed # Disabled because warning functionality is temporarily not needed
# @slash_command( # @slash_command(
@ -117,7 +126,7 @@ class Admin(commands.Cog):
ctx: ApplicationContext, ctx: ApplicationContext,
amount: int, amount: int,
user: User, user: User,
): ) -> None:
if ctx.user.id in self.client.owner_ids: if ctx.user.id in self.client.owner_ids:
logging.info( logging.info(
"User %s removed %s message(s) in %s", "User %s removed %s message(s) in %s",
@ -148,46 +157,51 @@ class Admin(commands.Cog):
embed=Embed( embed=Embed(
title="Відмовлено в доступі", title="Відмовлено в доступі",
description="Здається, це команда лише для модераторів", description="Здається, це команда лише для модераторів",
color=Color.fail, color=Color.FAIL,
) )
) )
mod_role = ds_utils.get( mod_role: Union[Role, None] = ds_utils.get(
ctx.user.guild.roles, id=await config_get("moderators", "roles") ctx.user.guild.roles, id=await config_get("moderators", "roles")
) )
admin_chan = ds_utils.get( admin_chan: Union[TextChannel, None] = ds_utils.get(
ctx.user.guild.channels, ctx.user.guild.channels,
id=await config_get("adminchat", "channels", "text"), id=await config_get("adminchat", "channels", "text"),
) )
await admin_chan.send( if admin_chan is not None:
content=f"{mod_role.mention}", await admin_chan.send(
embed=Embed( content="" if mod_role is None else mod_role.mention,
title="Неавторизований запит", embed=Embed(
description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.", title="Неавторизований запит",
color=Color.fail, description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
), color=Color.FAIL,
) ),
)
@slash_command( @slash_command(
name="reboot", name="reboot",
description="Перезапустити бота", description="Перезапустити бота",
guild_ids=[sync_config_get("guild")], guild_ids=[sync_config_get("guild")],
) )
async def reboot_cmd(self, ctx: ApplicationContext): async def reboot_cmd(self, ctx: ApplicationContext) -> None:
await ctx.defer(ephemeral=True) await ctx.defer(ephemeral=True)
if ctx.user.id in self.client.owner_ids: if ctx.user.id in self.client.owner_ids:
logging.info("Calling shutdown initiated by %s", guild_name(ctx.user)) logging.info("Calling shutdown initiated by %s", guild_name(ctx.user))
await ctx.respond( await ctx.respond(
embed=Embed( embed=Embed(
title="Вимикаюсь...", title="Вимикаюсь...",
description="Спробую перезавантажитись за 5 секунд", description="Спробую перезавантажитись за 5 секунд",
) )
) )
scheduler.shutdown() scheduler.shutdown()
await self.client.close() await self.client.close()
await waifu_pics._client_session.close() await waifu_pics._client_session.close()
sys.exit() sys.exit()
logging.warning( logging.warning(
@ -199,27 +213,28 @@ class Admin(commands.Cog):
embed=Embed( embed=Embed(
title="Відмовлено в доступі", title="Відмовлено в доступі",
description="Здається, це команда лише для модераторів", description="Здається, це команда лише для модераторів",
color=Color.fail, color=Color.FAIL,
) )
) )
mod_role = ds_utils.get( mod_role: Union[Role, None] = ds_utils.get(
ctx.user.guild.roles, id=await config_get("moderators", "roles") ctx.user.guild.roles, id=await config_get("moderators", "roles")
) )
admin_chan = ds_utils.get( admin_chan: Union[TextChannel, None] = ds_utils.get(
ctx.user.guild.channels, ctx.user.guild.channels,
id=await config_get("adminchat", "channels", "text"), id=await config_get("adminchat", "channels", "text"),
) )
await admin_chan.send( if admin_chan is not None:
content=f"{mod_role.mention}", await admin_chan.send(
embed=Embed( content="" if mod_role is None else mod_role.mention,
title="Неавторизований запит", embed=Embed(
description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.", title="Неавторизований запит",
color=Color.fail, description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
), color=Color.FAIL,
) ),
)
def setup(client: PycordBot): def setup(client: PycordBot) -> None:
client.add_cog(Admin(client)) client.add_cog(Admin(client))

13
cogs/analytics.py
View File

@ -1,4 +1,5 @@
import logging import logging
from typing import Dict, List, Any
from discord import Cog, Message from discord import Cog, Message
from discord.ext import commands from discord.ext import commands
@ -11,16 +12,17 @@ logger = logging.getLogger(__name__)
class Analytics(commands.Cog): class Analytics(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: PycordBot):
self.client = client self.client: PycordBot = client
@Cog.listener() @Cog.listener()
async def on_message(self, message: Message): async def on_message(self, message: Message) -> None:
if ( if (
(message.author != self.client.user) (message.author != self.client.user)
and (message.author.bot is False) and (message.author.bot is False)
and (message.author.system is False) and (message.author.system is False)
): ):
stickers = [] stickers: List[Dict[str, Any]] = []
for sticker in message.stickers: for sticker in message.stickers:
stickers.append( stickers.append(
{ {
@ -31,7 +33,8 @@ class Analytics(commands.Cog):
} }
) )
attachments = [] attachments: List[Dict[str, Any]] = []
for attachment in message.attachments: for attachment in message.attachments:
attachments.append( attachments.append(
{ {
@ -57,5 +60,5 @@ class Analytics(commands.Cog):
) )
def setup(client: PycordBot): def setup(client: PycordBot) -> None:
client.add_cog(Analytics(client)) client.add_cog(Analytics(client))

58
cogs/custom_channels.py
View File

@ -1,4 +1,6 @@
from discord import ApplicationContext, Embed, option from typing import Any, Dict, List, Union
from discord import ApplicationContext, Embed, option, TextChannel
from discord import utils as ds_utils from discord import utils as ds_utils
from discord.abc import GuildChannel from discord.abc import GuildChannel
from discord.commands import SlashCommandGroup from discord.commands import SlashCommandGroup
@ -8,24 +10,26 @@ from libbot.pycord.classes import PycordBot
from libbot.sync import config_get as sync_config_get from libbot.sync import config_get as sync_config_get
from classes.holo_user import HoloUser from classes.holo_user import HoloUser
from enums.colors import Color from enums import Color
from modules.database import col_users from modules.database import col_users
from modules.utils_sync import guild_name from modules.utils_sync import guild_name
class CustomChannels(commands.Cog): class CustomChannels(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: PycordBot):
self.client = client self.client: PycordBot = client
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_channel_delete(self, channel: GuildChannel): async def on_guild_channel_delete(self, channel: GuildChannel) -> None:
await col_users.find_one_and_update( await col_users.find_one_and_update(
{"customchannel": channel.id}, {"$set": {"customchannel": None}} {"customchannel": channel.id}, {"$set": {"customchannel": None}}
) )
customchannel = SlashCommandGroup("customchannel", "Керування особистим каналом") custom_channel_group: SlashCommandGroup = SlashCommandGroup(
"customchannel", "Керування особистим каналом"
)
@customchannel.command( @custom_channel_group.command(
name="get", name="get",
description="Отримати персональний текстовий канал", description="Отримати персональний текстовий канал",
guild_ids=[sync_config_get("guild")], guild_ids=[sync_config_get("guild")],
@ -35,8 +39,8 @@ class CustomChannels(commands.Cog):
@option("threads", description="Дозволити гілки") @option("threads", description="Дозволити гілки")
async def customchannel_get_cmd( async def customchannel_get_cmd(
self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
): ) -> None:
holo_user_ctx = HoloUser(ctx.user) holo_user_ctx: HoloUser = HoloUser(ctx.user)
# Return if the user already has a custom channel # Return if the user already has a custom channel
if holo_user_ctx.customchannel is not None: if holo_user_ctx.customchannel is not None:
@ -45,14 +49,14 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Помилка виконання", title="Помилка виконання",
description="У вас вже є особистий канал.\nДля редагування каналу є `/customchannel edit` або просто відкрийте меню керування вашим каналом.", description="У вас вже є особистий канал.\nДля редагування каналу є `/customchannel edit` або просто відкрийте меню керування вашим каналом.",
color=Color.fail, color=Color.FAIL,
) )
) )
return return
await ctx.defer() await ctx.defer()
created_channel = await ctx.user.guild.create_text_channel( created_channel: TextChannel = await ctx.user.guild.create_text_channel(
name=name, name=name,
reason=f"Користувач {guild_name(ctx.user)} отримав власний приватний канал", reason=f"Користувач {guild_name(ctx.user)} отримав власний приватний канал",
category=ds_utils.get( category=ds_utils.get(
@ -83,11 +87,11 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Створено канал", title="Створено канал",
description=f"Вітаємо! Ви створили канал {created_channel.mention}. Для керування ним користуйтесь меню налаштувань каналу а також командою `/customchannel edit`", description=f"Вітаємо! Ви створили канал {created_channel.mention}. Для керування ним користуйтесь меню налаштувань каналу а також командою `/customchannel edit`",
color=Color.success, color=Color.SUCCESS,
) )
) )
bots = await config_get("bots") bots: List[Dict[str, Any]] = await config_get("bots")
for bot in bots: for bot in bots:
await created_channel.set_permissions( await created_channel.set_permissions(
@ -95,7 +99,7 @@ class CustomChannels(commands.Cog):
view_channel=False, view_channel=False,
) )
@customchannel.command( @custom_channel_group.command(
name="edit", name="edit",
description="Змінити параметри особистого каналу", description="Змінити параметри особистого каналу",
guild_ids=[sync_config_get("guild")], guild_ids=[sync_config_get("guild")],
@ -105,10 +109,10 @@ class CustomChannels(commands.Cog):
@option("threads", description="Дозволити гілки") @option("threads", description="Дозволити гілки")
async def customchannel_edit_cmd( async def customchannel_edit_cmd(
self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
): ) -> None:
holo_user_ctx = HoloUser(ctx.user) holo_user_ctx: HoloUser = HoloUser(ctx.user)
custom_channel = ds_utils.get( custom_channel: Union[TextChannel, None] = ds_utils.get(
ctx.guild.channels, id=holo_user_ctx.customchannel ctx.guild.channels, id=holo_user_ctx.customchannel
) )
@ -118,7 +122,7 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Канал не знайдено", title="Канал не знайдено",
description="Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.", description="Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.",
color=Color.fail, color=Color.FAIL,
) )
) )
return return
@ -136,11 +140,11 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Канал змінено", title="Канал змінено",
description=f"Назва каналу тепер `{name}`, реакції `{reactions}` та дозволено треди `{threads}`", description=f"Назва каналу тепер `{name}`, реакції `{reactions}` та дозволено треди `{threads}`",
color=Color.fail, color=Color.FAIL,
) )
) )
@customchannel.command( @custom_channel_group.command(
name="remove", name="remove",
description="Відібрати канал, знищуючи його, та частково повернути кошти", description="Відібрати канал, знищуючи його, та частково повернути кошти",
guild_ids=[sync_config_get("guild")], guild_ids=[sync_config_get("guild")],
@ -148,8 +152,8 @@ class CustomChannels(commands.Cog):
@option("confirm", description="Підтвердження операції") @option("confirm", description="Підтвердження операції")
async def customchannel_remove_cmd( async def customchannel_remove_cmd(
self, ctx: ApplicationContext, confirm: bool = False self, ctx: ApplicationContext, confirm: bool = False
): ) -> None:
holo_user_ctx = HoloUser(ctx.user) holo_user_ctx: HoloUser = HoloUser(ctx.user)
# Return if the user does not have a custom channel # Return if the user does not have a custom channel
if holo_user_ctx.customchannel is None: if holo_user_ctx.customchannel is None:
@ -158,14 +162,14 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Помилка виконання", title="Помилка виконання",
description="У вас немає особистого каналу.", description="У вас немає особистого каналу.",
color=Color.fail, color=Color.FAIL,
) )
) )
return return
await ctx.defer() await ctx.defer()
custom_channel = ds_utils.get( custom_channel: Union[TextChannel, None] = ds_utils.get(
ctx.guild.channels, id=holo_user_ctx.customchannel ctx.guild.channels, id=holo_user_ctx.customchannel
) )
@ -175,7 +179,7 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Канал не знайдено", title="Канал не знайдено",
description="Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.", description="Канал, вказаний як ваш, не існує. Можливо, його було вручну видалено раніше.",
color=Color.fail, color=Color.FAIL,
) )
) )
await holo_user_ctx.set("customchannel", None) await holo_user_ctx.set("customchannel", None)
@ -187,7 +191,7 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Підтвердження не надано", title="Підтвердження не надано",
description="Для підтвердження операції додайте до команди параметр `confirm` зі значенням `True`.", description="Для підтвердження операції додайте до команди параметр `confirm` зі значенням `True`.",
color=Color.fail, color=Color.FAIL,
) )
) )
return return
@ -201,10 +205,10 @@ class CustomChannels(commands.Cog):
embed=Embed( embed=Embed(
title="Канал знищено", title="Канал знищено",
description="Ви відмовились від каналу та видалили його.", description="Ви відмовились від каналу та видалили його.",
color=Color.default, color=Color.DEFAULT,
) )
) )
def setup(client: PycordBot): def setup(client: PycordBot) -> None:
client.add_cog(CustomChannels(client)) client.add_cog(CustomChannels(client))

59
cogs/data.py
View File

@ -1,9 +1,10 @@
import logging import logging
from os import makedirs from os import makedirs
from pathlib import Path from pathlib import Path
from typing import Union, List, Dict, Any
from uuid import uuid4 from uuid import uuid4
from discord import ApplicationContext, Embed, File, option from discord import ApplicationContext, Embed, File, option, Role, TextChannel
from discord import utils as ds_utils from discord import utils as ds_utils
from discord.commands import SlashCommandGroup from discord.commands import SlashCommandGroup
from discord.ext import commands from discord.ext import commands
@ -13,7 +14,7 @@ from libbot.sync import config_get as sync_config_get
from libbot.sync import json_write as sync_json_write from libbot.sync import json_write as sync_json_write
from classes.holo_user import HoloUser from classes.holo_user import HoloUser
from enums.colors import Color from enums import Color
from modules.database import col_users from modules.database import col_users
from modules.utils_sync import guild_name from modules.utils_sync import guild_name
@ -22,9 +23,9 @@ logger = logging.getLogger(__name__)
class Data(commands.Cog): class Data(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: PycordBot):
self.client = client self.client: PycordBot = client
data = SlashCommandGroup("data", "Керування даними користувачів") data: SlashCommandGroup = SlashCommandGroup("data", "Керування даними користувачів")
@data.command( @data.command(
name="export", name="export",
@ -34,9 +35,8 @@ class Data(commands.Cog):
@option( @option(
"kind", description="Тип даних, які треба експортувати", choices=["Користувачі"] "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
) )
async def data_export_cmd(self, ctx: ApplicationContext, kind: str): async def data_export_cmd(self, ctx: ApplicationContext, kind: str) -> None:
await ctx.defer() await ctx.defer()
holo_user = HoloUser(ctx.author)
# Return if the user is not an owner and not in the council # Return if the user is not an owner and not in the council
if (ctx.user.id not in self.client.owner_ids) and not ( if (ctx.user.id not in self.client.owner_ids) and not (
@ -51,24 +51,24 @@ class Data(commands.Cog):
embed=Embed( embed=Embed(
title="Відмовлено в доступі", title="Відмовлено в доступі",
description="Здається, це команда лише для модераторів", description="Здається, це команда лише для модераторів",
color=Color.fail, color=Color.FAIL,
) )
) )
mod_role = ds_utils.get( mod_role: Union[Role, None] = ds_utils.get(
ctx.user.guild.roles, id=await config_get("moderators", "roles") ctx.user.guild.roles, id=await config_get("moderators", "roles")
) )
admin_chan = ds_utils.get( admin_chan: Union[TextChannel, None] = ds_utils.get(
ctx.user.guild.channels, ctx.user.guild.channels,
id=await config_get("adminchat", "channels", "text"), id=await config_get("adminchat", "channels", "text"),
) )
await admin_chan.send( await admin_chan.send(
content=f"{mod_role.mention}", content="" if mod_role is None else mod_role.mention,
embed=Embed( embed=Embed(
title="Неавторизований запит", title="Неавторизований запит",
description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.", description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
color=Color.fail, color=Color.FAIL,
), ),
) )
@ -78,10 +78,10 @@ class Data(commands.Cog):
makedirs("tmp", exist_ok=True) makedirs("tmp", exist_ok=True)
uuid = str(uuid4()) uuid: str = str(uuid4())
if kind == "Користувачі": if kind == "Користувачі":
users = [] users: List[Dict[str, Any]] = []
for member in ctx.guild.members: for member in ctx.guild.members:
users.append( users.append(
@ -105,7 +105,7 @@ class Data(commands.Cog):
@option( @option(
"kind", description="Тип даних, які треба експортувати", choices=["Користувачі"] "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
) )
async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str): async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str) -> None:
await ctx.defer() await ctx.defer()
# Return if the user is not an owner and not in the council # Return if the user is not an owner and not in the council
@ -121,26 +121,27 @@ class Data(commands.Cog):
embed=Embed( embed=Embed(
title="Відмовлено в доступі", title="Відмовлено в доступі",
description="Здається, це команда лише для модераторів", description="Здається, це команда лише для модераторів",
color=Color.fail, color=Color.FAIL,
) )
) )
mod_role = ds_utils.get( mod_role: Union[Role, None] = ds_utils.get(
ctx.user.guild.roles, id=await config_get("moderators", "roles") ctx.user.guild.roles, id=await config_get("moderators", "roles")
) )
admin_chan = ds_utils.get( admin_chan: Union[TextChannel, None] = ds_utils.get(
ctx.user.guild.channels, ctx.user.guild.channels,
id=await config_get("adminchat", "channels", "text"), id=await config_get("adminchat", "channels", "text"),
) )
await admin_chan.send( if admin_chan is not None:
content=f"{mod_role.mention}", await admin_chan.send(
embed=Embed( content="" if mod_role is None else mod_role.mention,
title="Неавторизований запит", embed=Embed(
description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.", title="Неавторизований запит",
color=Color.fail, description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
), color=Color.FAIL,
) ),
)
return return
@ -155,8 +156,8 @@ class Data(commands.Cog):
continue continue
if (await col_users.find_one({"user": member.id})) is None: if (await col_users.find_one({"user": member.id})) is None:
user = {} user: Dict[str, Any] = {}
defaults = await config_get("user", "defaults") defaults: Dict[str, Any] = await config_get("user", "defaults")
user["user"] = member.id user["user"] = member.id
@ -173,10 +174,10 @@ class Data(commands.Cog):
embed=Embed( embed=Embed(
title="Міграцію завершено", title="Міграцію завершено",
description="Всім користувачам сервера було створено записи в базі даних.", description="Всім користувачам сервера було створено записи в базі даних.",
color=Color.success, color=Color.SUCCESS,
) )
) )
def setup(client: PycordBot): def setup(client: PycordBot) -> None:
client.add_cog(Data(client)) client.add_cog(Data(client))

18
cogs/fun.py
View File

@ -14,7 +14,7 @@ logger = logging.getLogger(__name__)
class Fun(commands.Cog): class Fun(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: PycordBot):
self.client = client self.client: PycordBot = client
@slash_command( @slash_command(
name="action", name="action",
@ -27,13 +27,13 @@ class Fun(commands.Cog):
choices=sync_config_get("actions").keys(), choices=sync_config_get("actions").keys(),
) )
@option("user", description="Користувач") @option("user", description="Користувач")
async def action_cmd(self, ctx: ApplicationContext, type: str, user: User): async def action_cmd(self, ctx: ApplicationContext, type: str, user: User) -> None:
await ctx.defer() await ctx.defer()
action = await config_get("category", "actions", type) action: str = await config_get("category", "actions", type)
action_verb = await config_get("action", "actions", type) action_verb: str = await config_get("action", "actions", type)
image = await waifu_pics.sfw(action) image_url: str = await waifu_pics.sfw(action)
logger.info( logger.info(
"User %s (%s) %s %s (%s) with image %s", "User %s (%s) %s %s (%s) with image %s",
@ -42,17 +42,17 @@ class Fun(commands.Cog):
action_verb, action_verb,
guild_name(user), guild_name(user),
user.id, user.id,
image, image_url,
) )
embed = Embed( embed: Embed = Embed(
description=f"**{guild_name(ctx.user)}** {action_verb} **{guild_name(user)}**", description=f"**{guild_name(ctx.user)}** {action_verb} **{guild_name(user)}**",
color=0x2F3136, color=0x2F3136,
) )
embed.set_image(url=image) embed.set_image(url=image_url)
await ctx.respond(embed=embed) await ctx.respond(embed=embed)
def setup(client: PycordBot): def setup(client: PycordBot) -> None:
client.add_cog(Fun(client)) client.add_cog(Fun(client))

22
cogs/logger.py
View File

@ -1,4 +1,6 @@
from discord import Member, Message from typing import Dict, Any, Union
from discord import Member, Message, TextChannel
from discord import utils as ds_utils from discord import utils as ds_utils
from discord.ext import commands from discord.ext import commands
from libbot import config_get from libbot import config_get
@ -9,7 +11,7 @@ from modules.database import col_users
class Logger(commands.Cog): class Logger(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: PycordBot):
self.client = client self.client: PycordBot = client
@commands.Cog.listener() @commands.Cog.listener()
async def on_message(self, message: Message): async def on_message(self, message: Message):
@ -19,8 +21,8 @@ class Logger(commands.Cog):
and (message.author.system is False) and (message.author.system is False)
): ):
if (await col_users.find_one({"user": message.author.id})) is None: if (await col_users.find_one({"user": message.author.id})) is None:
user = {} user: Dict[str, Any] = {}
defaults = await config_get("user", "defaults") defaults: Dict[str, Any] = await config_get("user", "defaults")
user["user"] = message.author.id user["user"] = message.author.id
@ -30,12 +32,12 @@ class Logger(commands.Cog):
await col_users.insert_one(document=user) await col_users.insert_one(document=user)
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_join(self, member: Member): async def on_member_join(self, member: Member) -> None:
welcome_chan = ds_utils.get( welcome_chan: Union[TextChannel, None] = ds_utils.get(
self.client.get_guild(await config_get("guild")).channels, self.client.get_guild(await config_get("guild")).channels,
id=await config_get("welcome", "channels", "text"), id=await config_get("welcome", "channels", "text"),
) )
rules_chan = ds_utils.get( rules_chan: Union[TextChannel, None] = ds_utils.get(
self.client.get_guild(await config_get("guild")).channels, self.client.get_guild(await config_get("guild")).channels,
id=await config_get("rules", "channels", "text"), id=await config_get("rules", "channels", "text"),
) )
@ -52,8 +54,8 @@ class Logger(commands.Cog):
) )
if (await col_users.find_one({"user": member.id})) is None: if (await col_users.find_one({"user": member.id})) is None:
user = {} user: Dict[str, Any] = {}
defaults = await config_get("user", "defaults") defaults: Dict[str, Any] = await config_get("user", "defaults")
user["user"] = member.id user["user"] = member.id
@ -63,5 +65,5 @@ class Logger(commands.Cog):
await col_users.insert_one(document=user) await col_users.insert_one(document=user)
def setup(client: PycordBot): def setup(client: PycordBot) -> None:
client.add_cog(Logger(client)) client.add_cog(Logger(client))

1
enums/__init__.py Normal file
View File

@ -0,0 +1 @@
from .colors import Color

7
enums/colors.py
View File

@ -1,6 +1,7 @@
from enum import IntEnum from enum import IntEnum
class Color(IntEnum): class Color(IntEnum):
fail = 0xd6345b FAIL = 0xD6345B
success = 0x84d961 SUCCESS = 0x84D961
default = 0xa7a6ab DEFAULT = 0xA7A6AB

1
errors/__init__.py Normal file
View File

@ -0,0 +1 @@
from .user import UserNotFoundError

10
errors/user.py Normal file
View File

@ -0,0 +1,10 @@
class UserNotFoundError(Exception):
"""HoloUser could not find user with such an ID in database"""
def __init__(self, user, user_id):
self.user = user
self.user_id = user_id
super().__init__(
f"User of type {type(self.user)} with id {self.user_id} was not found"
)

14
main.py
View File

@ -1,4 +1,6 @@
import logging import logging
import sys
from logging import Logger
from discord import Activity, ActivityType from discord import Activity, ActivityType
from libbot import config_get from libbot import config_get
@ -13,7 +15,7 @@ logging.basicConfig(
datefmt="[%X]", datefmt="[%X]",
) )
logger = logging.getLogger(__name__) logger: Logger = logging.getLogger(__name__)
try: try:
import uvloop # type: ignore import uvloop # type: ignore
@ -24,11 +26,11 @@ except ImportError:
@client.event @client.event
async def on_ready(): async def on_ready() -> None:
logger.info("Logged in as %s", client.user) logger.info("Logged in as %s", client.user)
activity_type = await config_get("type", "status") activity_type: str = await config_get("type", "status")
activity_message = await config_get("message", "status") activity_message: str = await config_get("message", "status")
if activity_type == "playing": if activity_type == "playing":
await client.change_presence( await client.change_presence(
@ -62,7 +64,7 @@ async def on_ready():
) )
def main(): def main() -> None:
client.load_extension("cogs") client.load_extension("cogs")
try: try:
@ -70,7 +72,7 @@ def main():
client.run(sync_config_get("bot_token", "bot")) client.run(sync_config_get("bot_token", "bot"))
except KeyboardInterrupt: except KeyboardInterrupt:
scheduler.shutdown() scheduler.shutdown()
exit() sys.exit()
if __name__ == "__main__": if __name__ == "__main__":

6
modules/client.py
View File

@ -1,6 +1,8 @@
from discord import Intents from discord import Intents
from libbot.pycord.classes import PycordBot from libbot.pycord.classes import PycordBot
intents = Intents().all() intents: Intents = Intents().all()
intents.members = True intents.members = True
client = PycordBot(intents=intents)
client: PycordBot = PycordBot(intents=intents)

2
modules/scheduled.py
View File

@ -1,3 +1,3 @@
from apscheduler.schedulers.asyncio import AsyncIOScheduler from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler() scheduler: AsyncIOScheduler = AsyncIOScheduler()