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

Merge pull request 'v0.1.0-rc.4' (#19) from dev into main

Browse Source 
Reviewed-on: #19
This commit is contained in:
Profitroll 2024-12-27 21:31:46 +02:00
commit 6087349622
17 changed files with 329 additions and 242 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
.dockerignore Normal file
View File

@ -0,0 +1,40 @@
# Include any files or directories that you don't want to be copied to your
# container here (e.g., local build artifacts, temporary files, etc.).
#
# For more help, visit the .dockerignore file reference guide at
# https://docs.docker.com/go/build-context-dockerignore/
**/.DS_Store
**/__pycache__
**/.venv
**/.classpath
**/.dockerignore
**/.env
**/.git
**/.gitignore
**/.project
**/.settings
**/.toolstarget
**/.vs
**/.vscode
**/*.*proj.user
**/*.dbmdl
**/*.jfm
**/bin
**/charts
**/docker-compose*
**/compose.y*ml
**/Dockerfile*
**/node_modules
**/npm-debug.log
**/obj
**/secrets.dev.yaml
**/values.dev.yaml
LICENSE
README.md
config.json
.renovaterc
**/.idea
**/.mypy_cache
validation

27
Dockerfile Normal file
View File

@ -0,0 +1,27 @@
ARG PYTHON_VERSION=3.12.8
FROM python:${PYTHON_VERSION}-slim AS base
ENV PYTHONDONTWRITEBYTECODE=1
ENV PYTHONUNBUFFERED=1
WORKDIR /app
ARG UID=10001
RUN adduser \
--disabled-password \
--gecos "" \
--home "/nonexistent" \
--shell "/sbin/nologin" \
--no-create-home \
--uid "${UID}" \
appuser
RUN --mount=type=cache,target=/root/.cache/pip \
--mount=type=bind,source=requirements.txt,target=requirements.txt \
python -m pip install -r requirements.txt
USER appuser
COPY . .
ENTRYPOINT ["python", "main.py"]

51
README.md
View File

@ -7,14 +7,32 @@
<a href="https://git.end-play.xyz/HoloUA/Discord"><img alt="Code style: black" src="https://img.shields.io/badge/code%20style-black-000000.svg"></a> <a href="https://git.end-play.xyz/HoloUA/Discord"><img alt="Code style: black" src="https://img.shields.io/badge/code%20style-black-000000.svg"></a>
</p> </p>
## Installation ## Installation from release
1. Install MongoDB using the [official installation manual](https://www.mongodb.com/docs/manual/installation/). 1. Install MongoDB using the [official installation manual](https://www.mongodb.com/docs/manual/installation)
2. `git clone https://git.end-play.xyz/HoloUA/Discord.git` 2. Install Python 3.9+ (3.11+ is recommended)
3. `cd Discord` 3. Download the [latest release](https://git.end-play.xyz/HoloUA/Discord/releases/latest)'s archive
4. Install Python 3.9+ (at least 3.11 is recommended) for your OS 4. Extract the archive
5. `python3 -m pip install -r requirements.txt` 5. Navigate to the extracted folder and subfolder `Discord` in it
6. Run it with `python3 main.py` after configuring 6. Create a virtual environment:
`python -m venv .venv` or `virtualenv .venv`
7. Activate the virtual environment:
Windows: `.venv\Scripts\activate.bat`
Linux/macOS: `.venv/bin/activate`
8. Install dependencies:
`python -m pip install -r requirements.txt`
9. Run the bot with `python main.py` after completing the [configuration](#Configuration)
## Installation with Git
1. Install MongoDB using the [official installation manual](https://www.mongodb.com/docs/manual/installation)
2. Install Python 3.9+ (3.11+ is recommended)
3. Clone the repository:
`git clone https://git.end-play.xyz/HoloUA/Discord.git`
4. `cd Discord`
5. Install dependencies:
`python -m pip install -r requirements.txt`
6. Run the bot with `python main.py` after completing the [configuration](#Configuration)
## Configuration ## Configuration
@ -36,3 +54,22 @@ Mandatory keys to modify:
- roles.* - roles.*
After all of that you're good to go! Happy using :) After all of that you're good to go! Happy using :)
## Docker [Experimental]
As an experiment, Docker deployment option has been added.
### Building the image
1. `git clone https://git.end-play.xyz/HoloUA/Discord.git`
2. `cd Discord`
3. `docker build -t holoua-discord .`
### Starting the bot
1. Install MongoDB using the [official installation manual](https://www.mongodb.com/docs/manual/installation)
2. Download
the [configuration example file](https://git.end-play.xyz/HoloUA/Discord/src/branch/main/config_example.json) and
store it somewhere you would like your bot to access it
3. Complete the [configuration](#Configuration) step for this file
4. `docker run -d -v /path/to/config.json:/app/config.json holoua-discord`

6
classes/holo_bot.py Normal file
View File

@ -0,0 +1,6 @@
from libbot.pycord.classes import PycordBot
class HoloBot(PycordBot):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)

36
classes/holo_user.py
View File

@ -1,9 +1,9 @@
import logging import logging
from typing import Any, Union, Dict from typing import Any, Dict
from bson import ObjectId from bson import ObjectId
from discord import User, Member from discord import User, Member
from libbot import config_get from libbot.utils import config_get
from errors import UserNotFoundError 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
@ -12,11 +12,11 @@ logger = logging.getLogger(__name__)
class HoloUser: class HoloUser:
def __init__(self, user: Union[User, Member, int]) -> None: def __init__(self, user: User | 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[User, Member, int]): Object from which ID can be extracted * `user` (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
@ -24,17 +24,15 @@ class HoloUser:
self.id: int = user if not hasattr(user, "id") else user.id self.id: int = user if not hasattr(user, "id") else user.id
jav_user: Union[Dict[str, Any], None] = sync_col_users.find_one( jav_user: Dict[str, Any] | None = sync_col_users.find_one({"user": self.id})
{"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: ObjectId = jav_user["_id"] self.db_id: ObjectId = jav_user["_id"]
self.customrole: Union[int, None] = jav_user["customrole"] self.customrole: int | None = jav_user["customrole"]
self.customchannel: Union[int, None] = jav_user["customchannel"] self.customchannel: int | None = jav_user["customchannel"]
self.warnings: int = self.warns() self.warnings: int = self.warns()
def warns(self) -> int: def warns(self) -> int:
@ -43,9 +41,7 @@ class HoloUser:
### Returns: ### Returns:
* `int`: Number of warnings * `int`: Number of warnings
""" """
warns: Union[Dict[str, Any], None] = sync_col_warnings.find_one( warns: Dict[str, Any] | None = sync_col_warnings.find_one({"user": self.id})
{"user": self.id}
)
return 0 if warns is None else warns["warns"] return 0 if warns is None else warns["warns"]
@ -55,9 +51,7 @@ 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: Union[Dict[str, Any], None] = await col_warnings.find_one( warns: Dict[str, Any] | None = await col_warnings.find_one({"user": self.id})
{"user": self.id}
)
if warns is not None: if warns is not None:
await col_warnings.update_one( await col_warnings.update_one(
@ -88,11 +82,11 @@ class HoloUser:
logger.info("Set attribute %s of user %s to %s", key, self.id, value) logger.info("Set attribute %s of user %s to %s", key, self.id, value)
@staticmethod @staticmethod
async def is_moderator(member: Union[User, Member]) -> bool: async def is_moderator(member: User | Member) -> bool:
"""Check if user is moderator or council member """Check if user is moderator or council member
### Args: ### Args:
* `member` (Union[User, Member]): Member object * `member` (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
@ -100,8 +94,8 @@ class HoloUser:
if isinstance(member, User): if isinstance(member, User):
return False return False
moderator_role: Union[int, None] = await config_get("moderators", "roles") moderator_role: int | None = await config_get("moderators", "roles")
council_role: Union[int, None] = await config_get("council", "roles") council_role: int | None = await config_get("council", "roles")
for role in member.roles: for role in member.roles:
if role.id in (moderator_role, council_role): if role.id in (moderator_role, council_role):
@ -110,11 +104,11 @@ class HoloUser:
return False return False
@staticmethod @staticmethod
async def is_council(member: Union[User, Member]) -> bool: async def is_council(member: User | Member) -> bool:
"""Check if user is a member of council """Check if user is a member of council
### Args: ### Args:
* `member` (Union[User, Member]): Member object * `member` (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

115
cogs/admin.py
View File

@ -1,6 +1,5 @@
import logging import logging
import sys import sys
from typing import Union
from discord import ( from discord import (
ApplicationContext, ApplicationContext,
@ -13,10 +12,9 @@ from discord import (
) )
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.utils import config_get
from libbot.pycord.classes import PycordBot
from libbot.sync import config_get as sync_config_get
from classes.holo_bot import HoloBot
from enums import Color from enums import Color
from modules.scheduler import scheduler from modules.scheduler import scheduler
from modules.utils_sync import guild_name from modules.utils_sync import guild_name
@ -28,96 +26,13 @@ logger = logging.getLogger(__name__)
class Admin(commands.Cog): 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: HoloBot):
self.client: PycordBot = client self.client: HoloBot = client
# Disabled because warning functionality is temporarily not needed
# @slash_command(
# name="warning",
# description="Попередити юзера про порушення правил",
# guild_ids=[config_get_sync("guild")],
# )
# @option("user", description="Користувач")
# @option("reason", description="Причина")
# async def warn_cmd(
# self,
# ctx: ApplicationContext,
# user: User,
# reason: str = "Не вказана",
# ):
# logging.info(f"User {ctx.user.id} warned {user.id} for {reason}")
# await ctx.defer()
# jav_user = HoloUser(user)
# if ctx.user.id in await config_get("admins"):
# logging.info(
# f"Moderator {guild_name(ctx.user)} warned {guild_name(user)} for {reason} (has {jav_user.warnings} warns)"
# )
# if jav_user.warnings >= 5:
# logging.info(
# f"User {guild_name(user)} was banned due to a big amount of warns ({jav_user.warnings})"
# )
# await user.send(
# embed=Embed(
# title="Перманентне блокування",
# description=f"Вас було заблоковано за неодноразове порушення правил сервера.",
# color=Color.fail,
# )
# )
# await user.ban(reason=reason)
# elif jav_user.warnings >= 2:
# logging.info(
# f"User {guild_name(user)} was muted due to a big amount of warns ({jav_user.warnings})"
# )
# jav_user.warn(reason=reason)
# await user.send(
# embed=Embed(
# title="Тимчасове блокування",
# description=f"Причина: `{reason}`\n\nНа вашому рахунку вже {jav_user.warnings} попереджень. Вас було тимчасово заблоковано на **1 годину**.\n\nЯкщо Ви продовжите порушувати правила сервера – згодом Вас заблокують.",
# color=0xDED56B,
# )
# )
# await user.timeout_for(timedelta(hours=1), reason=reason)
# else:
# jav_user.warn()
# await ctx.respond(
# embed=Embed(
# title="Попередження",
# description=f"{user.mention} Будь ласка, не порушуйте правила. Ви отримали попередження з причини `{reason}`.\n\nЯкщо Ви продовжите порушувати правила – це може призвести до блокування в спільноті.",
# color=0xDED56B,
# )
# )
# else:
# logging.warning(
# f"User {guild_name(ctx.user)} tried to use /warn but permission denied"
# )
# await ctx.respond(
# embed=Embed(
# title="Відмовлено в доступі",
# description="Здається, це команда лише для модераторів",
# color=Color.fail,
# )
# )
# mod_role = ds_utils.get(
# ctx.user.guild.roles, id=await config_get("moderators", "roles")
# )
# admin_chan = ds_utils.get(
# ctx.user.guild.channels,
# id=await config_get("adminchat", "channels", "text"),
# )
# await admin_chan.send(
# content=f"{mod_role.mention}",
# embed=Embed(
# title="Неавторизований запит",
# description=f"Користувач {ctx.user.mention} запитав у каналі {ctx.channel.mention} команду, до якої не повинен мати доступу/бачити.",
# color=Color.fail,
# ),
# )
@slash_command( @slash_command(
name="clear", name="clear",
description="Видалити деяку кількість повідомлень в каналі", description="Видалити деяку кількість повідомлень в каналі",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
@option("amount", description="Кількість") @option("amount", description="Кількість")
@option("user", description="Користувач", default=None) @option("user", description="Користувач", default=None)
@ -127,6 +42,10 @@ class Admin(commands.Cog):
amount: int, amount: int,
user: User, user: User,
) -> None: ) -> None:
"""Command /clear <amount> [<user>]
Removes last <amount> messages in the current channel. Optionally from a specific user.
"""
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",
@ -161,10 +80,10 @@ class Admin(commands.Cog):
) )
) )
mod_role: Union[Role, None] = ds_utils.get( mod_role: 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: Union[TextChannel, None] = ds_utils.get( admin_chan: 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"),
) )
@ -182,9 +101,13 @@ class Admin(commands.Cog):
@slash_command( @slash_command(
name="reboot", name="reboot",
description="Перезапустити бота", description="Перезапустити бота",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
async def reboot_cmd(self, ctx: ApplicationContext) -> None: async def reboot_cmd(self, ctx: ApplicationContext) -> None:
"""Command /reboot
Stops the bot. Is called "reboot" because it's assumed that the bot has automatic restart.
"""
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:
@ -217,10 +140,10 @@ class Admin(commands.Cog):
) )
) )
mod_role: Union[Role, None] = ds_utils.get( mod_role: 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: Union[TextChannel, None] = ds_utils.get( admin_chan: 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"),
) )
@ -236,5 +159,5 @@ class Admin(commands.Cog):
) )
def setup(client: PycordBot) -> None: def setup(client: HoloBot) -> None:
client.add_cog(Admin(client)) client.add_cog(Admin(client))

12
cogs/analytics.py
View File

@ -3,24 +3,26 @@ 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
from libbot.pycord.classes import PycordBot
from classes.holo_bot import HoloBot
from modules.database import col_analytics from modules.database import col_analytics
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class Analytics(commands.Cog): class Analytics(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: HoloBot):
self.client: PycordBot = client self.client: HoloBot = client
@Cog.listener() @Cog.listener()
async def on_message(self, message: Message) -> None: async def on_message(self, message: Message) -> None:
"""Listener that collects analytical data (stickers, attachments, messages)."""
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)
): ):
# Handle stickers
stickers: List[Dict[str, Any]] = [] stickers: List[Dict[str, Any]] = []
for sticker in message.stickers: for sticker in message.stickers:
@ -33,6 +35,7 @@ class Analytics(commands.Cog):
} }
) )
# Handle attachments
attachments: List[Dict[str, Any]] = [] attachments: List[Dict[str, Any]] = []
for attachment in message.attachments: for attachment in message.attachments:
@ -49,6 +52,7 @@ class Analytics(commands.Cog):
} }
) )
# Insert entry into the database
await col_analytics.insert_one( await col_analytics.insert_one(
{ {
"user": message.author.id, "user": message.author.id,
@ -60,5 +64,5 @@ class Analytics(commands.Cog):
) )
def setup(client: PycordBot) -> None: def setup(client: HoloBot) -> None:
client.add_cog(Analytics(client)) client.add_cog(Analytics(client))

38
cogs/custom_channels.py
View File

@ -1,15 +1,14 @@
import logging import logging
from typing import Any, Dict, Union from typing import Any, Dict
from discord import ApplicationContext, Embed, option, TextChannel, Role from discord import ApplicationContext, Embed, option, TextChannel, Role
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
from discord.ext import commands from discord.ext import commands
from libbot import config_get from libbot.utils import config_get
from libbot.pycord.classes import PycordBot
from libbot.sync import config_get as sync_config_get
from classes.holo_bot import HoloBot
from classes.holo_user import HoloUser from classes.holo_user import HoloUser
from enums import Color from enums import Color
from modules.database import col_users from modules.database import col_users
@ -19,8 +18,8 @@ logger = logging.getLogger(__name__)
class CustomChannels(commands.Cog): class CustomChannels(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: HoloBot):
self.client: PycordBot = client self.client: HoloBot = client
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_channel_delete(self, channel: GuildChannel) -> None: async def on_guild_channel_delete(self, channel: GuildChannel) -> None:
@ -35,7 +34,7 @@ class CustomChannels(commands.Cog):
@custom_channel_group.command( @custom_channel_group.command(
name="get", name="get",
description="Отримати персональний текстовий канал", description="Отримати персональний текстовий канал",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
@option("name", description="Назва каналу") @option("name", description="Назва каналу")
@option("reactions", description="Дозволити реакції") @option("reactions", description="Дозволити реакції")
@ -43,6 +42,10 @@ class CustomChannels(commands.Cog):
async def custom_channel_get_cmd( async def custom_channel_get_cmd(
self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
) -> None: ) -> None:
"""Command /customchannel get <name> <reactions> <threads>
Command to create a custom channel for a user.
"""
holo_user_ctx: HoloUser = HoloUser(ctx.user) holo_user_ctx: HoloUser = HoloUser(ctx.user)
# Return if the user is using the command outside of a guild # Return if the user is using the command outside of a guild
@ -109,9 +112,7 @@ class CustomChannels(commands.Cog):
bots: Dict[str, Any] = await config_get("bots") bots: Dict[str, Any] = await config_get("bots")
for bot in bots: for bot in bots:
role: Union[Role, None] = ds_utils.get( role: Role | None = ds_utils.get(ctx.user.guild.roles, id=bots[bot]["role"])
ctx.user.guild.roles, id=bots[bot]["role"]
)
if role is not None: if role is not None:
await created_channel.set_permissions( await created_channel.set_permissions(
@ -122,7 +123,7 @@ class CustomChannels(commands.Cog):
@custom_channel_group.command( @custom_channel_group.command(
name="edit", name="edit",
description="Змінити параметри особистого каналу", description="Змінити параметри особистого каналу",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
@option("name", description="Назва каналу") @option("name", description="Назва каналу")
@option("reactions", description="Дозволити реакції") @option("reactions", description="Дозволити реакції")
@ -130,9 +131,13 @@ class CustomChannels(commands.Cog):
async def custom_channel_edit_cmd( async def custom_channel_edit_cmd(
self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool self, ctx: ApplicationContext, name: str, reactions: bool, threads: bool
) -> None: ) -> None:
"""Command /customchannel edit <name> <reactions> <threads>
Command to change properties of a custom channel.
"""
holo_user_ctx: HoloUser = HoloUser(ctx.user) holo_user_ctx: HoloUser = HoloUser(ctx.user)
custom_channel: Union[TextChannel, None] = ds_utils.get( custom_channel: TextChannel | None = ds_utils.get(
ctx.guild.channels, id=holo_user_ctx.customchannel ctx.guild.channels, id=holo_user_ctx.customchannel
) )
@ -167,12 +172,15 @@ class CustomChannels(commands.Cog):
@custom_channel_group.command( @custom_channel_group.command(
name="remove", name="remove",
description="Відібрати канал, знищуючи його, та частково повернути кошти", description="Відібрати канал, знищуючи його, та частково повернути кошти",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
@option("confirm", description="Підтвердження операції") @option("confirm", description="Підтвердження операції")
async def custom_channel_remove_cmd( async def custom_channel_remove_cmd(
self, ctx: ApplicationContext, confirm: bool = False self, ctx: ApplicationContext, confirm: bool = False
) -> None: ) -> None:
"""Command /customchannel remove [<confirm>]
Command to remove a custom channel. Requires additional confirmation."""
holo_user_ctx: HoloUser = 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
@ -189,7 +197,7 @@ class CustomChannels(commands.Cog):
await ctx.defer() await ctx.defer()
custom_channel: Union[TextChannel, None] = ds_utils.get( custom_channel: TextChannel | None = ds_utils.get(
ctx.guild.channels, id=holo_user_ctx.customchannel ctx.guild.channels, id=holo_user_ctx.customchannel
) )
@ -234,5 +242,5 @@ class CustomChannels(commands.Cog):
) )
def setup(client: PycordBot) -> None: def setup(client: HoloBot) -> None:
client.add_cog(CustomChannels(client)) client.add_cog(CustomChannels(client))

41
cogs/data.py
View File

@ -1,41 +1,43 @@
import logging import logging
from logging import Logger
from os import makedirs from os import makedirs
from pathlib import Path from pathlib import Path
from typing import Union, List, Dict, Any from typing import List, Dict, Any
from uuid import uuid4 from uuid import uuid4
from discord import ApplicationContext, Embed, File, option, Role, TextChannel 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
from libbot import config_get from libbot.utils import config_get, json_write
from libbot.pycord.classes import PycordBot
from libbot.sync import config_get as sync_config_get
from libbot.sync import json_write as sync_json_write
from classes.holo_bot import HoloBot
from classes.holo_user import HoloUser from classes.holo_user import HoloUser
from enums 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
logger = logging.getLogger(__name__) logger: Logger = logging.getLogger(__name__)
class Data(commands.Cog): class Data(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: HoloBot):
self.client: PycordBot = client self.client: HoloBot = client
data: SlashCommandGroup = SlashCommandGroup("data", "Керування даними користувачів") data: SlashCommandGroup = SlashCommandGroup("data", "Керування даними користувачів")
@data.command( @data.command(
name="export", name="export",
description="Експортувати дані", description="Експортувати дані",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
@option( @option(
"kind", description="Тип даних, які треба експортувати", choices=["Користувачі"] "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
) )
async def data_export_cmd(self, ctx: ApplicationContext, kind: str) -> None: async def data_export_cmd(self, ctx: ApplicationContext, kind: str) -> None:
"""Command /data export <kind>
Command to export specific kind of data."""
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
@ -55,10 +57,10 @@ class Data(commands.Cog):
) )
) )
mod_role: Union[Role, None] = ds_utils.get( mod_role: 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: Union[TextChannel, None] = ds_utils.get( admin_chan: 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"),
) )
@ -93,19 +95,26 @@ class Data(commands.Cog):
} }
) )
sync_json_write(users, Path(f"tmp/{uuid}")) # Temporary file must be written synchronously,
# otherwise it will not be there when ctx.respond() is be called
json_write(users, Path(f"tmp/{uuid}"))
await ctx.respond(file=File(Path(f"tmp/{uuid}"), filename="users.json")) await ctx.respond(file=File(Path(f"tmp/{uuid}"), filename="users.json"))
@data.command( @data.command(
name="migrate", name="migrate",
description="Мігрувати всіх користувачів до бази", description="Мігрувати всіх користувачів до бази",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
@option( @option(
"kind", description="Тип даних, які треба експортувати", choices=["Користувачі"] "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"]
) )
async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str) -> None: async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str) -> None:
"""Command /migrate <kind>
Command to migrate specific kind of data.
Migration of users in this case means creation of their DB entries."""
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
@ -125,10 +134,10 @@ class Data(commands.Cog):
) )
) )
mod_role: Union[Role, None] = ds_utils.get( mod_role: 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: Union[TextChannel, None] = ds_utils.get( admin_chan: 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"),
) )
@ -179,5 +188,5 @@ class Data(commands.Cog):
) )
def setup(client: PycordBot) -> None: def setup(client: HoloBot) -> None:
client.add_cog(Data(client)) client.add_cog(Data(client))

