krosh/Telegram-bot-schedule
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

code refactoring. add "admin" file

Browse Source
This commit is contained in:
kr0sh512
2023-11-02 16:18:50 +03:00
parent 81febd3e1a
commit 38adcea565
3 changed files with 164 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
admin.py
+28
View File
@@ -0,0 +1,28 @@
import schedules as bot
admin_id = '856850518'
def admin_command(func):
def wrapper(*args, **kwargs):
message = args[0]
if str(message.from_user.id) != admin_id:
bot.send_message(message.from_user.id, 'Команда доступна только администратору')
return
func(*args, **kwargs)
return
return wrapper
def is_admin(message):
if str(message.from_user.id) == admin_id and message.from_user.id == message.chat.id:
return True
return False
def send_admin_message(text):
bot.send_message(admin_id, '🛑 ' + text)
def send_admin_document(file):
bot.send_document(admin_id, file, '🛑 Admin file')
for_json.py
+10 -7
View File
@@ -11,6 +11,7 @@ def save_user(infos):
if type(infos[i]) == type(None): if type(infos[i]) == type(None):
infos[i] = '' infos[i] = ''
infos[i] = str(infos[i]) infos[i] = str(infos[i])
infos = { infos = {
"id" : infos[0], "id" : infos[0],
"first_name" : infos[1], "first_name" : infos[1],
@@ -39,6 +40,7 @@ def save_user(infos):
return return
def change_user_param(id, key, value): def change_user_param(id, key, value):
id = str(id)
data = {} data = {}
with open(path_users, 'r', encoding='utf-8') as json_file: with open(path_users, 'r', encoding='utf-8') as json_file:
data = json.load(json_file) data = json.load(json_file)
@@ -117,7 +119,6 @@ def create_schedule_tasks():
elif day == 'sat': elif day == 'sat':
schedule.every().saturday.at(start_task).do(bot.send_message, id, text, thread_id) schedule.every().saturday.at(start_task).do(bot.send_message, id, text, thread_id)
bot.send_message(bot.admin_id, '🛑 Произошёл update')
return return
@@ -129,19 +130,21 @@ def check_group_in_json(number):
return str(number) in schdl.keys() return str(number) in schdl.keys()
def return_infos(id): def return_infos(id):
id = str(id)
data = {} data = {}
with open(path_users, 'r', encoding='utf-8') as json_file: with open(path_users, 'r', encoding='utf-8') as json_file:
data = json.load(json_file) data = json.load(json_file)
data = data[id] data = data[id]
text = '<i>Выбранная группа</i>: <b>{}</b>\
\n<i>Время напоминания до урока</i>: <b>{}</b>\
\n<i>Разрешены ли напоминания</i>: <b>{}</b>'.format((data['group'] if data['group'] != 'other' else 'не выбрана'),
data['timeout'],
('да' if data['allow_message'] == 'yes' else 'нет'))
return text return data
def pause_bot():
schedule.clear()
return
def get_schedule(id): def get_schedule(id):
id = str(id)
text = '<u>Расписание на сегодня</u>:\n\n' text = '<u>Расписание на сегодня</u>:\n\n'
schdl_today = {} schdl_today = {}
schedules.py
+119 -81
View File
@@ -4,62 +4,79 @@ import os, sys
from telebot import types from telebot import types
from telegram.constants import ParseMode from telegram.constants import ParseMode
import for_json import for_json
from admin import admin_command, is_admin, send_admin_message, send_admin_document
bot = telebot.TeleBot("TOKEN_API") bot = telebot.TeleBot("TOKEN_API")
admin_id = '856850518'
start_txt = 'Привет! Это бот, который будет кидать тебе сообщения перед нужной парой с номером кабинета/фамилией препода\ start_txt = 'Привет! Это бот, который будет кидать тебе сообщения перед нужной парой с номером кабинета/фамилией препода\
\n\nТы можешь настроить отправку сообщений в лс или добавить меня в группу, куда я буду присылать расписание\ \n\nТы можешь настроить отправку сообщений в лс или добавить меня в группу, куда я буду присылать расписание\
\n\nCreated by: @Kr0sH_512' \n\nCreated by: @Kr0sH_512'
@bot.message_handler(commands=['test'])
@bot.message_handler(commands=['update']) @admin_command
def update_schedules(message): def update_schedules(message):
if (str(message.from_user.id) != admin_id): text = '<b>Жирный текст</b>\
bot.send_message(message.from_user.id, 'Команда доступна только администратору', parse_mode=ParseMode.HTML) \n<i>Курсивный текст</i>\
\n<s>Перечёркнутый текст</s>\
\n<a href="google.com">Ссылка</a>\
\n<code>Моноширинный текст</code>\
\n<pre>Форматированный с сохранением пробелов</pre>\
\n<blockquote>Цитата</blockquote>'
send_admin_message(text)
return return
@bot.message_handler(commands=['update'])
@admin_command
def update_schedules(message):
for_json.create_schedule_tasks() for_json.create_schedule_tasks()
send_admin_message('Произошёл update')
return return
@bot.message_handler(commands=['restart']) @bot.message_handler(commands=['restart'])
@admin_command
def restart_bot(message): def restart_bot(message):
if (str(message.from_user.id) != admin_id): send_admin_message('bye')
bot.send_message(message.from_user.id, 'Команда доступна только администратору', parse_mode=ParseMode.HTML)
return
bot.send_message(admin_id, '🛑 bye', parse_mode=ParseMode.HTML)
os.execv(sys.executable, ['python'] + sys.argv) os.execv(sys.executable, ['python'] + sys.argv)
return
@bot.message_handler(commands=['json']) @bot.message_handler(commands=['json'])
@admin_command
def send_json(message): def send_json(message):
if (str(message.from_user.id) != admin_id): with open(for_json.path_users, 'rb') as json_file:
bot.send_message(message.from_user.id, 'Команда доступна только администратору', parse_mode=ParseMode.HTML) send_admin_document(json_file)
with open(for_json.path_schedule, 'rb') as json_file:
send_admin_document(json_file)
return return
with open(for_json.path_users, 'rb') as json_file: @bot.message_handler(commands=['pause_all'])
bot.send_document(admin_id, json_file) @admin_command
with open(for_json.path_schedule, 'rb') as json_file: def pause_bot(message):
bot.send_document(admin_id, json_file) for_json.pause_bot()
send_admin_message('Бот больше не отправляет расписание')
return return
@bot.message_handler(commands=['stop'])
@admin_command
def pause_bot(message):
if len(str(message.text).split(' ')) != 2:
send_admin_message('Эта команда вида /stop <i>id_пользователя</i>')
return
for_json.change_user_param(str(message.text).split(' ')[1], 'allow_message', 'no')
send_admin_message('Успешно')
return
@bot.message_handler(commands=['start']) @bot.message_handler(commands=['start'])
def start(message): def start(message):
if message.chat.type == "supergroup": if message.chat.type == "supergroup":
print(message.text)
print(message.chat.id)
print(message.text.split(' '))
if len(message.text.split(' ')) != 2: if len(message.text.split(' ')) != 2:
bot.send_message(message.chat.id, start_txt, parse_mode='Markdown') send_message(message.chat.id, start_txt)
bot.send_message(message.chat.id, 'Похоже, что этот бот добавлен в группу. Чтобы он присылал расписание в чат, \ send_message(message.chat.id, 'Похоже, что этот бот добавлен в группу. Чтобы он присылал расписание в чат, \
укажите вместе с /start@vmk_schedule_bot номер своей группы.\ укажите вместе с /start@vmk_schedule_bot номер своей группы.\
\nПример:\n/start@vmk_schedule_bot 102', parse_mode=ParseMode.HTML) \nПример:\n/start@vmk_schedule_bot 102')
else: else:
temp = message.text.split(' ')[1] temp = message.text.split(' ')[1]
@@ -73,22 +90,24 @@ def start(message):
] ]
if for_json.check_group_in_json(temp): if for_json.check_group_in_json(temp):
bot.send_message(message.chat.id, \ send_message(message.chat.id,
'Отлично! Теперь я буду присылать вам расписание {} группы'.format(temp), parse_mode=ParseMode.HTML) 'Отлично! Теперь я буду присылать вам расписание {} группы'.format(temp))
else: else:
bot.send_message(message.chat.id, 'Похоже, что расписания для этой группы ещё не существует. \ send_message(message.chat.id, 'Похоже, что расписания для этой группы ещё не существует. \
\nРазработчик уже пинается, но можете дополнительно написать ему: @Kr0sH_512', parse_mode=ParseMode.HTML) \nРазработчик уже пинается, но можете дополнительно написать ему: @Kr0sH_512')
send_message(admin_id, '🛑 В группе {} был запрос на {} группу.\
send_admin_message('В группе {} был запрос на {} группу.\
\nid: {}'.format(message.chat.title, temp, message.chat.id)) \nid: {}'.format(message.chat.title, temp, message.chat.id))
temp = 'other' temp = 'other'
for_json.save_user(infos) for_json.save_user(infos)
else: else:
bot.send_message(message.chat.id, 'Введено не число. Попробуйте снова.\ send_message(message.chat.id, 'Введено не число. Попробуйте снова.\
\nПример:\n/start@vmk_schedule_bot 102', parse_mode=ParseMode.HTML) \nПример:\n/start@vmk_schedule_bot 102')
return return
bot.send_message(message.chat.id, start_txt, parse_mode=ParseMode.HTML) send_message(message.chat.id, start_txt)
markup = types.InlineKeyboardMarkup() markup = types.InlineKeyboardMarkup()
markup.add(types.InlineKeyboardButton(text="102", callback_data="102")) markup.add(types.InlineKeyboardButton(text="102", callback_data="102"))
markup.add(types.InlineKeyboardButton(text="103", callback_data="103")) markup.add(types.InlineKeyboardButton(text="103", callback_data="103"))
@@ -139,29 +158,31 @@ def help(message):
\n/source - Страница бота на Github\ \n/source - Страница бота на Github\
\n\nИли же можно всегда написать напрямую: @Kr0sH_512' \n\nИли же можно всегда написать напрямую: @Kr0sH_512'
bot.send_message(message.chat.id, help_msg, parse_mode=ParseMode.HTML) send_message(message.chat.id, help_msg)
if str(message.from_user.id) == admin_id and message.from_user.id == message.chat.id: if is_admin(message):
help_msg = '🛑 И команды только для админа:\ admin_help_msg = 'И команды только для админа:\
\n/restart - перезапуск бота.\ \n/restart - перезапуск бота.\
\n/update - обновление schedule задач\ \n/update - обновление schedule задач\
\n/json - получить файл пользователей и расписания\ \n/json - получить файл пользователей и расписания\
\n/info id_пользователя - узнать настройки пользователя' \n/info <i>id_пользователя</i> - узнать настройки пользователя\
bot.send_message(message.from_user.id, help_msg, parse_mode=ParseMode.HTML) \n/pause_all - приостановить бота для всех (каникулы/выходные)\
\n/stop <i>id_пользователя</i> - приостановить отправку сообщений для пользователя'
send_admin_message(admin_help_msg)
return return
@bot.message_handler(commands=['request']) @bot.message_handler(commands=['request'])
def request(message): def request(message):
bot.send_message(message.from_user.id, 'Отправь мне сообщение и я перешлю его разработчику\ send_message(message.from_user.id, 'Отправь мне сообщение и я перешлю его разработчику\
\n(или напиши stop, чтобы отменить отправку)', parse_mode=ParseMode.HTML) \n(или напиши stop, чтобы отменить отправку)')
bot.register_next_step_handler(message, send_request) bot.register_next_step_handler(message, send_request)
return return
def send_request(message): def send_request(message):
if message.text == "stop" or message.text == "стоп" or message.text[0] == '/': if message.text == "stop" or message.text == "стоп" or message.text[0] == '/':
bot.send_message(message.from_user.id, 'Отмена отпраки', parse_mode=ParseMode.HTML) send_message(message.from_user.id, 'Отмена отпраки')
return return
infos = [ infos = [
@@ -176,10 +197,10 @@ def send_request(message):
if type(infos[i]) == type(None): if type(infos[i]) == type(None):
infos[i] = '' infos[i] = ''
text = '🛑 Сообщение от {} {} (@{}):\n\n{}\nid: {}'.format(infos[1], infos[2], infos[3], infos[4], infos[0]) admin_text = 'Сообщение от {} {} (@{}):\n\n{}\nid: {}'.format(infos[1], infos[2], infos[3], infos[4], infos[0])
bot.send_message(admin_id, text, parse_mode=ParseMode.HTML) send_admin_message(admin_text)
bot.send_message(message.from_user.id, 'Ваше сообщение успешно доставлено!', parse_mode=ParseMode.HTML) send_message(message.from_user.id, 'Ваше сообщение успешно доставлено!')
return return
@@ -189,51 +210,56 @@ def send_schedule(message):
try: try:
text = for_json.get_schedule(str(message.chat.id)) text = for_json.get_schedule(str(message.chat.id))
except Exception as e: except Exception as e:
bot.send_message(admin_id, send_admin_message('Schedule error from: {} \n\n {}'.format(message.chat.id, e))
'🛑 schedule error from: {} \n\n {}'.format(message.chat.id, e), send_message(message.chat.id, text)
parse_mode=ParseMode.HTML)
bot.send_message(message.chat.id, text, parse_mode=ParseMode.HTML)
return return
@bot.message_handler(commands=['timeout']) @bot.message_handler(commands=['timeout'])
def set_timeout(message): def set_timeout(message):
bot.send_message(message.chat.id, 'Пришли мне число от 1 до 60. \ send_message(message.chat.id, 'Пришли мне число от 1 до 60. \
\nЭто будет количество минут, за которое я буду присылать тебе напоминание о уроке (По умолчанию: 10)', parse_mode=ParseMode.HTML) \nЭто будет количество минут, за которое я буду присылать тебе напоминание о уроке (По умолчанию: 10)')
bot.register_next_step_handler(message, save_timeout) bot.register_next_step_handler(message, save_timeout)
return return
def save_timeout(message): def save_timeout(message):
if message.text[0] == '/':
bot.send_message(message.chat.id, 'Изменения не были внесены', parse_mode=ParseMode.HTML)
return
if str(message.text).isdigit() and int(message.text) <= 60 and int(message.text) >= 1: if str(message.text).isdigit() and int(message.text) <= 60 and int(message.text) >= 1:
for_json.change_user_param(str(message.chat.id), 'timeout', str(message.text)) for_json.change_user_param(message.chat.id, 'timeout', str(message.text))
bot.send_message(message.chat.id, send_message(message.chat.id,
'Хорошо, теперь ты будешь получать напоминания за {} минут до урока'.format(str(message.text)), 'Хорошо, теперь ты будешь получать напоминания за {} минут до урока'.format(str(message.text)))
parse_mode=ParseMode.HTML)
else: else:
bot.send_message(message.chat.id, send_message(message.chat.id,
'Пожалуйста, пришли <u>число</u> от 1 до 60. \ 'Пожалуйста, пришли <u>число</u> от 1 до 60. \
\nПо умолчанию: 10', \nПо умолчанию: 10')
parse_mode=ParseMode.HTML) send_message(message.chat.id, 'Изменения не были внесены')
bot.send_message(message.chat.id, 'Изменения не были внесены', parse_mode=ParseMode.HTML)
return return
@bot.message_handler(commands=['info']) @bot.message_handler(commands=['info'])
def send_info(message): def send_info(message):
if len(message.text.split(' ')) == 2 and str(message.from_user.id) == admin_id: if len(message.text.split(' ')) == 2 and is_admin(message):
bot.send_message(message.chat.id, '🛑 Окей, держи настройки <i>{}</i>'.format(message.text.split(' ')[1]), send_admin_message('Окей, держи настройки <code>{}</code>'.format(message.text.split(' ')[1]))
parse_mode=ParseMode.HTML) infos = for_json.return_infos(message.text.split(' ')[1])
text = for_json.return_infos(message.text.split(' ')[1]) text = '<i>Никнейм</i>: <b>@{}</b>\
bot.send_message(message.chat.id, text, parse_mode=ParseMode.HTML) \n<i>Имя</i>: <b>{} {}</b>\
\n<i>Выбранная группа</i>: <b>{}</b>\
\n<i>Время напоминания до урока</i>: <b>{}</b>\
\n<i>Разрешены ли напоминания</i>: <b>{}</b>'.format(infos['username'], infos['first_name'], infos['last_name'],
(infos['group'] if infos['group'] != 'other' else 'не выбрана'),
infos['timeout'],
('да' if infos['allow_message'] == 'yes' else 'нет'))
send_admin_message(text)
return return
bot.send_message(message.chat.id, 'Хорошо, вот твои настройки:', parse_mode=ParseMode.HTML) send_message(message.chat.id, 'Хорошо, вот твои настройки:')
text = for_json.return_infos(str(message.chat.id)) infos = for_json.return_infos(message.chat.id)
bot.send_message(message.chat.id, text, parse_mode=ParseMode.HTML) text = '<i>Выбранная группа</i>: <b>{}</b>\
\n<i>Время напоминания до урока</i>: <b>{}</b>\
\n<i>Разрешены ли напоминания</i>: <b>{}</b>'.format((infos['group'] if infos['group'] != 'other' else 'не выбрана'),
infos['timeout'],
('да' if infos['allow_message'] == 'yes' else 'нет'))
send_message(message.chat.id, text)
return return
@@ -254,8 +280,7 @@ def change_thread(message):
for_json.change_user_param(str(message.chat.id), 'thread', thread_id) for_json.change_user_param(str(message.chat.id), 'thread', thread_id)
bot.send_message(message.chat.id, 'Хорошо, теперь я буду отправлять сообщения в этот чат', send_message(message.chat.id, 'Хорошо, теперь я буду отправлять сообщения в этот чат', thread_id)
parse_mode='Markdown', message_thread_id=thread_id)
return return
@@ -263,8 +288,8 @@ def change_thread(message):
def pause_schedule(message): def pause_schedule(message):
for_json.change_user_param(str(message.chat.id), 'allow_message', 'no') for_json.change_user_param(str(message.chat.id), 'allow_message', 'no')
bot.send_message(message.chat.id, 'Рассылка сообщений преращена. \ send_message(message.chat.id, 'Рассылка сообщений преращена. \
\nДля возобновления воспользуйтесь командой\n/start', parse_mode='Markdown') \nДля возобновления воспользуйтесь командой\n/start')
return return
@@ -272,8 +297,8 @@ def pause_schedule(message):
def text_message(message): def text_message(message):
if message.chat.type == "supergroup": if message.chat.type == "supergroup":
return return
bot.send_message(message.from_user.id, 'К сожалению, я ещё не умею обрабатывать сообщения.\ send_message(message.from_user.id, 'К сожалению, я ещё не умею обрабатывать сообщения.\
\nИспользуй команды из меню или напиши /help.', parse_mode='Markdown') \nИспользуй команды из меню или напиши /help.')
return return
@@ -282,23 +307,36 @@ def send_message(id, text, thread_id='General'):
thread_id = None thread_id = None
try: try:
bot.send_message(chat_id=id, text=text, parse_mode=ParseMode.HTML, message_thread_id=thread_id) bot.send_message(chat_id=id, text=text, parse_mode=ParseMode.HTML, message_thread_id=thread_id,disable_web_page_preview=True)
except Exception as e: except Exception as e:
text_error = '🛑 user:' + id + 'error.\n' + str(e)
bot.send_message(admin_id, text_error, parse_mode=ParseMode.HTML, message_thread_id=thread_id)
print(id, e)
time.sleep(5) time.sleep(5)
try: try:
bot.send_message(chat_id=id, text=text, parse_mode=ParseMode.HTML, message_thread_id=thread_id) bot.send_message(chat_id=id, text=text, parse_mode=ParseMode.HTML, message_thread_id=thread_id,disable_web_page_preview=True)
except Exception as e: except Exception as e:
text_error = 'Error from user: <code>{}</code>\n{}'.format(id, str(e))
send_admin_message(text_error)
print(id, e) print(id, e)
return return
def send_document(id, file, text = ''):
try:
bot.send_document(id, file, caption=text)
except Exception as e:
time.sleep(5)
try:
bot.send_document(id, file, caption=text)
except Exception as e:
text_error = 'Error from user: <code>{}</code>\n{}'.format(id, str(e))
send_admin_message(text_error)
print(id, e)
return
if __name__ == '__main__': if __name__ == '__main__':
for_json.create_schedule_tasks() for_json.create_schedule_tasks()
send_message(admin_id, '🛑 я перезапустился!')
send_admin_message('Я перезапустился!')
print("-------------------------") print("-------------------------")
threading.Thread(target=bot.infinity_polling, name='bot_infinity_polling', daemon=True).start() threading.Thread(target=bot.infinity_polling, name='bot_infinity_polling', daemon=True).start()