added requirements.txt

This commit is contained in:
2022-08-28 19:25:39 +03:00
parent a603898f2d
commit 78a477f21a
8 changed files with 175 additions and 125 deletions

View File

@@ -2,11 +2,12 @@ import logging
from asyncio import sleep from asyncio import sleep
import disnake import disnake
from disnake import Option, OptionType
from disnake.ext import commands, tasks from disnake.ext import commands, tasks
from lib.Comands import read_json from lib.Comands import read_json, write_json
from lib.DB import fill_bd, prepare_db from lib.DB import fill_bd, prepare_db, work_with_db
class Admin(commands.Cog, name='Admin'): class Admin(commands.Cog, name='Admin'):
@@ -39,6 +40,12 @@ class Admin(commands.Cog, name='Admin'):
) )
) )
@commands.Cog.listener()
async def on_member_update(self, before: disnake.Member, after: disnake.Member):
sql_update_query = f"""UPDATE "{after.guild.id}" set nick = ? where userid = ?"""
data_tuple = (after.nick, before.id)
await work_with_db(sql_update_query, data_tuple)
@commands.Cog.listener() @commands.Cog.listener()
async def on_guild_join(self, guild): async def on_guild_join(self, guild):
for g in guild.members: for g in guild.members:
@@ -59,6 +66,93 @@ class Admin(commands.Cog, name='Admin'):
logging.info(f"Adding to {member} role {role}") logging.info(f"Adding to {member} role {role}")
await member.add_roles(role) await member.add_roles(role)
@commands.slash_command(
name="set_guest_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.role, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_guest_role(self, inter, role):
await write_json(inter.guild.id, "guest_role", role.id)
await inter.response.send_message(f"Setted up dsss bot role to: `{role.name}`", ephemeral=True)
@commands.command(name="set_prefix")
@commands.has_permissions(administrator=True)
async def command_set_prefix(self, ctx, prefix: str):
await write_json(ctx.guild.id, "prefix", prefix)
await ctx.reply(f"Prefix set to: `{prefix}`")
@commands.guild_only()
@commands.slash_command(
name="set_prefix",
description="Setting up bot prefix",
options=[
Option("prefix", "Specify prefix", OptionType.string, required=True),
]
)
@commands.has_permissions(administrator=True)
async def slash_set_prefix(self, inter, prefix: str):
await write_json(inter.guild.id, "prefix", prefix)
await inter.response.send_message(f"Prefix set to: `{prefix}`", ephemeral=True)
@commands.guild_only()
@commands.slash_command(
name="set_trigger_role",
description="Setting up role to trigger bot",
options=[
Option("role", "Specify role", OptionType.role, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_trigger_role(self, inter, role):
await write_json(inter.guild.id, "tigger_role", role.id)
await inter.response.send_message(f"Role to trigger set to : `{role.name}`", ephemeral=True)
@commands.slash_command(
name="set_bot_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.role, required=True),
]
)
@commands.guild_only()
@commands.has_permissions(administrator=True)
async def set_bot_role(self, ctx, role):
await write_json(ctx.guild.id, "bot_role", role.id)
await ctx.send(f"Setted up bot role to: `{role.name}`", ephemeral=True)
@set_bot_role.error
@set_trigger_role.error
@slash_set_prefix.error
async def set_prefix_error(self, inter, prefix):
await inter.response.send_message("You don`t have permissions", ephemeral=True)
@commands.has_permissions(administrator=True)
@commands.command(
name="set_time",
description="Read list of tracks for user",
options=[
Option("seconds", "specify max duration", OptionType.integer, required=True),
]
)
async def set_time(self, inter, seconds: commands.Range[5, 30]):
await write_json(inter.guild.id, "seconds", seconds)
await inter.response.send_message(f"Change max audio duration to {seconds} sec", ephemeral=True)
@commands.has_permissions(administrator=True)
@commands.command(
name="set_bot_channel",
description="Set channel whitch itterate with bot",
options=[
Option("channel", "specify channel", OptionType.channel, required=True),
]
)
async def set_time(self, inter, _channel: commands.Range[5, 30]):
await write_json(inter.guild.id, "channel", _channel)
await inter.response.send_message(f"Channet setted up to{_channel} sec", ephemeral=True)
def setup(bot): # an extension must have a setup function def setup(bot): # an extension must have a setup function
bot.add_cog(Admin(bot)) # adding a cog bot.add_cog(Admin(bot)) # adding a cog

View File

@@ -1,14 +1,15 @@
import logging import logging
import random import random
import tempfile
from os import path, makedirs, rename, remove from os import path, makedirs, rename, remove
from disnake.ext import commands from disnake.ext import commands
from lib.Comands import determine_time
from lib.DB import read_db, check_exist_audio, add_audio from lib.DB import read_db, check_exist_audio, add_audio
from lib.Player import play_audio from lib.Player import play_audio
# todo: write chose audio from list by slash command
class Audio(commands.Cog): class Audio(commands.Cog):
def __init__(self, bot): def __init__(self, bot):
self.bot = bot self.bot = bot
@@ -17,46 +18,44 @@ class Audio(commands.Cog):
async def on_ready(self): async def on_ready(self):
logging.info(f'Cog {__name__.split(".")[1]} is ready!.') logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
# todo: complete check activity
@commands.Cog.listener() @commands.Cog.listener()
async def on_voice_state_update(self, member, before, after): async def on_voice_state_update(self, member, before, after):
if any('Escape from Tarkov' in str(user.activity) for user in after.channel.members): if before.channel is None and not member.bot:
logging.info('Skip playing by Game') if any('Escape from Tarkov' in str(user.activity) for user in after.channel.members):
else: logging.info('Skip playing by Game')
from lib.Comands import read_json else:
_role = await read_json(member.guild.id, 'tigger_role') from lib.Comands import read_json
# Read audio from DB _role = await read_json(member.guild.id, 'tigger_role')
audio_db = await read_db(member.guild.id, member.id, 'usertracks') # Read audio from DB
def_audio_db = await read_db(member.guild.id, member.id, 'defaulttracks') audio_db = await read_db(member.guild.id, member.id, 'usertracks')
if audio_db is not None: def_audio_db = await read_db(member.guild.id, member.id, 'defaulttracks')
audio_db = audio_db.split(', ') # Need to fix creating list if audio_db is not None:
for i in range(len(audio_db)): audio_db = audio_db.split(', ') # Need to fix creating list
audio_db[i] = f'{member.id}/{audio_db[i]}' for i in range(len(audio_db)):
if def_audio_db is not None: audio_db[i] = f'{member.id}/{audio_db[i]}'
def_audio_db = def_audio_db.split(', ') if def_audio_db is not None:
from lib.Comands import list_files def_audio_db = def_audio_db.split(', ')
def_audio_ls = await list_files() from lib.Comands import list_files
if before.channel is None and not member.bot: def_audio_ls = await list_files()
if def_audio_db or audio_db is not None: if def_audio_db or audio_db is not None:
if def_audio_db is None: def_audio_db = [] if def_audio_db is None: def_audio_db = []
if audio_db is None: audio_db = [] if audio_db is None: audio_db = []
logging.info(f'Play audio from DB') logging.info(f'Play audio from DB')
full_audio = def_audio_db + audio_db full_audio = def_audio_db + audio_db
audio = random.choice(full_audio) await play_audio(random.choice(full_audio), self.bot, after.channel)
await play_audio(audio, self.bot, after.channel)
elif len(member.roles) == 1 or _role is None: elif len(member.roles) == 1 or _role is None:
logging.info(f'Skip playing') logging.info(f'Skip playing')
elif any(str(role.id) in _role for role in member.roles): elif any(str(role.id) in _role for role in member.roles):
logging.info(f'Play audio from list by role') logging.info(f'Play audio from list by role')
audio = random.choice(def_audio_ls) await play_audio(random.choice(def_audio_ls), self.bot, after.channel)
await play_audio(audio, self.bot, after.channel)
else: else:
logging.info(f'Skip playing') logging.info(f'Skip playing')
@commands.command(name="upload_audio") @commands.command(name="upload_audio")
async def upload_audio(self, ctx, user=None): async def upload_audio(self, ctx, user=None):
user = user or ctx.author user = user or ctx.author
print(tempfile.tempdir)
if ctx.author.guild_permissions.administrator or user is ctx.author: if ctx.author.guild_permissions.administrator or user is ctx.author:
if ctx.message.attachments: if ctx.message.attachments:
from os import error from os import error
@@ -68,15 +67,16 @@ class Audio(commands.Cog):
for at in ctx.message.attachments: for at in ctx.message.attachments:
import mimetypes import mimetypes
await at.save(f'{tempfile.tempdir}/{tempfile.mkdtemp(dir=f"tmp/{user.id}")}/{at.filename}') await at.save(f'tmp/{user.id}")/{at.filename}')
guess = mimetypes.guess_type(f'{tempfile.tempdir}/{user.id}/{at.filename}') guess = mimetypes.guess_type(f'tmp/{user.id}")/{at.filename}')
if guess[0].split('/')[0] == 'audio': if guess[0].split('/')[0] == 'audio':
from pymediainfo import MediaInfo from pymediainfo import MediaInfo
file = f'{tempfile.tempdir}/{user.id}/{at.filename}' file = f'tmp/{user.id}")/{at.filename}'
duration = round(MediaInfo.parse(file).tracks[0].duration / 1000) duration = round(MediaInfo.parse(file).tracks[0].duration / 1000)
if duration > 20: max_duration = determine_time(ctx)
await ctx.reply(f'Audio duration is {duration}, but max is 15') if duration > max_duration:
remove(f'{tempfile.tempdir}/{user.id}/{at.filename}') await ctx.reply(f'Audio duration is {duration}, but max is {max_duration}')
remove(f'tmp/{user.id}")/{at.filename}')
else: else:
a = await read_db(ctx.guild.id, user.id, 'usertracks') a = await read_db(ctx.guild.id, user.id, 'usertracks')
if a: if a:
@@ -86,10 +86,10 @@ class Audio(commands.Cog):
await check_exist_audio(ctx, ctx.guild.id, user.id, 'usertracks', at.filename) await check_exist_audio(ctx, ctx.guild.id, user.id, 'usertracks', at.filename)
await add_audio(ctx.guild.id, user.id, audiolist) await add_audio(ctx.guild.id, user.id, audiolist)
rename(f'{tempfile.tempdir}/{user.id}/{at.filename}', f'audio/{user.id}/{at.filename}') rename(f'tmp/{user.id}")/{at.filename}', f'audio/{user.id}/{at.filename}')
elif guess[0].split('/')[0] != 'audio': elif guess[0].split('/')[0] != 'audio':
await ctx.reply(f'It not audio {at.filename}\n it`s {guess[0]}') await ctx.reply(f'It not audio {at.filename}\n it`s {guess[0]}')
remove(f'{tempfile.tempdir}/{user.id}/{at.filename}') remove(f'tmp/{user.id}")/{at.filename}')
else: else:
await ctx.reply("Has no Attachment") await ctx.reply("Has no Attachment")
else: else:

View File

@@ -16,12 +16,6 @@ class General(commands.Cog):
async def on_ready(self): async def on_ready(self):
logging.info(f'Cog {__name__.split(".")[1]} is ready!.') logging.info(f'Cog {__name__.split(".")[1]} is ready!.')
@commands.Cog.listener()
async def on_member_update(self, before: disnake.Member, after: disnake.Member):
sql_update_query = f"""UPDATE "{after.guild.id}" set nick = ? where userid = ?"""
data_tuple = (after.nick, before.id)
await work_with_db(sql_update_query, data_tuple)
@commands.slash_command( @commands.slash_command(
name="info", name="info",
description="Read list of tracks for user", description="Read list of tracks for user",
@@ -66,69 +60,6 @@ class General(commands.Cog):
await inter.response.send_message(embed=emb, ephemeral=True) await inter.response.send_message(embed=emb, ephemeral=True)
@commands.slash_command(
name="set_guest_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.role, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_guest_role(self, inter, role):
await write_json(inter.guild.id, "guest_role", role.id)
await inter.response.send_message(f"Setted up dsss bot role to: `{role.name}`", ephemeral=True)
@commands.command(name="set_prefix")
@commands.has_permissions(administrator=True)
async def command_set_prefix(self, ctx, prefix: str):
await write_json(ctx.guild.id, "prefix", prefix)
await ctx.reply(f"Prefix set to: `{prefix}`")
@commands.guild_only()
@commands.slash_command(
name="set_prefix",
description="Setting up bot prefix",
options=[
Option("prefix", "Specify prefix", OptionType.string, required=True),
]
)
@commands.has_permissions(administrator=True)
async def slash_set_prefix(self, inter, prefix: str):
await write_json(inter.guild.id, "prefix", prefix)
await inter.response.send_message(f"Prefix set to: `{prefix}`", ephemeral=True)
@commands.guild_only()
@commands.slash_command(
name="set_trigger_role",
description="Setting up role to trigger bot",
options=[
Option("role", "Specify role", OptionType.role, required=True),
]
)
@commands.has_permissions(administrator=True)
async def set_trigger_role(self, inter, role):
await write_json(inter.guild.id, "tigger_role", role.id)
await inter.response.send_message(f"Role to trigger set to : `{role.name}`", ephemeral=True)
@commands.slash_command(
name="set_bot_role",
description="Set Default bot role",
options=[
Option("role", "Specify role", OptionType.role, required=True),
]
)
@commands.guild_only()
@commands.has_permissions(administrator=True)
async def set_bot_role(self, ctx, role):
await write_json(ctx.guild.id, "bot_role", role.id)
await ctx.send(f"Setted up bot role to: `{role.name}`", ephemeral=True)
@set_bot_role.error
@set_trigger_role.error
@slash_set_prefix.error
async def set_prefix_error(self, inter, prefix):
await inter.response.send_message("You don`t have permissions", ephemeral=True)
def setup(bot): # an extension must have a setup function def setup(bot): # an extension must have a setup function
bot.add_cog(General(bot)) # adding a cog bot.add_cog(General(bot)) # adding a cog

View File

@@ -5,7 +5,7 @@ import disnake
import psutil import psutil
from disnake.ext import commands from disnake.ext import commands
from lib.Comands import determine_prefix from lib.Comands import determine_prefix, determine_time
class Bot_info(commands.Cog, name='Bot Info'): class Bot_info(commands.Cog, name='Bot Info'):
@@ -24,11 +24,13 @@ class Bot_info(commands.Cog, name='Bot Info'):
title=f"General information", title=f"General information",
description=f"General information on about bot", description=f"General information on about bot",
) )
# emb.set_thumbnail(self.bot.display_avatar) # emb.set_thumbnail(self.bot.user.avarat_url)
emb.add_field(name="System info:", value=f"Memory Usage: {round(_process.memory_info().rss / 2 ** 20, 2)} Mb\n" emb.add_field(name="System info:", value=f"Memory Usage: {round(_process.memory_info().rss / 2 ** 20, 2)} Mb\n"
f"CPU Usage: {_process.cpu_percent()}%\n" f"CPU Usage: {_process.cpu_percent()}%\n"
f'Bot ping: {round(self.bot.latency * 1000)}\n' f'Bot ping: {round(self.bot.latency * 1000)}\n'
f'Prefix: `{determine_prefix(self.bot, inter)}`') f'Prefix: `{determine_prefix(self.bot, inter)}\n`'
f"Max audio duration: {determine_time(inter)} sec\n"
)
emb.add_field(name="Bot owner", value=f"<@{self.bot.owner_id}>") emb.add_field(name="Bot owner", value=f"<@{self.bot.owner_id}>")
emb.set_footer(text="Information requested by: {}".format(inter.author.display_name)) emb.set_footer(text="Information requested by: {}".format(inter.author.display_name))

View File

@@ -1,5 +1,8 @@
import logging
from os import listdir from os import listdir
from disnake.ext import commands
""" """
Loads, unloads Cogs files Loads, unloads Cogs files
cog_list: return list of cog filenames cog_list: return list of cog filenames
@@ -18,7 +21,19 @@ def cog_list():
async def work_with_cogs(what_do, bot): async def work_with_cogs(what_do, bot):
for _filename in cog_list(): for _filename in cog_list():
if what_do == "load": if what_do == "load":
bot.load_extension(f'cogs.{_filename}') try:
bot.load_extension(f'cogs.{_filename}')
logging.info(f'Loaded cog {_filename}')
except commands.ExtensionNotFound:
logging.error(f"Error: {_filename} couldn't be found to load.")
except commands.ExtensionFailed as error:
logging.error(f'Error: {_filename} failed to load properly.', error)
except commands.ExtensionError:
logging.error(f'Error: unknown error with {_filename}')
elif what_do == 'unload': elif what_do == 'unload':
bot.unload_extension(f'cogs.{_filename}') bot.unload_extension(f'cogs.{_filename}')
elif what_do == 'reload': elif what_do == 'reload':

View File

@@ -68,8 +68,7 @@ def determine_prefix(bot, msg):
parameter = '$' parameter = '$'
with open('prefix.json', 'r', encoding='utf-8') as fp: # Open the JSON with open('prefix.json', 'r', encoding='utf-8') as fp: # Open the JSON
try: try:
from json import load _json = json.load(fp) # Load the custom prefixes
_json = load(fp) # Load the custom prefixes
except: except:
_json = {} _json = {}
try: try:
@@ -78,3 +77,23 @@ def determine_prefix(bot, msg):
except: except:
pass pass
return parameter return parameter
def determine_time(msg):
"""
Determite per-server bot prefix
:param msg: Disnake msg object
:return: prefix for server, default is $
"""
parameter = 15
with open('prefix.json', 'r', encoding='utf-8') as fp: # Open the JSON
try:
_json = json.load(fp) # Load the custom prefixes
except:
_json = {}
try:
parameter = _json[f"{msg.guild.id}"]["seconds"] # Read prefix from json if is setted up
except:
pass
return parameter

View File

@@ -1,5 +1,5 @@
disnake[audio]~=2.5.2 disnake[audio]~=2.5.2
setuptools==65.2.0 setuptools==65.3.0
psutil~=5.9.1 psutil~=5.9.1
pymediainfo~=5.1.0 pymediainfo~=5.1.0
pyNaCl==1.5.0 pyNaCl~=1.5.0

17
test.py
View File

@@ -1,3 +1,4 @@
import asyncio
import logging import logging
import sys import sys
from os import path from os import path
@@ -6,7 +7,7 @@ import disnake
from disnake import OptionChoice, OptionType, Option from disnake import OptionChoice, OptionType, Option
from disnake.ext import commands from disnake.ext import commands
from lib.CogsPrepare import cog_list, work_with_cogs from lib.CogsPrepare import work_with_cogs
from lib.Comands import determine_prefix from lib.Comands import determine_prefix
if not path.isfile('prefix.json'): if not path.isfile('prefix.json'):
@@ -29,19 +30,7 @@ bot = commands.Bot(command_prefix=determine_prefix,
logging.basicConfig(stream=sys.stdout, filemode='w', level='INFO', logging.basicConfig(stream=sys.stdout, filemode='w', level='INFO',
format='%(asctime)s - %(levelname)s - %(module)s - %(message)s') format='%(asctime)s - %(levelname)s - %(module)s - %(message)s')
for filename in cog_list(): asyncio.run(work_with_cogs('load', bot))
try:
bot.load_extension(f'cogs.{filename}')
logging.info(f'Loaded cog {filename}')
except commands.ExtensionNotFound:
logging.error(f"Error: {filename} couldn't be found to load.")
except commands.ExtensionFailed as error:
logging.error(f'Error: {filename} failed to load properly.', error)
except commands.ExtensionError:
logging.error(f'Error: unknown error with {filename}')
@bot.event @bot.event