mirror of
https://github.com/SantaSpeen/RoyaltyIT-tg-bot.git
synced 2026-10-27 06:20:13 +00:00
Basic bot
This commit is contained in:
1 parent
3fe5009086
commit
aeceb53346
3 files changed
+119
No files matched your search
@@ -0,0 +1,38 @@
|
|||||||
|
import logging
|
||||||
|
import json
|
||||||
|
import os
|
||||||
|
|
||||||
|
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(name)-29s - %(levelname)-5s - %(message)s")
|
||||||
|
|
||||||
|
|
||||||
|
class Config:
|
||||||
|
|
||||||
|
# noinspection PyTypeChecker
|
||||||
|
def __init__(self, config_file):
|
||||||
|
self.log = logging.getLogger(__name__)
|
||||||
|
self.debug = self.log.debug
|
||||||
|
|
||||||
|
self.config_file = config_file
|
||||||
|
self.raw_config: dict = None
|
||||||
|
|
||||||
|
self.bot_token: str = None
|
||||||
|
self.admin_list: list = None
|
||||||
|
self.remote_chat: int = None
|
||||||
|
|
||||||
|
self.messages_object: str = None
|
||||||
|
|
||||||
|
self._read_config()
|
||||||
|
|
||||||
|
def _read_config(self):
|
||||||
|
self.debug("_read_config(self)")
|
||||||
|
if os.path.isfile(self.config_file):
|
||||||
|
self.log.info(f"Config file: %s - found" % self.config_file)
|
||||||
|
with open(self.config_file, 'r') as f:
|
||||||
|
self.raw_config = json.load(f)
|
||||||
|
else:
|
||||||
|
raise FileNotFoundError("Cannot found config file at %s." % self.config_file)
|
||||||
|
|
||||||
|
self.bot_token = self.raw_config.get("bot_token")
|
||||||
|
self.admin_list = self.raw_config.get("admin_list")
|
||||||
|
self.remote_chat = self.raw_config.get("remote_chat")
|
||||||
|
self.messages_object = self.raw_config.get("messages")
|
||||||
+44
@@ -0,0 +1,44 @@
|
|||||||
|
import logging
|
||||||
|
|
||||||
|
from aiogram import Bot, types
|
||||||
|
from aiogram.dispatcher import Dispatcher
|
||||||
|
from aiogram.utils import executor
|
||||||
|
|
||||||
|
from config import Config
|
||||||
|
from tools import Tools
|
||||||
|
|
||||||
|
config = Config("config.json")
|
||||||
|
log = logging.getLogger("bot core")
|
||||||
|
|
||||||
|
bot = Bot(token=config.bot_token)
|
||||||
|
dp = Dispatcher(bot)
|
||||||
|
tools = Tools(config, dp)
|
||||||
|
|
||||||
|
|
||||||
|
@dp.message_handler(commands=['admins'])
|
||||||
|
async def bot_admins(msg: types.Message):
|
||||||
|
message = "%(owner)-s"
|
||||||
|
admin_list = await bot.get_chat_administrators(config.remote_chat)
|
||||||
|
i = 1
|
||||||
|
for admin_object in admin_list:
|
||||||
|
user = admin_object['user']
|
||||||
|
status = admin_object['status'].replace("administrator", "Администратор").replace("creator", "Создатель")
|
||||||
|
if status == "Администратор":
|
||||||
|
if not user['is_bot']:
|
||||||
|
message += f"{i}. {status:13} @{user['username']}\n"
|
||||||
|
i += 1
|
||||||
|
else:
|
||||||
|
message %= {"owner": f"0. {status:13} @{user['username']}\n"}
|
||||||
|
|
||||||
|
await msg.reply(message, disable_notification=True)
|
||||||
|
|
||||||
|
|
||||||
|
# @dp.message_handler(content_types=['text'])
|
||||||
|
# async def msg_logger(msg: types.Message):
|
||||||
|
# log.info(f"New message from {msg.from_user.id}({msg.from_user.username}) in {msg.chat.id}: '{msg.text}'")
|
||||||
|
|
||||||
|
|
||||||
|
tools.bind_static_messages()
|
||||||
|
|
||||||
|
if __name__ == '__main__':
|
||||||
|
executor.start_polling(dp)
|
||||||
@@ -0,0 +1,37 @@
|
|||||||
|
import logging
|
||||||
|
|
||||||
|
import aiogram
|
||||||
|
|
||||||
|
from config import Config
|
||||||
|
|
||||||
|
static_log = logging.getLogger('static messages')
|
||||||
|
|
||||||
|
|
||||||
|
class Tools:
|
||||||
|
|
||||||
|
def __init__(self, config: Config, dispatcher: aiogram.Dispatcher):
|
||||||
|
self.log = logging.getLogger("bot tools")
|
||||||
|
self.config = config
|
||||||
|
self.dispatcher = dispatcher
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
async def __message_handler(msg: aiogram.types.Message, **kwargs):
|
||||||
|
text = msg.text
|
||||||
|
static_log.info(f"New message from {msg.from_user.id}({msg.from_user.username}) in {msg.chat.id}: '{text}'")
|
||||||
|
for k, v in kwargs['__messages__'].items():
|
||||||
|
if k.startswith(text[1:len(k)]):
|
||||||
|
allow = v['allow']
|
||||||
|
send = False
|
||||||
|
if (allow == "all") or \
|
||||||
|
(allow == "private" and msg.chat.id > 0) or \
|
||||||
|
(allow == "chat" and msg.chat.id < 0):
|
||||||
|
send = True
|
||||||
|
if send:
|
||||||
|
await msg.reply(v['msg'])
|
||||||
|
break
|
||||||
|
|
||||||
|
def bind_static_messages(self):
|
||||||
|
__messages__: dict = self.config.raw_config['static_message']
|
||||||
|
__messages_keys__ = list(__messages__.keys())
|
||||||
|
l = lambda *x: self.__message_handler(*x, __messages__=__messages__)
|
||||||
|
self.dispatcher.register_message_handler(l, commands=__messages_keys__)
|
||||||
Reference in new issue
Block a user