Davte/davtelepot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Moved messages to dedicated module

Browse Source
This commit is contained in:
Davte 2020-02-18 18:50:19 +01:00
parent 61cebd0c1f
commit 086750c77e
4 changed files with 484 additions and 505 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
davtelepot/__init__.py
View File

@ -14,7 +14,7 @@ __author__ = "Davide Testa"
__email__ = "davide@davte.it" __email__ = "davide@davte.it"
__credits__ = ["Marco Origlia", "Nick Lee @Nickoala"] __credits__ = ["Marco Origlia", "Nick Lee @Nickoala"]
__license__ = "GNU General Public License v3.0" __license__ = "GNU General Public License v3.0"
__version__ = "2.4.5" __version__ = "2.4.6"
__maintainer__ = "Davide Testa" __maintainer__ = "Davide Testa"
__contact__ = "t.me/davte" __contact__ = "t.me/davte"

595
davtelepot/administration_tools.py
View File

@ -14,6 +14,7 @@ import datetime
import json import json
# Third party modules # Third party modules
from davtelepot import messages
from davtelepot.utilities import ( from davtelepot.utilities import (
async_wrapper, Confirmator, extract, get_cleaned_text, get_user, async_wrapper, Confirmator, extract, get_cleaned_text, get_user,
escape_html_chars, line_drawing_unordered_list, make_button, escape_html_chars, line_drawing_unordered_list, make_button,
@ -23,110 +24,6 @@ from davtelepot.utilities import (
from sqlalchemy.exc import ResourceClosedError from sqlalchemy.exc import ResourceClosedError
default_talk_messages = dict(
admin_session_ended=dict(
en=(
'Session with user {u} ended.'
),
it=(
'Sessione terminata con l\'utente {u}.'
),
),
admin_warning=dict(
en=(
'You are now talking to {u}.\n'
'Until you end this session, your messages will be '
'forwarded to each other.'
),
it=(
'Sei ora connesso con {u}.\n'
'Finché non chiuderai la connessione, i messaggi che scriverai '
'qui saranno inoltrati a {u}, e ti inoltrerò i suoi.'
),
),
end_session=dict(
en=(
'End session?'
),
it=(
'Chiudere la sessione?'
),
),
help_text=dict(
en='Press the button to search for user.',
it='Premi il pulsante per scegliere un utente.'
),
search_button=dict(
en="🔍 Search for user",
it="🔍 Cerca utente",
),
select_user=dict(
en='Which user would you like to talk to?',
it='Con quale utente vorresti parlare?'
),
user_not_found=dict(
en=(
"Sory, but no user matches your query for\n"
"<code>{q}</code>"
),
it=(
"Spiacente, ma nessun utente corrisponde alla ricerca per\n"
"<code>{q}</code>"
),
),
instructions=dict(
en=(
'Write a part of name, surname or username of the user you want '
'to talk to.'
),
it=(
'Scrivi una parte del nome, cognome o username dell\'utente con '
'cui vuoi parlare.'
),
),
stop=dict(
en=(
'End session'
),
it=(
'Termina la sessione'
),
),
user_session_ended=dict(
en=(
'Session with admin {u} ended.'
),
it=(
'Sessione terminata con l\'amministratore {u}.'
),
),
user_warning=dict(
en=(
'{u}, admin of this bot, wants to talk to you.\n'
'Until this session is ended by {u}, your messages will be '
'forwarded to each other.'
),
it=(
'{u}, amministratore di questo bot, vuole parlare con te.\n'
'Finché non chiuderà la connessione, i messaggi che scriverai '
'qui saranno inoltrati a {u}, e ti inoltrerò i suoi.'
),
),
# key=dict(
# en='',
# it='',
# ),
# key=dict(
# en=(
# ''
# ),
# it=(
# ''
# ),
# ),
)
async def _forward_to(update, bot, sender, addressee, is_admin=False): async def _forward_to(update, bot, sender, addressee, is_admin=False):
if update['text'].lower() in ['stop'] and is_admin: if update['text'].lower() in ['stop'] and is_admin:
with bot.db as db: with bot.db as db:
@ -411,7 +308,7 @@ async def end_session(bot, other_user_record, admin_record):
u=get_user(other_user_record) u=get_user(other_user_record)
), ),
) )
for record in (admin_record, other_user_record, ): for record in (admin_record, other_user_record,):
bot.remove_individual_text_message_handler(record['telegram_id']) bot.remove_individual_text_message_handler(record['telegram_id'])
return return
@ -434,8 +331,8 @@ async def _talk_button(bot, update, user_record, data):
reply_markup = None reply_markup = None
elif command == 'select': elif command == 'select':
if ( if (
len(arguments) < 1 len(arguments) < 1
or type(arguments[0]) is not int or type(arguments[0]) is not int
): ):
result = "Errore!" result = "Errore!"
else: else:
@ -453,8 +350,8 @@ async def _talk_button(bot, update, user_record, data):
) )
elif command == 'stop': elif command == 'stop':
if ( if (
len(arguments) < 1 len(arguments) < 1
or type(arguments[0]) is not int or type(arguments[0]) is not int
): ):
result = "Errore!" result = "Errore!"
elif not Confirmator.get('stop_bots').confirm(telegram_id): elif not Confirmator.get('stop_bots').confirm(telegram_id):
@ -490,245 +387,6 @@ async def _talk_button(bot, update, user_record, data):
return result return result
default_admin_messages = {
'talk_command': {
'description': {
'en': "Choose a user and forward messages to each other",
'it': "Scegli un utente e il bot farà da tramite inoltrando a "
"ognuno i messaggi dell'altro finché non terminerai la "
"sessione"
}
},
'restart_command': {
'description': {
'en': "Restart bots",
'it': "Riavvia i bot"
},
'restart_scheduled_message': {
'en': "Bots are being restarted, after pulling from repository.",
'it': "I bot verranno riavviati in pochi secondi, caricando "
"prima le eventuali modifiche al codice."
},
'restart_completed_message': {
'en': "<i>Restart was successful.</i>",
'it': "<i>Restart avvenuto con successo.</i>"
}
},
'stop_command': {
'description': {
'en': "Stop bots",
'it': "Ferma i bot"
},
'text': {
'en': "Are you sure you want to stop all bots?\n"
"To make them start again you will have to ssh-log "
"in server.\n\n"
"To restart the bots remotely use the /restart command "
"instead (before starting over, a <code>git pull</code> "
"is performed).",
'it': "Sei sicuro di voler fermare i bot?\n"
"Per farli ripartire dovrai accedere al server.\n\n"
"Per far ripartire i bot da remoto usa invece il comando "
"/restart (prima di ripartire farò un "
"<code>git pull</code>)."
}
},
'stop_button': {
'stop_text': {
'en': "Stop bots",
'it': "Ferma i bot"
},
'cancel': {
'en': "Cancel",
'it': "Annulla"
},
'confirm': {
'en': "Do you really want to stop all bots?",
'it': "Vuoi davvero fermare tutti i bot?"
},
'stopping': {
'en': "Stopping bots...",
'it': "Arresto in corso..."
},
'cancelled': {
'en': "Operation was cancelled",
'it': "Operazione annullata"
}
},
'db_command': {
'description': {
'en': "Ask for bot database via Telegram",
'it': "Ricevi il database del bot via Telegram"
},
'not_sqlite': {
'en': "Only SQLite databases may be sent via Telegram, since they "
"are single-file databases.\n"
"This bot has a `{db_type}` database.",
'it': "Via Telegram possono essere inviati solo database SQLite, "
"in quanto composti di un solo file.\n"
"Questo bot ha invece un database `{db_type}`."
},
'file_caption': {
'en': "Here is bot database.",
'it': "Ecco il database!"
},
'db_sent': {
'en': "Database sent.",
'it': "Database inviato."
}
},
'query_command': {
'description': {
'en': "Receive the result of a SQL query performed on bot "
"database",
'it': "Ricevi il risultato di una query SQL sul database del bot"
},
'help': {
'en': "Write a SQL query to be run on bot database.\n\n"
"<b>Example</b>\n"
"<code>/query SELECT * FROM users WHERE 0</code>",
'it': "Invia una query SQL da eseguire sul database del bot.\n\n"
"<b>Esempio</b>\n"
"<code>/query SELECT * FROM users WHERE 0</code>"
},
'no_iterable': {
'en': "No result to show was returned",
'it': "La query non ha restituito risultati da mostrare"
},
'exception': {
'en': "The query threw this error:",
'it': "La query ha dato questo errore:"
},
'result': {
'en': "Query result",
'it': "Risultato della query"
}
},
'select_command': {
'description': {
'en': "Receive the result of a SELECT query performed on bot "
"database",
'it': "Ricevi il risultato di una query SQL di tipo SELECT "
"sul database del bot"
}
},
'query_button': {
'error': {
'en': "Error!",
'it': "Errore!"
},
'file_name': {
'en': "Query result.csv",
'it': "Risultato della query.csv"
},
'empty_file': {
'en': "No result to show.",
'it': "Nessun risultato da mostrare."
}
},
'log_command': {
'description': {
'en': "Receive bot log file, if set",
'it': "Ricevi il file di log del bot, se impostato"
},
'no_log': {
'en': "Sorry but no log file is set.\n"
"To set it, use `bot.set_log_file_name` instance method or "
"`Bot.set_class_log_file_name` class method.",
'it': "Spiacente ma il file di log non è stato impostato.\n"
"Per impostarlo, usa il metodo d'istanza "
"`bot.set_log_file_name` o il metodo di classe"
"`Bot.set_class_log_file_name`."
},
'sending_failure': {
'en': "Sending log file failed!\n\n"
"<b>Error:</b>\n"
"<code>{e}</code>",
'it': "Inviio del messaggio di log fallito!\n\n"
"<b>Errore:</b>\n"
"<code>{e}</code>"
},
'here_is_log_file': {
'en': "Here is the complete log file.",
'it': "Ecco il file di log completo."
},
'log_file_first_lines': {
'en': "Here are the first {lines} lines of the log file.",
'it': "Ecco le prime {lines} righe del file di log."
},
'log_file_last_lines': {
'en': "Here are the last {lines} lines of the log file.\n"
"Newer lines are at the top of the file.",
'it': "Ecco le ultime {lines} righe del file di log.\n"
"L'ordine è cronologico, con i messaggi nuovi in alto."
}
},
'errors_command': {
'description': {
'en': "Receive bot error log file, if set",
'it': "Ricevi il file di log degli errori del bot, se impostato"
},
'no_log': {
'en': "Sorry but no errors log file is set.\n"
"To set it, use `bot.set_errors_file_name` instance method"
"or `Bot.set_class_errors_file_name` class method.",
'it': "Spiacente ma il file di log degli errori non è stato "
"impostato.\n"
"Per impostarlo, usa il metodo d'istanza "
"`bot.set_errors_file_name` o il metodo di classe"
"`Bot.set_class_errors_file_name`."
},
'empty_log': {
'en': "Congratulations! Errors log is empty!",
'it': "Congratulazioni! Il log degli errori è vuoto!"
},
'sending_failure': {
'en': "Sending errors log file failed!\n\n"
"<b>Error:</b>\n"
"<code>{e}</code>",
'it': "Inviio del messaggio di log degli errori fallito!\n\n"
"<b>Errore:</b>\n"
"<code>{e}</code>"
},
'here_is_log_file': {
'en': "Here is the complete errors log file.",
'it': "Ecco il file di log degli errori completo."
},
'log_file_first_lines': {
'en': "Here are the first {lines} lines of the errors log file.",
'it': "Ecco le prime {lines} righe del file di log degli errori."
},
'log_file_last_lines': {
'en': "Here are the last {lines} lines of the errors log file.\n"
"Newer lines are at the top of the file.",
'it': "Ecco le ultime {lines} righe del file di log degli "
"errori.\n"
"L'ordine è cronologico, con i messaggi nuovi in alto."
}
},
'maintenance_command': {
'description': {
'en': "Put the bot under maintenance",
'it': "Metti il bot in manutenzione"
},
'maintenance_started': {
'en': "<i>Bot has just been put under maintenance!</i>\n\n"
"Until further notice, it will reply to users "
"with the following message:\n\n"
"{message}",
'it': "<i>Il bot è stato messo in manutenzione!</i>\n\n"
"Fino a nuovo ordine, risponderà a tutti i comandi con il "
"seguente messaggio\n\n"
"{message}"
},
'maintenance_ended': {
'en': "<i>Maintenance ended!</i>",
'it': "<i>Manutenzione terminata!</i>"
}
}
}
async def _restart_command(bot, update, user_record): async def _restart_command(bot, update, user_record):
with bot.db as db: with bot.db as db:
db['restart_messages'].insert( db['restart_messages'].insert(
@ -801,22 +459,22 @@ async def _stop_button(bot, update, user_record, data):
if command == 'stop': if command == 'stop':
if not Confirmator.get('stop_bots').confirm(telegram_id): if not Confirmator.get('stop_bots').confirm(telegram_id):
return bot.get_message( return bot.get_message(
'admin', 'stop_button', 'confirm', 'admin', 'stop_button', 'confirm',
update=update, user_record=user_record
)
text = bot.get_message(
'admin', 'stop_button', 'stopping',
update=update, user_record=user_record update=update, user_record=user_record
) )
text = bot.get_message(
'admin', 'stop_button', 'stopping',
update=update, user_record=user_record
)
result = text result = text
# Do not stop bots immediately, otherwise callback query # Do not stop bots immediately, otherwise callback query
# will never be answered # will never be answered
asyncio.ensure_future(stop_bots(bot)) asyncio.ensure_future(stop_bots(bot))
elif command == 'cancel': elif command == 'cancel':
text = bot.get_message( text = bot.get_message(
'admin', 'stop_button', 'cancelled', 'admin', 'stop_button', 'cancelled',
update=update, user_record=user_record update=update, user_record=user_record
) )
result = text result = text
if text: if text:
return dict( return dict(
@ -833,10 +491,10 @@ async def _stop_button(bot, update, user_record, data):
async def _send_bot_database(bot, update, user_record): async def _send_bot_database(bot, update, user_record):
if not all( if not all(
[ [
bot.db_url.endswith('.db'), bot.db_url.endswith('.db'),
bot.db_url.startswith('sqlite:///') bot.db_url.startswith('sqlite:///')
] ]
): ):
return bot.get_message( return bot.get_message(
'admin', 'db_command', 'not_sqlite', 'admin', 'db_command', 'not_sqlite',
@ -905,14 +563,14 @@ async def _query_command(bot, update, user_record):
e=e e=e
) )
result = ( result = (
"<b>{first_line}</b>\n".format( "<b>{first_line}</b>\n".format(
first_line=bot.get_message( first_line=bot.get_message(
'admin', 'query_command', 'result', 'admin', 'query_command', 'result',
update=update, user_record=user_record update=update, user_record=user_record
)
) )
) + f"<code>{query}</code>\n\n"
+ f"<code>{query}</code>\n\n" f"{result}"
f"{result}"
) )
if query_id: if query_id:
reply_markup = make_inline_keyboard( reply_markup = make_inline_keyboard(
@ -1037,7 +695,7 @@ async def _errors_command(bot, update, user_record):
try: try:
# Check that error log is not empty # Check that error log is not empty
with open(bot.errors_file_path, 'r') as errors_file: with open(bot.errors_file_path, 'r') as errors_file:
for line in errors_file: for _ in errors_file:
break break
else: else:
return bot.get_message( return bot.get_message(
@ -1094,14 +752,15 @@ def get_maintenance_exception_criterion(bot, allowed_command):
`bot` : davtelepot.bot.Bot() instance `bot` : davtelepot.bot.Bot() instance
`allowed_command` : str (command to be allowed during maintenance) `allowed_command` : str (command to be allowed during maintenance)
""" """
def criterion(update): def criterion(update):
if 'message' not in update: if 'message' not in update:
return False return False
update = update['message'] update = update['message']
text = get_cleaned_text(update, bot, []) text = get_cleaned_text(update, bot, [])
if ( if (
'from' not in update 'from' not in update
or 'id' not in update['from'] or 'id' not in update['from']
): ):
return False return False
with bot.db as db: with bot.db as db:
@ -1109,164 +768,162 @@ def get_maintenance_exception_criterion(bot, allowed_command):
telegram_id=update['from']['id'] telegram_id=update['from']['id']
) )
if not bot.authorization_function( if not bot.authorization_function(
update=update, update=update,
user_record=user_record, user_record=user_record,
authorization_level=2 authorization_level=2
): ):
return False return False
return text == allowed_command.strip('/') return text == allowed_command.strip('/')
return criterion return criterion
def init(bot, talk_messages=None, admin_messages=None): def init(telegram_bot, talk_messages=None, admin_messages=None):
"""Assign parsers, commands, buttons and queries to given `bot`.""" """Assign parsers, commands, buttons and queries to given `bot`."""
if talk_messages is None: if talk_messages is None:
talk_messages = default_talk_messages talk_messages = messages.default_talk_messages
bot.messages['talk'] = talk_messages telegram_bot.messages['talk'] = talk_messages
if admin_messages is None: if admin_messages is None:
admin_messages = default_admin_messages admin_messages = messages.default_admin_messages
bot.messages['admin'] = admin_messages telegram_bot.messages['admin'] = admin_messages
with bot.db as db: db = telegram_bot.db
if 'talking_sessions' not in db.tables: if 'talking_sessions' not in db.tables:
db['talking_sessions'].insert( db['talking_sessions'].insert(
dict( dict(
user=0, user=0,
admin=0, admin=0,
cancelled=1 cancelled=1
)
) )
)
allowed_during_maintenance = [ allowed_during_maintenance = [
get_maintenance_exception_criterion(bot, command) get_maintenance_exception_criterion(telegram_bot, command)
for command in ['stop', 'restart', 'maintenance'] for command in ['stop', 'restart', 'maintenance']
] ]
@bot.additional_task(when='BEFORE') @telegram_bot.additional_task(when='BEFORE')
async def load_talking_sessions(): async def load_talking_sessions():
sessions = [] sessions = []
with bot.db as db: for session in db.query(
for session in db.query(
"""SELECT * """SELECT *
FROM talking_sessions FROM talking_sessions
WHERE NOT cancelled WHERE NOT cancelled
""" """
): ):
sessions.append( sessions.append(
dict( dict(
other_user_record=db['users'].find_one( other_user_record=db['users'].find_one(
id=session['user'] id=session['user']
), ),
admin_record=db['users'].find_one( admin_record=db['users'].find_one(
id=session['admin'] id=session['admin']
), ),
)
)
for session in sessions:
await start_session(
bot=bot,
other_user_record=session['other_user_record'],
admin_record=session['admin_record']
) )
)
for session in sessions:
await start_session(
bot=telegram_bot,
other_user_record=session['other_user_record'],
admin_record=session['admin_record']
)
@bot.command(command='/talk', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/talk', aliases=[], show_in_keyboard=False,
description=admin_messages['talk_command']['description'], description=admin_messages['talk_command']['description'],
authorization_level='admin') authorization_level='admin')
async def talk_command(bot, update, user_record): async def talk_command(bot, update, user_record):
return await _talk_command(bot, update, user_record) return await _talk_command(bot, update, user_record)
@bot.button(prefix='talk:///', separator='|', authorization_level='admin') @telegram_bot.button(prefix='talk:///', separator='|', authorization_level='admin')
async def talk_button(bot, update, user_record, data): async def talk_button(bot, update, user_record, data):
return await _talk_button(bot, update, user_record, data) return await _talk_button(bot, update, user_record, data)
@bot.command(command='/restart', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/restart', aliases=[], show_in_keyboard=False,
description=admin_messages['restart_command']['description'], description=admin_messages['restart_command']['description'],
authorization_level='admin') authorization_level='admin')
async def restart_command(bot, update, user_record): async def restart_command(bot, update, user_record):
return await _restart_command(bot, update, user_record) return await _restart_command(bot, update, user_record)
@bot.additional_task('BEFORE') @telegram_bot.additional_task('BEFORE')
async def send_restart_messages(): async def send_restart_messages():
"""Send restart messages at restart.""" """Send restart messages at restart."""
with bot.db as db: for restart_message in db['restart_messages'].find(sent=None):
for restart_message in db['restart_messages'].find(sent=None): asyncio.ensure_future(
asyncio.ensure_future( telegram_bot.send_message(
bot.send_message( **{
**{ key: val
key: val for key, val in restart_message.items()
for key, val in restart_message.items() if key in (
if key in ( 'chat_id',
'chat_id', 'text',
'text', 'parse_mode',
'parse_mode', 'reply_to_message_id'
'reply_to_message_id' )
) }
}
)
)
db['restart_messages'].update(
dict(
sent=datetime.datetime.now(),
id=restart_message['id']
),
['id'],
ensure=True
) )
)
db['restart_messages'].update(
dict(
sent=datetime.datetime.now(),
id=restart_message['id']
),
['id'],
ensure=True
)
return return
@bot.command(command='/stop', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/stop', aliases=[], show_in_keyboard=False,
description=admin_messages['stop_command']['description'], description=admin_messages['stop_command']['description'],
authorization_level='admin') authorization_level='admin')
async def stop_command(bot, update, user_record): async def stop_command(bot, update, user_record):
return await _stop_command(bot, update, user_record) return await _stop_command(bot, update, user_record)
@bot.button(prefix='stop:///', separator='|', @telegram_bot.button(prefix='stop:///', separator='|',
description=admin_messages['stop_command']['description'], description=admin_messages['stop_command']['description'],
authorization_level='admin') authorization_level='admin')
async def stop_button(bot, update, user_record, data): async def stop_button(bot, update, user_record, data):
return await _stop_button(bot, update, user_record, data) return await _stop_button(bot, update, user_record, data)
@bot.command(command='/db', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/db', aliases=[], show_in_keyboard=False,
description=admin_messages['db_command']['description'], description=admin_messages['db_command']['description'],
authorization_level='admin') authorization_level='admin')
async def send_bot_database(bot, update, user_record): async def send_bot_database(bot, update, user_record):
return await _send_bot_database(bot, update, user_record) return await _send_bot_database(bot, update, user_record)
@bot.command(command='/query', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/query', aliases=[], show_in_keyboard=False,
description=admin_messages['query_command']['description'], description=admin_messages['query_command']['description'],
authorization_level='admin') authorization_level='admin')
async def query_command(bot, update, user_record): async def query_command(bot, update, user_record):
return await _query_command(bot, update, user_record) return await _query_command(bot, update, user_record)
@bot.command(command='/select', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/select', aliases=[], show_in_keyboard=False,
description=admin_messages['select_command']['description'], description=admin_messages['select_command']['description'],
authorization_level='admin') authorization_level='admin')
async def select_command(bot, update, user_record): async def select_command(bot, update, user_record):
return await _query_command(bot, update, user_record) return await _query_command(bot, update, user_record)
@bot.button(prefix='db_query:///', separator='|', @telegram_bot.button(prefix='db_query:///', separator='|',
description=admin_messages['query_command']['description'], description=admin_messages['query_command']['description'],
authorization_level='admin') authorization_level='admin')
async def query_button(bot, update, user_record, data): async def query_button(bot, update, user_record, data):
return await _query_button(bot, update, user_record, data) return await _query_button(bot, update, user_record, data)
@bot.command(command='/log', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/log', aliases=[], show_in_keyboard=False,
description=admin_messages['log_command']['description'], description=admin_messages['log_command']['description'],
authorization_level='admin') authorization_level='admin')
async def log_command(bot, update, user_record): async def log_command(bot, update, user_record):
return await _log_command(bot, update, user_record) return await _log_command(bot, update, user_record)
@bot.command(command='/errors', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/errors', aliases=[], show_in_keyboard=False,
description=admin_messages['errors_command']['description'], description=admin_messages['errors_command']['description'],
authorization_level='admin') authorization_level='admin')
async def errors_command(bot, update, user_record): async def errors_command(bot, update, user_record):
return await _errors_command(bot, update, user_record) return await _errors_command(bot, update, user_record)
for exception in allowed_during_maintenance: for exception in allowed_during_maintenance:
bot.allow_during_maintenance(exception) telegram_bot.allow_during_maintenance(exception)
@bot.command(command='/maintenance', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/maintenance', aliases=[], show_in_keyboard=False,
description=admin_messages[ description=admin_messages['maintenance_command']['description'],
'maintenance_command']['description'], authorization_level='admin')
authorization_level='admin')
async def maintenance_command(bot, update, user_record): async def maintenance_command(bot, update, user_record):
return await _maintenance_command(bot, update, user_record) return await _maintenance_command(bot, update, user_record)

