hypoxie/krekbot-moderation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Исправлены все ошибки PyLance

Browse Source
This commit is contained in:
HypoxiE
2025-08-21 10:57:51 +07:00
parent ec655ba4ac
commit 044134041b
6 changed files with 126 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
src/CoreMod.py
View File

@@ -192,7 +192,7 @@ class AnyBots(commands.Bot):
matches = re.findall(pattern, time_str)
for value, _, unit in matches:
time_units[unit] += float(value)
time_units[unit] += value
return FormatedTime(time_units)
@@ -208,6 +208,8 @@ class AnyBots(commands.Bot):
krekchat = await self.fetch_guild(self.krekchat.id)
bt_channel = await krekchat.fetch_channel(self.bots_talk_protocol_channel_id)
if not isinstance(bt_channel, disnake.TextChannel):
raise ValueError("bt_channel не найден")
punishment_keys = ['type', 'options', 'severity', 'member', 'moderator']
complaint_keys = ['type', 'options', 'accepted', 'attack_member', 'defence_member', 'moderator']
@@ -321,7 +323,7 @@ class MainBot(AnyBots):
return wrapper
@tasks.loop(seconds=30)
@catch_exceptions.__func__
@catch_exceptions
async def watchdog(self):
for loop in self.loops:
@@ -335,7 +337,7 @@ class MainBot(AnyBots):
@tasks.loop(seconds=60)
@catch_exceptions.__func__
@catch_exceptions
async def SendingDeferredMessages(self):
async with self.DataBaseManager.session() as session:
async with session.begin():
@@ -351,16 +353,18 @@ class MainBot(AnyBots):
await session.delete(message)
@tasks.loop(seconds=3600)
@catch_exceptions.__func__
@catch_exceptions
async def MakeBackups(self):
backup_file = await self.DataBaseManager.pg_dump()
krekchat = await self.fetch_guild(self.krekchat.id)
backups_channel = await krekchat.fetch_channel(self.databases_backups_channel_id)
if not isinstance(backups_channel, disnake.TextChannel):
raise ValueError("backups_channel не найден")
await backups_channel.send(content=f"Бэкап бд за {datetime.datetime.now()}:", file=disnake.File(backup_file))
@tasks.loop(seconds=60)
@catch_exceptions.__func__
@catch_exceptions
async def CheckDataBases(self):
self.krekchat = await self.fetch_guild(self.krekchat.id)
members = [i async for i in self.krekchat.fetch_members(limit=None)]
@@ -547,29 +551,30 @@ class MainBot(AnyBots):
if not inter.response.is_done():...
#await inter.response.send_message(embed=self.ErrEmbed(description=f'Ответ не был отправлен, возможно, кнопка перестала действовать'), ephemeral=True)
async def on_message(self, msg):
async def on_message(self, message):
if msg.author.bot or not self.task_start or not self.ready_once.is_set():
return 0
if message.author.bot or not self.task_start or not self.ready_once.is_set():
return
if msg.author.id == 479210801891115009 and msg.content == "botsoff":
await msg.reply(embed=self.AnswEmbed(description=f'Бот отключён', colour=0xff9900))
if message.author.id == 479210801891115009 and message.content == "botsoff":
await message.reply(embed=self.AnswEmbed(description=f'Бот отключён', colour=0xff9900))
await self.BotOff()
return 0
if type(msg.channel).__name__!="DMChannel" and re.match(r"^⚠️?жалоба-от-(.+)-на-(.+)$", msg.channel.name):
log_reports = disnake.utils.get(msg.guild.channels, id=1242373230384386068)
return
if type(message.channel).__name__!="DMChannel" and re.match(r"^⚠️?жалоба-от-(.+)-на-(.+)$", message.channel.name):
log_reports = disnake.utils.get(message.guild.channels, id=1242373230384386068)
if isinstance(log_reports, disnake.TextChannel):
files=[]
for att in msg.attachments:
for att in message.attachments:
files = files + [await att.to_file()]
log_mess = await log_reports.send(f"Чат: `{msg.channel.name}`({msg.channel.id}).\n"
f"Автор: `{msg.author.name} ({msg.author.id})`\n" +
(f"Сообщение: ```{msg.content}```\n" if msg.content else ""),
log_mess = await log_reports.send(f"Чат: `{message.channel.name}`({message.channel.id}).\n"
f"Автор: `{message.author.name} ({message.author.id})`\n" +
(f"Сообщение: ```{message.content}```\n" if message.content else ""),
files = files)
return 0
return
async with self.DataBaseManager.session() as session:
if (await self.DataBaseManager.model_classes['staff_users'].is_admin_or_moder_by_id(msg.author.id, self.DataBaseManager, session)):
return 0
if (await self.DataBaseManager.model_classes['staff_users'].is_admin_or_moder_by_id(message.author.id, self.DataBaseManager, session)):
return
def extract_root_domain(url):
ext = tldextract.extract(url)
@@ -577,10 +582,12 @@ class MainBot(AnyBots):
return None
return f"{ext.domain}.{ext.suffix}".lower()
log = disnake.utils.get(msg.guild.channels, id=893065482263994378)
log = disnake.utils.get(message.guild.channels, id=893065482263994378)
if not isinstance(log, disnake.TextChannel):
raise TypeError("Проверь канал логов для сомнительных ссылок")
url_pattern = re.compile(r'https?://[^\s]+')
links = re.findall(url_pattern, msg.content)
links = re.findall(url_pattern, message.content)
аllowed_domains_model = self.DataBaseManager.model_classes['аllowed_domains']
async with self.DataBaseManager.session() as session:
for link in links:
@@ -589,28 +596,28 @@ class MainBot(AnyBots):
link_in_wl = (await session.execute(stmt)).scalars().first()
if link_in_wl is None:
await log.send(f"{msg.author.mention}({msg.author.id}) отправил в чат {msg.channel.mention} сомнительную ссылку, которой нет в вайлисте:```{msg.content}```")
mess = await msg.reply(embed=self.ErrEmbed(description=f'Этой ссылки нет в белом списке, но заявка на добавление уже отправлена. Если это срочно, свяжитесь с разработчиком или модераторами.'))
await msg.delete()
await log.send(f"{message.author.mention}({message.author.id}) отправил в чат {message.channel.mention} сомнительную ссылку, которой нет в вайлисте:```{message.content}```")
mess = await message.reply(embed=self.ErrEmbed(description=f'Этой ссылки нет в белом списке, но заявка на добавление уже отправлена. Если это срочно, свяжитесь с разработчиком или модераторами.'))
await message.delete()
await asyncio.sleep(20)
await mess.delete()
return 1
return
message_words = msg.content.replace("/", " ").split(" ")
message_words = message.content.replace("/", " ").split(" ")
if "discord.gg" in message_words:
for i in range(len(message_words)):
if message_words[i]=="discord.gg" and not msg.author.bot:
if message_words[i]=="discord.gg" and not message.author.bot:
try:
inv = await self.fetch_invite(url = "https://discord.gg/"+message_words[i+1])
if inv.guild.id != 490445877903622144:
await log.send(f"{msg.author.mention}({msg.author.id}) отправил в чат {msg.channel.mention} сомнительную ссылку на сервер '{inv.guild.name}':```{msg.content}```")
mess = await msg.reply(embed=self.ErrEmbed(description=f'Ссылки-приглашения запрещены!', colour=0xff9900))
await msg.delete()
if (isinstance(inv.guild, disnake.PartialInviteGuild) or isinstance(inv.guild, disnake.Guild)) and inv.guild.id != 490445877903622144:
await log.send(f"{message.author.mention}({message.author.id}) отправил в чат {message.channel.mention} сомнительную ссылку на сервер '{inv.guild.name}':```{message.content}```")
mess = await message.reply(embed=self.ErrEmbed(description=f'Ссылки-приглашения запрещены!', colour=0xff9900))
await message.delete()
await asyncio.sleep(20)
await mess.delete()
break
except disnake.errors.NotFound:
await log.send(f"{msg.author.mention}({msg.author.id}) отправил в чат {msg.channel.mention} [сомнительную ссылку]({msg.jump_url}) на неизвестный сервер:```{msg.content}```")
await log.send(f"{message.author.mention}({message.author.id}) отправил в чат {message.channel.mention} [сомнительную ссылку]({message.jump_url}) на неизвестный сервер:```{message.content}```")
@@ -681,12 +688,14 @@ async def main():
except Exception as e:
print(f"Произошла критическая ошибка: {e}")
finally:
if bot is not None:
await bot.BotOff()
for bot in all_bots:
if not bot.is_closed():
await bot.close()
if DataBase is not None:
await DataBase.close()
current_task = asyncio.current_task()

46
src/cogs/administrators.py
View File

@@ -30,14 +30,25 @@ class AdminModule(commands.Cog):
@commands.slash_command(name="bot_mod_off")
async def BotModOff(self, ctx: disnake.ApplicationCommandInteraction):
if isinstance(ctx.author, disnake.User):
await ctx.send(embed=self.client.ErrEmbed(description=f'Невозможно в личных сообщениях'), ephemeral=True)
return
if self.client.me in ctx.author.roles:
await ctx.send(embed=self.client.SuccessEmbed(description=f'Бот отключён', colour=0xff9900), ephemeral=True)
await ctx.send(embed=self.client.SuccessEmbed(description=f'Бот отключён'), ephemeral=True)
await self.client.BotOff()
else:
await ctx.send(embed=self.client.ErrEmbed(description=f'Не допустимо', colour=0xff9900), ephemeral=True)
await ctx.send(embed=self.client.ErrEmbed(description=f'Не допустимо'), ephemeral=True)
@commands.slash_command(name="очистка", administrator=True)
async def clear(self, ctx: disnake.AppCmdInter, count: int):
if isinstance(ctx.author, disnake.User):
await ctx.send(embed=self.client.ErrEmbed(description=f'Невозможно в личных сообщениях'), ephemeral=True)
return
if isinstance(ctx.channel, (disnake.DMChannel, disnake.GroupChannel, disnake.PartialMessageable)):
await ctx.send(embed=self.client.ErrEmbed(description=f'Неверный тип канала!'), ephemeral=True)
return
if self.client.me in ctx.author.roles:
await ctx.channel.purge(limit=count)
await ctx.send(embed = self.client.SuccessEmbed(description = f'очищено {count} сообщений!', colour = 0xff9900), ephemeral=True)
@@ -51,6 +62,9 @@ class AdminModule(commands.Cog):
is_admin: bool = commands.Param(description="Имеют ли пользователи в этой роли права администратора? *(только при создании)", name="администратор", default=False),
is_moder: bool = commands.Param(description="Имеют ли пользователи в этой роли права модератора? *(только при создании)", name="модератор", default=False),
delete_branch: str = commands.Param(description="Вы уверены, что хотите удалить ветвь? Для подтверждения впишите \"уверен\"", name="удаление", default=None)):
if isinstance(ctx.author, disnake.User):
await ctx.send(embed=self.client.ErrEmbed(description=f'Невозможно в личных сообщениях'), ephemeral=True)
return
if not self.client.me in ctx.author.roles:
await ctx.send(embed = self.client.ErrEmbed(description = f'Недостаточно прав', colour = 0xff9900), ephemeral=True)
@@ -101,16 +115,20 @@ class AdminModule(commands.Cog):
@commands.slash_command(description="Позволяет менять/создавать/удалять роли в системе персонала", name="правка_роли", administrator=True)
async def edit_role(self, ctx: disnake.AppCmdInter, roleid: str = commands.Param(description="Укажите id роли (используются идентификаторы дискорда)", name="id"),
async def edit_role(self, ctx: disnake.AppCmdInter, roleid_str: str = commands.Param(description="Укажите id роли (используются идентификаторы дискорда)", name="id"),
staffsalary: int = commands.Param(description="Укажите зарплату этой роли", name="зарплата", default=0),
branchid: int = commands.Param(description="Укажите id ветви для этой роли *(только при создании)", name="ветвь", default=None),
layer: int = commands.Param(description="Укажите слой этой роли в ветке (у ролей нижних слоёв есть власть над верхними)", name="слой", default=None),
delete_role: str = commands.Param(description="Вы уверены, что хотите удалить роль из системы? Для подтверждения впишите \"уверен\"", name="удаление", default=None),):
if isinstance(ctx.author, disnake.User):
await ctx.send(embed=self.client.ErrEmbed(description=f'Невозможно в личных сообщениях'), ephemeral=True)
return
if not self.client.me in ctx.author.roles:
await ctx.send(embed = self.client.ErrEmbed(description = f'Недостаточно прав', colour = 0xff9900), ephemeral=True)
return 1
roleid = int(roleid)
roleid = int(roleid_str)
staff_roles_model = self.DataBaseManager.models['staff_roles'].m
async with self.DataBaseManager.session() as session:
@@ -153,13 +171,17 @@ class AdminModule(commands.Cog):
return 0
@commands.slash_command(description="Позволяет создавать/удалять пользователей в системе персонала", name="правка_пользователя", administrator=True)
async def edit_member(self, ctx: disnake.AppCmdInter, userid: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="id"),
async def edit_member(self, ctx: disnake.AppCmdInter, userid_str: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="id"),
delete_user: str = commands.Param(description="Вы уверены, что хотите удалить пользователя из системы? Для подтверждения впишите \"уверен\"", name="удаление", default=None)):
if isinstance(ctx.author, disnake.User):
await ctx.send(embed=self.client.ErrEmbed(description=f'Невозможно в личных сообщениях'), ephemeral=True)
return
if not self.client.me in ctx.author.roles:
await ctx.send(embed = self.client.ErrEmbed(description = f'Недостаточно прав', colour = 0xff9900), ephemeral=True)
return 1
userid = int(userid)
userid = int(userid_str)
staff_users_model = self.DataBaseManager.models['staff_users'].m
async with self.DataBaseManager.session() as session:
@@ -187,15 +209,19 @@ class AdminModule(commands.Cog):
return 1
@commands.slash_command(description="!!ВАЖНО!! ИСПОЛЬЗОВАНИЕ ТОЛЬКО В ЭКСТРЕННЫХ СЛУЧАЯХ! Назначает пользователей на роль", name="назначить_пользователя", administrator=True)
async def appoint_member(self, ctx: disnake.AppCmdInter, userid: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="пользователь"),
roleid: str = commands.Param(description="Укажите id роли (используются идентификаторы дискорда)", name="роль"),
async def appoint_member(self, ctx: disnake.AppCmdInter, userid_str: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="пользователь"),
roleid_str: str = commands.Param(description="Укажите id роли (используются идентификаторы дискорда)", name="роль"),
description: str = commands.Param(description="Описание", name="описание", default=None)):
if isinstance(ctx.author, disnake.User):
await ctx.send(embed=self.client.ErrEmbed(description=f'Невозможно в личных сообщениях'), ephemeral=True)
return
if not self.client.me in ctx.author.roles:
await ctx.send(embed = self.client.ErrEmbed(description = f'Недостаточно прав', colour = 0xff9900), ephemeral=True)
return 1
userid = int(userid)
roleid = int(roleid)
userid = int(userid_str)
roleid = int(roleid_str)
async with self.DataBaseManager.session() as session:
async with session.begin():