18
cogs/fun.py
View File

@ -2,10 +2,9 @@ import logging
from discord import ApplicationContext, Embed, User, option, slash_command from discord import ApplicationContext, Embed, User, option, slash_command
from discord.ext import commands from discord.ext import commands
from libbot import config_get from libbot.utils import config_get
from libbot.pycord.classes import PycordBot
from libbot.sync import config_get as sync_config_get
from classes.holo_bot import HoloBot
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
@ -13,21 +12,24 @@ logger = logging.getLogger(__name__)
class Fun(commands.Cog): class Fun(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: HoloBot):
self.client: PycordBot = client self.client: HoloBot = client
@slash_command( @slash_command(
name="action", name="action",
description="Провести над користувачем РП дію", description="Провести над користувачем РП дію",
guild_ids=[sync_config_get("guild")], guild_ids=[config_get("guild")],
) )
@option( @option(
"type", "type",
description="Тип дії, яку хочете провести з користувачем", description="Тип дії, яку хочете провести з користувачем",
choices=sync_config_get("actions").keys(), choices=config_get("actions").keys(),
) )
@option("user", description="Користувач") @option("user", description="Користувач")
async def action_cmd(self, ctx: ApplicationContext, type: str, user: User) -> None: async def action_cmd(self, ctx: ApplicationContext, type: str, user: User) -> None:
"""Command /action <type> <user>
Command to perform some RP action on a user and send them a GIF."""
await ctx.defer() await ctx.defer()
action: str = await config_get("category", "actions", type) action: str = await config_get("category", "actions", type)
@ -54,5 +56,5 @@ class Fun(commands.Cog):
await ctx.respond(embed=embed) await ctx.respond(embed=embed)
def setup(client: PycordBot) -> None: def setup(client: HoloBot) -> None:
client.add_cog(Fun(client)) client.add_cog(Fun(client))