65
davtelepot/authorization.py
View File

@ -60,7 +60,7 @@ DEFAULT_ROLES[100] = {
} }
class Role(): class Role:
"""Authorization level for users of a bot.""" """Authorization level for users of a bot."""
roles = OrderedDict() roles = OrderedDict()
@ -202,11 +202,11 @@ class Role():
def __gt__(self, other): def __gt__(self, other):
"""Return True if self can appoint other.""" """Return True if self can appoint other."""
return ( return (
( (
self.code < other.code self.code < other.code
or other.code == 0 or other.code == 0
) )
and self.code in other.can_be_appointed_by and self.code in other.can_be_appointed_by
) )
def __ge__(self, other): def __ge__(self, other):
@ -232,6 +232,7 @@ class Role():
def get_authorization_function(bot): def get_authorization_function(bot):
"""Take a `bot` and return its authorization_function.""" """Take a `bot` and return its authorization_function."""
def is_authorized(update, user_record=None, authorization_level=2): def is_authorized(update, user_record=None, authorization_level=2):
"""Return True if user role is at least at `authorization_level`.""" """Return True if user role is at least at `authorization_level`."""
user_role = bot.Role.get_user_role(user_record=user_record) user_role = bot.Role.get_user_role(user_record=user_record)
@ -241,6 +242,7 @@ def get_authorization_function(bot):
if needed_role.code < user_role.code: if needed_role.code < user_role.code:
return False return False
return True return True
return is_authorized return is_authorized
@ -412,8 +414,8 @@ async def _authorization_button(bot, update, user_record, data):
elif command in ['set'] and len(arguments) > 1: elif command in ['set'] and len(arguments) > 1:
other_user_id, new_privileges, *_ = arguments other_user_id, new_privileges, *_ = arguments
if not Confirmator.get( if not Confirmator.get(
key=f'{user_id}_set_{other_user_id}', key=f'{user_id}_set_{other_user_id}',
confirm_timedelta=5 confirm_timedelta=5
).confirm: ).confirm:
return bot.get_message( return bot.get_message(
'authorization', 'auth_button', 'confirm', 'authorization', 'auth_button', 'confirm',
@ -497,58 +499,55 @@ async def _ban_command(bot, update, user_record):
return return
def init(bot: Bot, roles=None, authorization_messages=None): def init(telegram_bot: Bot, roles=None, authorization_messages=None):
"""Set bot roles and assign role-related commands. """Set bot roles and assign role-related commands.
Pass an OrderedDict of `roles` to get them set. Pass an OrderedDict of `roles` to get them set.
""" """
class _Role(Role): class _Role(Role):
roles = OrderedDict() roles = OrderedDict()
bot.set_role_class(_Role) telegram_bot.set_role_class(_Role)
if roles is None: if roles is None:
roles = DEFAULT_ROLES roles = DEFAULT_ROLES
# Cast roles to OrderedDict # Cast roles to OrderedDict
if isinstance(roles, list): if isinstance(roles, list):
roles = OrderedDict( roles = OrderedDict(
(i, element) (i, element)
for i, element in enumerate(list) for i, element in enumerate(roles)
) )
if not isinstance(roles, OrderedDict): if not isinstance(roles, OrderedDict):
raise TypeError("`roles` shall be a OrderedDict!") raise TypeError("`roles` shall be a OrderedDict!")
for id, role in roles.items(): for code, role in roles.items():
if 'code' not in role: if 'code' not in role:
role['code'] = id role['code'] = code
bot.Role(**role) telegram_bot.Role(**role)
bot.set_authorization_function( telegram_bot.set_authorization_function(
get_authorization_function(bot) get_authorization_function(telegram_bot)
) )
if authorization_messages is None: if authorization_messages is None:
authorization_messages = deafult_authorization_messages authorization_messages = deafult_authorization_messages
bot.messages['authorization'] = authorization_messages telegram_bot.messages['authorization'] = authorization_messages
@bot.command(command='/auth', aliases=[], show_in_keyboard=False, @telegram_bot.command(command='/auth', aliases=[], show_in_keyboard=False,
description=( description=(
authorization_messages['auth_command']['description'] authorization_messages['auth_command']['description']
), ),
authorization_level='moderator') authorization_level='moderator')
async def authorization_command(bot, update, user_record): async def authorization_command(bot, update, user_record):
return await _authorization_command(bot, update, user_record) return await _authorization_command(bot, update, user_record)
@bot.button('auth:///', @telegram_bot.button('auth:///',
description=( description=authorization_messages['auth_button']['description'],
authorization_messages['auth_button']['description'] separator='|',
), authorization_level='moderator')
separator='|',
authorization_level='moderator')
async def authorization_button(bot, update, user_record, data): async def authorization_button(bot, update, user_record, data):
return await _authorization_button(bot, update, user_record, data) return await _authorization_button(bot, update, user_record, data)
@bot.command('/ban', aliases=[], show_in_keyboard=False, @telegram_bot.command('/ban', aliases=[], show_in_keyboard=False,
description=( description=authorization_messages['ban_command']['description'],
authorization_messages['ban_command']['description'] authorization_level='admin')
),
authorization_level='admin')
async def ban_command(bot, update, user_record): async def ban_command(bot, update, user_record):
return await _ban_command(bot, update, user_record) return await _ban_command(bot, update, user_record)

327
davtelepot/messages.py
View File

@ -1,5 +1,248 @@
"""Default messages for bot functions.""" """Default messages for bot functions."""
default_admin_messages = {
'talk_command': {
'description': {
'en': "Choose a user and forward messages to each other",
'it': "Scegli un utente e il bot farà da tramite inoltrando a "
"ognuno i messaggi dell'altro finché non terminerai la "
"sessione"
}
},
'restart_command': {
'description': {
'en': "Restart bots",
'it': "Riavvia i bot"
},
'restart_scheduled_message': {
'en': "Bots are being restarted, after pulling from repository.",
'it': "I bot verranno riavviati in pochi secondi, caricando "
"prima le eventuali modifiche al codice."
},
'restart_completed_message': {
'en': "<i>Restart was successful.</i>",
'it': "<i>Restart avvenuto con successo.</i>"
}
},
'stop_command': {
'description': {
'en': "Stop bots",
'it': "Ferma i bot"
},
'text': {
'en': "Are you sure you want to stop all bots?\n"
"To make them start again you will have to ssh-log "
"in server.\n\n"
"To restart the bots remotely use the /restart command "
"instead (before starting over, a <code>git pull</code> "
"is performed).",
'it': "Sei sicuro di voler fermare i bot?\n"
"Per farli ripartire dovrai accedere al server.\n\n"
"Per far ripartire i bot da remoto usa invece il comando "
"/restart (prima di ripartire farò un "
"<code>git pull</code>)."
}
},
'stop_button': {
'stop_text': {
'en': "Stop bots",
'it': "Ferma i bot"
},
'cancel': {
'en': "Cancel",
'it': "Annulla"
},
'confirm': {
'en': "Do you really want to stop all bots?",
'it': "Vuoi davvero fermare tutti i bot?"
},
'stopping': {
'en': "Stopping bots...",
'it': "Arresto in corso..."
},
'cancelled': {
'en': "Operation was cancelled",
'it': "Operazione annullata"
}
},
'db_command': {
'description': {
'en': "Ask for bot database via Telegram",
'it': "Ricevi il database del bot via Telegram"
},
'not_sqlite': {
'en': "Only SQLite databases may be sent via Telegram, since they "
"are single-file databases.\n"
"This bot has a `{db_type}` database.",
'it': "Via Telegram possono essere inviati solo database SQLite, "
"in quanto composti di un solo file.\n"
"Questo bot ha invece un database `{db_type}`."
},
'file_caption': {
'en': "Here is bot database.",
'it': "Ecco il database!"
},
'db_sent': {
'en': "Database sent.",
'it': "Database inviato."
}
},
'query_command': {
'description': {
'en': "Receive the result of a SQL query performed on bot "
"database",
'it': "Ricevi il risultato di una query SQL sul database del bot"
},
'help': {
'en': "Write a SQL query to be run on bot database.\n\n"
"<b>Example</b>\n"
"<code>/query SELECT * FROM users WHERE 0</code>",
'it': "Invia una query SQL da eseguire sul database del bot.\n\n"
"<b>Esempio</b>\n"
"<code>/query SELECT * FROM users WHERE 0</code>"
},
'no_iterable': {
'en': "No result to show was returned",
'it': "La query non ha restituito risultati da mostrare"
},
'exception': {
'en': "The query threw this error:",
'it': "La query ha dato questo errore:"
},
'result': {
'en': "Query result",
'it': "Risultato della query"
}
},
'select_command': {
'description': {
'en': "Receive the result of a SELECT query performed on bot "
"database",
'it': "Ricevi il risultato di una query SQL di tipo SELECT "
"sul database del bot"
}
},
'query_button': {
'error': {
'en': "Error!",
'it': "Errore!"
},
'file_name': {
'en': "Query result.csv",
'it': "Risultato della query.csv"
},
'empty_file': {
'en': "No result to show.",
'it': "Nessun risultato da mostrare."
}
},
'log_command': {
'description': {
'en': "Receive bot log file, if set",
'it': "Ricevi il file di log del bot, se impostato"
},
'no_log': {
'en': "Sorry but no log file is set.\n"
"To set it, use `bot.set_log_file_name` instance method or "
"`Bot.set_class_log_file_name` class method.",
'it': "Spiacente ma il file di log non è stato impostato.\n"
"Per impostarlo, usa il metodo d'istanza "
"`bot.set_log_file_name` o il metodo di classe"
"`Bot.set_class_log_file_name`."
},
'sending_failure': {
'en': "Sending log file failed!\n\n"
"<b>Error:</b>\n"
"<code>{e}</code>",
'it': "Inviio del messaggio di log fallito!\n\n"
"<b>Errore:</b>\n"
"<code>{e}</code>"
},
'here_is_log_file': {
'en': "Here is the complete log file.",
'it': "Ecco il file di log completo."
},
'log_file_first_lines': {
'en': "Here are the first {lines} lines of the log file.",
'it': "Ecco le prime {lines} righe del file di log."
},
'log_file_last_lines': {
'en': "Here are the last {lines} lines of the log file.\n"
"Newer lines are at the top of the file.",
'it': "Ecco le ultime {lines} righe del file di log.\n"
"L'ordine è cronologico, con i messaggi nuovi in alto."
}
},
'errors_command': {
'description': {
'en': "Receive bot error log file, if set",
'it': "Ricevi il file di log degli errori del bot, se impostato"
},
'no_log': {
'en': "Sorry but no errors log file is set.\n"
"To set it, use `bot.set_errors_file_name` instance method"
"or `Bot.set_class_errors_file_name` class method.",
'it': "Spiacente ma il file di log degli errori non è stato "
"impostato.\n"
"Per impostarlo, usa il metodo d'istanza "
"`bot.set_errors_file_name` o il metodo di classe"
"`Bot.set_class_errors_file_name`."
},
'empty_log': {
'en': "Congratulations! Errors log is empty!",
'it': "Congratulazioni! Il log degli errori è vuoto!"
},
'sending_failure': {
'en': "Sending errors log file failed!\n\n"
"<b>Error:</b>\n"
"<code>{e}</code>",
'it': "Inviio del messaggio di log degli errori fallito!\n\n"
"<b>Errore:</b>\n"
"<code>{e}</code>"
},
'here_is_log_file': {
'en': "Here is the complete errors log file.",
'it': "Ecco il file di log degli errori completo."
},
'log_file_first_lines': {
'en': "Here are the first {lines} lines of the errors log file.",
'it': "Ecco le prime {lines} righe del file di log degli errori."
},
'log_file_last_lines': {
'en': "Here are the last {lines} lines of the errors log file.\n"
"Newer lines are at the top of the file.",
'it': "Ecco le ultime {lines} righe del file di log degli "
"errori.\n"
"L'ordine è cronologico, con i messaggi nuovi in alto."
}
},
'maintenance_command': {
'description': {
'en': "Put the bot under maintenance",
'it': "Metti il bot in manutenzione"
},
'maintenance_started': {
'en': "<i>Bot has just been put under maintenance!</i>\n\n"
"Until further notice, it will reply to users "
"with the following message:\n\n"
"{message}",
'it': "<i>Il bot è stato messo in manutenzione!</i>\n\n"
"Fino a nuovo ordine, risponderà a tutti i comandi con il "
"seguente messaggio\n\n"
"{message}"
},
'maintenance_ended': {
'en': "<i>Maintenance ended!</i>",
'it': "<i>Manutenzione terminata!</i>"
}
}
}
default_authorization_denied_message = {
'en': "You are not allowed to use this command, sorry.",
'it': "Non disponi di autorizzazioni sufficienti per questa richiesta, spiacente.",
}
default_help_messages = { default_help_messages = {
'help_command': { 'help_command': {
'header': { 'header': {
@ -33,8 +276,8 @@ default_help_messages = {
}, },
}, },
'commands_button_label': { 'commands_button_label': {
'en': "Commands 🤖", 'en': "Commands 🤖",
'it': "Comandi 🤖", 'it': "Comandi 🤖",
}, },
} }
@ -211,3 +454,83 @@ default_suggestion_messages = {
}, },
} }
} }
default_talk_messages = {
'admin_session_ended': {
'en': 'Session with user {u} ended.',
'it': 'Sessione terminata con l\'utente {u}.',
},
'admin_warning': {
'en': (
'You are now talking to {u}.\n'
'Until you end this session, your messages will be '
'forwarded to each other.'
),
'it': (
'Sei ora connesso con {u}.\n'
'Finché non chiuderai la connessione, i messaggi che scriverai '
'qui saranno inoltrati a {u}, e ti inoltrerò i suoi.'
),
},
'end_session': {
'en': 'End session?',
'it': 'Chiudere la sessione?',
},
'help_text': {
'en': 'Press the button to search for user.',
'it': 'Premi il pulsante per scegliere un utente.',
},
'search_button': {
'en': "🔍 Search for user",
'it': "🔍 Cerca utente",
},
'select_user': {
'en': 'Which user would you like to talk to?',
'it': 'Con quale utente vorresti parlare?',
},
'user_not_found': {
'en': (
"Sory, but no user matches your query for\n"
"<code>{q}</code>"
),
'it': (
"Spiacente, ma nessun utente corrisponde alla ricerca per\n"
"<code>{q}</code>"
),
},
'instructions': {
'en': (
'Write a part of name, surname or username of the user you want '
'to talk to.'
),
'it': (
'Scrivi una parte del nome, cognome o username dell\'utente con '
'cui vuoi parlare.'
),
},
'stop': {
'en': 'End session',
'it': 'Termina la sessione',
},
'user_session_ended': {
'en': 'Session with admin {u} ended.',
'it': 'Sessione terminata con l\'amministratore {u}.',
},
'user_warning': {
'en': (
'{u}, admin of this bot, wants to talk to you.\n'
'Until this session is ended by {u}, your messages will be '
'forwarded to each other.'
),
'it': (
'{u}, amministratore di questo bot, vuole parlare con te.\n'
'Finché non chiuderà la connessione, i messaggi che scriverai '
'qui saranno inoltrati a {u}, e ti inoltrerò i suoi.'
),
},
}
default_unknown_command_message = {
'en': "Unknown command! Touch /help to read the guide and available commands.",
'it': "Comando sconosciuto! Fai /help per leggere la guida e i comandi."
}