31
src/cogs/moderators.py
View File

@@ -68,6 +68,7 @@ class ModerModule(commands.Cog):
def __init__(self, title, member):
self.member = member
self.title = title
components = []
if title.split(":")[0] == "textmute" or title.split(":")[0] == "voicemute" or title.split(":")[0] == "ban":
components = [
disnake.ui.TextInput(label="Время", placeholder="Например: 1д15ч9мин", custom_id="time", style=disnake.TextInputStyle.short, max_length=100),
@@ -89,10 +90,13 @@ class ModerModule(commands.Cog):
async def callback(self, interaction: disnake.Interaction):
if interaction.guild is None:
raise TypeError("interaction.guild is None")
async def voicemute(interaction: disnake.MessageInteraction, member: disnake.Member, time, reason):
role = self.client.voice_mute
await member.add_roles(role)
await member.move_to(None)
await member.move_to(None) # type: ignore
async with DataBaseManager.session() as session:
async with session.begin():
new_punishment = models['punishment_mutes_voice'](user_id = member.id, reason = reason, time_end = time, time_warn = None, moderator_id = interaction.author.id)
@@ -106,7 +110,7 @@ class ModerModule(commands.Cog):
session.add(new_punishment)
async def ban(interaction: disnake.MessageInteraction, member: disnake.Member, time, reason):
role = self.client.ban_role
await member.move_to(None)
await member.move_to(None) # type: ignore
if time-datetime.datetime.timestamp(datetime.datetime.now())>0:
await member.add_roles(role)
async with DataBaseManager.session() as session:
@@ -119,6 +123,9 @@ class ModerModule(commands.Cog):
async with session.begin():
new_punishment = models['punishment_perms'](user_id = member.id, reason = reason, moderator_id = interaction.author.id)
session.add(new_punishment)
if interaction.guild is None:
raise TypeError("interaction.guild is None")
for channel in interaction.guild.channels:
if isinstance(channel, disnake.TextChannel):
await channel.purge(limit=10, check=lambda m: m.author==member)
@@ -147,6 +154,10 @@ class ModerModule(commands.Cog):
newnick = ""
if self.title.split(':')[0] == "reprimand":
embed.add_field(name = 'Ветка', value = self.title.split(':')[1], inline = False)
if isinstance(interaction, disnake.Interaction):
raise TypeError("modal interaction is interaction")
for key, value in interaction.text_values.items():
if key == "time":
formated_time = self.client.TimeFormater(value)
@@ -295,6 +306,8 @@ class ModerModule(commands.Cog):
return self
async def callback(self, interaction:disnake.MessageInteraction):
if interaction.values is None or interaction.guild is None:
raise TypeError("interaction.values is None or interaction.guild is None")
values = interaction.values[0]
try:
pentype, member, penaltid = values.split(":")
@@ -302,6 +315,8 @@ class ModerModule(commands.Cog):
return
if values:
logs_channel = disnake.utils.get(interaction.guild.channels, id = 490730651629387776)
if not isinstance(logs_channel, disnake.TextChannel):
raise ValueError("logs_channel is None")
async with DataBaseManager.session() as session:
async with session.begin():
@@ -471,9 +486,9 @@ class ModerModule(commands.Cog):
@commands.slash_command(description="Позволяет повысить пользователя в указанной ветке", name="повысить", administrator=True)
async def promote(self, ctx: disnake.AppCmdInter, branchid: int = commands.Param(description="Укажите id ветви", name="ветвь", default = None),
userid: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="пользователь")):
userid_str: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="пользователь")):
userid = int(userid)
userid = int(userid_str)
async with self.DataBaseManager.session() as session:
async with session.begin():
@@ -544,6 +559,8 @@ class ModerModule(commands.Cog):
if member_role is not None:
member_role_index = next((i for i, role in enumerate(branch_roles) if role.id == member_role.role.id), None)
if member_role_index is None:
raise ValueError("member_role_index is None")
if member_role_index + 1 >= len(branch_roles):
await ctx.send(embed = self.client.ErrEmbed(description = f'Этот пользователь уже находится на самой высокой должности в данной ветви'))
@@ -584,9 +601,9 @@ class ModerModule(commands.Cog):
@commands.slash_command(description="Позволяет понизить пользователя в указанной ветке", name="понизить", administrator=True)
async def demote(self, ctx: disnake.AppCmdInter, branchid: int = commands.Param(description="Укажите id ветви", name="ветвь", default = None),
userid: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="пользователь")):
userid_str: str = commands.Param(description="Укажите id пользователя (используются идентификаторы дискорда)", name="пользователь")):
userid = int(userid)
userid = int(userid_str)
async with self.DataBaseManager.session() as session:
async with session.begin():
@@ -657,6 +674,8 @@ class ModerModule(commands.Cog):
if member_role is not None:
member_role_index = next((i for i, role in enumerate(branch_roles) if role.id == member_role.role.id), None)
if member_role_index is None:
raise ValueError("member_role_index is None")
if member_role_index == 0:
stmt = (

12
src/cogs/users.py
View File

@@ -27,10 +27,12 @@ class UIModule(commands.Cog):
print(f'KrekModBot UI module activated')
@commands.slash_command(description="Показывает действительные наказания пользователя", name="наказания")
async def penalties(self, ctx: disnake.AppCmdInter, member: disnake.Member = None):
async def penalties(self, ctx: disnake.AppCmdInter, input_member: disnake.Member | None= None):
models = self.client.DataBaseManager.model_classes
if not member:
if input_member is None:
member = ctx.author
else:
member = input_member
embed = self.client.AnswEmbed(title="Наказания", description = f"{member.mention}")
embed.set_thumbnail(url=member.avatar)
@@ -181,7 +183,13 @@ class UIModule(commands.Cog):
return
mentions = []
if ctx.guild is None:
await ctx.send(embed = client.ErrEmbed(description = f'Нельзя подать жалобу в личных сообщениях!', colour = 0xFF4500), ephemeral=True)
return
report_channel = disnake.utils.get(ctx.guild.channels, id = 1219644036378394746)
if not isinstance(report_channel, disnake.TextChannel):
raise ValueError("not isinstance(report_channel, disnake.TextChannel)")
highest = [i for i in client.hierarchy if i in member.roles][0]
for i in range(0, client.hierarchy.index(highest)):

2
src/database/db_classes.py
View File

@@ -14,7 +14,7 @@ class Base(DeclarativeBase):
def get_table_name(self):
return self.__tablename__
def to_dict(self, exclude: list[str] = None):
def to_dict(self, exclude: list[str] | None = None):
"""Конвертирует модель в словарь, исключая указанные поля."""
if exclude is None:
exclude = []

12
src/managers/DataBaseManager.py
View File

@@ -100,12 +100,12 @@ class DatabaseManager:
return backup_file
async def pg_restore(self, echo = False, backup_file = 'backups/backup_file.backup'):
conn = await self.DataBaseManager.engine.connect()
db_name = self.DataBaseManager.engine.url.database
user = self.DataBaseManager.engine.url.username
host = self.DataBaseManager.engine.url.host
port = self.DataBaseManager.engine.url.port
password = self.DataBaseManager.engine.url.password
conn = await self.engine.connect()
db_name = self.engine.url.database
user = self.engine.url.username
host = self.engine.url.host
port = self.engine.url.port
password = self.engine.url.password
os.environ['PGPASSWORD'] = password # Установка пароля для подключения
command = [