101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
""" Respond to commands """
|
|
|
|
import telethon
|
|
|
|
import pgbotlib.api
|
|
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_d'])
|
|
|
|
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.config = config
|
|
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_entry(self, caller: int, command: str) -> bool:
|
|
if caller not in self.admins:
|
|
print('fuck off!')
|
|
return None
|
|
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())
|
|
return self.db_conn.update(query, values)
|
|
|
|
def __add_user(self, caller: int, userspec: str) -> bool:
|
|
if caller not in self.admins:
|
|
print('fuck off!')
|
|
return None
|
|
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 True
|
|
|
|
def __start_response(self) -> str:
|
|
if self.responder.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.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)
|
|
|
|
async def action(self,
|
|
event: telethon.events.common.EventBuilder) -> None:
|
|
command = event.message.text
|
|
sender = await event.get_sender()
|
|
response = None
|
|
if command.startswith('/add '):
|
|
if self.__add_entry(sender.id, command[5:]):
|
|
response = 'success'
|
|
else:
|
|
response = 'failure'
|
|
elif command.startswith('/adduser '):
|
|
self.__add_user(sender.id, command[9:])
|
|
elif command == '/list':
|
|
response = ', '.join(self.available_tokens)
|
|
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()
|
|
if response:
|
|
await self.client.send_message(event.message.peer_id, response)
|
|
return None
|