143 lines
5.3 KiB
Python
143 lines
5.3 KiB
Python
""" Respond to commands """
|
|
|
|
import telethon
|
|
import telethon.utils
|
|
|
|
import pgbotlib.dbstuff
|
|
import pgbotlib.misc
|
|
import pgbotlib.response
|
|
|
|
|
|
# TODO: quote via response?
|
|
# chat = await event.get_chat()
|
|
# result = await client.get_messages(chat.id, ids=[event.message.reply_to.reply_to_msg_id])
|
|
# print(result)
|
|
class Commander:
|
|
T_START = frozenset(['cmd_start'])
|
|
T_START_E = frozenset(['cmd_start_enabled'])
|
|
T_STOP = frozenset(['cmd_stop'])
|
|
T_STOP_D = frozenset(['cmd_stop_disabled'])
|
|
NOPE = "а ты что ещё за хуй с горы?"
|
|
YEP = "да, господин!"
|
|
DOC = """
|
|
Команды:
|
|
__.start__
|
|
запустить бота
|
|
__.stop__
|
|
остановить бота
|
|
__.list__
|
|
перечислить доступные токены
|
|
__.regex token__
|
|
перечислить регулярные выражения, относящиеся к токену
|
|
__.chat__
|
|
получить id текущего чата
|
|
__.users__
|
|
перечислить id пользователей
|
|
__.add token1[,token2,...] your phrase here__
|
|
добавить фразу your phrase here для реакции на токены
|
|
__.adduser id имя__
|
|
добавить пользователю имя
|
|
__.help__
|
|
вывести этот текст
|
|
"""
|
|
|
|
def __init__(self, config: dict,
|
|
client: telethon.TelegramClient,
|
|
admins: list,
|
|
db_conn: pgbotlib.dbstuff.DBConn,
|
|
namegen: pgbotlib.misc.NameGenerator,
|
|
responder: pgbotlib.response.Responder) -> None:
|
|
self.chats = config['chats']
|
|
self.client = client
|
|
self.admins = admins
|
|
self.db_conn = db_conn
|
|
self.namegen = namegen
|
|
self.responder = responder
|
|
self.available_tokens = [
|
|
str(token) for token, _ in self.responder.tokens]
|
|
|
|
def __add_response(self, caller: int, command: str) -> bool:
|
|
if caller not in self.admins:
|
|
return self.NOPE
|
|
input_tokens, phrase = command.strip().split(' ', 1)
|
|
input_tokenset = frozenset(input_tokens.split(','))
|
|
for token in input_tokenset:
|
|
if token not in self.available_tokens:
|
|
return False
|
|
query = 'INSERT INTO responses (tokens, response) values (%s,%s)'
|
|
values = (','.join(sorted(input_tokenset)), phrase.strip())
|
|
self.db_conn.update(query, values)
|
|
return self.YEP
|
|
|
|
def __add_user(self, caller: int, userspec: str) -> bool:
|
|
if caller not in self.admins:
|
|
return self.NOPE
|
|
user_id, names = userspec.strip().split(' ', 1)
|
|
for name in names.strip().split(','):
|
|
query = 'INSERT INTO names (tg_id, name) values(%s,%s)'
|
|
values = (user_id, name)
|
|
self.db_conn.update(query, values)
|
|
return self.YEP
|
|
|
|
|
|
def __start_response(self) -> str:
|
|
if self.responder.is_enabled():
|
|
return self.responder.get_response(self.T_START_E)
|
|
return self.responder.get_response(self.T_START)
|
|
|
|
def __stop_response(self) -> str:
|
|
if self.responder.is_enabled():
|
|
return self.responder.get_response(self.T_STOP)
|
|
return self.responder.get_response(self.T_STOP_D)
|
|
|
|
def __list_users(self, users: list) -> str:
|
|
userlist = [f'{user.id}: {self.namegen.get_tg_name(user)}'
|
|
for user in users]
|
|
return '\n'.join(userlist)
|
|
|
|
def __list_regex(self, token: str) -> str:
|
|
for t, r in self.responder.tokens:
|
|
if token == t:
|
|
regexlist = [i.pattern for i in r]
|
|
return '\n'.join(regexlist)
|
|
return 'not found!'
|
|
|
|
async def action(self,
|
|
event: telethon.events.common.EventBuilder) -> None:
|
|
chat_id = telethon.utils.get_peer_id(event.message.peer_id)
|
|
if chat_id not in self.chats:
|
|
return None
|
|
command = event.message.text
|
|
sender = await event.get_sender()
|
|
response = None
|
|
if command.startswith('.add '):
|
|
try:
|
|
response = self.__add_response(sender.id, command[5:])
|
|
except Exception as e:
|
|
response = str(e)
|
|
elif command.startswith('.adduser '):
|
|
try:
|
|
response = self.__add_user(sender.id, command[9:])
|
|
except Exception as e:
|
|
response = str(e)
|
|
elif command == '.chat':
|
|
response = str(chat_id)
|
|
elif command == '.list':
|
|
response = ', '.join(self.available_tokens)
|
|
elif command.startswith('.regex '):
|
|
response = self.__list_regex(command[7:].strip())
|
|
elif command == '.users':
|
|
users = await self.client.get_participants(
|
|
entity=event.message.peer_id)
|
|
response = self.__list_users(users)
|
|
elif command == '.start':
|
|
response = self.__start_response()
|
|
self.responder.enable()
|
|
elif command == '.stop':
|
|
response = self.__stop_response()
|
|
self.responder.disable()
|
|
elif command == '.help':
|
|
response = self.DOC
|
|
if response:
|
|
await self.client.send_message(event.message.peer_id, response)
|
|
return None
|