36
cogs/logger.py
View File

@ -1,20 +1,24 @@
from typing import Dict, Any, Union import logging
from typing import Dict, Any
from discord import Member, Message, TextChannel from discord import Member, Message, TextChannel, MessageType
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.utils import config_get
from libbot.pycord.classes import PycordBot
from classes.holo_bot import HoloBot
from modules.database import col_users from modules.database import col_users
logger = logging.getLogger(__name__)
class Logger(commands.Cog): class Logger(commands.Cog):
def __init__(self, client: PycordBot): def __init__(self, client: HoloBot):
self.client: PycordBot = client self.client: HoloBot = client
@commands.Cog.listener() @commands.Cog.listener()
async def on_message(self, message: Message): async def on_message(self, message: Message):
"""Message listener. All actions on messages remain here for now."""
if ( if (
(message.author != self.client.user) (message.author != self.client.user)
and (message.author.bot is False) and (message.author.bot is False)
@ -31,17 +35,31 @@ class Logger(commands.Cog):
await col_users.insert_one(document=user) await col_users.insert_one(document=user)
if (
(message.type == MessageType.thread_created)
and (message.channel is not None)
and (
await col_users.count_documents({"customchannel": message.channel.id})
> 0
)
):
await message.delete()
@commands.Cog.listener() @commands.Cog.listener()
async def on_member_join(self, member: Member) -> None: async def on_member_join(self, member: Member) -> None:
welcome_chan: Union[TextChannel, None] = ds_utils.get( """Member join handler. All actions on member join remain here for now."""
welcome_chan: 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: Union[TextChannel, None] = ds_utils.get( rules_chan: 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"),
) )
if welcome_chan is None:
logger.warning("Could not find a welcome channel by its id")
if ( if (
(member != self.client.user) (member != self.client.user)
and (member.bot is False) and (member.bot is False)
@ -65,5 +83,5 @@ class Logger(commands.Cog):
await col_users.insert_one(document=user) await col_users.insert_one(document=user)
def setup(client: PycordBot) -> None: def setup(client: HoloBot) -> None:
client.add_cog(Logger(client)) client.add_cog(Logger(client))

58
cogs/utility.py Normal file
View File

@ -0,0 +1,58 @@
import logging
from logging import Logger
from discord import Activity, ActivityType
from discord.ext import commands
from libbot.utils import config_get
from classes.holo_bot import HoloBot
logger: Logger = logging.getLogger(__name__)
class Utility(commands.Cog):
def __init__(self, client: HoloBot):
self.client: HoloBot = client
@commands.Cog.listener()
async def on_ready(self) -> None:
"""Listener for the event when bot connects to Discord and becomes "ready"."""
logger.info("Logged in as %s", self.client.user)
activity_type: str = await config_get("type", "status")
activity_message: str = await config_get("message", "status")
if activity_type == "playing":
await self.client.change_presence(
activity=Activity(type=ActivityType.playing, name=activity_message)
)
elif activity_type == "watching":
await self.client.change_presence(
activity=Activity(type=ActivityType.watching, name=activity_message)
)
elif activity_type == "listening":
await self.client.change_presence(
activity=Activity(type=ActivityType.listening, name=activity_message)
)
elif activity_type == "streaming":
await self.client.change_presence(
activity=Activity(type=ActivityType.streaming, name=activity_message)
)
elif activity_type == "competing":
await self.client.change_presence(
activity=Activity(type=ActivityType.competing, name=activity_message)
)
elif activity_type == "custom":
await self.client.change_presence(
activity=Activity(type=ActivityType.custom, name=activity_message)
)
else:
return
logger.info(
"Set activity type to %s with message %s", activity_type, activity_message
)
def setup(client: HoloBot) -> None:
client.add_cog(Utility(client))

64
main.py
View File

@ -1,12 +1,12 @@
import logging import logging
import sys import sys
from logging import Logger from logging import Logger
from pathlib import Path
from discord import Activity, ActivityType from discord import LoginFailure, Intents
from libbot import config_get from libbot.utils import config_get
from libbot.sync import config_get as sync_config_get
from modules.client import client from classes.holo_bot import HoloBot
from modules.scheduler import scheduler from modules.scheduler import scheduler
logging.basicConfig( logging.basicConfig(
@ -25,53 +25,25 @@ except ImportError:
pass pass
@client.event
async def on_ready() -> None:
logger.info("Logged in as %s", client.user)
activity_type: str = await config_get("type", "status")
activity_message: str = await config_get("message", "status")
if activity_type == "playing":
await client.change_presence(
activity=Activity(type=ActivityType.playing, name=activity_message)
)
elif activity_type == "watching":
await client.change_presence(
activity=Activity(type=ActivityType.watching, name=activity_message)
)
elif activity_type == "listening":
await client.change_presence(
activity=Activity(type=ActivityType.listening, name=activity_message)
)
elif activity_type == "streaming":
await client.change_presence(
activity=Activity(type=ActivityType.streaming, name=activity_message)
)
elif activity_type == "competing":
await client.change_presence(
activity=Activity(type=ActivityType.competing, name=activity_message)
)
elif activity_type == "custom":
await client.change_presence(
activity=Activity(type=ActivityType.custom, name=activity_message)
)
else:
return
logger.info(
"Set activity type to %s with message %s", activity_type, activity_message
)
def main() -> None: def main() -> None:
if not Path("config.json").exists():
logger.error(
"Config file is missing: Make sure the configuration file 'config.json' is in place."
)
sys.exit()
intents: Intents = Intents().all()
client: HoloBot = HoloBot(intents=intents, scheduler=scheduler)
client.load_extension("cogs") client.load_extension("cogs")
try: try:
scheduler.start() client.run(config_get("bot_token", "bot"))
client.run(sync_config_get("bot_token", "bot")) except LoginFailure as exc:
logger.error("Provided bot token is invalid: %s", exc)
except KeyboardInterrupt: except KeyboardInterrupt:
scheduler.shutdown() logger.info("KeyboardInterrupt received: Shutting down gracefully.")
finally:
sys.exit() sys.exit()

10
modules/client.py
View File

@ -1,10 +0,0 @@
from discord import Intents
from libbot.pycord.classes import PycordBot
from modules.scheduler import scheduler
intents: Intents = Intents().all()
intents.members = True
client: PycordBot = PycordBot(intents=intents, scheduler=scheduler)

4
modules/database.py
View File

@ -1,12 +1,12 @@
from typing import Dict, Any from typing import Dict, Any
from async_pymongo import AsyncClient, AsyncCollection, AsyncDatabase from async_pymongo import AsyncClient, AsyncCollection, AsyncDatabase
from libbot.sync import config_get as sync_config_get from libbot.utils import config_get
from pymongo import MongoClient from pymongo import MongoClient
from pymongo.synchronous.collection import Collection from pymongo.synchronous.collection import Collection
from pymongo.synchronous.database import Database from pymongo.synchronous.database import Database
db_config: Dict[str, Any] = sync_config_get("database") db_config: Dict[str, Any] = config_get("database")
con_string: str = ( con_string: str = (
"mongodb://{0}:{1}/{2}".format( "mongodb://{0}:{1}/{2}".format(

4
modules/utils_sync.py
View File

@ -1,9 +1,7 @@
from typing import Union
from discord import Member, User from discord import Member, User
def guild_name(member: Union[Member, User]) -> str: def guild_name(member: Member | User) -> str:
if isinstance(member, User): if isinstance(member, User):
return member.name return member.name

3
requirements.txt
View File

@ -5,6 +5,7 @@ requests>=2.32.2
aiofiles~=24.1.0 aiofiles~=24.1.0
apscheduler>=3.10.0 apscheduler>=3.10.0
async_pymongo==0.1.11 async_pymongo==0.1.11
libbot[speed,pycord]==3.2.3 libbot[speed,pycord]==4.0.0
typing-extensions~=4.12.2
ujson~=5.10.0 ujson~=5.10.0
WaifuPicsPython==0.2.0 WaifuPicsPython==0.2.0