import logging from os import makedirs from pathlib import Path from uuid import uuid4 from discord import ApplicationContext, Embed, File, option from discord import utils as ds_utils from discord.commands import SlashCommandGroup from discord.ext import commands from classes.holo_user import HoloUser from enums.colors import Color from modules.database import col_users from modules.utils import config_get from modules.utils_sync import config_get_sync, guild_name, json_write_sync logger = logging.getLogger(__name__) class Data(commands.Cog): def __init__(self, client): self.client = client data = SlashCommandGroup("data", "Керування даними користувачів") @data.command( name="export", description="Експортувати дані", guild_ids=[config_get_sync("guild")], ) @option( "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"] ) async def data_export_cmd(self, ctx: ApplicationContext, kind: str): await ctx.defer() holo_user = HoloUser(ctx.author) if (ctx.user.id in await config_get("admins")) or ( await holo_user.is_council(ctx.author) ): logging.info( f"Moderator {guild_name(ctx.user)} exported current users list" ) makedirs("tmp", exist_ok=True) uuid = str(uuid4()) if kind == "Користувачі": users = [] for member in ctx.guild.members: users.append( { "id": member.id, "nick": member.nick, "username": f"{member.name}#{member.discriminator}", "bot": member.bot, } ) json_write_sync(users, str(Path(f"tmp/{uuid}"))) await ctx.respond( file=File(str(Path(f"tmp/{uuid}")), filename="users.json") ) else: logging.info( f"User {guild_name(ctx.user)} tried to use /export 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("moderator", "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, ), ) @data.command( name="migrate", description="Мігрувати всіх користувачів до бази", guild_ids=[config_get_sync("guild")], ) @option( "kind", description="Тип даних, які треба експортувати", choices=["Користувачі"] ) async def data_migrate_cmd(self, ctx: ApplicationContext, kind: str): await ctx.defer() holo_user = HoloUser(ctx.author) if (ctx.user.id in await config_get("admins")) or ( await holo_user.is_council(ctx.author) ): logging.info( f"Moderator {guild_name(ctx.user)} started migration of all members to the database" ) if kind == "Користувачі": for member in ctx.guild.members: if member.bot: continue if col_users.find_one({"user": member.id}) is None: user = {} defaults = await config_get("user", "defaults") user["user"] = member.id for key in defaults: user[key] = defaults[key] col_users.insert_one(document=user) logging.info( f"Added DB record for user {member.id} during migration" ) await ctx.respond( embed=Embed( title="Міграцію завершено", description="Всім користувачам сервера було створено записи в базі даних.", color=Color.success, ) ) else: logging.info( f"User {guild_name(ctx.user)} tried to use /migrate 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("moderator", "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, ), )