Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
0946a7f51e | ||
|
|
3e0a0c197a | ||
|
|
b43c5cf5c1 | ||
|
|
a136491362 | ||
|
|
413736b44d | ||
|
|
78be10a88b | ||
|
|
c3140403b5 | ||
|
|
c60502a241 | ||
|
|
17f1972104 |
2
.gitignore
vendored
2
.gitignore
vendored
@@ -10,3 +10,5 @@
|
||||
/fun_and_admin_bot.egg-info/
|
||||
/.YMcache/
|
||||
config.json
|
||||
/env/
|
||||
/.project
|
||||
|
||||
@@ -10,58 +10,10 @@ stages:
|
||||
- 'code_quality'
|
||||
- 'deploy'
|
||||
|
||||
|
||||
sast:
|
||||
stage: 'test'
|
||||
include:
|
||||
- template: Security/SAST.gitlab-ci.yml
|
||||
- template: Security/Dependency-Scanning.gitlab-ci.yml
|
||||
- template: Security/SAST-IaC.gitlab-ci.yml
|
||||
|
||||
qodana:
|
||||
stage: 'code_quality'
|
||||
needs: ['gemnasium-python-dependency_scanning', 'semgrep-sast', 'kics-iac-sast']
|
||||
only:
|
||||
- master
|
||||
- merge_requests
|
||||
image:
|
||||
name: jetbrains/qodana-python-community
|
||||
entrypoint:
|
||||
- ''
|
||||
cache:
|
||||
- key: qodana-2023.3-$CI_DEFAULT_BRANCH-$CI_COMMIT_REF_SLUG
|
||||
fallback_keys:
|
||||
- qodana-2023.3-$CI_DEFAULT_BRANCH-
|
||||
- qodana-2023.3-
|
||||
paths:
|
||||
- ".qodana/cache"
|
||||
variables:
|
||||
QODANA_TOKEN: "$qodana_token"
|
||||
script:
|
||||
- qodana --save-report --results-dir=$CI_PROJECT_DIR/.qodana/results --cache-dir=$CI_PROJECT_DIR/.qodana/cache
|
||||
artifacts:
|
||||
expose_as: qodana_report
|
||||
expire_in: 1 week
|
||||
paths:
|
||||
- ".qodana/results/"
|
||||
|
||||
|
||||
push:
|
||||
stage: 'deploy'
|
||||
image: 'alpine'
|
||||
needs: ['qodana']
|
||||
only:
|
||||
- master
|
||||
before_script:
|
||||
- apk add openssh-client > /dev/null
|
||||
- eval $(ssh-agent -s)
|
||||
- chmod 400 $SSH_PRIVATE_KEY
|
||||
|
||||
- mkdir -p ~/.ssh
|
||||
- chmod 700 ~/.ssh
|
||||
|
||||
- cp $SSH_PRIVATE_KEY ~/.ssh/id_rsa
|
||||
|
||||
- ssh-add ~/.ssh/id_rsa
|
||||
script:
|
||||
- ssh $BOT_HOST -oStrictHostKeyChecking=accept-new 'cd /opt/discord/tarkov && git pull'
|
||||
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
__version__ = '0.0.7'
|
||||
__version__ = '0.0.8'
|
||||
__title__ = "Pisya_bot"
|
||||
__author__ = "baconborn"
|
||||
|
||||
@@ -13,4 +13,4 @@ class VersionInfo(NamedTuple):
|
||||
serial: int
|
||||
|
||||
|
||||
version_info: VersionInfo = VersionInfo(major=0, minor=0, micro=7, releaselevel="alpha", serial=0)
|
||||
version_info: VersionInfo = VersionInfo(major=0, minor=0, micro=8, releaselevel="alpha", serial=0)
|
||||
|
||||
2
bot.py
2
bot.py
@@ -6,11 +6,11 @@ from os.path import isfile
|
||||
from disnake import OptionType, Option, Localized, ApplicationCommandInteraction, Intents, __version__
|
||||
from disnake.ext.commands import Bot, is_owner
|
||||
from dotenv import load_dotenv
|
||||
from loguru import logger
|
||||
|
||||
from __init__ import __version__ as ver
|
||||
from lib.CogsPrep import work_with_cogs, cog_list
|
||||
from lib.Comands import determine_prefix
|
||||
from lib.Logger import logger
|
||||
|
||||
load_dotenv()
|
||||
if not isfile('.env') or not getenv('CONF_FILE'):
|
||||
|
||||
@@ -3,10 +3,10 @@ from asyncio import sleep
|
||||
import disnake
|
||||
from disnake import Option, OptionType, Localized
|
||||
from disnake.ext import commands, tasks
|
||||
from loguru import logger
|
||||
|
||||
from lib.Comands import read_json, write_json
|
||||
from lib.DB_Worker import fill_bd, prepare_db, work_with_db
|
||||
from lib.Logger import logger
|
||||
|
||||
|
||||
class Admin(commands.Cog, name='Admin'):
|
||||
@@ -57,6 +57,8 @@ class Admin(commands.Cog, name='Admin'):
|
||||
bot_role = read_json(member.guild.id, 'bot_role') # Get bot role
|
||||
guest_role = read_json(member.guild.id, 'guest_role') # Get guest role
|
||||
|
||||
logger.info(f"Bot role: {bot_role} | Guest role: {guest_role}")
|
||||
logger.info(f'type bot_role: {type(bot_role)} | type guest_role: {type(guest_role)}')
|
||||
if bot_role or guest_role:
|
||||
if member.bot == 0:
|
||||
role = disnake.utils.get(member.guild.roles, id=guest_role)
|
||||
@@ -77,6 +79,7 @@ class Admin(commands.Cog, name='Admin'):
|
||||
)
|
||||
@commands.has_permissions(administrator=True)
|
||||
async def set_guest_role(self, inter: disnake.ApplicationCommandInteraction, role):
|
||||
print(type(role.id))
|
||||
await write_json(inter.guild.id, "guest_role", role.id)
|
||||
await inter.response.send_message(f"Set up bot role to: `{role.name}`", ephemeral=True)
|
||||
|
||||
|
||||
@@ -3,10 +3,11 @@ import random
|
||||
import disnake
|
||||
from disnake import OptionChoice, Option, OptionType, Member, VoiceState
|
||||
from disnake.ext import commands
|
||||
from disnake.utils import get
|
||||
from loguru import logger
|
||||
|
||||
from lib.Logger import logger
|
||||
from lib.Player import play_audio
|
||||
from lib.ListGenerator import ListGenerator
|
||||
from lib.Player import play_audio
|
||||
|
||||
|
||||
class Audio(commands.Cog, name='Audio'):
|
||||
@@ -28,14 +29,14 @@ class Audio(commands.Cog, name='Audio'):
|
||||
else:
|
||||
# Prepare list of audio
|
||||
from lib.Comands import read_json
|
||||
_role = await read_json(member.guild.id, 'tigger_role')
|
||||
_role = get(member.guild.roles, id=read_json(member.guild.id, 'tigger_role'))
|
||||
audio: list = []
|
||||
for _a in ListGenerator('audio'):
|
||||
audio.append(_a.name)
|
||||
|
||||
if len(member.roles) == 1 or _role is None:
|
||||
logger.info('Skip playing by role')
|
||||
elif any(str(role.id) in _role for role in member.roles):
|
||||
elif _role in member.roles:
|
||||
logger.info('Play audio from list by role')
|
||||
await play_audio(f'audio/{random.choice(audio)}', self.bot, after.channel)
|
||||
else:
|
||||
@@ -54,10 +55,31 @@ class Audio(commands.Cog, name='Audio'):
|
||||
):
|
||||
if inter.author.voice is not None:
|
||||
await inter.response.send_message(f'Played {audio}', ephemeral=True)
|
||||
|
||||
await play_audio(audio, self.bot, inter.author.voice.channel)
|
||||
else:
|
||||
await inter.response.send_message('You`re not in voice', ephemeral=True)
|
||||
|
||||
@commands.slash_command(name="play_random",
|
||||
description="Make possible playing audio by command",
|
||||
options=[
|
||||
Option(name="num",
|
||||
type=OptionType.integer,
|
||||
)
|
||||
])
|
||||
async def playrandom(self, inter: disnake.ApplicationCommandInteraction,
|
||||
num: int
|
||||
):
|
||||
if inter.author.voice is not None:
|
||||
audio: list = []
|
||||
for _a in ListGenerator('audio'):
|
||||
audio.append(_a.name)
|
||||
for i in range(num):
|
||||
logger.info(f'Played {i+1} times')
|
||||
await play_audio(f'audio/{random.choice(audio)}', self.bot, inter.author.voice.channel)
|
||||
else:
|
||||
await inter.response.send_message('You`re not in voice', ephemeral=True)
|
||||
|
||||
@playaudio.autocomplete('audio')
|
||||
async def list_to_play(self, inter: disnake.ApplicationCommandInteraction, current: str):
|
||||
current = current.lower()
|
||||
|
||||
@@ -3,9 +3,9 @@ from typing import List
|
||||
import disnake
|
||||
from disnake import OptionChoice
|
||||
from disnake.ext import commands
|
||||
from loguru import logger
|
||||
|
||||
from lib.Comands import write_json
|
||||
from lib.Logger import logger
|
||||
|
||||
|
||||
class Fun(commands.Cog, name='Fun'):
|
||||
|
||||
@@ -1,9 +1,9 @@
|
||||
import disnake
|
||||
from disnake import Option
|
||||
from disnake.ext import commands
|
||||
from loguru import logger
|
||||
|
||||
from lib import YandexPlayer
|
||||
from lib.Logger import logger
|
||||
|
||||
|
||||
class Testing(commands.Cog, name='Testing'):
|
||||
|
||||
@@ -2,7 +2,8 @@ import disnake
|
||||
from disnake import Option, OptionType, Colour
|
||||
from disnake.ext import commands
|
||||
|
||||
from lib.Logger import logger
|
||||
# from lib.Logger import logger
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class General(commands.Cog):
|
||||
|
||||
@@ -1,13 +1,16 @@
|
||||
import os
|
||||
|
||||
import disnake
|
||||
import psutil
|
||||
|
||||
from disnake.ext import commands
|
||||
from loguru import logger
|
||||
|
||||
from __init__ import version_info as ver
|
||||
from lib.Comands import determine_prefix
|
||||
from lib.Logger import logger
|
||||
|
||||
|
||||
# from lib.Logger import logger
|
||||
|
||||
class BotInfo(commands.Cog, name='Bot Info'):
|
||||
def __init__(self, bot: commands.Bot):
|
||||
self.bot = bot # defining bot as global var in class
|
||||
|
||||
@@ -10,8 +10,8 @@ from os import listdir, rename
|
||||
from typing import List
|
||||
|
||||
from disnake.ext import commands
|
||||
|
||||
from .Logger import logger
|
||||
# from .Logger import logger
|
||||
from loguru import logger
|
||||
|
||||
|
||||
async def cog_list(fold: str = './cogs') -> List[str]:
|
||||
|
||||
@@ -9,7 +9,7 @@ from os import getenv
|
||||
from disnake.ext import commands
|
||||
|
||||
|
||||
async def read_json(guild: int, _param: str):
|
||||
def read_json(guild: int, _param: str):
|
||||
"""
|
||||
Reads Json file to determite config strings
|
||||
:param guild: ID of Guild
|
||||
@@ -35,6 +35,7 @@ async def read_json(guild: int, _param: str):
|
||||
|
||||
|
||||
async def write_json(guild: int, param_name: str, param: str or int):
|
||||
print(type(param))
|
||||
with open(getenv('CONF_FILE'), encoding='utf-8') as f:
|
||||
try:
|
||||
_json = load(f)
|
||||
@@ -45,7 +46,8 @@ async def write_json(guild: int, param_name: str, param: str or int):
|
||||
except KeyError:
|
||||
_json.update({f'{guild}': {}})
|
||||
_guild = _json[f'{guild}']
|
||||
_guild.update({f'{param_name}': f'{param}'})
|
||||
_guild.update({f'{param_name}': param})
|
||||
_json.update({f'{guild}': _guild})
|
||||
with open(getenv('CONF_FILE'), 'w', encoding='utf-8') as f:
|
||||
dump(_json, f, indent=4)
|
||||
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
from sqlite3 import connect, Error
|
||||
|
||||
from lib.Logger import logger
|
||||
# from lib.Logger import logger
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class DBReader:
|
||||
|
||||
@@ -1,52 +0,0 @@
|
||||
import logging
|
||||
|
||||
|
||||
class _CustomFormatter(logging.Formatter):
|
||||
grey = "\x1b[38;20m"
|
||||
yellow = "\x1b[33;20m"
|
||||
red = "\x1b[31;20m"
|
||||
bold_red = "\x1b[31;1m"
|
||||
reset = "\x1b[0m"
|
||||
fmt = "%(asctime)s - [%(levelname)s] -%(module)s- %(message)s"
|
||||
|
||||
FORMATS = {
|
||||
logging.DEBUG: grey + fmt + reset,
|
||||
logging.INFO: grey + fmt + reset,
|
||||
logging.WARNING: yellow + fmt + reset,
|
||||
logging.ERROR: red + fmt + reset,
|
||||
logging.CRITICAL: bold_red + fmt + reset
|
||||
}
|
||||
|
||||
old_factory = logging.getLogRecordFactory()
|
||||
|
||||
def _record_factory(*args, **kwargs):
|
||||
record = _CustomFormatter.old_factory(*args, **kwargs)
|
||||
_module = record.module
|
||||
_levelname = record.levelname
|
||||
if len(_module) % 2 == 0 and len(_module) < 12:
|
||||
_module = ' ' * ((12 - len(_module)) // 2) + _module + ' ' * ((12 - len(_module)) // 2)
|
||||
elif len(_module) % 2 == 1 and len(_module) < 12:
|
||||
_module = ' ' * ((12 - len(_module)) // 2) + record.module + ' ' * ((12 - len(_module)) // 2 + 1)
|
||||
if len(_levelname) % 2 == 0 and len(_levelname) < 8:
|
||||
_levelname = ' ' * ((8 - len(_levelname)) // 2) + _levelname + ' ' * ((8 - len(_levelname)) // 2)
|
||||
elif len(record.levelname) % 2 == 1 and len(record.module) < 8:
|
||||
_levelname = ' ' * ((8 - len(_levelname)) // 2) + _levelname + ' ' * ((8 - len(_levelname)) // 2 + 1)
|
||||
record.module = _module
|
||||
record.levelname = _levelname
|
||||
return record
|
||||
|
||||
logging.setLogRecordFactory(_record_factory)
|
||||
|
||||
def format(self, record):
|
||||
log_fmt = self.FORMATS.get(record.levelno)
|
||||
formatter = logging.Formatter(log_fmt, "%d-%m-%Y %H:%M:%S")
|
||||
return formatter.format(record)
|
||||
|
||||
|
||||
logger = logging.getLogger('Pisya_Bot')
|
||||
logger.setLevel(logging.DEBUG)
|
||||
|
||||
ch = logging.StreamHandler()
|
||||
ch.setLevel(logging.DEBUG)
|
||||
ch.setFormatter(_CustomFormatter())
|
||||
logger.addHandler(ch)
|
||||
@@ -1,16 +1,19 @@
|
||||
from asyncio import sleep
|
||||
|
||||
from disnake import FFmpegOpusAudio
|
||||
|
||||
from .Logger import logger
|
||||
from loguru import logger
|
||||
|
||||
|
||||
# from .Logger import logger
|
||||
|
||||
|
||||
@logger.catch
|
||||
async def play_audio(audio, bot, vc):
|
||||
if not bot.voice_clients:
|
||||
logger.error(f'Playing: {audio}')
|
||||
vp = await vc.connect()
|
||||
if not vp.is_playing():
|
||||
vp.play(FFmpegOpusAudio(f'{audio}', ))
|
||||
vp.play(FFmpegOpusAudio(f'{audio}', executable='ffmpeg', options='-nostats -loglevel 0'))
|
||||
while vp.is_playing():
|
||||
await sleep(0.5)
|
||||
await sleep(1)
|
||||
|
||||
@@ -2,3 +2,4 @@ disnake[audio]
|
||||
psutil
|
||||
python-dotenv
|
||||
yandex-music
|
||||
loguru==0.7.2
|
||||
Reference in New Issue
Block a user