Добавил проект с учетом .gitignore
This commit is contained in:
259
src/cogs/admin.py
Normal file
259
src/cogs/admin.py
Normal file
@@ -0,0 +1,259 @@
|
||||
|
||||
import disnake
|
||||
from disnake.ext import commands
|
||||
from disnake.ext import tasks
|
||||
import requests
|
||||
import numpy as np
|
||||
import aiohttp
|
||||
|
||||
import asyncio
|
||||
import sqlite3
|
||||
import sys
|
||||
import os
|
||||
import copy
|
||||
import datetime
|
||||
import math
|
||||
import random
|
||||
import json
|
||||
import shutil
|
||||
from constants.global_constants import *
|
||||
from constants.rimagochi_constants import *
|
||||
|
||||
|
||||
def setup(bot):
|
||||
bot.add_cog(MainAdminModule(bot))
|
||||
|
||||
class MainAdminModule(commands.Cog):
|
||||
def __init__(self, client):
|
||||
self.client = client
|
||||
self.DataBaseManager = client.DataBaseManager
|
||||
|
||||
@commands.Cog.listener()
|
||||
async def on_ready(self):
|
||||
print(f'KrekFunBot admin module activated')
|
||||
self.krekchat = await self.client.fetch_guild(constants["krekchat"])
|
||||
self.me = disnake.utils.get(self.krekchat.roles, id=constants["me"])
|
||||
|
||||
|
||||
@commands.slash_command(name="bot_fun_off")
|
||||
async def BotFunOff(self, ctx: disnake.ApplicationCommandInteraction):
|
||||
if self.me in ctx.author.roles:
|
||||
await ctx.send(embed=disnake.Embed(description=f'Бот отключён', colour=0xff9900), ephemeral=True)
|
||||
await self.client.BotOff()
|
||||
else:
|
||||
await ctx.send(embed=disnake.Embed(description=f'Не допустимо', colour=0xff9900), ephemeral=True)
|
||||
|
||||
|
||||
@commands.slash_command(name="зарегистрировать_призовую_роль")
|
||||
async def RegisterAPrizeRoleSlash(self, ctx, role: disnake.Role):
|
||||
await ctx.response.defer(ephemeral=True)
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.edit_original_message(
|
||||
embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
|
||||
async with self.DataBaseManager.session() as session:
|
||||
async with session.begin():
|
||||
async with self.DataBaseManager.models['roles_prize'] as roles_prize_model:
|
||||
stmt = self.DataBaseManager.select(roles_prize_model).where(roles_prize_model.id == role.id)
|
||||
role_prize = (await session.execute(stmt)).scalars().first()
|
||||
if role_prize is None:
|
||||
prize_role = roles_prize_model(id = role.id)
|
||||
session.add(prize_role)
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Теперь роль {role.mention} зарегистрирована!',colour=0x2F3136))
|
||||
else:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Эта роль уже зарегистрирована!',colour=0x2F3136))
|
||||
|
||||
|
||||
|
||||
@commands.slash_command(name="выдать_призовую_роль")
|
||||
async def GiveAPrizeRoleSlash(self, ctx, role: disnake.Role, member: disnake.Member):
|
||||
await ctx.response.defer(ephemeral=True)
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
|
||||
async with self.DataBaseManager.session() as session:
|
||||
await self.client.UserUpdate(member, session = session)
|
||||
async with session.begin():
|
||||
async with self.DataBaseManager.models['roles_prize'] as roles_prize_model:
|
||||
|
||||
stmt = self.DataBaseManager.select(roles_prize_model).where(roles_prize_model.id == role.id)
|
||||
role_prize = (await session.execute(stmt)).scalars().first()
|
||||
if role_prize is None:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Данная роль не зарегистрирована как призовая', colour=0x2F3136))
|
||||
return 1
|
||||
|
||||
async with self.DataBaseManager.models['received_roles_prize'] as received_roles_prize_model:
|
||||
stmt = self.DataBaseManager.select(received_roles_prize_model).where(
|
||||
received_roles_prize_model.role_id == role.id,
|
||||
received_roles_prize_model.user_id == member.id
|
||||
)
|
||||
received_roles_prize = (await session.execute(stmt)).scalars().first()
|
||||
if not received_roles_prize is None:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'У {member.mention} уже есть роль {role.mention}', colour=0x2F3136))
|
||||
return 1
|
||||
|
||||
received_roles_prize = received_roles_prize_model(role_id = role.id, user_id = member.id)
|
||||
session.add(received_roles_prize)
|
||||
|
||||
embed = disnake.Embed(description=f'Роль {role.mention} успешно передана {member.mention}!', colour=0x2F3136)
|
||||
await member.add_roles(role)
|
||||
await ctx.edit_original_message(embed=embed)
|
||||
|
||||
|
||||
@commands.slash_command(name="забрать_призовую_роль")
|
||||
async def TakeAwayAPrizeRoleSlash(self, ctx, role: disnake.Role, member: disnake.Member):
|
||||
await ctx.response.defer(ephemeral=True)
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
|
||||
async with self.DataBaseManager.session() as session:
|
||||
async with session.begin():
|
||||
async with self.DataBaseManager.models['received_roles_prize'] as received_roles_prize_model:
|
||||
stmt = self.DataBaseManager.select(received_roles_prize_model).where(
|
||||
received_roles_prize_model.user_id == member.id,
|
||||
received_roles_prize_model.role_id == role.id
|
||||
).with_for_update()
|
||||
user_role = (await session.execute(stmt)).scalars().first()
|
||||
if user_role is None:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Данный пользователь не связан с этой ролью', colour=0x2F3136))
|
||||
return
|
||||
|
||||
await session.delete(user_role)
|
||||
embed = disnake.Embed(description=f'Роль {role.mention} удалена из инвентаря {member.mention}!', colour=0x2F3136)
|
||||
await member.remove_roles(role)
|
||||
await ctx.edit_original_message(embed=embed)
|
||||
|
||||
|
||||
@commands.slash_command(name="удалить_роль")
|
||||
async def DeleteRoleSlash(self, ctx: disnake.AppCmdInter, roleid: str):
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.send(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
try:
|
||||
res = await self.client.DeleteRole(int(roleid))
|
||||
await ctx.send(embed=disnake.Embed(description=f'{res[1]}', colour=0xff9900), ephemeral=True)
|
||||
except ValueError:
|
||||
await ctx.send(embed=disnake.Embed(description=f'Введён неверный id', colour=0xff9900), ephemeral=True)
|
||||
|
||||
|
||||
@commands.slash_command(name="установить_описание_роли")
|
||||
async def RoleDescriptoinSetSlash(self, ctx, role: disnake.Role, description: str):
|
||||
await ctx.response.defer(ephemeral=True)
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
|
||||
async with self.DataBaseManager.session() as session:
|
||||
async with session.begin():
|
||||
async with self.DataBaseManager.models['roles_static'] as roles_static_model:
|
||||
stmt = self.DataBaseManager.select(roles_static_model).where(
|
||||
roles_static_model.id == role.id
|
||||
).with_for_update()
|
||||
role_static = (await session.execute(stmt)).scalars().first()
|
||||
if role_static is None:
|
||||
role = roles_static_model(id = role.id, description = description)
|
||||
session.add(role)
|
||||
else:
|
||||
role_static.description = description
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Для роли {role.mention} успешно установлено описание\n```{description}```', colour=0x2F3136))
|
||||
return
|
||||
|
||||
@commands.slash_command(name = "изменить_параметр")
|
||||
async def ChangeParamSlash(self, ctx: disnake.AppCmdInter, member: disnake.Member, vector: int, parameter: str = commands.Param(description="Какой параметр хотите изменить?",
|
||||
name="параметр",
|
||||
choices=['крошки', 'сообщения', 'секунды в голосовом канале', 'репутация', 'зарплата'])):
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
|
||||
async with self.DataBaseManager.session() as session:
|
||||
await self.client.UserUpdate(member, session = session)
|
||||
async with session.begin():
|
||||
async with self.DataBaseManager.models['users'] as user_model:
|
||||
stmt = self.DataBaseManager.select(user_model).where(
|
||||
user_model.id == member.id
|
||||
).with_for_update()
|
||||
user = (await session.execute(stmt)).scalars().first()
|
||||
count = 0
|
||||
match parameter:
|
||||
case 'крошки':
|
||||
user.crumbs += vector
|
||||
count = user.crumbs
|
||||
case 'сообщения':
|
||||
user.period_messages += vector
|
||||
user.summary_messages += vector
|
||||
count = user.period_messages
|
||||
case 'секунды в голосовом канале':
|
||||
user.period_voice_activity += vector
|
||||
user.summary_voice_activity += vector
|
||||
count = user.period_voice_activity
|
||||
case 'репутация':
|
||||
user.carma += vector
|
||||
count = user.carma
|
||||
case 'зарплата':
|
||||
user.staff_salary += vector
|
||||
count = user.staff_salary
|
||||
await ctx.send(embed=disnake.Embed(description=f'Количество {parameter} пользователя {member.mention} успешно изменено до `{count}`', colour=0xff9900), ephemeral=True)
|
||||
|
||||
@commands.slash_command(name = "дать_зверя")
|
||||
async def GiveRimagochiAnimalSlash(self, ctx: disnake.AppCmdInter, member: disnake.Member, animal_id: int):
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
|
||||
if not animal_id in rimagochi_animals.keys():
|
||||
await ctx.send(embed=disnake.Embed(description=f'Некорректный идентификатор животного', colour=0xff9900), ephemeral=True)
|
||||
return
|
||||
|
||||
async with self.DataBaseManager.session() as session:
|
||||
await self.client.RimagochiUserUpdate(member, session = session)
|
||||
async with session.begin():
|
||||
async with self.DataBaseManager.models['rimagochi_animals'] as rimagochi_animals_model:
|
||||
animal = rimagochi_animals_model(model_animal_id = animal_id, initial_owner_id = ctx.author.id, owner_id = member.id)
|
||||
session.add(animal)
|
||||
await ctx.send(embed=disnake.Embed(description=f'{member.mention} успешно получил `{rimagochi_animals[animal_id]["name"]}`', colour=0xff9900), ephemeral=True)
|
||||
|
||||
@commands.slash_command(name = "удалить_зверя")
|
||||
async def RemoveRimagochiAnimalSlash(self, ctx: disnake.AppCmdInter, inventory_id: int):
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.edit_original_message(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
|
||||
async with self.DataBaseManager.session() as session:
|
||||
async with session.begin():
|
||||
async with self.DataBaseManager.models['rimagochi_animals'] as rimagochi_animals_model:
|
||||
stmt = self.DataBaseManager.select(rimagochi_animals_model).where(
|
||||
rimagochi_animals_model.id == inventory_id
|
||||
).with_for_update()
|
||||
animal = (await session.execute(stmt)).scalars().first()
|
||||
|
||||
if animal is None:
|
||||
await ctx.send(embed=disnake.Embed(description=f'Животного с таким идентификатором не обнаружено', colour=0x2F3136), ephemeral=True)
|
||||
return 1
|
||||
else:
|
||||
await session.delete(animal)
|
||||
await ctx.send(embed=disnake.Embed(description=f"<@{animal.owner_id}> лишился `{rimagochi_animals[animal.model_animal_id]['name']}` с id `{inventory_id}`",
|
||||
colour=0xff9900), ephemeral=True)
|
||||
return 0
|
||||
|
||||
@commands.slash_command(name = "температура")
|
||||
async def RaspberryTemperature(self, ctx: disnake.AppCmdInter):
|
||||
if not self.me in ctx.author.roles:
|
||||
await ctx.send(embed=disnake.Embed(description=f'Эта команда доступна только администратору', colour=0x2F3136))
|
||||
return
|
||||
def redgreen(temper, minimum, maximum):
|
||||
if temper == ((maximum+minimum)/2):
|
||||
return [255, 255, 0]
|
||||
if temper > ((maximum+minimum)/2):
|
||||
return [255, max(0, min(int((1-((temper-minimum)/(maximum-minimum)))*255), 255)), 0]
|
||||
if temper < ((maximum+minimum)/2):
|
||||
return [max(0, min(int(((temper-minimum)/(maximum-minimum))*255), 255)), 255, 0]
|
||||
try:
|
||||
with open('/sys/class/thermal/thermal_zone0/temp', 'r') as f:
|
||||
temp = f.read()
|
||||
except:
|
||||
temp = "0"
|
||||
await ctx.send(embed=disnake.Embed(description=f"Температура CPU {str(int(temp) / 1000)} °C", colour=disnake.Color.from_rgb(*redgreen(int(temp)/1000, 30, 70))))
|
||||
Reference in New Issue
Block a user